Merge pull request #10 from chrisis58/dev

fix: 修复部分存在的问题，优化使用体验

Author: chrisis58

README.md

@@ -28,15 +28,13 @@ pip install kmoe-manga-downloader
 
 ```bash
 kmdr login -u <your_username> -p <your_password>
-```
-
-或者：
-
-```bash
+# 或者
 kmdr login -u <your_username>
 ```
 
-第二种方式会在程序运行时获取登录密码。如果登录成功，会同时显示当前登录用户及配额。
+第二种方式会在程序运行时获取登录密码，此时你输入的密码**不会显示**在终端中。
+
+如果登录成功，会同时显示当前登录用户及配额。
 
 ### 2. 下载漫画书籍
 

src/kmdr/core/__init__.py

@@ -2,4 +2,6 @@
 from .structure import VolInfo, BookInfo, VolumeType
 from .bases import AUTHENTICATOR, LISTERS, PICKERS, DOWNLOADER, CONFIGURER
 
-from .defaults import argument_parser
+from .defaults import argument_parser
+
+from .error import KmdrError, LoginError

src/kmdr/core/bases.py

@@ -2,17 +2,24 @@
 
 from typing import Callable, Optional
 
+from .error import LoginError
 from .registry import Registry
 from .structure import VolInfo, BookInfo
 from .utils import get_singleton_session, construct_callback
-from .defaults import Configurer as InnerConfigurer
+from .defaults import Configurer as InnerConfigurer, UserProfile
 
 class SessionContext:
 
     def __init__(self, *args, **kwargs):
         super().__init__()
         self._session = get_singleton_session()
 
+class UserProfileContext:
+
+    def __init__(self, *args, **kwargs):
+        super().__init__()
+        self._profile = UserProfile()
+
 class ConfigContext:
 
     def __init__(self, *args, **kwargs):
@@ -26,7 +33,7 @@ def __init__(self, *args, **kwargs):
 
     def operate(self) -> None: ...
 
