simon
d4a55bcaaa
- 新增 report_db/ 包（models/schema/db，复用 news 项目实现，幂等 upsert） - reporter.py: _build_report_data + generate_report 写库返回 report_id - pipeline/cli 适配 report_id 返回值；HTML 渲染/上传保留 deprecated - 新增 tests/test_report_db.py；.env.example 增加 NEWS_DB_* 配置 - 已部署 pi5 并验证：真实生成 report_id=184（12 事件）+ 幂等覆盖
257 lines
7.8 KiB
Python
257 lines
7.8 KiB
Python
"""English Financial News — CLI 入口"""
|
||
|
||
import logging
|
||
import sys
|
||
|
||
import typer
|
||
from dotenv import load_dotenv
|
||
|
||
# 加载 .env 环境变量（必须在所有模块导入之前）
|
||
load_dotenv()
|
||
|
||
app = typer.Typer(
|
||
name="en-news",
|
||
help="国际财经新闻抓取与深度研究平台",
|
||
no_args_is_help=True,
|
||
)
|
||
|
||
# ── 日志配置 ──────────────────────────────────────────
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S",
|
||
stream=sys.stderr,
|
||
)
|
||
|
||
# 抑制第三方库噪音日志
|
||
for _lib in ("httpx", "httpcore", "openai", "urllib3"):
|
||
logging.getLogger(_lib).setLevel(logging.WARNING)
|
||
|
||
logger = logging.getLogger("en-news")
|
||
|
||
|
||
@app.command()
|
||
def crawl(
|
||
source: str | None = typer.Option(
|
||
None, "--source", "-s",
|
||
help="只抓取指定 source_id（不传则全部）",
|
||
),
|
||
profile: str | None = typer.Option(
|
||
None, "--profile", "-p",
|
||
help="Profile 名称（2g_headless / 8g_headful）",
|
||
),
|
||
):
|
||
"""M1: 抓取英文财经新闻（部署海外）"""
|
||
import os
|
||
|
||
from crawler.orchestrator import run_crawl_sync
|
||
|
||
if profile:
|
||
os.environ["EN_NEWS_PROFILE"] = profile
|
||
logger.info("📋 使用 Profile: %s", profile)
|
||
|
||
try:
|
||
stats = run_crawl_sync(source_filter=source)
|
||
typer.echo(f"\n✅ 完成: {stats.sources_crawled} 源, {stats.total_articles} 篇文章")
|
||
if stats.sources_failed:
|
||
typer.echo(f"⚠️ {stats.sources_failed} 个源有错误")
|
||
except FileNotFoundError as e:
|
||
typer.echo(f"❌ {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
except Exception as e:
|
||
logger.exception("抓取失败")
|
||
typer.echo(f"❌ 抓取出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def extract(
|
||
source: str | None = typer.Option(
|
||
None, "--source", "-s",
|
||
help="只处理指定 source_id（不传则全部）",
|
||
),
|
||
):
|
||
"""M2: 英文正文提取（trafilatura）"""
|
||
from extractor.pipeline import process_all_sources
|
||
|
||
try:
|
||
stats = process_all_sources(source_filter=source)
|
||
typer.echo(
|
||
f"\n✅ 提取: {stats['sources_processed']} 源, "
|
||
f"{stats['total_articles']} 篇, {stats['elapsed_sec']:.1f}s"
|
||
)
|
||
except Exception as e:
|
||
logger.exception("正文提取失败")
|
||
typer.echo(f"❌ 提取出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def dedup():
|
||
"""M3: 三层去重"""
|
||
from dedup.pipeline import dedup_all_sources
|
||
|
||
try:
|
||
stats = dedup_all_sources()
|
||
typer.echo(
|
||
f"\n✅ 去重: {stats['sources_processed']} 源, "
|
||
f"唯一 {stats['unique']} / 重复 {stats['duplicate']} / "
|
||
f"总计 {stats['total_articles']}, {stats['elapsed_sec']:.1f}s"
|
||
)
|
||
except Exception as e:
|
||
logger.exception("去重失败")
|
||
typer.echo(f"❌ 去重出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def translate():
|
||
"""M4: 全文翻译 + 投资事件抽取（LLM）"""
|
||
from llm.pipeline import translate_all_deduped
|
||
|
||
try:
|
||
stats = translate_all_deduped()
|
||
typer.echo(
|
||
f"\n✅ 翻译+事件抽取: {stats['success']}/{stats['total']} 篇, "
|
||
f"{stats['elapsed_sec']:.1f}s ({stats['provider']}/{stats['model']})"
|
||
)
|
||
except Exception as e:
|
||
logger.exception("翻译失败")
|
||
typer.echo(f"❌ 翻译出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def embed():
|
||
"""M5: 向量生成"""
|
||
from embedding.pipeline import embed_all_events
|
||
|
||
try:
|
||
stats = embed_all_events()
|
||
typer.echo(
|
||
f"\n✅ 向量生成: {stats['success']}/{stats['total']} 篇, "
|
||
f"{stats['elapsed_sec']:.1f}s ({stats['provider']}/{stats['model']})"
|
||
)
|
||
except Exception as e:
|
||
logger.exception("向量生成失败")
|
||
typer.echo(f"❌ 向量生成出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def index(
|
||
recreate: bool = typer.Option(
|
||
False, "--recreate",
|
||
help="重建 collection（会删除已有数据）",
|
||
),
|
||
):
|
||
"""M6: Qdrant 入库"""
|
||
from vectorstore.pipeline import get_collection_info, ingest_all_embeddings
|
||
|
||
try:
|
||
stats = ingest_all_embeddings(recreate=recreate)
|
||
typer.echo(
|
||
f"\n✅ 入库: {stats['ingested']}/{stats['total']} 条, "
|
||
f"{stats['elapsed_sec']:.1f}s"
|
||
)
|
||
info = get_collection_info()
|
||
typer.echo(f"📊 Collection: {info['name']} — {info['vectors_count']} 条向量")
|
||
except Exception as e:
|
||
logger.exception("入库失败")
|
||
typer.echo(f"❌ 入库出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def search(
|
||
query: str = typer.Argument(..., help="中文检索查询"),
|
||
top_k: int = typer.Option(10, help="返回结果数"),
|
||
):
|
||
"""M6: Qdrant 语义检索"""
|
||
from vectorstore.pipeline import search_news
|
||
|
||
try:
|
||
results = search_news(query, top_k=top_k)
|
||
if not results:
|
||
typer.echo("🔍 未找到相关结果")
|
||
return
|
||
|
||
typer.echo(f"\n🔍 搜索: {query}\n")
|
||
for i, r in enumerate(results, 1):
|
||
typer.echo(f"{i}. [{r.source_id}] score={r.score:.4f}")
|
||
typer.echo(f" 📰 {r.title_zh or r.title}")
|
||
if r.events:
|
||
events_summary = ", ".join(
|
||
f"{ev.get('event_type','')}({ev.get('sentiment','')})"
|
||
for ev in r.events[:3]
|
||
)
|
||
typer.echo(f" 📌 事件: {events_summary}")
|
||
typer.echo(f" 🔗 {r.url}")
|
||
typer.echo()
|
||
except Exception as e:
|
||
logger.exception("搜索失败")
|
||
typer.echo(f"❌ 搜索出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def report():
|
||
"""M7: 日报生成（结构化入库，不再产出 HTML）"""
|
||
from scheduler.reporter import generate_report
|
||
|
||
try:
|
||
report_id = generate_report()
|
||
if report_id is not None:
|
||
typer.echo(f"\n✅ 日报已入库: report_id={report_id}")
|
||
else:
|
||
typer.echo("⚠️ 无数据或入库失败，跳过日报生成")
|
||
except Exception as e:
|
||
logger.exception("日报生成失败")
|
||
typer.echo(f"❌ 日报生成出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def pipeline(
|
||
skip_report: bool = typer.Option(
|
||
False, "--skip-report",
|
||
help="跳过日报生成",
|
||
),
|
||
):
|
||
"""M7: 一键运行完整管道 M2→M6（+ 可选日报）"""
|
||
from crawler.utils import get_news_day
|
||
from scheduler.pipeline import run_pipeline
|
||
|
||
date_str = get_news_day()
|
||
typer.echo(f"🚀 开始全链路管道，日期: {date_str}\n")
|
||
|
||
try:
|
||
result = run_pipeline(date_str, skip_report=skip_report)
|
||
|
||
typer.echo(f"\n{'='*50}")
|
||
typer.echo(f"Pipeline 完成: {result.success_count}/{len(result.steps)} 成功")
|
||
for s in result.steps:
|
||
flag = "✅" if s.success else "❌"
|
||
typer.echo(f" {flag} {s.name}: {s.message} ({s.elapsed_sec:.0f}s)")
|
||
if result.finished_at and result.started_at:
|
||
total = (result.finished_at - result.started_at).total_seconds()
|
||
typer.echo(f"\n⏱ 总耗时: {total:.0f}s")
|
||
except Exception as e:
|
||
logger.exception("管道执行失败")
|
||
typer.echo(f"❌ 管道执行出错: {e}", err=True)
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def mcp_server():
|
||
"""M8: 启动 MCP 服务（供 Claude Code / Cherry Studio 调用）"""
|
||
from mcp_server.server import mcp
|
||
|
||
typer.echo("🚀 启动国际财经 Deep Research MCP 服务...")
|
||
mcp.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app()
|