From 12bbe0058b9c4c01c6215cf2217ccddea0a33e0b Mon Sep 17 00:00:00 2001 From: lhr Date: Sat, 11 Apr 2026 23:47:08 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=9B=BD=E9=99=85?= =?UTF-8?q?=E6=96=B0=E9=97=BB=E8=8E=B7=E5=8F=96=E5=92=8C=E9=99=84=E8=A8=80?= =?UTF-8?q?=E7=94=9F=E6=88=90=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增国际新闻 RSS 源（BBC World、Sky News、NHK 国际） - 新增附言生成功能（AI 生成的"舔狗"风格小故事） - 优化邮件 HTML 模板（信纸风格、主题配色） - 新增天气/新闻缓存机制，避免重复获取 - 新增 NEWS_PROXY_URL 环境变量配置国际新闻代理 - 完善状态管理（monitor_state.json 替代 state.json） Co-Authored-By: Claude Opus 4.6 --- ping_monitor.py | 431 +++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 391 insertions(+), 40 deletions(-) diff --git a/ping_monitor.py b/ping_monitor.py index 1dbcc71..138c2cd 100644 --- a/ping_monitor.py +++ b/ping_monitor.py @@ -16,6 +16,7 @@ import json import re import urllib.parse import urllib.request +import xml.etree.ElementTree as ET from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header @@ -37,6 +38,9 @@ TARGET_IPS = [ # 天气位置配置 WEATHER_LOCATION = "成都" +# 国外新闻代理配置 +NEWS_PROXY_URL = os.getenv("NEWS_PROXY_URL", "http://127.0.0.1:7890").strip() or None + # 运行时间配置 START_HOUR = 2 # 凌晨2点开始 END_HOUR = 22 # 晚上10点结束 @@ -57,6 +61,27 @@ DEFAULT_HTTP_HEADERS = { "User-Agent": "Mozilla/5.0", } +NEWS_CACHE_VERSION = "mixed-news-v1" +DOMESTIC_NEWS_LIMIT = 20 +FOREIGN_NEWS_LIMIT = 40 +FOREIGN_NEWS_SOURCES = [ + { + "name": "BBC World", + "url": "https://feeds.bbci.co.uk/news/world/rss.xml", + "proxy": True, + }, + { + "name": "Sky News World", + "url": "https://feeds.skynews.com/feeds/rss/world.xml", + "proxy": True, + }, + { + "name": "NHK 国际", + "url": "https://www3.nhk.or.jp/rss/news/cat6.xml", + "proxy": False, + }, +] + # ============== 邮件发送函数 ============== def send_email(subject, html_body): @@ -124,6 +149,22 @@ def generate_morning_message(status, arrived_time=None, weather_summary=None, ne # 第三段直接用news_summary（已经是串联好的内容） return f"{first_paragraph}\n\n{second_paragraph}\n\n{news_summary}" +def generate_dog_section(): + """生成附言：问题式副标题 + 故事正文 + 原创舔狗收尾""" + dog_section = get_dog_intro() + if dog_section: + return dog_section + + return { + "subtitle": "喜欢为什么总慢半拍？", + "story": ( + f"以前我也总觉得，只要自己再往前走一步，很多话就会有回应。后来才明白，" + f"有些人不是听不懂，只是不想接住。那天路灯已经暗下来了，他低头笑了笑，" + f"我站在原地，忽然觉得很多没说出口的话，其实早就被看穿了。" + ), + "quote": "她说：\"我没有在等你回我，我只是想看看，我还能为你失眠到几点。\"", + } + def build_email_html(status, arrived_time=None, weather_summary=None, news_summary=None): """构建邮件HTML内容""" if status == "arrived": @@ -134,14 +175,107 @@ def build_email_html(status, arrived_time=None, weather_summary=None, news_summa # 获取AI生成的文案 morning_message = generate_morning_message(status, arrived_time, weather_summary, news_summary) + # 只切前两段，避免AI多换行时把新闻正文截断 + msg_parts = morning_message.split('\n\n', 2) + sun_part = msg_parts[0] if len(msg_parts) > 0 else "" + weather_part = msg_parts[1] if len(msg_parts) > 1 else "" + news_part = msg_parts[2] if len(msg_parts) > 2 else "" + + # 获取附言部分 + dog_section = generate_dog_section() + + theme = { + "seal": "太阳照常升起" if status == "arrived" else "吗喽也要休息", + "accent": "#A66A4C" if status == "arrived" else "#6B7485", + "ink": "#3B3028", + "sub_ink": "#6A5A4D", + "paper": "#FFF9F0", + "paper_edge": "#E8DCCB", + "line": "#E9DDCF", + "seal_bg": "#F3E1D2" if status == "arrived" else "#E2E8EF", + "seal_text": "#9C5F43" if status == "arrived" else "#5E7387", + } + html = f""" - -
- {morning_message.replace('\n', '
')} + + + + + +
+
+
+
+
+ 致：{YOUR_NAME} +
+
+ {theme['seal']} +
+
+ +
+ 今天给你写一封晨间小信 +
+
+ 把天气、热点和一点碎碎念都收进信里。打开它的时候，就当是有人替你先把今天轻轻铺开。 +
+
+ +
+
+
先和你说
+
+ {sun_part.replace('\n', '
')} +
+
+ +
+
今天天气
+
+ {weather_part.replace('\n', '
')} +
+
+ +
+
今天的事
+
+ {news_part.replace('\n', '
')} +
+
+
+ """ + + # 如果有附言，添加附言区 + if dog_section: + html += f""" +
+
附言
+
+ {dog_section['subtitle']} +
+
+ {dog_section['story'].replace('\n', '
')} +
+
+ {dog_section['quote'].replace('\n', '
')} +
+
+ """ + + html += f""" +
+
+ 愿今天也顺顺当当。 +
+
+ {datetime.datetime.now().strftime('%Y.%m.%d')} · 自动送达
+ 写信人：{FROM_NAME} +
+
+
-
-

此邮件由系统自动发送

""" @@ -242,16 +376,89 @@ def is_failed_content(content): """判断内容是否属于失败结果""" return not content or "获取失败" in content -def fetch_text(url, headers=None, timeout=10): +def fetch_text(url, headers=None, timeout=10, proxy_url=None): """发起HTTP请求并返回文本内容""" request_headers = dict(DEFAULT_HTTP_HEADERS) if headers: request_headers.update(headers) req = urllib.request.Request(url, headers=request_headers) - with urllib.request.urlopen(req, timeout=timeout) as response: + if proxy_url: + proxy_handler = urllib.request.ProxyHandler({ + "http": proxy_url, + "https": proxy_url, + }) + opener = urllib.request.build_opener(proxy_handler) + response = opener.open(req, timeout=timeout) + else: + response = urllib.request.urlopen(req, timeout=timeout) + + with response: return response.read().decode('utf-8', errors='ignore') +def get_domestic_news_titles(limit=DOMESTIC_NEWS_LIMIT): + """获取国内热点标题""" + html = fetch_text("https://top.baidu.com/board?tab=realtime") + raw_titles = re.findall(r'"word":"(.*?)"', html) + + news_items = [] + seen_titles = set() + for raw_title in raw_titles: + title = raw_title.strip() + if not title or title in seen_titles: + continue + seen_titles.add(title) + news_items.append(title) + if len(news_items) >= limit: + break + + return news_items + +def parse_rss_titles(rss_text, limit): + """从RSS内容中提取标题""" + root = ET.fromstring(rss_text) + titles = [] + seen_titles = set() + + for item in root.findall('.//item'): + title_element = item.find('title') + if title_element is None or not title_element.text: + continue + title = re.sub(r'\s+', ' ', title_element.text).strip() + if not title or title in seen_titles: + continue + seen_titles.add(title) + titles.append(title) + if len(titles) >= limit: + break + + return titles + +def get_foreign_news_titles(limit=FOREIGN_NEWS_LIMIT): + """获取国外新闻平台标题，必要时走代理""" + foreign_items = [] + seen_titles = set() + + for source in FOREIGN_NEWS_SOURCES: + proxy_url = NEWS_PROXY_URL if source.get("proxy") else None + try: + rss_text = fetch_text(source["url"], timeout=15, proxy_url=proxy_url) + titles = parse_rss_titles(rss_text, limit) + for title in titles: + if title in seen_titles: + continue + seen_titles.add(title) + foreign_items.append((source["name"], title)) + if len(foreign_items) >= limit: + break + except Exception as e: + print(f"[{datetime.datetime.now()}] 获取国际新闻失败 ({source['name']}): {e}") + + if len(foreign_items) >= limit: + break + + return foreign_items + def get_weather(): """获取当天天气信息（带缓存）""" today = get_today_str() @@ -274,16 +481,21 @@ def get_weather(): # 天气信息来源 - 使用wttr.in（免费无需API Key） weather_info = "" + weather_ok = False try: encoded_location = urllib.parse.quote(WEATHER_LOCATION) url = f"https://wttr.in/{encoded_location}?format=%l:+%C,+%t,+%h,+feeling:+%f,+%p" weather_info = fetch_text(url) if not weather_info.strip(): weather_info = "天气信息获取失败" + else: + weather_ok = True except Exception as e: print(f"[{datetime.datetime.now()}] 获取天气失败: {e}") weather_info = "天气信息获取失败" + print(f"[{datetime.datetime.now()}] - wttr.in ({WEATHER_LOCATION}): {'✓ 获取成功' if weather_ok else '✗ 获取失败'}") + # 保存缓存 os.makedirs(STATE_DIR, exist_ok=True) cache_data = { @@ -302,9 +514,10 @@ def get_weather(): # ============== 新闻获取函数 ============== -def get_hot_news(): - """获取当天热搜新闻（带缓存）""" +def get_hot_news_with_status(): + """获取当天热搜新闻（带缓存），同时返回各源状态""" today = get_today_str() + source_status = [] # 检查缓存 if os.path.exists(NEWS_CACHE_FILE): @@ -314,36 +527,49 @@ def get_hot_news(): cached_news = cache.get("news", "") if ( cache.get("date") == today + and cache.get("version") == NEWS_CACHE_VERSION and time.time() - cache.get("timestamp", 0) < CACHE_EXPIRE and not is_failed_content(cached_news) ): - return cached_news + return cached_news, [] except Exception: pass - # 从百度热搜页提取热榜关键词 news_info = "" try: - html = fetch_text("https://top.baidu.com/board?tab=realtime") - raw_titles = re.findall(r'"word":"(.*?)"', html) - news_items = [] - seen_titles = set() + domestic_titles = get_domestic_news_titles() + domestic_ok = bool(domestic_titles) + source_status.append(("国内热点 (百度)", domestic_ok, len(domestic_titles))) - for raw_title in raw_titles: - title = raw_title.strip() - if not title or title in seen_titles: - continue - seen_titles.add(title) - news_items.append(title) - if len(news_items) >= 10: - break + foreign_titles = get_foreign_news_titles() + foreign_sources = {} + for source_name, _ in foreign_titles: + foreign_sources[source_name] = foreign_sources.get(source_name, 0) + 1 + for src in FOREIGN_NEWS_SOURCES: + src_name = src["name"] + count = foreign_sources.get(src_name, 0) + source_status.append((src_name, count > 0, count)) - if not news_items: - raise ValueError("未能从百度热搜页面提取标题") + sections = [] + if domestic_titles: + sections.append( + "国内热点：\n" + "\n".join( + f"{index}. {title}" for index, title in enumerate(domestic_titles, 1) + ) + ) - news_info = "\n".join( - f"{index}. {title}" for index, title in enumerate(news_items, 1) - ) + if foreign_titles: + sections.append( + "国际热点：\n" + "\n".join( + f"{index}. {source} | {title}" + for index, (source, title) in enumerate(foreign_titles, 1) + ) + ) + + if not sections: + raise ValueError("未能获取国内或国际新闻") + + news_info = "\n\n".join(sections) except Exception as e: print(f"[{datetime.datetime.now()}] 获取新闻失败: {e}") @@ -354,6 +580,7 @@ def get_hot_news(): os.makedirs(STATE_DIR, exist_ok=True) cache_data = { "date": today, + "version": NEWS_CACHE_VERSION, "timestamp": time.time(), "news": news_info } @@ -363,8 +590,86 @@ def get_hot_news(): except Exception: pass + return news_info, source_status + +def get_hot_news(): + """获取当天热搜新闻（带缓存）""" + news_info, source_status = get_hot_news_with_status() + if source_status: + # source_status[0] = 国内热点, source_status[1:] = 国外各源 + domestic_name, domestic_ok, domestic_count = source_status[0] + print(f"[{datetime.datetime.now()}] - {domestic_name}: {'✓ 获取成功 ' + str(domestic_count) + ' 条' if domestic_ok else '✗ 获取失败'}") + for src_info in source_status[1:]: + src_name, ok, count = src_info + print(f"[{datetime.datetime.now()}] - {src_name}: {'✓ 获取成功 ' + str(count) + ' 条' if ok else '✗ 获取失败'}") return news_info +# ============== 附言生成函数 ============== + +def get_dog_intro(): + """让AI直接生成附言的副标题、故事和收尾台词""" + + prompt = """请直接原创一段适合放进早安邮件附言里的小故事，主题是单方面喜欢、过度投入、自我感动、错位理解这类情绪，但不要写得太油腻。 + +请严格按下面格式输出： +第一行：一个问题式副标题，像“为什么越卑微越容易自我感动？”、“喜欢为什么总在错误频道里发生？”，长度控制在10到22字，以问号结尾。 + +第二部分：写一段100到170字的小故事，像一个过来人在回忆自己以前经历过的一件小事。你要自己判断叙述者更像男生还是女生、喜欢的人是谁，并保持整段故事的人称、语气、细节一致。要自然、具体、克制，带一点自嘲和电影感，但不要像鸡汤，也不要空泛说理。正文结尾要停在一个情绪悬着的位置，像还差一句话没有落下来。 + +第三部分：另起一段，只写一句收尾台词。这里可以是女神/男神的冷淡回应，也可以是舔狗本人特别卑微、特别会自我感动的一句话。关键要求是：如果删掉这最后一句，前面只是一个普通的暗恋小故事；加上这最后一句，整段才会一下子变得很舔、很心酸、甚至有点好笑。这里要自然写成“我说：\"...\"”、“他说：\"...\"”或者“她说：\"...\"”。引号里的内容由你原创，但必须像真实聊天里会出现的话，不要故作矫情。 + +要求： +- 副标题不要加书名号、引号、编号或解释 +- 正文不要分段，不要列表，不要使用 markdown +- 正文和最后那句台词要有情绪上的衔接，像回忆走到那一刻时自然停下来 +- 最后一段单独成段，像信里被轻轻划出来的一句原话，不要额外解释 +- 台词要短一些，尽量控制在20到50字 +- 最后一句必须明显提升“舔狗感”，不能只是普通对白，不能和前文松散脱节 +- 只输出这三部分，每部分之间空一行""" + + intro = ask_deepseek( + prompt, + system_prompt="你是一个很会讲感情小故事的人，文风克制、真诚，带一点自嘲和电影感。你尤其擅长写那种最后一句一出来，整段故事突然显得很舔、很可怜、又有点好笑的附言。", + ) + if not intro: + return None + + parts = intro.split("\n\n", 2) + subtitle = parts[0].strip() if parts else "" + story = parts[1].strip() if len(parts) > 1 else "" + quote = parts[2].strip() if len(parts) > 2 else "" + + if not subtitle or not story or not quote: + lines = [line.strip() for line in intro.splitlines() if line.strip()] + if len(lines) >= 3: + subtitle = lines[0] + quote_index = None + for index, line in enumerate(lines[1:], start=1): + if line.startswith("我说：") or line.startswith("他说：") or line.startswith("她说："): + quote_index = index + break + if quote_index is not None: + story = "".join(lines[1:quote_index]) + quote = "".join(lines[quote_index:]) + else: + story = "".join(lines[1:-1]) + quote = lines[-1] + + if not subtitle or not story or not quote: + return None + + if not subtitle.endswith("？") and not subtitle.endswith("?"): + subtitle = subtitle.rstrip("。！!，,：:") + "？" + + if not (quote.startswith("我说：") or quote.startswith("他说：") or quote.startswith("她说：")): + quote = "我说：\"你回不回都没关系，我只是想让你的对话框里也有一点我的痕迹。\"" + + return { + "subtitle": subtitle, + "story": story, + "quote": quote, + } + # ============== DeepSeek AI 处理函数 ============== def ask_deepseek(prompt, system_prompt="你是一个有用的助手，用简洁温暖的语言回复。"): @@ -413,23 +718,35 @@ def get_weather_summary(weather_info): return summary if summary else weather_info def get_news_summary(news_info): - """让DeepSeek总结新闻要点，找出关联，串联成自然的段落""" + """让DeepSeek把国内外新闻统一整理成中文段落""" if not news_info or "获取失败" in news_info: return news_info - prompt = f"""请阅读以下今日热搜新闻，试着找出它们之间的关联或共同话题，然后写成一段200字左右的自然段落，像朋友聊天一样分享这些新闻： + prompt = f"""请阅读以下今日新闻，其中包含国内热点和国外新闻平台标题。请先理解并翻译其中的外文标题，再从中筛选出最值得聊的内容，写成适合早安邮件的三段式中文短评。 {news_info} 要求： -- 找出新闻之间的联系，用自己的话串起来讲，不要念稿 -- 讲的时候像跟朋友吐槽聊天那样，说到哪个新闻就直接说，别用引号包着 -- 语气像在说相声或脱口秀，自然调侃，不要端着 -- 不要用列表或条目，要连贯的段落 -- 字数200字左右 -- 只返回这段新闻解读，不要多余的话""" +- 固定写成三段，每段之间空一行，不要加小标题 +- 第一段：挑 2 到 3 条更偏宏观的新闻开场，可以是国际局势、科技、公共事件，负责把今天的大背景立起来 +- 第二段：转到更具体、更有人味的新闻，挑 2 到 3 条民生、人物、生活感更强的内容来写，让叙述落到具体的人和细节上 +- 第三段：不要再继续堆新新闻，改成基于前两段做一个自然的观察、总结或轻一点的评价，像看完今天新闻后的一个收束 +- 三段之间要有内在联系，像同一个人顺着一个念头慢慢讲下来，而不是三段独立的小作文 +- 只选真正值得聊的内容，不要为了凑数把每条新闻都塞进去 +- 把外文标题自然翻成中文，不要保留英文原句，也不要写“某媒体说”“我给你翻译几条”这种转述腔 +- 叙述顺序要自然，避免一会儿国外一会儿国内来回跳 +- 语气像一个信息敏感、表达克制的朋友，在早餐时顺手聊几件今天值得知道的事 +- 可以有一点感慨和观察，但不要故作幽默，不要玩梗过多，不要像相声稿 +- 不要用列表、序号、小标题，不要分成“国外新闻这边”“国内新闻这边”这种板块汇报 +- 不要出现“早上好啊”“对了”“先给你翻译几条”“把这些串起来看”这类口水开场或总结套话 +- 总字数控制在 260 到 360 字，第三段宁可短一点，也要起到收束作用 +- ⚠️重要：不要使用任何 markdown 符号，如**加粗**、*斜体*、- 列表等，只用纯文本 +- 只返回最终正文，不要解释你的写作思路""" - summary = ask_deepseek(prompt, system_prompt="你是一个贴心的早安助手，用温暖自然的语言回复。") + summary = ask_deepseek( + prompt, + system_prompt="你是一个擅长写晨间新闻短评的中文编辑，知道如何筛选重点、组织主线，并把信息写得自然、克制、有人味。你尤其擅长三段式写法：前两段给信息，最后一段做收束和观察。", + ) return summary if summary else news_info # ============== 时间判断 ============== @@ -466,13 +783,47 @@ def get_cached_content(): print(f"[{datetime.datetime.now()}] 正在获取天气信息...") weather_info = get_weather() _cached_weather_summary = get_weather_summary(weather_info) - print(f"[{datetime.datetime.now()}] 天气摘要: {_cached_weather_summary[:50]}...") + weather_ok = weather_info and "获取失败" not in weather_info + print(f"[{datetime.datetime.now()}] 天气 {'✓ 获取成功' if weather_ok else '✗ 获取失败'}: {_cached_weather_summary[:50] if _cached_weather_summary else '(无内容)'}...") + print(f"[{datetime.datetime.now()}] - wttr.in (成都): {'✓ 获取成功' if weather_ok else '✗ 获取失败'}") if _cached_news_summary is None: print(f"[{datetime.datetime.now()}] 正在获取新闻信息...") - news_info = get_hot_news() + news_info, source_status = get_hot_news_with_status() _cached_news_summary = get_news_summary(news_info) - print(f"[{datetime.datetime.now()}] 新闻摘要: {_cached_news_summary[:50]}...") + news_ok = news_info and "获取失败" not in news_info + print(f"[{datetime.datetime.now()}] 新闻 {'✓ 获取成功' if news_ok else '✗ 获取失败'}") + + # 如果有源状态（非缓存命中），打印各源状态 + if source_status: + for src_name, ok, count in source_status: + status = f"✓ 获取成功 {count} 条" if ok else "✗ 获取失败" + print(f"[{datetime.datetime.now()}] - {src_name}: {status}") + else: + # 缓存命中，单独检查各源 + domestic_titles = [] + try: + domestic_titles = get_domestic_news_titles() + except Exception: + pass + print(f"[{datetime.datetime.now()}] - 国内热点 (百度): {'✓ 获取成功 ' + str(len(domestic_titles)) + ' 条' if domestic_titles else '✗ 获取失败 (缓存命中)'}") + + foreign_items = [] + try: + foreign_items = get_foreign_news_titles() + except Exception: + pass + foreign_sources = {} + for source_name, _ in foreign_items: + foreign_sources[source_name] = foreign_sources.get(source_name, 0) + 1 + for src in FOREIGN_NEWS_SOURCES: + src_name = src["name"] + count = foreign_sources.get(src_name, 0) + print(f"[{datetime.datetime.now()}] - {src_name}: {'✓ 获取成功 ' + str(count) + ' 条' if count > 0 else '✗ 获取失败 (缓存命中)'}") + + print(f"[{datetime.datetime.now()}] 新闻摘要: {_cached_news_summary[:50] if _cached_news_summary else '(无内容)'}...") + + print(f"[{datetime.datetime.now()}] - 附言: 发送邮件时由 AI 即时生成") return _cached_weather_summary, _cached_news_summary