393 lines
16 KiB
Python
393 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
新闻智能跟踪系统 - 智能分析引擎
|
|
两级分析：
|
|
1. 规则打分（快、无需外部依赖）：兴趣相关度 + 重要度启发式 → total_score
|
|
2. LLM 深度分析（准、逐条）：对候选重要资讯调用 DeepSeek 输出重要度/分类/结论
|
|
"""
|
|
import json
|
|
import threading
|
|
|
|
import requests
|
|
|
|
import config
|
|
import db
|
|
|
|
# 重要度启发式信号词（命中越多分越高）
|
|
_STRONG_SIGNAL = [
|
|
"发布", "宣布", "推出", "开源", "突破", "融资", "IPO", "上市", "收购", "并购",
|
|
"合并", "裁员", "禁令", "制裁", "管制", "监管", "起诉", "反垄断", "罚款",
|
|
"重大", "首个", "最强", "新规", "停止", "暂停", "安全事件", "数据泄露",
|
|
"里程碑", "量产", "实施细则", "生效", "世界第一",
|
|
]
|
|
_MID_SIGNAL = [
|
|
"升级", "更新", "合作", "投资", "签署", "获批", "中标", "测试", "开放", "公测",
|
|
"翻倍", "增长", "新高", "新纪录", "接入", "部署", "扩大", "提升",
|
|
]
|
|
_WEAK_SIGNAL = ["讨论", "传闻", "预计", "可能", "或将", "消息人士", "知情人士"]
|
|
|
|
# 大额资金/规模信号：出现强金额词且伴随大数字 → 重要度加成
|
|
_MONEY_WORDS = ["亿美元", "亿元", "万亿", "估值", "融资", "收购", "IPO", "罚款", "大单", "募资", "基金"]
|
|
_MONEY_BIG = ["10亿", "20亿", "50亿", "百亿", "千亿", "万亿", "估值突破", "估值超", "亿美元",
|
|
"10 亿", "20 亿", "50 亿", "规模突破", "市值"]
|
|
|
|
_DOMAIN_RULES = {
|
|
"AI模型与算法": ["模型", "GPT", "DeepSeek", "大模型", "推理", "多模态", "智能体", "Agent",
|
|
"Scaling", "思维链", "视频生成", "扩散", "检索增强", "端侧模型", "开源模型"],
|
|
"芯片与硬件": ["芯片", "GPU", "半导体", "晶圆", "制程", "数据中心", "服务器", "AI芯片", "光刻机"],
|
|
"云计算与算力": ["云计算", "算力", "数据中心", "集群", "云厂商", "推理集群", "超算"],
|
|
"政策与监管": ["监管", "政策", "法案", "法规", "管制", "禁令", "制裁", "新规", "合规", "备案", "诉讼", "审查", "治理框架", "罚款"],
|
|
"投融资": ["融资", "IPO", "投资", "收购", "并购", "估值", "募资", "上市", "风投", "D轮", "大单"],
|
|
"企业动态": ["重组", "架构", "高管", "任命", "裁员", "人事", "部门", "收购"],
|
|
"学术研究": ["论文", "arXiv", "研究", "实验", "学者", "基准", "团队发布"],
|
|
"开源生态": ["开源", "GitHub", "权重", "社区", "代码", "star", "周榜"],
|
|
}
|
|
|
|
_IMPORTANCE_SCALE = {"strong": 32, "mid": 16, "weak": 6}
|
|
|
|
|
|
def _text_of(a):
|
|
return (a.get("title", "") + " " + a.get("content", "") + " " + a.get("summary", ""))
|
|
|
|
|
|
def classify_domain(a):
|
|
text = _text_of(a)
|
|
best, best_hits = "", 0
|
|
for dom, kws in _DOMAIN_RULES.items():
|
|
hits = sum(1 for kw in kws if kw.lower() in text.lower())
|
|
if hits > best_hits:
|
|
best, best_hits = dom, hits
|
|
return best
|
|
|
|
|
|
def extract_entities(a):
|
|
"""返回 (全部实体, 真实关注公司)"""
|
|
text = _text_of(a)
|
|
found, real = [], []
|
|
for c in db.list_companies():
|
|
name = c["name"]
|
|
if name and name.lower() in text.lower():
|
|
if name not in found:
|
|
found.append(name)
|
|
if name not in real:
|
|
real.append(name)
|
|
for e in a.get("entities") or []:
|
|
if e not in found:
|
|
found.append(e)
|
|
return found, real
|
|
|
|
|
|
def _keyword_hits(text):
|
|
hits, score = [], 0
|
|
for kw in db.list_keywords():
|
|
if not kw["enabled"]:
|
|
continue
|
|
if kw["keyword"].lower() in text.lower():
|
|
hits.append(kw["keyword"])
|
|
score += kw["weight"]
|
|
return hits, min(50, score)
|
|
|
|
|
|
def _domain_score(domain):
|
|
for dom in db.list_domains():
|
|
if dom["enabled"] and dom["name"] == domain:
|
|
return min(30, dom["weight"] * 4)
|
|
return 0
|
|
|
|
|
|
def _company_match(real_companies):
|
|
"""返回 (相关度加分, 重要度impact)"""
|
|
n = len(real_companies)
|
|
if n == 0:
|
|
return 0, 0
|
|
return min(20, 10 + (n - 1) * 3), min(14, 8 + (n - 1) * 4)
|
|
|
|
|
|
def _signal_score(text):
|
|
strong = [w for w in _STRONG_SIGNAL if w in text]
|
|
if strong:
|
|
return min(38, _IMPORTANCE_SCALE["strong"] + 6 * (len(strong) - 1))
|
|
mid = [w for w in _MID_SIGNAL if w in text]
|
|
if mid:
|
|
return min(22, _IMPORTANCE_SCALE["mid"] + 4 * (len(mid) - 1))
|
|
return _IMPORTANCE_SCALE["weak"] if any(w in text for w in _WEAK_SIGNAL) else 0
|
|
|
|
|
|
def _money_magnitude(text):
|
|
if any(k in text for k in _MONEY_WORDS):
|
|
if any(m in text for m in _MONEY_BIG):
|
|
return 12
|
|
return 6
|
|
return 0
|
|
|
|
|
|
def _recency_bonus(a):
|
|
try:
|
|
from datetime import datetime
|
|
pub = datetime.strptime(a.get("published_at", ""), "%Y-%m-%d %H:%M:%S")
|
|
hours = (datetime.now() - pub).total_seconds() / 3600
|
|
except Exception:
|
|
return 5
|
|
if hours <= 6:
|
|
return 12
|
|
if hours <= 24:
|
|
return 8
|
|
if hours <= 48:
|
|
return 4
|
|
return 0
|
|
|
|
|
|
def _source_weight(a):
|
|
sid = a.get("source_id") or 0
|
|
s = db.get_source(sid)
|
|
if not s:
|
|
return 0
|
|
if s.get("kind") == "custom":
|
|
# 定制监控无权重，是否推送由大模型按「推送标准」判断
|
|
return 0
|
|
return int(round(s["weight"] * 10))
|
|
|
|
|
|
def source_is_custom(a):
|
|
"""该资讯所属数据源是否为定制监控类型"""
|
|
sid = a.get("source_id") or 0
|
|
if not sid:
|
|
return False
|
|
s = db.get_source(sid)
|
|
return bool(s and s.get("kind") == "custom")
|
|
|
|
|
|
def analyze_article(aid):
|
|
"""规则打分（立即生效）"""
|
|
a = db.get_article(aid)
|
|
if not a:
|
|
return None
|
|
text = _text_of(a)
|
|
domain = classify_domain(a)
|
|
entities, real = extract_entities(a)
|
|
|
|
_, kw_score = _keyword_hits(text)
|
|
dom_score = _domain_score(domain)
|
|
comp_score, comp_impact = _company_match(real)
|
|
rel = min(100, kw_score + dom_score + comp_score)
|
|
|
|
sig = _signal_score(text)
|
|
money = _money_magnitude(text)
|
|
rec = _recency_bonus(a)
|
|
src = _source_weight(a)
|
|
imp = min(100, sig + money + rec + src + comp_impact + int(rel * 0.2))
|
|
|
|
total = min(100, int(round(0.4 * rel + 0.6 * imp)))
|
|
|
|
# 定制监控：规则分仅供展示，是否推送完全由大模型按「推送标准」判断（初始标记为不推送，等 LLM 结论）
|
|
if source_is_custom(a):
|
|
db.update_article(aid, domain=domain, entities=entities, relevance=rel,
|
|
total_score=total, is_important=0)
|
|
return {"id": aid, "domain": domain, "entities": entities, "relevance": rel,
|
|
"importance_rule": imp, "total_score": total, "is_important": 0,
|
|
"custom": True}
|
|
|
|
threshold = int(db.get_setting("realtime_threshold", config.AUTO_DEFAULTS["realtime_threshold"]))
|
|
is_important = 1 if (total >= threshold or (rel >= 65 and imp >= 70)) else 0
|
|
|
|
db.update_article(
|
|
aid, domain=domain, entities=entities, relevance=rel, total_score=total,
|
|
is_important=is_important,
|
|
)
|
|
return {"id": aid, "domain": domain, "entities": entities, "relevance": rel,
|
|
"importance_rule": imp, "total_score": total, "is_important": is_important}
|
|
|
|
|
|
def get_llm_cfg():
|
|
"""当前激活的大模型接口（网页可一键切换）；无则回退 config 默认"""
|
|
p = db.get_active_provider()
|
|
if p and p.get("base_url"):
|
|
return {"name": p["name"], "base_url": p["base_url"].rstrip("/"),
|
|
"api_key": p.get("api_key", ""), "model": p.get("model", "")}
|
|
return {"name": "DeepSeek 官方", "base_url": config.LLM_BASE_URL,
|
|
"api_key": config.LLM_API_KEY, "model": config.LLM_MODEL}
|
|
|
|
|
|
def _provider_chain():
|
|
"""按优先级排序的大模型接口调用链（priority 小优先，其次 id）。
|
|
全部由 llm_providers 表启用项决定；无任何启用接口时回退 config 默认。"""
|
|
chain = []
|
|
for p in db.list_providers_ordered():
|
|
chain.append({"id": p["id"], "name": p["name"],
|
|
"base_url": (p["base_url"] or "").rstrip("/"),
|
|
"api_key": p.get("api_key", ""), "model": p.get("model", "")})
|
|
if not chain:
|
|
chain.append({"id": 0, "name": "DeepSeek 官方", "base_url": config.LLM_BASE_URL,
|
|
"api_key": config.LLM_API_KEY, "model": config.LLM_MODEL})
|
|
return chain
|
|
|
|
|
|
def _llm_chat(prompt):
|
|
"""按优先级顺序调用大模型：高优先级失败 → 自动切换下一个；全部失败才抛异常。
|
|
每次调用记录统计（次数 / token / 失败）。返回 (content, provider_name)。"""
|
|
chain = _provider_chain()
|
|
last_err = ""
|
|
for cfg in chain:
|
|
if not cfg.get("base_url"):
|
|
continue
|
|
pname = cfg.get("name", cfg["base_url"])
|
|
pid = cfg.get("id", 0)
|
|
try:
|
|
resp = requests.post(
|
|
f"{cfg['base_url']}/chat/completions",
|
|
headers={"Authorization": f"Bearer {cfg['api_key']}",
|
|
"Content-Type": "application/json"},
|
|
json={"model": cfg["model"],
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": config.LLM_TEMPERATURE,
|
|
"max_tokens": config.LLM_MAX_TOKENS,
|
|
"response_format": {"type": "json_object"}},
|
|
timeout=config.LLM_TIMEOUT,
|
|
)
|
|
data = resp.json()
|
|
if not resp.ok or "choices" not in data:
|
|
raise RuntimeError(data.get("error", {}).get("message", f"HTTP {resp.status_code}"))
|
|
content = data["choices"][0]["message"]["content"]
|
|
usage = data.get("usage") or {}
|
|
pt = usage.get("prompt_tokens", 0) or 0
|
|
ct = usage.get("completion_tokens", 0) or 0
|
|
db.add_llm_stat(pname, pid, pt, ct, success=True)
|
|
return content, pname
|
|
except Exception as e:
|
|
db.add_llm_stat(pname, pid, 0, 0, success=False)
|
|
last_err = str(e)
|
|
continue
|
|
raise RuntimeError(f"所有大模型接口调用失败: {last_err}")
|
|
|
|
|
|
def llm_analyze(aid):
|
|
"""LLM 深度分析单条。
|
|
普通源：重要度/相关度/结论；定制监控源：按「推送标准」判断是否达到推送条件。"""
|
|
a = db.get_article(aid)
|
|
if not a:
|
|
return None
|
|
if source_is_custom(a):
|
|
src = db.get_source(a.get("source_id") or 0)
|
|
return _llm_standard_check(a, src)
|
|
profile = _profile_text()
|
|
prompt = (
|
|
"你是一位资深科技资讯分析师，专注AI领域。\n"
|
|
f"用户兴趣画像：\n{profile}\n\n"
|
|
f"资讯标题：{a['title']}\n"
|
|
f"资讯内容：{a.get('content') or a.get('summary')}\n\n"
|
|
"请只输出一个 JSON 对象（不要任何其他文字），格式：\n"
|
|
'{"importance": 1-10的整数, "relevance": 0-100的整数, '
|
|
'"is_important": true或false, "category": "分类名", "reason": "为什么对用户重要（40字内中文）"}'
|
|
)
|
|
try:
|
|
content, _provider = _llm_chat(prompt)
|
|
parsed = json.loads(content)
|
|
importance = max(1, min(10, int(parsed.get("importance", 5))))
|
|
relevance = max(0, min(100, int(parsed.get("relevance", 50))))
|
|
is_important = 1 if parsed.get("is_important") else 0
|
|
category = parsed.get("category", a.get("domain", ""))
|
|
reason = parsed.get("reason", "")
|
|
|
|
# LLM 结论与规则分融合
|
|
total = a.get("total_score", 0)
|
|
llm_component = int(round(importance * 10 * 0.5 + relevance * 0.2))
|
|
total = min(100, int(round(0.6 * total + 0.4 * llm_component)))
|
|
threshold = int(db.get_setting("realtime_threshold", config.AUTO_DEFAULTS["realtime_threshold"]))
|
|
if is_important == 0 and total >= threshold:
|
|
is_important = 1
|
|
|
|
db.update_article(
|
|
aid, importance=importance, relevance=max(relevance, a.get("relevance", 0)),
|
|
total_score=total, is_important=is_important, analysis=reason,
|
|
domain=category, llm_status="done",
|
|
)
|
|
return {"id": aid, "importance": importance, "relevance": relevance,
|
|
"total_score": total, "is_important": is_important, "reason": reason}
|
|
except Exception as e:
|
|
db.update_article(aid, llm_status="error")
|
|
if "所有大模型接口调用失败" in str(e):
|
|
try:
|
|
import notifier
|
|
notifier.report_error("分析", "LLM 接口全部不可用", str(e)[:300])
|
|
except Exception:
|
|
pass
|
|
return {"id": aid, "error": str(e)}
|
|
|
|
|
|
def _llm_standard_check(a, src):
|
|
"""定制监控源：按推送标准让大模型判断该条资讯是否达到推送条件。
|
|
达到 → is_important=1 → 实时邮件推送；未达到 → 不推送。"""
|
|
standard = (src.get("monitor_standard") or "").strip() or "重要资讯"
|
|
prompt = (
|
|
"你是一位资讯监控专员。用户配置了一个定制监控数据源，并设定了「推送标准」。\n"
|
|
f"【推送标准】\n{standard}\n\n"
|
|
f"【资讯标题】{a['title']}\n"
|
|
f"【资讯内容】{a.get('content') or a.get('summary')}\n\n"
|
|
"请严格对照推送标准判断：这条资讯是否达到应推送的程度？\n"
|
|
"只输出一个 JSON 对象（不要任何其他文字），格式：\n"
|
|
'{"meets_standard": true或false, "reason": "判断理由（40字内中文）"}'
|
|
)
|
|
try:
|
|
content, _provider = _llm_chat(prompt)
|
|
parsed = json.loads(content)
|
|
meets = 1 if parsed.get("meets_standard") else 0
|
|
reason = parsed.get("reason", "")
|
|
db.update_article(
|
|
a["id"], is_important=meets, analysis=reason, llm_status="done",
|
|
importance=7 if meets else 1,
|
|
)
|
|
return {"id": a["id"], "meets_standard": bool(meets), "reason": reason,
|
|
"total_score": a.get("total_score", 0), "is_important": meets,
|
|
"custom": True}
|
|
except Exception as e:
|
|
db.update_article(a["id"], llm_status="error")
|
|
if "所有大模型接口调用失败" in str(e):
|
|
try:
|
|
import notifier
|
|
notifier.report_error("分析", "LLM 接口全部不可用", str(e)[:300])
|
|
except Exception:
|
|
pass
|
|
return {"id": a["id"], "error": str(e)}
|
|
|
|
|
|
def _profile_text():
|
|
kws = "、".join(k["keyword"] for k in db.list_keywords() if k["enabled"])
|
|
comps = "、".join(c["name"] for c in db.list_companies() if c["enabled"])
|
|
doms = "、".join(d["name"] for d in db.list_domains() if d["enabled"])
|
|
return f"关注关键词：{kws}\n关注公司：{comps}\n关注领域：{doms}"
|
|
|
|
|
|
def batch_llm_analyze(limit=10):
|
|
"""后台线程：对 pending 的资讯做 LLM 深度分析
|
|
普通源：规则分达 llm_threshold 才分析；定制监控源：无条件分析（是否推送由大模型决定）。"""
|
|
threshold = int(db.get_setting("llm_threshold", config.AUTO_DEFAULTS["llm_threshold"]))
|
|
arts = db.pending_llm_articles(limit=limit)
|
|
results = {"done": 0, "error": 0, "skipped": 0}
|
|
for a in arts:
|
|
if a.get("total_score", 0) < threshold and not source_is_custom(a):
|
|
db.update_article(a["id"], llm_status="skipped")
|
|
results["skipped"] += 1
|
|
continue
|
|
r = llm_analyze(a["id"])
|
|
if r and "error" not in r:
|
|
results["done"] += 1
|
|
else:
|
|
results["error"] += 1
|
|
return results
|
|
|
|
|
|
def run_llm_background(limit=8):
|
|
def _job():
|
|
try:
|
|
batch_llm_analyze(limit=limit)
|
|
except Exception as e:
|
|
db.add_log("realtime", "LLM分析异常", 0, [], status="error", detail=str(e))
|
|
try:
|
|
import notifier
|
|
notifier.report_error("分析", "LLM 深度分析异常", str(e)[:300])
|
|
except Exception:
|
|
pass
|
|
t = threading.Thread(target=_job, daemon=True)
|
|
t.start()
|
|
return t
|