106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
"""Deep dive: pick top news and generate in-depth analysis via DeepSeek."""
|
|
import json, os, requests
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
API_KEY = "sk-bbca4a0380d549389f0d27cdea0b5228"
|
|
API_URL = "https://api.deepseek.com/v1/chat/completions"
|
|
MODEL = "deepseek-v4-flash"
|
|
|
|
NEWS_SUMMARY = "/var/www/nav/data/news_summary.json"
|
|
OUTPUT = "/var/www/nav/data/deep_dive.json"
|
|
tz = timezone(timedelta(hours=8))
|
|
|
|
|
|
def load_top_news():
|
|
if not os.path.exists(NEWS_SUMMARY):
|
|
return None
|
|
with open(NEWS_SUMMARY) as f:
|
|
data = json.load(f)
|
|
# Pick the #1 item from the first category (usually 国际要闻 or most important)
|
|
for cat in data.get("categories", []):
|
|
if cat.get("items"):
|
|
return cat["items"][0], cat["name"]
|
|
return None, None
|
|
|
|
|
|
def generate_deep_dive(top_item, category):
|
|
title = top_item.get("title", "")
|
|
subtitle = top_item.get("subtitle", "")
|
|
summary = top_item.get("summary", "")
|
|
source = top_item.get("source", "")
|
|
link = top_item.get("link", "")
|
|
|
|
prompt = f"""你是一位资深的国际事务/财经分析员。用户看到了一条新闻的简要摘要，请你写一篇深度解读，帮助用户理解这条新闻的**完整背景、深层含义和未来走向**。
|
|
|
|
## 新闻信息
|
|
- 分类：{category}
|
|
- 标题：{subtitle}
|
|
- 原文标题：{title}
|
|
- 来源：{source}
|
|
- 原文链接：{link}
|
|
- 简要摘要：{summary}
|
|
|
|
## 写作要求
|
|
1. **800-1200 字**，中文，客观专业
|
|
2. **结构清晰**：事件背景 → 关键事实 → 各方立场 → 深层分析 → 可能走向 → 对中国/市场的影响
|
|
3. **不重复摘要内容**，要在摘要基础上做增量分析
|
|
4. **引用具体数据/事件**支撑论点，不要空泛议论
|
|
5. **结尾一句话总结**这条新闻对用户（中国中层干部、量化交易者）的核心启示
|
|
6. 关键信息用 **加粗** 标出
|
|
|
|
请直接输出正文，不要输出 JSON。"""
|
|
|
|
resp = requests.post(
|
|
API_URL,
|
|
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
|
|
json={
|
|
"model": MODEL,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": 0.5,
|
|
"max_tokens": 4000,
|
|
},
|
|
timeout=120,
|
|
)
|
|
|
|
if resp.status_code != 200:
|
|
print(f"API error: {resp.status_code}")
|
|
return None
|
|
|
|
content = resp.json()["choices"][0]["message"]["content"]
|
|
return content.strip()
|
|
|
|
|
|
def run():
|
|
top_item, category = load_top_news()
|
|
if not top_item:
|
|
print("No news data available for deep dive")
|
|
return
|
|
|
|
print(f"Generating deep dive for: {top_item.get('subtitle', '')[:50]}...")
|
|
|
|
analysis = generate_deep_dive(top_item, category)
|
|
if not analysis:
|
|
print("Deep dive generation failed")
|
|
return
|
|
|
|
result = {
|
|
"updated_at": datetime.now(tz).strftime("%Y-%m-%d %H:%M"),
|
|
"source_news": {
|
|
"subtitle": top_item.get("subtitle", ""),
|
|
"title": top_item.get("title", ""),
|
|
"link": top_item.get("link", ""),
|
|
"category": category,
|
|
},
|
|
"analysis": analysis,
|
|
}
|
|
|
|
os.makedirs(os.path.dirname(OUTPUT), exist_ok=True)
|
|
with open(OUTPUT, "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"Deep dive written -> {OUTPUT}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|