fix(review-feedback-1265): address latest review comments

Author: ZhuLinsen

data_provider/akshare_fetcher.py

@@ -1961,8 +1961,9 @@ def get_limit_up_pool(
                 if col in df.columns:
                     df[col] = pd.to_numeric(df[col], errors='coerce')
             if '首次封板时间' in df.columns:
-                df['首次封板时间'] = df['首次封板时间'].astype(str)
-            sort_cols = [col for col in ('连板数', '首次封板时间') if col in df.columns]
+                df['首次封板时间'] = df['首次封板时间'].map(self._normalize_limit_time_value)
+                df['_首次封板时间排序'] = df['首次封板时间'].where(df['首次封板时间'] != '', '999999')
+            sort_cols = [col for col in ('连板数', '_首次封板时间排序') if col in df.columns]
             if sort_cols:
                 ascending = [False if col == '连板数' else True for col in sort_cols]
                 df = df.sort_values(sort_cols, ascending=ascending)
@@ -1989,6 +1990,35 @@ def get_limit_up_pool(
             logger.warning(f"[Akshare] 获取涨停池失败: {e}")
             return None
 
+    @staticmethod
+    def _normalize_limit_time_value(value: Any) -> str:
+        """Normalize AkShare HHMMSS-like seal time values to zero-padded HHMMSS."""
+        try:
+            if pd.isna(value):
+                return ""
+        except TypeError:
+            pass
+
+        text = str(value).strip()
+        if not text or text.lower() in {"nan", "nat", "none", "null", "-", "--"}:
+            return ""
+
+        if ":" in text:
+            parts = text.split(":")
+            try:
+                hour = int(parts[0])
+                minute = int(parts[1]) if len(parts) > 1 else 0
+                second = int(parts[2]) if len(parts) > 2 else 0
+                return f"{hour:02d}{minute:02d}{second:02d}"
+            except (TypeError, ValueError):
+                return text
+
+        try:
+            return f"{int(float(text)):06d}"
+        except (TypeError, ValueError):
+            digits = "".join(ch for ch in text if ch.isdigit())
+            return digits.zfill(6) if digits else text
+
     @staticmethod
     def _safe_float(value: Any) -> Optional[float]:
         try:

src/formatters.py

@@ -265,6 +265,43 @@ def _bytes(s: str) -> int:
     return len(s.encode('utf-8'))
 
 
+def _is_escaped_pipe(value: str, index: int) -> bool:
+    """Return whether value[index] is escaped by an odd number of backslashes."""
+    slash_count = 0
+    pos = index - 1
+    while pos >= 0 and value[pos] == "\\":
+        slash_count += 1
+        pos -= 1
+    return slash_count % 2 == 1
+
+
+def _parse_markdown_table_row(row: str) -> List[str]:
+    """Split a Markdown table row while preserving escaped literal pipes."""
+    value = row.strip()
+    if value.startswith("|"):
+        value = value[1:]
+    if value.endswith("|") and not _is_escaped_pipe(value, len(value) - 1):
+        value = value[:-1]
+
+    cells: List[str] = []
+    current: List[str] = []
+    index = 0
+    while index < len(value):
+        char = value[index]
+        if char == "\\" and index + 1 < len(value) and value[index + 1] == "|":
+            current.append("|")
+            index += 2
+            continue
+        if char == "|":
+            cells.append("".join(current).strip())
+            current = []
+        else:
+            current.append(char)
+        index += 1
+    cells.append("".join(current).strip())
+    return cells
+
+
 def _chunk_by_max_bytes(content: str, max_bytes: int) -> List[str]:
     if _bytes(content) <= max_bytes:
         return [content]
@@ -421,7 +458,7 @@ def format_feishu_markdown(content: str) -> str:
         >>> print(formatted)
         **标题**
         > 引用
-        ```text
+        ```
         列1  列2
         值1  值2
         ```
@@ -458,8 +495,7 @@ def _flush_table_rows(buffer: List[str], output: List[str]) -> None:
 
         def _parse_row(row: str) -> List[str]:
             """解析表格行，提取单元格"""
-            cells = [c.strip() for c in row.strip().strip('|').split('|')]
-            return cells
+            return _parse_markdown_table_row(row)
 
         rows = []
         for raw in buffer:
@@ -545,7 +581,7 @@ def _is_table_separator(row: str) -> bool:
         return bool(re.match(r'^\s*\|?\s*:?-{3,}:?\s*(\|\s*:?-{3,}:?\s*)+\|?\s*$', row))
 
     def _parse_table_row(row: str) -> List[str]:
-        return [cell.strip() for cell in row.strip().strip("|").split("|")]
+        return _parse_markdown_table_row(row)
 
     def _text(content_value: str, tag: str = "lark_md") -> Dict[str, str]:
         return {"tag": tag, "content": content_value}

src/notification_sender/feishu_sender.py

@@ -190,11 +190,11 @@ def _send_feishu_message(self, content: str, *, timeout_seconds: Optional[float]
         prepared_content = self._apply_keyword_prefix(content)
         security_fields = self._build_security_fields()
 
-        def _post_payload(payload: Dict[str, Any]) -> bool:
+        def _post_payload(payload: Dict[str, Any], payload_text: str = prepared_content) -> bool:
             request_payload = dict(payload)
             request_payload.update(security_fields)
             logger.debug(f"飞书请求 URL: {self._feishu_url}")
-            logger.debug(f"飞书请求 payload 长度: {len(prepared_content)} 字符")
+            logger.debug(f"飞书请求 payload 长度: {len(payload_text)} 字符")
 
             response = requests.post(
                 self._feishu_url,
@@ -223,6 +223,63 @@ def _post_payload(payload: Dict[str, Any]) -> bool:
                 logger.error(f"响应内容: {response.text}")
                 return False
 
+        def _post_text_payload(text_content: str) -> bool:
+            prepared_text = self._apply_keyword_prefix(text_content)
+            text_payload = {
+                "msg_type": "text",
+                "content": {
+                    "text": prepared_text
+                }
+            }
+            return _post_payload(text_payload, prepared_text)
+
+        def _send_fallback_text(text_content: str) -> bool:
+            max_bytes = self._feishu_max_bytes
+            keyword_overhead = len(self._get_keyword_prefix().encode('utf-8'))
+            effective_max_bytes = max_bytes - keyword_overhead
+
+            if effective_max_bytes <= 0:
+                logger.error("飞书关键词过长，超过单条消息允许的最大字节数，无法发送")
+                return False
+
+            text_bytes = len(text_content.encode('utf-8')) + keyword_overhead
+            if text_bytes <= max_bytes:
+                return _post_text_payload(text_content)
+
+            min_chunk_bytes = MIN_MAX_BYTES + PAGE_MARKER_SAFE_BYTES
+            if effective_max_bytes < min_chunk_bytes:
+                logger.error(
+                    "飞书回退文本分片预算(%s字节)不足以安全分页发送，至少需要 %s 字节",
+                    effective_max_bytes,
+                    min_chunk_bytes,
+                )
+                return False
+
+            try:
+                chunks = chunk_content_by_max_bytes(
+                    text_content,
+                    effective_max_bytes,
+                    add_page_marker=True,
+                )
+            except ValueError as e:
+                logger.error("飞书回退文本分片失败: %s", e)
+                return False
+
+            success_count = 0
+            total_chunks = len(chunks)
+            logger.info("飞书回退文本超长(%s字节/%s字符)，将分批发送：共 %s 批", text_bytes, len(text_content), total_chunks)
+            for i, chunk in enumerate(chunks):
+                if _post_text_payload(chunk):
+                    success_count += 1
+                    logger.info("飞书回退文本第 %s/%s 批发送成功", i + 1, total_chunks)
+                else:
+                    logger.error("飞书回退文本第 %s/%s 批发送失败", i + 1, total_chunks)
+
+                if i < total_chunks - 1:
+                    time.sleep(1)
+
+            return success_count == total_chunks
+
         # 1) 优先使用结构化交互卡片。飞书的 lark_md 不是完整 Markdown，
         # 表格/代码块/引用需要拆成卡片元素才会真正渲染。
         card_elements = build_feishu_card_elements(prepared_content)
@@ -243,12 +300,6 @@ def _post_payload(payload: Dict[str, Any]) -> bool:
         if _post_payload(card_payload):
             return True
 
-        # 2) 回退为普通文本消息
-        text_payload = {
-            "msg_type": "text",
-            "content": {
-                "text": format_feishu_markdown(prepared_content)
-            }
-        }
-
-        return _post_payload(text_payload)
+        # 2) 回退为普通文本消息。回退格式可能比原始 Markdown 更长，
+        # 发送前按真实回退文本重新校验并必要时分片。
+        return _send_fallback_text(format_feishu_markdown(content))

tests/test_akshare_realtime_logging.py

@@ -1,5 +1,8 @@
 import logging
+import sys
+from types import SimpleNamespace
 
+import pandas as pd
 import pytest
 import requests
 
@@ -150,7 +153,7 @@ def test_tencent_realtime_success_logs_endpoint(caplog, monkeypatch, akshare_fet
     assert "[实时行情-腾讯] 601006 大秦铁路:" in caplog.text
 
 
-def test_hot_stocks_falls_back_to_eastmoney(monkeypatch, akshare_fetcher):
+def test_hot_stocks_uses_eastmoney_hot_ranking_when_available(monkeypatch, akshare_fetcher):
     monkeypatch.setattr(
         akshare_fetcher,
         "_get_eastmoney_hot_stocks",
@@ -170,3 +173,62 @@ def test_hot_stocks_falls_back_to_eastmoney(monkeypatch, akshare_fetcher):
 
     assert result[0]["source"] == "东方财富人气榜"
     assert result[0]["name"] == "中国长城"
+
+
+def test_limit_up_pool_zero_pads_first_seal_times_before_sorting(monkeypatch, akshare_fetcher):
+    df = pd.DataFrame(
+        [
+            {
+                "代码": "000002",
+                "名称": "午后股",
+                "涨跌幅": 10.0,
+                "最新价": 12.3,
+                "成交额": 1,
+                "换手率": 2,
+                "封板资金": 3,
+                "首次封板时间": 141354,
+                "最后封板时间": 141500,
+                "炸板次数": 0,
+                "涨停统计": "1/1",
+                "连板数": 1,
+                "所属行业": "地产",
+            },
+            {
+                "代码": "000001",
+                "名称": "竞价股",
+                "涨跌幅": 10.0,
+                "最新价": 10.0,
+                "成交额": 1,
+                "换手率": 2,
+                "封板资金": 3,
+                "首次封板时间": 92500,
+                "最后封板时间": 93000,
+                "炸板次数": 0,
+                "涨停统计": "1/1",
+                "连板数": 1,
+                "所属行业": "计算机",
+            },
+            {
+                "代码": "000003",
+                "名称": "早盘股",
+                "涨跌幅": 10.0,
+                "最新价": 11.0,
+                "成交额": 1,
+                "换手率": 2,
+                "封板资金": 3,
+                "首次封板时间": 101500,
+                "最后封板时间": 102000,
+                "炸板次数": 0,
+                "涨停统计": "1/1",
+                "连板数": 1,
+                "所属行业": "电子",
+            },
+        ]
+    )
+    fake_akshare = SimpleNamespace(stock_zt_pool_em=lambda date: df)
+    monkeypatch.setitem(sys.modules, "akshare", fake_akshare)
+
+    result = akshare_fetcher.get_limit_up_pool(date="20260511", n=3)
+
+    assert [row["code"] for row in result] == ["000001", "000003", "000002"]
+    assert result[0]["first_limit_time"] == "092500"

tests/test_formatters.py

@@ -220,3 +220,16 @@ def test_card_elements_render_tables_as_columns(self):
         self.assertIn('"tag": "note"', serialized)
         self.assertNotIn("```", serialized)
         self.assertNotIn("| 指标 |", serialized)
+
+    def test_card_table_preserves_escaped_literal_pipes(self):
+        content = """| 类型 | 标题 | 备注 |
+| --- | --- | --- |
+| 新闻 | AI \\| chips | 后续 |
+"""
+        elements = build_feishu_card_elements(content)
+        column_set = next(element for element in elements if element["tag"] == "column_set")
+        serialized = json.dumps(elements, ensure_ascii=False)
+
+        self.assertEqual(len(column_set["columns"]), 3)
+        self.assertIn("AI | chips", serialized)
+        self.assertIn("后续", serialized)

tests/test_notification_sender.py

@@ -236,6 +236,28 @@ def test_send_error_response_returns_false(self, mock_post):
         self.assertFalse(result)
         self.assertEqual(mock_post.call_count, 2)
 
+    @mock.patch("src.notification_sender.feishu_sender.time.sleep", return_value=None)
+    @mock.patch("src.notification_sender.feishu_sender.format_feishu_markdown", return_value="F" * 180)
+    @mock.patch("src.notification_sender.feishu_sender.requests.post")
+    def test_card_failure_chunks_oversized_fallback_text(self, mock_post, mock_format, _mock_sleep):
+        mock_post.side_effect = (
+            [_response(200, {"code": 19024, "msg": "card failed"})]
+            + [_response(200, {"code": 0}) for _ in range(10)]
+        )
+        cfg = _config(feishu_webhook_url="https://feishu.example/hook", feishu_max_bytes=100)
+        sender = FeishuSender(cfg)
+
+        result = sender.send_to_feishu("short")
+
+        self.assertTrue(result)
+        mock_format.assert_called_once_with("short")
+        self.assertGreater(mock_post.call_count, 2)
+        self.assertEqual(mock_post.call_args_list[0].kwargs["json"]["msg_type"], "interactive")
+        for call in mock_post.call_args_list[1:]:
+            payload = call.kwargs["json"]
+            self.assertEqual(payload["msg_type"], "text")
+            self.assertLessEqual(len(payload["content"]["text"].encode("utf-8")), 100)
+
     @mock.patch("src.notification_sender.feishu_sender.requests.post")
     def test_send_with_keyword_that_leaves_too_little_chunk_budget_returns_false(self, mock_post):
         cfg = _config(