simon
9aa44e610d
- system.yaml: llm.max_retries(死配置) 改为 llm.max_attempts；embedding 段新增 max_attempts - llm/client.py: LLMConfig 新增 max_attempts（默认 3 兜底），load_llm_config 读取配置 - llm/extractor.py: translate_and_extract(_async) max_attempts=None 时取 config.max_attempts - embedding/client.py: load_embedding_config 读取 embedding.max_attempts - scheduler/reporter.py: _call_llm_simple max_retries=None 时取 max_attempts-1 - 新增 2 个配置读取测试；已同步 pi5 验证
1042 lines
39 KiB
Python
1042 lines
39 KiB
Python
"""每日 AI 摘要日报生成器。
|
||
|
||
输出 HTML 日报，包含:
|
||
一、AI 摘要（LLM 根据当日重要事件生成）
|
||
二、重要事件（importance ≥ 4，最多 20 篇）
|
||
三、数据总览（管道统计、情绪分布、重要度分布、事件类型分布）
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from collections import Counter
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import markdown
|
||
import yaml
|
||
|
||
from report_db.models import EventRow, ReportData # noqa: F401 - 供 _build_report_data 注解使用
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_MAX_HIGH_EVENTS = 30
|
||
_REPORT_DIR = Path("data/reports")
|
||
|
||
|
||
def _load_source_names() -> dict[str, str]:
|
||
"""从 sources.yaml 加载源名称映射。"""
|
||
try:
|
||
with open("configs/sources.yaml", encoding="utf-8") as f:
|
||
data = yaml.safe_load(f)
|
||
return {s["id"]: s["name"] for s in (data.get("sources") or []) if s.get("id")}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _source_name(src_id: str) -> str:
|
||
return _load_source_names().get(src_id, src_id)
|
||
|
||
|
||
def _load_domain_name_map() -> dict[str, str]:
|
||
"""构建域名 → 来源展示名称映射。
|
||
|
||
来源：
|
||
1. sources.yaml 中各源 homepage 的域名
|
||
2. 已知域名别名（如 investinglive.com → ForexLive）
|
||
"""
|
||
domain_map: dict[str, str] = {}
|
||
try:
|
||
with open("configs/sources.yaml", encoding="utf-8") as f:
|
||
data = yaml.safe_load(f)
|
||
for s in data.get("sources") or []:
|
||
name = s.get("name", "")
|
||
homepage = s.get("homepage", "")
|
||
if not name or not homepage:
|
||
continue
|
||
try:
|
||
from urllib.parse import urlparse
|
||
domain = urlparse(homepage).netloc.removeprefix("www.")
|
||
if domain:
|
||
domain_map[domain] = name
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
return domain_map
|
||
|
||
|
||
def _url_source_label(url: str, source_id: str = "") -> str:
|
||
"""从 URL 提取域名，映射为来源展示名称。
|
||
|
||
确保日报中链接域名与来源标注一致。
|
||
未知域名直接显示域名（去 www 前缀）。
|
||
|
||
Args:
|
||
url: 文章 URL
|
||
source_id: 回退用的 source_id
|
||
|
||
Returns:
|
||
来源展示名称
|
||
"""
|
||
if not url:
|
||
return _source_name(source_id) if source_id else "?"
|
||
|
||
try:
|
||
from urllib.parse import urlparse
|
||
domain = urlparse(url).netloc.removeprefix("www.")
|
||
except Exception:
|
||
return _source_name(source_id) if source_id else "?"
|
||
|
||
if not domain:
|
||
return _source_name(source_id) if source_id else "?"
|
||
|
||
domain_map = _load_domain_name_map()
|
||
return domain_map.get(domain, domain)
|
||
|
||
|
||
# 日报覆盖时间窗口（小时）
|
||
_REPORT_WINDOW_HOURS = 25
|
||
|
||
|
||
def _dates_in_window(now: datetime) -> list[str]:
|
||
"""返回 now - 25h 到 now 之间覆盖的所有 YYYYMMDD 日期。
|
||
|
||
跨天场景：now=06-21 03:00 → now-25h=06-20 02:00 → 返回 ["20260620", "20260621"]
|
||
"""
|
||
start = now - timedelta(hours=_REPORT_WINDOW_HOURS)
|
||
dates: list[str] = []
|
||
current = start.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
end = now.replace(hour=23, minute=59, second=59)
|
||
while current <= end:
|
||
dates.append(current.strftime("%Y%m%d"))
|
||
current += timedelta(days=1)
|
||
return dates
|
||
|
||
|
||
def _try_parse_time(time_str: str) -> datetime | None:
|
||
"""尝试解析多种 ISO 8601 变体，失败返回 None。"""
|
||
if not time_str or not time_str.strip():
|
||
return None
|
||
s = time_str.strip()
|
||
# 按长度尝试常见格式
|
||
candidates = [s]
|
||
if "T" not in s and len(s) == 8: # YYYYMMDD
|
||
candidates.append(f"{s[:4]}-{s[4:6]}-{s[6:8]}T00:00:00")
|
||
elif "T" not in s and len(s) == 10: # YYYY-MM-DD
|
||
candidates.append(f"{s}T00:00:00")
|
||
for c in candidates:
|
||
try:
|
||
dt = datetime.fromisoformat(c)
|
||
# 统一转为 UTC-aware：
|
||
# - 有时区 → astimezone 转为 UTC
|
||
# - 无时区 → 假设为 UTC（多数财经新闻 API 使用 UTC）
|
||
if dt.tzinfo is None:
|
||
return dt.replace(tzinfo=UTC)
|
||
return dt.astimezone(UTC)
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _extract_date_from_url(url: str) -> str:
|
||
"""从 URL 中提取日期（回退策略）。
|
||
|
||
匹配模式: /YYYY/MM/DD/、/YYYYMMDD/、-YYYYMMDD、-YYYY-MM-DD
|
||
"""
|
||
import re
|
||
patterns = [
|
||
r"/(\d{4})/(\d{2})/(\d{2})/", # /2026/06/19/
|
||
r"/(\d{4})(\d{2})(\d{2})/", # /20260619/
|
||
r"-(\d{4})(\d{2})(\d{2})(?:[/-]|$)", # -20260619/ or -20260619- or -20260619
|
||
r"-(\d{4})-(\d{2})-(\d{2})[/-]", # -2026-06-19/
|
||
]
|
||
for pat in patterns:
|
||
m = re.search(pat, url)
|
||
if m:
|
||
y, mo, d = m.group(1), m.group(2), m.group(3)
|
||
try:
|
||
dt = datetime(int(y), int(mo), int(d))
|
||
return dt.isoformat()
|
||
except ValueError:
|
||
continue
|
||
return ""
|
||
|
||
|
||
def _load_events_window(now: datetime) -> list[dict]:
|
||
"""加载过去 25 小时内所有事件文章（跨天聚合）。
|
||
|
||
Returns:
|
||
文章 dict 列表，含 title/title_zh/url/source_id/events 等
|
||
"""
|
||
# cutoff 使用 UTC-aware，与 _try_parse_time 返回的 UTC datetime 对齐
|
||
cutoff = now.astimezone(UTC) - timedelta(hours=_REPORT_WINDOW_HOURS)
|
||
articles: list[dict] = []
|
||
skipped_empty_pt = 0
|
||
for day_str in _dates_in_window(now):
|
||
ev_dir = Path(f"data/events/{day_str}")
|
||
if not ev_dir.is_dir():
|
||
continue
|
||
for fp in sorted(ev_dir.glob("*.json")):
|
||
if fp.name == "index.json":
|
||
continue
|
||
try:
|
||
data = json.loads(fp.read_text(encoding="utf-8"))
|
||
# 时间过滤：publish_time 在 25 小时内
|
||
pt_str = data.get("publish_time", "")
|
||
pt = _try_parse_time(pt_str)
|
||
|
||
# 回退：尝试从 URL 提取日期
|
||
if pt is None:
|
||
url_pt = _extract_date_from_url(data.get("url", ""))
|
||
pt = _try_parse_time(url_pt)
|
||
if pt is not None:
|
||
logger.debug(
|
||
"publish_time 缺失，从 URL 提取: %s → %s",
|
||
data.get("url", "")[:60], url_pt,
|
||
)
|
||
|
||
# 最后回退：使用事件目录日期（粗略近似）
|
||
if pt is None:
|
||
dir_pt = _try_parse_time(day_str)
|
||
if dir_pt is not None:
|
||
logger.debug(
|
||
"publish_time 缺失，使用目录日期回退: %s → %s",
|
||
fp.name, day_str,
|
||
)
|
||
pt = dir_pt
|
||
|
||
# 仍然无法确定时间 → 排除
|
||
if pt is None:
|
||
skipped_empty_pt += 1
|
||
continue
|
||
|
||
if pt < cutoff:
|
||
continue
|
||
articles.append(data)
|
||
except (json.JSONDecodeError, OSError):
|
||
pass
|
||
if skipped_empty_pt > 0:
|
||
logger.warning(
|
||
"日报时间过滤: 排除 %d 篇 publish_time 缺失的文章（窗口 %dh）",
|
||
skipped_empty_pt, _REPORT_WINDOW_HOURS,
|
||
)
|
||
return articles
|
||
|
||
|
||
def _collect_stats_window(now: datetime) -> dict:
|
||
"""收集过去 25 小时全链路统计数据（跨天聚合）。"""
|
||
proc_count = 0
|
||
deduped = 0
|
||
emb_count = 0
|
||
raw_by_source: dict[str, int] = {}
|
||
|
||
for day_str in _dates_in_window(now):
|
||
for d in Path("data/processed").glob(f"*/{day_str}"):
|
||
proc_count += len(list(d.glob("*.json")))
|
||
dedup_dir = Path(f"data/deduped/{day_str}/uniques")
|
||
if dedup_dir.is_dir():
|
||
deduped += len(list(dedup_dir.glob("*.json")))
|
||
emb_dir = Path(f"data/embeddings/{day_str}")
|
||
if emb_dir.is_dir():
|
||
emb_count += len(list(emb_dir.glob("*.json")))
|
||
for idx in Path("data/raw").glob(f"*/{day_str}/index.jsonl"):
|
||
src = idx.parent.parent.name
|
||
n = sum(1 for _ in open(idx, encoding="utf-8"))
|
||
name = _source_name(src)
|
||
raw_by_source[name] = raw_by_source.get(name, 0) + n
|
||
|
||
qdrant_count = 0
|
||
try:
|
||
from vectorstore.client import VectorStore, make_qdrant_client
|
||
c = make_qdrant_client()
|
||
s = VectorStore(c)
|
||
qdrant_count = s.count()
|
||
s.close()
|
||
except Exception:
|
||
pass
|
||
|
||
return {
|
||
"proc": proc_count,
|
||
"deduped": deduped,
|
||
"emb_count": emb_count,
|
||
"qdrant_count": qdrant_count,
|
||
"raw_total": sum(raw_by_source.values()),
|
||
"raw_by_source": raw_by_source,
|
||
}
|
||
|
||
|
||
# AI 摘要分批阈值：单批最多处理的事件条数
|
||
_MAX_EVENTS_PER_BATCH = 10
|
||
# 单批摘要目标字数（软上限，最后一条不允许截断）
|
||
_BATCH_SUMMARY_TARGET_CHARS = 300
|
||
# 最终摘要目标字数（软上限，最后一条不允许截断）
|
||
_FINAL_SUMMARY_TARGET_CHARS = 500
|
||
|
||
|
||
# LLM 客户端缓存（避免每次调用都创建新客户端）
|
||
_llm_client_cache: dict = {}
|
||
|
||
|
||
def _call_llm_simple(
|
||
system_prompt: str,
|
||
user_prompt: str,
|
||
max_tokens: int = 600,
|
||
max_retries: int | None = None,
|
||
) -> str:
|
||
"""封装 LLM 调用，带重试和客户端复用。
|
||
|
||
Args:
|
||
system_prompt: system role 内容
|
||
user_prompt: user role 内容
|
||
max_tokens: 最大输出 token
|
||
max_retries: 重试次数（不含首次）；None 时取
|
||
config.max_attempts - 1（system.yaml llm.max_attempts）
|
||
|
||
Returns:
|
||
LLM 输出文本；所有重试均失败返回空字符串
|
||
"""
|
||
import time as _time
|
||
|
||
# 复用客户端（同 provider/model 只创建一次）
|
||
cache_key = "default"
|
||
if cache_key not in _llm_client_cache:
|
||
from llm.client import load_llm_config, make_sync_client
|
||
_llm_client_cache["config"] = load_llm_config()
|
||
_llm_client_cache[cache_key] = make_sync_client(_llm_client_cache["config"])
|
||
|
||
config = _llm_client_cache["config"]
|
||
client = _llm_client_cache[cache_key]
|
||
|
||
if max_retries is None:
|
||
max_retries = max(config.max_attempts - 1, 0)
|
||
|
||
last_err: str = ""
|
||
for attempt in range(1, max_retries + 2): # 首次 + max_retries 次重试
|
||
try:
|
||
resp = client.chat.completions.create(
|
||
model=config.model,
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt},
|
||
],
|
||
temperature=0.3,
|
||
max_tokens=max_tokens,
|
||
)
|
||
content = (resp.choices[0].message.content or "").strip()
|
||
if content:
|
||
return content
|
||
# 内容为空也视为失败
|
||
last_err = "LLM 返回空内容"
|
||
logger.warning(
|
||
"LLM 返回空内容 (attempt %d/%d, max_tokens=%d)",
|
||
attempt, max_retries + 1, max_tokens,
|
||
)
|
||
except Exception as e:
|
||
last_err = f"{type(e).__name__}: {e}"
|
||
logger.warning(
|
||
"LLM 调用失败 (attempt %d/%d): %s",
|
||
attempt, max_retries + 1, last_err,
|
||
)
|
||
|
||
if attempt <= max_retries:
|
||
wait = min(1.0 * (2 ** (attempt - 1)), 4.0)
|
||
_time.sleep(wait)
|
||
|
||
logger.error("LLM 调用最终失败（共 %d 次尝试）: %s", max_retries + 1, last_err)
|
||
return ""
|
||
|
||
|
||
def _is_truncated(text: str) -> bool:
|
||
"""检测文本是否被截断（最后一条要点不完整）。
|
||
|
||
判断标准：
|
||
- 以完整中文/英文句末标点结尾 → 未截断
|
||
- 以换行结尾 → 未截断（LLM 自然换行说明已写完当前要点）
|
||
- 最后一行是 "- " 开头的要点但无句末标点 → 截断
|
||
- 其他不以标点结尾的情况 → 截断
|
||
"""
|
||
if not text or not text.strip():
|
||
return False
|
||
_SENTENCE_END = ("。", "！", "？", "）", ")", "」", "』", "”", ".", "!", "?")
|
||
# 以完整句末标点结尾 → 未截断
|
||
if text.rstrip().endswith(_SENTENCE_END):
|
||
return False
|
||
# 以换行结尾 → 大概率未截断（先检查原始文本，避免 rstrip 去掉换行）
|
||
if text.endswith("\n"):
|
||
return False
|
||
# 最后一行以 "- " 开头但无句末标点 → 截断
|
||
last_line = text.rstrip().split("\n")[-1].strip()
|
||
if last_line.startswith("- ") and not last_line.endswith(_SENTENCE_END):
|
||
return True
|
||
# 不以任何已知完整标记结尾 → 截断
|
||
return True
|
||
|
||
|
||
def _build_event_lines(articles: list[dict]) -> list[str]:
|
||
"""从文章列表构建事件摘要行列表（仅 importance ≥ 4）。"""
|
||
lines: list[str] = []
|
||
for a in articles:
|
||
for ev in a.get("events", []):
|
||
if ev.get("importance", 0) >= 4:
|
||
sentiment_label = {
|
||
"positive": "利好", "negative": "利空", "neutral": "中性",
|
||
}.get(ev.get("sentiment", ""), "")
|
||
codes = ",".join(ev.get("stock_codes", [])[:3])
|
||
code_str = f" [{codes}]" if codes else ""
|
||
lines.append(
|
||
f"- [{sentiment_label}][{ev.get('event_type', '')}] "
|
||
f"{a.get('title_zh', a.get('title', ''))}。{ev.get('summary_zh', '')}{code_str}"
|
||
)
|
||
return lines
|
||
|
||
|
||
def _fallback_summary(events_info: list[str], max_items: int = 8) -> str:
|
||
"""LLM 全部失败时的规则回退摘要：直接列出当日 Top 事件。
|
||
|
||
不依赖 LLM，直接从 events_info 提取前 max_items 条展示。
|
||
"""
|
||
if not events_info:
|
||
return "⚠️ 暂无重要事件数据"
|
||
|
||
top = events_info[:max_items]
|
||
lines = [
|
||
f"- {line.lstrip('- ')}"
|
||
for line in top
|
||
]
|
||
header = (
|
||
"⚠️ AI 摘要暂时无法生成（LLM 服务异常），以下是当日重要事件原始列表：\n\n"
|
||
)
|
||
return header + "\n".join(lines)
|
||
|
||
|
||
def _generate_ai_summary(articles: list[dict], day_str: str) -> str:
|
||
"""LLM 生成日报 AI 摘要（≤ 500 字要点列表）。
|
||
|
||
当高重要度事件超过 _MAX_EVENTS_PER_BATCH 条时，分批生成部分摘要，
|
||
最后合并为最终摘要。避免单次输入内容过长导致 LLM 失败。
|
||
"""
|
||
events_info = _build_event_lines(articles)
|
||
|
||
if not events_info:
|
||
return ""
|
||
|
||
# ── 事件少：直接生成 ──
|
||
if len(events_info) <= _MAX_EVENTS_PER_BATCH:
|
||
input_text = "\n".join(events_info)
|
||
prompt = f"""以下是国际财经新闻的当日重要事件摘要 ({day_str}):
|
||
|
||
{input_text}
|
||
|
||
请用要点总结，每条以 "- " 开头，要求:
|
||
1. 前 3 条为当日影响最大的事件，说明为什么重要
|
||
2. 市场情绪基调（利好/利空/中性分布）
|
||
3. 值得持续关注的行业、主题或地缘政治动向
|
||
4. 纯要点，不要开场白/结束语/标题
|
||
5. 约 {_FINAL_SUMMARY_TARGET_CHARS} 字左右，但最后一条要点必须完整输出，严禁截断
|
||
|
||
直接输出要点列表:"""
|
||
|
||
result = _call_llm_simple(
|
||
"你是国际财经日报撰写助手，输出简洁、有洞察的新闻摘要。",
|
||
prompt,
|
||
max_tokens=800,
|
||
)
|
||
if result:
|
||
# 截断检测：若被截断，以更高 max_tokens 重试一次
|
||
if _is_truncated(result):
|
||
logger.warning("AI 摘要疑似截断，以更高 max_tokens 重试")
|
||
retry_prompt = prompt + "\n\n⚠️ 注意：上次输出被截断了，请确保最后一条要点完整结束。"
|
||
retry_result = _call_llm_simple(
|
||
"你是国际财经日报撰写助手。务必确保输出完整，不以不完整句子结尾。",
|
||
retry_prompt,
|
||
max_tokens=1200,
|
||
)
|
||
if retry_result:
|
||
return retry_result
|
||
return result
|
||
logger.warning("AI 摘要：单次 LLM 调用失败，使用规则回退")
|
||
return _fallback_summary(events_info)
|
||
|
||
# ── 事件多：分批处理 ──
|
||
logger.info(
|
||
"AI 摘要分批处理: 共 %d 条高重要度事件，每批 ≤ %d 条",
|
||
len(events_info), _MAX_EVENTS_PER_BATCH,
|
||
)
|
||
|
||
# 分批生成部分摘要
|
||
partial_summaries: list[str] = []
|
||
for batch_idx in range(0, len(events_info), _MAX_EVENTS_PER_BATCH):
|
||
batch = events_info[batch_idx:batch_idx + _MAX_EVENTS_PER_BATCH]
|
||
batch_num = batch_idx // _MAX_EVENTS_PER_BATCH + 1
|
||
total_batches = (len(events_info) + _MAX_EVENTS_PER_BATCH - 1) // _MAX_EVENTS_PER_BATCH
|
||
|
||
batch_text = "\n".join(batch)
|
||
prompt = f"""以下是国际财经新闻的当日重要事件摘要 第 {batch_num}/{total_batches} 批 ({day_str}):
|
||
|
||
{batch_text}
|
||
|
||
请用要点总结本批事件，每条以 "- " 开头，要求:
|
||
1. 提取本批最重要的 3-5 条事件
|
||
2. 说明这些事件的市场影响方向
|
||
3. 纯要点，不要开场白/结束语/标题
|
||
4. 约 {_BATCH_SUMMARY_TARGET_CHARS} 字左右，但最后一条要点必须完整输出，严禁截断
|
||
|
||
直接输出要点列表:"""
|
||
|
||
result = _call_llm_simple(
|
||
"你是国际财经日报撰写助手，输出简洁、有洞察的新闻摘要。",
|
||
prompt,
|
||
max_tokens=500,
|
||
)
|
||
if result:
|
||
# 截断检测：若被截断，以更高 max_tokens 重试一次
|
||
if _is_truncated(result):
|
||
logger.warning(
|
||
"AI 摘要分批: 第 %d/%d 批疑似截断，重试", batch_num, total_batches,
|
||
)
|
||
retry_prompt = prompt + "\n\n⚠️ 注意：上次输出被截断了，请确保最后一条要点完整结束。"
|
||
retry_result = _call_llm_simple(
|
||
"你是国际财经日报撰写助手。务必确保输出完整，不以不完整句子结尾。",
|
||
retry_prompt,
|
||
max_tokens=800,
|
||
)
|
||
if retry_result:
|
||
partial_summaries.append(retry_result)
|
||
logger.info("AI 摘要分批: 第 %d/%d 批重试完成 (%d 字)",
|
||
batch_num, total_batches, len(retry_result))
|
||
continue
|
||
partial_summaries.append(result)
|
||
logger.info("AI 摘要分批: 第 %d/%d 批完成 (%d 字)",
|
||
batch_num, total_batches, len(result))
|
||
else:
|
||
logger.warning("AI 摘要分批: 第 %d/%d 批失败", batch_num, total_batches)
|
||
|
||
if not partial_summaries:
|
||
# 全部批次 LLM 调用失败 → 回退：直接列出 Top 事件
|
||
logger.warning("AI 摘要：所有分批 LLM 调用均失败，使用规则回退")
|
||
return _fallback_summary(events_info)
|
||
|
||
# ── 合并部分摘要为最终摘要 ──
|
||
merged_input = "\n\n---\n\n".join(
|
||
f"第 {i+1} 批摘要:\n{s}" for i, s in enumerate(partial_summaries)
|
||
)
|
||
merge_prompt = f"""以下是当日国际财经新闻的多批摘要 ({day_str})，请合并为一份简洁的最终日报摘要:
|
||
|
||
{merged_input}
|
||
|
||
请合并为要点总结，每条以 "- " 开头，要求:
|
||
1. 前 3 条为当日影响最大的事件，说明为什么重要
|
||
2. 市场情绪基调（利好/利空/中性分布）
|
||
3. 值得持续关注的行业、主题或地缘政治动向
|
||
4. 纯要点，不要开场白/结束语/标题
|
||
5. 约 {_FINAL_SUMMARY_TARGET_CHARS} 字左右，但最后一条要点必须完整输出，严禁截断
|
||
|
||
直接输出要点列表:"""
|
||
|
||
result = _call_llm_simple(
|
||
"你是国际财经日报撰写助手，输出简洁、有洞察的新闻摘要。合并多批摘要时注意去重。",
|
||
merge_prompt,
|
||
max_tokens=1000,
|
||
)
|
||
if result:
|
||
# 截断检测：若被截断，以更高 max_tokens 重试一次
|
||
if _is_truncated(result):
|
||
logger.warning("AI 摘要合并疑似截断，以更高 max_tokens 重试")
|
||
retry_merge_prompt = merge_prompt + "\n\n⚠️ 注意：上次输出被截断了，请确保最后一条要点完整结束。"
|
||
retry_result = _call_llm_simple(
|
||
"你是国际财经日报撰写助手。务必确保输出完整，不以不完整句子结尾。合并多批摘要时注意去重。",
|
||
retry_merge_prompt,
|
||
max_tokens=1500,
|
||
)
|
||
if retry_result:
|
||
return retry_result
|
||
return result
|
||
# 合并失败 → 拼接所有部分摘要作为回退
|
||
logger.warning("AI 摘要：合并 LLM 调用失败，使用部分摘要拼接")
|
||
return "⚠️ AI 摘要合并失败，以下为各批次原始摘要：\n\n" + "\n\n".join(partial_summaries)
|
||
|
||
|
||
def _build_report_data(
|
||
now: datetime,
|
||
stats: dict,
|
||
high_events: list[dict],
|
||
sentiments: Counter,
|
||
importances: Counter,
|
||
event_types: Counter,
|
||
sources: Counter,
|
||
ai_summary: str,
|
||
) -> ReportData:
|
||
"""组装结构化日报数据（M9：写入 MySQL 的前置步骤）。
|
||
|
||
与 news 项目 report_db_design.md 数据契约一致：
|
||
- 事件全部归入 `intl` 板块（intl 日报无 xwlb/news/cninfo 板块）；
|
||
- report_type="intl"、file_name="" → 唯一键退化为 (report_date, intl, "")，
|
||
同一天重复生成 = UPDATE 覆盖（幂等）；
|
||
- 数据总览统计以 JSON 快照存入 stats（前端自行解析）。
|
||
"""
|
||
events: list[EventRow] = []
|
||
for i, ev in enumerate(high_events, 1):
|
||
article = ev.get("article", {})
|
||
title = (article.get("title_zh") or article.get("title") or "").strip()[:512]
|
||
url = article.get("url") or None
|
||
src = _url_source_label(url, article.get("source_id", ""))
|
||
# 归一化："" / "?" 不入库，留 None（DB 仅存 positive/negative/neutral）
|
||
sentiment = ev.get("sentiment") or None
|
||
if sentiment in ("", "?"):
|
||
sentiment = None
|
||
event_type = ev.get("event_type") or None
|
||
if event_type in ("", "?"):
|
||
event_type = None
|
||
events.append(
|
||
EventRow(
|
||
section="intl",
|
||
rank=i,
|
||
importance=ev.get("importance") or None,
|
||
event_type=event_type,
|
||
title=title or "（无标题）",
|
||
summary=ev.get("summary_zh") or None,
|
||
sentiment=sentiment,
|
||
source=src if src not in (None, "", "?") else None,
|
||
url=url,
|
||
)
|
||
)
|
||
|
||
stats_snapshot: dict[str, Any] = {
|
||
"pipeline": {
|
||
"raw_total": stats.get("raw_total", 0),
|
||
"processed": stats.get("proc", 0),
|
||
"deduped": stats.get("deduped", 0),
|
||
"embedded": stats.get("emb_count", 0),
|
||
"qdrant": stats.get("qdrant_count", 0),
|
||
},
|
||
"sentiment": {k: v for k, v in sentiments.items() if k not in ("", "?")},
|
||
"importance": [
|
||
{"importance": k, "count": v} for k, v in sorted(importances.items())
|
||
],
|
||
"event_types": [
|
||
{"event_type": k, "count": v}
|
||
for k, v in event_types.most_common(10)
|
||
if k not in ("", "?")
|
||
],
|
||
"source_dist": [
|
||
{"source": _source_name(k), "count": v}
|
||
for k, v in sources.most_common(15)
|
||
if k not in ("", "?")
|
||
],
|
||
}
|
||
|
||
return ReportData(
|
||
report_date=now.date(),
|
||
report_type="intl",
|
||
file_name="", # 新生成日报唯一键退化为 (report_date, intl, "")
|
||
generated_at=now,
|
||
ai_summary=ai_summary or None,
|
||
stats=stats_snapshot,
|
||
events=events,
|
||
)
|
||
|
||
|
||
def generate_report() -> int | None:
|
||
"""生成日报并结构化入库（覆盖过去 25 小时数据）。
|
||
|
||
M9 起完全切换：不再产出 HTML，日报内容写入 MySQL
|
||
（news_report / news_event，report_type="intl"，file_name=""，
|
||
同一天重复生成 → 幂等覆盖，不产生多行）。
|
||
|
||
Returns:
|
||
report_id（成功）或 None（无数据/失败）
|
||
"""
|
||
now = datetime.now()
|
||
ts = now.strftime("%Y%m%d_%H%M%S")
|
||
logger.info("生成日报: %s（窗口: 过去 %d 小时）", ts, _REPORT_WINDOW_HOURS)
|
||
|
||
# 加载过去 25 小时数据
|
||
articles = _load_events_window(now)
|
||
stats = _collect_stats_window(now)
|
||
|
||
if not articles and stats["raw_total"] == 0:
|
||
logger.warning("过去 %d 小时无数据，跳过日报生成", _REPORT_WINDOW_HOURS)
|
||
return None
|
||
|
||
# 收集事件统计
|
||
all_events: list[dict] = []
|
||
sentiments: Counter = Counter()
|
||
importances: Counter = Counter()
|
||
event_types: Counter = Counter()
|
||
sources: Counter = Counter()
|
||
|
||
for a in articles:
|
||
sources[a.get("source_id", "?")] += 1
|
||
for ev in a.get("events", []):
|
||
all_events.append({**ev, "article": a})
|
||
sentiments[ev.get("sentiment", "?")] += 1
|
||
importances[ev.get("importance", 0)] += 1
|
||
event_types[ev.get("event_type", "?")] += 1
|
||
|
||
# 高重要度事件（importance ≥ 4，不足逐级回退）
|
||
def _get_high(evs, threshold):
|
||
return sorted(
|
||
[e for e in evs if e.get("importance", 0) >= threshold],
|
||
key=lambda e: -e.get("importance", 0),
|
||
)
|
||
|
||
high = _get_high(all_events, 4)
|
||
if len(high) < 3:
|
||
high = _get_high(all_events, 3)
|
||
if len(high) < 3:
|
||
high = sorted(all_events, key=lambda e: -e.get("importance", 0))
|
||
# 事件级去重：同一 URL + 同一标题 → 合并
|
||
high = _dedup_events(high)
|
||
high = high[:_MAX_HIGH_EVENTS]
|
||
|
||
# AI 摘要
|
||
ai_summary = _generate_ai_summary(articles, ts)
|
||
|
||
# 结构化入库（替代原 HTML 渲染 + 上传）
|
||
report = _build_report_data(
|
||
now, stats, high, sentiments, importances, event_types, sources, ai_summary
|
||
)
|
||
try:
|
||
from report_db import connect, save_report
|
||
|
||
conn = connect()
|
||
try:
|
||
report_id = save_report(conn, report)
|
||
finally:
|
||
conn.close()
|
||
except Exception as e:
|
||
logger.exception("日报入库失败: %s", e)
|
||
return None
|
||
|
||
logger.info("日报已入库: report_id=%s", report_id)
|
||
return report_id
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 上传
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _load_report_config() -> dict:
|
||
"""从 system.yaml 加载 report 段配置。"""
|
||
config_path = Path("configs/system.yaml")
|
||
if config_path.exists():
|
||
try:
|
||
with open(config_path, encoding="utf-8") as f:
|
||
raw = yaml.safe_load(f)
|
||
return raw.get("report", {})
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def _upload_report(html_path: Path, day_str: str) -> bool:
|
||
"""上传日报到 Web 服务器（M9 起弃用：日报已改为写库，保留以便回退）。"""
|
||
import subprocess
|
||
|
||
config = _load_report_config()
|
||
host = config.get("upload_host", "").strip()
|
||
base = config.get("upload_path", "").strip()
|
||
|
||
if not host or not base:
|
||
logger.debug("未配置 report.upload_host/upload_path，跳过上传")
|
||
return False
|
||
|
||
remote_dir = f"{host}:{base}/{day_str}/"
|
||
|
||
try:
|
||
# 创建远程目录
|
||
subprocess.run(
|
||
["ssh", host, f"mkdir -p {base}/{day_str}/"],
|
||
timeout=15, capture_output=True, text=True,
|
||
)
|
||
# 上传
|
||
subprocess.run(
|
||
["scp", str(html_path), remote_dir],
|
||
timeout=30, capture_output=True, text=True,
|
||
)
|
||
logger.info(
|
||
"日报已上传: https://echart.doorcome.cn/research/%s/", day_str
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("日报上传失败（不阻塞）: %s", e)
|
||
return False
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# HTML 渲染（M9 起弃用：日报已改为写库，以下渲染函数保留以便回退）
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
_HTML_TEMPLATE = """<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||
<title>国际财经 Deep Research 日报 — {date}</title>
|
||
<style>
|
||
:root {{ --bg: #f8f9fa; --card: #fff; --text: #212529; --muted: #6c757d;
|
||
--accent: #2563eb; --border: #dee2e6; --pos: #059669; --neg: #dc2626;
|
||
--neu: #6b7280; --radius: 10px; }}
|
||
* {{ box-sizing: border-box; margin: 0; padding: 0; }}
|
||
body {{ font-family: -apple-system, BlinkMacSystemFont, "PingFang SC", "Microsoft YaHei", sans-serif; background: var(--bg); color: var(--text); line-height: 1.7; padding-bottom: 3rem; }}
|
||
.container {{ max-width: 1000px; margin: 0 auto; padding: 1.2rem; }}
|
||
header {{ background: linear-gradient(135deg, #0f172a, #1e3a5f); color: #fff; padding: 2.5rem 0 1.8rem; text-align: center; }}
|
||
header h1 {{ font-size: 1.8rem; }}
|
||
header p {{ color: #94a3b8; margin-top: .4rem; }}
|
||
h2 {{ font-size: 1.3rem; margin: 2rem 0 .8rem; padding-bottom: .4rem; border-bottom: 2px solid var(--accent); }}
|
||
.ai-summary {{ background: linear-gradient(135deg, #eff6ff, #f0fdf4); border: 1px solid #93c5fd; border-radius: var(--radius); padding: 1.2rem 1.5rem; margin: 1rem 0; line-height: 1.9; font-size: .95em; }}
|
||
.stats-grid {{ display: grid; grid-template-columns: repeat(5, 1fr); gap: .8rem; margin: 1rem 0; }}
|
||
.stat-card {{ background: var(--card); border: 1px solid var(--border); border-radius: var(--radius); padding: 1rem .8rem; text-align: center; }}
|
||
.stat-card .num {{ font-size: 1.6rem; font-weight: 700; }}
|
||
.stat-card .label {{ font-size: .8em; color: var(--muted); margin-top: .2rem; }}
|
||
.sentiment-bar {{ display: flex; height: 24px; border-radius: 6px; overflow: hidden; margin: .5rem 0; }}
|
||
.sentiment-bar .pos {{ background: var(--pos); }}
|
||
.sentiment-bar .neg {{ background: var(--neg); }}
|
||
.sentiment-bar .neu {{ background: var(--neu); }}
|
||
.sentiment-legend {{ display: flex; gap: 1rem; margin: .5rem 0; font-size: .9em; }}
|
||
table {{ width: 100%; border-collapse: collapse; margin: .8rem 0; font-size: .93em; }}
|
||
th, td {{ border: 1px solid var(--border); padding: .5rem .7rem; text-align: left; }}
|
||
th {{ background: #f1f5f9; font-weight: 600; white-space: nowrap; }}
|
||
.event-row:hover {{ background: #f8fafc; }}
|
||
.badge {{ display: inline-block; padding: .1em .5em; border-radius: 10px; font-size: .78em; font-weight: 600; }}
|
||
.badge-pos {{ background: #d1fae5; color: #065f46; }}
|
||
.badge-neg {{ background: #fee2e2; color: #991b1b; }}
|
||
.badge-neu {{ background: #e5e7eb; color: #374151; }}
|
||
.imp {{ font-weight: 700; }} .imp-5 {{ color: #dc2626; }} .imp-4 {{ color: #ea580c; }}
|
||
footer {{ text-align: center; color: var(--muted); font-size: .82em; margin-top: 3rem; padding-top: 1.2rem; border-top: 1px solid var(--border); }}
|
||
a {{ color: var(--accent); text-decoration: none; }}
|
||
a:hover {{ text-decoration: underline; }}
|
||
@media (max-width: 768px) {{
|
||
.stats-grid {{ grid-template-columns: repeat(3, 1fr); }}
|
||
}}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<header>
|
||
<div class="container">
|
||
<h1>🌍 国际财经 Deep Research 日报</h1>
|
||
<p>{date} · 生成于 {generated_at}</p>
|
||
</div>
|
||
</header>
|
||
<main class="container">
|
||
|
||
<!-- ====== 一、AI 摘要 ====== -->
|
||
<h2>一、🤖 AI 摘要</h2>
|
||
<div class="ai-summary">{ai_summary}</div>
|
||
|
||
<!-- ====== 二、重要事件 ====== -->
|
||
<h2>二、🔥 重要事件 <small style="color:var(--muted)">(importance ≥ {hi_threshold}, {high_count} 条)</small></h2>
|
||
{events_table}
|
||
|
||
<!-- ====== 三、数据总览 ====== -->
|
||
<h2>三、📊 数据总览</h2>
|
||
|
||
<h3>3.1 M1→M6 管道</h3>
|
||
<div class="stats-grid">
|
||
<div class="stat-card"><div class="num">{raw_total}</div><div class="label">M1 原始文章</div></div>
|
||
<div class="stat-card"><div class="num">{proc}</div><div class="label">M2 正文提取</div></div>
|
||
<div class="stat-card"><div class="num">{deduped}</div><div class="label">M3 去重唯一</div></div>
|
||
<div class="stat-card"><div class="num">{emb_count}</div><div class="label">M5 向量</div></div>
|
||
<div class="stat-card"><div class="num">{qdrant_count}</div><div class="label">M6 Qdrant</div></div>
|
||
</div>
|
||
|
||
<h3>3.2 情绪分布 <small style="color:var(--muted)">(当日事件)</small></h3>
|
||
{sentiment_section}
|
||
|
||
<h3>3.3 重要度分布</h3>
|
||
{importance_table}
|
||
|
||
<h3>3.4 事件类型 TOP 10</h3>
|
||
{event_type_table}
|
||
|
||
<h3>3.5 文章来源分布</h3>
|
||
{source_table}
|
||
|
||
</main>
|
||
<footer>
|
||
<div class="container">
|
||
<p>国际财经 Deep Research 私有投研平台 · 自动生成于 {generated_at}</p>
|
||
</div>
|
||
</footer>
|
||
</body>
|
||
</html>"""
|
||
|
||
|
||
def _md_to_html(text: str) -> str:
|
||
"""Markdown → HTML（使用 python-markdown，开启常用扩展）。"""
|
||
if not text.strip():
|
||
return ""
|
||
return markdown.markdown(
|
||
text,
|
||
extensions=["nl2br"], # 单换行 → <br>
|
||
)
|
||
|
||
|
||
def _render_html(
|
||
day_str: str,
|
||
stats: dict,
|
||
articles: list[dict],
|
||
high_events: list[dict],
|
||
hi_threshold: int,
|
||
sentiments: Counter,
|
||
importances: Counter,
|
||
event_types: Counter,
|
||
sources: Counter,
|
||
ai_summary: str,
|
||
) -> str:
|
||
"""组装完整 HTML（M9 起弃用，保留以便回退）。"""
|
||
|
||
# AI 摘要 Markdown → HTML
|
||
summary_html = _md_to_html(ai_summary) if ai_summary.strip() else "<p>暂无 AI 摘要</p>"
|
||
|
||
# 事件表格
|
||
events_table = _render_event_table(high_events)
|
||
|
||
# 情绪
|
||
pos = sentiments.get("positive", 0)
|
||
neg = sentiments.get("negative", 0)
|
||
neu = sentiments.get("neutral", 0)
|
||
total_s = max(pos + neg + neu, 1)
|
||
sentiment_section = (
|
||
f'<div class="sentiment-bar">'
|
||
f'<div class="pos" style="width:{pos/total_s*100:.0f}%"></div>'
|
||
f'<div class="neg" style="width:{neg/total_s*100:.0f}%"></div>'
|
||
f'<div class="neu" style="width:{neu/total_s*100:.0f}%"></div>'
|
||
f'</div>'
|
||
f'<div class="sentiment-legend">'
|
||
f'<span>🟢 利好 {pos} ({pos/total_s:.0%})</span>'
|
||
f'<span>🔴 利空 {neg} ({neg/total_s:.0%})</span>'
|
||
f'<span>⚪ 中性 {neu} ({neu/total_s:.0%})</span>'
|
||
f'</div>'
|
||
)
|
||
|
||
# 重要度
|
||
imp_rows = "".join(
|
||
f"<tr><td>等级 {k}</td><td>{v}</td></tr>"
|
||
for k, v in sorted(importances.items())
|
||
)
|
||
importance_table = f"<table><tr><th>重要度</th><th>数量</th></tr>{imp_rows}</table>"
|
||
|
||
# 事件类型
|
||
et_rows = "".join(
|
||
f"<tr><td>{k}</td><td>{v}</td></tr>"
|
||
for k, v in event_types.most_common(10)
|
||
)
|
||
event_type_table = f"<table><tr><th>事件类型</th><th>数量</th></tr>{et_rows}</table>"
|
||
|
||
# 文章来源
|
||
src_rows = "".join(
|
||
f"<tr><td>{_source_name(k)}</td><td>{v}</td></tr>"
|
||
for k, v in sources.most_common(15)
|
||
)
|
||
source_table = f"<table><tr><th>来源</th><th>文章数</th></tr>{src_rows}</table>"
|
||
|
||
return _HTML_TEMPLATE.format(
|
||
date=day_str,
|
||
generated_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
ai_summary=summary_html,
|
||
hi_threshold=hi_threshold,
|
||
high_count=len(high_events),
|
||
events_table=events_table,
|
||
raw_total=stats["raw_total"],
|
||
proc=stats["proc"],
|
||
deduped=stats["deduped"],
|
||
emb_count=stats["emb_count"],
|
||
qdrant_count=stats["qdrant_count"],
|
||
sentiment_section=sentiment_section,
|
||
importance_table=importance_table,
|
||
event_type_table=event_type_table,
|
||
source_table=source_table,
|
||
)
|
||
|
||
|
||
def _dedup_events(events: list[dict]) -> list[dict]:
|
||
"""事件级去重：同一 URL + 同一中文标题的事件合并。
|
||
|
||
合并策略：
|
||
- stock_codes 取并集
|
||
- importance / sentiment / summary_zh / event_type 保留 importance 更高的一条
|
||
- 按 importance 降序排列
|
||
|
||
Args:
|
||
events: 含 article 属性的事件列表
|
||
|
||
Returns:
|
||
去重合并后的事件列表
|
||
"""
|
||
groups: dict[tuple[str, str], dict] = {}
|
||
for ev in events:
|
||
article = ev.get("article", {})
|
||
url = article.get("url", "")
|
||
title_zh = article.get("title_zh", "") or article.get("title", "")
|
||
key = (url, title_zh)
|
||
|
||
if key in groups:
|
||
existing = groups[key]
|
||
# 合并 stock_codes
|
||
existing_codes = set(existing.get("stock_codes", []))
|
||
new_codes = set(ev.get("stock_codes", []))
|
||
existing["stock_codes"] = sorted(existing_codes | new_codes)
|
||
# 保留 importance 更高的事件详情
|
||
if ev.get("importance", 0) > existing.get("importance", 0):
|
||
for field in ("importance", "sentiment", "summary_zh", "event_type"):
|
||
if field in ev:
|
||
existing[field] = ev[field]
|
||
elif ev.get("importance", 0) == existing.get("importance", 0):
|
||
# 同等重要度：保留更长的 summary_zh（信息量更大）
|
||
if len(ev.get("summary_zh", "")) > len(existing.get("summary_zh", "")):
|
||
existing["summary_zh"] = ev["summary_zh"]
|
||
existing["event_type"] = ev.get("event_type", existing.get("event_type", ""))
|
||
else:
|
||
groups[key] = {
|
||
"importance": ev.get("importance", 0),
|
||
"sentiment": ev.get("sentiment", ""),
|
||
"summary_zh": ev.get("summary_zh", ""),
|
||
"event_type": ev.get("event_type", ""),
|
||
"stock_codes": sorted(ev.get("stock_codes", [])),
|
||
"article": ev.get("article", {}),
|
||
}
|
||
|
||
return sorted(groups.values(), key=lambda e: -e.get("importance", 0))
|
||
|
||
|
||
def _render_event_table(events: list[dict]) -> str:
|
||
"""渲染事件表格（M9 起弃用，保留以便回退）。"""
|
||
if not events:
|
||
return "<p>暂无符合条件的数据</p>"
|
||
|
||
rows: list[str] = []
|
||
for i, ev in enumerate(events, 1):
|
||
sentiment = ev.get("sentiment", "")
|
||
icon = {"positive": "🟢", "negative": "🔴", "neutral": "⚪"}.get(sentiment, "")
|
||
badge_cls = {"positive": "badge-pos", "negative": "badge-neg"}.get(
|
||
sentiment, "badge-neu"
|
||
)
|
||
imp = ev.get("importance", 0)
|
||
imp_cls = f"imp-{imp}" if imp >= 4 else ""
|
||
|
||
article = ev.get("article", {})
|
||
title = article.get("title_zh") or article.get("title", "")[:80]
|
||
url = article.get("url", "")
|
||
codes = ",".join(ev.get("stock_codes", [])[:5])
|
||
code_str = f" <small>[{codes}]</small>" if codes else ""
|
||
# 从 URL 域名推导来源名称，确保链接域名与来源标注一致
|
||
src_name = _url_source_label(url, article.get("source_id", ""))
|
||
|
||
cols = [
|
||
f"<td>{i}</td>",
|
||
f'<td><span class="badge {badge_cls}">{icon}</span></td>',
|
||
f'<td><a href="{url}" target="_blank">{title}</a>{code_str}</td>',
|
||
f'<td class="imp {imp_cls}">{imp}</td>',
|
||
f"<td>{ev.get('event_type', '')}</td>",
|
||
f"<td>{(ev.get('summary_zh', '') or '')[:80]} <small>[{src_name}]</small></td>",
|
||
]
|
||
rows.append(f'<tr class="event-row">{"".join(cols)}</tr>')
|
||
|
||
headers = ["#", "", "标题", "重要度", "事件类型", "摘要"]
|
||
header_row = "".join(f"<th>{h}</th>" for h in headers)
|
||
return f"<table><tr>{header_row}</tr>{''.join(rows)}</table>"
|