github/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis.git synced 2026-10-06 14:33:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve notification report rendering (#1531)

Browse Source 
* fix: improve notification report rendering

* fix: polish chat report rendering output

* fix: add chat-optimized report layout

* fix: use chat report for im image routing

* fix: adapt im report formatting per channel

* fix: make im reports decision cards

* chore: reserve notification renderer presets

* docs: trim notification rendering guidance

* fix(review-feedback-1531): preserve valid links but escape non-link Telegram Markdown characters

* fix(review-feedback-1531): Chunk Telegram fallback before sending raw text

* fix(review-feedback-1531): avoid stripping arbitrary leading whitespace here

* fix: preserve legacy notification report formatting

* fix(review-feedback-1531): 补齐外部模型/API 与运行时配置迁移命中的排除说明或兼容性证据
This commit is contained in:
mumu
2026-05-31 19:45:12 +08:00
committed by GitHub
parent b22d7e0277
commit 9178e03762
18 changed files with 1369 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/CHANGELOG.md
+5
View File
@@ -20,6 +20,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
- [修复] Windows 桌面端自动更新静默安装时显式复用当前安装目录，避免自定义安装目录场景下卸载旧版本文件失败。
- [修复] Windows 安装器重试旧卸载器时对 `_?=` 安装目录参数加引号，修复旧版本安装在带空格路径时返回 2 导致自动更新失败。
- [修复] Windows 桌面端自动更新传给 NSIS 的 `/D=` 目录参数在包含空格时自动加引号，避免安装位置注册表被截断。
- [chore] 新增通知报告渠道能力画像、PreparedMessage 与结构感知 Markdown 分片基础设施，为 #1311 全渠道渲染适配打底。
- [chore] 预置企业微信、飞书、Telegram、钉钉、Slack 平台 renderer 元数据，暂不改变默认推送报告入口和可见版式。
- [改进] 通知报告默认发送路径恢复既有渠道兼容转换与分片逻辑，新增 renderer 能力仅保留为未来扩展基础。
- [文档] 明确 #1311 的兼容性边界：渲染层仅消费分析结果 `model_used` 展示字段，未改动 `wechat/slack/feishu/telegram` sender 发送链路，不触发 provider/model/base_url 兼容迁移。
- [改进] 关联板块缺少类型数据时改为单行展示板块名称，避免生成整列 `N/A` 的板块表格。
- [修复] 加固 LLM channel base_url 校验，避免解析差异导致 SSRF 绕过。
- [修复] 修正 efinance ETF 日线 Eastmoney secid 路由，避免沪市 ETF 被按深市 quote id 查询导致日线为空。
docs/notifications.md
+13
View File
@@ -35,6 +35,19 @@
- WebPush、Apprise、更细粒度路由、跨进程降噪和真实每日摘要暂不进入运行时实现；相关配置如未来引入，应先更新本文档、`.env.example`、Web 元数据与回归测试。
- Bark 保持 custom webhook 基线，不新增 `BARK_*` 一等配置。
## 报告渲染与分片
当前默认推送报告的入口、内容来源和整体版式保持不变。本阶段只收敛通知渲染的技术路线：沉淀渠道能力画像、发送前消息结构和结构感知分片能力，避免后续按渠道扩展时继续在各 sender 中堆叠平行逻辑。
默认发送路径沿用既有 sender 行为，不接入新增 renderer：飞书和 Telegram 继续使用原有兼容转换，企业微信、Slack 继续使用原有分片逻辑，避免改变线上可见报告版式。新增的渠道能力画像、PreparedMessage、renderer preset 和结构感知分片仅作为后续扩展基础；如需启用企业微信、飞书、Telegram、Slack 等渠道专用 renderer，应通过显式配置、真实发送验证和回归测试逐步接入。
兼容性排除说明：
- 本轮未改动 `src/notification_sender/wechat_sender.py`、`src/notification_sender/slack_sender.py`、`src/notification_sender/feishu_sender.py`、`src/notification_sender/telegram_sender.py` 的发送路径；现有 `send_to_*` 调用链（`src/notification.py -> sender method`）沿用既有行为。
- `model_used` 只在报告渲染末尾展示，不参与 provider/model/base_url 的 runtime 选择、保存、清理或迁移。若某次 CI 扫描到“provider/API 兼容迁移”类关键词，命中范围应优先回归到测试夹具中的 `model_used` 示例与报告快照 fixture（`tests/fixtures/notification_reports/*.md`），以及 `src/notification.py` 对 `report_show_llm_model` 的仅展示开关逻辑。
- `REPORT_SHOW_LLM_MODEL` 与 `report_renderer_enabled` 均为展示/降级策略开关：关闭仅影响报告可见结构，不会触发配置迁移或运行时参数回退；回退方式为恢复 `true`（或移除该项）或恢复默认配置。
关联板块渲染保持报告正文生成阶段处理：当板块表现数据不可用且所有板块类型均缺失时，只输出一行板块名称；有板块类型或板块涨跌榜信号时继续使用表格。
## GitHub Actions 映射
仓库自带 `.github/workflows/00-daily-analysis.yml` 只显式导入固定变量名。P0/P3/P4/P6 已把 Body 模板、安全项、PushPlus topic、路由、降噪、ntfy 和 Gotify 等通知 key 纳入默认 workflow。下面的表格由 `scripts/generate_notification_actions_env_table.py` 从 workflow `env:` 和通知诊断元数据生成，避免手写对照表和真实 Actions 映射继续漂移。
src/core/pipeline.py
-2
View File
@@ -34,10 +34,8 @@ from src.analyzer import (
normalize_chip_structure_availability,
stabilize_decision_with_structure,
)
from src.data.stock_mapping import STOCK_NAME_MAP
from src.notification import NotificationService, NotificationChannel
from src.report_language import (
get_unknown_text,
infer_decision_type_from_advice,
localize_confidence_level,
localize_operation_advice,
src/formatters.py
+415 -5
View File
@@ -8,7 +8,7 @@
"""
import re
from typing import List
from typing import Callable, List, Optional
import markdown2
@@ -18,6 +18,8 @@ PAGE_MARKER_SAFE_BYTES = 16 # "\n\n📄 9999/9999"
PAGE_MARKER_SAFE_LEN = 13 # "\n\n📄 9999/9999"
MIN_MAX_WORDS = 10
MIN_MAX_BYTES = 40
FENCED_CODE_BLOCK_RE = re.compile(r"(^```[^\n]*\n.*?^```[ \t]*$)", re.MULTILINE | re.DOTALL)
FENCED_CODE_BLOCK_PLACEHOLDER = "@@DSA_FENCED_CODE_BLOCK_{}@@"
# Unicode code point ranges for special characters.
_SPECIAL_CHAR_RANGE = (0x10000, 0xFFFFF)
@@ -264,6 +266,306 @@ def _bytes(s: str) -> int:
return len(s.encode('utf-8'))
def utf8_len(s: str) -> int:
"""Return the number of UTF-8 bytes used by ``s``."""
return len(s.encode("utf-8"))
def utf16_len(s: str) -> int:
"""Return the number of UTF-16 code units used by ``s``.
Telegram's 4096-character message limit is effectively counted in UTF-16
units, so astral-plane characters such as emoji consume two units.
"""
return len(s.encode("utf-16-le")) // 2
def _custom_unit_to_index(text: str, budget: int, len_fn: Callable[[str], int]) -> int:
"""Map a custom-unit budget to the largest safe Python string index."""
if len_fn(text) <= budget:
return len(text)
lo, hi = 0, len(text)
while lo < hi:
mid = (lo + hi + 1) // 2
if len_fn(text[:mid]) <= budget:
lo = mid
else:
hi = mid - 1
return lo
def _has_unclosed_inline_code(text: str) -> bool:
"""Return whether ``text`` ends inside a single-backtick inline code span."""
escaped = False
count = 0
i = 0
while i < len(text):
ch = text[i]
if ch == "\\" and not escaped:
escaped = True
i += 1
continue
if ch == "`" and not escaped:
# Triple backticks are handled as fenced code by the caller; skip
# them here so fence delimiters do not look like inline spans.
if text[i:i + 3] == "```":
i += 3
escaped = False
continue
count += 1
escaped = False
i += 1
return count % 2 == 1
def _last_unclosed_markdown_link_start(text: str) -> int:
"""Return the start index of an inline Markdown link split in progress."""
last_open_paren = text.rfind("](")
last_close_paren = text.rfind(")")
if last_open_paren > last_close_paren:
label_start = text.rfind("[", 0, last_open_paren)
return label_start if label_start >= 0 else last_open_paren
last_open_bracket = text.rfind("[")
last_close_bracket = text.rfind("]")
if last_open_bracket > last_close_bracket:
return last_open_bracket
return -1
def chunk_markdown_preserving_blocks(
content: str,
max_units: int,
*,
len_fn: Optional[Callable[[str], int]] = None,
add_page_marker: bool = False,
) -> List[str]:
"""Split Markdown while preserving common formatting boundaries.
The splitter is intentionally conservative and does not alter report
semantics. If a split lands inside a fenced code block, the current chunk is
closed and the next chunk reopens the same fence language. It also avoids
splitting inside inline code spans and Markdown links, and supports custom
length functions such as :func:`utf16_len`.
"""
measure = len_fn or len
if max_units < MIN_MAX_WORDS:
raise ValueError(f"max_units={max_units} < {MIN_MAX_WORDS}, 可能陷入无限递归。")
if measure(content) <= max_units:
return [content]
marker_reserve = measure(_page_marker(9998, 9998)) if add_page_marker else 0
indicator_reserve = measure("\n\n(9999/9999)")
fence_close = "\n```"
chunks: List[str] = []
remaining = content
carry_lang: Optional[str] = None
while remaining:
prefix = f"```{carry_lang}\n" if carry_lang is not None else ""
headroom = max_units - marker_reserve - indicator_reserve - measure(prefix) - measure(fence_close)
if headroom < MIN_MAX_WORDS:
headroom = max(MIN_MAX_WORDS, max_units - marker_reserve - indicator_reserve - measure(prefix))
if headroom <= 0:
raise ValueError("max_units is too small for markdown-preserving chunking")
if measure(prefix) + measure(remaining) <= max_units - marker_reserve - indicator_reserve:
chunks.append(prefix + remaining)
break
cp_limit = (
_custom_unit_to_index(remaining, headroom, measure)
if measure is not len else min(headroom, len(remaining))
)
region = remaining[:cp_limit]
split_at = region.rfind("\n\n")
if split_at < cp_limit // 2:
split_at = region.rfind("\n")
if split_at < cp_limit // 2:
split_at = region.rfind(" ")
if split_at < 1:
split_at = cp_limit
candidate = remaining[:split_at]
unsafe_start = len(candidate)
if _has_unclosed_inline_code(candidate):
last_tick = candidate.rfind("`")
if last_tick >= 0:
unsafe_start = min(unsafe_start, last_tick)
link_start = _last_unclosed_markdown_link_start(candidate)
if link_start >= 0:
unsafe_start = min(unsafe_start, link_start)
if unsafe_start < len(candidate):
safe_split = max(candidate.rfind(" ", 0, unsafe_start), candidate.rfind("\n", 0, unsafe_start))
if safe_split > 0:
split_at = safe_split
chunk_body = remaining[:split_at].rstrip()
in_code = carry_lang is not None
lang = carry_lang or ""
for line in chunk_body.split("\n"):
stripped = line.strip()
if stripped.startswith("```"):
if in_code:
in_code = False
lang = ""
else:
in_code = True
tag = stripped[3:].strip()
lang = tag.split()[0] if tag else ""
next_remaining = remaining[split_at:]
if next_remaining.startswith("\n"):
next_remaining = next_remaining[1:]
elif not in_code and next_remaining.startswith(" "):
line_start = remaining.rfind("\n", 0, split_at) + 1
if remaining[line_start:split_at].strip(" \t"):
next_remaining = next_remaining[1:]
remaining = next_remaining
full_chunk = prefix + chunk_body
if in_code:
full_chunk += fence_close
carry_lang = lang
else:
carry_lang = None
chunks.append(full_chunk)
if len(chunks) > 1:
total = len(chunks)
for i, chunk in enumerate(chunks):
suffix = f"\n\n({i + 1}/{total})"
if add_page_marker:
suffix += _page_marker(i, total)
chunks[i] = chunk + suffix
elif add_page_marker:
chunks[0] = chunks[0] + _page_marker(0, 1)
return chunks
def _is_markdown_table_separator(row: str) -> bool:
return bool(re.match(r'^\s*\|?\s*[:-]+\s*(\|\s*[:-]+\s*)+\|?\s*$', row))
def _parse_markdown_table_row(row: str) -> List[str]:
cells = [c.strip() for c in row.strip().strip('|').split('|')]
return [c for c in cells if c]
def _strip_inline_markdown(text: str) -> str:
text = text.strip()
text = re.sub(r'^\*\*(.+)\*\*$', r'\1', text)
text = re.sub(r'^\*(.+)\*$', r'\1', text)
text = text.replace("**", "")
return text.strip()
def _format_two_column_table_row(header: List[str], row: List[str]) -> str:
key = _strip_inline_markdown(row[0]) if row else ""
value = _strip_inline_markdown(row[1]) if len(row) > 1 else ""
value_header = _strip_inline_markdown(header[1]) if len(header) > 1 else ""
if not value:
return key
if value.upper() == "N/A" and value_header in {"类型", "Type"}:
return key
return f"{key}：{value}"
def _flush_table_as_key_value_rows(buffer: List[str], output: List[str], *, bullet: str) -> None:
if not buffer:
return
rows = []
for raw in buffer:
if _is_markdown_table_separator(raw):
continue
parsed = _parse_markdown_table_row(raw)
if parsed:
rows.append(parsed)
if not rows:
return
header = rows[0]
data_rows = rows[1:] if len(rows) > 1 else []
for row in data_rows:
if len(header) == 2 and len(row) >= 2:
output.append(f"{bullet} {_format_two_column_table_row(header, row)}")
continue
pairs = []
for idx, cell in enumerate(row):
key = _strip_inline_markdown(header[idx]) if idx < len(header) else f"列{idx + 1}"
pairs.append(f"{key}：{_strip_inline_markdown(cell)}")
output.append(f"{bullet} {' | '.join(pairs)}")
def _protect_fenced_code_blocks(content: str) -> tuple[str, List[str]]:
blocks: List[str] = []
def _replace(match: re.Match) -> str:
blocks.append(match.group(0))
return FENCED_CODE_BLOCK_PLACEHOLDER.format(len(blocks) - 1)
return FENCED_CODE_BLOCK_RE.sub(_replace, content), blocks
def _restore_fenced_code_blocks(content: str, blocks: List[str]) -> str:
restored = content
for idx, block in enumerate(blocks):
restored = restored.replace(FENCED_CODE_BLOCK_PLACEHOLDER.format(idx), block)
return restored
def _transform_outside_fenced_code_blocks(content: str, transform: Callable[[str], str]) -> str:
protected, blocks = _protect_fenced_code_blocks(content)
return _restore_fenced_code_blocks(transform(protected), blocks)
def _markdown_tables_to_key_value_rows_unprotected(content: str, *, bullet: str) -> str:
"""Convert pipe tables to compact key-value rows for chat clients."""
lines: List[str] = []
table_buffer: List[str] = []
for raw_line in content.splitlines():
line = raw_line.rstrip()
if line.strip().startswith('|'):
table_buffer.append(line)
continue
if table_buffer:
_flush_table_as_key_value_rows(table_buffer, lines, bullet=bullet)
table_buffer = []
lines.append(line)
if table_buffer:
_flush_table_as_key_value_rows(table_buffer, lines, bullet=bullet)
return "\n".join(lines).strip()
def markdown_tables_to_key_value_rows(content: str, *, bullet: str = "•") -> str:
"""Convert pipe tables to compact key-value rows outside fenced code blocks."""
return _transform_outside_fenced_code_blocks(
content,
lambda text: _markdown_tables_to_key_value_rows_unprotected(text, bullet=bullet),
).strip()
def _chunk_by_max_bytes(content: str, max_bytes: int) -> List[str]:
if _bytes(content) <= max_bytes:
return [content]
@@ -398,22 +700,63 @@ def slice_at_max_bytes(text: str, max_bytes: int) -> tuple[str, str]:
return truncated, text[len(truncated):]
def _format_feishu_markdown_unprotected(content: str) -> str:
lines = []
table_buffer: List[str] = []
for raw_line in content.splitlines():
line = raw_line.rstrip()
# 处理表格行
if line.strip().startswith('|'):
table_buffer.append(line)
continue
# 刷新表格缓冲区
if table_buffer:
_flush_table_as_key_value_rows(table_buffer, lines, bullet="•")
table_buffer = []
# 转换标题（# ## ### 等）
if re.match(r'^#{1,6}\s+', line):
title = re.sub(r'^#{1,6}\s+', '', line).strip()
line = f"**{title}**" if title else ""
# 转换引用块
elif line.startswith('> '):
quote = line[2:].strip()
line = quote
# 转换分隔线
elif line.strip() == '---':
line = '────────'
# 转换列表项
elif line.startswith('- '):
line = f"• {line[2:].strip()}"
lines.append(line)
# 处理末尾的表格
if table_buffer:
_flush_table_as_key_value_rows(table_buffer, lines, bullet="•")
return "\n".join(lines).strip()
def format_feishu_markdown(content: str) -> str:
"""
将通用 Markdown 转换为飞书 lark_md 更友好的格式
转换规则：
- 飞书不支持 Markdown 标题（# / ## / ###），用加粗代替
- 引用块使用前缀替代
- 分隔线统一为细线
- 表格转换为条目列表
Args:
content: 原始 Markdown 内容
Returns:
转换后的飞书 Markdown 格式内容
Example:
>>> markdown = "# 标题\\n> 引用\\n| 列1 | 列2 |"
>>> formatted = format_feishu_markdown(markdown)
@@ -493,6 +836,73 @@ def format_feishu_markdown(content: str) -> str:
return "\n".join(lines).strip()
def _format_telegram_markdown_unprotected(content: str) -> str:
"""Convert common report Markdown to Telegram legacy Markdown."""
result = _markdown_tables_to_key_value_rows_unprotected(content, bullet="-")
result = re.sub(r'^#{1,6}\s+(.+)$', r'*\1*', result, flags=re.MULTILINE)
result = re.sub(r'\*\*(.+?)\*\*', r'*\1*', result)
result = re.sub(r'^\s*---+\s*$', '────────', result, flags=re.MULTILINE)
result = _escape_telegram_non_link_markdown_chars(result)
return result.strip()
def _escape_telegram_non_link_markdown_chars(content: str) -> str:
"""Escape Telegram Markdown link metacharacters outside valid links."""
links: list[str] = []
def _save_link(match: re.Match) -> str:
links.append(match.group(0))
return f"@@DSA_TELEGRAM_LINK_{len(links) - 1}@@"
result = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', _save_link, content)
for char in ("[", "]", "(", ")"):
result = result.replace(char, f"\\{char}")
for index, link in enumerate(links):
result = result.replace(f"@@DSA_TELEGRAM_LINK_{index}@@", link)
return result
def format_telegram_markdown(content: str) -> str:
"""Convert common report Markdown to Telegram legacy Markdown."""
return _transform_outside_fenced_code_blocks(
content,
_format_telegram_markdown_unprotected,
).strip()
def format_wechat_markdown(content: str) -> str:
"""Keep WeChat Markdown style while making pipe tables mobile-readable."""
result = markdown_tables_to_key_value_rows(content, bullet="•")
result = re.sub(r'^\s*---+\s*$', '────────', result, flags=re.MULTILINE)
return result.strip()
def _format_slack_mrkdwn_unprotected(content: str) -> str:
"""Convert common report Markdown to Slack mrkdwn."""
result = _markdown_tables_to_key_value_rows_unprotected(content, bullet="•")
result = re.sub(r'!\[([^\]]*)\]\(([^)]+)\)', r'<\2|\1>', result)
result = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'<\2|\1>', result)
result = re.sub(r'^#{1,6}\s+(.+)$', r'*\1*', result, flags=re.MULTILINE)
result = re.sub(r'\*\*(.+?)\*\*', r'*\1*', result)
result = re.sub(r'^\s*---+\s*$', '────────', result, flags=re.MULTILINE)
return result.strip()
def format_slack_mrkdwn(content: str) -> str:
"""Convert common report Markdown to Slack mrkdwn."""
return _transform_outside_fenced_code_blocks(
content,
_format_slack_mrkdwn_unprotected,
).strip()
def _chunk_by_separators(content: str) -> tuple[list[str], str]:
"""
通过分割线等特殊字符将消息内容分割为多个区块
src/notification.py
+4
View File
@@ -2035,6 +2035,10 @@ class NotificationService(
change = "--" if change_pct is None else f"{change_pct:+.2f}%"
lines.append(f"| {name} | {board_type} | {status} | {change} |")
else:
if all(board_type == "N/A" for _, board_type, _, _ in prepared):
lines.append(" / ".join(name for name, _, _, _ in prepared))
lines.append("")
return
lines.append(f"| {labels['board_name_label']} | {labels['board_type_label']} |")
lines.append("|:-----|:----:|")
for name, board_type, _, _ in prepared:
src/notification_capabilities.py
+269
View File
@@ -0,0 +1,269 @@
# -*- coding: utf-8 -*-
"""Notification channel rendering capability profiles.
This module intentionally uses plain channel strings instead of importing
``NotificationChannel`` from ``src.notification``. The notification service may
import these profiles later without creating a circular dependency.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Mapping, Optional, Tuple
@dataclass(frozen=True)
class ChannelProfile:
"""Static rendering capabilities for one notification channel."""
channel: str
markdown: str
default_mode: str
max_text_chars: Optional[int] = None
max_text_bytes: Optional[int] = None
supports_card: bool = False
supports_image: bool = False
supports_file: bool = False
supports_link: bool = True
notes: str = ""
@dataclass(frozen=True)
class PreparedMessage:
"""A channel-specific prepared notification message.
The object describes how a message is ready to be sent for a channel without
changing the original report semantics. Senders can consume the fields they
support and fall back to ``fallback_text`` or ``text`` when a richer payload
is unavailable.
"""
channel: str
text: str
formatted_text: Optional[str] = None
card_payload: Optional[Mapping[str, Any]] = None
fallback_text: Optional[str] = None
attachments: Tuple[Any, ...] = ()
diagnostics: Tuple[str, ...] = ()
@property
def content_for_text_send(self) -> str:
"""Return the best text payload for legacy text senders."""
return self.formatted_text or self.fallback_text or self.text
@dataclass(frozen=True)
class RendererPreset:
"""Reserved renderer plan for one notification channel.
Presets document the intended renderer shape without changing today's
runtime send path. A future opt-in implementation can use these names to
wire platform-specific renderers while keeping the legacy text fallback.
"""
channel: str
text_renderer: str
markdown: str
enabled_by_default: bool = False
rich_renderer: Optional[str] = None
image_renderer: Optional[str] = None
fallback_renderer: str = "legacy_text"
notes: str = ""
CHANNEL_PROFILES: Dict[str, ChannelProfile] = {
"wechat": ChannelProfile(
channel="wechat",
markdown="wechat_markdown",
default_mode="full_report",
max_text_bytes=4096,
supports_image=True,
supports_link=True,
notes="Enterprise WeChat receives the full report by default and relies on safe chunking.",
),
"feishu": ChannelProfile(
channel="feishu",
markdown="lark_md",
default_mode="full_report",
max_text_bytes=20000,
supports_card=True,
supports_file=True,
supports_link=True,
notes="Feishu uses lark_md/card payloads and needs table fallbacks.",
),
"telegram": ChannelProfile(
channel="telegram",
markdown="markdown_v2",
default_mode="full_report",
max_text_chars=4096,
supports_image=True,
supports_link=True,
notes="Telegram length limits are measured in UTF-16 code units.",
),
"email": ChannelProfile(
channel="email",
markdown="html",
default_mode="full_html",
supports_image=True,
supports_file=True,
supports_link=True,
notes="Email remains the high-fidelity full-report carrier.",
),
"pushover": ChannelProfile(
channel="pushover",
markdown="plain_text",
default_mode="plain_fallback",
max_text_chars=1024,
supports_link=True,
),
"ntfy": ChannelProfile(
channel="ntfy",
markdown="plain_text",
default_mode="plain_fallback",
supports_link=True,
),
"gotify": ChannelProfile(
channel="gotify",
markdown="markdown",
default_mode="full_report",
supports_link=True,
),
"pushplus": ChannelProfile(
channel="pushplus",
markdown="markdown",
default_mode="full_report",
supports_link=True,
),
"serverchan3": ChannelProfile(
channel="serverchan3",
markdown="markdown",
default_mode="full_report",
supports_link=True,
),
"custom": ChannelProfile(
channel="custom",
markdown="channel_specific",
default_mode="full_report",
supports_image=True,
supports_link=True,
notes="Custom webhook payload shape can be configured by templates.",
),
"discord": ChannelProfile(
channel="discord",
markdown="discord_markdown",
default_mode="full_report",
max_text_chars=2000,
supports_link=True,
),
"slack": ChannelProfile(
channel="slack",
markdown="mrkdwn",
default_mode="full_report",
max_text_chars=39000,
supports_image=True,
supports_file=True,
supports_link=True,
notes="Slack sections should avoid splitting markdown blocks.",
),
"astrbot": ChannelProfile(
channel="astrbot",
markdown="plain_text",
default_mode="plain_fallback",
supports_link=True,
),
"unknown": ChannelProfile(
channel="unknown",
markdown="plain_text",
default_mode="plain_fallback",
supports_link=False,
),
}
CHANNEL_RENDERER_PRESETS: Dict[str, RendererPreset] = {
"wechat": RendererPreset(
channel="wechat",
text_renderer="wecom_markdown",
markdown="wechat_markdown",
rich_renderer="wecom_card",
image_renderer="png_poster",
notes="Preset only; current runtime keeps the legacy WeCom dashboard text.",
),
"feishu": RendererPreset(
channel="feishu",
text_renderer="feishu_lark_md",
markdown="lark_md",
rich_renderer="feishu_interactive_card",
image_renderer="png_poster",
notes="Preset only; native card rendering is not enabled by default.",
),
"telegram": RendererPreset(
channel="telegram",
text_renderer="telegram_markdown_v2",
markdown="markdown_v2",
rich_renderer="telegram_html",
image_renderer="png_poster",
notes="Preset only; future renderer must handle Telegram escaping and UTF-16 length limits.",
),
"dingtalk": RendererPreset(
channel="dingtalk",
text_renderer="dingtalk_markdown",
markdown="dingtalk_markdown",
rich_renderer="dingtalk_action_card",
image_renderer="png_poster",
notes="Preset only; DingTalk is not an active NotificationChannel in this runtime yet.",
),
"slack": RendererPreset(
channel="slack",
text_renderer="slack_mrkdwn",
markdown="mrkdwn",
rich_renderer="slack_blocks",
image_renderer="png_poster",
notes="Preset only; current runtime keeps the legacy report text fallback.",
),
}
def normalize_channel_name(channel: Any) -> str:
"""Normalize enum-like or string channel values into profile keys."""
value = getattr(channel, "value", channel)
return str(value or "").strip().lower() or "unknown"
def get_channel_profile(channel: Any) -> ChannelProfile:
"""Return the channel profile, falling back to ``unknown``."""
name = normalize_channel_name(channel)
return CHANNEL_PROFILES.get(name, CHANNEL_PROFILES["unknown"])
def all_channel_profiles() -> Tuple[ChannelProfile, ...]:
"""Return all profiles in deterministic declaration order."""
return tuple(CHANNEL_PROFILES.values())
def get_renderer_preset(channel: Any) -> RendererPreset:
"""Return the reserved renderer preset for ``channel``.
Unknown channels use a plain text fallback preset and stay disabled.
"""
name = normalize_channel_name(channel)
return CHANNEL_RENDERER_PRESETS.get(
name,
RendererPreset(
channel=name,
text_renderer="plain_text",
markdown="plain_text",
notes="Fallback preset for channels without a dedicated renderer plan.",
),
)
def all_renderer_presets() -> Tuple[RendererPreset, ...]:
"""Return all reserved renderer presets in deterministic declaration order."""
return tuple(CHANNEL_RENDERER_PRESETS.values())
src/notification_sender/feishu_sender.py
+21 -21
View File
@@ -27,7 +27,7 @@ logger = logging.getLogger(__name__)
class FeishuSender:
def __init__(self, config: Config):
"""
初始化飞书配置
@@ -71,12 +71,12 @@ class FeishuSender:
"timestamp": timestamp,
"sign": sign,
}
def send_to_feishu(self, content: str, *, timeout_seconds: Optional[float] = None) -> bool:
"""
推送消息到飞书机器人
飞书自定义机器人 Webhook 消息格式：
{
"msg_type": "interactive",
@@ -99,22 +99,22 @@ class FeishuSender:
}
}
}
说明：飞书文本消息不会渲染 Markdown，需使用交互卡片（lark_md）格式
注意：飞书文本消息限制约 20KB，超长内容会自动分批发送
可通过环境变量 FEISHU_MAX_BYTES 调整限制值
Args:
content: 消息内容（Markdown 会转为纯文本）
Returns:
是否发送成功
"""
if not self._feishu_url:
logger.warning("飞书 Webhook 未配置，跳过推送")
return False
# 飞书 lark_md 支持有限，先做格式转换
formatted_content = format_feishu_markdown(content)
@@ -125,7 +125,7 @@ class FeishuSender:
if effective_max_bytes <= 0:
logger.error("飞书关键词过长，超过单条消息允许的最大字节数，无法发送")
return False
# 检查字节长度，超长则分批发送
content_bytes = len(formatted_content.encode('utf-8')) + keyword_overhead
if content_bytes > max_bytes:
@@ -139,23 +139,23 @@ class FeishuSender:
return False
logger.info(f"飞书消息内容超长({content_bytes}字节/{len(content)}字符)，将分批发送")
return self._send_feishu_chunked(formatted_content, effective_max_bytes)
try:
return self._send_feishu_message(formatted_content, timeout_seconds=timeout_seconds)
except Exception as e:
logger.error(f"发送飞书消息失败: {e}")
return False
def _send_feishu_chunked(self, content: str, max_bytes: int) -> bool:
"""
分批发送长消息到飞书
按股票分析块（以 --- 或 ### 分隔）智能分割，确保每批不超过限制
Args:
content: 完整消息内容
max_bytes: 单条消息最大字节数
Returns:
是否全部发送成功
"""
@@ -164,13 +164,13 @@ class FeishuSender:
except ValueError as e:
logger.error("飞书消息分片失败，单片预算不足以安全分页（关键词过长或 max_bytes 过小）: %s", e)
return False
# 分批发送
total_chunks = len(chunks)
success_count = 0
logger.info(f"飞书分批发送：共 {total_chunks} 批")
for i, chunk in enumerate(chunks):
try:
if self._send_feishu_message(chunk):
@@ -180,13 +180,13 @@ class FeishuSender:
logger.error(f"飞书第 {i+1}/{total_chunks} 批发送失败")
except Exception as e:
logger.error(f"飞书第 {i+1}/{total_chunks} 批发送异常: {e}")
# 批次间隔，避免触发频率限制
if i < total_chunks - 1:
time.sleep(1)
return success_count == total_chunks
def _send_feishu_message(self, content: str, *, timeout_seconds: Optional[float] = None) -> bool:
"""发送单条飞书消息（优先使用 Markdown 卡片）"""
prepared_content = self._apply_keyword_prefix(content)
src/notification_sender/telegram_sender.py
+27 -28
View File
@@ -19,7 +19,7 @@ logger = logging.getLogger(__name__)
class TelegramSender:
def __init__(self, config: Config):
"""
初始化 Telegram 配置
@@ -32,11 +32,11 @@ class TelegramSender:
'chat_id': getattr(config, 'telegram_chat_id', None),
'message_thread_id': getattr(config, 'telegram_message_thread_id', None),
}
def _is_telegram_configured(self) -> bool:
"""检查 Telegram 配置是否完整"""
return bool(self._telegram_config['bot_token'] and self._telegram_config['chat_id'])
def send_to_telegram(
self,
content: str,
@@ -47,7 +47,7 @@ class TelegramSender:
) -> bool:
"""
推送消息到 Telegram 机器人
Telegram Bot API 格式：
POST https://api.telegram.org/bot<token>/sendMessage
{
@@ -55,10 +55,10 @@ class TelegramSender:
"text": "消息内容",
"parse_mode": "Markdown"
}
Args:
content: 消息内容（Markdown 格式）
Returns:
是否发送成功
"""
@@ -76,27 +76,27 @@ class TelegramSender:
bot_token = self._telegram_config['bot_token']
chat_id = target_chat_id
message_thread_id = target_message_thread_id
try:
# Telegram API 端点
api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
# Telegram 消息最大长度 4096 字符
max_length = 4096
if len(content) <= max_length:
# 单条消息发送
return self._send_telegram_message(api_url, chat_id, content, message_thread_id, timeout_seconds=timeout_seconds)
else:
# 分段发送长消息
return self._send_telegram_chunked(api_url, chat_id, content, max_length, message_thread_id, timeout_seconds=timeout_seconds)
except Exception as e:
logger.error(f"发送 Telegram 消息失败: {e}")
import traceback
logger.debug(traceback.format_exc())
return False
def _send_telegram_message(
self,
api_url: str,
@@ -109,7 +109,7 @@ class TelegramSender:
"""Send a single Telegram message with exponential backoff retry (Fixes #287)"""
# Convert Markdown to Telegram-compatible format
telegram_text = self._convert_to_telegram_markdown(text)
payload = {
"chat_id": chat_id,
"text": telegram_text,
@@ -134,7 +134,7 @@ class TelegramSender:
else:
logger.error(f"Telegram request failed after {max_retries} attempts: {e}")
return False
if response.status_code == 200:
result = response.json()
if result.get('ok'):
@@ -143,12 +143,12 @@ class TelegramSender:
else:
error_desc = result.get('description', '未知错误')
logger.error(f"Telegram 返回错误: {error_desc}")
# If Markdown parsing failed, fall back to plain text
if self._should_fallback_to_plain_text(error_desc=error_desc):
if self._send_plain_text_fallback(api_url, payload, text, timeout_seconds=timeout_seconds):
return True
return False
elif response.status_code == 429:
# Rate limited — respect Retry-After header
@@ -230,7 +230,7 @@ class TelegramSender:
logger.error(f"Telegram 纯文本回退失败: HTTP {response.status_code}")
logger.error(f"响应内容: {response.text}")
return False
def _send_telegram_chunked(
self,
api_url: str,
@@ -244,15 +244,15 @@ class TelegramSender:
"""分段发送长 Telegram 消息"""
# 按段落分割
sections = content.split("\n---\n")
current_chunk = []
current_length = 0
all_success = True
chunk_index = 1
for section in sections:
section_length = len(section) + 5 # +5 for "\n---\n"
if current_length + section_length > max_length:
# 发送当前块
if current_chunk:
@@ -261,21 +261,21 @@ class TelegramSender:
if not self._send_telegram_message(api_url, chat_id, chunk_content, message_thread_id, timeout_seconds=timeout_seconds):
all_success = False
chunk_index += 1
# 重置
current_chunk = [section]
current_length = section_length
else:
current_chunk.append(section)
current_length += section_length
# 发送最后一块
if current_chunk:
chunk_content = "\n---\n".join(current_chunk)
logger.info(f"发送 Telegram 消息块 {chunk_index}...")
if not self._send_telegram_message(api_url, chat_id, chunk_content, message_thread_id, timeout_seconds=timeout_seconds):
all_success = False
return all_success
def _send_telegram_photo(self, image_bytes: bytes) -> bool:
@@ -304,20 +304,20 @@ class TelegramSender:
def _convert_to_telegram_markdown(self, text: str) -> str:
"""
将标准 Markdown 转换为 Telegram 支持的格式
Telegram Markdown 限制：
- 不支持 # 标题
- 使用 *bold* 而非 **bold**
- 使用 _italic_
- 使用 _italic_
"""
result = text
# 移除 # 标题标记（Telegram 不支持）
result = re.sub(r'^#{1,6}\s+', '', result, flags=re.MULTILINE)
# 转换 **bold** 为 *bold*
result = re.sub(r'\*\*(.+?)\*\*', r'*\1*', result)
# Escape special characters for Telegram Markdown, but preserve link syntax [text](url)
# Step 1: temporarily protect markdown links
import uuid as _uuid
@@ -337,4 +337,3 @@ class TelegramSender:
result = result.replace(f"{_link_placeholder}{i}", link)
return result
tests/fixtures/notification_reports/aggregate_report.md
Vendored
+21
View File
@@ -0,0 +1,21 @@
# A股智能分析日报
## 今日结论
- 市场风险偏好回升，短线资金偏向高股息与 AI 算力。
- 默认仍发送完整报告，本样例用于验证不同通知渠道的格式化保真。
| 股票 | 信号 | 建议 |
| --- | --- | --- |
| 600519 | 趋势强势 | 逢低关注 |
| AAPL | 震荡整理 | 等待放量 |
## 风险提示
> 样例不构成投资建议。
```text
volume_signal = "watch"
```
[查看完整报告](https://example.com/reports/daily-stock-analysis)
tests/fixtures/notification_reports/market_review_report.md
Vendored
+17
View File
@@ -0,0 +1,17 @@
# 大盘复盘
## 指数表现
| 指数 | 涨跌幅 | 量能 |
| --- | --- | --- |
| 上证指数 | +0.8% | 放量 |
| 创业板指 | -0.2% | 缩量 |
## 资金方向
- 北向资金小幅净流入。
- 高股息、半导体、消费电子轮动活跃。
## 明日观察
关注成交额能否维持在万亿附近，以及权重板块是否继续承接。
tests/fixtures/notification_reports/single_stock_report.md
Vendored
+18
View File
@@ -0,0 +1,18 @@
# 贵州茅台 600519 分析报告
## 核心判断
- 技术面：均线多头排列，成交量温和放大。
- 基本面：盈利质量稳定，估值处于历史中位附近。
| 指标 | 当前值 | 解释 |
| --- | --- | --- |
| RSI | 58 | 中性偏强 |
| MACD | 金叉 | 动能改善 |
## 操作计划
1. 回踩关键均线不破时观察。
2. 跌破风控位时降低仓位。
`risk_level = medium`
tests/test_feishu_stream.py
+20 -15
View File
@@ -1,6 +1,5 @@
import pytest
from bot.platforms.feishu_stream import FeishuReplyClient
from src.formatters import format_feishu_markdown
class DummyFeishuReplyClient(FeishuReplyClient):
@@ -31,19 +30,6 @@ class DummyFeishuReplyClient(FeishuReplyClient):
return True
@pytest.fixture(autouse=True)
def patch_format_feishu_markdown(monkeypatch):
# Keep formatting simple so byte length is predictable
import bot.platforms.feishu_stream as feishu_stream
monkeypatch.setattr(
feishu_stream,
"format_feishu_markdown",
lambda text: text,
)
yield
def test_reply_text_chunked_keeps_reply_and_at_user(monkeypatch):
client = DummyFeishuReplyClient(max_bytes=1000)
@@ -64,6 +50,16 @@ def test_reply_text_chunked_keeps_reply_and_at_user(monkeypatch):
assert call["user_id"] == user_id
def test_reply_text_uses_legacy_feishu_markdown_formatter():
client = DummyFeishuReplyClient(max_bytes=1000)
text = "# 日报\n\n## 📊 分析结果摘要\n\n| 股票 | 信号 |\n| --- | --- |\n| 600519 | 强势 |"
result = client.reply_text(message_id="msg_123", text=text)
assert result is True
assert client.calls[0]["content"] == format_feishu_markdown(text)
def test_send_to_chat_chunked_uses_chat_id(monkeypatch):
client = DummyFeishuReplyClient(max_bytes=1000)
@@ -82,3 +78,12 @@ def test_send_to_chat_chunked_uses_chat_id(monkeypatch):
assert call["at_user"] is False
assert call["user_id"] is None
def test_send_to_chat_uses_legacy_feishu_markdown_formatter():
client = DummyFeishuReplyClient(max_bytes=1000)
text = "# 日报\n\n[详情](https://example.com/report)"
result = client.send_to_chat(chat_id="chat_123", text=text)
assert result is True
assert client.calls[0]["content"] == format_feishu_markdown(text)
tests/test_formatters.py
+267
View File
@@ -11,12 +11,19 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")
from src.formatters import (
chunk_content_by_max_words,
chunk_content_by_max_bytes,
chunk_markdown_preserving_blocks,
format_feishu_markdown,
format_slack_mrkdwn,
format_telegram_markdown,
format_wechat_markdown,
markdown_tables_to_key_value_rows,
slice_at_max_bytes,
TRUNCATION_SUFFIX,
MIN_MAX_WORDS,
MIN_MAX_BYTES,
_slice_at_effective_len,
_chunk_by_max_words,
utf16_len,
)
@@ -176,3 +183,263 @@ class TestChunkContentByMaxBytes(unittest.TestCase):
chunk, remaining = slice_at_max_bytes("测试ABC", 7)
self.assertEqual(chunk, "测试A")
self.assertEqual(remaining, "BC")
class TestChunkMarkdownPreservingBlocks(unittest.TestCase):
"""Tests for markdown-aware notification chunking."""
@staticmethod
def _strip_chunk_suffix(chunk):
return chunk.rsplit("\n\n(", 1)[0] if "\n\n(" in chunk else chunk
def test_preserves_fenced_code_boundaries_when_split_inside_block(self):
text = (
"## Report\n\n"
"```python\n"
+ "\n".join(f"print({i})" for i in range(20))
+ "\n```\n\n"
"Done"
)
chunks = chunk_markdown_preserving_blocks(text, 120)
self.assertGreater(len(chunks), 1)
for chunk in chunks:
body = self._strip_chunk_suffix(chunk)
self.assertEqual(body.count("```") % 2, 0)
bodies = [self._strip_chunk_suffix(chunk) for chunk in chunks]
self.assertTrue(any(body.endswith("```") for body in bodies[:-1]))
self.assertTrue(any(body.startswith("```python\n") for body in bodies[1:]))
def test_preserves_indentation_when_carrying_fenced_code(self):
text = (
"## Report\n\n"
"```python\n"
+ "\n".join(f" print({i})" for i in range(20))
+ "\n```\n\n"
"Done"
)
chunks = chunk_markdown_preserving_blocks(text, 120)
carried_bodies = [
self._strip_chunk_suffix(chunk)
for chunk in chunks[1:]
if self._strip_chunk_suffix(chunk).startswith("```python\n")
]
self.assertTrue(carried_bodies)
self.assertTrue(
any(body.startswith("```python\n print(") for body in carried_bodies)
)
self.assertFalse(
any(body.startswith("```python\nprint(") for body in carried_bodies)
)
def test_preserves_space_delimiter_inside_fenced_code(self):
text = "```python\n" + "value =" + (" item" * 30) + "\n```"
chunks = chunk_markdown_preserving_blocks(text, 70)
carried_bodies = [
self._strip_chunk_suffix(chunk)
for chunk in chunks[1:]
if self._strip_chunk_suffix(chunk).startswith("```python\n")
]
self.assertTrue(any(body.startswith("```python\n item") for body in carried_bodies))
self.assertFalse(any(body.startswith("```python\nitem") for body in carried_bodies))
def test_preserves_nested_list_indentation_after_split(self):
text = (
"- parent with enough words here\n"
" - nested child keeps indentation after chunking "
+ ("word " * 20)
)
chunks = chunk_markdown_preserving_blocks(text, 70)
bodies = [self._strip_chunk_suffix(chunk) for chunk in chunks]
self.assertTrue(any(body.startswith(" - nested child") for body in bodies[1:]))
self.assertFalse(any(body.startswith("- nested child") for body in bodies[1:]))
def test_avoids_splitting_inside_inline_code(self):
text = "Intro " + ("word " * 8) + "`inline code with spaces and (parens)` tail " + ("more " * 20)
chunks = chunk_markdown_preserving_blocks(text, 70)
self.assertGreater(len(chunks), 1)
for chunk in chunks:
body = self._strip_chunk_suffix(chunk)
self.assertEqual(body.count("`") % 2, 0)
def test_avoids_splitting_inside_markdown_links(self):
link = "[full report](https://example.com/reports/2026/05/30/daily-stock-analysis?channel=feishu)"
text = (
"Intro "
+ ("word " * 8)
+ link
+ " tail "
+ ("more " * 20)
)
chunks = chunk_markdown_preserving_blocks(text, 130)
self.assertGreater(len(chunks), 1)
bodies = [self._strip_chunk_suffix(chunk) for chunk in chunks]
self.assertTrue(any(link in body for body in bodies))
self.assertEqual([body for body in bodies if "[full report" in body and link not in body], [])
def test_supports_utf16_length_function(self):
text = "😀" * 20
chunks = chunk_markdown_preserving_blocks(text, 20, len_fn=utf16_len)
self.assertGreater(len(chunks), 1)
for chunk in chunks:
self.assertLessEqual(utf16_len(chunk), 20)
def test_add_page_marker_appends_stable_markers(self):
text = "A" * 180
chunks = chunk_markdown_preserving_blocks(text, 80, add_page_marker=True)
self.assertGreater(len(chunks), 1)
for i, chunk in enumerate(chunks):
self.assertIn(f"{i + 1}/{len(chunks)}", chunk)
class TestNotificationMarkdownFormatters(unittest.TestCase):
"""Tests for chat-friendly report Markdown conversions."""
@staticmethod
def _first_fenced_code_body(text):
return text.split("```markdown\n", 1)[1].split("\n```", 1)[0]
def test_markdown_tables_to_key_value_rows(self):
text = "| 股票 | 评级 |\n| --- | --- |\n| 600519 | 买入 |\n| AAPL | 观望 |"
result = markdown_tables_to_key_value_rows(text)
self.assertNotIn("| --- |", result)
self.assertIn("• 600519：买入", result)
self.assertIn("• AAPL：观望", result)
def test_markdown_tables_to_key_value_rows_compacts_report_metric_tables(self):
text = (
"| 价格指标 | 当前价 |\n"
"|---------|------|\n"
"| MA5 | 1292.85 |\n"
"| 支撑位 | 1302.77 |\n\n"
"| 板块 | 类型 |\n"
"|:-----|:----:|\n"
"| 白酒Ⅲ | N/A |"
)
result = markdown_tables_to_key_value_rows(text)
self.assertIn("• MA5：1292.85", result)
self.assertIn("• 支撑位：1302.77", result)
self.assertIn("• 白酒Ⅲ", result)
self.assertNotIn("价格指标：MA5", result)
self.assertNotIn("类型：N/A", result)
def test_markdown_tables_to_key_value_rows_keeps_fenced_code_tables(self):
text = (
"```markdown\n"
"| 股票 | 评级 |\n"
"| --- | --- |\n"
"| 示例 | 不应转换 |\n"
"```\n\n"
"| 股票 | 评级 |\n"
"| --- | --- |\n"
"| 600519 | 买入 |"
)
result = markdown_tables_to_key_value_rows(text)
fenced = self._first_fenced_code_body(result)
self.assertIn("| --- |", fenced)
self.assertIn("| 示例 | 不应转换 |", fenced)
self.assertIn("• 600519：买入", result)
self.assertNotIn("@@DSA_FENCED_CODE_BLOCK_", result)
def test_feishu_formatter_keeps_legacy_structure_and_converts_table(self):
text = "# 日报\n\n> 风险提示\n\n| 股票 | 信号 |\n| --- | --- |\n| 600519 | 强势 |\n\n- 关注量能"
result = format_feishu_markdown(text)
self.assertIn("**日报**", result)
self.assertIn("💬 风险提示", result)
self.assertIn("• 股票：600519 | 信号：强势", result)
self.assertIn("• 关注量能", result)
def test_telegram_formatter_uses_supported_markdown(self):
text = "## 日报\n\n| 股票 | 信号 |\n| --- | --- |\n| 600519 | 强势 |\n\n[详情](https://example.com/report)"
result = format_telegram_markdown(text)
self.assertIn("*日报*", result)
self.assertIn("- 600519：强势", result)
self.assertIn("[详情](https://example.com/report)", result)
def test_telegram_formatter_escapes_non_link_metacharacters(self):
text = (
"## [P4] 日报\n\n"
"| 股票 | 信号 |\n"
"| --- | --- |\n"
"| 600519 | [P4] 强势 (观察) |\n\n"
"详见 [详情](https://example.com/report)"
)
result = format_telegram_markdown(text)
self.assertIn("*\\[P4\\] 日报*", result)
self.assertIn("- 600519：\\[P4\\] 强势 \\(观察\\)", result)
self.assertIn("[详情](https://example.com/report)", result)
def test_slack_formatter_uses_mrkdwn_links_and_tables(self):
text = "## 日报\n\n| 股票 | 信号 |\n| --- | --- |\n| 600519 | 强势 |\n\n[详情](https://example.com/report)"
result = format_slack_mrkdwn(text)
self.assertIn("*日报*", result)
self.assertIn("• 600519：强势", result)
self.assertIn("<https://example.com/report|详情>", result)
def test_wechat_formatter_keeps_markdown_but_converts_tables(self):
text = "## 日报\n\n| 股票 | 信号 |\n| --- | --- |\n| 600519 | 强势 |"
result = format_wechat_markdown(text)
self.assertIn("## 日报", result)
self.assertIn("• 600519：强势", result)
self.assertNotIn("| --- |", result)
def test_platform_formatters_do_not_rewrite_fenced_code_blocks(self):
text = (
"## 日报\n\n"
"```markdown\n"
"| 股票 | 信号 |\n"
"| --- | --- |\n"
"| 示例 | 不应转换 |\n"
"# not heading\n"
"[详情](https://example.com/raw)\n"
"```\n\n"
"| 股票 | 信号 |\n"
"| --- | --- |\n"
"| 600519 | 强势 |"
)
for formatter in (
format_telegram_markdown,
format_wechat_markdown,
format_slack_mrkdwn,
):
with self.subTest(formatter=formatter.__name__):
result = formatter(text)
fenced = self._first_fenced_code_body(result)
self.assertIn("| --- |", fenced)
self.assertIn("| 示例 | 不应转换 |", fenced)
self.assertIn("# not heading", fenced)
self.assertIn("[详情](https://example.com/raw)", fenced)
self.assertIn("600519：强势", result)
self.assertNotIn("@@DSA_FENCED_CODE_BLOCK_", result)
tests/test_notification.py
+33
View File
@@ -1033,6 +1033,39 @@ class TestNotificationServiceReportGeneration(unittest.TestCase):
# And no leftover "--" cells either.
self.assertNotIn("| -- | -- |", out)
@mock.patch("src.notification.get_config")
def test_related_boards_without_type_renders_one_line(
self, mock_get_config: mock.MagicMock
):
mock_get_config.return_value = _make_config(report_renderer_enabled=False)
service = NotificationService()
result = AnalysisResult(
code="600519",
name="贵州茅台",
sentiment_score=72,
trend_prediction="看多",
operation_advice="持有",
analysis_summary="稳健",
)
result.fundamental_context = {
"earnings": {"status": "ok", "data": {}},
"growth": {"status": "ok", "data": {}},
"belong_boards": [
{"name": "白酒Ⅲ"},
{"name": "白酒Ⅱ"},
{"name": "食品饮料"},
{"name": "贵州板块"},
{"name": "酿酒概念"},
],
}
out = service.generate_single_stock_report(result)
self.assertIn("关联板块", out)
self.assertIn("白酒Ⅲ / 白酒Ⅱ / 食品饮料 / 贵州板块 / 酿酒概念", out)
self.assertNotIn("| 板块 | 类型 |", out)
self.assertNotIn("| 白酒Ⅲ | N/A |", out)
@mock.patch("src.notification.get_config")
def test_related_boards_keeps_signal_columns_when_any_board_has_data(
self, mock_get_config: mock.MagicMock
tests/test_notification_capabilities.py
+74
View File
@@ -0,0 +1,74 @@
# -*- coding: utf-8 -*-
"""Tests for notification channel rendering capability profiles."""
import os
import sys
import unittest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.notification import NotificationChannel
from src.notification_capabilities import (
CHANNEL_PROFILES,
CHANNEL_RENDERER_PRESETS,
PreparedMessage,
all_channel_profiles,
all_renderer_presets,
get_channel_profile,
get_renderer_preset,
normalize_channel_name,
)
class NotificationCapabilityProfileTestCase(unittest.TestCase):
def test_profiles_cover_all_notification_channels(self):
profile_channels = {profile.channel for profile in all_channel_profiles()}
expected = {channel.value for channel in NotificationChannel}
self.assertTrue(expected.issubset(profile_channels))
def test_get_channel_profile_accepts_enum_or_string(self):
self.assertEqual(get_channel_profile(NotificationChannel.FEISHU).channel, "feishu")
self.assertEqual(get_channel_profile("SLACK").markdown, "mrkdwn")
self.assertEqual(get_channel_profile("missing").channel, "unknown")
def test_core_channels_keep_full_report_defaults(self):
self.assertEqual(CHANNEL_PROFILES["feishu"].default_mode, "full_report")
self.assertEqual(CHANNEL_PROFILES["telegram"].default_mode, "full_report")
self.assertEqual(CHANNEL_PROFILES["slack"].default_mode, "full_report")
self.assertEqual(CHANNEL_PROFILES["wechat"].default_mode, "full_report")
self.assertEqual(CHANNEL_PROFILES["email"].default_mode, "full_html")
def test_prepared_message_keeps_legacy_text_fallback(self):
prepared = PreparedMessage(
channel="feishu",
text="raw report",
formatted_text="formatted report",
fallback_text="fallback report",
)
self.assertEqual(prepared.content_for_text_send, "formatted report")
fallback = PreparedMessage(channel="custom", text="raw report", fallback_text="fallback report")
self.assertEqual(fallback.content_for_text_send, "fallback report")
def test_normalize_channel_name_handles_empty_values(self):
self.assertEqual(normalize_channel_name(None), "unknown")
self.assertEqual(normalize_channel_name(" Telegram "), "telegram")
def test_renderer_presets_are_reserved_and_disabled_by_default(self):
preset_channels = {preset.channel for preset in all_renderer_presets()}
self.assertTrue({"wechat", "feishu", "telegram", "slack", "dingtalk"}.issubset(preset_channels))
self.assertEqual(CHANNEL_RENDERER_PRESETS["feishu"].rich_renderer, "feishu_interactive_card")
self.assertEqual(CHANNEL_RENDERER_PRESETS["telegram"].markdown, "markdown_v2")
self.assertTrue(all(not preset.enabled_by_default for preset in all_renderer_presets()))
def test_get_renderer_preset_accepts_enum_or_string(self):
self.assertEqual(get_renderer_preset(NotificationChannel.WECHAT).text_renderer, "wecom_markdown")
self.assertEqual(get_renderer_preset("DINGTALK").rich_renderer, "dingtalk_action_card")
self.assertEqual(get_renderer_preset("missing").fallback_renderer, "legacy_text")
if __name__ == "__main__":
unittest.main()
tests/test_notification_report_fixtures.py
+54
View File
@@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
"""Golden-style checks for notification report formatting fixtures."""
import os
import sys
import unittest
from pathlib import Path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.formatters import (
chunk_markdown_preserving_blocks,
utf16_len,
utf8_len,
)
FIXTURE_DIR = Path(__file__).parent / "fixtures" / "notification_reports"
class NotificationReportFixtureTestCase(unittest.TestCase):
def test_all_report_fixtures_exist(self):
expected = {
"aggregate_report.md",
"single_stock_report.md",
"market_review_report.md",
}
self.assertEqual(expected, {path.name for path in FIXTURE_DIR.glob("*.md")})
def test_report_fixtures_keep_original_markdown_constructs(self):
for path in FIXTURE_DIR.glob("*.md"):
content = path.read_text(encoding="utf-8")
self.assertIn("##", content)
self.assertTrue("风险" in content or "操作" in content or "观察" in content)
if "|" in content:
self.assertRegex(content, r"\|[^\n]+\|")
def test_fixture_chunking_preserves_markdown_boundaries(self):
content = (FIXTURE_DIR / "aggregate_report.md").read_text(encoding="utf-8")
for chunks in (
chunk_markdown_preserving_blocks(content, 220),
chunk_markdown_preserving_blocks(content, 360, len_fn=utf8_len),
chunk_markdown_preserving_blocks(content, 220, len_fn=utf16_len),
):
self.assertGreater(len(chunks), 1)
for chunk in chunks:
self.assertEqual(chunk.count("```") % 2, 0)
if __name__ == "__main__":
unittest.main()
tests/test_notification_sender.py
+86
View File
@@ -208,6 +208,31 @@ class TestFeishuSender(unittest.TestCase):
"股票日报\nhello",
)
@mock.patch("src.notification_sender.feishu_sender.requests.post")
def test_send_uses_legacy_feishu_report_formatter(self, mock_post):
mock_post.return_value = _response(200, {"code": 0})
cfg = _config(feishu_webhook_url="https://feishu.example/hook")
sender = FeishuSender(cfg)
content = (
"# 日报\n\n"
"## 📊 分析结果摘要\n\n"
"| 股票 | 信号 |\n"
"| --- | --- |\n"
"| 600519 | 强势 |\n\n"
"[详情](https://example.com/report)"
)
result = sender.send_to_feishu(content)
self.assertTrue(result)
payload = mock_post.call_args.kwargs["json"]
rendered = payload["card"]["elements"][0]["text"]["content"]
self.assertIn("**日报**", rendered)
self.assertIn("**📊 分析结果摘要**", rendered)
self.assertIn("• 股票：600519 | 信号：强势", rendered)
self.assertIn("[详情](https://example.com/report)", rendered)
self.assertNotIn("| --- |", rendered)
@mock.patch("src.notification_sender.feishu_sender.requests.post")
def test_send_error_response_returns_false(self, mock_post):
mock_post.return_value = _response(200, {"code": 19024, "msg": "keyword not found"})
@@ -909,6 +934,22 @@ class TestSlackSender(unittest.TestCase):
self.assertEqual(blocks[0]["type"], "section")
self.assertEqual(blocks[0]["text"]["type"], "mrkdwn")
@mock.patch("src.notification_sender.slack_sender.requests.post")
def test_send_preserves_legacy_text_payload(self, mock_post):
resp = mock.MagicMock()
resp.status_code = 200
resp.text = "ok"
mock_post.return_value = resp
cfg = _config(slack_webhook_url="https://hooks.slack.com/services/T/B/xxx")
sender = SlackSender(cfg)
result = sender.send_to_slack("## 日报\n\n[详情](https://example.com/report)")
self.assertTrue(result)
payload = json.loads(mock_post.call_args.kwargs["data"].decode("utf-8"))
self.assertIn("## 日报", payload["text"])
self.assertIn("[详情](https://example.com/report)", payload["text"])
@mock.patch("src.notification_sender.slack_sender.requests.post")
def test_send_text_prefers_bot_when_both_configured(self, mock_post):
"""When both webhook and bot are configured, text must go via bot
@@ -995,6 +1036,51 @@ class TestTelegramSender(unittest.TestCase):
self.assertNotIn("parse_mode", second_payload)
self.assertEqual(second_payload["text"], "*ST宝实")
@mock.patch("src.notification_sender.telegram_sender.requests.post")
def test_send_plain_text_fallback_keeps_original_text_after_legacy_markdown_error(self, mock_post):
markdown_error = _response(400)
markdown_error.text = (
'{"ok":false,"error_code":400,"description":"Bad Request: can\'t parse entities"}'
)
plain_text_success = _response(200, {"ok": True})
mock_post.side_effect = [markdown_error, plain_text_success]
cfg = _config(telegram_bot_token="BOT", telegram_chat_id="CHAT")
sender = TelegramSender(cfg)
content = "关注 **AAPL** (未闭合)"
result = sender.send_to_telegram(content)
self.assertTrue(result)
first_payload = mock_post.call_args_list[0][1]["json"]
second_payload = mock_post.call_args_list[1][1]["json"]
self.assertEqual(first_payload["text"], "关注 *AAPL* \\(未闭合\\)")
self.assertEqual(second_payload["text"], content)
@mock.patch("src.notification_sender.telegram_sender.requests.post")
def test_send_uses_legacy_telegram_report_formatter(self, mock_post):
mock_post.return_value = _response(200, {"ok": True})
cfg = _config(telegram_bot_token="BOT", telegram_chat_id="CHAT")
sender = TelegramSender(cfg)
content = (
"# 日报\n\n"
"## 📊 分析结果摘要\n\n"
"| 股票 | 信号 |\n"
"| --- | --- |\n"
"| 600519 | 强势 |\n\n"
"[详情](https://example.com/report)"
)
result = sender.send_to_telegram(content)
self.assertTrue(result)
payload = mock_post.call_args.kwargs["json"]
rendered = payload["text"]
self.assertIn("日报", rendered)
self.assertIn("📊 分析结果摘要", rendered)
self.assertIn("| 股票 | 信号 |", rendered)
self.assertIn("[详情](https://example.com/report)", rendered)
self.assertNotIn("# 日报", rendered)
@mock.patch("src.notification_sender.telegram_sender.requests.post")
def test_send_plain_text_fallback_handles_non_json_200(self, mock_post):
markdown_error = _response(400)
tests/test_pipeline_notification_image_routing.py
+25 -5
View File
@@ -137,6 +137,7 @@ class _FakeWechatNotifier:
self._markdown_to_image_channels = {"wechat"}
self._markdown_to_image_max_chars = 15000
self.generate_dashboard_report = MagicMock(return_value="dashboard-report")
self.generate_wechat_dashboard = MagicMock(return_value="dashboard-report")
self.save_report_to_file = MagicMock(return_value="/tmp/report.md")
self.is_available = MagicMock(return_value=True)
self.get_available_channels = MagicMock(return_value=[NotificationChannel.WECHAT])
@@ -146,7 +147,7 @@ class _FakeWechatNotifier:
)
)
self.send_to_context = MagicMock(return_value=False)
self.generate_wechat_dashboard = MagicMock(return_value="wechat-dashboard")
self.generate_brief_report = MagicMock(return_value="brief-report")
self._should_use_image_for_channel = MagicMock(
side_effect=lambda channel, image_bytes: (
channel.value in self._markdown_to_image_channels and image_bytes is not None
@@ -157,7 +158,7 @@ class _FakeWechatNotifier:
class TestPipelineWechatOnlyImageRouting(unittest.TestCase):
def test_send_notifications_wechat_only_skips_full_report_conversion(self):
def test_send_notifications_wechat_only_converts_legacy_dashboard_for_image(self):
pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
pipeline.notifier = _FakeWechatNotifier()
pipeline.config = SimpleNamespace(stock_email_groups=[])
@@ -167,7 +168,7 @@ class TestPipelineWechatOnlyImageRouting(unittest.TestCase):
pipeline._send_notifications(results, ReportType.SIMPLE)
mock_md2img.assert_called_once_with(
"wechat-dashboard", max_chars=pipeline.notifier._markdown_to_image_max_chars
"dashboard-report", max_chars=pipeline.notifier._markdown_to_image_max_chars
)
pipeline.notifier._send_wechat_image.assert_called_once()
pipeline.notifier.send_to_wechat.assert_not_called()
@@ -184,7 +185,7 @@ class TestPipelineWechatOnlyImageRouting(unittest.TestCase):
pipeline._send_notifications(results, ReportType.SIMPLE)
pipeline.notifier._send_wechat_image.assert_not_called()
pipeline.notifier.send_to_wechat.assert_called_once_with("wechat-dashboard")
pipeline.notifier.send_to_wechat.assert_called_once_with("dashboard-report")
self.assertTrue(
any("企业微信 Markdown 转图片失败" in str(call.args[0]) for call in mock_warning.call_args_list)
)
@@ -195,6 +196,7 @@ class _FakeRoutedNotifier:
self._markdown_to_image_channels = set(image_channels or [])
self._markdown_to_image_max_chars = 15000
self.generate_dashboard_report = MagicMock(side_effect=self._generate_dashboard_report)
self.generate_wechat_dashboard = MagicMock(side_effect=self._generate_dashboard_report)
self.save_report_to_file = MagicMock(return_value="/tmp/report.md")
self.is_available = MagicMock(return_value=True)
self.get_available_channels = MagicMock(
@@ -221,7 +223,7 @@ class _FakeRoutedNotifier:
channel.value in self._markdown_to_image_channels and image_bytes is not None
)
)
self.generate_wechat_dashboard = MagicMock(return_value="wechat-dashboard")
self.generate_brief_report = MagicMock(return_value="brief-report")
self._send_wechat_image = MagicMock(return_value=True)
self.send_to_wechat = MagicMock(return_value=True)
self._send_telegram_photo = MagicMock(return_value=True)
@@ -280,6 +282,24 @@ class TestPipelineReportRouteFiltering(unittest.TestCase):
pipeline.notifier.send_to_email.assert_called_once_with("report:000001")
pipeline.notifier.send_to_telegram.assert_not_called()
def test_telegram_image_route_converts_full_report(self):
pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
pipeline.notifier = _FakeRoutedNotifier(
[NotificationChannel.TELEGRAM],
image_channels={"telegram"},
)
pipeline.config = SimpleNamespace(stock_email_groups=[])
results = [SimpleNamespace(code="000001")]
with patch("src.md2img.markdown_to_image", return_value=b"png") as mock_md2img:
pipeline._send_notifications(results, ReportType.SIMPLE)
mock_md2img.assert_called_once_with(
"report:000001", max_chars=pipeline.notifier._markdown_to_image_max_chars
)
pipeline.notifier._send_telegram_photo.assert_called_once_with(b"png")
pipeline.notifier.send_to_telegram.assert_not_called()
def test_ntfy_route_uses_text_report_without_image_conversion(self):
pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
pipeline.notifier = _FakeRoutedNotifier(