# coding=utf-8 """ AI 分析结果格式化模块 将 AI 分析结果格式化为各推送渠道的样式 """ import html as html_lib import re from .analyzer import AIAnalysisResult def _escape_html(text: str) -> str: """转义 HTML 特殊字符，防止 XSS 攻击""" return html_lib.escape(text) if text else "" def _format_list_content(text: str) -> str: """ 格式化列表内容，确保序号前有换行 例如将 "1. xxx 2. yyy" 转换为: 1. xxx 2. yyy """ if not text: return "" # 去除首尾空白，防止 AI 返回的内容开头就有换行导致显示空行 text = text.strip() # 0. 合并序号与紧随的【标签】（防御性处理） # 将 "1.\n【投资者】：" 或 "1. 【投资者】：" 合并为 "1. 投资者：" text = re.sub(r'(\d+\.)\s*【([^】]+)】([:：]?)', r'\1 \2：', text) # 1. 规范化：确保 "1." 后面有空格 result = re.sub(r'(\d+)\.([^ \d])', r'\1. \2', text) # 2. 强制换行：匹配 "数字."，且前面不是换行符 result = re.sub(r'(?<=[^\n])\s+(\d+\.)', r'\n\1', result) # 3. 处理 "1.**粗体**" 这种情况（虽然 Prompt 要求不输出 Markdown，但防御性处理） result = re.sub(r'(?<=[^\n])(\d+\.\*\*)', r'\n\1', result) # 4. 处理中文标点后的换行 result = re.sub(r'([：:;,。；，])\s*(\d+\.)', r'\1\n\2', result) # 5. 处理 "XX方面："、"XX领域：" 等子标题换行 # 只有在中文标点（句号、逗号、分号等）后才触发换行，避免破坏 "1. XX领域：" 格式 result = re.sub(r'([。！？；，、])\s*([a-zA-Z0-9\u4e00-\u9fa5]+(方面|领域)[:：])', r'\1\n\2', result) # 6. 处理 【标签】 格式 # 6a. 标签前确保空行分隔（文本开头除外） result = re.sub(r'(?<=\S)\n*(【[^】]+】)', r'\n\n\1', result) # 6b. 合并标签与被换行拆开的冒号：【tag】\n： → 【tag】： result = re.sub(r'(【[^】]+】)\n+([:：])', r'\1\2', result) # 6c. 标签后（含可选冒号），如果紧跟非空白非冒号内容则另起一行 # 用 (?=[^\s:：]) 避免正则回溯将冒号误判为"内容"而拆开 【tag】： result = re.sub(r'(【[^】]+】[:：]?)[ \t]*(?=[^\s:：])', r'\1\n', result) # 7. 在列表项之间增加视觉空行 # 排除 【标签】 行（以】结尾）和子标题行（以冒号结尾）之后的情况，避免标题与首项之间出现空行 result = re.sub(r'(? str: """格式化独立展示区概括为纯文本行，每个源名称单独一行""" if not summaries: return "" lines = [] for source_name, summary in summaries.items(): if summary: lines.append(f"[{source_name}]:\n{summary}") return "\n\n".join(lines) def render_ai_analysis_markdown(result: AIAnalysisResult) -> str: """渲染为通用 Markdown 格式（Telegram、企业微信、ntfy、Bark、Slack）""" if not result.success: return f"⚠️ AI 分析失败: {result.error}" lines = ["**✨ AI 热点分析**", ""] if result.core_trends: lines.extend(["**核心热点态势**", _format_list_content(result.core_trends), ""]) if result.sentiment_controversy: lines.extend( ["**舆论风向争议**", _format_list_content(result.sentiment_controversy), ""] ) if result.signals: lines.extend(["**异动与弱信号**", _format_list_content(result.signals), ""]) if result.rss_insights: lines.extend( ["**RSS 深度洞察**", _format_list_content(result.rss_insights), ""] ) if result.outlook_strategy: lines.extend( ["**研判策略建议**", _format_list_content(result.outlook_strategy), ""] ) if result.standalone_summaries: summaries_text = _format_standalone_summaries(result.standalone_summaries) if summaries_text: lines.extend(["**独立源点速览**", summaries_text]) return "\n".join(lines) def render_ai_analysis_feishu(result: AIAnalysisResult) -> str: """渲染为飞书卡片 Markdown 格式""" if not result.success: return f"⚠️ AI 分析失败: {result.error}" lines = ["**✨ AI 热点分析**", ""] if result.core_trends: lines.extend(["**核心热点态势**", _format_list_content(result.core_trends), ""]) if result.sentiment_controversy: lines.extend( ["**舆论风向争议**", _format_list_content(result.sentiment_controversy), ""] ) if result.signals: lines.extend(["**异动与弱信号**", _format_list_content(result.signals), ""]) if result.rss_insights: lines.extend( ["**RSS 深度洞察**", _format_list_content(result.rss_insights), ""] ) if result.outlook_strategy: lines.extend( ["**研判策略建议**", _format_list_content(result.outlook_strategy), ""] ) if result.standalone_summaries: summaries_text = _format_standalone_summaries(result.standalone_summaries) if summaries_text: lines.extend(["**独立源点速览**", summaries_text]) return "\n".join(lines) def render_ai_analysis_dingtalk(result: AIAnalysisResult) -> str: """渲染为钉钉 Markdown 格式""" if not result.success: return f"⚠️ AI 分析失败: {result.error}" lines = ["### ✨ AI 热点分析", ""] if result.core_trends: lines.extend( ["#### 核心热点态势", _format_list_content(result.core_trends), ""] ) if result.sentiment_controversy: lines.extend( [ "#### 舆论风向争议", _format_list_content(result.sentiment_controversy), "", ] ) if result.signals: lines.extend(["#### 异动与弱信号", _format_list_content(result.signals), ""]) if result.rss_insights: lines.extend( ["#### RSS 深度洞察", _format_list_content(result.rss_insights), ""] ) if result.outlook_strategy: lines.extend( ["#### 研判策略建议", _format_list_content(result.outlook_strategy), ""] ) if result.standalone_summaries: summaries_text = _format_standalone_summaries(result.standalone_summaries) if summaries_text: lines.extend(["#### 独立源点速览", summaries_text]) return "\n".join(lines) def render_ai_analysis_html(result: AIAnalysisResult) -> str: """渲染为 HTML 格式（邮件）""" if not result.success: return ( f'
⚠️ AI 分析失败: {_escape_html(result.error)}
' ) html_parts = ['
', "

