197 lines
7.0 KiB
Python
197 lines
7.0 KiB
Python
"""
|
||
资讯子图API调用工具
|
||
News Analysis API Client
|
||
支持 async 和真实数据库缓存
|
||
"""
|
||
|
||
from typing import Dict, Any, Optional, List
|
||
import random
|
||
from datetime import datetime
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass
|
||
class NewsAPIClient:
|
||
"""
|
||
资讯API客户端 - 可扩展支持多种API和数据库缓存
|
||
"""
|
||
|
||
# 可以配置多个API（如 NewsAPI, 今日头条, 百度新闻等）
|
||
newsapi_key: Optional[str] = None
|
||
|
||
# 数据库 Repository（可选，用于缓存新闻）
|
||
news_repository: Optional[Any] = None
|
||
|
||
async def query_news_db(self, user_id: str, keyword: str) -> Optional[List[Dict[str, Any]]]:
|
||
"""从数据库缓存查询新闻"""
|
||
if not self.news_repository:
|
||
return None
|
||
try:
|
||
entities = await self.news_repository.search_by_keywords(user_id, keyword)
|
||
if entities:
|
||
return [
|
||
{
|
||
"title": e.title,
|
||
"source": e.source,
|
||
"summary": e.content,
|
||
"keywords": e.keywords.split(",") if e.keywords else [],
|
||
"author": "",
|
||
"published_at": e.created_at
|
||
}
|
||
for e in entities
|
||
]
|
||
except Exception as e:
|
||
print(f"从数据库查询新闻失败：{e}")
|
||
return None
|
||
|
||
async def cache_news_db(self, user_id: str, news: Dict[str, Any]):
|
||
"""把新闻缓存到数据库"""
|
||
if not self.news_repository:
|
||
return
|
||
try:
|
||
from ...db.models import NewsEntity
|
||
entity = NewsEntity(
|
||
user_id=user_id,
|
||
title=news.get("title", ""),
|
||
content=news.get("summary", ""),
|
||
url=news.get("url", ""),
|
||
source=news.get("source", ""),
|
||
keywords=",".join(news.get("keywords", []))
|
||
)
|
||
await self.news_repository.insert(entity)
|
||
except Exception as e:
|
||
print(f"缓存新闻到数据库失败：{e}")
|
||
|
||
def query_news_mock(self, query: str) -> List[Dict[str, Any]]:
|
||
"""
|
||
模拟查询资讯 - 目前用于演示
|
||
"""
|
||
# 模拟资讯数据库
|
||
mock_news = [
|
||
{
|
||
"title": "OpenAI发布GPT-5：智能再升级",
|
||
"source": "Tech News",
|
||
"summary": "最新消息，OpenAI刚刚发布了GPT-5模型，智能水平再次取得重大突破...",
|
||
"keywords": ["AI", "GPT-5", "OpenAI"],
|
||
"author": "AI Team",
|
||
"published_at": datetime.now().isoformat()
|
||
},
|
||
{
|
||
"title": "大模型在医疗领域的应用",
|
||
"source": "Health Tech",
|
||
"summary": "大模型AI技术正在医疗领域展现巨大潜力，从辅助诊断到药物研发...",
|
||
"keywords": ["医疗", "大模型", "应用"],
|
||
"author": "Medical Team",
|
||
"published_at": datetime.now().isoformat()
|
||
},
|
||
{
|
||
"title": "2026年AI行业发展趋势报告",
|
||
"source": "Business Daily",
|
||
"summary": "最新行业报告显示，AI行业将继续保持高速增长，企业数字化转型加速...",
|
||
"keywords": ["趋势", "AI", "商业"],
|
||
"author": "Business Team",
|
||
"published_at": datetime.now().isoformat()
|
||
}
|
||
]
|
||
|
||
# 根据查询词简单过滤
|
||
results = []
|
||
query_lower = query.lower()
|
||
|
||
for news in mock_news:
|
||
if (query_lower in news["title"].lower() or
|
||
query_lower in news["summary"].lower() or
|
||
any(keyword.lower() in query_lower for keyword in news["keywords"])):
|
||
results.append(news)
|
||
|
||
# 如果没有匹配到，返回前两条
|
||
if not results:
|
||
results = mock_news[:2]
|
||
|
||
return results
|
||
|
||
def analyze_url_mock(self, url: str) -> Dict[str, Any]:
|
||
"""
|
||
模拟URL分析 - 目前用于演示
|
||
"""
|
||
return {
|
||
"title": f"分析结果：{url}",
|
||
"source": "URL Analyzer",
|
||
"summary": "已完成对该URL的内容分析，包含文章摘要和情感倾向判断...",
|
||
"keywords": ["News", "Analysis", url.split("/")[-1] if url else "unknown"]
|
||
}
|
||
|
||
def extract_keywords_mock(self, text: str) -> List[str]:
|
||
"""
|
||
模拟关键词提取 - 目前用于演示
|
||
"""
|
||
# 简单的关键词提取模拟
|
||
common_keywords = ["AI", "大模型", "应用场景", "行业趋势", "创新", "技术"]
|
||
result = []
|
||
|
||
for keyword in common_keywords:
|
||
if keyword.lower() in text.lower():
|
||
result.append(keyword)
|
||
|
||
# 如果没找到，返回默认关键词
|
||
if not result:
|
||
result = ["AI", "大模型", "应用场景", "行业趋势"]
|
||
|
||
return result
|
||
|
||
def generate_report_mock(self, query: str) -> str:
|
||
"""
|
||
模拟报告生成 - 目前用于演示
|
||
"""
|
||
report = f"""═══════════════════════════════════════════
|
||
📊 资讯分析报告
|
||
═══════════════════════════════════════════
|
||
|
||
主题：{query}
|
||
|
||
📋 摘要：
|
||
这是一份关于 {query} 的资讯分析综合报告，包含最新行业动态和趋势分析。
|
||
|
||
🔍 主要发现：
|
||
1. AI技术持续快速发展
|
||
2. 大模型应用场景不断拓展
|
||
3. 行业数字化转型加速
|
||
|
||
🏷️ 关键词：
|
||
- AI
|
||
- 大模型
|
||
- 数字化转型
|
||
- 创新
|
||
|
||
═══════════════════════════════════════════
|
||
💡 建议：继续关注行业动态，把握发展机遇！
|
||
"""
|
||
return report
|
||
|
||
# ========== 统一入口（优先查缓存） ==========
|
||
async def query_news(self, user_id: str = "default", query: str = "", use_cache: bool = True) -> List[Dict[str, Any]]:
|
||
"""查询新闻（统一入口，优先查数据库缓存）"""
|
||
# 1. 先查数据库缓存
|
||
if use_cache:
|
||
cached = await self.query_news_db(user_id, query)
|
||
if cached:
|
||
return cached
|
||
|
||
# 2. 查第三方 API（暂未实现）
|
||
# api_result = await self.query_news_api(query)
|
||
# if api_result:
|
||
# for news in api_result:
|
||
# await self.cache_news_db(user_id, news)
|
||
# return api_result
|
||
|
||
# 3. 用模拟数据（兜底）
|
||
mock_result = self.query_news_mock(query)
|
||
if use_cache:
|
||
for news in mock_result:
|
||
await self.cache_news_db(user_id, news)
|
||
return mock_result
|
||
|
||
|
||
# 单例实例（模拟模式，保持向后兼容）
|
||
news_api = NewsAPIClient()
|