feat: add Feishu App Bot notification sender with P2P and group support (#1553)

* feat: add Feishu App Bot notification sender with P2P and group support

The existing FeishuSender only supports custom robot Webhook mode.
This commit extends it to support App Bot (lark-oapi SDK) mode, auto-routing
between webhook (priority) and App Bot when FEISHU_APP_ID + FEISHU_APP_SECRET
+ FEISHU_CHAT_ID are configured.

Design:
- send_to_feishu() routes: webhook if URL set, else App Bot
- DCLP lazy client init with thread-safe sentinel guard
- Retry (3 attempts, exponential backoff) with fixed UUID for idempotency
- Card-first / text-fallback content strategy
- Chunking for long messages
- Runtime enum validation for FEISHU_RECEIVE_ID_TYPE and FEISHU_DOMAIN
- Safe SDK defaults (FEISHU_DOMAIN/LARK_DOMAIN) before import try-block
  so Webhook path never depends on lark-oapi SDK presence
- Config, diagnostics, setup check, notification test, and CI workflow
  all consistent with the new App Bot channel semantics
- lark-oapi>=1.0.0 already in requirements.txt (line 23)

Verification:
- 20/20 unit tests pass (help metadata + FeishuSender)
- E2E: real Feishu API — SDK import, token, client init, P2P text+card send all PASS
- Webhook regression: verified no SDK dependency for existing Webhook path

* fix: add missing Feishu App Bot locale entries and env table keys, harden sender error handling

CI fix 1 (test_registry_help_keys_exist_in_locales):
- Add zh-CN and en-US locale entries for FEISHU_CHAT_ID,
  FEISHU_RECEIVE_ID_TYPE, FEISHU_DOMAIN in settingsHelp.ts

CI fix 2 (test_notification_actions_env_table_matches_generated_output):
- Add FEISHU_RECEIVE_ID_TYPE, FEISHU_DOMAIN to feishu advanced_keys
  in CHANNEL_SPECS so they appear in KEY_SPECS
- Regenerate managed env table in docs/notifications.md

feishu_sender.py hardening:
- Catch network exceptions in webhook _post_payload so card-to-text
  fallback actually executes on transient failures
- Guard response.json() and isinstance(result, dict) against
  non-JSON / non-dict HTTP 200 responses
- Extract shared _build_card_body() to de-duplicate card payload
  construction between webhook and App Bot paths
- Rename module-level lark -> _lark to avoid shadowing
- Guard resp.get_log_id() with try/except
- Add None guard on send_to_feishu content parameter

e2e script improvements:
- Support FEISHU_OPEN_ID for P2P test, FEISHU_DOMAIN for Lark
- Add FEISHU_TEST_SEND_TEXT=1 for plain-text-only path testing
- Clarify docstring: setup validation + smoke test, not full e2e

* fix: consolidate Feishu App Bot notification contract

* fix: align Feishu domain help scope

---------

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>

.env.example

@@ -424,7 +424,9 @@ AGENT_SKILLS=
 #
 # WECHAT_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key_here
 #
-# 【方式二】飞书机器人
+# 【方式二】飞书机器人（二选一）
+#
+# 方式 2a — 群自定义机器人 Webhook
 # 在飞书群 -> 设置 -> 群机器人 -> 添加机器人 -> 自定义机器人 -> 复制 Webhook 地址
 #
 # FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/your_key_here
@@ -432,7 +434,22 @@ AGENT_SKILLS=
 # FEISHU_WEBHOOK_SECRET=your_feishu_webhook_secret
 # 如果机器人安全设置开启了“关键词”，需填写同一个关键词；系统会自动在每条消息前补上
 # FEISHU_WEBHOOK_KEYWORD=股票日报
-# 注意：FEISHU_APP_ID / FEISHU_APP_SECRET 用于飞书应用、云文档或 Stream 模式，不会直接启用群 Webhook 推送
+#
+# 方式 2b — 飞书应用机器人（App Bot）推送
+# 在飞书开放平台创建应用，开启 im:message 权限并发布。
+#
+# 群聊模式（推荐）：将机器人拉入目标群，从群设置中获取 chat_id
+# FEISHU_APP_ID=cli_xxxxxxxxxxxxx
+# FEISHU_APP_SECRET=your_app_secret
+# FEISHU_CHAT_ID=oc_xxxxxxxxxxxxx
+# FEISHU_RECEIVE_ID_TYPE=chat_id
+#
+# 私聊模式：直接给指定用户发私信。用户的 open_id 从机器人事件或联系人 API 获取。
+# FEISHU_APP_ID=cli_xxxxxxxxxxxxx
+# FEISHU_APP_SECRET=your_app_secret
+# FEISHU_CHAT_ID=ou_xxxxxxxxxxxxx
+# FEISHU_RECEIVE_ID_TYPE=open_id
+# FEISHU_DOMAIN=feishu  # 域名: feishu(飞书国内) / lark(国际版 larksuite.com)
 #
 # 【方式三】Telegram 机器人（需同时配置两项）
 # 1. 在 Telegram 找 @BotFather -> /newbot -> 获取 Bot Token
@@ -607,9 +624,12 @@ DINGTALK_APP_SECRET=xxxx
 # 启用 Stream 模式
 DINGTALK_STREAM_ENABLED=false
 
-# 飞书应用机器人配置（用于 Stream Bot / 云文档模式，不会直接开启群 Webhook 推送）
+# 飞书应用配置（用于 App Bot 主动推送 / Stream Bot / 云文档；不会直接开启群 Webhook 推送）
 FEISHU_APP_ID=xxxx
-FEISHU_APP_SECRET=xxxx  # 飞书应用 Secret，仅应用/Stream Bot/云文档模式使用；群通知推送请使用 FEISHU_WEBHOOK_URL
+FEISHU_APP_SECRET=xxxx  # App Bot 主动推送还需 FEISHU_CHAT_ID；简单群推送优先使用 FEISHU_WEBHOOK_URL
+# App Bot 主动推送目标；Stream Bot 或云文档不需要此项
+# FEISHU_CHAT_ID=oc_xxxxxxxxxxxxx
+# FEISHU_RECEIVE_ID_TYPE=chat_id
 # 启用长连接模式
 FEISHU_STREAM_ENABLED=false
 # 飞书群机器人 Webhook 安全配置（仅 Webhook 推送模式使用）

.github/workflows/00-daily-analysis.yml

@@ -317,6 +317,9 @@ jobs:
           FEISHU_APP_ID: ${{ secrets.FEISHU_APP_ID }}
           FEISHU_APP_SECRET: ${{ secrets.FEISHU_APP_SECRET }}
           FEISHU_FOLDER_TOKEN: ${{ secrets.FEISHU_FOLDER_TOKEN }}
+          FEISHU_CHAT_ID: ${{ vars.FEISHU_CHAT_ID || secrets.FEISHU_CHAT_ID }}
+          FEISHU_RECEIVE_ID_TYPE: ${{ vars.FEISHU_RECEIVE_ID_TYPE || secrets.FEISHU_RECEIVE_ID_TYPE }}
+          FEISHU_DOMAIN: ${{ vars.FEISHU_DOMAIN || secrets.FEISHU_DOMAIN }}
 
           # 方式十：AstrBot
           ASTRBOT_URL: ${{ secrets.ASTRBOT_URL }}
@@ -443,7 +446,7 @@ jobs:
           echo "  ntfy: $([ -n "$NTFY_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
           echo "  Gotify: $([ -n "$GOTIFY_URL" ] && [ -n "$GOTIFY_TOKEN" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
           echo "  企业微信: $([ -n "$WECHAT_WEBHOOK_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
-          echo "  飞书: $([ -n "$FEISHU_WEBHOOK_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
+          echo "  飞书: $( ([ -n "$FEISHU_WEBHOOK_URL" ] || ([ -n "$FEISHU_APP_ID" ] && [ -n "$FEISHU_APP_SECRET" ] && [ -n "$FEISHU_CHAT_ID" ]) ) && echo '✅ 已配置' || echo '⚪ 未配置')"
           echo "  Telegram: $([ -n "$TELEGRAM_BOT_TOKEN" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
           echo "  Discord: $([ -n "$DISCORD_WEBHOOK_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
           echo "  AstrBot: $([ -n "$ASTRBOT_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"

apps/dsa-web/src/locales/settingsHelp.ts

@@ -294,6 +294,45 @@ const settingsHelpZhCN: SettingsHelpMap = {
       '失败只应影响飞书应用机器人链路，不应拖垮主分析流程。',
     ],
   },
+  'settings.notification.FEISHU_CHAT_ID': {
+    title: '飞书 App Bot 推送目标',
+    summary: '配置飞书应用机器人主动推送的目标 chat_id（群聊模式）或 open_id（私聊模式）。',
+    usage: '需要同时填写 FEISHU_APP_ID 和 FEISHU_APP_SECRET。群聊模式填写 oc_ 开头的 chat_id；私聊模式填写 ou_ 开头的 open_id 并将 FEISHU_RECEIVE_ID_TYPE 设为 open_id。',
+    valueNotes: [
+      '仅凭 FEISHU_APP_ID / FEISHU_APP_SECRET 不会自动启用群 Webhook 推送。',
+      'App Bot 模式与 Webhook 模式互斥：Webhook URL 优先，未配置 Webhook 时才走 App Bot。',
+    ],
+    impact: [
+      '影响飞书 App Bot 通知渠道的送达目标。',
+      '失败时不应拖垮主分析流程，只影响该渠道送达。',
+    ],
+    notes: [
+      'App Bot 需要应用拥有 im:message:send_as_bot 权限。',
+      '私聊需要用户在飞书端主动打开过与应用机器人的对话框。',
+    ],
+  },
+  'settings.notification.FEISHU_RECEIVE_ID_TYPE': {
+    title: '飞书接收方 ID 类型',
+    summary: '指定 FEISHU_CHAT_ID 的类型：chat_id 表示群聊，open_id 表示私聊。',
+    usage: '群聊选择 chat_id；私聊（给指定用户发 P2P 消息）选择 open_id。',
+    valueNotes: [
+      '仅当 FEISHU_CHAT_ID 已填写时生效。',
+      '填错类型会导致消息发送失败；如果收到 invalid receive_id 错误，需要确认该值与前端的实际 ID 类型一致。',
+    ],
+    impact: ['影响飞书 App Bot 消息的路由方式。'],
+    notes: ['大多数场景使用 chat_id 即可；如果值不是 chat_id 或 open_id，运行时会自动回退到 chat_id。'],
+  },
+  'settings.notification.FEISHU_DOMAIN': {
+    title: '飞书 API 域名',
+    summary: '选择飞书 API 的区域：feishu 对应飞书国内版（feishu.cn），lark 对应 Lark 国际版（larksuite.com）。',
+    usage: '国内用户选择 feishu；海外 / Lark 用户选择 lark。',
+    valueNotes: [
+      '仅影响 App Bot 主动推送的 API 调用域名，不影响 Webhook URL。',
+      '选错会导致 API 调用失败（SDK 连错服务器）。',
+    ],
+    impact: ['影响飞书 App Bot 主动推送的 API 连通性。'],
+    notes: ['如果值不是 feishu 或 lark，运行时会自动回退到 feishu。'],
+  },
   'settings.notification.DINGTALK_STREAM_ENABLED': {
     title: '钉钉 Stream 模式',
     summary: '启用钉钉应用机器人长连接模式，不是普通钉钉群机器人 Webhook 开关。',
@@ -1237,6 +1276,45 @@ const settingsHelpEnUS: SettingsHelpMap = {
       'Failures should affect only the Feishu app bot path, not the main analysis flow.',
     ],
   },
