第 7 章 · 04 触达链路细节：代码转换、名称查询与容错

第 7 章 · 04 触达链路细节：代码转换、名称查询与容错

本节摘要：本节讲飞书触达链路的「最后三件小事」——把前面三节的内容串成一条「从选股结果到飞书群」的完整链路。三个细节看似琐碎，每件都关系用户体验：① 雪球代码转换——A 股 6 开头是上海、4/8 开头是北京、其余是深圳，错一个前缀链接就点不进去；② 股票名称查询——选股结果是「6 位代码」对人不友好，用 baostock 二次查询出「贵州茅台」这样的中文名，飞书消息才看着舒服；③ HTTP 失败容错——飞书服务偶发不可达，仅记 ERROR 日志，不抛异常、不影响下个策略。本节用「链路图」把这三件事拼起来，再用一段可运行的代码完整复现 Sequoia-X 的飞书发送逻辑（Mock 飞书响应）。

内容来源：原项目飞书通知模块精读。

💡 核心心法：好的「用户感知层」的代码，不只做「能跑」的事，还做「让用户感觉舒服」的细节——名字而不是代码、可点击而不是裸 URL、失败不打扰。

学习目标

阅读完本节，你应当能够：

1. 讲清 A 股代码到雪球代码的转换规则（6/4/8/9 开头）。
2. 理解股票名称查询的工程价值与实现路径。
3. 读懂飞书 HTTP 失败时仅记日志不抛异常的容错设计。
4. 写一段可运行的完整飞书发送 Mock 代码。

一、链路全景图

把 Sequoia-X 触达层的完整流程画成「从选股结果到飞书群」的链路：

5 个步骤 + 1 个错误兜底——整条链路就这么简单。

二、步骤 1：雪球代码转换

雪球是中国主流的股票社区，飞书卡片里的链接直接用雪球代码拼接：

https://xueqiu.com/S/<雪球代码>
​

A 股 6 位代码到雪球代码的转换规则：

Sequoia-X 的实现：

@staticmethod
def _to_xueqiu_code(code: str) -> str:
    """6/9 开头 → SH；4/8 开头 → BJ；其余 → SZ。"""
    if code.startswith("6"):
        return f"SH{code}"
    elif code.startswith(("4", "8")):
        return f"BJ{code}"
    return f"SZ{code}"
​

注意：Sequoia-X 把 9 开头也算到 SH——上海 B 股（沪 B）虽然代码以 9 开头但仍属上交所。

三、步骤 2：股票名称查询

为什么需要「代码 → 名称」的查询？

「代码 vs 名称」是用户体验的核心分水岭。

Sequoia-X 用 baostock 查名称：

@staticmethod
def _get_stock_names(symbols: list[str]) -> dict[str, str]:
    """通过 baostock 批量查询股票名称。"""
    import baostock as bs
    bs.login()
    mapping = {}
    for code in symbols:
        prefix = "sh" if code.startswith(("6", "9")) else "sz"
        rs = bs.query_stock_basic(code=f"{prefix}.{code}")
        while rs.next():
            row = rs.get_row_data()
            mapping[code] = row[1]    # 第 2 个字段是股票名称
    bs.logout()
    return mapping
​

注意细节：

• 每个代码单独查询（baostock 接口不支持批量）——简单但慢；
• 用 bs.login() / bs.logout() 包住——避免长连接；
• 返回 dict[str, str]（代码 → 名称）——O(1) 查询。

一个潜在问题：如果某只股票当天停牌或新上市，baostock 可能查不到——mapping.get(code, xq_code) 兜底用「雪球代码」作为名称。

这是一个**「用户能容忍」**的兜底——总比「没消息」好。

四、步骤 3 + 4：构造卡片

把代码转链接、嵌入到卡片 JSON 里。前面第 02 节已经讲过卡片结构，这里只补充链接生成的细节：

