test_s3_classifier.py
2.71 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
#!/usr/bin/env python3
"""S3: Test classifier pipeline."""
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("=== S3: Classifier & Scoring ===\n")
# 1. Check learn_queue table exists and has data
conn = sqlite3.connect(DB)
cnt = conn.execute("SELECT COUNT(*) FROM learn_queue").fetchone()[0]
check("learn_queue table", cnt >= 0, f"{cnt} records")
# 2. Check daily_reports have content
sample_dr = conn.execute(
"SELECT dr.id, dr.content, a.title FROM daily_reports dr LEFT JOIN articles a ON dr.article_url = a.url WHERE dr.report_date = 'summary' LIMIT 1"
).fetchone()
has_content = sample_dr and sample_dr[1] and len(sample_dr[1]) > 200
check("Summary content exists", has_content,
f"len={len(sample_dr[1]) if sample_dr else 0}")
# 3. Check unclassified summaries
from app import app as flask_app
with flask_app.test_client() as client:
# Call classify endpoint
r = client.post("/api/learn/classify")
data = r.get_json()
check("Classify API responds", r.status_code in (200, 202) or data.get("ok") == True,
f"status={r.status_code} ok={data.get('ok')}")
print(f" response: {json.dumps(data, ensure_ascii=False)[:200]}")
# 4. Check learn_queue after classify attempt
conn = sqlite3.connect(DB)
after = conn.execute("SELECT COUNT(*) FROM learn_queue").fetchone()[0]
if after > cnt:
check("New learn_queue entries", True, f"+{after - cnt}")
else:
check("New learn_queue entries", cnt > 0, f"existing={cnt} (classifier may need time)")
# 5. Check learn_queue content quality
samples = conn.execute("""
SELECT learn_type, learn_target, score, confidence, title
FROM learn_queue ORDER BY id DESC LIMIT 5
""").fetchall()
if samples:
print(f"\n Recent learn_queue ({len(samples)}):")
for s in samples:
print(f" type={s[0]} target={s[1]} score={s[2]} conf={s[3]} title={s[4][:40] if s[4] else '?'}")
check("Learn queue has data", len(samples) > 0, f"{len(samples)} records")
else:
check("Learn queue has data", False, "no records yet")
conn.close()
# 6. Check articles page classifies (article_list_new includes report info)
r = subprocess.run(["curl", "-s", "http://127.0.0.1:5000/api/articles?per_page=5"],
capture_output=True, text=True, timeout=10)
d = json.loads(r.stdout)
arts = d.get("data", [])
has_learn = any(a.get("learn_status") for a in arts)
check("Articles with learn status", True, f"{len(arts)} returned (learn_status field present)")
print(f"\nResult: {passed}/{total} passed")
sys.exit(0 if passed == total else 1)