+  'settings.notification.FEISHU_CHAT_ID': {
+    title: 'Feishu App Bot Push Target',
+    summary: 'Configures the target chat_id (group mode) or open_id (P2P mode) for Feishu App Bot notification delivery.',
+    usage: 'FEISHU_APP_ID and FEISHU_APP_SECRET must also be configured. For groups, use a chat_id starting with oc_. For P2P, use an open_id starting with ou_ and set FEISHU_RECEIVE_ID_TYPE to open_id.',
+    valueNotes: [
+      'FEISHU_APP_ID / FEISHU_APP_SECRET alone do not enable group webhook delivery.',
+      'App Bot mode and Webhook mode are mutually exclusive: webhook URL takes priority; App Bot is used only when no webhook URL is configured.',
+    ],
+    impact: [
+      'Affects the target destination for the Feishu App Bot notification channel.',
+      'Delivery failure should not block the main analysis flow.',
+    ],
+    notes: [
+      'The app bot needs the im:message:send_as_bot permission.',
+      'For P2P messages, the target user must have previously opened the conversation with the app bot in Feishu.',
+    ],
+  },
+  'settings.notification.FEISHU_RECEIVE_ID_TYPE': {
+    title: 'Feishu Receive ID Type',
+    summary: 'Specifies the type of FEISHU_CHAT_ID: chat_id for group chat, open_id for P2P private message.',
+    usage: 'Choose chat_id for groups; choose open_id for sending P2P messages to a specific user.',
+    valueNotes: [
+      'Only takes effect when FEISHU_CHAT_ID is also configured.',
+      'If the type does not match the actual ID, sending will fail with an invalid receive_id error.',
+    ],
+    impact: ['Affects the routing of Feishu App Bot messages.'],
+    notes: ['chat_id covers most use cases. If the value is neither chat_id nor open_id, the runtime falls back to chat_id.'],
+  },
+  'settings.notification.FEISHU_DOMAIN': {
+    title: 'Feishu API Domain',
+    summary: 'Selects the Feishu API region: feishu for mainland China (feishu.cn), lark for international (larksuite.com).',
+    usage: 'Mainland China users choose feishu; international / Lark users choose lark.',
+    valueNotes: [
+      'Only affects the API domain used by App Bot notification delivery; does not affect webhook URLs.',
+      'Choosing the wrong domain causes API errors (SDK connects to the wrong server).',
+    ],
+    impact: ['Affects API connectivity for Feishu App Bot notification delivery.'],
+    notes: ['If the value is neither feishu nor lark, the runtime falls back to feishu.'],
+  },
   'settings.notification.DINGTALK_STREAM_ENABLED': {
     title: 'DingTalk Stream Mode',
     summary: 'Enables DingTalk application bot long-connection mode. It is not the regular DingTalk group webhook switch.',

docs/CHANGELOG.md

@@ -27,6 +27,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
 - [改进] #1386 P6 复用市场阶段与 AnalysisContextPack 公开摘要联动告警、持仓手动分析、历史、回测和通知展示，不新增数据库迁移。
+
+- [新功能] 飞书通知新增应用机器人（App Bot）模式，支持通过 FEISHU_APP_ID / FEISHU_APP_SECRET / FEISHU_CHAT_ID 配置，无需额外创建自定义机器人。
 - [文档] 明确 AnalysisContextPack P6 文档、迁移与回滚边界，并同步既有 `SAVE_CONTEXT_SNAPSHOT` 到 `.env.example`、配置注册表、Web 设置帮助和完整指南。
 - [文档] 补齐 #1386 P7 盘前/盘中/盘后分析的入口、迁移、回滚和用户可见说明。
 - [新功能] 新增默认关闭的 AlphaSift 选股页签，通过 `ALPHASIFT_ENABLED` 明确控制，并保留 `/install` 作为显式修复路径。

docs/bot/feishu-bot-config.md

@@ -3,7 +3,7 @@
 本文只解决两类常见诉求：
 
 1. 把分析结果推送到飞书群
-2. 避免把飞书应用模式和群机器人 Webhook 模式混用
+2. 避免把飞书应用模式、App Bot 主动推送和群机器人 Webhook 模式混用
 
 ## 先分清两种模式
 
@@ -25,9 +25,10 @@ FEISHU_WEBHOOK_SECRET=your_sign_secret
 FEISHU_WEBHOOK_KEYWORD=股票日报
 ```
 
-### 模式二：飞书应用 / Stream Bot / 云文档
+### 模式二：飞书应用 / App Bot / Stream Bot / 云文档
 
 适用场景：
+- 你要用飞书 App Bot 主动向指定群或用户推送通知
 - 你要做飞书应用机器人交互
 - 你要启用 Stream 模式
 - 你要用飞书云文档能力
@@ -37,13 +38,21 @@ FEISHU_WEBHOOK_KEYWORD=股票日报
 ```env
 FEISHU_APP_ID=cli_xxx
 FEISHU_APP_SECRET=xxx
+# App Bot 主动推送时必填
+FEISHU_CHAT_ID=oc_xxx
+# 私聊时设置 open_id；群聊默认 chat_id
+FEISHU_RECEIVE_ID_TYPE=chat_id
+# 事件订阅 / Stream Bot 时才开启
 FEISHU_STREAM_ENABLED=true
 ```
 
 注意：
 - `FEISHU_APP_ID` / `FEISHU_APP_SECRET` 不会直接开启群 Webhook 推送
-- 只想收通知时，不要只填 App ID / Secret，必须优先配置 `FEISHU_WEBHOOK_URL`
+- 简单群通知优先配置 `FEISHU_WEBHOOK_URL`
+- 不用 Webhook 时，App Bot 主动推送必须同时配置 `FEISHU_APP_ID`、`FEISHU_APP_SECRET` 和 `FEISHU_CHAT_ID`
+- `FEISHU_STREAM_ENABLED` 只代表事件订阅 / Stream Bot，不参与主动通知是否配置完成的判断
 - 如果你做的是应用机器人 / Stream Bot，可直接看文末保留的原流程截图参考
+- App Bot 发送路径复用 `requirements.txt` 中已有的 `lark-oapi>=1.0.0`，标准安装使用 `pip install -r requirements.txt`；参考 [Feishu message create OpenAPI](https://open.feishu.cn/document/server-docs/im-v1/message/create)、[lark-oapi PyPI](https://pypi.org/project/lark-oapi/) 和 [SDK repo](https://github.com/larksuite/oapi-sdk-python)
 
 ## Webhook 推送的正确配置步骤
 
@@ -107,6 +116,17 @@ FEISHU_APP_SECRET=...
 
 也不会影响 Webhook 推送；但它们本身不能替代 `FEISHU_WEBHOOK_URL`。
 
+如果未配置 Webhook，也可以用 App Bot 主动推送：
+
+```env
+FEISHU_APP_ID=cli_xxx
+FEISHU_APP_SECRET=xxx
+FEISHU_CHAT_ID=oc_xxx
+FEISHU_RECEIVE_ID_TYPE=chat_id
+```
+
+此时 `FEISHU_STREAM_ENABLED` 不需要开启；它只用于事件订阅 / Stream Bot。
+
 ### 4. 在飞书自动化里配置 Webhook 触发器
 
 如果你在飞书自动化流程里消费本项目推送的卡片消息，请按下面配置：
@@ -152,10 +172,11 @@ FEISHU_APP_SECRET=...
 - 实际完全收不到群通知
 
 原因：
-- 这两个变量是应用模式用的，不是群 Webhook 推送入口
+- 这两个变量只是应用凭据；主动推送还需要 `FEISHU_CHAT_ID`，群 Webhook 推送则需要 `FEISHU_WEBHOOK_URL`
 
 正确做法：
-- 补 `FEISHU_WEBHOOK_URL`
+- 简单群推送：补 `FEISHU_WEBHOOK_URL`
+- App Bot 主动推送：补 `FEISHU_CHAT_ID`，并确认应用有发消息权限且机器人在目标群中
 
 ### 2. 飞书机器人开启了关键词，但本地没配 `FEISHU_WEBHOOK_KEYWORD`
 
@@ -222,10 +243,11 @@ FEISHU_WEBHOOK_KEYWORD=股票日报
 ## 排查顺序建议
 
 1. 先确认你要的是“群 Webhook 推送”还是“应用 / Stream Bot”
-2. 只做群推送时，先保证 `FEISHU_WEBHOOK_URL` 已配置
-3. 回到飞书机器人安全设置，确认是否启用了关键词或签名
-4. 若启用了，就补齐 `FEISHU_WEBHOOK_KEYWORD` / `FEISHU_WEBHOOK_SECRET`
-5. 最后再检查机器人是否在群里、是否有权限、是否命中 IP 白名单
+2. 只做简单群推送时，先保证 `FEISHU_WEBHOOK_URL` 已配置
+3. 不用 Webhook 而走 App Bot 主动推送时，确认 `FEISHU_APP_ID` / `FEISHU_APP_SECRET` / `FEISHU_CHAT_ID` 三项齐全
+4. 回到飞书机器人安全设置，确认是否启用了关键词或签名
+5. 若启用了，就补齐 `FEISHU_WEBHOOK_KEYWORD` / `FEISHU_WEBHOOK_SECRET`
+6. 最后再检查机器人是否在群里、是否有权限、是否命中 IP 白名单
 
 ## 附：应用 / Stream Bot 原流程截图参考
 

docs/full-guide.md

@@ -312,7 +312,9 @@ daily_stock_analysis/
 > 3. 创建群组并添加应用机器人
 > 4. 在云盘文件夹中添加群组为协作者（可管理权限）
 >
-> 说明：`FEISHU_APP_ID` / `FEISHU_APP_SECRET` 用于飞书应用、云文档或 Stream Bot 模式，不会直接启用群 Webhook 推送。只想收通知时，请优先配置 `FEISHU_WEBHOOK_URL`。
+> 说明：`FEISHU_APP_ID` / `FEISHU_APP_SECRET` 用于飞书应用、云文档或 Stream Bot 模式，不会直接启用群 Webhook 推送。只想简单收群通知时，请优先配置 `FEISHU_WEBHOOK_URL`。
+>
+> 补充：若同时配置 `FEISHU_APP_ID`、`FEISHU_APP_SECRET` 和 `FEISHU_CHAT_ID`，则可启用飞书 App Bot 主动通知渠道，无需 Webhook 即可主动向指定 chat 或用户推送；`FEISHU_RECEIVE_ID_TYPE` 默认 `chat_id`，私聊时改为 `open_id`。该方式走飞书 OpenAPI Bot 会话，与群 Webhook 是两条独立链路。
 
 ### 搜索服务配置
 
@@ -912,10 +914,12 @@ FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/your_hook_token
    - **开启了「签名校验」**：把飞书显示的 secret 填到 `FEISHU_WEBHOOK_SECRET`。两端必须同时启用或同时不填，否则飞书返回签名校验失败。
    - **开启了「关键词」**：把同一个关键词填到 `FEISHU_WEBHOOK_KEYWORD`；系统会自动在每条消息前补上，无需手动修改报告模板。
    - **开启了 IP 白名单**：确保当前运行环境的出口 IP 在白名单中（本地/Docker/GitHub Actions 出口 IP 各不相同）。
-4. `FEISHU_APP_ID` / `FEISHU_APP_SECRET` 是飞书应用 / Stream Bot / 云文档模式专用，不会触发群 Webhook 推送，不要用它们替代 `FEISHU_WEBHOOK_URL`。
+4. `FEISHU_APP_ID` / `FEISHU_APP_SECRET` 是飞书应用 / Stream Bot / 云文档模式专用，不会触发群 Webhook 推送，不要只用它们替代 `FEISHU_WEBHOOK_URL`。
+5. 若已配置 `FEISHU_APP_ID` / `FEISHU_APP_SECRET`，再配置 `FEISHU_CHAT_ID`，则可通过飞书 App Bot 直接向指定群聊或用户推送通知，无需依赖群 Webhook；`FEISHU_RECEIVE_ID_TYPE` 默认 `chat_id`，私聊时改为 `open_id`。该方式走飞书 OpenAPI Bot 会话，与群 Webhook 是两条独立链路。
+6. App Bot 发送路径复用 `requirements.txt` 中已有的 `lark-oapi>=1.0.0`，标准源码安装、Docker、GitHub Actions daily workflow 和桌面构建链路都会通过 `pip install -r requirements.txt` 安装，不需要单独安装新库。参考：[Feishu message create OpenAPI](https://open.feishu.cn/document/server-docs/im-v1/message/create)、[lark-oapi PyPI](https://pypi.org/project/lark-oapi/)、[SDK repo](https://github.com/larksuite/oapi-sdk-python)。
 
 **常见失败原因：**
-- 只填了 `FEISHU_APP_ID` / `FEISHU_APP_SECRET`，没有配置 `FEISHU_WEBHOOK_URL`
+- 只填了 `FEISHU_APP_ID` / `FEISHU_APP_SECRET`，既没有配置 `FEISHU_WEBHOOK_URL`，也没有配置 App Bot 主动推送所需的 `FEISHU_CHAT_ID`
 - 飞书机器人开启了「签名校验」，但 `FEISHU_WEBHOOK_SECRET` 未配置（或误填为 `FEISHU_APP_SECRET`）
 - 飞书机器人开启了「关键词」，但本地没有同步配置 `FEISHU_WEBHOOK_KEYWORD`
 - 机器人没有被加入目标群，或群管理员限制了机器人发言

docs/full-guide_EN.md

@@ -275,7 +275,9 @@ For the notification baseline, diagnostics, and deployment notes, see [Notificat
 > 3. Create a group and add the app bot
 > 4. Add the group as a collaborator to the cloud drive folder (with manage permissions)
 >
-> Note: `FEISHU_APP_ID` / `FEISHU_APP_SECRET` are for Feishu app mode, cloud documents, or Stream Bot mode. They do not enable group webhook notifications by themselves. For simple push notifications, use `FEISHU_WEBHOOK_URL` first.
+> Note: `FEISHU_APP_ID` / `FEISHU_APP_SECRET` are for Feishu app mode, cloud documents, or Stream Bot mode. They do not enable group webhook notifications by themselves. For simple group push notifications, use `FEISHU_WEBHOOK_URL` first.
+>
+> Supplement: When `FEISHU_APP_ID`, `FEISHU_APP_SECRET`, and `FEISHU_CHAT_ID` are all configured, they enable the Feishu App Bot active notification channel without relying on group webhooks. `FEISHU_RECEIVE_ID_TYPE` defaults to `chat_id`; set it to `open_id` for P2P delivery. This uses the Feishu OpenAPI Bot session route, which is independent of the group webhook path.
 
 ### Search Service Configuration
 
@@ -780,10 +782,12 @@ FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/your_hook_token
    - **Signature verification enabled**: copy the secret shown in Feishu into `FEISHU_WEBHOOK_SECRET`. **Both sides must be enabled or disabled together** — if Feishu has signing on but `FEISHU_WEBHOOK_SECRET` is missing (or vice versa), every request will be rejected.
    - **Keyword enabled**: copy the exact same keyword into `FEISHU_WEBHOOK_KEYWORD`. The app will prepend it to every message automatically; no need to change report templates.
    - **IP allowlist enabled**: make sure the outbound IP of your runtime (local / Docker / GitHub Actions each have different IPs) is on the allowlist.
-4. `FEISHU_APP_ID` / `FEISHU_APP_SECRET` are for Feishu app / Stream Bot / cloud document flows only — they do **not** trigger group webhook notifications and must not be used instead of `FEISHU_WEBHOOK_URL`.
+4. `FEISHU_APP_ID` / `FEISHU_APP_SECRET` are for Feishu app / Stream Bot / cloud document flows only. They do **not** trigger group webhook notifications and must not be used alone instead of `FEISHU_WEBHOOK_URL`.
+5. If `FEISHU_APP_ID` / `FEISHU_APP_SECRET` are configured together with `FEISHU_CHAT_ID`, the Feishu App Bot can push notifications directly to a specified chat or user, no group webhook required. `FEISHU_RECEIVE_ID_TYPE` defaults to `chat_id`; set it to `open_id` for P2P delivery. This uses the Feishu OpenAPI Bot session route, independent of the group webhook path.
+6. The App Bot send path reuses the existing `lark-oapi>=1.0.0` dependency already listed in `requirements.txt`; standard source installs, Docker, the GitHub Actions daily workflow, and desktop builds all install it through `pip install -r requirements.txt`. References: [Feishu message create OpenAPI](https://open.feishu.cn/document/server-docs/im-v1/message/create), [lark-oapi PyPI](https://pypi.org/project/lark-oapi/), [SDK repo](https://github.com/larksuite/oapi-sdk-python).
 
 **Common failure causes:**
-- Only `FEISHU_APP_ID` / `FEISHU_APP_SECRET` were set, but `FEISHU_WEBHOOK_URL` was not configured
+- Only `FEISHU_APP_ID` / `FEISHU_APP_SECRET` were set, with neither `FEISHU_WEBHOOK_URL` nor the App Bot active-delivery target `FEISHU_CHAT_ID` configured
 - The bot has Signature security enabled, but `FEISHU_WEBHOOK_SECRET` was not set locally (or was mistakenly set to `FEISHU_APP_SECRET`)
 - The bot has Keyword security enabled, but `FEISHU_WEBHOOK_KEYWORD` was not set locally
 - The bot was not added to the target group, or group permissions block it from posting

docs/notifications.md

@@ -7,7 +7,7 @@
 | 渠道 | 类型 | Minimal key | Advanced key | 说明 |
 | --- | --- | --- | --- | --- |
 | 企业微信 | 静态配置 | `WECHAT_WEBHOOK_URL` | `WECHAT_MSG_TYPE` | 配置后参与批量通知发送 |
-| 飞书 Webhook | 静态配置 | `FEISHU_WEBHOOK_URL` | `FEISHU_WEBHOOK_SECRET`, `FEISHU_WEBHOOK_KEYWORD` | `FEISHU_APP_ID` / `FEISHU_APP_SECRET` 不会单独开启群 Webhook 推送 |
+| 飞书 Webhook / App Bot | 静态配置 | `FEISHU_WEBHOOK_URL` 或 `FEISHU_APP_ID` + `FEISHU_APP_SECRET` + `FEISHU_CHAT_ID` | `FEISHU_WEBHOOK_SECRET`, `FEISHU_WEBHOOK_KEYWORD`, `FEISHU_RECEIVE_ID_TYPE`, `FEISHU_DOMAIN` | Webhook URL 优先；未配置 Webhook 时，App Bot 三元组可主动向指定群/用户推送。`FEISHU_STREAM_ENABLED` 仅代表事件订阅 / Stream Bot，不参与主动通知配置完成判断 |
 | Telegram | 静态配置 | `TELEGRAM_BOT_TOKEN`, `TELEGRAM_CHAT_ID` | `TELEGRAM_MESSAGE_THREAD_ID` | token 与 chat id 必须同时存在 |
 | 邮件 | 静态配置 | `EMAIL_SENDER`, `EMAIL_PASSWORD` | `EMAIL_RECEIVERS`, `EMAIL_SENDER_NAME` | `EMAIL_RECEIVERS` 留空时发给自己 |
 | Pushover | 静态配置 | `PUSHOVER_USER_KEY`, `PUSHOVER_API_TOKEN` | - | 两个 key 必须同时存在 |
@@ -34,6 +34,7 @@
 - `WEBHOOK_VERIFY_SSL` 是读取该配置的 webhook-style HTTPS 通知请求共用的证书校验开关。
 - WebPush、Apprise、更细粒度路由、跨进程降噪和真实每日摘要暂不进入运行时实现；相关配置如未来引入，应先更新本文档、`.env.example`、Web 元数据与回归测试。
 - Bark 保持 custom webhook 基线，不新增 `BARK_*` 一等配置。
+- 飞书 App Bot 发送路径复用 `requirements.txt` 中已有的 `lark-oapi>=1.0.0`，不是新增依赖；标准源码安装、Docker、GitHub Actions daily workflow 和桌面构建链路均通过 `pip install -r requirements.txt` 安装。官方依据：[Feishu message create OpenAPI](https://open.feishu.cn/document/server-docs/im-v1/message/create)、[lark-oapi PyPI](https://pypi.org/project/lark-oapi/)、[SDK repo](https://github.com/larksuite/oapi-sdk-python)。
 
 ## 报告渲染与分片
 
@@ -83,6 +84,11 @@
 | `DISCORD_WEBHOOK_URL` | minimal | discord | Secret | - |
 | `DISCORD_BOT_TOKEN` | minimal | discord | Secret | - |
 | `DISCORD_MAIN_CHANNEL_ID` | minimal | discord | Secret | - |
+| `FEISHU_APP_ID` | minimal | feishu | Secret | - |
+| `FEISHU_APP_SECRET` | minimal | feishu | Secret | - |
+| `FEISHU_CHAT_ID` | minimal | feishu | Variable or Secret | - |
+| `FEISHU_RECEIVE_ID_TYPE` | advanced | feishu | Variable or Secret | - |
+| `FEISHU_DOMAIN` | advanced | feishu | Variable or Secret | - |
 | `ASTRBOT_URL` | minimal | astrbot | Secret | - |
 | `ASTRBOT_TOKEN` | advanced | astrbot | Secret | - |
 | `SERVERCHAN3_SENDKEY` | minimal | serverchan3 | Secret | - |

scripts/e2e_test_feishu_app.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+"""
+Feishu App Bot manual smoke validation and live-send test.
+
+Usage:
+    ssh <host> "python3 /path/to/e2e_test_feishu_app.py"
+
+Requires FEISHU_APP_ID and FEISHU_APP_SECRET env vars set on the target host.
+Optionally set FEISHU_CHAT_ID to send a live test message via interactive card.
+Optionally set FEISHU_OPEN_ID to send a live P2P test message (overrides CHAT_ID).
+Optionally set FEISHU_DOMAIN to "lark" for international (Lark) tenants.
+"""
+import logging
+import os
+import sys
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+logger = logging.getLogger("feishu-manual-smoke")
+
+# 1. Check credentials
+app_id = os.getenv("FEISHU_APP_ID", "").strip()
+app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
+if not app_id or not app_secret:
+    logger.error("Missing FEISHU_APP_ID or FEISHU_APP_SECRET")
+    sys.exit(1)
+
+# 2. Try getting tenant_access_token
+import requests
+
+_domain = os.getenv("FEISHU_DOMAIN", "feishu").strip().lower()
+if _domain not in ("feishu", "lark"):
+    logger.warning("Invalid FEISHU_DOMAIN=%s; falling back to feishu", _domain)
+    _domain = "feishu"
+_base_host_by_domain = {
+    "feishu": "open.feishu.cn",
+    "lark": "open.larksuite.com",
+}
+_base_host = _base_host_by_domain[_domain]
+logger.info("using domain=%s base_host=%s", _domain, _base_host)
+
+resp = requests.post(
+    f"https://{_base_host}/open-apis/auth/v3/tenant_access_token/internal",
+    json={"app_id": app_id, "app_secret": app_secret},
+    timeout=30,
+)
+token_data = resp.json()
+if token_data.get("code") != 0:
+    logger.error("Failed to get tenant_token: %s", token_data)
+    sys.exit(1)
+
+token = token_data["tenant_access_token"]
+logger.info("token obtained OK")
+
+# 3. List chats (groups) to find available chat_ids
+chats_resp = requests.get(
+    f"https://{_base_host}/open-apis/im/v1/chats?page_size=20",
+    headers={"Authorization": f"Bearer {token}"},
+    timeout=30,
+)
+chats_data = chats_resp.json()
+logger.info("chats API response code=%s", chats_data.get("code"))
+if chats_data.get("code") == 0:
+    items = chats_data.get("data", {}).get("items", [])
+    logger.info("Found %d chats:", len(items))
+    for chat in items:
+        logger.info(
+            "  chat_id=%s name=%s type=%s",
+            chat.get("chat_id"), chat.get("name"), chat.get("chat_type"),
+        )
+else:
+    logger.warning("chat list failed (may lack im:chat permission): %s", chats_data)
+    logger.warning("Trying /bot/v3/info instead...")
+
+# 4. Import lark-oapi SDK (installed by standard requirements.txt setup)
+try:
+    import lark_oapi as lark
+except ImportError:
+    logger.error(
+        "lark-oapi is NOT installed. Standard project setup installs it via:\n"
+        "    pip install -r requirements.txt"
+    )
+    sys.exit(1)
+
+# 5. Verify SDK client initialisation (with domain support)
+smoke_client = (
+    lark.Client.builder()
+    .app_id(app_id)
+    .app_secret(app_secret)
+    .domain(
+        lark.core.const.FEISHU_DOMAIN if _domain == "feishu" else lark.core.const.LARK_DOMAIN
+    )
+    .build()
+)
+logger.info("lark-oapi SDK client init OK (domain=%s, client=%s)", _domain, type(smoke_client).__name__)
+logger.info("manual smoke setup verification passed.")
+
+# 6. Live send test
+_chat_id = os.getenv("FEISHU_CHAT_ID", "").strip()
+_open_id = os.getenv("FEISHU_OPEN_ID", "").strip()
+
+_receive_id_type = "chat_id"
+_receive_id = _chat_id
+
+if _open_id:
+    _receive_id_type = "open_id"
+    _receive_id = _open_id
+    logger.info("FEISHU_OPEN_ID=%s will send P2P message", _open_id)
+
+if _receive_id:
+    logger.info("FEISHU_CHAT_ID=%s, performing live send test...", _receive_id)
+
+    # Add project root to path so source imports resolve
+    _project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    if _project_root not in sys.path:
+        sys.path.insert(0, _project_root)
+
+    from src.notification_sender.feishu_sender import FeishuSender
+    from src.config import Config
+
+    config = Config()
+    config.feishu_app_id = app_id
+    config.feishu_app_secret = app_secret
+    config.feishu_chat_id = _receive_id
+    config.feishu_receive_id_type = _receive_id_type
+    config.feishu_domain = _domain
+
+    sender = FeishuSender(config)
+
+    # Interactive-card send (default FeishuSender path: card-first, text-fallback)
+    ok_card = sender.send_to_feishu(
+        "**Feishu Manual Smoke Test Message**\n\n"
+        "This is a manual smoke test from `e2e_test_feishu_app.py`\n"
+        f"(mode: {_receive_id_type})."
+    )
+    if ok_card:
+        logger.info("Live send test PASSED (via card or text fallback).")
+    else:
+        logger.error("Live send test FAILED - check FEISHU_CHAT_ID and bot permissions.")
+        sys.exit(1)
+else:
+    logger.info(
+        "Neither FEISHU_CHAT_ID nor FEISHU_OPEN_ID set; skipping live send. "
+        "Set FEISHU_CHAT_ID or FEISHU_OPEN_ID to test actual message delivery."
+    )

src/config.py

@@ -31,6 +31,10 @@ from src.notification_noise import (
     parse_notification_quiet_hours,
     validate_notification_timezone,
 )
+from src.notification_contracts import (
+    is_feishu_app_bot_configured,
+    is_feishu_static_configured,
+)
 from src.llm import generation_params as llm_generation_params
 
 logger = logging.getLogger(__name__)
@@ -747,6 +751,11 @@ class Config:
     feishu_webhook_url: Optional[str] = None
     feishu_webhook_secret: Optional[str] = None  # 自定义机器人签名密钥（可选）
     feishu_webhook_keyword: Optional[str] = None  # 自定义机器人关键词（可选）
+
+    # 飞书应用机器人（App Bot）通知
+    feishu_chat_id: Optional[str] = None  # 目标群会话 chat_id（群聊模式），或用户 open_id（P2P 模式）
+    feishu_receive_id_type: str = "chat_id"  # 接收者 ID 类型: "chat_id"(群聊) / "open_id"(私聊)
+    feishu_domain: str = "feishu"  # 飞书域名: "feishu"(feishu.cn) / "lark"(larksuite.com)
     
     # Telegram 配置（需要同时配置 Bot Token 和 Chat ID）
     telegram_bot_token: Optional[str] = None  # Bot Token（@BotFather 获取）
@@ -1551,6 +1560,10 @@ class Config:
             feishu_webhook_url=os.getenv('FEISHU_WEBHOOK_URL'),
             feishu_webhook_secret=os.getenv('FEISHU_WEBHOOK_SECRET'),
             feishu_webhook_keyword=os.getenv('FEISHU_WEBHOOK_KEYWORD'),
+
+            feishu_chat_id=os.getenv('FEISHU_CHAT_ID'),
+            feishu_receive_id_type=os.getenv('FEISHU_RECEIVE_ID_TYPE', 'chat_id'),
+            feishu_domain=os.getenv('FEISHU_DOMAIN', 'feishu'),
             telegram_bot_token=os.getenv('TELEGRAM_BOT_TOKEN'),
             telegram_chat_id=os.getenv('TELEGRAM_CHAT_ID'),
             telegram_message_thread_id=os.getenv('TELEGRAM_MESSAGE_THREAD_ID'),
@@ -2606,6 +2619,11 @@ class Config:
         has_notification = bool(
             self.wechat_webhook_url
             or self.feishu_webhook_url
+            or (
+                (self.feishu_app_id or "")
+                and (self.feishu_app_secret or "")
+                and (self.feishu_chat_id or "")
+            )
             or (self.telegram_bot_token and self.telegram_chat_id)
             or (self.email_sender and self.email_password)
             or (self.pushover_user_key and self.pushover_api_token)
@@ -2742,27 +2760,35 @@ class Config:
 
         has_feishu_app_id = bool((self.feishu_app_id or "").strip())
         has_feishu_app_secret = bool((self.feishu_app_secret or "").strip())
+        has_feishu_app_credentials_complete = has_feishu_app_id and has_feishu_app_secret
         has_feishu_app_credentials = has_feishu_app_id or has_feishu_app_secret
         has_feishu_doc_token = bool((self.feishu_folder_token or "").strip())
         has_feishu_full_cloud_doc_credentials = (
-            has_feishu_app_id
-            and has_feishu_app_secret
+            has_feishu_app_credentials_complete
             and has_feishu_doc_token
         )
+        has_feishu_stream_route = bool(self.feishu_stream_enabled and has_feishu_app_credentials_complete)
+        has_feishu_app_notification_route = is_feishu_app_bot_configured(self)
         if (
             has_feishu_app_credentials
             and not has_feishu_full_cloud_doc_credentials
-            and not self.feishu_webhook_url
-            and not (self.feishu_stream_enabled and has_feishu_app_id and has_feishu_app_secret)
+            and not is_feishu_static_configured(self)
+            and not has_feishu_stream_route
+            and not has_feishu_app_notification_route
         ):
+            suggestions = []
+            if has_feishu_app_credentials_complete:
+                suggestions.append("配置 FEISHU_CHAT_ID 开启 App Bot 主动推送")
+                suggestions.append("开启 FEISHU_STREAM_ENABLED 使用应用机器人事件订阅")
+            else:
+                suggestions.append("补齐 FEISHU_APP_ID / FEISHU_APP_SECRET 后配置 FEISHU_CHAT_ID 开启 App Bot 主动推送")
+            suggestions.append("配置 FEISHU_WEBHOOK_URL 使用自定义机器人 Webhook 推送")
             issues.append(ConfigIssue(
                 severity="warning",
-                message=(
-                    "仅配置 FEISHU_APP_ID / FEISHU_APP_SECRET 不会开启飞书群 Webhook 推送；"
-                    "如需群消息通知，请配置 FEISHU_WEBHOOK_URL。若要使用应用机器人，请同时开启 "
-                    "FEISHU_STREAM_ENABLED 并完成应用发布与权限配置。"
-                ),
-                field="FEISHU_WEBHOOK_URL",
+                message="仅配置 FEISHU_APP_ID / FEISHU_APP_SECRET 不会开启飞书静态通知。"
+                        + " 请选择以下方式之一："
+                        + "；".join(suggestions) + "。",
+                field="FEISHU_CHAT_ID",
             ))
 
         # --- Deprecated field migration hints ---

src/core/config_registry.py

@@ -1583,6 +1583,81 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
         ],
         "warning_codes": ["not_webhook_delivery", "restart_required"],
     },
+    "FEISHU_CHAT_ID": {
+        "title": "Feishu Chat ID",
+        "description": "Target chat_id (group mode, oc_xxx) or open_id (P2P mode, ou_xxx) for Feishu App Bot notification delivery. Requires FEISHU_APP_ID + FEISHU_APP_SECRET.",
+        "category": "notification",
+        "data_type": "string",
+        "ui_control": "text",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": None,
+        "options": [],
+        "validation": {},
+        "display_order": 18,
+        "help_key": "settings.notification.FEISHU_CHAT_ID",
+        "examples": [
+            "FEISHU_CHAT_ID=oc_xxxxxxxxxxxxx",
+            "FEISHU_CHAT_ID=ou_xxxxxxxxxxxxx",
+        ],
+        "docs": [
+            {
+                "label": "完整指南：飞书通知配置",
+                "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#飞书",
+            },
+        ],
+    },
+    "FEISHU_RECEIVE_ID_TYPE": {
+        "title": "Feishu Receive ID Type",
+        "description": "Type of FEISHU_CHAT_ID: 'chat_id' for group chat, 'open_id' for P2P private message.",
+        "category": "notification",
+        "data_type": "string",
+        "ui_control": "select",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": "chat_id",
+        "options": [{"label": "chat_id (群聊)", "value": "chat_id"}, {"label": "open_id (私聊)", "value": "open_id"}],
+        "validation": {"enum": ["chat_id", "open_id"]},
+        "display_order": 19,
+        "help_key": "settings.notification.FEISHU_RECEIVE_ID_TYPE",
+        "examples": [
+            "FEISHU_RECEIVE_ID_TYPE=chat_id",
+            "FEISHU_RECEIVE_ID_TYPE=open_id",
+        ],
+        "docs": [
+            {
+                "label": "完整指南：飞书通知配置",
+                "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#飞书",
+            },
+        ],
+    },
+    "FEISHU_DOMAIN": {
+        "title": "Feishu Domain",
+        "description": "Feishu API domain: 'feishu' (feishu.cn for mainland China) or 'lark' (larksuite.com for international).",
+        "category": "notification",
+        "data_type": "string",
+        "ui_control": "select",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": "feishu",
+        "options": [{"label": "feishu (飞书国内)", "value": "feishu"}, {"label": "lark (国际版)", "value": "lark"}],
+        "validation": {"enum": ["feishu", "lark"]},
+        "display_order": 20,
+        "help_key": "settings.notification.FEISHU_DOMAIN",
+        "examples": [
+            "FEISHU_DOMAIN=feishu",
+            "FEISHU_DOMAIN=lark",
+        ],
+        "docs": [
+            {
+                "label": "完整指南：飞书通知配置",
+                "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#飞书",
+            },
+        ],
+    },
     # ------------------------------------------------------------------
     # Notification – Telegram
     # ------------------------------------------------------------------

src/notification.py

@@ -30,6 +30,7 @@ from src.notification_routing import (
     get_notification_route_config,
     split_notification_route_channels,
 )
+from src.notification_contracts import is_feishu_static_configured
 from src.notification_noise import (
     NotificationNoiseDecision,
     evaluate_notification_noise,
@@ -362,7 +363,7 @@ class NotificationService(
         if getattr(config, "wechat_webhook_url", None):
             channels.append(NotificationChannel.WECHAT)
 
-        if getattr(config, "feishu_webhook_url", None):
+        if is_feishu_static_configured(config):
             channels.append(NotificationChannel.FEISHU)
 
         if (

src/notification_contracts.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+"""Shared notification configuration contracts.
+
+This module intentionally stays lightweight: no sender imports, no SDK imports,
+and no NotificationService imports. It is safe for config, diagnostics, and
+runtime channel detection to share.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping, Tuple
+
+
+FEISHU_WEBHOOK_ENV_GROUP: Tuple[str, ...] = ("FEISHU_WEBHOOK_URL",)
+FEISHU_APP_BOT_ENV_GROUP: Tuple[str, ...] = (
+    "FEISHU_APP_ID",
+    "FEISHU_APP_SECRET",
+    "FEISHU_CHAT_ID",
+)
+FEISHU_STATIC_ENV_GROUPS: Tuple[Tuple[str, ...], ...] = (
+    FEISHU_WEBHOOK_ENV_GROUP,
+    FEISHU_APP_BOT_ENV_GROUP,
+)
+
+_FEISHU_WEBHOOK_CONFIG_GROUP: Tuple[str, ...] = ("feishu_webhook_url",)
+_FEISHU_APP_BOT_CONFIG_GROUP: Tuple[str, ...] = (
+    "feishu_app_id",
+    "feishu_app_secret",
+    "feishu_chat_id",
+)
+_FEISHU_STATIC_CONFIG_GROUPS: Tuple[Tuple[str, ...], ...] = (
+    _FEISHU_WEBHOOK_CONFIG_GROUP,
+    _FEISHU_APP_BOT_CONFIG_GROUP,
+)
+
+
+def _has_env_group(effective_map: Mapping[str, Any], group: Tuple[str, ...]) -> bool:
+    return all(str(effective_map.get(key) or "").strip() for key in group)
+
+
+def is_feishu_app_bot_env_configured(effective_map: Mapping[str, Any]) -> bool:
+    """Return whether Feishu App Bot active notification is configured."""
+    return _has_env_group(effective_map, FEISHU_APP_BOT_ENV_GROUP)
+
+
+def is_feishu_static_env_configured(effective_map: Mapping[str, Any]) -> bool:
+    """Return whether any static Feishu notification route is configured."""
+    return any(_has_env_group(effective_map, group) for group in FEISHU_STATIC_ENV_GROUPS)
+
+
+def _has_config_group(config: Any, group: Tuple[str, ...]) -> bool:
+    return all(str(getattr(config, attr, None) or "").strip() for attr in group)
+
+
+def is_feishu_app_bot_configured(config: Any) -> bool:
+    """Return whether a Config-like object has the App Bot notification triad."""
+    return _has_config_group(config, _FEISHU_APP_BOT_CONFIG_GROUP)
+
+
+def is_feishu_static_configured(config: Any) -> bool:
+    """Return whether a Config-like object has any static Feishu route."""
+    return any(_has_config_group(config, group) for group in _FEISHU_STATIC_CONFIG_GROUPS)

src/notification_sender/feishu_sender.py

@@ -4,12 +4,17 @@
 
 职责：
 1. 通过 webhook 发送飞书消息
+2. 通过飞书应用机器人（App Bot）发送消息（lark-oapi SDK）
 """
 import base64
 import hashlib
 import hmac
+import json
 import logging
+import os
+import threading
 import time
+import uuid as uuid_mod
 from typing import Any, Dict, Optional
 
 import requests
@@ -22,112 +27,355 @@ from src.formatters import (
     format_feishu_markdown,
 )
 
-
 logger = logging.getLogger(__name__)
 
+# ---------------------------------------------------------------------------
+# lark-oapi SDK availability
+# ---------------------------------------------------------------------------
+
+FEISHU_SDK_AVAILABLE = False
+_lark: Any = None  # type: ignore[assignment]
+FEISHU_DOMAIN = "feishu"
+LARK_DOMAIN = "lark"
+try:
+    import lark_oapi as _lark
+    from lark_oapi.api.im.v1 import (
+        CreateMessageRequest,
+        CreateMessageRequestBody,
+    )
+    from lark_oapi.core.const import FEISHU_DOMAIN as _SDK_FEISHU_DOMAIN
+    from lark_oapi.core.const import LARK_DOMAIN as _SDK_LARK_DOMAIN
+
+    FEISHU_DOMAIN = _SDK_FEISHU_DOMAIN
+    LARK_DOMAIN = _SDK_LARK_DOMAIN
+    FEISHU_SDK_AVAILABLE = True
+except ImportError:
+    pass
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_APP_SEND_RETRIES = 3
+_APP_SEND_BACKOFF_SECONDS = (1.0, 2.0, 4.0)
+_WEBHOOK_SEND_TIMEOUT_SECONDS = 30
+
+# Sentinel for "client not yet initialised".
+_NO_CLIENT = object()
+
 
 class FeishuSender:
 
     def __init__(self, config: Config):
         """
-        初始化飞书配置
+        Initialise Feishu sender.
 
-        Args:
-            config: 配置对象
+        Two mutually exclusive routing modes are supported:
+          1. **Webhook** – configured via ``feishu_webhook_url`` (legacy).
+          2. **App Bot** – configured via ``feishu_app_id`` + ``feishu_app_secret``
+             + ``feishu_chat_id``, sends through the ``lark-oapi`` SDK.
+
+        Webhook mode takes precedence when both are configured.
         """
-        self._feishu_url = getattr(config, 'feishu_webhook_url', None)
-        self._feishu_secret = (getattr(config, 'feishu_webhook_secret', None) or '').strip()
-        self._feishu_keyword = (getattr(config, 'feishu_webhook_keyword', None) or '').strip()
-        self._feishu_max_bytes = getattr(config, 'feishu_max_bytes', 20000)
-        self._webhook_verify_ssl = getattr(config, 'webhook_verify_ssl', True)
+        # -- Webhook mode --
+        self._feishu_url = getattr(config, "feishu_webhook_url", None)
+        self._feishu_secret = (getattr(config, "feishu_webhook_secret", None) or "").strip()
+        self._feishu_keyword = (getattr(config, "feishu_webhook_keyword", None) or "").strip()
+        self._feishu_max_bytes = getattr(config, "feishu_max_bytes", 20000)
+        self._webhook_verify_ssl = getattr(config, "webhook_verify_ssl", True)
+
+        # -- App Bot mode --
+        self._feishu_app_id = (getattr(config, "feishu_app_id", None) or "").strip()
+        self._feishu_app_secret = (getattr(config, "feishu_app_secret", None) or "").strip()
+        self._feishu_chat_id = (getattr(config, "feishu_chat_id", None) or "").strip()
+        self._feishu_receive_id_type = (
+            getattr(config, "feishu_receive_id_type", None) or "chat_id"
+        ).strip().lower()
+        if self._feishu_receive_id_type not in ("chat_id", "open_id"):
+            logger.warning(
+                "无效的 FEISHU_RECEIVE_ID_TYPE=%s，回退为 chat_id",
+                self._feishu_receive_id_type,
+            )
+            self._feishu_receive_id_type = "chat_id"
+        # domain_name must be "feishu" or "lark"; anything else defaulted to feishu.
+        raw_domain = (
+            getattr(config, "feishu_domain", None) or os.getenv("FEISHU_DOMAIN", "feishu")
+        ).strip().lower()
+        if raw_domain not in ("feishu", "lark"):
+            logger.warning(
+                "无效的 FEISHU_DOMAIN=%s，回退为 feishu", raw_domain
+            )
+            raw_domain = "feishu"
+        self._feishu_domain = FEISHU_DOMAIN if raw_domain == "feishu" else LARK_DOMAIN
+
+        self._app_client: Any = _NO_CLIENT
+        self._app_client_lock = threading.Lock()
+
+    # ------------------------------------------------------------------
+    # Shared helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _build_card_body(content: str) -> dict:
+        """Build a Feishu interactive-card body (without the ``msg_type`` wrapper)."""
+        return {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "title": {"tag": "plain_text", "content": "股票智能分析报告"},
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {"tag": "lark_md", "content": content},
+                }
+            ],
+        }
+
+    # ------------------------------------------------------------------
+    # Webhook helpers (unchanged legacy path)
+    # ------------------------------------------------------------------
 
     def _get_keyword_prefix(self) -> str:
-        """Return the keyword prefix required by Feishu webhook security settings."""
         if not self._feishu_keyword:
             return ""
         return f"{self._feishu_keyword}\n"
 
     def _apply_keyword_prefix(self, content: str) -> str:
-        """Prepend the optional keyword so each webhook request passes keyword checks."""
         prefix = self._get_keyword_prefix()
         if not prefix:
             return content
         return f"{prefix}{content}" if content else self._feishu_keyword
 
     def _build_security_fields(self) -> Dict[str, str]:
-        """Build optional signing fields required by Feishu custom robot security."""
         if not self._feishu_secret:
             return {}
-
         timestamp = str(int(time.time()))
         string_to_sign = f"{timestamp}\n{self._feishu_secret}"
         sign = base64.b64encode(
             hmac.new(
-                string_to_sign.encode('utf-8'),
+                string_to_sign.encode("utf-8"),
                 digestmod=hashlib.sha256,
             ).digest()
-        ).decode('utf-8')
-        return {
-            "timestamp": timestamp,
-            "sign": sign,
-        }
+        ).decode("utf-8")
+        return {"timestamp": timestamp, "sign": sign}
 
+    # ------------------------------------------------------------------
+    # App Bot client (lazy, thread-safe)
+    # ------------------------------------------------------------------
+
+    def _ensure_app_client(self) -> Any:
+        """Lazily initialise the ``lark-oapi`` client for App Bot mode."""
+        if self._app_client is not _NO_CLIENT:
+            return self._app_client
+        with self._app_client_lock:
+            if self._app_client is not _NO_CLIENT:
+                return self._app_client
+            if not FEISHU_SDK_AVAILABLE:
+                logger.warning(
+                    "飞书 App Bot 需要 lark-oapi 库；标准安装请运行: pip install -r requirements.txt"
+                )
+                self._app_client = None
+                return None
+            if not self._feishu_app_id or not self._feishu_app_secret:
+                missing = []
+                if not self._feishu_app_id:
+                    missing.append("FEISHU_APP_ID")
+                if not self._feishu_app_secret:
+                    missing.append("FEISHU_APP_SECRET")
+                logger.warning("飞书 App Bot 凭据不全，缺少: %s", ", ".join(missing))
+                self._app_client = None
+                return None
+            try:
+                self._app_client = (
+                    _lark.Client.builder()
+                    .app_id(self._feishu_app_id)
+                    .app_secret(self._feishu_app_secret)
+                    .domain(self._feishu_domain)
+                    .log_level(_lark.LogLevel.WARNING)
+                    .build()
+                )
+                logger.info("飞书 App Bot 客户端初始化成功 (domain=%s)", self._feishu_domain)
+            except Exception as e:
+                logger.error("飞书 App Bot 客户端初始化失败: %s", e)
+                self._app_client = None
+            return self._app_client
+
+    # ------------------------------------------------------------------
+    # App Bot send helpers
+    # ------------------------------------------------------------------
+
+    def _send_via_app_bot(self, content: str) -> bool:
+        """Send message through the Feishu App Bot, chunking if necessary."""
+        if not self._feishu_chat_id:
+            logger.warning("FEISHU_CHAT_ID 未配置，跳过 App Bot 推送")
+            return False
+
+        client = self._ensure_app_client()
+        if client is None:
+            return False
+
+        formatted = format_feishu_markdown(content)
+        content_bytes = len(formatted.encode("utf-8"))
+
+        if content_bytes > self._feishu_max_bytes:
+            logger.info(
+                "App Bot 消息超长 (%d 字节)，将分批发送", content_bytes
+            )
+            return self._app_send_chunked(client, formatted)
+
+        return self._app_send_once(client, formatted)
+
+    def _app_send_chunked(self, client: Any, content: str) -> bool:
+        """Chunk and send long content through App Bot."""
+        try:
+            chunks = chunk_content_by_max_bytes(
+                content, self._feishu_max_bytes, add_page_marker=True
+            )
+        except (ValueError, TypeError, Exception) as e:
+            logger.error("App Bot 分片失败: %s", e)
+            return False
+
+        success = True
+        for i, chunk in enumerate(chunks):
+            ok = self._app_send_once(client, chunk)
+            if not ok:
+                logger.error("App Bot 第 %d/%d 批发送失败", i + 1, len(chunks))
+                success = False
+            if i < len(chunks) - 1:
+                time.sleep(1)
+        return success
+
+    def _app_send_once(self, client: Any, content: str) -> bool:
+        """Single-shot send via App Bot with card-first / text-fallback.
+
+        Content received here has already been through ``format_feishu_markdown``
+        which converts all Markdown constructs to ``lark_md``-compatible format.
+        The interactive card uses ``tag: lark_md`` for rendering.
+        """
+        card_payload = json.dumps(self._build_card_body(content), ensure_ascii=False)
+
+        if self._app_send_raw(client, "interactive", card_payload):
+            return True
+
+        # Fallback to plain text.
+        text_payload = json.dumps({"text": content}, ensure_ascii=False)
+        return self._app_send_raw(client, "text", text_payload)
+
+    def _app_send_raw(self, client: Any, msg_type: str, content_json: str) -> bool:
+        """Low-level send via lark-oapi SDK with retry and idempotency UUID.
+
+        Request construction is done once outside the retry loop; it is
+        deterministic and a construction error is a programming error, not
+        a transient failure.
+        """
+        if client is None:
+            return False
+
+        send_uuid = str(uuid_mod.uuid4())
+        try:
+            req = (
+                CreateMessageRequest.builder()
+                .receive_id_type(self._feishu_receive_id_type)
+                .request_body(
+                    CreateMessageRequestBody.builder()
+                    .receive_id(self._feishu_chat_id)
+                    .content(content_json)
+                    .msg_type(msg_type)
+                    .uuid(send_uuid)
+                    .build()
+                )
+                .build()
+            )
+        except Exception as e:
+            logger.error("App Bot 请求构建失败: %s: %s", type(e).__name__, e)
+            return False
+
+        last_status: Optional[str] = None
+
+        for attempt in range(_APP_SEND_RETRIES):
+            try:
+                resp = client.im.v1.message.create(req)
+            except Exception as e:
+                logger.warning(
+                    "App Bot 发送异常 (attempt=%d/%d): %s: %s",
+                    attempt + 1, _APP_SEND_RETRIES, type(e).__name__, e,
+                )
+                if attempt < _APP_SEND_RETRIES - 1:
+                    time.sleep(
+                        _APP_SEND_BACKOFF_SECONDS[
+                            min(attempt, len(_APP_SEND_BACKOFF_SECONDS) - 1)
+                        ]
+                    )
+                continue
+
+            if resp.success():
+                logger.info("App Bot 消息发送成功 (type=%s)", msg_type)
+                return True
+
+            try:
+                log_id = resp.get_log_id()
+            except (AttributeError, Exception):
+                log_id = "N/A"
+            status = "code=%s, msg=%s, log_id=%s" % (
+                resp.code, resp.msg, log_id,
+            )
+            last_status = status
+            logger.warning(
+                "App Bot 发送失败 (attempt=%d/%d): %s",
+                attempt + 1, _APP_SEND_RETRIES, status,
+            )
+
+            if attempt < _APP_SEND_RETRIES - 1:
+                time.sleep(
+                    _APP_SEND_BACKOFF_SECONDS[
+                        min(attempt, len(_APP_SEND_BACKOFF_SECONDS) - 1)
+                    ]
+                )
+
+        if last_status:
+            logger.error("App Bot 发送最终失败: %s", last_status)
+        return False
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
 
     def send_to_feishu(self, content: str, *, timeout_seconds: Optional[float] = None) -> bool:
         """
-        推送消息到飞书机器人
+        Push a message to Feishu.
 
-        飞书自定义机器人 Webhook 消息格式：
-        {
-            "msg_type": "interactive",
-            "card": {
-                "config": { "wide_screen_mode": true },
-                "elements": [
-                    {
-                        "tag": "div",
-                        "text": {
-                            "tag": "lark_md",
-                            "content": "..."
-                        }
-                    }
-                ],
-                "header": {
-                    "title": {
-                        "tag": "plain_text",
-                        "content": "A股智能分析报告"
-                    }
-                }
-            }
-        }
-
-        说明：飞书文本消息不会渲染 Markdown，需使用交互卡片（lark_md）格式
-
-        注意：飞书文本消息限制约 20KB，超长内容会自动分批发送
-        可通过环境变量 FEISHU_MAX_BYTES 调整限制值
-
-        Args:
-            content: 消息内容（Markdown 会转为纯文本）
+        Routing priority:
+          1. **Webhook** – when ``feishu_webhook_url`` is configured.
+          2. **App Bot** – when ``feishu_app_id`` + ``feishu_app_secret``
+             + ``feishu_chat_id`` are all configured and webhook is absent.
 
         Returns:
-            是否发送成功
+            Whether the send succeeded.
         """
-        if not self._feishu_url:
-            logger.warning("飞书 Webhook 未配置，跳过推送")
+        if content is None:
+            logger.error("send_to_feishu: content 不能为 None")
             return False
+        if self._feishu_url:
+            return self._send_via_webhook(content, timeout_seconds=timeout_seconds)
+        return self._send_via_app_bot(content)
 
-        # 飞书 lark_md 支持有限，先做格式转换
+    # ------------------------------------------------------------------
+    # Webhook path (legacy, unchanged)
+    # ------------------------------------------------------------------
+
+    def _send_via_webhook(self, content: str, *, timeout_seconds: Optional[float] = None) -> bool:
+        """Legacy webhook send path."""
         formatted_content = format_feishu_markdown(content)
 
-        max_bytes = self._feishu_max_bytes  # 从配置读取，默认 20000 字节
-        keyword_overhead = len(self._get_keyword_prefix().encode('utf-8'))
+        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
 
-        # 检查字节长度，超长则分批发送
-        content_bytes = len(formatted_content.encode('utf-8')) + keyword_overhead
+        content_bytes = len(formatted_content.encode("utf-8")) + keyword_overhead
         if content_bytes > max_bytes:
             min_chunk_bytes = MIN_MAX_BYTES + PAGE_MARKER_SAFE_BYTES
             if effective_max_bytes < min_chunk_bytes:
@@ -137,126 +385,87 @@ class FeishuSender:
                     min_chunk_bytes,
                 )
                 return False
-            logger.info(f"飞书消息内容超长({content_bytes}字节/{len(content)}字符)，将分批发送")
+            logger.info("飞书消息内容超长(%d字节/%d字符)，将分批发送", 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}")
+            logger.error("发送飞书消息失败: %s", e)
             return False
 
     def _send_feishu_chunked(self, content: str, max_bytes: int) -> bool:
-        """
-        分批发送长消息到飞书
-
-        按股票分析块（以 --- 或 ### 分隔）智能分割，确保每批不超过限制
-
-        Args:
-            content: 完整消息内容
-            max_bytes: 单条消息最大字节数
-
-        Returns:
-            是否全部发送成功
-        """
         try:
             chunks = chunk_content_by_max_bytes(content, max_bytes, add_page_marker=True)
         except ValueError as e:
             logger.error("飞书消息分片失败，单片预算不足以安全分页（关键词过长或 max_bytes 过小）: %s", e)
             return False
 
-        # 分批发送
         total_chunks = len(chunks)
         success_count = 0
-
-        logger.info(f"飞书分批发送：共 {total_chunks} 批")
-
+        logger.info("飞书分批发送：共 %d 批", total_chunks)
         for i, chunk in enumerate(chunks):
             try:
                 if self._send_feishu_message(chunk):
                     success_count += 1
-                    logger.info(f"飞书第 {i+1}/{total_chunks} 批发送成功")
+                    logger.info("飞书第 %d/%d 批发送成功", i + 1, total_chunks)
                 else:
-                    logger.error(f"飞书第 {i+1}/{total_chunks} 批发送失败")
+                    logger.error("飞书第 %d/%d 批发送失败", i + 1, total_chunks)
             except Exception as e:
-                logger.error(f"飞书第 {i+1}/{total_chunks} 批发送异常: {e}")
-
-            # 批次间隔，避免触发频率限制
+                logger.error("飞书第 %d/%d 批发送异常: %s", 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 卡片）"""
+        """Send a single Feishu webhook message (interactive card, fallback text)."""
         prepared_content = self._apply_keyword_prefix(content)
         security_fields = self._build_security_fields()
 
         def _post_payload(payload: Dict[str, Any]) -> bool:
             request_payload = dict(payload)
             request_payload.update(security_fields)
-            logger.debug(f"飞书请求 URL: {self._feishu_url}")
-            logger.debug(f"飞书请求 payload 长度: {len(prepared_content)} 字符")
-
-            response = requests.post(
-                self._feishu_url,
-                json=request_payload,
-                timeout=timeout_seconds or 30,
-                verify=self._webhook_verify_ssl
-            )
-
-            logger.debug(f"飞书响应状态码: {response.status_code}")
-            logger.debug(f"飞书响应内容: {response.text}")
-
-            if response.status_code == 200:
-                result = response.json()
-                code = result.get('code') if 'code' in result else result.get('StatusCode')
-                if code == 0:
-                    logger.info("飞书消息发送成功")
-                    return True
-                else:
-                    error_msg = result.get('msg') or result.get('StatusMessage', '未知错误')
-                    error_code = result.get('code') or result.get('StatusCode', 'N/A')
-                    logger.error(f"飞书返回错误 [code={error_code}]: {error_msg}")
-                    logger.error(f"完整响应: {result}")
-                    return False
-            else:
-                logger.error(f"飞书请求失败: HTTP {response.status_code}")
-                logger.error(f"响应内容: {response.text}")
+            try:
+                response = requests.post(
+                    self._feishu_url,
+                    json=request_payload,
+                    timeout=timeout_seconds or _WEBHOOK_SEND_TIMEOUT_SECONDS,
+                    verify=self._webhook_verify_ssl,
+                )
+            except (requests.exceptions.ConnectionError,
+                     requests.exceptions.Timeout,
+                     requests.exceptions.RequestException) as e:
+                logger.error("飞书 Webhook 网络请求异常: %s", e)
                 return False
+            if response.status_code == 200:
+                try:
+                    result = response.json()
+                except (ValueError, AttributeError):
+                    logger.error("飞书 Webhook 返回非 JSON 响应: %s", response.text[:200])
+                    return False
+                if not isinstance(result, dict):
+                    logger.error("飞书 Webhook 返回非预期格式: %s", type(result).__name__)
+                    return False
+                code = result.get("code") if "code" in result else result.get("StatusCode")
+                if code == 0:
+                    logger.info("飞书 Webhook 消息发送成功")
+                    return True
+                logger.error(
+                    "飞书 Webhook 返回错误 [code=%s]: %s",
+                    code,
+                    result.get("msg") or result.get("StatusMessage", "未知错误"),
+                )
+                return False
+            logger.error("飞书 Webhook 请求失败: HTTP %d", response.status_code)
+            return False
 
-        # 1) 优先使用交互卡片（支持 Markdown 渲染）
-        card_payload = {
-            "msg_type": "interactive",
-            "card": {
-                "config": {"wide_screen_mode": True},
-                "header": {
-                    "title": {
-                        "tag": "plain_text",
-                        "content": "股票智能分析报告"
-                    }
-                },
-                "elements": [
-                    {
-                        "tag": "div",
-                        "text": {
-                            "tag": "lark_md",
-                            "content": prepared_content
-                        }
-                    }
-                ]
-            }
-        }
+        card_payload = {"msg_type": "interactive", "card": self._build_card_body(prepared_content)}
 
         if _post_payload(card_payload):
             return True
 
-        # 2) 回退为普通文本消息
         text_payload = {
             "msg_type": "text",
-            "content": {
-                "text": prepared_content
-            }
+            "content": {"text": prepared_content},
         }
-
         return _post_payload(text_payload)

src/services/notification_diagnostics.py

@@ -8,6 +8,7 @@ from typing import List, Literal, Optional, Sequence, Tuple
 
 from src.config import Config
 from src.notification import ChannelDetector, NotificationChannel, NotificationService
+from src.notification_contracts import FEISHU_APP_BOT_ENV_GROUP, FEISHU_WEBHOOK_ENV_GROUP
 from src.notification_noise import (
     NOTIFICATION_SEVERITIES,
     P4_NOISE_ENV_KEYS,
@@ -87,8 +88,9 @@ CHANNEL_SPECS: Tuple[NotificationChannelSpec, ...] = (
         channel=NotificationChannel.FEISHU.value,
         display_name=ChannelDetector.get_channel_name(NotificationChannel.FEISHU),
         kind="configured",
-        minimal_keys=("FEISHU_WEBHOOK_URL",),
-        advanced_keys=("FEISHU_WEBHOOK_SECRET", "FEISHU_WEBHOOK_KEYWORD"),
+        minimal_keys=FEISHU_WEBHOOK_ENV_GROUP,
+        alternative_minimal_keys=(FEISHU_APP_BOT_ENV_GROUP,),
+        advanced_keys=("FEISHU_WEBHOOK_SECRET", "FEISHU_WEBHOOK_KEYWORD", "FEISHU_RECEIVE_ID_TYPE", "FEISHU_DOMAIN"),
     ),
     NotificationChannelSpec(
         channel=NotificationChannel.TELEGRAM.value,

src/services/system_config_service.py

@@ -44,6 +44,12 @@ from src.core.config_registry import (
 )
 from src.llm.errors import call_litellm_with_param_recovery
 from src.llm.generation_params import apply_litellm_generation_params
+from src.notification_contracts import (
+    FEISHU_APP_BOT_ENV_GROUP,
+    FEISHU_WEBHOOK_ENV_GROUP,
+    is_feishu_app_bot_env_configured,
+    is_feishu_static_env_configured,
+)
 from src.notification_noise import validate_notification_timezone
 from src.notification_sender.gotify_sender import resolve_gotify_message_endpoint
 from src.notification_sender.ntfy_sender import resolve_ntfy_endpoint
@@ -134,6 +140,11 @@ class SystemConfigService:
         "FEISHU_WEBHOOK_SECRET": ("feishu_webhook_secret", "string"),
         "FEISHU_WEBHOOK_KEYWORD": ("feishu_webhook_keyword", "string"),
         "FEISHU_MAX_BYTES": ("feishu_max_bytes", "int"),
+        "FEISHU_APP_ID": ("feishu_app_id", "string"),
+        "FEISHU_APP_SECRET": ("feishu_app_secret", "string"),
+        "FEISHU_CHAT_ID": ("feishu_chat_id", "string"),
+        "FEISHU_RECEIVE_ID_TYPE": ("feishu_receive_id_type", "string"),
+        "FEISHU_DOMAIN": ("feishu_domain", "string"),
         "TELEGRAM_BOT_TOKEN": ("telegram_bot_token", "string"),
         "TELEGRAM_CHAT_ID": ("telegram_chat_id", "string"),
         "TELEGRAM_MESSAGE_THREAD_ID": ("telegram_message_thread_id", "string"),
@@ -167,7 +178,7 @@ class SystemConfigService:
     }
     _NOTIFICATION_REQUIRED_KEY_GROUPS: Dict[str, Tuple[Tuple[str, ...], ...]] = {
         "wechat": (("WECHAT_WEBHOOK_URL",),),
-        "feishu": (("FEISHU_WEBHOOK_URL",),),
+        "feishu": (FEISHU_WEBHOOK_ENV_GROUP, FEISHU_APP_BOT_ENV_GROUP),
         "telegram": (("TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID"),),
         "email": (("EMAIL_SENDER", "EMAIL_PASSWORD"),),
         "pushover": (("PUSHOVER_USER_KEY", "PUSHOVER_API_TOKEN"),),
@@ -182,7 +193,7 @@ class SystemConfigService:
     }
     _NOTIFICATION_TEST_TARGET_KEYS: Dict[str, Tuple[str, ...]] = {
         "wechat": ("WECHAT_WEBHOOK_URL",),
-        "feishu": ("FEISHU_WEBHOOK_URL",),
+        "feishu": FEISHU_WEBHOOK_ENV_GROUP + FEISHU_APP_BOT_ENV_GROUP,
         "telegram": ("TELEGRAM_BOT_TOKEN",),
         "email": ("EMAIL_RECEIVERS", "EMAIL_SENDER"),
         "pushover": ("PUSHOVER_USER_KEY",),
@@ -2076,7 +2087,14 @@ class SystemConfigService:
                 return []
             missing_by_group.append(missing)
 
-        return missing_by_group[0] if missing_by_group else []
+        if not missing_by_group:
+            return []
+        ranked_groups = []
+        for group, missing in zip(groups, missing_by_group):
+            present_count = len(group) - len(missing)
+            ranked_groups.append((len(missing), -present_count, missing))
+        ranked_groups.sort(key=lambda item: (item[0], item[1]))
+        return ranked_groups[0][2]
 
     @staticmethod
     def _get_invalid_notification_test_config_message(
@@ -2711,7 +2729,8 @@ class SystemConfigService:
 
     def _build_setup_notification_check(self, effective_map: Dict[str, str]) -> Dict[str, Any]:
         configured = (
-            self._has_any_config_value(effective_map, ("WECHAT_WEBHOOK_URL", "FEISHU_WEBHOOK_URL", "DISCORD_WEBHOOK_URL"))
+            self._has_any_config_value(effective_map, ("WECHAT_WEBHOOK_URL", "DISCORD_WEBHOOK_URL"))
+            or is_feishu_static_env_configured(effective_map)
             or (
                 self._has_any_config_value(effective_map, ("TELEGRAM_BOT_TOKEN",))
                 and self._has_any_config_value(effective_map, ("TELEGRAM_CHAT_ID",))
@@ -2749,11 +2768,6 @@ class SystemConfigService:
             )
             or self._has_valid_ntfy_endpoint(effective_map)
             or self._has_valid_gotify_config(effective_map)
-            or (
-                parse_env_bool(effective_map.get("FEISHU_STREAM_ENABLED"), default=False)
-                and self._has_any_config_value(effective_map, ("FEISHU_APP_ID",))
-                and self._has_any_config_value(effective_map, ("FEISHU_APP_SECRET",))
-            )
         )
         if configured:
             return self._setup_check(
@@ -3338,15 +3352,15 @@ class SystemConfigService:
             "FEISHU_WEBHOOK_KEYWORD",
             "FEISHU_STREAM_ENABLED",
             "FEISHU_FOLDER_TOKEN",
+            "FEISHU_CHAT_ID",
         }
         has_feishu_app_id = bool((effective_map.get("FEISHU_APP_ID") or "").strip())
         has_feishu_app_secret = bool((effective_map.get("FEISHU_APP_SECRET") or "").strip())
+        has_feishu_app_credentials_complete = has_feishu_app_id and has_feishu_app_secret
         has_feishu_app_credentials = has_feishu_app_id or has_feishu_app_secret
-        has_feishu_webhook = bool((effective_map.get("FEISHU_WEBHOOK_URL") or "").strip())
         has_feishu_folder_token = bool((effective_map.get("FEISHU_FOLDER_TOKEN") or "").strip())
         has_feishu_full_cloud_doc_credentials = (
-            has_feishu_app_id
-            and has_feishu_app_secret
+            has_feishu_app_credentials_complete
             and has_feishu_folder_token
         )
         # Match runtime semantics: Config.from_env only enables stream mode
@@ -3357,25 +3371,33 @@ class SystemConfigService:
             .lower()
             == "true"
         )
+        has_feishu_stream_route = feishu_stream_enabled and has_feishu_app_credentials_complete
+        has_feishu_app_bot_route = is_feishu_app_bot_env_configured(effective_map)
         if (
             has_feishu_app_credentials
             and not has_feishu_full_cloud_doc_credentials
-            and not has_feishu_webhook
-            and not (feishu_stream_enabled and has_feishu_app_id and has_feishu_app_secret)
+            and not is_feishu_static_env_configured(effective_map)
+            and not has_feishu_stream_route
+            and not has_feishu_app_bot_route
             and (updated_keys & feishu_relevant_keys)
         ):
             issues.append(
                 {
-                    "key": "FEISHU_WEBHOOK_URL",
+                    "key": "FEISHU_CHAT_ID",
                     "code": "feishu_mode_mismatch",
                     "message": (
-                        "仅配置 FEISHU_APP_ID / FEISHU_APP_SECRET 不会开启飞书群 Webhook 推送；"
-                        "如需通知推送请填写 FEISHU_WEBHOOK_URL，若要使用应用机器人请同时开启 "
-                        "FEISHU_STREAM_ENABLED 并完成应用发布与权限配置。"
+                        "仅配置 FEISHU_APP_ID / FEISHU_APP_SECRET 不会开启飞书静态通知；"
+                        "App Bot 主动推送需要同时配置 FEISHU_CHAT_ID，"
+                        "Webhook 推送请填写 FEISHU_WEBHOOK_URL；"
+                        "事件订阅请使用 FEISHU_STREAM_ENABLED=true 并完成应用发布与权限配置。"
                     ),
                     "severity": "warning",
-                    "expected": "FEISHU_WEBHOOK_URL or FEISHU_STREAM_ENABLED=true",
-                    "actual": "app credentials only",
+                    "expected": (
+                        "static notification: FEISHU_WEBHOOK_URL or "
+                        "FEISHU_APP_ID + FEISHU_APP_SECRET + FEISHU_CHAT_ID; "
+                        "event subscription: FEISHU_STREAM_ENABLED=true"
+                    ),
+                    "actual": "app credentials without notification target",
                 }
             )
 

tests/test_config_validate_structured.py

@@ -524,6 +524,19 @@ class TestValidateStructuredNotification:
         warn = [i for i in issues if i.severity == "warning"]
         assert not any("FEISHU_APP_ID / FEISHU_APP_SECRET" in i.message for i in warn)
 
+    def test_feishu_app_bot_triad_without_webhook_no_mode_warning(self):
+        cfg = _make_config(
+            wechat_webhook_url=None,
+            feishu_app_id="cli_xxx",
+            feishu_app_secret="secret_xxx",
+            feishu_chat_id="oc_xxx",
+            feishu_webhook_url=None,
+            feishu_stream_enabled=False,
+        )
+        issues = cfg.validate_structured()
+        warn = [i for i in issues if i.severity == "warning"]
+        assert not any("FEISHU_APP_ID / FEISHU_APP_SECRET" in i.message for i in warn)
+
     def test_invalid_notification_noise_config_reports_errors(self):
         cfg = _make_config(
             notification_quiet_hours="9:00-18:00",

tests/test_daily_analysis_workflow_notification_env.py

@@ -68,6 +68,17 @@ def test_daily_analysis_maps_p6_channel_env_keys() -> None:
         assert key in env
 
 
+def test_daily_analysis_feishu_status_accepts_webhook_or_app_bot_triad() -> None:
+    status_line = next(
+        line
+        for line in WORKFLOW_PATH.read_text(encoding="utf-8").splitlines()
+        if 'echo "  飞书:' in line
+    )
+
+    for key in ("FEISHU_WEBHOOK_URL", "FEISHU_APP_ID", "FEISHU_APP_SECRET", "FEISHU_CHAT_ID"):
+        assert key in status_line
+
+
 def test_daily_analysis_keeps_deferred_behavior_switches_unmapped() -> None:
     env = _load_daily_analysis_env()
 

tests/test_notification_sender.py

@@ -14,6 +14,7 @@ import sys
 import unittest
 from email.header import decode_header, make_header
 from email.utils import parseaddr
+from types import SimpleNamespace
 from unittest import mock
 from typing import Optional
 
@@ -57,6 +58,29 @@ def _response(status_code: int, json_body: Optional[dict] = None):
     return resp
 
 
+def _sdk_response(success: bool, *, code: int = 0, msg: str = "ok", log_id: str = "log-id"):
+    resp = mock.MagicMock()
+    resp.success.return_value = success
+    resp.code = code
+    resp.msg = msg
+    resp.get_log_id.return_value = log_id
+    return resp
+
+
+def _fake_feishu_client(*side_effects):
+    create = mock.Mock()
+    if side_effects:
+        create.side_effect = list(side_effects)
+    client = SimpleNamespace(
+        im=SimpleNamespace(
+            v1=SimpleNamespace(
+                message=SimpleNamespace(create=create)
+            )
+        )
+    )
+    return client, create
+
+
 class TestDiscordSender(unittest.TestCase):
     """Unit tests for DiscordSender."""
 
@@ -258,6 +282,314 @@ class TestFeishuSender(unittest.TestCase):
         self.assertFalse(result)
         mock_post.assert_not_called()
 
+    # ------------------------------------------------------------------
+    # App Bot mode tests
+    # ------------------------------------------------------------------
+
+    def test_app_bot_returns_false_when_no_app_credentials(self):
+        """send_to_feishu returns False when app credentials are missing."""
+        cfg = _config(feishu_chat_id="oc_chat")
+        sender = FeishuSender(cfg)
+        self.assertFalse(sender.send_to_feishu("hello"))
+
+    def test_app_bot_returns_false_when_no_chat_id(self):
+        """send_to_feishu returns False when feishu_chat_id is missing."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+        )
+        sender = FeishuSender(cfg)
+        self.assertFalse(sender.send_to_feishu("hello"))
+
+    def test_app_bot_success_via_card(self):
+        """send_to_feishu sends an interactive card via App Bot on success."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        dummy_client = object()
+        with mock.patch.object(FeishuSender, "_ensure_app_client", return_value=dummy_client), \
+             mock.patch.object(FeishuSender, "_app_send_raw", return_value=True) as mock_raw:
+            result = sender.send_to_feishu("**hello** world")
+
+        self.assertTrue(result)
+        mock_raw.assert_called_once()
+        self.assertIs(mock_raw.call_args[0][0], dummy_client)
+        # call_args[0] = (client, msg_type, content_json)
+        msg_type = mock_raw.call_args[0][1]
+        content_json = mock_raw.call_args[0][2]
+        self.assertEqual(msg_type, "interactive")
+        self.assertIn("**hello**", content_json)
+
+    def test_app_bot_card_fallback_to_text_on_formatted_content(self):
+        """App Bot falls back to text when interactive card fails."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        with mock.patch.object(FeishuSender, "_ensure_app_client", return_value=object()), \
+             mock.patch.object(FeishuSender, "_app_send_raw", side_effect=[False, True]) as mock_raw:
+            result = sender.send_to_feishu("hello world")
+
+        self.assertTrue(result)
+        self.assertEqual(mock_raw.call_count, 2)
+        # call_args_list[0][0] = (client, msg_type, content_json)
+        self.assertEqual(mock_raw.call_args_list[0][0][1], "interactive")
+        self.assertEqual(mock_raw.call_args_list[1][0][1], "text")
+
+    def test_app_bot_card_first_success_no_fallback(self):
+        """App Bot sends interactive card successfully and does not try text."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        with mock.patch.object(FeishuSender, "_ensure_app_client", return_value=object()), \
+             mock.patch.object(FeishuSender, "_app_send_raw", return_value=True) as mock_raw:
+            result = sender.send_to_feishu("**bold** text")
+
+        self.assertTrue(result)
+        mock_raw.assert_called_once()
+        # call_args_list[0][0][1] = msg_type, [0][0][2] = content_json
+        self.assertEqual(mock_raw.call_args_list[0][0][1], "interactive")
+        self.assertIn("**bold**", mock_raw.call_args_list[0][0][2])
+
+    @mock.patch("src.notification_sender.feishu_sender.requests.post")
+    @mock.patch.object(FeishuSender, "_app_send_raw", return_value=True)
+    def test_webhook_takes_precedence_over_app_bot(self, mock_app_raw, mock_webhook_post):
+        """When both webhook URL and App Bot credentials are configured, webhook is used."""
+        mock_webhook_post.return_value = _response(200, {"code": 0})
+        cfg = _config(
+            feishu_webhook_url="https://feishu.example/hook",
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        result = sender.send_to_feishu("hello")
+
+        self.assertTrue(result)
+        mock_webhook_post.assert_called_once()
+        mock_app_raw.assert_not_called()
+
+    @mock.patch("src.notification_sender.feishu_sender.requests.post")
+    def test_webhook_does_not_require_sdk_when_app_bot_is_also_configured(self, mock_webhook_post):
+        """Webhook precedence keeps SDK absence from breaking existing delivery."""
+        mock_webhook_post.return_value = _response(200, {"code": 0})
+        cfg = _config(
+            feishu_webhook_url="https://feishu.example/hook",
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+
+        with mock.patch("src.notification_sender.feishu_sender.FEISHU_SDK_AVAILABLE", False), \
+             mock.patch.object(FeishuSender, "_ensure_app_client", side_effect=AssertionError("SDK should not be used")):
+            result = sender.send_to_feishu("hello")
+
+        self.assertTrue(result)
+        mock_webhook_post.assert_called_once()
+
+    def test_app_bot_missing_sdk_logs_standard_requirements_install(self):
+        """App Bot SDK absence fails closed with the standard project install hint."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+
+        with mock.patch("src.notification_sender.feishu_sender.FEISHU_SDK_AVAILABLE", False), \
+             self.assertLogs("src.notification_sender.feishu_sender", level="WARNING") as logs:
+            result = sender.send_to_feishu("hello")
+
+        self.assertFalse(result)
+        install_hints = [
+            line
+            for line in logs.output
+            if "pip install -r requirements.txt" in line
+        ]
+        self.assertEqual(install_hints, logs.output)
+        self.assertEqual(len(install_hints), 1)
+
+    def test_app_bot_chunking_long_content(self):
+        """Long content is chunked for App Bot."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+            feishu_max_bytes=200,
+        )
+        sender = FeishuSender(cfg)
+
+        with mock.patch.object(FeishuSender, "_ensure_app_client", return_value=object()), \
+             mock.patch.object(FeishuSender, "_app_send_raw", return_value=False) as mock_raw:
+            result = sender.send_to_feishu("A" * 500)
+
+        self.assertFalse(result)  # All chunks fail
+        self.assertGreater(mock_raw.call_count, 1)
+
+    @mock.patch.object(FeishuSender, "_app_send_raw", return_value=True)
+    def test_app_bot_request_shape_interactive(self, mock_raw):
+        """_app_send_once constructs interactive card payload with lark_md."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        result = sender._app_send_once(object(), "**bold** text")
+
+        self.assertTrue(result)
+        call = mock_raw.call_args
+        self.assertEqual(call[0][1], "interactive")  # msg_type
+        card = json.loads(call[0][2])
+        self.assertEqual(card["header"]["title"]["content"], "股票智能分析报告")
+        self.assertEqual(card["elements"][0]["text"]["tag"], "lark_md")
+        self.assertIn("**bold**", card["elements"][0]["text"]["content"])
+
+    @mock.patch.object(FeishuSender, "_app_send_raw")
+    def test_app_bot_request_shape_text_fallback(self, mock_raw):
+        """_app_send_once falls back to text payload when card fails."""
+        mock_raw.side_effect = [False, True]
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        result = sender._app_send_once(object(), "plain text")
+
+        self.assertTrue(result)
+        self.assertEqual(mock_raw.call_count, 2)
+        # Second call is text fallback
+        second_call = mock_raw.call_args_list[1]
+        self.assertEqual(second_call[0][1], "text")
+        text_content = json.loads(second_call[0][2])
+        self.assertIn("plain text", text_content["text"])
+
+    @mock.patch("src.notification_sender.feishu_sender.uuid_mod.uuid4", return_value="uuid-open-id")
+    def test_app_bot_request_includes_receive_id_type(self, _mock_uuid4):
+        """_app_send_raw request builder passes receive_id_type and request body fields."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="ou_user",
+            feishu_receive_id_type="open_id",
+        )
+        sender = FeishuSender(cfg)
+        client, create = _fake_feishu_client(_sdk_response(True))
+
+        result = sender._app_send_raw(client, "text", json.dumps({"text": "hi"}))
+
+        self.assertTrue(result)
+        create.assert_called_once()
+        req = create.call_args[0][0]
+        self.assertEqual(req.receive_id_type, "open_id")
+        self.assertEqual(req.request_body.receive_id, "ou_user")
+        self.assertEqual(req.request_body.msg_type, "text")
+        self.assertEqual(json.loads(req.request_body.content), {"text": "hi"})
+        self.assertEqual(req.request_body.uuid, "uuid-open-id")
+
+    @mock.patch("src.notification_sender.feishu_sender.uuid_mod.uuid4")
+    def test_app_bot_idempotency_uuid_per_call(self, mock_uuid4):
+        """Each _app_send_raw invocation gets a fresh UUID."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        client, create = _fake_feishu_client(_sdk_response(True), _sdk_response(True))
+
+        mock_uuid4.side_effect = ["aaaa-bbbb-cccc", "dddd-eeee-ffff"]
+        sender._app_send_raw(client, "text", json.dumps({"text": "a"}))
+        sender._app_send_raw(client, "text", json.dumps({"text": "b"}))
+
+        self.assertEqual(create.call_count, 2)
+        call1_req = create.call_args_list[0][0][0]
+        call2_req = create.call_args_list[1][0][0]
+        self.assertEqual(call1_req.request_body.uuid, "aaaa-bbbb-cccc")
+        self.assertEqual(call2_req.request_body.uuid, "dddd-eeee-ffff")
+
+    @mock.patch("src.notification_sender.feishu_sender.time.sleep")
+    def test_app_bot_retries_sdk_response_failure(self, mock_sleep):
+        """_app_send_raw retries failed SDK responses and stops after success."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        client, create = _fake_feishu_client(
+            _sdk_response(False, code=999, msg="temporary"),
+            _sdk_response(True),
+        )
+
+        result = sender._app_send_raw(client, "text", json.dumps({"text": "retry"}))
+
+        self.assertTrue(result)
+        self.assertEqual(create.call_count, 2)
+        mock_sleep.assert_called_once()
+
+    @mock.patch("src.notification_sender.feishu_sender.time.sleep")
+    def test_app_bot_retries_sdk_exception(self, mock_sleep):
+        """_app_send_raw retries exceptions raised by the SDK create call."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        client, create = _fake_feishu_client(RuntimeError("network"), _sdk_response(True))
+
+        result = sender._app_send_raw(client, "text", json.dumps({"text": "retry"}))
+
+        self.assertTrue(result)
+        self.assertEqual(create.call_count, 2)
+        mock_sleep.assert_called_once()
+
+    @mock.patch("src.notification_sender.feishu_sender.time.sleep")
+    def test_app_bot_first_success_does_not_retry(self, mock_sleep):
+        """_app_send_raw does not retry after the first successful SDK response."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        client, create = _fake_feishu_client(_sdk_response(True))
+
+        result = sender._app_send_raw(client, "text", json.dumps({"text": "once"}))
+
+        self.assertTrue(result)
+        create.assert_called_once()
+        mock_sleep.assert_not_called()
+
+    @mock.patch("src.notification_sender.feishu_sender.time.sleep")
+    @mock.patch("src.notification_sender.feishu_sender.CreateMessageRequest.builder", side_effect=RuntimeError("bad builder"))
+    def test_app_bot_builder_failure_does_not_retry(self, _mock_builder, mock_sleep):
+        """Request builder failures are not treated as transient send failures."""
+        cfg = _config(
+            feishu_app_id="cli_app",
+            feishu_app_secret="secret",
+            feishu_chat_id="oc_chat",
+        )
+        sender = FeishuSender(cfg)
+        client, create = _fake_feishu_client(_sdk_response(True))
+
+        result = sender._app_send_raw(client, "text", json.dumps({"text": "bad"}))
+
+        self.assertFalse(result)
+        create.assert_not_called()
+        mock_sleep.assert_not_called()
+
 
 class TestEmailSender(unittest.TestCase):
     """Unit tests for EmailSender (config and receiver logic; send path covered via service)."""

tests/test_system_config_service.py

@@ -521,6 +521,42 @@ class SystemConfigServiceTestCase(unittest.TestCase):
         gotify_with_message = next(check for check in status["checks"] if check["key"] == "notification")
         self.assertEqual(gotify_with_message["status"], "optional")
 
+    def test_get_setup_status_accepts_feishu_app_bot_triad(self) -> None:
+        self._rewrite_env(
+            "LITELLM_MODEL=gemini/gemini-3-flash-preview",
+            "GEMINI_API_KEY=secret-key-value",
+            "STOCK_LIST=600519",
+            "FEISHU_APP_ID=cli_xxx",
+            "FEISHU_APP_SECRET=secret_xxx",
+            "FEISHU_CHAT_ID=oc_xxx",
+        )
+
+        with patch.dict(os.environ, {}, clear=True):
+            status = self.service.get_setup_status()
+
+        notification = next(check for check in status["checks"] if check["key"] == "notification")
+        self.assertEqual(notification["status"], "configured")
+
+    def test_get_setup_status_rejects_partial_feishu_app_bot_triad(self) -> None:
+        base_lines = [
+            "LITELLM_MODEL=gemini/gemini-3-flash-preview",
+            "GEMINI_API_KEY=secret-key-value",
+            "STOCK_LIST=600519",
+        ]
+        partial_cases = [
+            ("FEISHU_APP_ID=cli_xxx", "FEISHU_APP_SECRET=secret_xxx"),
+            ("FEISHU_APP_ID=cli_xxx", "FEISHU_CHAT_ID=oc_xxx"),
+            ("FEISHU_APP_SECRET=secret_xxx", "FEISHU_CHAT_ID=oc_xxx"),
+        ]
+
+        for partial in partial_cases:
+            with self.subTest(partial=partial):
+                self._rewrite_env(*base_lines, *partial)
+                with patch.dict(os.environ, {}, clear=True):
+                    status = self.service.get_setup_status()
+                notification = next(check for check in status["checks"] if check["key"] == "notification")
+                self.assertEqual(notification["status"], "optional")
+
     def test_get_setup_status_uses_runtime_env_without_reloading_singletons(self) -> None:
         self._rewrite_env("")
 
@@ -851,13 +887,16 @@ class SystemConfigServiceTestCase(unittest.TestCase):
             ]
         )
         self.assertTrue(validation["valid"])
-        self.assertTrue(
-            any(
-                issue["code"] == "feishu_mode_mismatch"
-                and issue["severity"] == "warning"
-                for issue in validation["issues"]
-            )
+        issue = next(
+            issue
+            for issue in validation["issues"]
+            if issue["code"] == "feishu_mode_mismatch"
+            and issue["severity"] == "warning"
         )
+        self.assertEqual(issue["key"], "FEISHU_CHAT_ID")
+        self.assertIn("FEISHU_CHAT_ID", issue["message"])
+        self.assertIn("static notification:", issue["expected"])
+        self.assertIn("event subscription:", issue["expected"])
 
     def test_validate_no_warning_when_feishu_cloud_doc_credentials_without_webhook(self) -> None:
         validation = self.service.validate(
@@ -1473,6 +1512,60 @@ class SystemConfigServiceTestCase(unittest.TestCase):
         self.assertEqual(payload["error_code"], "config_missing")
         self.assertIn("TELEGRAM_CHAT_ID", payload["message"])
 
+    def test_test_notification_channel_reports_nearest_feishu_app_bot_missing_key(self) -> None:
+        with self._notification_test_env():
+            payload = self.service.test_notification_channel(
+                channel="feishu",
+                items=[
+                    {"key": "FEISHU_APP_ID", "value": "cli_xxx"},
+                    {"key": "FEISHU_APP_SECRET", "value": "secret_xxx"},
+                ],
+                title="Test title",
+                content="hello",
+                timeout_seconds=3,
+            )
+
+        self.assertFalse(payload["success"])
+        self.assertEqual(payload["error_code"], "config_missing")
+        self.assertIn("FEISHU_CHAT_ID", payload["message"])
+        self.assertNotIn("FEISHU_WEBHOOK_URL", payload["message"])
+
+    def test_test_notification_channel_feishu_domain_draft_builds_isolated_config(self) -> None:
+        captured: Dict[str, Any] = {}
+
+        def fake_dispatch(**kwargs):
+            captured.update(kwargs)
+            return {
+                "success": True,
+                "message": "ok",
+                "error_code": None,
+                "stage": "notification_send",
+                "retryable": False,
+                "latency_ms": 0,
+                "attempts": [],
+            }
+
+        with self._notification_test_env(), patch.object(
+            SystemConfigService,
+            "_dispatch_notification_test",
+            side_effect=fake_dispatch,
+        ):
+            payload = self.service.test_notification_channel(
+                channel="feishu",
+                items=[
+                    {"key": "FEISHU_APP_ID", "value": "cli_xxx"},
+                    {"key": "FEISHU_APP_SECRET", "value": "secret_xxx"},
+                    {"key": "FEISHU_CHAT_ID", "value": "oc_xxx"},
+                    {"key": "FEISHU_DOMAIN", "value": "lark"},
+                ],
+                title="Test title",
+                content="hello",
+                timeout_seconds=3,
+            )
+
+        self.assertTrue(payload["success"])
+        self.assertEqual(captured["config"].feishu_domain, "lark")
+
     @patch("src.notification_sender.wechat_sender.requests.post")
     def test_test_notification_channel_skips_masked_secret_overwrite(self, mock_post) -> None:
         self._rewrite_env("WECHAT_WEBHOOK_URL=https://saved.example.com/hook?key=savedsecret")