"""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")