新增 Pixiv token 刷新守护:每 30-60 分钟随机续期,refresh_token 轮换原子落盘
- refresh_token.py:启动即刷新,random 30~60min 循环,失败 5min 重试 - 每次保存新 access_token+refresh_token(Pixiv 会轮换),chmod 600 - pixiv-token-refresh.service:常驻守护,Restart=always - README 同步服务清单与部署说明
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# refresh_token.py — Pixiv OAuth token 定时刷新守护
|
||||
# 逻辑(照旧): 启动立即刷新一次, 之后每 random 30~60 分钟刷新一次, 失败 5 分钟后重试。
|
||||
# 每次刷新把新的 access_token + refresh_token 原子落盘:
|
||||
# Pixiv 每次刷新会轮换 refresh_token, 必须保存新的, 否则续期链会断。
|
||||
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
import urllib3
|
||||
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
|
||||
TOKEN_PATH = Path(__file__).parent / "tokens.json"
|
||||
UA = "PixivIOSApp/7.13.0 (iOS 14.6; iPhone)"
|
||||
INTERVAL_RANGE = (30 * 60, 60 * 60) # 半小时 ~ 一小时 之间随机
|
||||
RETRY_INTERVAL = 5 * 60 # 失败后 5 分钟重试
|
||||
|
||||
|
||||
def log(msg):
|
||||
print(time.strftime("[%Y-%m-%d %H:%M:%S]") + " " + str(msg), flush=True)
|
||||
|
||||
|
||||
def do_refresh():
|
||||
data = json.loads(TOKEN_PATH.read_text(encoding="utf-8"))
|
||||
rt = data.get("refresh_token")
|
||||
if not rt:
|
||||
raise RuntimeError("tokens.json 缺少 refresh_token")
|
||||
|
||||
payload = {
|
||||
"client_id": "MOBrBDS8blbauoSck0ZfDbtuzpyT",
|
||||
"client_secret": "lsACyCD94FhDUtGTXi3QzcFE2uU1hqtDaKeqrdwj",
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": rt,
|
||||
"include_policy": 1,
|
||||
"get_secure_url": 1,
|
||||
}
|
||||
headers = {"User-Agent": UA, "App-OS": "iOS", "App-OS-Version": "14.6"}
|
||||
try:
|
||||
resp = requests.post("https://oauth.secure.pixiv.net/auth/token",
|
||||
data=payload, headers=headers, timeout=20, verify=True)
|
||||
verify_mode = "CA验证"
|
||||
except requests.exceptions.SSLError:
|
||||
resp = requests.post("https://oauth.secure.pixiv.net/auth/token",
|
||||
data=payload, headers=headers, timeout=20, verify=False)
|
||||
verify_mode = "CA验证失败,回退不验证"
|
||||
|
||||
if resp.status_code != 200:
|
||||
raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:120]}")
|
||||
j = resp.json()
|
||||
new_at = j.get("access_token")
|
||||
new_rt = j.get("refresh_token")
|
||||
if not new_at or not new_rt:
|
||||
raise RuntimeError(f"响应缺少 token 字段: keys={list(j.keys())}")
|
||||
|
||||
data["access_token"] = new_at
|
||||
data["refresh_token"] = new_rt
|
||||
tmp = TOKEN_PATH.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
os.chmod(tmp, 0o600)
|
||||
os.replace(tmp, TOKEN_PATH)
|
||||
log(f"刷新成功 ({verify_mode}) expires_in={j.get('expires_in')}s "
|
||||
f"access_token={new_at[:8]}... refresh_token={new_rt[:8]}... 已落盘")
|
||||
|
||||
|
||||
def main():
|
||||
log(f"守护启动, 刷新间隔 random {INTERVAL_RANGE[0]//60}~{INTERVAL_RANGE[1]//60} min")
|
||||
while True:
|
||||
try:
|
||||
do_refresh()
|
||||
time.sleep(random.uniform(*INTERVAL_RANGE))
|
||||
except Exception as e:
|
||||
log(f"刷新失败: {e}")
|
||||
traceback.print_exc()
|
||||
time.sleep(RETRY_INTERVAL)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user