Proxy support and refactor codes

README.md

@@ -41,6 +41,10 @@ aapi.auth(refresh_token=REFRESH_TOKEN)
 ...
 ```
 
+## Enable Proxy
+
+Set `ALL_PROXY` or `HTTPS_PROXY` to your environment variables.
+
 ### From Docker
 
 ```shellsession
@@ -81,11 +85,12 @@ expires_in: 3600
 
 ```python
 from gppt import GetPixivToken
+
 g = GetPixivToken()
-res = g.login(headless=True, user="...", pass_="...")
+res = g.login(headless=True, username="...", password="...")
 ```
 
-- `res.response` returns
+- `res.response` returns:
 
 ```json
 {

gppt/__init__.py

@@ -5,7 +5,6 @@
     CLIENT_SECRET,
     LOGIN_URL,
     REDIRECT_URI,
-    REQUESTS_KWARGS,
     USER_AGENT,
     GetPixivToken,
 )
@@ -36,7 +35,6 @@
     "CLIENT_SECRET",
     "LOGIN_URL",
     "REDIRECT_URI",
-    "REQUESTS_KWARGS",
     "USER_AGENT",
     "GetPixivToken",
 ]

gppt/_selenium.py

@@ -14,6 +14,7 @@
 from time import sleep
 from typing import Any, cast
 from urllib.parse import urlencode
+from urllib.request import getproxies
 
 import pyderman
 import requests
@@ -34,17 +35,13 @@
 AUTH_TOKEN_URL = "https://oauth.secure.pixiv.net/auth/token"
 CLIENT_ID = "MOBrBDS8blbauoSck0ZfDbtuzpyT"
 CLIENT_SECRET = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj"
-REQUESTS_KWARGS: dict[str, Any] = {
-    # 'proxies': {
-    #     'https': 'http://127.0.0.1:1087',
-    # },
-    # 'verify': False
-}
+PROXIES = getproxies()
+
+OptionsType = webdriver.chrome.options.Options
 
 
 class GetPixivToken:
     def __init__(self) -> None:
-
         self.caps = DesiredCapabilities.CHROME.copy()
         self.caps["goog:loggingPrefs"] = {
             "performance": "ALL"
@@ -53,25 +50,22 @@ def __init__(self) -> None:
     def login(
         self,
         headless: bool | None = False,
-        user: str | None = None,
-        pass_: str | None = None,
+        username: str | None = None,
+        password: str | None = None,
     ) -> LoginInfo:
-        self.headless, self.user, self.pass_ = headless, user, pass_
+        self.headless = headless
+        self.username = username
+        self.password = password
+
         executable_path = pyderman.install(verbose=False, browser=pyderman.chrome)
         if type(executable_path) is not str:
             raise ValueError("Executable path is not str somehow.")
-        if headless is not None and headless:
-            opts = self.__get_headless_option()
-            self.driver = webdriver.Chrome(
-                executable_path=executable_path,
-                options=opts,
-                desired_capabilities=self.caps,
-            )
 
-        else:
-            self.driver = webdriver.Chrome(
-                executable_path=executable_path, desired_capabilities=self.caps
-            )
+        self.driver = webdriver.Chrome(
+            executable_path=executable_path,
+            options=self.__get_chrome_option(headless),
+            desired_capabilities=self.caps,
+        )
 
         code_verifier, code_challenge = self.__oauth_pkce()
         login_params = {
@@ -110,7 +104,7 @@ def login(
                 "app-os-version": "14.6",
                 "app-os": "ios",
             },
-            **REQUESTS_KWARGS,
+            proxies=PROXIES,
         )
 
         return cast(LoginInfo, response.json())
@@ -131,20 +125,20 @@ def refresh(refresh_token: str) -> LoginInfo:
                 "app-os-version": "14.6",
                 "app-os": "ios",
             },
-            **REQUESTS_KWARGS,
+            proxies=PROXIES,
         )
         return cast(LoginInfo, response.json())
 
     def __fill_login_form(self) -> None:
-        if self.user is not None:
+        if self.username:
             el = self.driver.find_element(By.XPATH, "//input[@autocomplete='username']")
-            self.__slow_type(el, self.user)
+            self.__slow_type(el, self.username)
 
-        if self.pass_ is not None:
+        if self.password:
             el = self.driver.find_element(
                 By.XPATH, "//input[@autocomplete='current-password']"
             )
-            self.__slow_type(el, self.pass_)
+            self.__slow_type(el, self.password)
 
     @staticmethod
     def __slow_type(elm: Any, text: str) -> None:
@@ -154,8 +148,12 @@ def __slow_type(elm: Any, text: str) -> None:
 
     def __try_login(self) -> None:
         if self.headless:
+            label_selectors = [
+                f"contains(text(), '{label}')"
+                for label in ["ログイン", "Login", "登录", "로그인", "登入"]
+            ]
             el = self.driver.find_element(
-                By.XPATH, "//button[@type='submit'][contains(text(), 'Login')]"
+                By.XPATH, f"//button[@type='submit'][{' or '.join(label_selectors)}]"
             )
             el.send_keys(Keys.ENTER)
 
@@ -176,22 +174,35 @@ def __try_login(self) -> None:
             )
 
     @staticmethod
-    def __get_headless_option() -> webdriver.chrome.options.Options:
+    def __get_chrome_option(headless: bool | None) -> OptionsType:
         options = webdriver.ChromeOptions()
-        options.add_argument("--headless")
-        options.add_argument("--disable-gpu")
-        options.add_argument("--disable-extensions")
-        options.add_argument("--disable-infobars")
-        options.add_argument("--disable-dev-shm-usage")
-        options.add_argument("--disable-browser-side-navigation")
-        options.add_argument('--proxy-server="direct://"')
-        options.add_argument("--proxy-bypass-list=*")
-        options.add_argument("--start-maximized")
-        options.add_argument("--no-sandbox")
-        options.add_argument("--disable-dev-shm-usage")
-        options.add_argument("--user-agent=" + USER_AGENT)
-        options.add_experimental_option("excludeSwitches", ["enable-automation"])
-        options.add_experimental_option("useAutomationExtension", False)
+
+        if headless:
+            options.add_argument("--headless")
+            options.add_argument("--disable-gpu")
+            options.add_argument("--disable-extensions")
+            options.add_argument("--disable-infobars")
+            options.add_argument("--disable-dev-shm-usage")
+            options.add_argument("--disable-browser-side-navigation")
+            options.add_argument('--proxy-server="direct://"')
+            options.add_argument("--proxy-bypass-list=*")
+            options.add_argument("--start-maximized")
+            options.add_argument("--no-sandbox")
+            options.add_argument("--disable-dev-shm-usage")
+            options.add_argument("--user-agent=" + USER_AGENT)
+            options.add_experimental_option("excludeSwitches", ["enable-automation"])
+            options.add_experimental_option("useAutomationExtension", False)
+
+        if "all" in PROXIES:
+            options.add_argument(f"--proxy-server={PROXIES['all']}")
+        elif "https" in PROXIES:
+            options.add_argument(f"--proxy-server={PROXIES['https']}")
+        elif "http" in PROXIES:
+            options.add_argument(f"--proxy-server={PROXIES['http']}")
+        else:
+            options.add_argument('--proxy-server="direct://"')
+            options.add_argument("--proxy-bypass-list=*")
+
         return options
 
     @staticmethod
@@ -229,24 +240,5 @@ def __parse_log(self) -> str | None:
             if url is not None and str(url).startswith("pixiv://"):
                 m = re.search(r"code=([^&]*)", url)
                 return None if m is None else str(m.group(1))
-        return None
 
-    # Example of pref log:
-    #
-    # {
-    #     'level': 'INFO',
-    #     'message': '{
-    #             "message":
-    #                 {
-    #                     "method": "Network.loadingFinished",
-    #                     "params":{
-    #                         "encodedDataLength": 0,
-    #                         "requestId": "B13E7DBAD4EB28AAD6B05EEA5D628CE5",
-    #                         "shouldReportCorbBlocking": false,
-    #                         "timestamp":105426.696689
-    #                     }
-    #                 },
-    #             "webview": "9D08CF686401F5B87539217E751861DD"
-    #     }',
-    #     'timestamp': 1653700994895
-    # }
+        return None

gppt/auth.py

@@ -38,7 +38,7 @@ def __auth(self, cnt: int) -> tuple[AppPixivAPI, LoginInfo]:
             print("\x1b[?25l[+]: Login...")
             login_info = aapi.auth(refresh_token=ref)
         elif login_cred is None or cnt > 0:
-            print("[+]: ID is mail address, userid, account name.")
+            print("[+]: ID is mail address, username, or account name.")
             stdin_login = (
                 pwinput.pwinput(prompt="[?]: ID: ", mask=" "),
                 pwinput.pwinput(prompt="[?]: PW: ", mask=" "),
@@ -55,7 +55,7 @@ def __auth(self, cnt: int) -> tuple[AppPixivAPI, LoginInfo]:
     @staticmethod
     def get_refresh_token(pixiv_id: str, pixiv_pass: str) -> str:
         g = GetPixivToken()
-        res = g.login(headless=True, user=pixiv_id, pass_=pixiv_pass)
+        res = g.login(headless=True, username=pixiv_id, password=pixiv_pass)
         return res["refresh_token"]
 
     def read_client_cred(self) -> LoginCred | None:

gppt/main.py

@@ -40,7 +40,7 @@ def print_auth_token_response(res: LoginInfo, json: bool | None = False) -> None
 def func_login(ns: argparse.Namespace) -> None:
     g = GetPixivToken()
     print("[!]: Chrome browser will be launched. Please login.", file=stderr)
-    res = g.login(user=ns.username, pass_=ns.password)
+    res = g.login(username=ns.username, password=ns.password)
     print("[+]: Success!", file=stderr)
     print_auth_token_response(res, json=ns.json)
 
@@ -54,7 +54,7 @@ def func_logini(ns: argparse.Namespace) -> None:
 
 def func_loginh(ns: argparse.Namespace) -> None:
     g = GetPixivToken()
-    res = g.login(headless=True, user=ns.username, pass_=ns.password)
+    res = g.login(headless=True, username=ns.username, password=ns.password)
     print("[+]: Success!", file=stderr)
     print_auth_token_response(res, json=ns.json)