"""每日 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 = """
国际财经 Deep Research 日报 — {date}
一、🤖 AI 摘要
{ai_summary}
二、🔥 重要事件 (importance ≥ {hi_threshold}, {high_count} 条)
{events_table}
三、📊 数据总览
3.1 M1→M6 管道
3.2 情绪分布 (当日事件)
{sentiment_section}
3.3 重要度分布
{importance_table}
3.4 事件类型 TOP 10
{event_type_table}
3.5 文章来源分布
{source_table}
"""
def _md_to_html(text: str) -> str:
"""Markdown → HTML（使用 python-markdown，开启常用扩展）。"""
if not text.strip():
return ""
return markdown.markdown(
text,
extensions=["nl2br"], # 单换行 →
)
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 "暂无 AI 摘要
"
# 事件表格
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''
f''
f'🟢 利好 {pos} ({pos/total_s:.0%})'
f'🔴 利空 {neg} ({neg/total_s:.0%})'
f'⚪ 中性 {neu} ({neu/total_s:.0%})'
f'
'
)
# 重要度
imp_rows = "".join(
f"| 等级 {k} | {v} |
"
for k, v in sorted(importances.items())
)
importance_table = f""
# 事件类型
et_rows = "".join(
f"| {k} | {v} |
"
for k, v in event_types.most_common(10)
)
event_type_table = f""
# 文章来源
src_rows = "".join(
f"| {_source_name(k)} | {v} |
"
for k, v in sources.most_common(15)
)
source_table = f""
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 "暂无符合条件的数据
"
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" [{codes}]" if codes else ""
# 从 URL 域名推导来源名称，确保链接域名与来源标注一致
src_name = _url_source_label(url, article.get("source_id", ""))
cols = [
f"{i} | ",
f'{icon} | ',
f'{title}{code_str} | ',
f'{imp} | ',
f"{ev.get('event_type', '')} | ",
f"{(ev.get('summary_zh', '') or '')[:80]} [{src_name}] | ",
]
rows.append(f'{"".join(cols)}
')
headers = ["#", "", "标题", "重要度", "事件类型", "摘要"]
header_row = "".join(f"{h} | " for h in headers)
return f"{header_row}
{''.join(rows)}
"