test_articles.py
4.05 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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
#!/usr/bin/env python3
"""自动测试文章列表：链接可用性 → 摘要 → 学习"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import json, time, urllib.request, urllib.error
BASE = "http://127.0.0.1:5000"
def test_url(title, url):
"""测试链接是否可访问"""
try:
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)'
})
with urllib.request.urlopen(req, timeout=15) as resp:
return resp.status, len(resp.read())
except urllib.error.HTTPError as e:
return e.code, 0
except Exception as e:
return str(e), 0
def api_post(path, data):
"""调 Flask API"""
req = urllib.request.Request(f"{BASE}{path}",
data=json.dumps(data).encode(),
headers={'Content-Type': 'application/json'},
method='POST')
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
def api_get(path):
with urllib.request.urlopen(f"{BASE}{path}", timeout=10) as resp:
return json.loads(resp.read())
# ── Step 1: 获取文章列表 ──
print("=" * 60)
print("Step 1: 获取文章列表")
print("=" * 60)
articles = api_get("/api/articles")
data = articles.get("data", [])
print(f"共 {len(data)} 篇文章\n")
# ── Step 2: 测试链接可用性 ──
print("=" * 60)
print("Step 2: 测试链接（前10篇）")
print("=" * 60)
bad = []
for i, a in enumerate(data[:10]):
status, size = test_url(a['title'], a['url'])
status_str = f"HTTP {status}" if isinstance(status, int) else f"ERR: {status}"
ok = isinstance(status, int) and 200 <= status < 400
mark = "✅" if ok else "❌"
print(f" {mark} [{i+1}] {a['title'][:50]}...")
print(f" {status_str} | {size}B")
if not ok:
bad.append(a)
time.sleep(0.5)
if bad:
print(f"\n⚠️ {len(bad)} 篇链接有问题")
else:
print(f"\n✅ 全部链接正常")
# ── Step 3: 测试摘要（前3篇） ──
print("\n" + "=" * 60)
print("Step 3: 测试摘要（前3篇）")
print("=" * 60)
for i, a in enumerate(data[:3]):
print(f"\n [{i+1}] 提交摘要: {a['title'][:50]}...")
r = api_post("/api/articles/summarize", {"title": a['title'], "url": a['url']})
print(f" API: OK={r['ok']}, status={r['data']['status']}")
time.sleep(2)
print("\n⏳ 等待 Hermes 子进程完成（最多 120s）...")
for attempt in range(24):
time.sleep(5)
reports = api_get("/api/reports")
summary_reports = [r for r in reports['data'] if r.get('report_date') == 'summary']
if len(summary_reports) >= 3:
print(f"\n✅ 3 篇摘要已生成！")
for sr in summary_reports[:3]:
print(f" #{sr['id']} {sr['chatroom_name'][:40]}: {len(sr.get('summary','') or '')} 字")
break
if attempt % 4 == 3:
print(f" 等待中... ({len(summary_reports)}/3)")
# ── Step 4: 测试学习（前2篇） ──
print("\n" + "=" * 60)
print("Step 4: 测试学习（前2篇）")
print("=" * 60)
for i, a in enumerate(data[:2]):
print(f"\n [{i+1}] 提交学习: {a['title'][:50]}...")
r = api_post("/api/articles/learn", {"title": a['title'], "url": a['url']})
print(f" API: OK={r['ok']}, status={r['data']['status']}")
time.sleep(2)
print("\n⏳ 等待 Hermes 子进程完成（最多 300s）...")
for attempt in range(60):
time.sleep(5)
reports = api_get("/api/reports")
learn_reports = [r for r in reports['data'] if r.get('report_date') == 'learn']
if len(learn_reports) >= 2:
print(f"\n✅ 2 篇学习结果已生成！")
for lr in learn_reports[:2]:
print(f" #{lr['id']} {lr['chatroom_name'][:40]}: {len(lr.get('summary','') or '')} 字")
break
if attempt % 6 == 5:
print(f" 等待中... ({len(learn_reports)}/2)")
print("\n" + "=" * 60)
print("测试完成！")
print("=" * 60)
print(f"链接: {len(data) - len(bad)}/{len(data[:10])} 正常")
print(f"摘要: 已提交 3 篇")
print(f"学习: 已提交 2 篇")
print(f"查看结果: {BASE}/reports")