feat(douyu): add daily danmu report pipeline

plugins/douyu/main.py

@@ -1,7 +1,9 @@
 import asyncio
 import json
-from datetime import datetime
+from collections import Counter
+from datetime import datetime, timedelta
 import os
+from pathlib import Path
 import threading
 import time
 from typing import Dict, Any, List, Optional, Tuple, Set
@@ -19,9 +21,13 @@ except ImportError:
 from base.plugin_common.message_plugin_interface import MessagePluginInterface
 from base.plugin_common.plugin_interface import PluginStatus
 from db.connection import DBConnectionManager
+from plugins.ai_auto_response.llm_client import LLMClient
+from plugins.douyu.danmu_summary import DouyuDanmuSummaryHelper
+from plugins.douyu.report_template import render_daily_report_html
 from utils.decorator.async_job import async_job
 from utils.decorator.plugin_decorators import plugin_stats_decorator
 from utils.decorator.points_decorator import plugin_points_cost
+from utils.markdown_to_image import convert_md_str_to_image, html_to_image
 from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
 from wechat_ipad import WechatAPIClient
 from wechat_ipad.models.appmsg_xml import DOUYU_MESSAGE_XML
@@ -316,6 +322,61 @@ class DouyuRedisManager:
         key = f"{self.prefix}room_status:{room_id}"
         return self.redis.set(key, json.dumps(status, ensure_ascii=False))
 
+    def get_room_session(self, room_id: str, session_id: str) -> Optional[Dict[str, Any]]:
+        key = f"{self.prefix}room:{room_id}:session:{session_id}"
+        data = self.redis.get(key)
+        if not data:
+            return None
+        if isinstance(data, bytes):
+            data = data.decode("utf-8")
+        try:
+            return json.loads(data)
+        except Exception:
+            return None
+
+    def save_room_session(self, room_id: str, session: Dict[str, Any]) -> bool:
+        session_id = str(session.get("session_id") or "").strip()
+        if not session_id:
+            return False
+        payload = json.dumps(session, ensure_ascii=False)
+        session_key = f"{self.prefix}room:{room_id}:session:{session_id}"
+        latest_key = f"{self.prefix}room:{room_id}:latest_session"
+        index_key = f"{self.prefix}room:{room_id}:session_ids"
+        pipe = self.redis.pipeline()
+        pipe.set(session_key, payload)
+        pipe.set(latest_key, session_id)
+        pipe.lrem(index_key, 0, session_id)
+        pipe.lpush(index_key, session_id)
+        pipe.ltrim(index_key, 0, 29)
+        result = pipe.execute()
+        return bool(result)
+
+    def get_latest_room_session(self, room_id: str) -> Optional[Dict[str, Any]]:
+        latest_key = f"{self.prefix}room:{room_id}:latest_session"
+        session_id = self.redis.get(latest_key)
+        if not session_id:
+            return None
+        if isinstance(session_id, bytes):
+            session_id = session_id.decode("utf-8")
+        return self.get_room_session(room_id, str(session_id))
+
+    def list_room_session_ids(self, room_id: str, limit: int = 10) -> List[str]:
+        key = f"{self.prefix}room:{room_id}:session_ids"
+        rows = self.redis.lrange(key, 0, max(limit - 1, 0)) or []
+        result = []
+        for row in rows:
+            result.append(row.decode("utf-8") if isinstance(row, bytes) else str(row))
+        return result
+
+    def get_text_value(self, key: str) -> Optional[str]:
+        data = self.redis.get(key)
+        if not data:
+            return None
+        return data.decode("utf-8") if isinstance(data, bytes) else str(data)
+
+    def set_text_value(self, key: str, value: str) -> bool:
+        return bool(self.redis.set(key, value))
+
 
 class DouyuPlugin(MessagePluginInterface):
     FEATURE_KEY = "DOUYU_MONITOR"
@@ -364,18 +425,44 @@ class DouyuPlugin(MessagePluginInterface):
         self._yuba_api = "https://yuba.douyu.com/wgapi/yubanc/api/feed/getUserFeedList"
         self._user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
         self._check_interval = 5
+        self._session_cutoff_hour = 6
+        self._merge_gap_hours = 4
+        self._daily_report_enable = True
+        self._daily_report_time = "10:05"
+        self._daily_report_min_messages = 120
+        self._daily_report_use_llm = False
+        self._daily_report_max_sessions = 4
+        self._daily_report_max_length = 1800
+        self._daily_report_send_image = True
+        self._daily_report_llm_client: Optional[LLMClient] = None
         self._danmu_recorders: Dict[str, DouyuDanmuRecorder] = {}
         async_job.every_minutes(self._check_interval)(self._scheduled_unified_check_job)
+        async_job.every_minutes(5)(self._scheduled_daily_report_tick)
 
     async def _scheduled_unified_check_job(self):
         """统一检查直播和鱼吧动态"""
         await self._scheduled_check_job()
         await self._scheduled_yuba_check_job()
 
