From e57e5219000d7446d8dbf929a86bac98674a2a80 Mon Sep 17 00:00:00 2001 From: liuwei Date: Thu, 16 Apr 2026 17:59:26 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96Epic=E5=85=8D=E8=B4=B9?= =?UTF-8?q?=E6=B8=B8=E6=88=8F=E6=8E=A8=E9=80=81=EF=BC=9A=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E5=AE=98=E7=BD=91=E6=B4=BB=E5=8A=A8=E5=B9=B6=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E5=8F=98=E5=8C=96=E6=8E=A8=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 变更项：1) 数据源切换为Epic官方freeGamesPromotions接口，不再依赖第三方页面抓取。2) 活动识别升级：区分当前可领与即将免费，仅提取0%折扣活动，按活动窗口过滤脏数据。3) 推送策略从周五固定时点改为小时级检查，并支持 only_on_change 变化去重，避免重复刷屏。4) 新增地区/语言参数（locale、country、allow_countries）与是否包含即将免费配置。5) 增强推送内容：发行商、原价到现价、开始/截止时间、直达链接，信息更贴近官网展示。6) 增加Redis摘要缓存与中文注释，保证活动变化判断稳定可追踪。 --- plugins/epic_free/main.py | 335 +++++++++++++++++++++++++++++++------- 1 file changed, 280 insertions(+), 55 deletions(-) diff --git a/plugins/epic_free/main.py b/plugins/epic_free/main.py index 00388d9..8f6ba9c 100644 --- a/plugins/epic_free/main.py +++ b/plugins/epic_free/main.py @@ -1,13 +1,14 @@ # -*- coding: utf-8 -*- -from datetime import datetime +import hashlib +import json +from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple import requests -from bs4 import BeautifulSoup from base.plugin_common.message_plugin_interface import MessagePluginInterface from base.plugin_common.plugin_interface import PluginStatus -from utils.robot_cmd.robot_command import GroupBotManager +from utils.robot_cmd.robot_command import GroupBotManager, PermissionStatus class EpicFreePlugin(MessagePluginInterface): @@ -22,7 +23,7 @@ class EpicFreePlugin(MessagePluginInterface): @property def version(self) -> str: - return "1.0.0" + return "1.1.0" @property def description(self) -> str: @@ -47,8 +48,11 @@ class EpicFreePlugin(MessagePluginInterface): def __init__(self): super().__init__() self.feature = self.register_feature() + self.db_manager = None def initialize(self, context: Dict[str, Any]) -> bool: + # 保留数据库管理器，用于读取 Redis 做“活动变化去重推送”。 + self.db_manager = context.get("db_manager") return True def start(self) -> bool: @@ -70,12 +74,20 @@ class EpicFreePlugin(MessagePluginInterface): { "action_key": "weekly_free_games_push", "name": "Epic免费游戏推送", - "description": "每周五推送 Epic 当周免费游戏", - "trigger_type": "every_weekday_time", - "trigger_config": {"weekday": 4, "time_str": "10:00"}, + "description": "按官网活动信息检查 Epic 免费游戏，变化时推送", + # 用小时级轮询替代“仅周五固定时刻”，避免错过非该时点仍可领取的活动窗口。 + "trigger_type": "every_seconds", + "trigger_config": {"seconds": 3600}, "target_scope": "all_enabled_groups", "target_config": {}, - "payload": {"force": False}, + "payload": { + "force": False, + "only_on_change": True, + "include_upcoming": True, + "locale": "zh-CN", + "country": "CN", + "allow_countries": "CN", + }, "default_enabled": True, } ] @@ -92,24 +104,55 @@ class EpicFreePlugin(MessagePluginInterface): payload = context.get("payload") or {} force = bool(payload.get("force", False)) - if not force and not self._is_friday(): - # 非周五时默认跳过；手动触发可通过 payload.force 强制执行。 - return {"success": True, "summary": "今天不是周五，已跳过 Epic 播报", "detail": {"skipped": True}} + only_on_change = bool(payload.get("only_on_change", True)) + include_upcoming = bool(payload.get("include_upcoming", True)) + locale = str(payload.get("locale") or "zh-CN").strip() or "zh-CN" + country = str(payload.get("country") or "CN").strip() or "CN" + allow_countries = str(payload.get("allow_countries") or country).strip() or country target_groups = [str(g).strip() for g in (context.get("target_groups") or []) if str(g).strip()] if not target_groups: target_groups = [ gid for gid in GroupBotManager.get_group_list() - if GroupBotManager.get_group_permission(gid, self.feature).value == "enabled" + if GroupBotManager.get_group_permission(gid, self.feature) == PermissionStatus.ENABLED ] if not target_groups: return {"success": False, "summary": "没有可推送目标群", "detail": {"target_count": 0}} try: - text = self._get_free_games() + games = self._fetch_official_free_games( + locale=locale, + country=country, + allow_countries=allow_countries, + include_upcoming=include_upcoming, + ) except Exception as e: return {"success": False, "summary": f"获取 Epic 免费游戏失败: {e}", "detail": {"error": str(e)}} + current_count = sum(1 for item in games if item.get("state") == "current") + upcoming_count = sum(1 for item in games if item.get("state") == "upcoming") + digest = self._build_games_digest(games) + if only_on_change and (not force): + last_digest = self._get_last_digest(locale=locale, country=country, include_upcoming=include_upcoming) + if last_digest and last_digest == digest: + return { + "success": True, + "summary": "Epic 活动无变化，已跳过推送", + "detail": { + "skipped": True, + "reason": "no_change", + "current_count": current_count, + "upcoming_count": upcoming_count, + }, + } + + text = self._render_push_text( + games=games, + locale=locale, + country=country, + include_upcoming=include_upcoming, + ) + success_groups = [] failed_groups = {} for gid in target_groups: @@ -119,6 +162,14 @@ class EpicFreePlugin(MessagePluginInterface): except Exception as e: failed_groups[gid] = str(e) + if len(success_groups) > 0: + self._set_last_digest( + locale=locale, + country=country, + include_upcoming=include_upcoming, + digest=digest, + ) + return { "success": len(failed_groups) == 0, "summary": f"Epic播报完成: 成功{len(success_groups)}群, 失败{len(failed_groups)}群", @@ -127,60 +178,234 @@ class EpicFreePlugin(MessagePluginInterface): "success_groups": success_groups, "failed_groups": failed_groups, "force": force, + "only_on_change": only_on_change, + "include_upcoming": include_upcoming, + "locale": locale, + "country": country, + "current_count": current_count, + "upcoming_count": upcoming_count, }, } @staticmethod - def _is_friday() -> bool: - """判断是否周五（插件内实现）。""" - return datetime.today().weekday() == 4 + def _parse_iso_to_utc(text: str) -> Optional[datetime]: + if not text: + return None + value = str(text).strip() + try: + # Epic 常见格式：2026-04-16T15:00:00.000Z + if value.endswith("Z"): + value = value[:-1] + "+00:00" + return datetime.fromisoformat(value).astimezone(timezone.utc) + except Exception: + return None @staticmethod - def _get_free_games() -> str: - """抓取 Epic 免费游戏列表（插件内实现）。""" - url = "https://steamstats.cn/xi" + def _fmt_utc(dt: Optional[datetime]) -> str: + if not dt: + return "-" + return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + + @staticmethod + def _safe_offer_link(locale: str, element: Dict[str, Any]) -> str: + locale_text = str(locale or "en-US") + mappings = ((element.get("catalogNs") or {}).get("mappings") or []) + if isinstance(mappings, list) and mappings: + page_slug = str((mappings[0] or {}).get("pageSlug") or "").strip() + if page_slug: + return f"https://store.epicgames.com/{locale_text}/p/{page_slug}" + + product_slug = str(element.get("productSlug") or "").strip() + if product_slug: + product_slug = product_slug.replace("/home", "").strip("/") + if product_slug: + return f"https://store.epicgames.com/{locale_text}/p/{product_slug}" + + return f"https://store.epicgames.com/{locale_text}/free-games" + + def _fetch_official_free_games( + self, + locale: str, + country: str, + allow_countries: str, + include_upcoming: bool, + ) -> List[Dict[str, Any]]: + """从 Epic 官方 freeGamesPromotions 接口提取免费活动。 + + 提取规则： + 1. 只保留折扣类型为 PERCENTAGE 且折扣为 0 的活动（官网“免费领取”语义）。 + 2. promotionalOffers -> 当前可领；upcomingPromotionalOffers -> 即将可领。 + 3. 默认返回当前+即将，按活动时间排序，便于推送阅读。 + """ + api = ( + "https://store-site-backend-static.ak.epicgames.com/freeGamesPromotions" + f"?locale={locale}&country={country}&allowCountries={allow_countries}" + ) headers = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/90.0.4430.72 Safari/537.36 Edg/90.0.818.41" + "(KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36" ) } - resp = requests.get(url, headers=headers, timeout=20) - resp.raise_for_status() - resp.encoding = resp.apparent_encoding - soup = BeautifulSoup(resp.text, "html.parser") - text = "今日喜加一 :https://store.epicgames.com/en-US/free-games\n" + response = requests.get(api, headers=headers, timeout=20) + response.raise_for_status() + payload = response.json() or {} + elements = (((payload.get("data") or {}).get("Catalog") or {}).get("searchStore") or {}).get("elements") or [] - tbody = soup.find("tbody") - if not tbody: - return text + "未抓取到免费游戏列表" - - rows = tbody.find_all("tr") - idx = 1 - for row in rows: - cols = row.find_all("td") - if len(cols) < 7: + now_utc = datetime.now(timezone.utc) + records: List[Dict[str, Any]] = [] + for element in elements: + title = str(element.get("title") or "").strip() + if not title: continue - name = (cols[1].string or "").strip() - gametype = (cols[2].string or "").replace(" ", "").strip() - start = (cols[3].string or "").replace(" ", "").strip() - end = (cols[4].string or "").replace(" ", "").strip() - permanent = (cols[5].string or "").replace(" ", "").strip() - origin_span = cols[6].find("span") - origin = (origin_span.string or "").replace(" ", "").strip() if origin_span else "" - href_value = "" - for a in cols[6].find_all("a"): - href_value = a.get("href", "") or href_value - text += ( - f"序号：{idx}\n" - f"游戏名称：{name}\n" - f"DLC/game：{gametype}\n" - f"开始时间：{start}\n" - f"结束时间：{end}\n" - f"是否永久：{permanent}\n" - f"平台：{origin}\n" - f"URL：{href_value}\n" + seller_name = str((element.get("seller") or {}).get("name") or "").strip() + total_price = ((element.get("price") or {}).get("totalPrice") or {}) + fmt_price = total_price.get("fmtPrice") or {} + original_price_text = str(fmt_price.get("originalPrice") or "").strip() + discount_price_text = str(fmt_price.get("discountPrice") or "").strip() + + # 先拆出当前活动窗口。 + promotions = element.get("promotions") or {} + promo_groups = [ + ("current", promotions.get("promotionalOffers") or []), + ("upcoming", promotions.get("upcomingPromotionalOffers") or []), + ] + for state, buckets in promo_groups: + if state == "upcoming" and not include_upcoming: + continue + for bucket in buckets: + for promo in (bucket or {}).get("promotionalOffers", []) or []: + discount_setting = promo.get("discountSetting") or {} + discount_type = str(discount_setting.get("discountType") or "").upper() + discount_percent = int(discount_setting.get("discountPercentage") or -1) + # 仅保留“0%折扣（即免费）”活动，避免把普通折扣活动混进来。 + if not (discount_type == "PERCENTAGE" and discount_percent == 0): + continue + + start_utc = self._parse_iso_to_utc(promo.get("startDate")) + end_utc = self._parse_iso_to_utc(promo.get("endDate")) + # 即将活动需在未来；当前活动需在窗口内，避免脏数据导致状态错乱。 + if state == "upcoming" and start_utc and start_utc <= now_utc: + continue + if state == "current" and start_utc and end_utc and not (start_utc <= now_utc < end_utc): + continue + + record = { + "offer_id": str(element.get("id") or ""), + "namespace": str(element.get("namespace") or ""), + "title": title, + "description": str(element.get("description") or "").strip(), + "seller": seller_name, + "state": state, + "start_utc": start_utc, + "end_utc": end_utc, + "original_price_text": original_price_text or "-", + "discount_price_text": discount_price_text or "0", + "link": self._safe_offer_link(locale=locale, element=element), + } + records.append(record) + + # 去重：同一 offer 在同一状态下可能有重复 bucket，按活动窗口去重。 + uniq = {} + for item in records: + key = ( + item.get("offer_id"), + item.get("state"), + self._fmt_utc(item.get("start_utc")), + self._fmt_utc(item.get("end_utc")), ) - idx += 1 - return text + uniq[key] = item + records = list(uniq.values()) + records.sort( + key=lambda x: ( + 0 if x.get("state") == "current" else 1, + x.get("end_utc") or datetime.max.replace(tzinfo=timezone.utc), + x.get("start_utc") or datetime.max.replace(tzinfo=timezone.utc), + x.get("title") or "", + ) + ) + return records + + def _build_games_digest(self, games: List[Dict[str, Any]]) -> str: + # 用稳定字段计算摘要，做“活动变化去重推送”。 + norm = [] + for item in games: + norm.append( + { + "offer_id": item.get("offer_id", ""), + "state": item.get("state", ""), + "start": self._fmt_utc(item.get("start_utc")), + "end": self._fmt_utc(item.get("end_utc")), + "original_price": item.get("original_price_text", ""), + "discount_price": item.get("discount_price_text", ""), + } + ) + payload = json.dumps(norm, ensure_ascii=False, sort_keys=True) + return hashlib.md5(payload.encode("utf-8")).hexdigest() + + def _redis_key_digest(self, locale: str, country: str, include_upcoming: bool) -> str: + return f"abot:epic_free:last_digest:{locale}:{country}:{1 if include_upcoming else 0}" + + def _get_last_digest(self, locale: str, country: str, include_upcoming: bool) -> str: + if not self.db_manager: + return "" + try: + redis_conn = self.db_manager.get_redis_connection() + value = redis_conn.get(self._redis_key_digest(locale, country, include_upcoming)) + if isinstance(value, bytes): + return value.decode("utf-8", errors="ignore") + return str(value or "") + except Exception: + return "" + + def _set_last_digest(self, locale: str, country: str, include_upcoming: bool, digest: str): + if not self.db_manager: + return + try: + redis_conn = self.db_manager.get_redis_connection() + redis_conn.set(self._redis_key_digest(locale, country, include_upcoming), digest) + except Exception: + # 去重写失败不影响主流程推送。 + return + + def _render_push_text( + self, + games: List[Dict[str, Any]], + locale: str, + country: str, + include_upcoming: bool, + ) -> str: + current_games = [item for item in games if item.get("state") == "current"] + upcoming_games = [item for item in games if item.get("state") == "upcoming"] + + lines = [] + lines.append("Epic 官方免费活动播报") + lines.append(f"地区: {country} 语言: {locale}") + lines.append(f"官方页: https://store.epicgames.com/{locale}/free-games") + lines.append("") + + if current_games: + lines.append(f"当前可领取 ({len(current_games)}):") + for idx, item in enumerate(current_games, 1): + lines.append(f"{idx}. {item.get('title')}") + lines.append(f" 发行商: {item.get('seller') or '-'}") + lines.append(f" 价格: {item.get('original_price_text')} -> {item.get('discount_price_text')}") + lines.append(f" 截止: {self._fmt_utc(item.get('end_utc'))}") + lines.append(f" 链接: {item.get('link')}") + else: + lines.append("当前可领取: 暂无") + + if include_upcoming: + lines.append("") + if upcoming_games: + lines.append(f"即将免费 ({len(upcoming_games)}):") + for idx, item in enumerate(upcoming_games, 1): + lines.append(f"{idx}. {item.get('title')}") + lines.append(f" 开始: {self._fmt_utc(item.get('start_utc'))}") + lines.append(f" 结束: {self._fmt_utc(item.get('end_utc'))}") + lines.append(f" 链接: {item.get('link')}") + else: + lines.append("即将免费: 暂无") + + return "\n".join(lines)