Skip to content

Commit

Permalink
manual token acquisition added
Browse files Browse the repository at this point in the history
- 新增 引导手动获取 Token 功能
- 感谢 @github/ZipFile 的解决方案与代码分享
  • Loading branch information
txperl committed Feb 13, 2021
1 parent d618da5 commit b1d340d
Show file tree
Hide file tree
Showing 3 changed files with 179 additions and 59 deletions.
88 changes: 88 additions & 0 deletions app/core/biu/login_token.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
# thanks to @github/ZipFile, https://gist.github.com/ZipFile/c9ebedb224406f4f11845ab700124362
import platform
import requests
from base64 import urlsafe_b64encode
from hashlib import sha256
from secrets import token_urlsafe
from urllib.parse import urlencode

from ...lib.common.msg import biuMsg
from ...lib.common.util import util

USER_AGENT = "PixivAndroidApp/5.0.234 (Android 11; Pixel 5)"
REDIRECT_URI = "https://app-api.pixiv.net/web/v1/users/auth/pixiv/callback"
LOGIN_URL = "https://app-api.pixiv.net/web/v1/login"
AUTH_TOKEN_URL = "https://oauth.secure.pixiv.net/auth/token"
CLIENT_ID = "MOBrBDS8blbauoSck0ZfDbtuzpyT"
CLIENT_SECRET = "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj"


class login_with_token(object):
def __init__(self):
self.proxy = util.getSystemProxy(platform.system())
self.code_verifier, self.code_challenge = self.oauth_pkce(self.s256)
self.login_params = {
"code_challenge": self.code_challenge,
"code_challenge_method": "S256",
"client": "pixiv-android",
}
self.msger = biuMsg("Login")

def run(self):
if self.proxy == "":
tip = self.msger.msg("无法获取代理监听地址,是否要手动输入? (y / n): ", out=False)
else:
tip = self.msger.msg(f"获取到为 {self.proxy} 的代理监听地址,是否需要修改? (y / n): ", out=False)
tmp = input(tip)
if tmp == "y":
self.proxy = input("请输入代理监听地址: ")

self.msger.arr(
"请按以下步骤进行操作:",
f"1. 访问「{LOGIN_URL}?{urlencode(self.login_params)}」",
"2. 打开浏览器的「开发者工具 / dev console / F12」,切换至「Network」标签",
"3. 开启「Preserve log」",
"4. 在「Filter」文本框中输入「callback?」",
"5. 登入您的 Pixiv 账号",
"6. 成功登陆后,会出现一个类似「https://app-api.pixiv.net/.../callback?state=...&code=...」的字段,"
"将「code」后面的参数输入本程序"
)

return self.login()

def s256(self, data):
"""S256 transformation method."""

return urlsafe_b64encode(sha256(data).digest()).rstrip(b"=").decode("ascii")

def oauth_pkce(self, transform):
"""Proof Key for Code Exchange by OAuth Public Clients (RFC7636)."""

code_verifier = token_urlsafe(32)
code_challenge = transform(code_verifier.encode("ascii"))

return code_verifier, code_challenge

def login(self):
code = input("code: ").strip()

response = requests.post(
AUTH_TOKEN_URL,
data={
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"code": code,
"code_verifier": self.code_verifier,
"grant_type": "authorization_code",
"include_policy": "true",
"redirect_uri": REDIRECT_URI,
},
headers={"User-Agent": USER_AGENT},
proxies={"https": self.proxy}
)
rst = response.json()

if "refresh_token" in rst:
return rst["refresh_token"]
else:
return None
88 changes: 29 additions & 59 deletions app/core/biu/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,10 @@
import requests
from pixivpy3 import *

from .login_token import login_with_token
from ..file.main import core_module_file as ifile
from ...lib.common.msg import biuMsg
from ...lib.common.util import util
from ...platform import CMDProcessor


Expand All @@ -39,6 +41,8 @@ def __init__(self, info=None):
self.STATUS = {"rate_search": {}, "rate_download": {}}
# lib-common 类
self.msger = biuMsg("PixivBiu")
# 暂时
self.sets["account"]["isToken"] = True

def __del__(self):
self.pool_srh.shutdown(False)
Expand Down Expand Up @@ -81,67 +85,16 @@ def __preCheck(self):
sys.exit(0)

def __getSystemProxy(self):
"""
检测系统本地设置中的代理地址,并验证是否可用。
@Windows: 通过注册表项获取
@macOS: 通过 scutil 获取
@Linux: 暂时未实现
"""
if self.sets["biu"]["common"]["proxy"] == "no":
return ""

if self.apiType == "byPassSni" or self.sets["biu"]["common"]["proxy"] != "":
return self.sets["biu"]["common"]["proxy"]