+    async def _scheduled_daily_report_tick(self):
+        """每 5 分钟检查一次，命中配置时间后发送前一天日报。"""
+        if not self._daily_report_enable or not self.redis_manager or not self.bot:
+            return
+        now_dt = datetime.now()
+        if not self._should_run_daily_report(now_dt):
+            return
+        anchor_day = (now_dt - timedelta(days=1)).strftime("%Y-%m-%d")
+        try:
+            await self._send_daily_reports(anchor_day)
+            self.redis_manager.set_text_value(self._daily_report_job_key(now_dt.strftime("%Y-%m-%d")), now_dt.strftime("%Y-%m-%d %H:%M:%S"))
+        except Exception as e:
+            logger.error(f"斗鱼每日报告任务失败(anchor_day={anchor_day}): {e}")
+
     def initialize(self, context: Dict[str, Any]) -> bool:
         try:
             dbm = DBConnectionManager.get_instance()
             self.redis_manager = DouyuRedisManager(dbm)
+            self.bot = context.get("bot", self.bot)
             cfg = self._config.get("Douyu", {})
             cfg_cmds = cfg.get("command", [])
             if isinstance(cfg_cmds, list) and cfg_cmds:
@@ -383,6 +470,21 @@ class DouyuPlugin(MessagePluginInterface):
             self._api_template = cfg.get("api_url_template", self._api_template)
             self._user_agent = cfg.get("user_agent", self._user_agent)
             self._check_interval = int(cfg.get("check_interval_minutes", self._check_interval))
+            self._session_cutoff_hour = int(cfg.get("session_cutoff_hour", self._session_cutoff_hour))
+            self._merge_gap_hours = int(cfg.get("merge_gap_hours", self._merge_gap_hours))
+            self._daily_report_enable = bool(cfg.get("daily_report_enable", self._daily_report_enable))
+            self._daily_report_time = str(cfg.get("daily_report_time", self._daily_report_time) or self._daily_report_time)
+            self._daily_report_min_messages = int(
+                cfg.get("daily_report_min_messages", self._daily_report_min_messages)
+            )
+            self._daily_report_use_llm = bool(cfg.get("daily_report_use_llm", self._daily_report_use_llm))
+            self._daily_report_max_sessions = int(cfg.get("daily_report_max_sessions", self._daily_report_max_sessions))
+            self._daily_report_max_length = int(cfg.get("daily_report_max_length", self._daily_report_max_length))
+            self._daily_report_send_image = bool(cfg.get("daily_report_send_image", self._daily_report_send_image))
+
+            report_api_cfg = cfg.get("report_api", {}) or {}
+            if report_api_cfg:
+                self._daily_report_llm_client = LLMClient(report_api_cfg)
             return True
         except Exception as e:
             logger.error(f"{self.name} 初始化失败: {e}")
@@ -544,6 +646,12 @@ class DouyuPlugin(MessagePluginInterface):
                             continue
                         if prev_live is True and curr_live is True and room_id not in self._danmu_recorders:
                             try:
+                                session = self._open_or_resume_session(room_id, nickname, room_name)
+                                if session:
+                                    logger.info(
+                                        f"检测到持续直播状态，续接斗鱼直播会话({room_id}): "
+                                        f"session={session.get('session_id')}"
+                                    )
                                 logger.info(f"检测到持续直播状态，补偿启动斗鱼弹幕记录({room_id})")
                                 self._start_danmu_record(room_id)
                             except Exception as e:
@@ -573,6 +681,12 @@ class DouyuPlugin(MessagePluginInterface):
                     logger.error(f"发送斗鱼开播提醒失败: {e}")
                     continue
         try:
+            session = self._open_or_resume_session(room_id, nickname, room_name)
+            if session:
+                logger.info(
+                    f"斗鱼直播会话开启/续接: room={room_id}, session={session.get('session_id')}, "
+                    f"segments={len(session.get('segments', []))}, anchor_day={session.get('anchor_day')}"
+                )
             logger.info(f"启动斗鱼弹幕记录({room_id})")
             self._start_danmu_record(room_id)
         except Exception as e:
@@ -591,6 +705,12 @@ class DouyuPlugin(MessagePluginInterface):
                     logger.error(f"发送斗鱼下播提醒失败: {e}")
                     continue
         try:
+            session = self._close_active_session(room_id, nickname, room_name)
+            if session:
+                logger.info(
+                    f"斗鱼直播会话关闭片段: room={room_id}, session={session.get('session_id')}, "
+                    f"segments={len(session.get('segments', []))}, is_live={session.get('is_live')}"
+                )
             logger.info(f"停止斗鱼弹幕记录({room_id})")
             self._stop_danmu_record(room_id)
         except Exception as e:
@@ -685,6 +805,655 @@ class DouyuPlugin(MessagePluginInterface):
             self._danmu_recorders[room_id] = recorder
         return recorder
 
