main
1#!/venv/bin/python
2# -*- coding: utf-8 -*-
3import json
4import tempfile
5from collections import defaultdict
6from contextlib import suppress
7from datetime import datetime, timedelta
8from pathlib import Path
9from zoneinfo import ZoneInfo
10
11import feedparser
12from feedgen.feed import FeedGenerator
13from glom import glom
14from loguru import logger
15from pyrogram.types import Chat, Message
16from pyrogram.types.messages_and_media.message import Str
17
18from ai.main import ai_text_generation
19from config import DB, PREFIX, TZ
20from database.d1 import query_d1
21from database.r2 import get_cf_r2, head_cf_r2, set_cf_r2
22from utils import convert2html, nowdt, rand_number
23
24DAILY_KEY = "AI-NEWS/daily.xml"
25WEEKLY_KEY = "AI-NEWS/weekly.xml"
26
27
28JSON_SCHEMA = {
29 "title": "List of News Summary",
30 "type": "array",
31 "items": {
32 "type": "object",
33 "title": "News",
34 "properties": {
35 "category": {"description": "News category", "enum": ["Model", "Tech", "Finance", "Policy", "Application", "Industry", "Others"], "title": "Category", "type": "string"},
36 "abstract": {"description": "A concise summary (50-100 words) in Chinese. Do NOT include URL strings here. Focus on facts and impact.", "title": "Abstract", "type": "string"},
37 "title": {"description": "A concise, objective Chinese title (max 20 chars). Remove clickbait words.", "title": "Title", "type": "string"},
38 "urls": {"description": "Select 1-2 most authoritative source URLs.", "items": {"type": "string"}, "title": "Urls", "type": "array"},
39 },
40 "required": ["category", "title", "abstract", "urls"],
41 "additionalProperties": False,
42 },
43}
44
45
46async def daily_ainews(num_days: int = 7):
47 """Get daily AI summary (num_days)."""
48 parsed = await parse_finished(DAILY_KEY)
49 if not parsed:
50 logger.warning("No Daily AI news parsed.")
51 return
52 now = nowdt(TZ)
53 begin = now - timedelta(days=num_days)
54 items = {}
55 for i in range(num_days):
56 day = begin + timedelta(days=i)
57 day_str = day.strftime("%Y-%m-%d")
58 if day_str in parsed:
59 continue
60 news_summary = await get_summary(day_str, day_str)
61 if not news_summary:
62 continue
63 await set_cf_r2(key=f"AI-NEWS/daily/{day_str}-summary.json", data=news_summary) # save news summary
64 # publish the summary
65 title = f"AI日报 | {day_str}"
66 html = convert_news_summary_to_html(news_summary)
67 full_html = f'<!DOCTYPE html><html><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>{title}</title><link rel="stylesheet" href="https://cdn.jsdmirror.cn/npm/water.css@2/out/water.css" integrity="sha512-GW7j11RXZmdio87hQsKNjomKEy/DwDjh6J2Z1JnI5Z4FNP791QfZP7Iut25vA+L+YJLZipI2BZhEpkvBkfr8cw==" crossorigin="anonymous"></head><body><h1>{title}</h1>{html}</body></html>'
68 html_key = f"AI-NEWS/daily/{day_str}.html"
69 await set_cf_r2(key=html_key, data=full_html, mime_type="text/html")
70 items[day_str] = {"title": title, "summary": html, "url": f"{DB.CF_R2_PUBLIC_URL}/{html_key}"}
71 if not items:
72 return
73 items |= parsed
74 fg = FeedGenerator()
75 fg.link(href=f"{DB.CF_R2_PUBLIC_URL}/{DAILY_KEY}", rel="self", type="application/rss+xml")
76 fg.title("AI News")
77 fg.language("zh-CN")
78 fg.category(term="AI")
79 fg.copyright("DNKT AI News")
80 fg.description("DNKT AI News")
81 fg.ttl(60)
82 for day_str, item in sorted(items.items()):
83 entry = fg.add_entry()
84 entry.title(item["title"])
85 entry.link(href=item["url"])
86 entry.guid(item["url"], permalink=True)
87 pub_dt = datetime.strptime(day_str, "%Y-%m-%d").replace(hour=23, minute=59, second=59, tzinfo=ZoneInfo(TZ))
88 entry.published(pub_dt)
89 entry.content(item["summary"], type="CDATA")
90
91 with tempfile.NamedTemporaryFile("w", suffix=".xml", delete=False) as tempf:
92 fg.rss_file(tempf.name, pretty=True)
93 # add rss beauty
94 xml_str = Path(tempf.name).read_text()
95 xml_str = xml_str.replace("<?xml version='1.0' encoding='UTF-8'?>", "<?xml version='1.0' encoding='UTF-8'?>\n<?xml-stylesheet type='text/xsl' href='/rss.xsl'?>")
96 await set_cf_r2(DAILY_KEY, xml_str, mime_type="application/xml", silent=True)
97
98
99async def get_summary(start: str, end: str) -> list[dict]:
100 """Get AI summary.
101
102 Args:
103 start (str): The start time in YYYY-MM-DD.
104 end (str): The end time in YYYY-MM-DD.
105
106 Returns:
107 list[dict]: list of news summary
108 """
109 start_dt = datetime.strptime(start, "%Y-%m-%d").replace(hour=0, minute=0, second=0, tzinfo=ZoneInfo(TZ))
110 end_dt = datetime.strptime(end, "%Y-%m-%d").replace(hour=23, minute=59, second=59, tzinfo=ZoneInfo(TZ))
111 start_ts = int(start_dt.timestamp())
112 end_ts = int(end_dt.timestamp())
113 d1 = await query_d1(sql=f"SELECT timestamp,title,url,summary FROM readhub WHERE timestamp >= {start_ts} AND timestamp <= {end_ts} ORDER BY timestamp ASC", db_name="dnkt", silent=True)
114 items = glom(d1, "result.0.results", default=[]) # From old to new
115 if not items:
116 return []
117 prompt = generate_ai_prompt(items)
118 # save sources to R2
119 if start == end: # daily
120 await set_cf_r2(f"AI-NEWS/daily/{start}.json", items, silent=True)
121 ai_msg = Message( # Construct a message for AI
122 id=rand_number(),
123 chat=Chat(id=rand_number()),
124 text=Str(f"{PREFIX.AI_TEXT_GENERATION} @general {prompt}"),
125 )
126
127 ai_res = await ai_text_generation(
128 "fake-client", # type: ignore
129 ai_msg,
130 openai_responses_config={
131 "instructions": daily_system_prompt(start),
132 "text": {
133 "format": {
134 "type": "json_schema",
135 "name": "NewsSummary",
136 "strict": True,
137 "description": "A list of news summary",
138 "schema": JSON_SCHEMA,
139 }
140 },
141 },
142 gemini_generate_content_config={
143 "system_instruction": daily_system_prompt(start),
144 "responseMimeType": "application/json",
145 "responseJsonSchema": JSON_SCHEMA,
146 },
147 openai_append_tool_results=False,
148 gemini_append_grounding=False,
149 anthropic_append_citation=False,
150 silent=True,
151 )
152 if not ai_res.get("texts", ""):
153 return []
154 return json.loads(ai_res.get("texts", ""))
155
156
157def convert_news_summary_to_html(news_summary: list[dict]) -> str:
158 """Convert news summary to html."""
159 emoji = {"Model": "🤖", "Tech": "🛠️", "Finance": "💰", "Policy": "📜", "Application": "💡", "Industry": "🏢", "Others": "✨"}
160 category_order = ["Model", "Tech", "Application", "Industry", "Finance", "Policy", "Others"]
161 category_map = {
162 "Model": "模型 (Models)",
163 "Tech": "技术 (Technologies)",
164 "Application": "应用 (Applications)",
165 "Industry": "行业动态 (Industry)",
166 "Finance": "投融资 (Finance)",
167 "Policy": "政策 (Policy)",
168 "Others": "其他 (Others)",
169 }
170
171 # Group by category
172 grouped = defaultdict(list)
173 for item in news_summary:
174 category = item.get("category", "Others")
175 if category not in category_order:
176 continue
177 grouped[category].append(item)
178
179 html = ""
180 for category in category_order:
181 items = grouped[category]
182 if not items:
183 continue
184 html += f"<h2>{emoji[category]} {category_map[category]}</h2>"
185 for idx, item in enumerate(items):
186 title = item["title"]
187 summary = item["abstract"]
188 if len(item["urls"]) == 1:
189 html += f'<p><strong>{idx + 1}. <a href="{item["urls"][0]}">{title}</a></strong><br>{summary}</p>'
190 else:
191 urls = ""
192 for uidx, url in enumerate(item["urls"]):
193 urls += f'<a href="{url}">[link-{uidx + 1}]</a>,'
194 urls = urls.rstrip(",")
195 html += f"<p><strong>{idx + 1}. {title}</strong> {urls}<br>{summary}</p>"
196 return html
197
198
199async def parse_finished(key: str = DAILY_KEY) -> dict:
200 parsed = {}
201 with suppress(Exception):
202 feed_url = f"{DB.CF_R2_PUBLIC_URL.rstrip('/')}/{key}"
203 feed = feedparser.parse(feed_url)
204 for item in glom(feed, "entries", default=[]):
205 title = item["title"]
206 summary = item["summary"]
207 url = item["link"]
208 date = title[-10:] # YYYY-MM-DD
209 parsed[date] = {"title": title, "summary": summary, "url": url}
210 return parsed
211
212
213def generate_ai_prompt(items: list[dict]) -> str:
214 """Generate AI prompt in json format."""
215 prompt = []
216 for idx, item in enumerate(items): # noqa: B007
217 title = item["title"]
218 summary = item["summary"]
219 url = item["url"].removeprefix("?f=rss")
220 prompt.append({"title": title, "summary": summary, "url": url})
221 return json.dumps(prompt, ensure_ascii=False)
222
223
224def daily_system_prompt(day: str) -> str:
225 prompt = f"""# Role
226你是一位资深的人工智能行业分析师和主编。你具备敏锐的新闻嗅觉，擅长从海量、碎片化的源数据中识别关键信息，并将其转化为结构化、高价值的行业简报。
227
228# Context
229当前日期：{day}
230输入数据：一份包含本日 AI 相关新闻的 JSON 列表（包含标题 title、摘要 summary、链接 url）。
231输出目标：符合 JSON Schema 定义的清洗后的新闻列表。
232
233# Workflow & Guidelines (必须严格执行)
234
235## 1. 深度清洗与去重 (核心任务)
236 - **语义聚合**：不要仅仅比对标题。如果多条新闻报道了同一个核心事件（例如 "OpenAI 发布 Sora" 和 "Sora 正式官宣"），必须将其合并为一个条目。
237 - **去除噪音**：剔除毫无实质内容的标题党、重复的营销软文、极其边缘的小道消息。
238 - **选取最佳来源**：在合并时，保留 1-2 个最具信息量的 URL 放入 `urls` 字段。
239
240## 2. 智能分类 (Category)
241 请根据新闻核心内容，严格按以下标准分类：
242 - **Model**: 大模型发布、权重开源、算法更新 (如 GPT-5, Llama 3)。
243 - **Tech**: 具体的 AI 技术突破、学术论文、RAG/Agent 等技术架构 (非具体模型)。
244 - **Application**: AI 产品落地、新功能上线、具体场景应用 (如 Copilot 更新)。
245 - **Finance**: 融资、收购、上市、财报、股价大幅波动。
246 - **Industry**: 行业大事件、人事变动、监管与法律诉讼、算力芯片 (Nvidia/AMD)。
247 - **Policy**: 政府法规、AI 安全倡议、白皮书。
248 - **Others**: 无法归类的内容。
249
250## 3. 摘要撰写规范 (Abstract)
251 - **禁止包含链接**：摘要文本中**绝对不要**包含 http 链接，链接仅需放入 JSON 的 `urls` 列表即可。
252 - **客观陈述**：使用第三方新闻语调。例如："OpenAI 宣布推出..." 而不是 "我们很高兴地宣布..."。
253 - **信息密度**：50-100 字。涵盖 "Who (谁) + What (做了什么) + Impact (有什么关键影响/参数)"。
254 - **中文输出**：即使输入是英文，摘要和标题也必须是流畅的中文。
255
256## 4. 标题重写 (Title)
257 - 必须简练（20字以内）。
258 - 去除 "突发"、"重磅" 等修饰词，直接陈述事实。
259
260## 5. 关注重点
261 - 仅保留你认为比较重要的事件，无需囊括所有新闻条目。
262 - 每个分类中最多保留 10 条新闻。
263
264请处理输入数据，并生成符合 Schema 的 JSON 输出。
265""" # noqa: RUF001
266 return prompt.strip()
267
268
269def get_last_week() -> tuple[datetime, datetime]:
270 """Get last monday and sunday's date."""
271 today = nowdt(TZ)
272 this_monday = today - timedelta(days=today.weekday())
273 previous_monday = this_monday - timedelta(days=7)
274 previous_sunday = this_monday - timedelta(days=1)
275 return previous_monday.replace(hour=0, minute=0, second=0, tzinfo=ZoneInfo(TZ)), previous_sunday.replace(hour=23, minute=59, second=59, tzinfo=ZoneInfo(TZ))
276
277
278async def weekly_ainews():
279 last_monday, last_sunday = get_last_week()
280 html_key = f"AI-NEWS/weekly/{last_monday:%Y-%m-%d}~{last_sunday:%Y-%m-%d}.html"
281 if await head_cf_r2(html_key):
282 return
283
284 parsed = await parse_finished(WEEKLY_KEY)
285 # if not parsed:
286 # logger.warning("No Weekly AI news parsed.")
287 # return
288
289 # gather news
290 news = []
291 for i in range(7):
292 day = last_monday + timedelta(days=i)
293 list_news: list[dict] = await get_cf_r2(key=f"AI-NEWS/daily/{day:%Y-%m-%d}-summary.json", silent=True) # ty:ignore[invalid-assignment]
294 news.extend(list_news)
295 for x in news:
296 x.pop("category", None)
297
298 ai_msg = Message( # Construct a message for AI
299 id=rand_number(),
300 chat=Chat(id=rand_number()),
301 text=Str(f"{PREFIX.AI_TEXT_GENERATION} @doubao {json.dumps(news, ensure_ascii=False)}"),
302 )
303 ai_res = await ai_text_generation(
304 "fake-client", # type: ignore
305 ai_msg,
306 openai_responses_config={"instructions": weekly_system_prompt()},
307 gemini_generate_content_config={"system_instruction": weekly_system_prompt()},
308 silent=True,
309 )
310 weekly_report = ai_res.get("texts", "")
311 if not weekly_report:
312 return
313 title = f"AI周报 | {last_monday:%Y-%m-%d}~{last_sunday:%Y-%m-%d}"
314 html = convert2html(weekly_report)
315 full_html = f'<!DOCTYPE html><html><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>{title}</title><link rel="stylesheet" href="https://cdn.jsdmirror.cn/npm/water.css@2/out/water.css" integrity="sha512-GW7j11RXZmdio87hQsKNjomKEy/DwDjh6J2Z1JnI5Z4FNP791QfZP7Iut25vA+L+YJLZipI2BZhEpkvBkfr8cw==" crossorigin="anonymous"></head><body><h1>{title}</h1>{html}</body></html>'
316 await set_cf_r2(key=html_key, data=full_html, mime_type="text/html")
317
318 parsed[f"{last_sunday:%Y-%m-%d}"] = {"title": f"AI周报 | {last_monday:%Y-%m-%d}~{last_sunday:%Y-%m-%d}", "summary": html, "url": f"{DB.CF_R2_PUBLIC_URL}/{html_key}"}
319 fg = FeedGenerator()
320 fg.link(href=f"{DB.CF_R2_PUBLIC_URL}/{DAILY_KEY}", rel="self", type="application/rss+xml")
321 fg.title("AI周报")
322 fg.language("zh-CN")
323 fg.category(term="AI")
324 fg.copyright("DNKT AI News")
325 fg.description("DNKT AI News 周报")
326 fg.ttl(480)
327 for sunday, item in sorted(parsed.items()):
328 entry = fg.add_entry()
329 entry.title(item["title"])
330 entry.link(href=item["url"])
331 entry.guid(item["url"], permalink=True)
332 pub_dt = datetime.strptime(sunday, "%Y-%m-%d").replace(hour=23, minute=59, second=59, tzinfo=ZoneInfo(TZ))
333 entry.published(pub_dt)
334 entry.content(item["summary"], type="CDATA")
335
336 with tempfile.NamedTemporaryFile("w", suffix=".xml", delete=False) as tempf:
337 fg.rss_file(tempf.name, pretty=True)
338 # add rss beauty
339 xml_str = Path(tempf.name).read_text()
340 xml_str = xml_str.replace("<?xml version='1.0' encoding='UTF-8'?>", "<?xml version='1.0' encoding='UTF-8'?>\n<?xml-stylesheet type='text/xsl' href='/rss.xsl'?>")
341 await set_cf_r2(WEEKLY_KEY, xml_str, mime_type="application/xml", silent=True)
342
343
344def weekly_system_prompt() -> str:
345 prompt = """# Role Definition
346你是一名为顶级汽车软件公司服务的**首席AI技术战略官**，精通 **QNX、Linux (AGL)、Android** 混合架构，并对**智能座舱 HMI（仪表/中控）**开发有深刻理解。你的核心能力结合了对 AI 前沿技术（AGI、LLM、Multimodal）的敏锐嗅觉。
347
348# Context & Goal
349用户是服务于丰田系汽车的软件开发工程师，主要负责**仪表盘 (Instrument Cluster)** 和 **中控屏 (IVI)** 的软件开发。
350你需要根据用户提供的 JSON 数据（包含 title, summary, urls），撰写一份具有深度的《AI行业周报》。
351你的目标不是简单的“新闻搬运”，而是“情报分析”。你需要帮助用户回答两个核心问题：
3521. 本周 AI 行业发生了什么可能会改变未来的大事？
3532. 这些技术如何具体落地到我们的汽车软件代码、架构或用户体验中？
354
355# Capabilities & Constraints
356由于你具备**联网搜索能力**，请严格遵守以下工作流：
357
358## Phase 1: Data Ingestion & Enrichment (数据摄入与增强)
3591. **去重与聚类**：阅读输入的 JSON 数据。识别报道同一事件的多条新闻（例如多家媒体报道同一个模型发布），将其聚合为一个条目。
3602. **主动搜索 (关键步骤)**：
361 - 如果 JSON 中的 `summary` 信息量不足，或者该新闻通过标题判断极其重要（Tier 1级别），**请务必使用搜索工具搜索该新闻的最新深度解读或技术白皮书**。
362 - 不要仅依赖提供的 JSON，要确保你的分析是基于最准确、最全面的技术细节（例如：模型参数量、推理延迟数据、上下文长度等）。
363
364## Phase 2: Strategic Filtering (战略筛选)
365请将新闻分为以下三类，并按此优先级排序：
366
367- **Tier 1: 行业范式转移 (Paradigm Shifts)**
368 - *标准*：基座模型的重大迭代（如 GPT-5, Gemini 1.5 Ultra）、颠覆性的新架构（如 SSM, Mamba 变体）、关键开源项目、或可能重塑行业的政策/商业动态。
369 - *处理*：即使与汽车无关，也**必须收录**。这是周报的核心价值。
370
371- **Tier 2: 车载软件强相关 (Automotive Relevant)**
372 - *标准*：直接涉及智能座舱、语音交互、端侧推理 (On-device AI)、RAG 在边缘端的应用、AI 辅助编程 (Copilot for Devs)、实时渲染等。
373 - *处理*：重点分析落地可行性。
374
375- **Tier 3: 噪音过滤 (Drop)**
376 - *标准*：单纯的股价波动、无技术细节的营销通稿、重复的低质量内容。
377 - *处理*：直接丢弃。
378
379## Phase 3: Analysis & Writing (分析与撰写)
380在撰写每条新闻时，必须包含 **[深度技术解析]** 和 **[车机软件启示]**。
381
382**关于 [车机软件启示] 的特殊指令：**
383- **禁止**说空话（如“这将提升用户体验”）。
384- **必须**联系具体的软件开发场景。请思考以下维度：
385 - **性能与架构**：是否涉及 NPU 算力占用？模型量化后是否能在高通/英伟达车规芯片上跑通？
386 - **交互创新**：是否支持多模态（手势+语音）？能否用于生成动态 UI？
387 - **开发效率**：该工具能否帮助车企软件团队自动生成单元测试或重构代码？
388 - **隐私与安全**：端侧处理对用户隐私意味着什么？
389
390# Output Format (Markdown)
391
392## 🚨 行业头条 (Headline Events)
393*(针对 Tier 1 新闻，深入剖析)*
394
395### [1. 新闻标题]
396> **核心事实**：[融合了 JSON 信息和你联网搜索补充的关键技术参数]
397> **行业震级**：⭐⭐⭐⭐⭐ (简述为什么这是行业拐点)
398> **对车机软件的降维打击**：[发散思维，例如：*“虽然这是云端大模型，但其蒸馏版本可能在明年进入车机，改变现有的语音助手架构...”*]
399
400---
401
402## 🚘 智能座舱与工程应用 (Cockpit & Engineering)
403*(针对 Tier 2 新闻，侧重落地)*
404
405### [2. 新闻标题]
406- **技术摘要**：[简述]
407- **落地分析 (Actionable Insight)**：
408 - *场景*：[具体场景，如：导航搜索、儿童模式生成、DMS疲劳监测]
409 - *挑战*：[提及延迟、功耗或内存限制]
410
411---
412
413## 🔍 速览 (Quick Bites)
414*(值得注意但篇幅较短的更新)*
415- **[标题]**：一句话核心观点。🔗[Link]
416
417---
418
419## 📝 本周总结 (Weekly Synthesis)
420[用一段话总结本周趋势。直接对作为 AI 工程师的用户喊话，给出下周关注的技术栈建议。]
421""" # noqa: RUF001
422 return prompt.strip()