links: list[str] = []
for code in symbols:
    xq_code = self._to_xueqiu_code(code)
    name = names.get(code, xq_code)    # 优先用中文名，缺则用雪球代码
    links.append(f"[{name}](https://xueqiu.com/S/{xq_code})")

symbol_text = " ".join(links)
​

关键：

• name = names.get(code, xq_code)：缺名称时用雪球代码——保证总有名字。
• " ".join(links)：用空格分隔多个链接——飞书会渲染为可点击列表。

五、步骤 5：HTTP 发送 + 容错

Sequoia-X 用 requests.post 发送，重点是错误处理：

def send(self, symbols, strategy_name, webhook_key="default") -> None:
    url = self.settings.get_webhook_url(webhook_key)
    payload = self._build_card(symbols, strategy_name)
    
    try:
        resp = requests.post(
            url,
            data=json.dumps(payload),
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
        resp_json = resp.json()
        
        # 飞书真正的成功标志是内部的 code == 0（**不是 HTTP 200**）
        if resp.status_code != 200 or resp_json.get("code") != 0:
            logger.error(f"飞书推送失败 [{webhook_key}] ...")
        else:
            logger.info(f"飞书推送成功 [{webhook_key}]，共 {len(symbols)} 只")
    
    except requests.RequestException as exc:
        # 网络异常：仅记日志，**不抛异常**——不影响主流程
        logger.error(f"飞书推送请求异常 [{webhook_key}]：{exc}")
​

三个关键点：

5.1 飞书的「双层成功标志」

if resp.status_code != 200 or resp_json.get("code") != 0:
    # 失败
​

• HTTP 200 只代表「请求到达飞书」；
• 飞书返回的 code == 0 才代表「飞书真的把消息发出去了」。

两者必须同时为真才算成功。这是飞书 webhook 的双层成功判定——很多新手栽在这里。

5.2 timeout=10

不设 timeout 可能永远卡住——网络偶发半开状态会让程序挂起。

10 秒是**「足够慢但不永久」**的经验值。

5.3 失败仅记日志，不抛异常

except requests.RequestException as exc:
    logger.error(...)         # 只记日志
    # 不 raise，不 return False——调用方继续执行
​

为什么不抛异常？

• 飞书推送不是核心功能（选股结果才是）；
• 某次推送失败不应该影响后续策略推送；
• 错误信息已经进了日志系统——运维能排查。

「失败隔离」是 Sequoia-X 整套系统的核心哲学——任何一层挂都不影响整体。

六、可运行教学代码（Mock 飞书）

下面这段代码完整演示「从选股结果到飞书发送」的完整链路。它Mock 飞书的 HTTP 响应——不实际发送，只演示：

# feishu_send_demo.py  # 可运行：完整飞书发送链路（Mock）
import json
import logging
from unittest.mock import patch, MagicMock


def to_xueqiu_code(code: str) -> str:
    if code.startswith("6"):
        return f"SH{code}"
    elif code.startswith(("4", "8")):
        return f"BJ{code}"
    return f"SZ{code}"


def build_card(symbols: list[str], names: dict[str, str], strategy_name: str) -> dict:
    today = "2024-06-15"
    links = [f"[{names.get(s, to_xueqiu_code(s))}](https://xueqiu.com/S/{to_xueqiu_code(s)})" for s in symbols]
    return {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {"tag": "plain_text", "content": f"📈 Sequoia-X 选股播报 | {strategy_name}"},
                "template": "blue",
            },
            "elements": [
                {"tag": "div", "text": {"tag": "lark_md", "content": f"**日期：** {today}\n**策略：** {strategy_name}\n**选股数量：** {len(symbols)}"}},
                {"tag": "hr"},
                {"tag": "div", "text": {"tag": "lark_md", "content": "**选股列表：**\n" + " ".join(links)}},
            ],
        },
    }


