mixue_monitor.py
12.1 KB| 1 | #!/usr/bin/env python3 |
| 2 | # -*- coding: utf-8 -*- |
| 3 | # mixue_monitor.py —— 蜜雪冰城新品/活动监测（QQ 推送） |
| 4 | # 机制：cn.bing.com RSS 搜索（免 key）轮询多个关键词，过滤出"新品/上新/联名/新口味/限定"类 |
| 5 | # 时效条目（默认最近 3 天），去重后通过 OneBot 推送到老师 QQ。 |
| 6 | # 用法： |
| 7 | # python3 mixue_monitor.py # 正常巡检：抓取→过滤→去重→推送 |
| 8 | # python3 mixue_monitor.py --dry-run # 只打印将推送内容，不推送不改状态 |
| 9 | # python3 mixue_monitor.py --days 7 # 放宽时效窗口到 7 天 |
| 10 | # python3 mixue_monitor.py --list # 查看 state 里已推送记录 |
| 11 | import argparse |
| 12 | import json |
| 13 | import os |
| 14 | import re |
| 15 | import subprocess |
| 16 | import sys |
| 17 | import time |
| 18 | import urllib.parse |
| 19 | from datetime import datetime, timedelta, timezone as dt_tz |
| 20 | |
| 21 | BASE = os.path.dirname(os.path.abspath(__file__)) |
| 22 | CONFIG_FILE = os.path.join(BASE, 'mixue_config.json') |
| 23 | STATE_FILE = os.path.join(BASE, 'mixue_state.json') |
| 24 | LOG = os.path.join(BASE, 'mixue_monitor.log') |
| 25 | UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36' |
| 26 | |
| 27 | # 搜索关键词（覆盖新品/上新/联名/回归/限定等角度） |
| 28 | QUERIES = [ |
| 29 | '蜜雪冰城 新品', |
| 30 | '蜜雪冰城 上新', |
| 31 | '蜜雪冰城 联名', |
| 32 | '蜜雪冰城 新品上市', |
| 33 | '蜜雪冰城 新口味', |
| 34 | '蜜雪冰城 限定', |
| 35 | ] |
| 36 | |
| 37 | # 命中即视为"值得看"的关键词（标题里出现任一即候选） |
| 38 | INTEREST_KW = ['新品', '上新', '上市', '联名', '新口味', '限定', '回归', '新包装', |
| 39 | '新杯', '新系列', '首发', '奶麻薯', '雪王', '周边', '快闪', '联名款'] |
| 40 | # 相关性必须命中（防止"蜜"字等无关结果混入） |
| 41 | RELEVANT_KW = ['蜜雪', '雪王', 'mixue', 'Mixue', 'MXBC', '蜜雪冰城'] |
| 42 | # 排除词（纯字典/百科/无关页/财经/导购/回顾文/实测文） |
| 43 | EXCLUDE_KW = ['百度百科', '汉语', '字典', '国学', '爱企查', '企查查', '加盟', '招商', '招聘', |
| 44 | '股票', '股吧', '财务报表', '港股', '开户', '十大股东', '工商信息', '财报', |
| 45 | '怎么选', '购买指南', '评测', '测评报告', '哪个牌子', '对比测评', |
| 46 | '实测', '攻略', '教程', |
| 47 | '必喝Top5', '必喝Top'] |
| 48 | |
| 49 | # 已核实的老官宣/噪音链接（精确 URL 排除，不用宽泛标题句式——避免误杀真新品官宣） |
| 50 | EXCLUDE_LINKS = [ |
| 51 | 'https://www.weibo.com/1704709632/ReZUptowL', # 8/25 奶麻薯首发官宣（Bing 伪标 10/27） |
| 52 | ] |
| 53 | |
| 54 | WINDOW_DAYS = 7 # 默认只看最近 N 天的条目 |
| 55 | |
| 56 | |
| 57 | def load_cfg(): |
| 58 | try: |
| 59 | with open(CONFIG_FILE) as f: |
| 60 | return json.load(f) |
| 61 | except Exception: |
| 62 | return {} |
| 63 | |
| 64 | |
| 65 | CFG = load_cfg() |
| 66 | OWNER_QQ = os.environ.get('QQ_USER_ID') or str(CFG.get('qq_user_id') or '') |
| 67 | ONE_BOT_URL = os.environ.get('ONE_BOT_URL') or CFG.get('one_bot_url') or 'http://127.0.0.1:3000/send_private_msg' |
| 68 | ONE_BOT_TOKEN = os.environ.get('ONE_BOT_TOKEN') or CFG.get('one_bot_token') or '' |
| 69 | |
| 70 | |
| 71 | def log(msg): |
| 72 | line = f'{datetime.now().strftime("%Y-%m-%d %H:%M:%S")} {msg}' |
| 73 | print(line, flush=True) |
| 74 | try: |
| 75 | with open(LOG, 'a') as f: |
| 76 | f.write(line + '\n') |
| 77 | except Exception: |
| 78 | pass |
| 79 | |
| 80 | |
| 81 | def http_get(url): |
| 82 | try: |
| 83 | r = subprocess.run(['curl', '-sL', '--max-time', '20', '-A', UA, url], |
| 84 | capture_output=True, text=True, timeout=25) |
| 85 | return r.stdout |
| 86 | except Exception: |
| 87 | return '' |
| 88 | |
| 89 | |
| 90 | def bing_rss(query, count=20): |
| 91 | """cn.bing.com RSS 搜索（免 key）。返回 [(title, link, pub_date, desc)]""" |
| 92 | url = 'https://cn.bing.com/search?q=%s&format=rss&count=%d' % ( |
| 93 | urllib.parse.quote(query), count) |
| 94 | xml = http_get(url) |
| 95 | if not xml or '<item>' not in xml: |
| 96 | return [] |
| 97 | out = [] |
| 98 | for it in re.findall(r'<item>(.*?)</item>', xml, flags=re.S): |
| 99 | def g(tag): |
| 100 | m = re.search(r'<%s[^>]*>(.*?)</%s>' % (tag, tag), it, flags=re.S) |
| 101 | return m.group(1).strip() if m else '' |
| 102 | title = g('title') |
| 103 | link = g('link') |
| 104 | pub = g('pubDate') |
| 105 | desc = g('description') |
| 106 | if title and link: |
| 107 | out.append({'title': title, 'link': link, 'pub': pub, 'desc': desc}) |
| 108 | return out |
| 109 | |
| 110 | |
| 111 | def parse_pub(s): |
| 112 | """Bing RSS 中文格式日期：'周二, 27 10月 2026 20:16:00 GMT' → datetime 或 None""" |
| 113 | if not s: |
| 114 | return None |
| 115 | # 中文格式：'周X, 27 10月 2026 20:16:00 GMT' |
| 116 | m = re.search(r'(\d{1,2}) (\d{1,2})月 (\d{4}) (\d{2}):(\d{2}):(\d{2})', s) |
| 117 | if m: |
| 118 | try: |
| 119 | return datetime(int(m.group(3)), int(m.group(2)), int(m.group(1)), |
| 120 | int(m.group(4)), int(m.group(5)), int(m.group(6))) |
| 121 | except ValueError: |
| 122 | return None |
| 123 | # 也可能 '27 Oct 2026 ...' 或 '2026-10-27' |
| 124 | for fmt in ('%a, %d %b %Y %H:%M:%S %Z', '%d %b %Y %H:%M:%S %Z', |
| 125 | '%Y-%m-%d %H:%M:%S', '%Y-%m-%d'): |
| 126 | try: |
| 127 | return datetime.strptime(s.strip(), fmt) |
| 128 | except ValueError: |
| 129 | continue |
| 130 | return None |
| 131 | |
| 132 | |
| 133 | def is_interesting(item): |
| 134 | t = item['title'] |
| 135 | low = t |
| 136 | for ex in EXCLUDE_KW: |
| 137 | if ex in low: |
| 138 | return False |
| 139 | if not any(k in low for k in RELEVANT_KW): |
| 140 | return False |
| 141 | if not any(k in low for k in INTEREST_KW): |
| 142 | return False |
| 143 | # 精确 URL 黑名单（老官宣/已核实噪音，按链接排不误伤真新品） |
| 144 | if any(item['link'].startswith(u) for u in EXCLUDE_LINKS): |
| 145 | return False |
| 146 | return True |
| 147 | |
| 148 | |
| 149 | # ---- 多源交叉验证（治本方案）---- |
| 150 | # 原理：真正的新品官宣会被多个来源同时报道（微博官宣 + B站/抖音视频 + sina 新闻转载）， |
| 151 | # 而 Bing 伪标日期的老官宣通常是"孤立微博"（只有微博一条被标成近期）。 |
| 152 | # 做法：对每条候选，提取标题里的产品词，用「蜜雪冰城 + 产品词」二次搜索， |
| 153 | # 统计窗口内出现的非微博来源域名数：≥1 源佐证 = 真新品；0 源 = 疑似旧闻。 |
| 154 | |
| 155 | # 已知具体产品词（提取产品关键词时优先匹配） |
| 156 | PROD_WORDS = ['芋泥', '奶麻薯', '大樱桃', '樱桃酸奶', '抹茶大新地', '雪王大新地', |
| 157 | '大新地', '抹茶大圣代', '大圣代', '鲜牛乳', '小丸子', '冰杯'] |
| 158 | |
| 159 | |
| 160 | def extract_prod_kw(title): |
| 161 | """从标题提取产品关键词：优先取 #话题#（去品牌/模板词），其次已知产品词，最后标题清洗兜底""" |
| 162 | for tag in re.findall(r'#([^#]+)#', title): |
| 163 | tt = re.sub(r'蜜雪冰城|雪王|mixue|Mixue|新品|上市|上新|联名|回归|限定|新口味|' |
| 164 | r'新包装|新系列|首发|快闪|周边|啦', '', tag).strip() |
| 165 | if len(tt) >= 2: |
| 166 | return tt[:10] |
| 167 | for w in PROD_WORDS: |
| 168 | if w in title: |
| 169 | return w |
| 170 | t = re.sub(r'蜜雪冰城|雪王|新品|上市啦|上新|联名|回归|限定|新口味|新包装|新系列|' |
| 171 | r'首发|快闪|周边|Get|get|一杯|双重口感|转\+关|请喝|杯|！|啦|上市|' |
| 172 | r'，|,|。|\.|（|）|\(|\)', ' ', title) |
| 173 | t = re.sub(r'\s+', ' ', t).strip() |
| 174 | if t and len(t) <= 12: |
| 175 | return t |
| 176 | return '' |
| 177 | |
| 178 | |
| 179 | def cross_verify(it, cutoff): |
| 180 | """多源交叉验证：返回 (通过与否, 说明)。窗口内 ≥1 个非微博来源佐证 → 通过""" |
| 181 | kw = extract_prod_kw(it['title']) |
| 182 | if not kw: |
| 183 | return False, '提取不出产品词' |
| 184 | hits = bing_rss('蜜雪冰城 ' + kw, count=15) |
| 185 | sources = set() |
| 186 | for h in hits: |
| 187 | if h['link'] == it['link']: |
| 188 | continue # 自己不算佐证 |
| 189 | dom = urllib.parse.urlparse(h['link']).netloc |
| 190 | if 'weibo.com' in dom: |
| 191 | continue # 微博不算佐证（Bing 伪标日期重灾区） |
| 192 | if not is_interesting(h): |
| 193 | continue # 佐证也要是蜜雪相关（挡字典/百科噪音） |
| 194 | p = parse_pub(h.get('pub', '')) |
| 195 | if p and p.replace(tzinfo=dt_tz.utc) >= cutoff: |
| 196 | sources.add(dom) |
| 197 | if sources: |
| 198 | return True, '佐证: ' + ','.join(sorted(sources)) |
| 199 | return False, '无其他来源佐证（疑似伪标旧闻）' |
| 200 | |
| 201 | |
| 202 | def load_state(): |
| 203 | if os.path.exists(STATE_FILE): |
| 204 | try: |
| 205 | with open(STATE_FILE, encoding='utf-8') as f: |
| 206 | return json.load(f) |
| 207 | except Exception: |
| 208 | pass |
| 209 | return {'seen': {}, 'last_run': None} |
| 210 | |
| 211 | |
| 212 | def save_state(st): |
| 213 | with open(STATE_FILE, 'w', encoding='utf-8') as f: |
| 214 | json.dump(st, f, ensure_ascii=False, indent=2) |
| 215 | |
| 216 | |
| 217 | def send_qq(text): |
| 218 | if not ONE_BOT_TOKEN or not OWNER_QQ: |
| 219 | log('(通知未配置：ONE_BOT_TOKEN / QQ_USER_ID 未设置，仅打印)') |
| 220 | return '' |
| 221 | payload = {'user_id': str(OWNER_QQ), 'message': text} |
| 222 | try: |
| 223 | r = subprocess.run( |
| 224 | ['curl', '-s', '--max-time', '15', '-X', 'POST', ONE_BOT_URL, |
| 225 | '-H', 'Authorization: Bearer %s' % ONE_BOT_TOKEN, |
| 226 | '-H', 'Content-Type: application/json', |
| 227 | '-d', json.dumps(payload, ensure_ascii=False)], |
| 228 | capture_output=True, text=True, timeout=20) |
| 229 | return r.stdout |
| 230 | except Exception as exc: |
| 231 | log('推送失败: %s' % exc) |
| 232 | return '' |
| 233 | |
| 234 | |
| 235 | def fmt_item(it): |
| 236 | return f"・{it['title']}\n {it['link']}" |
| 237 | |
| 238 | |
| 239 | def main(): |
| 240 | ap = argparse.ArgumentParser() |
| 241 | ap.add_argument('--dry-run', action='store_true') |
| 242 | ap.add_argument('--days', type=int, default=WINDOW_DAYS) |
| 243 | ap.add_argument('--list', action='store_true') |
| 244 | args = ap.parse_args() |
| 245 | |
| 246 | if args.list: |
| 247 | st = load_state() |
| 248 | for k, v in st['seen'].items(): |
| 249 | print(f'{v.get("time", "?")} {k}') |
| 250 | return |
| 251 | |
| 252 | days = args.days if args.days > 0 else WINDOW_DAYS |
| 253 | cutoff = datetime.now(dt_tz.utc) - timedelta(days=days) |
| 254 | st = load_state() |
| 255 | seen = st.get('seen', {}) |
| 256 | |
| 257 | all_items = [] |
| 258 | for q in QUERIES: |
| 259 | try: |
| 260 | items = bing_rss(q) |
| 261 | except Exception as e: |
| 262 | log(f'搜索 {q} 失败: {e}') |
| 263 | items = [] |
| 264 | log(f'搜索「{q}」→ {len(items)} 条') |
| 265 | all_items.extend(items) |
| 266 | |
| 267 | # 去重（同 link 只留一次） |
| 268 | by_link = {} |
| 269 | for it in all_items: |
| 270 | by_link.setdefault(it['link'], it) |
| 271 | |
| 272 | # 过滤：相关性 + 时效 |
| 273 | fresh = [] |
| 274 | for it in by_link.values(): |
| 275 | if not is_interesting(it): |
| 276 | continue |
| 277 | pub = parse_pub(it.get('pub', '')) |
| 278 | if pub is None: |
| 279 | # 日期解析不了就当作近期，靠去重兜底 |
| 280 | fresh.append((it, None)) |
| 281 | elif pub.replace(tzinfo=dt_tz.utc) >= cutoff: |
| 282 | fresh.append((it, pub)) |
| 283 | fresh.sort(key=lambda x: ((x[1] or datetime.now(dt_tz.utc)).replace(tzinfo=dt_tz.utc) if x[1] else datetime.now(dt_tz.utc)), reverse=True) |
| 284 | |
| 285 | # 多源交叉验证：窗口内需有 ≥1 个其他来源佐证，孤立微博(伪标旧闻)不推 |
| 286 | verified = [] |
| 287 | for it, pub in fresh: |
| 288 | ok, why = cross_verify(it, cutoff) |
| 289 | if ok: |
| 290 | verified.append((it, pub)) |
| 291 | else: |
| 292 | log(f'交叉验证未通过（{why}）：{it["title"][:40]}') |
| 293 | fresh = verified |
| 294 | |
| 295 | # 从未推送的（按 title+link 去重） |
| 296 | new_items = [] |
| 297 | for it, pub in fresh: |
| 298 | key = it['link'] |
| 299 | if key in seen: |
| 300 | continue |
| 301 | new_items.append((it, pub, key)) |
| 302 | |
| 303 | log(f'过滤后 {len(fresh)} 条相关，其中未推送 {len(new_items)} 条') |
| 304 | |
| 305 | if not new_items: |
| 306 | log('没有新的蜜雪新品条目，跳过') |
| 307 | st['last_run'] = datetime.now().isoformat() |
| 308 | save_state(st) |
| 309 | return |
| 310 | |
| 311 | lines = ['🍦 蜜雪冰城动态（最近 %d 天）' % days] |
| 312 | for it, pub, key in new_items[:8]: |
| 313 | lines.append(fmt_item(it)) |
| 314 | if pub: |
| 315 | lines[-1] += f"\n 时间：{pub.strftime('%m-%d %H:%M')}" |
| 316 | text = '\n'.join(lines) |
| 317 | |
| 318 | if args.dry_run: |
| 319 | log('[dry-run] 将推送：\n' + text) |
| 320 | log(f'[dry-run] 共 {len(new_items[:8])} 条（不推送、不写 state）') |
| 321 | return |
| 322 | |
| 323 | resp = send_qq(text) |
| 324 | log(f'推送响应: {resp}') |
| 325 | for it, pub, key in new_items[:8]: |
| 326 | seen[key] = {'time': datetime.now().isoformat(), 'title': it['title']} |
| 327 | st['last_run'] = datetime.now().isoformat() |
| 328 | save_state(st) |
| 329 | log(f'已推送并记录 {len(new_items[:8])} 条') |
| 330 | |
| 331 | |
| 332 | if __name__ == '__main__': |
| 333 | main() |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
