125 lines
4.6 KiB
Python
125 lines
4.6 KiB
Python
"""
|
||
验证签到流程是否正常。
|
||
在服务器上执行: docker exec -it weibo-scheduler python -m test_fetch_topics
|
||
或本地: cd backend && python ../test_fetch_topics.py
|
||
|
||
会从数据库读取第一个 active 账号，解密 Cookie，模拟 _fetch_topics 流程。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import re
|
||
import asyncio
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "backend"))
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
|
||
|
||
# 兼容容器内运行（/app 目录）
|
||
if os.path.exists("/app/shared"):
|
||
sys.path.insert(0, "/app")
|
||
|
||
from shared.config import shared_settings
|
||
from shared.crypto import decrypt_cookie, derive_key
|
||
|
||
WEIBO_HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||
"Referer": "https://weibo.com/",
|
||
"Accept": "*/*",
|
||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
||
}
|
||
|
||
|
||
async def main():
|
||
import httpx
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
from shared.models.account import Account
|
||
|
||
engine = create_async_engine(shared_settings.DATABASE_URL, echo=False)
|
||
Session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
||
async with Session() as session:
|
||
result = await session.execute(
|
||
select(Account).where(Account.status.in_(["active", "pending"])).limit(1)
|
||
)
|
||
account = result.scalar_one_or_none()
|
||
if not account:
|
||
print("❌ 没有 active/pending 账号")
|
||
return
|
||
|
||
print(f"📱 账号: {account.remark or account.weibo_user_id} (status={account.status})")
|
||
|
||
key = derive_key(shared_settings.COOKIE_ENCRYPTION_KEY)
|
||
cookie_str = decrypt_cookie(account.encrypted_cookies, account.iv, key)
|
||
cookies = {}
|
||
for pair in cookie_str.split(";"):
|
||
pair = pair.strip()
|
||
if "=" in pair:
|
||
k, v = pair.split("=", 1)
|
||
cookies[k.strip()] = v.strip()
|
||
|
||
print(f"🍪 Cookie 数量: {len(cookies)}, keys: {list(cookies.keys())}")
|
||
|
||
# 检查 Cookie 有效期
|
||
from datetime import datetime
|
||
alf = cookies.get("ALF", "")
|
||
if alf and alf.isdigit():
|
||
expire_time = datetime.fromtimestamp(int(alf))
|
||
remain = (expire_time - datetime.now()).days
|
||
print(f"📅 Cookie 过期时间: {expire_time.strftime('%Y-%m-%d %H:%M:%S')} (还剩 {remain} 天)")
|
||
else:
|
||
print(f"📅 ALF 字段: {alf or '无'} (无法判断过期时间)")
|
||
|
||
await engine.dispose()
|
||
|
||
# 测试 1: 访问 weibo.com
|
||
print("\n--- 测试 1: GET https://weibo.com/ ---")
|
||
async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:
|
||
resp = await client.get("https://weibo.com/", headers=WEIBO_HEADERS, cookies=cookies)
|
||
final_url = str(resp.url)
|
||
print(f" 状态码: {resp.status_code}")
|
||
print(f" 最终URL: {final_url}")
|
||
print(f" 是否登录页: {'login.sina.com.cn' in final_url or 'passport' in final_url}")
|
||
|
||
if "login.sina.com.cn" in final_url or "passport" in final_url:
|
||
print("\n❌ Cookie 已失效，被重定向到登录页")
|
||
return
|
||
|
||
xsrf = client.cookies.get("XSRF-TOKEN", "")
|
||
print(f" XSRF-TOKEN: {'有' if xsrf else '无'}")
|
||
|
||
# 测试 2: 获取超话列表
|
||
print("\n--- 测试 2: 获取超话列表 ---")
|
||
headers = {**WEIBO_HEADERS, "X-Requested-With": "XMLHttpRequest"}
|
||
if xsrf:
|
||
headers["X-XSRF-TOKEN"] = xsrf
|
||
|
||
resp = await client.get(
|
||
"https://weibo.com/ajax/profile/topicContent",
|
||
params={"tabid": "231093_-_chaohua", "page": "1"},
|
||
headers=headers, cookies=cookies,
|
||
)
|
||
print(f" 状态码: {resp.status_code}")
|
||
print(f" 最终URL: {resp.url}")
|
||
|
||
try:
|
||
data = resp.json()
|
||
print(f" ok: {data.get('ok')}")
|
||
topics = data.get("data", {}).get("list", [])
|
||
print(f" 超话数量: {len(topics)}")
|
||
for t in topics[:5]:
|
||
title = t.get("topic_name", "") or t.get("title", "")
|
||
print(f" - {title}")
|
||
if topics:
|
||
print("\n✅ Cookie 有效，超话获取正常")
|
||
else:
|
||
print("\n⚠️ Cookie 可能有效但没有关注超话")
|
||
except Exception as e:
|
||
print(f" ❌ 响应非 JSON: {resp.text[:300]}")
|
||
print(f"\n❌ 获取超话失败: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|