110 lines
4.0 KiB
Python
110 lines
4.0 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"))
|
|||
|
|
|
|||
|
|
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())}")
|
|||
|
|
|
|||
|
|
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())
|