fix: resolve Feishu cross-app notification routing bug
Implement per-app open_id storage via user_feishu_open_ids table with union_id-based cross-app user identification. WS handlers now auto-capture open_id+union_id and resolve/associate user accounts. Schedule notifications route through the correct bot's open_id instead of always falling back to 苹果. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -194,6 +194,44 @@ def lookup_user_by_email(email: str) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
def lookup_union_id_by_open_id(open_id: str) -> Optional[str]:
|
||||
"""通过 open_id 查询飞书用户的 union_id(跨应用唯一标识)。
|
||||
|
||||
同一个用户的 union_id 在所有应用下相同,可用于跨应用关联。
|
||||
|
||||
Args:
|
||||
open_id: 飞书用户的 open_id(必须在当前苹果应用下有权限访问)
|
||||
|
||||
Returns:
|
||||
union_id 字符串,未找到返回 None
|
||||
"""
|
||||
token = _get_tenant_access_token()
|
||||
if not token:
|
||||
return None
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=10) as client:
|
||||
resp = client.get(
|
||||
f"https://open.feishu.cn/open-apis/contact/v3/users/{open_id}",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
result = resp.json()
|
||||
if resp.is_success and result.get("code") == 0:
|
||||
user_data = result.get("data", {}).get("user", {})
|
||||
union_id = user_data.get("union_id")
|
||||
if union_id:
|
||||
logger.info("飞书 union_id 查询成功: open_id=%s union_id=%s", open_id[:20], union_id)
|
||||
return union_id
|
||||
logger.info("飞书用户未找到 union_id: open_id=%s", open_id[:20])
|
||||
return None
|
||||
else:
|
||||
logger.warning("飞书 union_id 查询失败: code=%s msg=%s", result.get("code"), result.get("msg"))
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.warning("飞书 union_id 查询异常: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def get_verification_token() -> str:
|
||||
"""获取飞书应用的 Verification Token(用于验证事件回调)。"""
|
||||
return settings.FEISHU_VERIFICATION_TOKEN
|
||||
|
||||
Reference in New Issue
Block a user