freedakgmail
107 lines
4.7 KiB
Python
107 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Extract cross-article restaurant industry signals from cached Qwen analyses."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timedelta
|
||
|
||
from app import DB_PATH, ensure_schema, now_iso
|
||
|
||
|
||
def call_qwen(items: list[dict], model: str) -> list[dict]:
|
||
key = os.getenv("DASHSCOPE_API_KEY", "").strip()
|
||
if not key:
|
||
raise RuntimeError("尚未配置 DASHSCOPE_API_KEY")
|
||
prompt = f"""你是餐饮产业首席分析师。根据以下多篇文章的结构化分析，识别跨文章、可验证、有经营意义的行业信号。
|
||
只返回合法 JSON：{{"signals":[...]}}。每个 signal 必须包含：title（短标题）、summary（100-180字）、trend（emerging/accelerating/stable/declining）、confidence（0到1）、article_ids（至少2个证据文章ID；确实只有单篇强信号时可为1个）、implications（2-4条经营启示）、tags（3-6个标签）。
|
||
不要把单一品牌新闻简单改写成行业信号；合并重复主题；最多输出8条；没有充分证据就少输出。
|
||
输入：{json.dumps(items, ensure_ascii=False)}"""
|
||
body = json.dumps({
|
||
"model": model,
|
||
"messages": [
|
||
{"role": "system", "content": "你进行基于证据的餐饮行业趋势聚类，避免空泛结论。"},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
"temperature": 0.2,
|
||
"response_format": {"type": "json_object"},
|
||
}, ensure_ascii=False).encode("utf-8")
|
||
req = urllib.request.Request(
|
||
"https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
|
||
data=body,
|
||
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=120) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
raise RuntimeError(f"千问接口返回 {exc.code}：{exc.read().decode('utf-8', errors='replace')[:500]}") from exc
|
||
text = result["choices"][0]["message"]["content"].strip()
|
||
if text.startswith("```"):
|
||
text = text.strip("`").removeprefix("json").strip()
|
||
return json.loads(text).get("signals", [])
|
||
|
||
|
||
def run(days: int) -> int:
|
||
ensure_schema()
|
||
model = os.getenv("QWEN_MODEL", "qwen-plus")
|
||
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT a.id,a.title,a.published_at,x.result_json
|
||
FROM articles a JOIN ai_analyses x ON x.article_id=a.id
|
||
WHERE a.crawl_status='success' AND a.published_at>=?
|
||
ORDER BY a.published_at DESC
|
||
""",
|
||
(cutoff,),
|
||
).fetchall()
|
||
items = []
|
||
for row in rows:
|
||
analysis = json.loads(row["result_json"])
|
||
items.append({
|
||
"article_id": row["id"], "title": row["title"], "published_at": row["published_at"],
|
||
"summary": analysis.get("summary"), "key_points": analysis.get("key_points", []),
|
||
"industry_signal": analysis.get("industry_signal"), "tags": analysis.get("tags", []),
|
||
})
|
||
if not items:
|
||
print(json.dumps({"status": "skipped", "reason": "没有已分析文章"}, ensure_ascii=False))
|
||
return 0
|
||
signals = call_qwen(items, model)
|
||
signal_date = datetime.now().astimezone().date().isoformat()
|
||
timestamp = now_iso()
|
||
conn.execute("DELETE FROM industry_signals WHERE signal_date=?", (signal_date,))
|
||
for signal in signals:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO industry_signals
|
||
(signal_date,title,summary,trend,confidence,implications_json,article_ids_json,tags_json,model,created_at,updated_at)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?)
|
||
""",
|
||
(
|
||
signal_date, signal["title"], signal["summary"], signal.get("trend", "emerging"),
|
||
float(signal.get("confidence", 0.5)), json.dumps(signal.get("implications", []), ensure_ascii=False),
|
||
json.dumps(signal.get("article_ids", [])), json.dumps(signal.get("tags", []), ensure_ascii=False),
|
||
model, timestamp, timestamp,
|
||
),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
print(json.dumps({"status": "success", "input_articles": len(items), "signals": len(signals), "date": signal_date}, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description="提炼红餐文章跨文行业信号")
|
||
parser.add_argument("--days", type=int, default=30, help="聚合最近多少天的文章")
|
||
args = parser.parse_args()
|
||
raise SystemExit(run(args.days))
|
||
|