✨ AI 热点分析

"] if result.core_trends: content = _format_list_content(result.core_trends) content_html = _escape_html(content).replace("\n", "
") html_parts.extend( [ '
', "

核心热点态势

", f'
{content_html}
', "
", ] ) if result.sentiment_controversy: content = _format_list_content(result.sentiment_controversy) content_html = _escape_html(content).replace("\n", "
") html_parts.extend( [ '
', "

舆论风向争议

", f'
{content_html}
', "
", ] ) if result.signals: content = _format_list_content(result.signals) content_html = _escape_html(content).replace("\n", "
") html_parts.extend( [ '
', "

异动与弱信号

", f'
{content_html}
', "
", ] ) if result.rss_insights: content = _format_list_content(result.rss_insights) content_html = _escape_html(content).replace("\n", "
") html_parts.extend( [ '
', "

RSS 深度洞察

", f'
{content_html}
', "
", ] ) if result.outlook_strategy: content = _format_list_content(result.outlook_strategy) content_html = _escape_html(content).replace("\n", "
") html_parts.extend( [ '
', "

研判策略建议

", f'
{content_html}
', "
", ] ) if result.standalone_summaries: summaries_text = _format_standalone_summaries(result.standalone_summaries) if summaries_text: summaries_html = _escape_html(summaries_text).replace("\n", "
") html_parts.extend( [ '
', "

独立源点速览

", f'
{summaries_html}
', "
", ] ) html_parts.append("
") return "\n".join(html_parts) def render_ai_analysis_plain(result: AIAnalysisResult) -> str: """渲染为纯文本格式""" if not result.success: return f"AI 分析失败: {result.error}" lines = ["【✨ AI 热点分析】", ""] if result.core_trends: lines.extend(["[核心热点态势]", _format_list_content(result.core_trends), ""]) if result.sentiment_controversy: lines.extend( ["[舆论风向争议]", _format_list_content(result.sentiment_controversy), ""] ) if result.signals: lines.extend(["[异动与弱信号]", _format_list_content(result.signals), ""]) if result.rss_insights: lines.extend(["[RSS 深度洞察]", _format_list_content(result.rss_insights), ""]) if result.outlook_strategy: lines.extend(["[研判策略建议]", _format_list_content(result.outlook_strategy), ""]) if result.standalone_summaries: summaries_text = _format_standalone_summaries(result.standalone_summaries) if summaries_text: lines.extend(["[独立源点速览]", summaries_text]) return "\n".join(lines) def get_ai_analysis_renderer(channel: str): """根据渠道获取对应的渲染函数""" renderers = { "feishu": render_ai_analysis_feishu, "dingtalk": render_ai_analysis_dingtalk, "wework": render_ai_analysis_markdown, "telegram": render_ai_analysis_markdown, "email": render_ai_analysis_html_rich, # 邮件使用丰富样式，配合 HTML 报告的 CSS "ntfy": render_ai_analysis_markdown, "bark": render_ai_analysis_plain, "slack": render_ai_analysis_markdown, } return renderers.get(channel, render_ai_analysis_markdown) def render_ai_analysis_html_rich(result: AIAnalysisResult) -> str: """渲染为丰富样式的 HTML 格式（HTML 报告用）""" if not result: return "" # 检查是否成功 if not result.success: error_msg = result.error or "未知错误" return f"""
⚠️ AI 分析失败: {_escape_html(str(error_msg))}
""" ai_html = """
✨ AI 热点分析
AI
""" if result.core_trends: content = _format_list_content(result.core_trends) content_html = _escape_html(content).replace("\n", "
") ai_html += f"""
核心热点态势
{content_html}
""" if result.sentiment_controversy: content = _format_list_content(result.sentiment_controversy) content_html = _escape_html(content).replace("\n", "
") ai_html += f"""
舆论风向争议
{content_html}
""" if result.signals: content = _format_list_content(result.signals) content_html = _escape_html(content).replace("\n", "
") ai_html += f"""
异动与弱信号
{content_html}
""" if result.rss_insights: content = _format_list_content(result.rss_insights) content_html = _escape_html(content).replace("\n", "
") ai_html += f"""
RSS 深度洞察
{content_html}
""" if result.outlook_strategy: content = _format_list_content(result.outlook_strategy) content_html = _escape_html(content).replace("\n", "
") ai_html += f"""
研判策略建议
{content_html}
""" if result.standalone_summaries: summaries_text = _format_standalone_summaries(result.standalone_summaries) if summaries_text: summaries_html = _escape_html(summaries_text).replace("\n", "
") ai_html += f"""
独立源点速览
{summaries_html}
""" ai_html += """
""" return ai_html