proxies = []
cmd = ""

if self.sysPlc == "Windows":
cmd = r'reg query "HKEY_CURRENT_USER\Software\Microsoft\Windows\CurrentVersion\Internet Settings" | ' \
r'findstr "ProxyServer AutoConfigURL" '
elif self.sysPlc == "Darwin":
cmd = "scutil --proxy"
else:
return ""
url = util.getSystemProxy(self.sysPlc)
if url != "":
self.msger.msg(f"已启用系统代理地址: {url}")

# 获取系统 console 执行结果
cmdRstObj = os.popen(cmd)
cmdRst = cmdRstObj.read()
cmdRstObj.close()
cmdRstArr = cmdRst.split("\n")[:-1]
proxies = [re.split(r"\s+", x)[1:] for x in cmdRstArr]

# 筛选出可用代理地址
for i in range(len(proxies) - 1, -1, -1):
x = proxies[i]
if len(x) < 3:
continue
add = prt = None

if self.sysPlc == "Windows":
tmp = re.match(r"https?:\/\/(.*?):(\d+)", x[2], re.IGNORECASE)
if tmp is None:
continue
add = tmp.group(1)
prt = int(tmp.group(2))
elif self.sysPlc == "Darwin":
tmp = re.match(r"https?proxy", x[0], re.IGNORECASE)
if tmp is None:
continue
add = proxies[i][2]
prt = int(proxies[i - 1][2])

# 检测本地是否可通
if add and prt:
try:
telnetlib.Telnet(add, port=prt, timeout=1)
url = f"http://{add}:{prt}/"
self.msger.msg(f"已启用系统代理地址: {url}")
return url
except:
pass

return ""
return url

def __getBiuInfo(self):
"""
Expand Down Expand Up @@ -202,10 +155,27 @@ def __login(self):
else:
self.__loginPublicAPI(**args)
except Exception as e:
self.msger.error(e, header=False)
self.msger.red("Pixiv 登陆失败")
input("按任意键退出...")
sys.exit(0)
try:
self.msger.msg("由于 Pixiv 禁止了目前使用的 Login API 账号密码登陆方式,暂时只能使用 Token 进行登陆")
if input("是否开始手动获取 Token 后继续? (y / n): ") != "y":
raise Exception("user cancelled")
login = login_with_token()
token = login.run()
if token is None:
raise Exception("request error")
if self.sets["account"]["isToken"]:
ifile.aout(
self.ENVORON["ROOTPATH"] + "usr/.token.json",
{"token": token, },
dRename=False,
)
self.__login()
except Exception as ee:
self.msger.error(e, header=False)
self.msger.error(ee, header=False)
self.msger.red("Pixiv 登陆失败")
input("按任意键退出...")
sys.exit(0)

def __loadAccountInfo(self):
"""
Expand Down
62 changes: 62 additions & 0 deletions app/lib/common/util.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
import os
import re
import telnetlib


class util(object):
@staticmethod
def getSystemProxy(sysPlc):
"""
检测系统本地设置中的代理地址,并验证是否可用。
@Windows: 通过注册表项获取
@macOS: 通过 scutil 获取
@Linux: 暂时未实现
"""
proxies = []
cmd = ""

if sysPlc == "Windows":
cmd = r'reg query "HKEY_CURRENT_USER\Software\Microsoft\Windows\CurrentVersion\Internet Settings" | ' \
r'findstr "ProxyServer AutoConfigURL" '
elif sysPlc == "Darwin":
cmd = "scutil --proxy"
else:
return ""

# 获取系统 console 执行结果
cmdRstObj = os.popen(cmd)
cmdRst = cmdRstObj.read()
cmdRstObj.close()
cmdRstArr = cmdRst.split("\n")[:-1]
proxies = [re.split(r"\s+", x)[1:] for x in cmdRstArr]

# 筛选出可用代理地址
for i in range(len(proxies) - 1, -1, -1):
x = proxies[i]
if len(x) < 3:
continue
add = prt = None

if sysPlc == "Windows":
tmp = re.match(r"https?:\/\/(.*?):(\d+)", x[2], re.IGNORECASE)
if tmp is None:
continue
add = tmp.group(1)
prt = int(tmp.group(2))
elif sysPlc == "Darwin":
tmp = re.match(r"https?proxy", x[0], re.IGNORECASE)
if tmp is None:
continue
add = proxies[i][2]
prt = int(proxies[i - 1][2])

# 检测本地是否可通
if add and prt:
try:
telnetlib.Telnet(add, port=prt, timeout=1)
url = f"http://{add}:{prt}/"
return url
except:
pass

return ""

0 comments on commit b1d340d

Please sign in to comment.