member_context: split daily weekly monthly digest scheduling

plugins/member_context/digest_service.py

@@ -11,7 +11,6 @@ from db.contacts_db import ContactsDBOperator
 from db.member_digest_db import MemberDigestDBOperator
 from db.message_storage import MessageStorageDB
 from plugins.member_context.dify_client import DifyClient
-from plugins.member_context.prompt_builder import MemberContextPromptBuilder
 from utils.compress_chat_data import compress_chat_data
 
 
@@ -107,43 +106,94 @@ class MemberDigestService:
             self.LOG.warning(f"[成员交互摘要] 检查群初始化状态失败，按增量处理: group={chatroom_id}, error={e}")
             return False
 
-    def ensure_member_digest_pipeline(self, chatroom_id: str, wxid: str, force: bool = False) -> Dict:
+    def ensure_member_digest_pipeline(self, chatroom_id: str, wxid: str, force: bool = False,
+                                      enable_weekly: bool = True, enable_monthly: bool = True) -> Dict:
         member = self.contacts_db.get_chatroom_member_info(chatroom_id, wxid) or {}
         display_name = member.get("display_name") or member.get("nick_name") or wxid
 
-        daily_digests = self.digest_db.list_digests(chatroom_id, wxid, "daily", limit=400)
-        if not daily_digests:
+        all_daily_digests = self.digest_db.list_digests(chatroom_id, wxid, "daily", limit=400)
+        if not all_daily_digests:
             return {
                 "display_name": display_name,
                 "daily_digests": [],
                 "weekly_digests": [],
                 "monthly_digests": [],
+                "all_daily_digests": [],
+                "all_weekly_digests": [],
+                "all_monthly_digests": [],
                 "stats": {"daily": 0, "weekly": 0, "monthly": 0, "active_days": 0, "built_daily": 0},
             }
 
-        built_weekly = self._ensure_weekly_digests(chatroom_id, wxid, display_name, force=force)
-        built_monthly = self._ensure_monthly_digests(chatroom_id, wxid, display_name, force=force)
+        latest_daily_date = self._extract_latest_daily_date(all_daily_digests)
+        built_weekly = 0
+        built_monthly = 0
+        if enable_weekly and (force or self._should_run_weekly(latest_daily_date)):
+            built_weekly = self._ensure_weekly_digests(chatroom_id, wxid, display_name, force=force)
+        elif enable_weekly:
+            self.LOG.debug(
+                f"[成员交互摘要][周摘要] 本次跳过(未到周处理窗口): "
+                f"group={chatroom_id}, wxid={wxid}, latest_daily_date={latest_daily_date}"
+            )
 
-        daily_digests = self.digest_db.list_digests(chatroom_id, wxid, "daily", limit=self.final_daily_limit)
-        weekly_digests = self.digest_db.list_digests(chatroom_id, wxid, "weekly", limit=self.final_weekly_limit)
-        monthly_digests = self.digest_db.list_digests(chatroom_id, wxid, "monthly", limit=self.final_monthly_limit)
+        if enable_monthly and (force or self._should_run_monthly(latest_daily_date)):
+            built_monthly = self._ensure_monthly_digests(chatroom_id, wxid, display_name, force=force)
+        elif enable_monthly:
+            self.LOG.debug(
+                f"[成员交互摘要][月摘要] 本次跳过(未到月处理窗口): "
+                f"group={chatroom_id}, wxid={wxid}, latest_daily_date={latest_daily_date}"
+            )
+
+        all_weekly_digests = self.digest_db.list_digests(chatroom_id, wxid, "weekly", limit=200)
+        all_monthly_digests = self.digest_db.list_digests(chatroom_id, wxid, "monthly", limit=120)
+
+        daily_digests = all_daily_digests[:self.final_daily_limit]
+        weekly_digests = all_weekly_digests[:self.final_weekly_limit]
+        monthly_digests = all_monthly_digests[:self.final_monthly_limit]
 
         return {
             "display_name": display_name,
             "daily_digests": daily_digests,
             "weekly_digests": weekly_digests,
             "monthly_digests": monthly_digests,
+            "all_daily_digests": all_daily_digests,
+            "all_weekly_digests": all_weekly_digests,
+            "all_monthly_digests": all_monthly_digests,
             "stats": {
-                "daily": len(daily_digests),
-                "weekly": len(weekly_digests),
-                "monthly": len(monthly_digests),
-                "active_days": len(self.digest_db.list_digest_keys(chatroom_id, wxid, "daily")),
+                "daily": len(all_daily_digests),
+                "weekly": len(all_weekly_digests),
+                "monthly": len(all_monthly_digests),
+                "active_days": len(all_daily_digests),
                 "built_daily": 0,
                 "built_weekly": built_weekly,
                 "built_monthly": built_monthly,
             },
         }
 
+    def _extract_latest_daily_date(self, daily_digests: List[Dict]) -> Optional[datetime]:
+        if not daily_digests:
+            return None
+        latest_key = daily_digests[0].get("period_key") or daily_digests[0].get("period_end")
+        return self._parse_period_date(latest_key)
+
+    @staticmethod
+    def _parse_period_date(value: Optional[str]) -> Optional[datetime]:
+        if not value:
+            return None
+        try:
+            return datetime.strptime(str(value)[:10], "%Y-%m-%d")
+        except Exception:
+            return None
+
+    def _should_run_weekly(self, latest_daily_date: Optional[datetime]) -> bool:
+        if not latest_daily_date:
+            return False
+        return latest_daily_date.weekday() == 6
+
+    def _should_run_monthly(self, latest_daily_date: Optional[datetime]) -> bool:
+        if not latest_daily_date:
+            return False
+        return (latest_daily_date + timedelta(days=1)).day == 1
+
     @staticmethod
     def _normalize_profile_item(item: Dict) -> Dict:
         normalized = {}
