test_s2_wechat_viewer.py
3.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
#!/usr/bin/env python3
"""S2: Check wechat-viewer data source health. (FIXED: handle 404 gracefully)"""
import json, subprocess, sys, sqlite3
passed = 0; total = 0
def check(n, c, d=""):
global passed, total; total += 1
print(f" {'✅' if c else '❌'} {n}{' - '+d if d else ''}")
if c: passed += 1
print("=== S2: WeChat Viewer Data Source ===\n")
# 1. Articles API
r = subprocess.run(["curl", "-s", "http://127.0.0.1:5000/api/articles?per_page=1"],
capture_output=True, text=True, timeout=10)
d = json.loads(r.stdout)
check("Articles API", d.get("total", 0) > 100, f"total={d.get('total')}")
# 2. Recent articles (24h)
r = subprocess.run(["curl", "-s",
"http://127.0.0.1:5000/api/articles?per_page=5&date_from=2026-07-10"],
capture_output=True, text=True, timeout=10)
d = json.loads(r.stdout)
arts24h = d.get("data", [])
check("Recent articles (24h)", len(arts24h) > 0, f"{len(arts24h)} in last 24h")
# 3. By-group query (TGO) — check if endpoint exists
r = subprocess.run(["curl", "-s", "-w", "|HTTP=%{http_code}",
"http://127.0.0.1:5000/api/articles/by-group?chatroom_id=54655425669@chatroom&per_page=5"],
capture_output=True, text=True, timeout=10)
http_code = r.stdout.split("|HTTP=")[-1].strip() if "|HTTP=" in r.stdout else "?"
body = r.stdout.split("|HTTP=")[0]
has_by_group = http_code == "200"
check("By-group endpoint", has_by_group, f"HTTP {http_code}")
if not has_by_group:
print(" ⚠️ /api/articles/by-group not implemented — daily digest may be affected")
# 4. Summarize API
r = subprocess.run(["curl", "-s", "-X", "POST",
"http://127.0.0.1:5000/api/articles/summarize",
"-H", "Content-Type: application/json",
"-d", '{"url":"https://mp.weixin.qq.com/s/test123","title":"test"}'],
capture_output=True, text=True, timeout=15)
d = json.loads(r.stdout)
check("Summarize API", d.get("ok") == True or "already" in str(d).lower(),
f"ok={d.get('ok')}")
# 5. Daily reports
DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
conn = sqlite3.connect(DB)
cnt = conn.execute("SELECT COUNT(*) FROM daily_reports").fetchone()[0]
recent = conn.execute(
"SELECT COUNT(*) FROM daily_reports WHERE created_at > datetime('now', '-1 day')"
).fetchone()[0]
# Check summary content quality
sample = conn.execute("SELECT article_url, content FROM daily_reports WHERE report_date='summary' LIMIT 1").fetchone()
has_content = sample and sample[1] and len(sample[1]) > 100
conn.close()
check("Daily reports total", cnt > 100, f"{cnt}")
check("Daily reports (24h)", recent > 0, f"{recent}")
check("Summary content quality", has_content, f"sample len={len(sample[1]) if sample else 0}")
# 6. Check-summaries API
r = subprocess.run(["curl", "-s", "-X", "POST",
"http://127.0.0.1:5000/api/articles/check-summaries",
"-H", "Content-Type: application/json",
"-d", json.dumps({"urls": [{"url": a.get("url",""), "title": a.get("title","")} for a in arts24h[:2]], "type": "summary"})],
capture_output=True, text=True, timeout=10)
d = json.loads(r.stdout)
check("Check-summaries API", d.get("ok") == True,
f"results={len(d.get('data',{}).get('results',{}))}, missing={len(d.get('data',{}).get('missing',[]))}")
# 7. Unsummarized API
r = subprocess.run(["curl", "-s", "http://127.0.0.1:5000/api/articles/unsummarized?limit=5"],
capture_output=True, text=True, timeout=10)
d = json.loads(r.stdout)
unsum = d.get("data", [])
check("Unsummarized API", len(unsum) >= 0, f"{len(unsum)} pending")
# 8. Hermes jobs completion
conn = sqlite3.connect(DB)
jobs = conn.execute(
"SELECT status, COUNT(*) FROM hermes_jobs GROUP BY status"
).fetchall()
conn.close()
job_stats = dict(jobs)
check("Hermes jobs summary", job_stats.get("completed", 0) > 50,
f"completed={job_stats.get('completed',0)}, failed={job_stats.get('failed',0)}")
print(f"\nResult: {passed}/{total} passed")
sys.exit(0 if passed == total else 1)