freedakgmail
84 lines
3.5 KiB
Python
84 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Batch Qwen analysis for all successfully crawled articles."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from app import DB_PATH, ensure_schema, now_iso, qwen_analyze
|
|
|
|
|
|
def run(force: bool, pause: float, retries: int) -> int:
|
|
ensure_schema()
|
|
model = os.getenv("QWEN_MODEL", "qwen-plus")
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT a.id,a.title,a.content,a.content_hash,x.content_hash analyzed_hash
|
|
FROM articles a
|
|
LEFT JOIN ai_analyses x
|
|
ON x.article_id=a.id AND x.model=? AND x.analysis_type='editorial'
|
|
WHERE a.crawl_status='success'
|
|
ORDER BY a.published_at DESC,a.id DESC
|
|
""",
|
|
(model,),
|
|
).fetchall()
|
|
pending = [r for r in rows if force or not r["analyzed_hash"] or r["analyzed_hash"] != r["content_hash"]]
|
|
print(json.dumps({"total": len(rows), "pending": len(pending), "model": model}, ensure_ascii=False), flush=True)
|
|
success = skipped = failed = 0
|
|
for index, row in enumerate(rows, start=1):
|
|
if not force and row["analyzed_hash"] == row["content_hash"]:
|
|
skipped += 1
|
|
continue
|
|
last_error = None
|
|
for attempt in range(1, retries + 2):
|
|
try:
|
|
result = qwen_analyze(row["title"], row["content"])
|
|
timestamp = now_iso()
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO ai_analyses(article_id,model,analysis_type,result_json,content_hash,created_at,updated_at)
|
|
VALUES (?,?, 'editorial', ?,?,?,?)
|
|
ON CONFLICT(article_id,model,analysis_type) DO UPDATE SET
|
|
result_json=excluded.result_json,content_hash=excluded.content_hash,updated_at=excluded.updated_at
|
|
""",
|
|
(row["id"], model, json.dumps(result, ensure_ascii=False), row["content_hash"], timestamp, timestamp),
|
|
)
|
|
conn.commit()
|
|
success += 1
|
|
print(f"[{index}/{len(rows)}] OK {row['title']}", flush=True)
|
|
last_error = None
|
|
break
|
|
except Exception as exc:
|
|
last_error = exc
|
|
if attempt <= retries:
|
|
wait = attempt * 2
|
|
print(f"[{index}/{len(rows)}] RETRY {attempt}/{retries} {exc}", flush=True)
|
|
time.sleep(wait)
|
|
if last_error is not None:
|
|
failed += 1
|
|
print(f"[{index}/{len(rows)}] FAIL {row['title']}: {last_error}", flush=True)
|
|
time.sleep(pause)
|
|
total_analyzed = conn.execute("SELECT COUNT(DISTINCT article_id) FROM ai_analyses").fetchone()[0]
|
|
conn.close()
|
|
print(json.dumps({
|
|
"success": success, "skipped": skipped, "failed": failed,
|
|
"total_analyzed": total_analyzed,
|
|
}, ensure_ascii=False), flush=True)
|
|
return 0 if failed == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="批量执行红餐文章千问分析")
|
|
parser.add_argument("--force", action="store_true", help="忽略缓存并重新分析")
|
|
parser.add_argument("--pause", type=float, default=0.35, help="请求间隔秒数")
|
|
parser.add_argument("--retries", type=int, default=2, help="单篇失败重试次数")
|
|
args = parser.parse_args()
|
|
raise SystemExit(run(args.force, args.pause, args.retries))
|