Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -9,5 +9,12 @@ FEISHU_APP_SECRET=
BINDING_DB_PATH=data/stickerhub.db
BIND_MAGIC_TTL_SECONDS=600

# 飞书 Webhook 域名白名单(JSON 格式的字符串数组)
# 不设置:使用默认白名单 ["open.feishu.cn","open.larksuite.com"]
# 设为 []:禁用白名单校验(允许任意域名,请谨慎使用)
# 设为自定义列表:如 ["open.feishu.cn","custom.domain.com"]
# FEISHU_WEBHOOK_ALLOWED_HOSTS=["open.feishu.cn","open.larksuite.com"]
FEISHU_WEBHOOK_ALLOWED_HOSTS=

# 日志级别(DEBUG / INFO / WARNING / ERROR)
LOG_LEVEL=INFO
6 changes: 5 additions & 1 deletion .github/copilot-instructions.md
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,13 @@

## 不可破坏的业务行为

- `/bind`(无参数)应创建或复用当前平台身份,并返回魔法字符串。
- 配置 `FEISHU_APP_ID`/`FEISHU_APP_SECRET` 时,Telegram `/bind`(无参数)应先展示两种绑定方式按钮:`飞书机器人`、`飞书 Webhook`。
- 选择 `飞书机器人` 后,流程与原有魔法字符串一致:生成魔法字符串并提示在飞书机器人执行 `/bind <code>`。
- 选择 `飞书 Webhook` 后,应提示用户输入 webhook 地址并完成绑定。
- `/bind <code>` 应将当前平台账号绑定到该魔法字符串对应身份。
- 绑定冲突策略为可覆盖:以魔法字符串所属身份为高优先级。
- 飞书机器人绑定与 webhook 绑定互斥,切换方式后仅新方式生效。
- 未配置 `FEISHU_APP_ID`/`FEISHU_APP_SECRET` 时,不支持 webhook 绑定(与飞书能力关闭保持一致)。
- Telegram 单贴纸流程必须保持两步:
- 先立即转发表情。
- 再提示是否发送整包。
Expand Down
37 changes: 29 additions & 8 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@ StickerHub 是一个 Telegram 表情素材(Sticker/Image/GIF/Video)转换与

配置飞书应用后,额外支持:

- 双端绑定机制(`/bind`),将 Telegram 身份与飞书身份关联
- `/bind` 双模式绑定(飞书机器人 / 飞书自定义机器人 Webhook)
- 绑定模式二选一:切换后仅新配置生效
- 单个贴纸自动转发到飞书
- 整包发送时可选「📤 发送到飞书」,按每批 10 个并发发送
- 飞书事件接收方式:**长连接(Long Connection)**,项目不对外开放 HTTP 端口
Expand Down Expand Up @@ -54,13 +55,24 @@ FEISHU_APP_SECRET=

BINDING_DB_PATH=data/stickerhub.db
BIND_MAGIC_TTL_SECONDS=600

# 飞书 Webhook 域名白名单(JSON 格式的字符串数组)
# 不设置:使用默认白名单 ["open.feishu.cn","open.larksuite.com"]
# 设为 []:禁用白名单校验(允许任意域名,请谨慎使用)
# 设为自定义列表:如 ["open.feishu.cn","custom.domain.com"]
FEISHU_WEBHOOK_ALLOWED_HOSTS=

LOG_LEVEL=INFO
```

说明:

- `TELEGRAM_BOT_API_TOKEN`:必填,Telegram Bot API Token
- `FEISHU_APP_ID` / `FEISHU_APP_SECRET`:可选,填写后启用飞书转发和 `/bind` 功能
- `FEISHU_APP_ID` / `FEISHU_APP_SECRET`:可选,填写后启用飞书转发和 `/bind` 功能(包括 webhook 绑定所需的图片上传能力)
- `FEISHU_WEBHOOK_ALLOWED_HOSTS`:飞书 Webhook 域名白名单(JSON 格式,防止 SSRF 攻击)
- 不设置:使用默认白名单 `["open.feishu.cn", "open.larksuite.com"]`
- 设为 `[]`:禁用白名单校验(允许任意域名,**请谨慎使用**)
- 设为自定义列表:如 `["open.feishu.cn", "custom.domain.com"]`
- 飞书需在应用后台开启机器人收发消息权限(im:message)以及获取与上传图片或文件资源权限(im:resource) ![lark_permission.png](docs/lark_permission.png)
- 飞书事件添加接收消息(im.message.receive_v1)并启用长连接事件能力。 ![lark_event.png](docs/lark_event.png)

Expand Down Expand Up @@ -91,12 +103,21 @@ docker compose -f docker-compose.local.yml up -d --build
docker compose -f docker-compose.local.yml logs -f stickerhub
```

## 绑定流程(需配置飞书)
![20260213-050307.gif](../../Downloads/20260213-050307.gif)
1. 在 Telegram 或飞书任一端发送:`/bind`
2. 机器人返回一串魔法字符串
3. 在另一端发送:`/bind <魔法字符串>`
4. 绑定成功后,Telegram 发来的素材会路由到对应飞书身份
## 绑定流程(需配置飞书 App)

