test_s1_cron_pipeline.py
2.68 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
#!/usr/bin/env python3
"""S1 diagnostic test: verify cron pipeline health. (FIXED: use per_page not limit)"""
import json, subprocess, sys, os
passed = 0
total = 0
def check(name, condition, detail=""):
global passed, total
total += 1
if condition:
print(f" ✅ {name}{' - ' + detail if detail else ''}")
passed += 1
else:
print(f" ❌ {name}{' - ' + detail if detail else ''}")
print("=== S1: Cron Pipeline Health Check (FIXED) ===\n")
# 1. Flask HTTP status
r = subprocess.run(["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
"http://127.0.0.1:5000/"],
capture_output=True, text=True, timeout=10)
check("Flask :5000", r.stdout == "200", f"HTTP {r.stdout}")
# 2. Articles API (FIXED: per_page not limit)
r = subprocess.run(["curl", "-s",
"http://127.0.0.1:5000/api/articles?per_page=3"],
capture_output=True, text=True, timeout=15)
data = json.loads(r.stdout)
total_articles = data.get("total", 0)
arts = data.get("data", [])
check("Articles API", len(arts) > 0, f"total={total_articles}, returned={len(arts)}")
# 3. Unsummarized articles (FIXED: different endpoint)
r = subprocess.run(["curl", "-s", "http://127.0.0.1:5000/api/articles/unsummarized?limit=5"],
capture_output=True, text=True, timeout=15)
data = json.loads(r.stdout)
unsum = data.get("data", data.get("articles", []))
check("Unsummarized API", isinstance(unsum, list), f"{len(unsum)} pending")
# 4. Daily reports (direct DB check)
DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
import sqlite3
conn = sqlite3.connect(DB)
cnt = conn.execute("SELECT COUNT(*) FROM daily_reports").fetchone()[0]
conn.close()
check("Daily reports table", cnt > 100, f"{cnt} records")
# 5. scan_articles.py is runnable
scan_path = os.path.expanduser("~/.dsh/skills/wechat-group-digest/scripts/scan_articles.py")
check("scan_articles.py exists", os.path.exists(scan_path))
# 6. Cron job active
r = subprocess.run(["hermes", "cron", "list"], capture_output=True, text=True, timeout=10)
has_tgo = "ae77adaeefa9" in r.stdout and "active" in r.stdout
check("TGO cron job active", has_tgo)
# 7. Articles in DB match API
conn = sqlite3.connect(DB)
db_count = conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
conn.close()
check("DB articles count", db_count > 100, f"{db_count} articles in articles table")
# 8. article_shares join works
conn = sqlite3.connect(DB)
share_count = conn.execute("SELECT COUNT(*) FROM article_shares").fetchone()[0]
conn.close()
check("article_shares table", share_count > 500, f"{share_count} shares")
print(f"\nResult: {passed}/{total} passed")
sys.exit(0 if passed == total else 1)