+    def _resolve_anchor_day(self, target_dt: datetime) -> str:
+        if target_dt.hour < self._session_cutoff_hour:
+            target_dt = target_dt - timedelta(days=1)
+        return target_dt.strftime("%Y-%m-%d")
+
+    @staticmethod
+    def _parse_session_time(value: str) -> Optional[datetime]:
+        if not value:
+            return None
+        try:
+            return datetime.strptime(str(value), "%Y-%m-%d %H:%M:%S")
+        except Exception:
+            return None
+
+    @staticmethod
+    def _find_open_segment(session: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        for segment in reversed(session.get("segments", []) or []):
+            if not str(segment.get("end_time") or "").strip():
+                return segment
+        return None
+
+    def _should_merge_with_latest_session(self, latest_session: Optional[Dict[str, Any]], now_dt: datetime) -> bool:
+        if not latest_session:
+            return False
+        if latest_session.get("is_live"):
+            return True
+        segments = latest_session.get("segments", []) or []
+        if not segments:
+            return False
+        last_segment = segments[-1]
+        end_dt = self._parse_session_time(last_segment.get("end_time", ""))
+        if not end_dt:
+            return False
+        gap_seconds = (now_dt - end_dt).total_seconds()
+        return 0 <= gap_seconds <= self._merge_gap_hours * 3600
+
+    def _open_or_resume_session(self, room_id: str, nickname: str, room_name: str) -> Optional[Dict[str, Any]]:
+        if not self.redis_manager:
+            return None
+        now_dt = datetime.now()
+        now_str = now_dt.strftime("%Y-%m-%d %H:%M:%S")
+        latest_session = self.redis_manager.get_latest_room_session(room_id) or {}
+
+        if self._should_merge_with_latest_session(latest_session, now_dt):
+            session = dict(latest_session)
+            open_segment = self._find_open_segment(session)
+            if not open_segment:
+                segments = list(session.get("segments", []) or [])
+                segments.append({"start_time": now_str, "end_time": ""})
+                session["segments"] = segments
+        else:
+            anchor_day = self._resolve_anchor_day(now_dt)
+            session = {
+                "session_id": f"{room_id}_{anchor_day.replace('-', '')}_{now_dt.strftime('%H%M%S')}",
+                "room_id": room_id,
+                "anchor_day": anchor_day,
+                "nickname": nickname,
+                "room_name": room_name,
+                "segments": [{"start_time": now_str, "end_time": ""}],
+                "is_live": True,
+                "summary_status": "pending",
+                "summary_generated_at": "",
+                "created_at": now_str,
+            }
+
+        session["nickname"] = nickname or session.get("nickname", "")
+        session["room_name"] = room_name or session.get("room_name", "")
+        session["is_live"] = True
+        session["updated_at"] = now_str
+        session["last_live_at"] = now_str
+        self.redis_manager.save_room_session(room_id, session)
+        return session
+
+    def _close_active_session(self, room_id: str, nickname: str, room_name: str) -> Optional[Dict[str, Any]]:
+        if not self.redis_manager:
+            return None
+        session = self.redis_manager.get_latest_room_session(room_id)
+        if not session:
+            return None
+
+        now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        open_segment = self._find_open_segment(session)
+        if open_segment:
+            open_segment["end_time"] = now_str
+
+        session["nickname"] = nickname or session.get("nickname", "")
+        session["room_name"] = room_name or session.get("room_name", "")
+        session["is_live"] = False
+        session["updated_at"] = now_str
+        session["last_offline_at"] = now_str
+        self.redis_manager.save_room_session(room_id, session)
+        return session
+
+    def get_room_session(self, room_id: str, session_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
+        if not self.redis_manager or not room_id:
+            return None
+        if session_id:
+            return self.redis_manager.get_room_session(room_id, session_id)
+        return self.redis_manager.get_latest_room_session(room_id)
+
+    def build_session_danmu_material(self, room_id: str, session_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
+        """
+        旁路能力：从已有日文件中按直播 session 抽取有效弹幕，并压缩成可供后续总结使用的材料。
+        当前不影响提醒、采集、群消息发送主流程。
+        """
+        session = self.get_room_session(room_id, session_id=session_id)
+        if not session:
+            return None
+        messages = DouyuDanmuSummaryHelper.load_session_messages(room_id, session)
+        material = DouyuDanmuSummaryHelper.build_summary_material(room_id, session, messages)
+        material["session"] = {
+            "session_id": session.get("session_id", ""),
+            "anchor_day": session.get("anchor_day", ""),
+            "nickname": session.get("nickname", ""),
+            "room_name": session.get("room_name", ""),
+            "is_live": bool(session.get("is_live")),
+        }
+        return material
+
+    def build_session_llm_payload(self, room_id: str, session_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
+        """
+        旁路能力：构造可直接发送给 LLM 的弹幕总结载荷。
+        不改变现有弹幕采集和通知主流程。
+        """
+        session = self.get_room_session(room_id, session_id=session_id)
+        if not session:
+            return None
+        messages = DouyuDanmuSummaryHelper.load_session_messages(room_id, session)
+        return DouyuDanmuSummaryHelper.build_llm_payload(room_id, session, messages)
+
+    def _daily_report_job_key(self, day_key: str) -> str:
+        return f"{self.redis_manager.prefix}daily_report_job:{day_key}"
+
+    def _daily_report_room_key(self, room_id: str, anchor_day: str) -> str:
+        return f"{self.redis_manager.prefix}daily_report:{room_id}:{anchor_day}"
+
+    def _should_run_daily_report(self, now_dt: datetime) -> bool:
+        time_text = str(self._daily_report_time or "").strip()
+        try:
+            target_hour, target_minute = [int(part) for part in time_text.split(":", 1)]
+        except Exception:
+            return False
+        target_dt = now_dt.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
+        if now_dt < target_dt or now_dt > target_dt + timedelta(minutes=4, seconds=59):
+            return False
+        last_run = self.redis_manager.get_text_value(self._daily_report_job_key(now_dt.strftime("%Y-%m-%d")))
+        return not last_run
+
+    def _load_sessions_for_anchor_day(self, room_id: str, anchor_day: str) -> List[Dict[str, Any]]:
+        if not self.redis_manager:
+            return []
+        sessions = []
+        for session_id in self.redis_manager.list_room_session_ids(room_id, limit=30):
+            session = self.redis_manager.get_room_session(room_id, session_id)
+            if not session:
+                continue
+            if str(session.get("anchor_day") or "") != anchor_day:
+                continue
+            sessions.append(session)
+        sessions.sort(
+            key=lambda item: str(((item.get("segments") or [{}])[0]).get("start_time", "")),
+        )
+        return sessions[:self._daily_report_max_sessions]
+
+    def _build_daily_report_payload(self, room_id: str, anchor_day: str, sessions: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+        if not sessions:
+            return None
+
+        session_payloads: List[Dict[str, Any]] = []
+        total_message_count = 0
+        total_noise_filtered_count = 0
+        total_organized_message_count = 0
+        total_unique_users: Set[str] = set()
+        merged_templates: List[Dict[str, Any]] = []
+        repeated_messages: List[Dict[str, Any]] = []
+        peak_buckets: List[Dict[str, Any]] = []
+        representative_messages: List[Dict[str, Any]] = []
+        raw_window_samples: List[Dict[str, Any]] = []
+        top_terms_counter = Counter()
+        burst_terms_counter = Counter()
+        operator_totals = {
+            "fans_badge_user_count": 0,
+            "fans_badge_message_count": 0,
+            "high_room_level_user_count": 0,
+            "high_fans_level_user_count": 0,
+            "noble_user_count": 0,
+            "noble_message_count": 0,
+            "active_users_5plus": 0,
+            "active_users_10plus": 0,
+        }
+        top_badge_counter = Counter()
+        top_badge_message_counter = Counter()
+
+        nickname = ""
+        room_name = ""
+
+        for session in sessions:
+            messages = DouyuDanmuSummaryHelper.load_session_messages(room_id, session)
+            if len(messages) < self._daily_report_min_messages:
+                continue
+            payload = DouyuDanmuSummaryHelper.build_llm_payload(room_id, session, messages)
+            session_payloads.append(payload)
+            meta = payload.get("session_meta", {}) or {}
+            operator_metrics = payload.get("operator_metrics", {}) or {}
+            total_message_count += int(meta.get("message_count", 0) or 0)
+            total_noise_filtered_count += int(meta.get("noise_filtered_count", 0) or 0)
+            total_organized_message_count += int(meta.get("organized_message_count", 0) or 0)
+            nickname = nickname or str(meta.get("nickname") or session.get("nickname") or "")
+            room_name = room_name or str(meta.get("room_name") or session.get("room_name") or "")
+            for key in operator_totals:
+                operator_totals[key] += int(operator_metrics.get(key, 0) or 0)
+
+            for item in payload.get("merged_templates", []) or []:
+                merged_templates.append(dict(item))
+            for item in payload.get("repeated_messages", []) or []:
+                repeated_messages.append(dict(item))
+            for item in payload.get("peak_buckets", []) or []:
+                peak_buckets.append(dict(item))
+            for item in payload.get("representative_messages", []) or []:
+                representative_messages.append(dict(item))
+            for item in payload.get("raw_window_samples", []) or []:
+                raw_window_samples.append(dict(item))
+            for item in payload.get("top_terms", []) or []:
+                term = str(item.get("term") or "").strip()
+                if term:
+                    top_terms_counter[term] += int(item.get("count", 0) or 0)
+            for item in payload.get("burst_terms", []) or []:
+                term = str(item.get("text") or "").strip()
+                if term:
+                    burst_terms_counter[term] += int(item.get("count", 0) or 0)
+            for item in operator_metrics.get("top_badges", []) or []:
+                badge_name = str(item.get("badge_name") or "").strip()
+                if badge_name:
+                    top_badge_counter[badge_name] += int(item.get("user_count", 0) or 0)
+                    top_badge_message_counter[badge_name] += int(item.get("message_count", 0) or 0)
+
+            for session_message in messages:
+                uid = str(session_message.get("uid") or "").strip()
+                if uid:
+                    total_unique_users.add(uid)
+
+        if not session_payloads:
+            return None
+
+        merged_templates.sort(key=lambda item: int(item.get("count", 0) or 0), reverse=True)
+        repeated_messages.sort(key=lambda item: int(item.get("count", 0) or 0), reverse=True)
+        peak_buckets.sort(key=lambda item: int(item.get("message_count", 0) or 0), reverse=True)
+
+        artifact_dir = os.path.join("temp", "douyu_materials")
+        os.makedirs(artifact_dir, exist_ok=True)
+        payload = {
+            "report_meta": {
+                "room_id": room_id,
+                "anchor_day": anchor_day,
+                "nickname": nickname,
+                "room_name": room_name,
+                "session_count": len(session_payloads),
+                "message_count": total_message_count,
+                "noise_filtered_count": total_noise_filtered_count,
+                "organized_message_count": total_organized_message_count,
+                "unique_user_count": len(total_unique_users),
+            },
+            "operator_metrics": {
+                **operator_totals,
+                "fans_badge_user_ratio": round(operator_totals["fans_badge_user_count"] / max(len(total_unique_users), 1), 4),
+                "top_badges": [
+                    {
+                        "badge_name": badge_name,
+                        "user_count": user_count,
+                        "message_count": int(top_badge_message_counter.get(badge_name, 0) or 0),
+                    }
+                    for badge_name, user_count in top_badge_counter.most_common(10)
+                ],
+            },
+            "sessions": [
+                {
+                    "session_id": (item.get("session_meta", {}) or {}).get("session_id", ""),
+                    "segments": (item.get("session_meta", {}) or {}).get("segments", []),
+                    "message_count": (item.get("session_meta", {}) or {}).get("message_count", 0),
+                    "organized_message_count": (item.get("session_meta", {}) or {}).get("organized_message_count", 0),
+                }
+                for item in session_payloads
+            ],
+            "merged_templates": merged_templates[:24],
+            "repeated_messages": repeated_messages[:24],
+            "top_terms": [{"term": term, "count": count} for term, count in top_terms_counter.most_common(24)],
+            "burst_terms": [{"text": term, "count": count} for term, count in burst_terms_counter.most_common(16)],
+            "peak_buckets": peak_buckets[:10],
+            "representative_messages": representative_messages[:24],
+            "raw_window_samples": raw_window_samples[:10],
+        }
+        artifact_path = os.path.join(artifact_dir, f"{room_id}_{anchor_day.replace('-', '')}_daily_report_payload.json")
+        with open(artifact_path, "w", encoding="utf-8") as f:
+            json.dump(payload, f, ensure_ascii=False, indent=2)
+        return payload
+
+    def _build_daily_report_prompt(self, payload: Dict[str, Any]) -> Tuple[str, str]:
+        meta = payload.get("report_meta", {}) or {}
+        system_prompt = (
+            "你是斗鱼直播日报助手。请基于给定的结构化弹幕材料，输出一份适合发群的中文日报。"
+            "要求简洁、自然、信息密度高，不要编造，不要使用代码块。"
+        )
+        user_prompt = (
+            "请输出一份斗鱼每日报告，格式要求：\n"
+            "1. 第一行写标题，包含主播名和日期。\n"
+            "2. 用 3-5 条概括直播主线、弹幕情绪、观众关注点。\n"
+            "3. 单独补充运营视角观察，比如带牌活跃用户、高等级用户、核心发言用户、活跃牌子分布。\n"
+            "4. 单独列出高频梗/复读内容（不超过 5 条）。\n"
+            "5. 单独列出 2-3 个热点时段。\n"
+            "6. 整体控制在 600 字以内。\n\n"
+            f"材料如下：\n{json.dumps(payload, ensure_ascii=False, indent=2)}"
+        )
+        return system_prompt, user_prompt
+
+    def _build_danmu_summary_prompt(self, payload: Dict[str, Any]) -> Tuple[str, str]:
+        meta = payload.get("report_meta", {}) or {}
+        system_prompt = (
+            "你是直播弹幕总结助手。请只根据给定材料，总结这场直播的弹幕内容与氛围。"
+            "不要输出运营数据，不要编造，不要写空话套话。"
+        )
+        user_prompt = (
+            "请输出一段适合放在日报图片上半部分的弹幕总结，要求：\n"
+            "1. 先用 1 段总述直播氛围与主线。\n"
+            "2. 再用 3-5 条要点总结观众关注点、情绪变化、反复出现的梗。\n"
+            "3. 语言像运营复盘，简洁自然。\n"
+            "4. 不要写标题，不要写“根据数据”。\n\n"
+            f"主播：{meta.get('nickname') or meta.get('room_name') or meta.get('room_id')}\n"
+            f"日期：{meta.get('anchor_day', '')}\n"
+            f"材料：\n{json.dumps(payload, ensure_ascii=False, indent=2)}"
+        )
+        return system_prompt, user_prompt
+
+    def _build_fallback_daily_report(self, payload: Dict[str, Any]) -> str:
+        meta = payload.get("report_meta", {}) or {}
+        title_name = str(meta.get("nickname") or meta.get("room_name") or meta.get("room_id") or "主播")
+        lines = [
+            f"斗鱼每日报告 | {title_name} | {meta.get('anchor_day', '')}",
+            f"共 {meta.get('session_count', 0)} 场，弹幕 {meta.get('message_count', 0)} 条，参与用户 {meta.get('unique_user_count', 0)} 人。",
+        ]
+        operator_metrics = payload.get("operator_metrics", {}) or {}
+
+        sessions = payload.get("sessions", []) or []
+        if sessions:
+            session_parts = []
+            for item in sessions[:4]:
+                segments = item.get("segments", []) or []
+                if not segments:
+                    continue
+                start_time = str(segments[0].get("start_time", ""))[-8:-3]
+                end_time = str(segments[-1].get("end_time", ""))[-8:-3]
+                session_parts.append(f"{start_time}-{end_time}")
+            if session_parts:
+                lines.append("场次时间：" + " / ".join(session_parts))
+
+        top_terms = payload.get("top_terms", []) or []
+        if top_terms:
+            lines.append("关注焦点：" + "、".join([str(item.get("term") or "") for item in top_terms[:8] if str(item.get("term") or "").strip()]))
+
+        if operator_metrics:
+            op_parts = []
+            fans_badge_user_count = int(operator_metrics.get("fans_badge_user_count", 0) or 0)
+            high_room_level_user_count = int(operator_metrics.get("high_room_level_user_count", 0) or 0)
+            high_fans_level_user_count = int(operator_metrics.get("high_fans_level_user_count", 0) or 0)
+            active_users_10plus = int(operator_metrics.get("active_users_10plus", 0) or 0)
+            if fans_badge_user_count:
+                op_parts.append(f"带牌活跃用户 {fans_badge_user_count}")
+            if high_room_level_user_count:
+                op_parts.append(f"30级+活跃用户 {high_room_level_user_count}")
+            if high_fans_level_user_count:
+                op_parts.append(f"10级+粉丝牌用户 {high_fans_level_user_count}")
+            if active_users_10plus:
+                op_parts.append(f"高活跃核心用户 {active_users_10plus}")
+            if op_parts:
+                lines.append("运营侧：" + "，".join(op_parts))
+
+            top_badges = operator_metrics.get("top_badges", []) or []
+            if top_badges:
+                lines.append("活跃粉丝牌：")
+                for item in top_badges[:5]:
+                    badge_name = str(item.get("badge_name") or "").strip()
+                    user_count = int(item.get("user_count", 0) or 0)
+                    message_count = int(item.get("message_count", 0) or 0)
+                    if badge_name:
+                        lines.append(f"- {badge_name}：{user_count}人，{message_count}条")
+
+        merged_templates = payload.get("merged_templates", []) or []
+        if merged_templates:
+            lines.append("高频梗：")
+            for item in merged_templates[:5]:
+                text = str(item.get("text") or "").strip()
+                count = int(item.get("count", 0) or 0)
+                if text:
+                    lines.append(f"- {text[:42]}（{count}次）")
+
+        peak_buckets = payload.get("peak_buckets", []) or []
+        if peak_buckets:
+            lines.append("热点时段：")
+            for item in peak_buckets[:3]:
+                start_time = str(item.get("start_time") or "")[-8:-3]
+                message_count = int(item.get("message_count", 0) or 0)
+                terms = "、".join(
+                    [str(term.get("term") or "") for term in (item.get("top_terms", []) or [])[:4] if str(term.get("term") or "").strip()]
+                )
+                lines.append(f"- {start_time}，{message_count}条，关键词：{terms}")
+
+        representative_messages = payload.get("representative_messages", []) or []
+        if representative_messages:
+            lines.append("代表弹幕：")
+            for item in representative_messages[:4]:
+                nickname = str(item.get("nickname") or "").strip()
+                content = str(item.get("content") or "").strip()
+                if content:
+                    lines.append(f"- {nickname}：{content[:60]}")
+
+        text = "\n".join(lines).strip()
+        if len(text) > self._daily_report_max_length:
+            text = text[: self._daily_report_max_length - 20].rstrip() + "\n...(已截断)"
+        return text
+
+    def _build_fallback_danmu_summary(self, payload: Dict[str, Any]) -> str:
+        meta = payload.get("report_meta", {}) or {}
+        top_terms = [str(item.get("term") or "").strip() for item in (payload.get("top_terms", []) or [])[:6] if str(item.get("term") or "").strip()]
+        merged_templates = payload.get("merged_templates", []) or []
+        peak_buckets = payload.get("peak_buckets", []) or []
+        representative_messages = payload.get("representative_messages", []) or []
+
+        lines = [
+            f"{meta.get('anchor_day', '')} 这场直播弹幕整体比较密集，讨论重心主要围绕 {'、'.join(top_terms[:4]) or '对局过程'} 展开，观众互动意愿较强，梗和复读内容持续出现。"
+        ]
+        if merged_templates:
+            sample_templates = "；".join(
+                [str(item.get("text") or "").strip()[:26] for item in merged_templates[:3] if str(item.get("text") or "").strip()]
+            )
+            if sample_templates:
+                lines.append(f"- 主线观察：直播间共识梗很强，重复刷屏内容主要集中在 {sample_templates}。")
+            for item in merged_templates[:4]:
+                break
+        if peak_buckets:
+            top_bucket = peak_buckets[0]
+            terms = "、".join(
+                [str(term.get("term") or "") for term in (top_bucket.get("top_terms", []) or [])[:4] if str(term.get("term") or "").strip()]
+            )
+            lines.append(
+                f"- 节奏变化：高峰集中在 {str(top_bucket.get('start_time') or '')[-8:-3]} 前后，单时段弹幕 {int(top_bucket.get('message_count', 0) or 0)} 条，关键词偏向 {terms}。"
+            )
+            if len(peak_buckets) > 1:
+                second_bucket = peak_buckets[1]
+                second_terms = "、".join(
+                    [str(term.get("term") or "") for term in (second_bucket.get("top_terms", []) or [])[:4] if str(term.get("term") or "").strip()]
+                )
+                lines.append(
+                    f"- 热点补充：{str(second_bucket.get('start_time') or '')[-8:-3]} 也出现明显抬升，弹幕讨论继续围绕 {second_terms} 展开。"
+                )
+        if representative_messages:
+            lines.append("- 情绪特点：代表性发言里既有对操作和决策的即时反馈，也有大量玩梗、调侃和情绪宣泄。")
+        if top_terms:
+            lines.append(f"- 关注焦点：高频词主要落在 {'、'.join(top_terms[:6])}，说明观众注意力相对集中。")
+        return "\n".join(lines).strip()
+
+    def _build_operator_summary_text(self, payload: Dict[str, Any]) -> str:
+        meta = payload.get("report_meta", {}) or {}
+        operator_metrics = payload.get("operator_metrics", {}) or {}
+        total_users = int(meta.get("unique_user_count", 0) or 0)
+        fans_badge_users = int(operator_metrics.get("fans_badge_user_count", 0) or 0)
+        high_room_users = int(operator_metrics.get("high_room_level_user_count", 0) or 0)
+        high_fans_users = int(operator_metrics.get("high_fans_level_user_count", 0) or 0)
+        active_users_5plus = int(operator_metrics.get("active_users_5plus", 0) or 0)
+        active_users_10plus = int(operator_metrics.get("active_users_10plus", 0) or 0)
+        fans_badge_ratio = float(operator_metrics.get("fans_badge_user_ratio", 0) or 0)
+
+        lines = [
+            f"- 活跃用户规模：{total_users} 人，其中发言 5 次以上 {active_users_5plus} 人，10 次以上 {active_users_10plus} 人。",
+            f"- 粉丝粘性：带粉丝牌活跃用户 {fans_badge_users} 人，占活跃用户 {fans_badge_ratio * 100:.1f}%；10 级以上粉丝牌用户 {high_fans_users} 人。",
+            f"- 用户质量：房间等级 30 级以上活跃用户 {high_room_users} 人，说明高等级老观众参与度不低。",
+        ]
+
+        top_badges = payload.get("operator_metrics", {}).get("top_badges", []) or []
+        if top_badges:
+            badge_parts = []
+            for item in top_badges[:5]:
+                badge_name = str(item.get("badge_name") or "").strip()
+                if not badge_name:
+                    continue
+                badge_parts.append(f"{badge_name} {int(item.get('user_count', 0) or 0)}人/{int(item.get('message_count', 0) or 0)}条")
+            if badge_parts:
+                lines.append(f"- 活跃牌子分布：{'；'.join(badge_parts)}。")
+
+        top_active_users = payload.get("operator_metrics", {}).get("top_active_users", []) or []
+        if top_active_users:
+            core_parts = []
+            for item in top_active_users[:5]:
+                nickname = str(item.get("nickname") or item.get("uid") or "").strip()
+                msg_count = int(item.get("message_count", 0) or 0)
+                fans_name = str(item.get("fans_name") or "").strip()
+                if fans_name:
+                    core_parts.append(f"{nickname}（{fans_name}，{msg_count}条）")
+                else:
+                    core_parts.append(f"{nickname}（{msg_count}条）")
+            if core_parts:
+                lines.append(f"- 核心发言用户：{'；'.join(core_parts)}。")
+
+        return "\n".join(lines).strip()
+
+    def _build_operator_summary_lines(self, payload: Dict[str, Any]) -> List[str]:
+        return [line.strip()[2:].strip() for line in self._build_operator_summary_text(payload).splitlines() if line.strip().startswith("- ")]
+
+    async def _generate_danmu_summary_text(self, payload: Dict[str, Any]) -> str:
+        if self._daily_report_use_llm and self._daily_report_llm_client:
+            system_prompt, user_prompt = self._build_danmu_summary_prompt(payload)
+            result = await asyncio.to_thread(
+                self._daily_report_llm_client.chat,
+                system_prompt,
+                user_prompt,
+                f"douyu_danmu_summary_{(payload.get('report_meta', {}) or {}).get('room_id', '')}",
+            )
+            if result:
+                return result.strip()
+            logger.warning(
+                f"斗鱼弹幕总结 LLM 生成失败: model={self._daily_report_llm_client.model}, "
+                f"last_error={self._daily_report_llm_client.last_error}"
+            )
+        return self._build_fallback_danmu_summary(payload)
+
+    async def _build_daily_report_markdown(self, payload: Dict[str, Any]) -> str:
+        meta = payload.get("report_meta", {}) or {}
+        title_name = str(meta.get("nickname") or meta.get("room_name") or meta.get("room_id") or "主播")
+        danmu_summary = await self._generate_danmu_summary_text(payload)
+        operator_summary = self._build_operator_summary_text(payload)
+
+        lines = [
+            f"# {title_name} 直播每日报告",
+            f"{meta.get('anchor_day', '')}｜场次 {meta.get('session_count', 0)}｜弹幕 {meta.get('message_count', 0)}｜活跃用户 {meta.get('unique_user_count', 0)}",
+            "",
+            "## 弹幕总结",
+            danmu_summary,
+            "",
+            "## 运营数据总结",
+            operator_summary,
+        ]
+
+        peak_buckets = payload.get("peak_buckets", []) or []
+        if peak_buckets:
+            lines.extend([
+                "",
+                "## 热点时段",
+            ])
+            for item in peak_buckets[:3]:
+                terms = "、".join(
+                    [str(term.get("term") or "") for term in (item.get("top_terms", []) or [])[:4] if str(term.get("term") or "").strip()]
+                )
+                lines.append(
+                    f"- `{str(item.get('start_time') or '')[-8:-3]}` 弹幕 {int(item.get('message_count', 0) or 0)} 条，关键词：{terms}"
+                )
+
+        merged_templates = payload.get("merged_templates", []) or []
+        if merged_templates:
+            lines.extend([
+                "",
+                "## 高频梗",
+            ])
+            for item in merged_templates[:5]:
+                text = str(item.get("text") or "").strip()
+                count = int(item.get("count", 0) or 0)
+                if text:
+                    lines.append(f"- {text[:72]}（{count}次）")
+
+        return "\n".join(lines).strip()
+
+    async def _render_daily_report_image(self, payload: Dict[str, Any]) -> Optional[str]:
+        markdown = await self._build_daily_report_markdown(payload)
+        room_id = str((payload.get("report_meta", {}) or {}).get("room_id", "") or "room")
+        anchor_day = str((payload.get("report_meta", {}) or {}).get("anchor_day", "") or "").replace("-", "")
+        filename = f"douyu_daily_report_{room_id}_{anchor_day}.png"
+        try:
+            danmu_summary = await self._generate_danmu_summary_text(payload)
+            html_content = render_daily_report_html(
+                payload=payload,
+                danmu_summary=danmu_summary,
+                operator_summary_lines=self._build_operator_summary_lines(payload),
+            )
+            output_dir = os.path.join(os.getcwd(), "temp", "md2image")
+            os.makedirs(output_dir, exist_ok=True)
+            output_path = os.path.join(output_dir, filename)
+            await html_to_image(html_content, output_path)
+            return str(Path(output_path).resolve())
+        except Exception as e:
+            logger.error(f"斗鱼专用模板图片生成失败(room={room_id}, day={anchor_day}): {e}")
+        try:
+            return await convert_md_str_to_image(markdown, filename)
+        except Exception as e:
+            logger.error(f"斗鱼每日报告图片生成失败(room={room_id}, day={anchor_day}): {e}")
+            return None
+
+    async def _generate_daily_report_text(self, payload: Dict[str, Any]) -> str:
+        if self._daily_report_use_llm and self._daily_report_llm_client:
+            system_prompt, user_prompt = self._build_daily_report_prompt(payload)
+            result = await asyncio.to_thread(
+                self._daily_report_llm_client.chat,
+                system_prompt,
+                user_prompt,
+                f"douyu_daily_report_{(payload.get('report_meta', {}) or {}).get('room_id', '')}",
+            )
+            if result:
+                text = result.strip()
+                if len(text) > self._daily_report_max_length:
+                    return text[: self._daily_report_max_length - 20].rstrip() + "\n...(已截断)"
+                return text
+            logger.warning(
+                f"斗鱼每日报告 LLM 生成失败: model={self._daily_report_llm_client.model}, "
+                f"last_error={self._daily_report_llm_client.last_error}"
+            )
+        return self._build_fallback_daily_report(payload)
+
+    async def _send_daily_reports(self, anchor_day: str):
+        rooms = self.redis_manager.all_subscribed_rooms()
+        for room_id in rooms:
+            if self.redis_manager.get_text_value(self._daily_report_room_key(room_id, anchor_day)):
+                continue
+            sessions = self._load_sessions_for_anchor_day(room_id, anchor_day)
+            if not sessions:
+                continue
+            if any(bool(session.get("is_live")) for session in sessions):
+                continue
+            payload = self._build_daily_report_payload(room_id, anchor_day, sessions)
+            if not payload:
+                continue
+            report_text = await self._generate_daily_report_text(payload)
+            report_image = None
+            if self._daily_report_send_image:
+                report_image = await self._render_daily_report_image(payload)
+            groups = self.redis_manager.groups_for_room(room_id)
+            delivered = False
+            for gid in groups:
+                if GroupBotManager.get_group_permission(gid, self.feature) != PermissionStatus.ENABLED:
+                    continue
+                try:
+                    if report_image:
+                        await self.bot.send_image_message(gid, Path(report_image))
+                    else:
+                        await self.bot.send_text_message(gid, report_text)
+                    delivered = True
+                except Exception as e:
+                    logger.error(f"发送斗鱼每日报告失败(room={room_id}, group={gid}): {e}")
+            if delivered:
+                self.redis_manager.set_text_value(
+                    self._daily_report_room_key(room_id, anchor_day),
+                    datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                )
+
     def _start_danmu_record(self, room_id: str):
         recorder = self._get_danmu_recorder(room_id)
         recorder.start()