Files
daily_stock_analysis/src/notification_contracts.py
Delicious233 3471afbd98 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>
2026-06-05 08:56:42 +08:00

63 lines
2.1 KiB
Python

# -*- 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)