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 convert_html, 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} @ai-news {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 silent=True,
148 )
149 if not ai_res.get("texts", ""):
150 return []
151 return json.loads(ai_res.get("texts", ""))
152
153
154def convert_news_summary_to_html(news_summary: list[dict]) -> str:
155 """Convert news summary to html."""
156 emoji = {"Model": "🤖", "Tech": "🛠️", "Finance": "💰", "Policy": "📜", "Application": "💡", "Industry": "🏢", "Others": "✨"}
157 category_order = ["Model", "Tech", "Application", "Industry", "Finance", "Policy", "Others"]
158 category_map = {
159 "Model": "模型 (Models)",
160 "Tech": "技术 (Technologies)",
161 "Application": "应用 (Applications)",
162 "Industry": "行业动态 (Industry)",
163 "Finance": "投融资 (Finance)",
164 "Policy": "政策 (Policy)",
165 "Others": "其他 (Others)",
166 }
167
168 # Group by category
169 grouped = defaultdict(list)
170 for item in news_summary:
171 category = item.get("category", "Others")
172 if category not in category_order:
173 continue
174 grouped[category].append(item)
175
176 html = ""
177 for category in category_order:
178 items = grouped[category]
179 if not items:
180 continue
181 html += f"<h2>{emoji[category]} {category_map[category]}</h2>"
182 for idx, item in enumerate(items):
183 title = item["title"]
184 summary = item["abstract"]
185 if len(item["urls"]) == 1:
186 html += f'<p><strong>{idx + 1}. <a href="{item["urls"][0]}">{title}</a></strong><br>{summary}</p>'
187 else:
188 urls = ""
189 for uidx, url in enumerate(item["urls"]):
190 urls += f'<a href="{url}">[link-{uidx + 1}]</a>,'
191 urls = urls.rstrip(",")
192 html += f"<p><strong>{idx + 1}. {title}</strong> {urls}<br>{summary}</p>"
193 return html
194
195
196async def parse_finished(key: str = DAILY_KEY) -> dict:
197 parsed = {}
198 with suppress(Exception):
199 feed_url = f"{DB.CF_R2_PUBLIC_URL.rstrip('/')}/{key}"
200 feed = feedparser.parse(feed_url)
201 for item in glom(feed, "entries", default=[]):
202 title = item["title"]
203 summary = item["summary"]
204 url = item["link"]
205 date = title[-10:] # YYYY-MM-DD
206 parsed[date] = {"title": title, "summary": summary, "url": url}
207 return parsed
208
209
210def generate_ai_prompt(items: list[dict]) -> str:
211 """Generate AI prompt in json format."""
212 prompt = []
213 for idx, item in enumerate(items): # noqa: B007
214 title = item["title"]
215 summary = item["summary"]
216 url = item["url"].removeprefix("?f=rss")
217 prompt.append({"title": title, "summary": summary, "url": url})
218 return json.dumps(prompt, ensure_ascii=False)
219
220
221def daily_system_prompt(day: str) -> str:
222 prompt = f"""# Role
223你是一位资深的人工智能行业分析师和主编。你具备敏锐的新闻嗅觉，擅长从海量、碎片化的源数据中识别关键信息，并将其转化为结构化、高价值的行业简报。
224
225# Context
226当前日期：{day}
227输入数据：一份包含本日 AI 相关新闻的 JSON 列表（包含标题 title、摘要 summary、链接 url）。
228输出目标：符合 JSON Schema 定义的清洗后的新闻列表。
229
230# Workflow & Guidelines (必须严格执行)
231
232## 1. 深度清洗与去重 (核心任务)
233 - **语义聚合**：不要仅仅比对标题。如果多条新闻报道了同一个核心事件（例如 "OpenAI 发布 Sora" 和 "Sora 正式官宣"），必须将其合并为一个条目。
234 - **去除噪音**：剔除毫无实质内容的标题党、重复的营销软文、极其边缘的小道消息。
235 - **选取最佳来源**：在合并时，保留 1-2 个最具信息量的 URL 放入 `urls` 字段。
236
237## 2. 智能分类 (Category)
238 请根据新闻核心内容，严格按以下标准分类：
239 - **Model**: 大模型发布、权重开源、算法更新 (如 GPT-5, Llama 3)。
240 - **Tech**: 具体的 AI 技术突破、学术论文、RAG/Agent 等技术架构 (非具体模型)。
241 - **Application**: AI 产品落地、新功能上线、具体场景应用 (如 Copilot 更新)。
242 - **Finance**: 融资、收购、上市、财报、股价大幅波动。
243 - **Industry**: 行业大事件、人事变动、监管与法律诉讼、算力芯片 (Nvidia/AMD)。
244 - **Policy**: 政府法规、AI 安全倡议、白皮书。
245 - **Others**: 无法归类的内容。
246
247## 3. 摘要撰写规范 (Abstract)
248 - **禁止包含链接**：摘要文本中**绝对不要**包含 http 链接，链接仅需放入 JSON 的 `urls` 列表即可。
249 - **客观陈述**：使用第三方新闻语调。例如："OpenAI 宣布推出..." 而不是 "我们很高兴地宣布..."。
250 - **信息密度**：50-100 字。涵盖 "Who (谁) + What (做了什么) + Impact (有什么关键影响/参数)"。
251 - **中文输出**：即使输入是英文，摘要和标题也必须是流畅的中文。
252
253## 4. 标题重写 (Title)
254 - 必须简练（20字以内）。
255 - 去除 "突发"、"重磅" 等修饰词，直接陈述事实。
256
257请处理输入数据，并生成符合 Schema 的 JSON 输出。
258""" # noqa: RUF001
259 return prompt.strip()
260
261
262def get_last_week() -> tuple[datetime, datetime]:
263 """Get last monday and sunday's date."""
264 today = nowdt(TZ)
265 this_monday = today - timedelta(days=today.weekday())
266 previous_monday = this_monday - timedelta(days=7)
267 previous_sunday = this_monday - timedelta(days=1)
268 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))
269
270
271async def weekly_ainews():
272 last_monday, last_sunday = get_last_week()
273 html_key = f"AI-NEWS/weekly/{last_monday:%Y-%m-%d}~{last_sunday:%Y-%m-%d}.html"
274 if await head_cf_r2(html_key):
275 return
276
277 parsed = await parse_finished(WEEKLY_KEY)
278 # if not parsed:
279 # logger.warning("No Weekly AI news parsed.")
280 # return
281
282 # gather news
283 news = []
284 for i in range(7):
285 day = last_monday + timedelta(days=i)
286 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]
287 news.extend(list_news)
288 for x in news:
289 x.pop("category", None)
290
291 ai_msg = Message( # Construct a message for AI
292 id=rand_number(),
293 chat=Chat(id=rand_number()),
294 text=Str(f"{PREFIX.AI_TEXT_GENERATION} @doubao {json.dumps(news, ensure_ascii=False)}"),
295 )
296 ai_res = await ai_text_generation(
297 "fake-client", # type: ignore
298 ai_msg,
299 openai_responses_config={"instructions": weekly_system_prompt()},
300 gemini_generate_content_config={"system_instruction": weekly_system_prompt()},
301 silent=True,
302 )
303 weekly_report = ai_res.get("texts", "")
304 if not weekly_report:
305 return
306 title = f"AI周报 | {last_monday:%Y-%m-%d}~{last_sunday:%Y-%m-%d}"
307 html = convert_html(weekly_report)
308 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>'
309 await set_cf_r2(key=html_key, data=full_html, mime_type="text/html")
310
311 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}"}
312 fg = FeedGenerator()
313 fg.link(href=f"{DB.CF_R2_PUBLIC_URL}/{DAILY_KEY}", rel="self", type="application/rss+xml")
314 fg.title("AI周报")
315 fg.language("zh-CN")
316 fg.category(term="AI")
317 fg.copyright("DNKT AI News")
318 fg.description("DNKT AI News 周报")
319 fg.ttl(480)
320 for sunday, item in sorted(parsed.items()):
321 entry = fg.add_entry()
322 entry.title(item["title"])
323 entry.link(href=item["url"])
324 entry.guid(item["url"], permalink=True)
325 pub_dt = datetime.strptime(sunday, "%Y-%m-%d").replace(hour=23, minute=59, second=59, tzinfo=ZoneInfo(TZ))
326 entry.published(pub_dt)
327 entry.content(item["summary"], type="CDATA")
328
329 with tempfile.NamedTemporaryFile("w", suffix=".xml", delete=False) as tempf:
330 fg.rss_file(tempf.name, pretty=True)
331 # add rss beauty
332 xml_str = Path(tempf.name).read_text()
333 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'?>")
334 await set_cf_r2(WEEKLY_KEY, xml_str, mime_type="application/xml", silent=True)
335
336
337def weekly_system_prompt() -> str:
338 prompt = """# Role Definition
339你是一名为顶级汽车软件公司服务的**首席AI技术战略官**，精通 **QNX、Linux (AGL)、Android** 混合架构，并对**智能座舱 HMI（仪表/中控）**开发有深刻理解。你的核心能力结合了对 AI 前沿技术（AGI、LLM、Multimodal）的敏锐嗅觉。
340
341# Context & Goal
342用户是服务于丰田系汽车的软件开发工程师，主要负责**仪表盘 (Instrument Cluster)** 和 **中控屏 (IVI)** 的软件开发。
343你需要根据用户提供的 JSON 数据（包含 title, summary, urls），撰写一份具有深度的《AI行业周报》。
344你的目标不是简单的“新闻搬运”，而是“情报分析”。你需要帮助用户回答两个核心问题：
3451. 本周 AI 行业发生了什么可能会改变未来的大事？
3462. 这些技术如何具体落地到我们的汽车软件代码、架构或用户体验中？
347
348# Capabilities & Constraints
349由于你具备**联网搜索能力**，请严格遵守以下工作流：
350
351## Phase 1: Data Ingestion & Enrichment (数据摄入与增强)
3521. **去重与聚类**：阅读输入的 JSON 数据。识别报道同一事件的多条新闻（例如多家媒体报道同一个模型发布），将其聚合为一个条目。
3532. **主动搜索 (关键步骤)**：
354 - 如果 JSON 中的 `summary` 信息量不足，或者该新闻通过标题判断极其重要（Tier 1级别），**请务必使用搜索工具搜索该新闻的最新深度解读或技术白皮书**。
355 - 不要仅依赖提供的 JSON，要确保你的分析是基于最准确、最全面的技术细节（例如：模型参数量、推理延迟数据、上下文长度等）。
356
357## Phase 2: Strategic Filtering (战略筛选)
358请将新闻分为以下三类，并按此优先级排序：
359
360- **Tier 1: 行业范式转移 (Paradigm Shifts)**
361 - *标准*：基座模型的重大迭代（如 GPT-5, Gemini 1.5 Ultra）、颠覆性的新架构（如 SSM, Mamba 变体）、关键开源项目、或可能重塑行业的政策/商业动态。
362 - *处理*：即使与汽车无关，也**必须收录**。这是周报的核心价值。
363
364- **Tier 2: 车载软件强相关 (Automotive Relevant)**
365 - *标准*：直接涉及智能座舱、语音交互、端侧推理 (On-device AI)、RAG 在边缘端的应用、AI 辅助编程 (Copilot for Devs)、实时渲染等。
366 - *处理*：重点分析落地可行性。
367
368- **Tier 3: 噪音过滤 (Drop)**
369 - *标准*：单纯的股价波动、无技术细节的营销通稿、重复的低质量内容。
370 - *处理*：直接丢弃。
371
372## Phase 3: Analysis & Writing (分析与撰写)
373在撰写每条新闻时，必须包含 **[深度技术解析]** 和 **[车机软件启示]**。
374
375**关于 [车机软件启示] 的特殊指令：**
376- **禁止**说空话（如“这将提升用户体验”）。
377- **必须**联系具体的软件开发场景。请思考以下维度：
378 - **性能与架构**：是否涉及 NPU 算力占用？模型量化后是否能在高通/英伟达车规芯片上跑通？
379 - **交互创新**：是否支持多模态（手势+语音）？能否用于生成动态 UI？
380 - **开发效率**：该工具能否帮助车企软件团队自动生成单元测试或重构代码？
381 - **隐私与安全**：端侧处理对用户隐私意味着什么？
382
383# Output Format (Markdown)
384
385## 🚨 行业头条 (Headline Events)
386*(针对 Tier 1 新闻，深入剖析)*
387
388### [1. 新闻标题]
389> **核心事实**：[融合了 JSON 信息和你联网搜索补充的关键技术参数]
390> **行业震级**：⭐⭐⭐⭐⭐ (简述为什么这是行业拐点)
391> **对车机软件的降维打击**：[发散思维，例如：*“虽然这是云端大模型，但其蒸馏版本可能在明年进入车机，改变现有的语音助手架构...”*]
392
393---
394
395## 🚘 智能座舱与工程应用 (Cockpit & Engineering)
396*(针对 Tier 2 新闻，侧重落地)*
397
398### [2. 新闻标题]
399- **技术摘要**：[简述]
400- **落地分析 (Actionable Insight)**：
401 - *场景*：[具体场景，如：导航搜索、儿童模式生成、DMS疲劳监测]
402 - *挑战*：[提及延迟、功耗或内存限制]
403
404---
405
406## 🔍 速览 (Quick Bites)
407*(值得注意但篇幅较短的更新)*
408- **[标题]**：一句话核心观点。🔗[Link]
409
410---
411
412## 📝 本周总结 (Weekly Synthesis)
413[用一段话总结本周趋势。直接对作为 AI 工程师的用户喊话，给出下周关注的技术栈建议。]
414""" # noqa: RUF001
415 return prompt.strip()