abot/plugins/member_context/main.py

from typing import Dict, Any, Tuple, Optional, List

from base.plugin_common.message_plugin_interface import MessagePluginInterface
from base.plugin_common.plugin_interface import PluginStatus
from plugins.member_context.service import MemberContextService


class MemberContextPlugin(MessagePluginInterface):
    """群成员交互摘要后台插件"""

    FEATURE_KEY = "MEMBER_CONTEXT_CAPABILITY"
    FEATURE_DESCRIPTION = "🧠 成员交互摘要 [后台AI提取，仅对启用群生效]"

    @property
    def name(self) -> str:
        return "成员交互摘要"

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

    @property
    def description(self) -> str:
        return "为群成员生成后台交互摘要，并按群功能开关控制"

    @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()
        self.service: Optional[MemberContextService] = None

    def initialize(self, context: Dict[str, Any]) -> bool:
        self.LOG.debug(f"正在初始化 {self.name} 插件...")
        self.service = MemberContextService(context["db_manager"], self._config)
        self.LOG.debug(f"{self.name} 插件初始化完成")
        return True

    def start(self) -> bool:
        self.LOG.debug(f"[{self.name}] 插件已启动")
        self.status = PluginStatus.RUNNING
        return True

    def stop(self) -> bool:
        self.LOG.info(f"[{self.name}] 插件已停止")
        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]]:
        """把成员画像日/周/月刷新动作声明给统一调度中心。"""
        schedule_cfg = self._config.get("schedule", {}) or {}
        refresh_times = schedule_cfg.get("refresh_times", []) or []
        weekly_refresh_time = str(schedule_cfg.get("weekly_refresh_time", "") or "").strip()
        monthly_refresh_time = str(schedule_cfg.get("monthly_refresh_time", "") or "").strip()

        actions: List[Dict[str, Any]] = []
        if refresh_times:
            actions.append(
                {
                    "action_key": "daily_refresh",
                    "name": "成员画像日刷新",
                    "description": "刷新启用群的成员交互摘要（日任务）",
                    "trigger_type": "at_times",
                    "trigger_config": {"time_list": refresh_times},
                    "target_scope": "all_enabled_groups",
                    "target_config": {},
                    "payload": {},
                    "default_enabled": True,
                }
            )
        if weekly_refresh_time:
            actions.append(
                {
                    "action_key": "weekly_refresh",
                    "name": "成员画像周刷新",
                    "description": "刷新启用群的成员交互摘要（周任务，含周摘要）",
                    "trigger_type": "every_week_time",
                    "trigger_config": {"weekday": 6, "time_str": weekly_refresh_time},
                    "target_scope": "all_enabled_groups",
                    "target_config": {},
                    "payload": {},
                    "default_enabled": True,
                }
            )
        if monthly_refresh_time:
            actions.append(
                {
                    "action_key": "monthly_refresh",
                    "name": "成员画像月刷新",
                    "description": "刷新启用群的成员交互摘要（月任务，含月摘要）",
                    "trigger_type": "every_month_last_day_time",
                    "trigger_config": {"time_str": monthly_refresh_time},
                    "target_scope": "all_enabled_groups",
                    "target_config": {},
                    "payload": {},
                    "default_enabled": True,
                }
            )
        return actions

    async def run_scheduled_action(self, action_key: str, context: Dict[str, Any]) -> Dict[str, Any]:
        """执行成员画像后台调度动作。"""
        if not self.service:
            return {"success": False, "summary": "服务未初始化", "detail": {}}
        if action_key not in {"daily_refresh", "weekly_refresh", "monthly_refresh"}:
            return {"success": False, "summary": f"不支持的动作: {action_key}", "detail": {"action_key": action_key}}

        # 兼容“指定群执行”的场景；若未指定则沿用全量刷新逻辑。
        target_groups = [str(g).strip() for g in (context.get("target_groups") or []) if str(g).strip()]
        enable_weekly = action_key == "weekly_refresh"
        enable_monthly = action_key == "monthly_refresh"

        try:
            if target_groups:
                groups = 0
                members = 0
                skipped = 0
                for group_id in target_groups:
                    result = self.service.refresh_group_contexts(
                        group_id,
                        enable_weekly_digest=enable_weekly,
                        enable_monthly_digest=enable_monthly,
                    )
                    if result.get("disabled"):
                        continue
                    groups += 1
                    members += int(result.get("refreshed", 0))
                    skipped += int(result.get("skipped", 0))
                detail = {"groups": groups, "members": members, "skipped": skipped, "targeted": True}
            else:
                detail = self.service.refresh_all_chatrooms(
                    enable_weekly_digest=enable_weekly,
                    enable_monthly_digest=enable_monthly,
                )
                detail["targeted"] = False
            return {
                "success": True,
                "summary": f"成员画像刷新完成: 群{detail.get('groups', 0)}，成员{detail.get('members', 0)}",
                "detail": detail,
            }
        except Exception as e:
            return {"success": False, "summary": f"执行异常: {e}", "detail": {"error": str(e)}}