feat: 檔案名稱模板支援 ULID 和資料夾路徑 (#146 @lekoOwO)

* feat: Enhance HTML report generation and dispatching with captions

* chore: apply ruff fixes

* fix(ReportGenerator): update HTML caption for report generation

* fix(ReportGenerator): encode filename in HTML report caption URL

* fix(ReportGenerator): refactor golden quotes handling for clarity and consistency

* feat(ReportGenerator): enhance filename formatting with ulid support and path sanitization

* fix(ReportGenerator): enhance path validation and filename uniqueness in report generation

* fix: pylance type error and ruff

* fix: 路径默认使用 ulid 变量

---------

Co-authored-by: SXP-Simon <sxp20061207@163.com>

_conf_schema.json

@@ -356,8 +356,8 @@
             "pdf_filename_format": {
                 "type": "string",
                 "description": "PDF文件名格式",
-                "default": "群聊分析报告_{group_id}_{date}.pdf",
+                "default": "群聊分析报告_{group_id}_{date}_{ulid}.pdf",
-                "hint": "PDF文件名格式，支持变量：{group_id}(群号)、{date}(日期)"
+                "hint": "PDF文件名格式，支持变量：{group_id}(群号)、{date}(日期)、{ulid}(时间排序ID)，支持子目录/层级。"
             }
         }
     },
@@ -381,8 +381,8 @@
             "html_filename_format": {
                 "type": "string",
                 "description": "HTML文件名格式",
-                "default": "群聊分析报告_{group_id}_{date}.html",
+                "default": "群聊分析报告_{group_id}_{date}_{ulid}.html",
-                "hint": "HTML文件名格式，支持变量：{group_id}(群号)、{date}(日期)"
+                "hint": "HTML文件名格式，支持变量：{group_id}(群号)、{date}(日期)、{ulid}(时间排序ID)，支持子目录/层级。"
             }
         }
     },

main.py

@@ -32,6 +32,7 @@ from .src.domain.services.incremental_merge_service import IncrementalMergeServi
 from .src.domain.services.statistics_service import StatisticsService
 from .src.infrastructure.analysis.llm_analyzer import LLMAnalyzer
 from .src.infrastructure.config.config_manager import ConfigManager