def mock_send(payload: dict, url: str, scenario: str = "success") -> dict:
    """Mock 飞书响应，模拟不同场景。"""
    if scenario == "success":
        return {"status_code": 200, "code": 0, "msg": "success"}
    elif scenario == "feishu_reject":
        return {"status_code": 200, "code": 99991663, "msg": "敏感词拦截"}
    elif scenario == "http_error":
        return {"status_code": 500, "code": -1, "msg": "server error"}
    return {}


def send_with_logging(url: str, payload: dict, scenario: str) -> str:
    """完整发送流程：构造 → POST → 解析响应 → 记日志 → 不抛异常。"""
    logging.basicConfig(level=logging.INFO, format="%(name)s - %(message)s")
    log = logging.getLogger("feishu")
    
    resp = mock_send(payload, url, scenario)
    
    if resp.get("status_code") == 200 and resp.get("code") == 0:
        log.info(f"推送成功：{len(payload['card']['elements'])} 个块")
        return "success"
    else:
        log.error(f"推送失败 [{scenario}] status={resp.get('status_code')} code={resp.get('code')}")
        return "failed"     # 不抛异常，调用方继续


if __name__ == "__main__":
    symbols = ["600519", "000001"]
    names = {"600519": "贵州茅台", "000001": "平安银行"}
    card = build_card(symbols, names, "海龟突破")
    
    print("=" * 60)
    print("演示 1：成功推送")
    print("=" * 60)
    print(send_with_logging("https://feishu/hook/turtle", card, "success"))
    print()
    
    print("=" * 60)
    print("演示 2：飞书敏感词拦截（HTTP 200 但 code != 0）")
    print("=" * 60)
    print(send_with_logging("https://feishu/hook/turtle", card, "feishu_reject"))
    print()
    
    print("=" * 60)
    print("演示 3：HTTP 500 错误")
    print("=" * 60)
    print(send_with_logging("https://feishu/hook/turtle", card, "http_error"))
​

预期输出：

============================================================
演示 1：成功推送
============================================================
feishu - 推送成功：3 个块
success

============================================================
演示 2：飞书敏感词拦截（HTTP 200 但 code != 0）
============================================================
feishu - 推送失败 [feishu_reject] status=200 code=99991663
failed

============================================================
演示 3：HTTP 500 错误
============================================================
feishu - 推送失败 [http_error] status=500 code=-1
failed
​

注意：所有失败都「仅记日志、不抛异常」——主流程继续。

七、双层成功标志的实操经验

飞书 webhook 的双层成功标志是新手最容易踩的坑：

飞书常见错误码：

八、触达链路的工程哲学

Sequoia-X 的触达链路设计体现了三条工程哲学：

「用户能感知」的代码（推送、卡片、日志）值得花时间打磨——这是用户对你的第一印象。

九、本节在全景图的位置

触达链路细节 ✅ ← 本节
        │
        ▼
下一章（第 8 章 主流程编排与运行模式）：双模式分发 + 策略注册与执行
​

本节要点回顾

1. 完整链路 5 步：路由 → 名称查询 → 代码转雪球 → 构造卡片 → POST 发送 + 容错。
2. 雪球代码转换：6/9 → SH，4/8 → BJ，其余 → SZ——前缀决定交易所。
3. 股票名称查询：baostock 二次查询，缺则用雪球代码兜底——保证总有名字。
4. 双层成功标志：HTTP 200 + 飞书 code == 0——两者缺一不可。
5. 失败仅记日志：except 后只记日志，不抛异常——失败隔离的核心。
6. timeout=10：避免网络半开状态永久挂起。
7. 可运行示例：3 种场景（成功/敏感词/HTTP 500），亲眼看容错行为。

配套教学脚本：images/feishu_send_demo.py——Mock 飞书响应，完整链路演示。

至此，第 7 章事件驱动策略与消息推送四节全部完成。下一章，我们把所有零件串起来——讲主流程编排与双运行模式。