1. 在 Telegram 发送:`/bind`
2. 选择绑定方式:
- `飞书机器人`:走魔法字符串绑定
- `飞书 Webhook`:输入飞书自定义机器人的 webhook 地址
3. `飞书机器人`流程:
- Telegram 返回魔法字符串
- 在飞书机器人里发送 `/bind <魔法字符串>` 完成绑定
4. `飞书 Webhook`流程:
- Telegram 提示输入 webhook 地址
- 输入后立即生效,后续素材通过 webhook 发送
5. 两种绑定方式互斥,切换后仅新方式生效

> 注意:未配置 `FEISHU_APP_ID` / `FEISHU_APP_SECRET` 时,不支持飞书 webhook 绑定与飞书转发。

## 已知限制

Expand Down
77 changes: 73 additions & 4 deletions src/stickerhub/adapters/feishu_sender.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
import json
import logging
from typing import Literal

import httpx

from stickerhub.core.models import StickerAsset
from stickerhub.utils.url_masking import mask_url

logger = logging.getLogger(__name__)

Expand All @@ -18,18 +20,29 @@ def __init__(
self._app_secret = app_secret
self._base_url = "https://open.feishu.cn/open-apis"

async def send(self, asset: StickerAsset, target_user_id: str) -> None:
async def send(
self, asset: StickerAsset, target_mode: Literal["bot", "webhook"], target: str
) -> None:
# 避免在日志中暴露 webhook URL 中的敏感 token
safe_target = target if target_mode == "bot" else mask_url(target)
logger.debug(
"准备发送图片到飞书: receive_id=%s file=%s mime=%s size=%s",
target_user_id,
"准备发送图片到飞书: mode=%s target=%s file=%s mime=%s size=%s",
target_mode,
safe_target,
asset.file_name,
Comment on lines 28 to 32
Copy link

Copilot AI Feb 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里的 debug 日志会在 webhook 模式下把完整 webhook_url 打到日志里(target=%s)。Webhook URL 通常包含敏感 token,属于凭据泄露风险;建议对 webhook 目标做脱敏(如仅保留 host + 末尾几位),或在 webhook 模式下完全不要记录 target。

Copilot uses AI. Check for mistakes.
asset.mime_type,
len(asset.content),
)
async with httpx.AsyncClient(timeout=30.0) as client:
token = await self._get_tenant_token(client)
image_key = await self._upload_image(client, token, asset)
await self._send_image_message(client, token, image_key, target_user_id)
if target_mode == "bot":
await self._send_image_message(client, token, image_key, target)
return
if target_mode == "webhook":
await self._send_webhook_image(client, image_key=image_key, webhook_url=target)
return
raise RuntimeError(f"不支持的飞书目标类型: {target_mode}")

async def send_text(
self,
Expand Down Expand Up @@ -62,6 +75,18 @@ async def send_text(
raise RuntimeError(f"发送飞书文本消息失败: {payload}")
logger.info("飞书文本消息已发送: receive_id=%s", receive_id)

async def send_webhook_text(self, text: str, webhook_url: str) -> None:
async with httpx.AsyncClient(timeout=30.0) as client:
await self._send_webhook_message(
client,
webhook_url=webhook_url,
message={
"msg_type": "text",
"content": {"text": text},
},
)
logger.info("飞书 webhook 文本消息已发送")

async def _get_tenant_token(self, client: httpx.AsyncClient) -> str:
response = await client.post(
f"{self._base_url}/auth/v3/tenant_access_token/internal",
Expand Down Expand Up @@ -129,3 +154,47 @@ async def _send_image_message(
raise RuntimeError(f"发送飞书消息失败: {payload}")

logger.info("素材已发送到飞书: receive_id=%s", receive_id)

async def _send_webhook_image(
self,
client: httpx.AsyncClient,
image_key: str,
webhook_url: str,
) -> None:
await self._send_webhook_message(
client,
webhook_url=webhook_url,
message={
"msg_type": "image",
"content": {"image_key": image_key},
},
)
logger.info("素材已发送到飞书 webhook")

async def _send_webhook_message(
self,
client: httpx.AsyncClient,
webhook_url: str,
message: dict[str, object],
) -> None:
response = await client.post(
webhook_url,
headers={"Content-Type": "application/json"},
json=message,
)
try:
payload = response.json()
except json.JSONDecodeError:
payload = {"raw_body": response.text}

if response.status_code != 200:
raise RuntimeError(
"发送飞书 webhook 消息失败: " f"status={response.status_code} payload={payload}"
)

status_code = payload.get("StatusCode")
code = payload.get("code")
status_ok = status_code in (None, 0, "0")
code_ok = code in (None, 0, "0")
if not status_ok or not code_ok:
raise RuntimeError(f"发送飞书 webhook 消息失败: {payload}")
Loading