+from .src.infrastructure.messaging.message_sender import MessageSender
 from .src.infrastructure.persistence.history_manager import HistoryManager
 from .src.infrastructure.persistence.incremental_store import IncrementalStore
 from .src.infrastructure.persistence.telegram_group_registry import (
@@ -72,6 +73,7 @@ class GroupDailyAnalysis(Star):
     template_preview_router: TemplatePreviewRouter
     retry_manager: RetryManager
     auto_scheduler: AutoScheduler
+    message_sender: MessageSender
 
     def __init__(self, context: Context, config: AstrBotConfig):
         super().__init__(context)
@@ -133,6 +135,9 @@ class GroupDailyAnalysis(Star):
         self.retry_manager = RetryManager(
             self.bot_manager, self.html_render, self.report_generator
         )
+        self.message_sender = MessageSender(
+            self.bot_manager, self.config_manager, self.retry_manager
+        )
         self.auto_scheduler = AutoScheduler(
             self.config_manager,
             self.analysis_service,
@@ -638,20 +643,29 @@ class GroupDailyAnalysis(Star):
                 nickname_getter=nickname_getter,
             )
             if html_path:
+                caption = self.report_generator.build_html_caption(html_path)
+
                 # 发送 HTML 文件
-                if not await adapter.send_file(group_id, html_path):
+                sender = getattr(self, "message_sender", None)
+                if sender:
+                    sent = await sender.send_file(
+                        group_id,
+                        html_path,
+                        caption=caption,
+                        platform_id=platform_id,
+                    )
+                else:
+                    sent = await adapter.send_file(group_id, html_path)
+                    if sent and caption:
+                        await adapter.send_text(group_id, caption)
+
+                if not sent:
                     yield event.chain_result(
                         [File(name=Path(html_path).name, file=html_path)]
                     )
 
-                # 如果配置了外链 Base URL，则也发送超链接
+                    if caption:
-                base_url = self.config_manager.get_html_base_url()
+                        yield event.plain_result(caption)
-                if base_url:
-                    filename = Path(html_path).name
-                    url = f"{base_url.rstrip('/')}/{filename}"
-                    link_message = f"报告已生成: {url}"
-                    if not await adapter.send_text(group_id, link_message):
-                        yield event.plain_result(link_message)
             else:
                 yield event.plain_result("⚠️ HTML 生成失败。")
 

requirements.txt

@@ -1,2 +1,3 @@
 playwright>=1.40.0
-diskcache
+diskcache
+ulid-py

src/infrastructure/messaging/message_sender.py

@@ -18,7 +18,7 @@ class MessageSender:
         self.retry_manager = retry_manager
 
     async def send_text(
-        self, group_id: str, text: str, platform_id: str = None
+        self, group_id: str, text: str, platform_id: str | None = None
     ) -> bool:
         """发送文本消息"""
         adapter = self.bot_manager.get_adapter(platform_id)
@@ -28,7 +28,11 @@ class MessageSender:
         return await adapter.send_text(group_id, text)
 
     async def send_image_smart(
-        self, group_id: str, image_url: str, caption: str = "", platform_id: str = None
+        self,
+        group_id: str,
+        image_url: str,
+        caption: str = "",
+        platform_id: str | None = None,
     ) -> bool:
         """智能发送图片，支持自动选择适配器"""
         adapter = self.bot_manager.get_adapter(platform_id)
@@ -37,15 +41,38 @@ class MessageSender:
             return False
         return await adapter.send_image(group_id, image_url, caption)
 
-    async def send_pdf(
+    async def send_file(
-        self, group_id: str, pdf_path: str, caption: str = "", platform_id: str = None
+        self,
+        group_id: str,
+        file_path: str,
+        caption: str = "",
+        platform_id: str | None = None,
     ) -> bool:
-        """发送 PDF 文件"""
+        """发送文件（HTML/PDF/其它文件）。支持可选 caption。"""
         adapter = self.bot_manager.get_adapter(platform_id)
         if not adapter:
             logger.error(f"[MessageSender] 未找到平台 {platform_id} 的适配器")
             return False
-        return await adapter.send_file(group_id, pdf_path)
+
+        # 首先发送文件，本方法的返回值只代表文件是否发送成功。
+        file_sent = await adapter.send_file(group_id, file_path)
+
+        if not file_sent:
+            # 适配器返回 False，表示文件未成功发送
+            return False
+
+        # 文件已成功发送，下面的 caption 发送为尽力而为，不影响整体成功与否。
+        if caption:
+            try:
+                caption_sent = await adapter.send_text(group_id, f"{caption}")
+                if not caption_sent:
+                    logger.warning(
+                        "[MessageSender] 文件已发送，但 caption 发送失败（适配器返回 False）"
+                    )
+            except Exception as e:
+                logger.warning(f"[MessageSender] 文件已发送，但 caption 发送异常: {e}")
+
+        return True
 
     def _get_available_platforms(self, group_id: str):
         """获取可用的平台列表 (Helper for Dispatcher)"""

src/infrastructure/reporting/dispatcher.py

@@ -46,6 +46,8 @@ class ReportDispatcher:
             success = await self._dispatch_image(group_id, analysis_result, platform_id)
         elif output_format == "pdf":
             success = await self._dispatch_pdf(group_id, analysis_result, platform_id)
+        elif output_format == "html":
+            success = await self._dispatch_html(group_id, analysis_result, platform_id)
         else:
             success = await self._dispatch_text(group_id, analysis_result, platform_id)
 
@@ -148,8 +150,11 @@ class ReportDispatcher:
 
         # 3. 发送 PDF
         if pdf_path:
-            sent = await self.message_sender.send_pdf(
+            sent = await self.message_sender.send_file(
-                group_id, pdf_path, "📊 每日群聊分析报告已生成：", platform_id
+                group_id,
+                pdf_path,
+                caption="📊 每日群聊分析报告已生成：",
+                platform_id=platform_id,
             )
             if sent:
                 return True
@@ -160,6 +165,36 @@ class ReportDispatcher:
         )
         return await self._dispatch_text(group_id, analysis_result, platform_id)
 
+    async def _dispatch_html(
+        self, group_id: str, analysis_result: dict[str, Any], platform_id: str | None
+    ) -> bool:
+        trace_id = TraceContext.get()
+
+        html_path = None
+        try:
+            html_path, json_path = await self.report_generator.generate_html_report(
+                analysis_result, group_id
+            )
+        except Exception as e:
+            logger.error(f"[{trace_id}] Failed to generate HTML report: {e}")
+
+        if html_path:
+            caption = self.report_generator.build_html_caption(html_path)
+
+            sent = await self.message_sender.send_file(
+                group_id,
+                html_path,
+                caption=caption,
+                platform_id=platform_id,
+            )
+            if sent:
+                return True
+
+        logger.warning(
+            f"[{trace_id}] HTML dispatch failed, falling back to text report."
+        )
+        return await self._dispatch_text(group_id, analysis_result, platform_id)
+
     async def _dispatch_text(
         self, group_id: str, analysis_result: dict[str, Any], platform_id: str | None
     ) -> bool:

src/infrastructure/reporting/generators.py

@@ -12,8 +12,10 @@ from dataclasses import asdict, is_dataclass
 from datetime import date, datetime
 from enum import Enum
 from pathlib import Path
+from urllib.parse import quote
 
 import aiohttp
+import ulid
 from diskcache import Cache
 from markupsafe import Markup
 
@@ -49,6 +51,77 @@ class ReportGenerator(IReportGenerator):
         )
         self._avatar_session = None
 
+    @staticmethod
+    def _sanitize_path_component(name: str) -> str:
+        """消毒单个路径/文件名片段，禁止路径穿越和非法字符。"""
+        # 禁止空组件、相对路径控制符："."、".."
+        if not name or name in {".", ".."}:
+            raise ValueError(f"无效的路径片段: {name!r}")
+
+        # 不允许包含路径分隔符
+        name = name.replace("/", "_")
+        name = name.replace("\\", "_")
+
+        # 去除非打印字符和非法文件名字符
+        name = re.sub(r'[\x00-\x1f<>:"|?*]', "_", name)
+
+        # 保留中文、字母、数字、下划线、横线和点
+        name = name.strip()
+        if not name:
+            raise ValueError("路径片段经过消毒后为空")
+
+        return name
+
+    def _build_safe_report_path(
+        self,
+        output_dir: Path,
+        filename_format: str,
+        group_id: str,
+        date: str,
+    ) -> Path:
+        """根据格式构建安全输出路径，支持子目录和 {ulid}。"""
+        generated_ulid = str(ulid.new())
+        safe_context = {
+            "group_id": group_id,
+            "date": date,
+            "ulid": generated_ulid,
+        }
+
+        try:
+            formatted = filename_format.format(**safe_context)
+        except Exception as e:
+            raise ValueError(f"文件名格式化失败: {e}") from e
+
+        if os.path.isabs(formatted):
+            raise ValueError("文件名格式不得为绝对路径")
+
+        relative_path = Path(formatted)
+        sanitized_parts = []
+        for part in relative_path.parts:
+            if part in {".", ".."}:
+                raise ValueError("路径中不得包含 '.' 或 '..'。")
+            sanitized_parts.append(self._sanitize_path_component(part))
+
+        safe_relative = Path(*sanitized_parts)
+
+        output_dir_resolved = output_dir.resolve(strict=False)
+        target_path = (output_dir_resolved / safe_relative).resolve(strict=False)
+
+        # 防止回退到上级目录（使用 Path.relative_to 进行目录包含校验）
+        try:
+            target_path.relative_to(output_dir_resolved)
+        except ValueError:
+            raise ValueError("文件路径不在输出目录之内，可能包含路径穿越")
+
+        # 防止与已有文件覆盖（如果用户格式没有唯一标记），追加 ULID 后缀
+        if target_path.exists():
+            suffix = target_path.suffix
+            stem = target_path.stem
+            target_path = target_path.with_name(f"{stem}_{generated_ulid}{suffix}")
+
+        target_path.parent.mkdir(parents=True, exist_ok=True)
+        return target_path
+
     async def generate_image_report(
         self,
         analysis_result: dict,
@@ -221,12 +294,14 @@ class ReportGenerator(IReportGenerator):
             output_dir = Path(self.config_manager.get_pdf_output_dir())
             await asyncio.to_thread(output_dir.mkdir, parents=True, exist_ok=True)
 
-            # 生成文件名
+            # 生成文件路径，支持 {group_id}/{date}/{ulid} 自定义子目录
             current_date = datetime.now().strftime("%Y%m%d")
-            filename = self.config_manager.get_pdf_filename_format().format(
+            pdf_path = self._build_safe_report_path(
-                group_id=group_id, date=current_date
+                output_dir,
+                self.config_manager.get_pdf_filename_format(),
+                group_id=group_id,
+                date=current_date,
             )
-            pdf_path = output_dir / filename
 
             # 准备渲染数据
             render_data = await self._prepare_render_data(
@@ -287,19 +362,22 @@ class ReportGenerator(IReportGenerator):
             output_dir = Path(self.config_manager.get_html_output_dir())
             await asyncio.to_thread(output_dir.mkdir, parents=True, exist_ok=True)
 
-            # 生成文件名
+            # 生成文件路径
             current_date = datetime.now().strftime("%Y%m%d")
-            current_time = datetime.now().strftime("%H%M%S")
+            base_html_path = self._build_safe_report_path(
-            html_filename = self.config_manager.get_html_filename_format().format(
+                output_dir,
-                group_id=group_id, date=current_date
+                self.config_manager.get_html_filename_format(),
+                group_id=group_id,
+                date=current_date,
             )
-            # 为避免同一天多次分析覆盖，添加时间戳
-            html_filename_base = html_filename.rsplit(".", 1)[0]
-            html_filename = f"{html_filename_base}_{current_time}.html"
-            json_filename = f"{html_filename_base}_{current_time}.json"
 
-            html_path = output_dir / html_filename
+            html_path = base_html_path
-            json_path = output_dir / json_filename
+            if not html_path.suffix:
+                html_path = html_path.with_suffix(".html")
+
+            json_path = html_path.with_suffix(".json")
+
+            html_path.parent.mkdir(parents=True, exist_ok=True)
 
             # 准备渲染数据
             render_data = await self._prepare_render_data(
@@ -378,6 +456,29 @@ class ReportGenerator(IReportGenerator):
             logger.error(f"生成 HTML 报告失败: {e}", exc_info=True)
             return None, None
 
+    def build_html_caption(self, html_path: str) -> str:
+        """根据 html_base_url 生成 HTML 报告链接 caption"""
+
+        caption = "📊 每日群聊分析报告已生成"
+        base_url = self.config_manager.get_html_base_url()
+        if not base_url or not html_path:
+            return caption
+
+        # 支持 html_filename_format 中的子目录，保持相对路径
+        output_dir = Path(self.config_manager.get_html_output_dir()).resolve(
+            strict=False
+        )
+        try:
+            relative_path = (
+                Path(html_path).resolve(strict=False).relative_to(output_dir)
+            )
+            relative_url = str(relative_path).replace(os.sep, "/")
+        except Exception:
+            relative_url = Path(html_path).name
+
+        encoded_relative_url = quote(relative_url, safe="/")
+        return caption + f"\n{base_url.rstrip('/')}/{encoded_relative_url}"
+
     def generate_text_report(self, analysis_result: dict) -> str:
         """生成文本格式的分析报告"""
         stats = analysis_result["statistics"]
@@ -413,9 +514,9 @@ class ReportGenerator(IReportGenerator):
 
         report += "💬 群圣经\n"
         max_golden_quotes = self.config_manager.get_max_golden_quotes()
-        for i, quote in enumerate(stats.golden_quotes[:max_golden_quotes], 1):
+        for i, golden_quote in enumerate(stats.golden_quotes[:max_golden_quotes], 1):
-            report += f'{i}. "{quote.content}" —— {quote.sender}\n'
+            report += f'{i}. "{golden_quote.content}" —— {golden_quote.sender}\n'
-            report += f"   {quote.reason}\n\n"
+            report += f"   {golden_quote.reason}\n\n"
 
         return report
 
@@ -481,20 +582,22 @@ class ReportGenerator(IReportGenerator):
         # 使用Jinja2模板构建金句HTML（批量渲染）
         max_golden_quotes = self.config_manager.get_max_golden_quotes()
         quotes_list = []
-        for quote in stats.golden_quotes[:max_golden_quotes]:
+        for golden_quote in stats.golden_quotes[:max_golden_quotes]:
             avatar_url = (
-                await self._get_user_avatar(str(quote.user_id), avatar_url_getter)
+                await self._get_user_avatar(
-                if quote.user_id
+                    str(golden_quote.user_id), avatar_url_getter
+                )
+                if golden_quote.user_id
                 else None
             )
             # 处理解析锐评中的用户引用头像
             processed_reason = await self._render_mentions(
-                quote.reason, avatar_url_getter, nickname_getter, user_analysis
+                golden_quote.reason, avatar_url_getter, nickname_getter, user_analysis
             )
             quotes_list.append(
                 {
-                    "content": quote.content,
+                    "content": golden_quote.content,
-                    "sender": quote.sender,
+                    "sender": golden_quote.sender,
                     "reason": processed_reason,
                     "avatar_url": avatar_url,
                 }