@@ -312,17 +362,24 @@ class MemberDigestService:
     def _build_period_digest(self, digest_type: str, chatroom_id: str, wxid: str, display_name: str,
                              period_key: str, period_start: str, period_end: str,
                              items: List[Dict]) -> Optional[Dict]:
-        prompt = MemberContextPromptBuilder.build_period_digest_prompt(
-            digest_type, chatroom_id, wxid, display_name, period_key, items
+        parsed = self._request_period_json(
+            digest_type=digest_type,
+            chatroom_id=chatroom_id,
+            wxid=wxid,
+            display_name=display_name,
+            period_key=period_key,
+            items=items,
         )
-        parsed = self._request_ai_json(prompt, tag=f"{digest_type}:{period_key}", chatroom_id=chatroom_id, wxid=wxid)
         if not parsed:
             self.LOG.warning(
                 f"[成员交互摘要][{digest_type}] 跳过周期摘要(未提取到有效结果): "
-                f"group={chatroom_id}, wxid={wxid}, period={period_key}, source_count={len(items)}"
+                f"group={chatroom_id}, wxid={wxid}, period={period_key}, source_count={len(items)}, "
+                f"last_error={self.dify_client.last_error}"
             )
             return None
 
+        parsed = self._normalize_profile_item(parsed)
+
         return {
             "chatroom_id": chatroom_id,
             "wxid": wxid,
@@ -355,18 +412,42 @@ class MemberDigestService:
             parsed["ai_usage"] = response.get("usage", {}) or {}
         return parsed
 
+    def _request_period_json(self, digest_type: str, chatroom_id: str, wxid: str,
+                             display_name: str, period_key: str, items: List[Dict]) -> Optional[Dict]:
+        if not self.dify_client.is_available():
+            return None
+
+        inputs = {
+            "digest_type": digest_type,
+            "chatroom_id": chatroom_id,
+            "wxid": wxid,
+            "display_name": display_name,
+            "period_key": period_key,
+            "source_items_json": json.dumps(self._build_period_source_items(items), ensure_ascii=False),
+            "source_item_count": str(len(items)),
+        }
+        response = self.dify_client.run(
+            prompt="",
+            user=f"member-digest:{chatroom_id}:{wxid}:{digest_type}:{period_key}",
+            inputs=inputs,
+            tag=f"{digest_type}:{period_key}",
+        )
+        if not response:
+            return None
+        parsed = self._parse_ai_answer(response.get("text", ""))
+        if parsed:
+            parsed["ai_usage"] = response.get("usage", {}) or {}
+        return parsed
+
     def _request_group_daily_json(self, chatroom_id: str, digest_date: str,
                                   member_labels: List[str], compressed_chat: str) -> List[Dict]:
         if not self.dify_client.is_available():
             return []
-        prompt = MemberContextPromptBuilder.build_group_daily_digest_prompt(
-            chatroom_id, digest_date, member_labels, compressed_chat
-        )
         response = self.dify_client.run(
-            prompt=prompt,
+            prompt="",
             user=f"member-digest:{chatroom_id}:group-daily:{digest_date}",
             inputs={
-                "query": prompt,
+                "digest_type": "daily",
                 "chatroom_id": chatroom_id,
                 "digest_date": digest_date,
                 "member_labels": "\n".join(member_labels),
@@ -379,6 +460,34 @@ class MemberDigestService:
         parsed = self._parse_group_daily_answer(response.get("text", ""))
         return parsed
 
+    @staticmethod
+    def _build_period_source_items(items: List[Dict]) -> List[Dict]:
+        source_items = []
+        for item in items:
+            structured = item.get("structured", {}) or {}
+            source_items.append({
+                "period_key": item.get("period_key"),
+                "summary_text": item.get("summary_text", ""),
+                "topics": structured.get("topics") or structured.get("stable_topics") or structured.get("long_term_topics") or [],
+                "discussion_scenarios": structured.get("discussion_scenarios") or structured.get("common_scenarios") or [],
+                "identity_clues": structured.get("identity_clues") or structured.get("identity_traits") or [],
+                "skill_signals": structured.get("skill_signals") or structured.get("skill_profile") or [],
+                "problem_solving_signals": structured.get("problem_solving_signals") or structured.get("problem_solving_profile") or [],
+                "family_signals": structured.get("family_signals") or structured.get("family_profile") or [],
+                "life_stage_signals": structured.get("life_stage_signals") or structured.get("life_stage_profile") or [],
+                "value_preferences": structured.get("value_preferences") or structured.get("value_profile") or [],
+                "habit_signals": structured.get("habit_signals") or structured.get("habit_patterns") or [],
+                "expression_markers": structured.get("expression_markers") or structured.get("expression_profile") or [],
+                "engagement_traits": structured.get("engagement_traits") or structured.get("stable_traits") or [],
+                "reply_entry_points": structured.get("reply_entry_points") or structured.get("reply_entry_profile") or [],
+                "reply_preferences": structured.get("reply_preferences") or structured.get("long_term_reply_preferences") or [],
+                "social_role": structured.get("social_role") or structured.get("group_role") or "",
+                "decision_style": structured.get("decision_style") or structured.get("decision_profile") or "",
+                "temperament_signal": structured.get("temperament_signal") or structured.get("temperament_tendency") or "",
+                "recent_state": structured.get("recent_state") or structured.get("phase_state") or [],
+            })
+        return source_items
+
     def _parse_ai_answer(self, answer: str) -> Optional[Dict]:
         if not answer:
             return None