-class Authenticator(SessionContext, ConfigContext):
+class Authenticator(SessionContext, ConfigContext, UserProfileContext):
 
     def __init__(self, proxy: Optional[str] = None, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -41,8 +48,13 @@ def __init__(self, proxy: Optional[str] = None, *args, **kwargs):
     # 主站正常情况下不使用代理也能登录成功。但是不排除特殊的网络环境下需要代理。
     # 所以暂时保留代理登录的功能，如果后续确认是代理的问题，可以考虑启用 @no_proxy 装饰器。
     # @no_proxy
-    def authenticate(self) -> bool:
-        return self._authenticate()
+    def authenticate(self) -> None:
+        try:
+            assert self._authenticate()
+        except LoginError as e:
+            print("Authentication failed. Please check your login credentials or session cookies.")
+            print(f"Details: {e}")
+            exit(1)
 
     def _authenticate(self) -> bool: ...
 
@@ -60,7 +72,7 @@ def __init__(self, *args, **kwargs):
 
     def pick(self, volumes: list[VolInfo]) -> list[VolInfo]: ...
 
-class Downloader(SessionContext):
+class Downloader(SessionContext, UserProfileContext):
 
     def __init__(self, 
             dest: str = '.',

src/kmdr/core/defaults.py

@@ -58,6 +58,29 @@ def parse_args():
 
     return args
 
+@singleton
+class UserProfile:
+
+    def __init__(self):
+        self._is_vip: Optional[int] = None
+        self._user_level: Optional[int] = None
+
+    @property
+    def is_vip(self) -> Optional[int]:
+        return self._is_vip
+
+    @property
+    def user_level(self) -> Optional[int]:
+        return self._user_level
+    
+    @is_vip.setter
+    def is_vip(self, value: Optional[int]):
+        self._is_vip = value
+
+    @user_level.setter
+    def user_level(self, value: Optional[int]):
+        self._user_level = value
+
 @singleton
 class Configurer:
 

src/kmdr/core/error.py

@@ -0,0 +1,15 @@
+from typing import Optional
+
+class KmdrError(RuntimeError):
+    def __init__(self, message: str, solution: Optional[list[str]] = None, *args: object, **kwargs: object):
+        super().__init__(message, *args, **kwargs)
+        self.message = message
+
+        self._solution = "" if solution is None else "\nSuggested Solution: \n" + "\n".join(f">>> {sol}" for sol in solution)
+
+class LoginError(KmdrError):
+    def __init__(self, message, solution: Optional[list[str]] = None):
+        super().__init__(message, solution)
+
+    def __str__(self):
+        return f"{self.message}\n{self._solution}"

src/kmdr/main.py

@@ -7,16 +7,13 @@
 def main(args: Namespace, fallback: Callable[[], None] = lambda: print('NOT IMPLEMENTED!')) -> None:
 
     if args.command == 'login':
-        if not AUTHENTICATOR.get(args).authenticate():
-            raise RuntimeError("Authentication failed. Please check your credentials.")
+        AUTHENTICATOR.get(args).authenticate()
 
     elif args.command == 'status':
-        if not AUTHENTICATOR.get(args).authenticate():
-            raise RuntimeError("Authentication failed. Please check your credentials.")
+        AUTHENTICATOR.get(args).authenticate()
 
-    elif args.command == 'download':    
-        if not AUTHENTICATOR.get(args).authenticate():
-            raise RuntimeError("Authentication failed. Please check your credentials.")
+    elif args.command == 'download':
+        AUTHENTICATOR.get(args).authenticate()
 
         book, volumes = LISTERS.get(args).list()
 

src/kmdr/module/authenticator/CookieAuthenticator.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from kmdr.core import Authenticator, AUTHENTICATOR
+from kmdr.core import Authenticator, AUTHENTICATOR, LoginError
 
 from .utils import check_status
 
@@ -18,8 +18,12 @@ def _authenticate(self) -> bool:
         cookie = self._configurer.cookie
 
         if not cookie:
-            print("No cookie found. Please login first.")
-            return False
+            raise LoginError("No cookie found, please login first.", ['kmdr login -u <username>'])
 
         self._session.cookies.update(cookie)
-        return check_status(self._session, show_quota=self._show_quota)
+        return check_status(
+            self._session,
+            show_quota=self._show_quota,
+            is_vip_setter=lambda value: setattr(self._profile, 'is_vip', value),
+            level_setter=lambda value: setattr(self._profile, 'user_level', value),
+        )

src/kmdr/module/authenticator/LoginAuthenticator.py

@@ -1,10 +1,19 @@
 from typing import Optional
 import re
+from getpass import getpass
 
-from kmdr.core import Authenticator, AUTHENTICATOR
+from kmdr.core import Authenticator, AUTHENTICATOR, LoginError
 
 from .utils import check_status
 
+CODE_OK = 'm100'
+
+CODE_MAPPING = {
+    'e400': "帳號或密碼錯誤。",
+    'e401': "非法訪問，請使用瀏覽器正常打開本站",
+    'e402': "帳號已經註銷。不會解釋原因，無需提問。",
+    'e403': "驗證失效，請刷新頁面重新操作。",
+}
 
 @AUTHENTICATOR.register(
     hasvalues = {'command': 'login'}
@@ -16,7 +25,7 @@ def __init__(self, username: str, proxy: Optional[str] = None, password: Optiona
         self._show_quota = show_quota
 
         if password is None:
-            password = input("please input your password: \n")
+            password = getpass("please input your password: ")
 
         self._password = password
 
@@ -31,22 +40,15 @@ def _authenticate(self) -> bool:
             },
         )
         response.raise_for_status()
-        
-        match = re.search('"\w+"', response.text)
+        match = re.search(r'"\w+"', response.text)
+
         if not match:
-            raise RuntimeError("Failed to extract authentication code from response.")
-        code = match.group(0).split('"')[1]
-        if code != 'm100':
-            if code == 'e400':
-                print("帳號或密碼錯誤。")
-            elif code == 'e401':
-                print("非法訪問，請使用瀏覽器正常打開本站")
-            elif code == 'e402':
-                print("帳號已經註銷。不會解釋原因，無需提問。")
-            elif code == 'e403':
-                print("驗證失效，請刷新頁面重新操作。")
-            raise RuntimeError("Authentication failed with code: " + code)
+            raise LoginError("Failed to extract authentication code from response.")
 
+        code = match.group(0).split('"')[1]
+        if code != CODE_OK:
+            raise LoginError(f"Authentication failed with error code: {code} " + CODE_MAPPING.get(code, "Unknown error."))
+
         if check_status(self._session, show_quota=self._show_quota):
             self._configurer.cookie = self._session.cookies.get_dict()
             return True

src/kmdr/module/authenticator/utils.py

@@ -1,25 +1,79 @@
+from typing import Optional, Callable
+
 from requests import Session
 
-def check_status(session: Session, show_quota: bool = False) -> bool:
-    response = session.get(url = 'https://kox.moe/my.php')
+from kmdr.core.error import LoginError
+
+PROFILE_URL = 'https://kox.moe/my.php'
+LOGIN_URL = 'https://kox.moe/login.php'
+
+NICKNAME_ID = 'div_nickname_display'
+
+VIP_ID = 'div_user_vip'
+NOR_ID = 'div_user_nor'
+LV1_ID = 'div_user_lv1'
+
+def check_status(
+        session: Session,
+        show_quota: bool = False,
+        is_vip_setter: Optional[Callable[[int], None]] = None,
+        level_setter: Optional[Callable[[int], None]] = None
+) -> bool:
+    response = session.get(url = PROFILE_URL)
 
     try:
         response.raise_for_status()
     except Exception as e:
         print(f"Error: {type(e).__name__}: {e}")
         return False
+    
+    if response.history and any(resp.status_code in (301, 302, 307) for resp in response.history) \
+        and response.url == LOGIN_URL:
+        raise LoginError("Invalid credentials, please login again.", ['kmdr config -c cookie', 'kmdr login -u <username>'])
 
-    if not show_quota:
+    if not is_vip_setter and not level_setter and not show_quota:
         return True
 
     from bs4 import BeautifulSoup
 
     soup = BeautifulSoup(response.text, 'html.parser')
+
+    script = soup.find('script', language="javascript")
+
+    if script:
+        var_define = extract_var_define(script.text[:100])
+
+        is_vip = int(var_define.get('is_vip', '0'))
+        user_level = int(var_define.get('user_level', '0'))
+
+        if is_vip_setter:
+            is_vip_setter(is_vip)
+        if level_setter:
+            level_setter(user_level)
 
-    nickname = soup.find('div', id='div_nickname_display').text.strip().split(' ')[0]
-    print(f"=========================\n\nLogged in as {nickname}\n\n=========================\n")
-    
-    quota = soup.find('div', id='div_user_vip').text.strip()
-    print(f"=========================\n\n{quota}\n\n=========================\n")
+    if not show_quota:
+        return True
+
+    nickname = soup.find('div', id=NICKNAME_ID).text.strip().split(' ')[0]
+    quota = soup.find('div', id=__resolve_quota_id(is_vip, user_level)).text.strip()
+
+    print(f"\n当前登录为 {nickname}\n\n{quota}")
     return True
+
+def extract_var_define(script_text) -> dict[str, str]:
+    var_define = {}
+    for line in script_text.splitlines():
+        line = line.strip()
+        if line.startswith("var ") and "=" in line:
+            var_name, var_value = line[4:].split("=", 1)
+            var_define[var_name.strip()] = var_value.strip().strip(";").strip('"')
+    return var_define
+
+def __resolve_quota_id(is_vip: Optional[int] = None, user_level: Optional[int] = None):
+    if is_vip is not None and is_vip >= 1:
+        return VIP_ID
+    
+    if user_level is not None and user_level <= 1:
+        return LV1_ID
 
+    return NOR_ID

src/kmdr/module/configurer/ConfigUnsetter.py

@@ -1,5 +1,7 @@
 from kmdr.core import Configurer, CONFIGURER
 
+from .option_validate import check_key
+
 @CONFIGURER.register()
 class ConfigUnsetter(Configurer):
     def __init__(self, unset: str, *args, **kwargs):
@@ -10,6 +12,7 @@ def operate(self) -> None:
         if not self._unset:
             print("No option specified to unset.")
             return
-        
+
+        check_key(self._unset)
         self._configurer.unset_option(self._unset)
         print(f"Unset configuration: {self._unset}")