358 lines
13 KiB
Python
Executable File
358 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
手动执行新闻抓取和分析脚本
|
||
|
||
使用方法：
|
||
# 抓取所有新闻
|
||
python scripts/manual_news_fetch.py
|
||
|
||
# 只抓取加密货币新闻
|
||
python scripts/manual_news_fetch.py --category crypto
|
||
|
||
# 只抓取股票新闻
|
||
python scripts/manual_news_fetch.py --category stock
|
||
|
||
# 不使用 LLM 分析（仅关键词过滤）
|
||
python scripts/manual_news_fetch.py --no-llm
|
||
|
||
# 显示详细输出
|
||
python scripts/manual_news_fetch.py --verbose
|
||
"""
|
||
import asyncio
|
||
import argparse
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
|
||
# 添加项目路径
|
||
script_dir = Path(__file__).parent
|
||
project_root = script_dir.parent
|
||
backend_dir = project_root / "backend"
|
||
|
||
sys.path.insert(0, str(backend_dir))
|
||
|
||
# 切换到 backend 目录作为工作目录
|
||
os.chdir(backend_dir)
|
||
|
||
from app.utils.logger import logger
|
||
from app.news_agent.news_agent import get_news_agent
|
||
from app.news_agent.fetcher import NewsFetcher, NewsItem
|
||
from app.news_agent.filter import NewsDeduplicator, NewsFilter
|
||
from app.news_agent.analyzer import NewsAnalyzer, NewsAnalyzerSimple
|
||
from app.news_agent.news_db_service import get_news_db_service
|
||
from app.news_agent.notifier import get_news_notifier
|
||
|
||
|
||
async def main():
|
||
parser = argparse.ArgumentParser(description="手动执行新闻抓取和分析")
|
||
parser.add_argument(
|
||
"--category",
|
||
choices=["crypto", "stock"],
|
||
help="指定新闻分类（默认：全部）"
|
||
)
|
||
parser.add_argument(
|
||
"--no-llm",
|
||
action="store_true",
|
||
help="不使用 LLM 分析，仅使用关键词过滤"
|
||
)
|
||
parser.add_argument(
|
||
"--no-notify",
|
||
action="store_true",
|
||
help="不发送飞书通知"
|
||
)
|
||
parser.add_argument(
|
||
"--verbose",
|
||
action="store_true",
|
||
help="显示详细输出"
|
||
)
|
||
parser.add_argument(
|
||
"--limit",
|
||
type=int,
|
||
default=50,
|
||
help="每个源最多抓取数量（默认：50）"
|
||
)
|
||
parser.add_argument(
|
||
"--analyze-existing",
|
||
action="store_true",
|
||
help="分析数据库中未分析的文章（不抓取新新闻）"
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
print("=" * 70)
|
||
print("📰 新闻智能体 - 手动抓取模式")
|
||
print("=" * 70)
|
||
print(f"📅 时间: {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
if args.category:
|
||
print(f"📂 分类: {args.category}")
|
||
else:
|
||
print(f"📂 分类: 全部")
|
||
print(f"🤖 LLM 分析: {'禁用' if args.no_llm else '启用'}")
|
||
print(f"📢 通知: {'禁用' if args.no_notify else '启用'}")
|
||
print("=" * 70)
|
||
print()
|
||
|
||
# 初始化组件
|
||
fetcher = NewsFetcher()
|
||
deduplicator = NewsDeduplicator()
|
||
filter = NewsFilter()
|
||
db_service = get_news_db_service()
|
||
notifier = get_news_notifier()
|
||
|
||
# 选择分析器
|
||
if args.no_llm:
|
||
analyzer = NewsAnalyzerSimple()
|
||
print("使用规则分析器（关键词匹配）")
|
||
else:
|
||
analyzer = NewsAnalyzer()
|
||
print("使用 LLM 分析器")
|
||
|
||
try:
|
||
# 如果是分析已有文章模式
|
||
if args.analyze_existing:
|
||
print("\n" + "─" * 70)
|
||
print("📂 分析数据库中未分析的文章")
|
||
print("─" * 70)
|
||
|
||
unanalyzed = db_service.get_unanalyzed_articles(limit=args.limit, hours=48)
|
||
print(f"✅ 找到 {len(unanalyzed)} 条未分析的文章")
|
||
|
||
if not unanalyzed:
|
||
print("没有需要分析的文章")
|
||
return
|
||
|
||
# 转换为 NewsItem 对象
|
||
saved_articles = []
|
||
for article in unanalyzed:
|
||
# 创建临时 NewsItem
|
||
item = NewsItem(
|
||
title=article.title,
|
||
content=article.content or "",
|
||
url=article.url,
|
||
source=article.source,
|
||
category=article.category,
|
||
published_at=article.published_at,
|
||
crawled_at=article.crawled_at,
|
||
content_hash=article.content_hash,
|
||
author=article.author,
|
||
tags=article.tags
|
||
)
|
||
# 附加属性
|
||
item.quality_score = article.quality_score or 0.5
|
||
saved_articles.append((article, item))
|
||
|
||
else:
|
||
# 1. 抓取新闻
|
||
print("\n" + "─" * 70)
|
||
print("📡 第一步: 抓取新闻")
|
||
print("─" * 70)
|
||
|
||
items = await fetcher.fetch_all_news(category=args.category)
|
||
|
||
if not items:
|
||
print("❌ 没有获取到任何新闻")
|
||
return
|
||
|
||
print(f"✅ 获取到 {len(items)} 条新闻")
|
||
|
||
if args.verbose:
|
||
for i, item in enumerate(items[:10], 1):
|
||
print(f" {i}. [{item.category}] {item.title[:60]}...")
|
||
|
||
# 2. 去重
|
||
print("\n" + "─" * 70)
|
||
print("🔍 第二步: 去重")
|
||
print("─" * 70)
|
||
|
||
items = deduplicator.deduplicate_list(items)
|
||
print(f"✅ 去重后剩余 {len(items)} 条")
|
||
|
||
# 3. 过滤
|
||
print("\n" + "─" * 70)
|
||
print("⚙️ 第三步: 关键词过滤")
|
||
print("─" * 70)
|
||
|
||
filtered_items = filter.filter_news(items, min_quality=0.3)
|
||
print(f"✅ 过滤后剩余 {len(filtered_items)} 条")
|
||
|
||
if args.verbose and filtered_items:
|
||
print("\n通过过滤的新闻:")
|
||
for i, item in enumerate(filtered_items[:10], 1):
|
||
impact_score = getattr(item, 'impact_score', 0)
|
||
quality_score = getattr(item, 'quality_score', 0)
|
||
print(f" {i}. [{item.category}] 影响:{impact_score:.1f} 质量:{quality_score:.2f}")
|
||
print(f" {item.title[:60]}...")
|
||
|
||
# 4. 保存到数据库
|
||
print("\n" + "─" * 70)
|
||
print("💾 第四步: 保存到数据库")
|
||
print("─" * 70)
|
||
|
||
saved_articles = []
|
||
for item in filtered_items:
|
||
# 检查是否已存在
|
||
if db_service.check_duplicate_by_hash(item.content_hash):
|
||
if args.verbose:
|
||
print(f" ⊗ 已存在: {item.title[:50]}...")
|
||
continue
|
||
|
||
# 保存
|
||
article_data = {
|
||
'title': item.title,
|
||
'content': item.content,
|
||
'url': item.url,
|
||
'source': item.source,
|
||
'author': item.author,
|
||
'category': item.category,
|
||
'tags': item.tags,
|
||
'published_at': item.published_at,
|
||
'crawled_at': item.crawled_at,
|
||
'content_hash': item.content_hash,
|
||
'quality_score': getattr(item, 'quality_score', 0.5),
|
||
}
|
||
|
||
article = db_service.save_article(article_data)
|
||
if article:
|
||
saved_articles.append((article, item))
|
||
if args.verbose:
|
||
print(f" ✓ 已保存: {item.title[:50]}...")
|
||
|
||
if not args.analyze_existing:
|
||
print(f"✅ 保存了 {len(saved_articles)} 条新文章")
|
||
|
||
if not saved_articles:
|
||
print("\n没有新文章需要处理")
|
||
return
|
||
|
||
# 5. LLM 分析
|
||
print("\n" + "─" * 70)
|
||
print("🤖 第五步: AI 分析")
|
||
print("─" * 70)
|
||
|
||
analyzed_count = 0
|
||
high_priority_articles = []
|
||
|
||
# 批量分析
|
||
batch_size = 10 if not args.no_llm else 20
|
||
for i in range(0, len(saved_articles), batch_size):
|
||
batch = saved_articles[i:i + batch_size]
|
||
|
||
if args.no_llm:
|
||
# 规则分析
|
||
for article, item in batch:
|
||
result = analyzer.analyze_single(item)
|
||
priority = result.get('confidence', 50)
|
||
|
||
db_service.mark_as_analyzed(article.id, result, priority)
|
||
analyzed_count += 1
|
||
|
||
print(f" ✓ [{result['market_impact']}] {article.title[:50]}...")
|
||
print(f" 情绪: {result['sentiment']} | 相关: {', '.join(result.get('relevant_symbols', [])[:3])}")
|
||
|
||
if priority >= 40:
|
||
# 构造包含分析结果的完整字典
|
||
article_dict = article.to_dict()
|
||
article_dict.update({
|
||
'llm_analyzed': True,
|
||
'market_impact': result.get('market_impact'),
|
||
'impact_type': result.get('impact_type'),
|
||
'sentiment': result.get('sentiment'),
|
||
'summary': result.get('summary'),
|
||
'key_points': result.get('key_points'),
|
||
'trading_advice': result.get('trading_advice'),
|
||
'relevant_symbols': result.get('relevant_symbols'),
|
||
'priority': priority,
|
||
})
|
||
high_priority_articles.append(article_dict)
|
||
|
||
else:
|
||
# LLM 批量分析
|
||
items_to_analyze = [item for _, item in batch]
|
||
results = analyzer.analyze_batch(items_to_analyze)
|
||
|
||
for (article, _), result in zip(batch, results):
|
||
if result:
|
||
priority = analyzer.calculate_priority(
|
||
result,
|
||
getattr(article, 'quality_score', 0.5)
|
||
)
|
||
db_service.mark_as_analyzed(article.id, result, priority)
|
||
|
||
analyzed_count += 1
|
||
print(f" ✓ [{result['market_impact']}] {article.title[:50]}...")
|
||
print(f" {result.get('summary', 'N/A')[:60]}...")
|
||
|
||
if priority >= 40:
|
||
# 构造包含分析结果的完整字典
|
||
article_dict = article.to_dict()
|
||
# 更新为最新的分析结果
|
||
article_dict.update({
|
||
'llm_analyzed': True,
|
||
'market_impact': result.get('market_impact'),
|
||
'impact_type': result.get('impact_type'),
|
||
'sentiment': result.get('sentiment'),
|
||
'summary': result.get('summary'),
|
||
'key_points': result.get('key_points'),
|
||
'trading_advice': result.get('trading_advice'),
|
||
'relevant_symbols': result.get('relevant_symbols'),
|
||
'priority': priority,
|
||
})
|
||
high_priority_articles.append(article_dict)
|
||
|
||
print(f"\n✅ 分析了 {analyzed_count} 条文章")
|
||
print(f"🔥 发现 {len(high_priority_articles)} 条高优先级新闻")
|
||
|
||
# 6. 发送通知
|
||
if not args.no_notify and high_priority_articles:
|
||
print("\n" + "─" * 70)
|
||
print("📢 第六步: 发送通知")
|
||
print("─" * 70)
|
||
|
||
# 按优先级排序
|
||
high_priority_articles.sort(
|
||
key=lambda x: x.get('priority', 0),
|
||
reverse=True
|
||
)
|
||
|
||
# 如果只有1-2条，单独发送；否则批量发送
|
||
if len(high_priority_articles) <= 2:
|
||
for article in high_priority_articles:
|
||
await notifier.notify_single_news(article)
|
||
db_service.mark_as_notified(article['id'])
|
||
print(f" ✓ 已发送: {article['title'][:50]}...")
|
||
else:
|
||
await notifier.notify_news_batch(high_priority_articles[:10])
|
||
for article in high_priority_articles[:10]:
|
||
db_service.mark_as_notified(article['id'])
|
||
print(f" ✓ 已发送批量通知 ({len(high_priority_articles[:10])} 条)")
|
||
|
||
# 7. 统计报告
|
||
print("\n" + "=" * 70)
|
||
print("📊 处理完成 - 统计报告")
|
||
print("=" * 70)
|
||
|
||
db_stats = db_service.get_stats(hours=24)
|
||
print(f"📰 最近24小时:")
|
||
print(f" • 总文章数: {db_stats['total_articles']}")
|
||
print(f" • 已分析: {db_stats['analyzed']}")
|
||
print(f" • 高影响: {db_stats['high_impact']}")
|
||
print(f" • 已通知: {db_stats['notified']}")
|
||
|
||
print("\n" + "=" * 70)
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 执行失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return 1
|
||
|
||
finally:
|
||
await fetcher.close()
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
exit_code = asyncio.run(main())
|
||
sys.exit(exit_code)
|