abot/plugins/epic_free/main.py

# -*- coding: utf-8 -*-
from datetime import datetime
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


class EpicFreePlugin(MessagePluginInterface):
    """Epic 免费游戏自动播报插件。"""

    FEATURE_KEY = "EPIC"
    FEATURE_DESCRIPTION = "📊 EPIC自动播报 [每周五自动发送]"

    @property
    def name(self) -> str:
        return "Epic播报"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def description(self) -> str:
        return "将 Epic 免费游戏播报从系统任务迁移到插件任务。"

    @property
    def author(self) -> str:
        return "ABOT Team"

    @property
    def commands(self) -> List[str]:
        return []

    @property
    def feature_key(self) -> Optional[str]:
        return self.FEATURE_KEY

    @property
    def feature_description(self) -> Optional[str]:
        return self.FEATURE_DESCRIPTION

    def __init__(self):
        super().__init__()
        self.feature = self.register_feature()

    def initialize(self, context: Dict[str, Any]) -> bool:
        return True

    def start(self) -> bool:
        self.status = PluginStatus.RUNNING
        return True

    def stop(self) -> bool:
        self.status = PluginStatus.STOPPED
        return True

    def can_process(self, message: Dict[str, Any]) -> bool:
        return False

    async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
        return False, None

    def get_schedule_actions(self) -> List[Dict[str, Any]]:
        return [
            {
                "action_key": "weekly_free_games_push",
                "name": "Epic免费游戏推送",
                "description": "每周五推送 Epic 当周免费游戏",
                "trigger_type": "every_weekday_time",
                "trigger_config": {"weekday": 4, "time_str": "10:00"},
                "target_scope": "all_enabled_groups",
                "target_config": {},
                "payload": {"force": False},
                "default_enabled": True,
            }
        ]

    async def run_scheduled_action(self, action_key: str, context: Dict[str, Any]) -> Dict[str, Any]:
        if action_key != "weekly_free_games_push":
            return {
                "success": False,
                "summary": f"不支持的动作: {action_key}",
                "detail": {"action_key": action_key},
            }
        if not self.bot:
            return {"success": False, "summary": "bot 未注入", "detail": {}}

        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}}

        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 not target_groups:
            return {"success": False, "summary": "没有可推送目标群", "detail": {"target_count": 0}}

        try:
            text = self._get_free_games()
        except Exception as e:
            return {"success": False, "summary": f"获取 Epic 免费游戏失败: {e}", "detail": {"error": str(e)}}

        success_groups = []
        failed_groups = {}
        for gid in target_groups:
            try:
                await self.bot.send_text_message(gid, text)
                success_groups.append(gid)
            except Exception as e:
                failed_groups[gid] = str(e)

        return {
            "success": len(failed_groups) == 0,
            "summary": f"Epic播报完成: 成功{len(success_groups)}群, 失败{len(failed_groups)}群",
            "detail": {
                "target_count": len(target_groups),
                "success_groups": success_groups,
                "failed_groups": failed_groups,
                "force": force,
            },
        }

    @staticmethod
    def _is_friday() -> bool:
        """判断是否周五（插件内实现）。"""
        return datetime.today().weekday() == 4

    @staticmethod
    def _get_free_games() -> str:
        """抓取 Epic 免费游戏列表（插件内实现）。"""
        url = "https://steamstats.cn/xi"
        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"
            )
        }
        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"

        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:
                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"
            )
            idx += 1
        return text