114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import random
|
|
import sys
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
from bilibili_api import Credential, sync
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
COOKIE_FILE = Path("cookies.txt")
|
|
|
|
def load_credential() -> Credential:
|
|
"""从 KEY=VALUE 格式的 txt 文件加载 Credential 对象"""
|
|
if not COOKIE_FILE.exists():
|
|
logger.error("Cookie file not found: %s", COOKIE_FILE)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
data = {}
|
|
with open(COOKIE_FILE, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if "=" in line:
|
|
key, _, value = line.partition("=")
|
|
data[key.strip()] = value.strip()
|
|
# 构建 Credential 对象（key 名大小写不敏感）
|
|
cred = Credential(
|
|
sessdata=data.get("SESSDATA") or data.get("sessdata"),
|
|
bili_jct=data.get("bili_jct") or data.get("BILI_JCT"),
|
|
buvid3=data.get("buvid3") or data.get("BUVID3"),
|
|
dedeuserid=data.get("dedeuserid") or data.get("DEDEUSERID"),
|
|
ac_time_value=data.get("ac_time_value") or data.get("AC_TIME_VALUE"),
|
|
)
|
|
logger.info("Credential loaded from %s", COOKIE_FILE)
|
|
return cred
|
|
except Exception as e:
|
|
logger.error("Failed to load credential: %s", e)
|
|
sys.exit(1)
|
|
|
|
async def check_and_refresh(cred: Credential) -> bool:
|
|
"""检查有效性，如果有效则刷新 token。返回是否成功刷新（或仍有效）"""
|
|
try:
|
|
valid = await cred.check_valid()
|
|
if not valid:
|
|
logger.warning("Credential is invalid (maybe expired). Refresh may fail.")
|
|
else:
|
|
logger.info("Credential is valid, proceeding to refresh...")
|
|
|
|
# 无论 valid 状态如何，都尝试 refresh（refresh 接口本身会处理）
|
|
await cred.refresh()
|
|
logger.info("Token refresh successful. New SESSDATA: %s...", cred.sessdata[:20])
|
|
# 保存更新后的凭证到文件
|
|
save_credential(cred)
|
|
return True
|
|
except Exception as e:
|
|
logger.error("Refresh failed: %s", e)
|
|
return False
|
|
|
|
def save_credential(cred: Credential):
|
|
"""将 credential 的关键字段保存回 txt 文件（KEY=VALUE 格式）"""
|
|
lines = []
|
|
if cred.sessdata:
|
|
lines.append(f"SESSDATA={cred.sessdata}")
|
|
if cred.bili_jct:
|
|
lines.append(f"bili_jct={cred.bili_jct}")
|
|
if cred.buvid3:
|
|
lines.append(f"buvid3={cred.buvid3}")
|
|
if cred.dedeuserid:
|
|
lines.append(f"dedeuserid={cred.dedeuserid}")
|
|
if cred.ac_time_value:
|
|
lines.append(f"ac_time_value={cred.ac_time_value}")
|
|
|
|
with open(COOKIE_FILE, "w") as f:
|
|
f.write("\n".join(lines) + "\n")
|
|
logger.info("Credential saved to %s", COOKIE_FILE)
|
|
|
|
async def main_loop():
|
|
logger.info("Starting token refresh daemon...")
|
|
cred = load_credential()
|
|
|
|
while True:
|
|
# 随机间隔 50~70 分钟（单位：秒）
|
|
interval = random.randint(50 * 60, 70 * 60)
|
|
next_refresh = datetime.now().timestamp() + interval
|
|
logger.info("Next refresh in %d minutes (at %s)", interval // 60, datetime.fromtimestamp(next_refresh).strftime("%Y-%m-%d %H:%M:%S"))
|
|
|
|
await asyncio.sleep(interval)
|
|
|
|
logger.info("Refreshing token...")
|
|
success = await check_and_refresh(cred)
|
|
if not success:
|
|
logger.error("Refresh failed, will retry in next cycle.")
|
|
else:
|
|
logger.info("Refresh cycle completed.")
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main_loop())
|
|
except KeyboardInterrupt:
|
|
logger.info("Daemon stopped by user.")
|
|
except Exception as e:
|
|
logger.exception("Unexpected error: %s", e)
|
|
sys.exit(1)
|