test_s4_e2e.py
3.77 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
92
93
#!/usr/bin/env python3
"""S4: End-to-end pipeline verification."""
import json, subprocess, sys, sqlite3
DB = "/data/data/com.termux/files/home/wechat-viewer/circle.db"
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("=== S4: End-to-End Verification ===\n")
# 1. Find a scored article in learn_queue
conn = sqlite3.connect(DB)
rows = conn.execute("""
SELECT lq.id, lq.article_url, lq.title, lq.learn_type, lq.learn_target,
lq.score, lq.confidence, lq.status,
a.url as article_url_full, dr.content as summary
FROM learn_queue lq
LEFT JOIN articles a ON lq.dedup_key = a.dedup_key
LEFT JOIN daily_reports dr ON dr.article_url = lq.article_url AND dr.report_date = 'summary'
WHERE lq.score >= 6 AND lq.status = 'pending'
ORDER BY lq.score DESC LIMIT 1
""").fetchone()
if not rows:
# Try without status filter
rows = conn.execute("""
SELECT lq.id, lq.article_url, lq.title, lq.learn_type, lq.learn_target,
lq.score, lq.confidence, lq.status
FROM learn_queue lq
WHERE lq.score >= 6
ORDER BY lq.score DESC LIMIT 1
""").fetchone()
if rows:
lid = rows[0]
ltype = rows[3]
ltarget = rows[4]
score = rows[5]
status = rows[7] if len(rows) > 7 else "?"
title = (rows[2] or "?")[:50]
check("Find classified article", True, f"id={lid} type={ltype} score={score} status={status} title={title}")
print(f" target={ltarget}")
# 2. Check summary exists for this article
summary_row = conn.execute(
"SELECT content FROM daily_reports WHERE article_url = ? AND report_date = 'summary' LIMIT 1",
(rows[1],)
).fetchone()
has_summary = summary_row and summary_row[0] and len(summary_row[0]) > 200
check("Summary available", has_summary,
f"len={len(summary_row[0]) if summary_row else 0}")
# 3. Check dedup logic - is this already in memory/skill?
if ltype == "skill":
has_dedup = conn.execute(
"SELECT COUNT(*) FROM learn_queue WHERE dedup_key = (SELECT dedup_key FROM learn_queue WHERE id=?) AND status IN ('completed','processing')", (lid,)
).fetchone()[0]
check("Dedup check", has_dedup == 0, f"duplicates={has_dedup}")
# 4. Check learn-and-build skill exists
from pathlib import Path
lb_skill = Path("/data/data/com.termux/files/home/.hermes/skills/wechat/learn-and-build/SKILL.md")
check("learn-and-build skill exists", lb_skill.exists())
# 5. Verify cron pipeline can self-heal
scan_path = Path("/data/data/com.termux/files/home/.hermes/skills/wechat-group-digest/scripts/scan_articles.py")
check("scan_articles.py exists", scan_path.exists())
# 6. Overall pipeline check
conn2 = sqlite3.connect(DB)
total_arts = conn2.execute("SELECT COUNT(*) FROM articles").fetchone()[0]
total_reports = conn2.execute("SELECT COUNT(*) FROM daily_reports WHERE report_date='summary'").fetchone()[0]
total_learn = conn2.execute("SELECT COUNT(*) FROM learn_queue WHERE score >= 6").fetchone()[0]
total_pending_classify = conn2.execute("""
SELECT COUNT(*) FROM learn_queue WHERE status = 'pending' AND score >= 6
""").fetchone()[0]
conn2.close()
check("Pipeline: articles → summaries", total_reports > 0,
f"{total_arts} arts → {total_reports} summaries")
check("Pipeline: summaries → classified", total_learn > 0,
f"{total_reports} summaries → {total_learn} classified (≥6)")
check("Pipeline: pending review", total_pending_classify >= 0,
f"{total_pending_classify} pending")
conn.close()
print(f"\nResult: {passed}/{total} passed")
sys.exit(0 if passed == total else 1)