210 lines
7.6 KiB
Python
210 lines
7.6 KiB
Python
"""Use DeepSeek to summarize top news into categorized Chinese briefs (5 per category).
|
||
v2: back-fills pub_time from raw RSS data after AI summarization.
|
||
"""
|
||
import json, os, requests, re
|
||
from datetime import datetime, timezone, timedelta
|
||
from dateutil import parser as dtparser
|
||
|
||
API_KEY = "sk-bbca4a0380d549389f0d27cdea0b5228"
|
||
API_URL = "https://api.deepseek.com/v1/chat/completions"
|
||
MODEL = "deepseek-v4-flash"
|
||
|
||
NEWS_FILE = "/var/www/nav/data/news.json"
|
||
OUTPUT = "/var/www/nav/data/news_summary.json"
|
||
tz = timezone(timedelta(hours=8))
|
||
|
||
|
||
def load_raw_news():
|
||
"""Load raw news and build title->published lookup."""
|
||
if not os.path.exists(NEWS_FILE):
|
||
return []
|
||
with open(NEWS_FILE) as f:
|
||
data = json.load(f)
|
||
return data.get("items", [])
|
||
|
||
|
||
def build_time_lookup(raw_items):
|
||
"""Build lookup: title -> pub_time (Beijing time string like '06-15 10:30')."""
|
||
lookup = {}
|
||
for item in raw_items:
|
||
title = item.get("title", "").strip()
|
||
published = item.get("published", "").strip()
|
||
if not title or not published:
|
||
continue
|
||
try:
|
||
dt = dtparser.parse(published)
|
||
# Convert to Beijing time if not already
|
||
if dt.tzinfo:
|
||
dt = dt.astimezone(tz)
|
||
else:
|
||
dt = dt.replace(tzinfo=tz)
|
||
# Format: MM-DD HH:MM for today, MM-DD for older
|
||
now = datetime.now(tz)
|
||
if dt.date() == now.date():
|
||
time_str = dt.strftime("%H:%M")
|
||
else:
|
||
time_str = dt.strftime("%m-%d %H:%M")
|
||
lookup[title] = time_str
|
||
except Exception:
|
||
continue
|
||
return lookup
|
||
|
||
|
||
def match_pub_time(items, lookup):
|
||
"""Match AI output items back to raw RSS pub_time by title similarity."""
|
||
for item in items:
|
||
if "pub_time" in item:
|
||
continue
|
||
title = item.get("title", "").strip()
|
||
subtitle = item.get("subtitle", "").strip()
|
||
matched = None
|
||
# Exact match on original title first
|
||
if title in lookup:
|
||
matched = lookup[title]
|
||
else:
|
||
# Fuzzy: check if subtitle contains original title keywords
|
||
for raw_title, time_val in lookup.items():
|
||
# Try partial match
|
||
short = min(len(raw_title), len(title))
|
||
if short > 10 and raw_title[:short] in title or title[:short] in raw_title:
|
||
matched = time_val
|
||
break
|
||
# Check if first 20 chars match
|
||
if raw_title[:20].strip() == title[:20].strip():
|
||
matched = time_val
|
||
break
|
||
if matched:
|
||
item["pub_time"] = matched
|
||
|
||
|
||
def summarize(items):
|
||
news_text = ""
|
||
for i, item in enumerate(items[:200], 1):
|
||
title = item.get("title", "")
|
||
summary = item.get("summary", "")[:150].replace("<", "").replace(">", "")
|
||
source = item.get("source", "")
|
||
link = item.get("link", "")
|
||
news_text += f"{i}. [{source}] {title}\n {summary}\n url: {link}\n\n"
|
||
|
||
prompt = f"""你是一个资深的新闻编辑助手。你的任务是精选出当下最重要的新闻，按分类组织。
|
||
|
||
## 选稿标准（满足任一即可）
|
||
- 地缘政治新闻（战争、条约、制裁、外交、选举）
|
||
- 宏观经济决策（央行、通胀、贸易、就业）
|
||
- 科技重大突破（AI、航天、医药、能源）
|
||
- 重大自然灾害/公共卫生事件
|
||
- **中国官方媒体头条**（新华社、人民日报的置顶/头条报道——这些非常重要，必须收录）
|
||
- 中国国内引发广泛讨论的社会事件
|
||
- 任何具有长期影响的政策变化
|
||
|
||
## 分类（每个分类**至少15条，最多30条**）
|
||
- 国际要闻
|
||
- 科技前沿
|
||
- 财经动向
|
||
- 中国热点（必须包含新华社、人民日报的重要报道）
|
||
- 社会生活
|
||
|
||
## 输出格式
|
||
每条输出到 items 数组，包含以下字段：
|
||
- subtitle：**15-30字完整标题**（不要缩写）
|
||
- title：原标题（必须与原始新闻标题完全一致，这是时间匹配的关键）
|
||
- source：来源
|
||
- link：原文链接（如不可用则留空）
|
||
- summary：**300-500字完整新闻摘要**，用自然段落叙述，关键信息用**加粗**标出。不在前面写"核心事实""背景分析"等前缀，直接写内容，段落清晰流畅。
|
||
|
||
## 原始新闻
|
||
{news_text}
|
||
|
||
请输出JSON格式。注意：宁多勿少，确保每个分类**至少15条，最多30条**。
|
||
|
||
JSON格式: {{"categories": [{{"name":"分类名","items":[{{"rank":1,"subtitle":"小标题","title":"原标题","source":"来源","link":"https://...","summary":"正文..."}}]}}]}}"""
|
||
|
||
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.3,
|
||
"max_tokens": 40000,
|
||
},
|
||
timeout=180,
|
||
)
|
||
|
||
if resp.status_code != 200:
|
||
print(f"API error: {resp.status_code} {resp.text[:200]}")
|
||
return None
|
||
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
content = content.strip()
|
||
if content.startswith("```"):
|
||
lines = content.split("\n")
|
||
content = "\n".join(lines[1:-1]) if lines[-1].strip() == "```" else "\n".join(lines[1:])
|
||
|
||
try:
|
||
result = json.loads(content)
|
||
except:
|
||
content = content.strip()
|
||
content = re.sub(r',\s*}', '}', content)
|
||
content = re.sub(r',\s*]', ']', content)
|
||
for attempt in range(3):
|
||
try:
|
||
result = json.loads(content)
|
||
break
|
||
except:
|
||
if attempt == 0:
|
||
idx = content.rfind('{"rank"')
|
||
if idx > 0:
|
||
content = content[:content.rfind('}', 0, idx)] + '}]}]}'
|
||
elif attempt == 1:
|
||
lines = content.split('\n')
|
||
fixed = []
|
||
for line in lines:
|
||
if line.strip().endswith(',') and not any(x in line for x in ['"', ']', '}']):
|
||
line = line.rstrip(',')
|
||
fixed.append(line)
|
||
content = '\n'.join(fixed)
|
||
else:
|
||
print(f"JSON parse failed, len={len(content)}")
|
||
print(f"First 800 chars: {content[:800]}")
|
||
return None
|
||
|
||
return result
|
||
|
||
|
||
if __name__ == "__main__":
|
||
now = datetime.now(tz).strftime("%Y-%m-%d %H:%M")
|
||
raw_items = load_raw_news()
|
||
if not raw_items:
|
||
print("No news data")
|
||
exit(0)
|
||
|
||
print(f"Loaded {len(raw_items)} raw news items. AI summarizing by category...")
|
||
data = summarize(raw_items)
|
||
|
||
if not data:
|
||
print("Summarization failed")
|
||
exit(1)
|
||
|
||
# Back-fill pub_time from raw RSS data
|
||
time_lookup = build_time_lookup(raw_items)
|
||
matched = 0
|
||
for cat in data.get("categories", []):
|
||
match_pub_time(cat["items"], time_lookup)
|
||
matched += sum(1 for item in cat["items"] if "pub_time" in item)
|
||
print(f"Matched pub_time for {matched} articles")
|
||
|
||
result = {
|
||
"updated_at": now,
|
||
"categories": data.get("categories", []),
|
||
}
|
||
|
||
total = sum(len(c["items"]) for c in result["categories"])
|
||
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"Written {total} articles across {len(result['categories'])} categories")
|
||
for c in result["categories"]:
|
||
print(f" {c['name']}: {len(c['items'])} items")
|