feat(integrations): 微信里发一句话,变成一条任务或知识
CI / backend (push) Canceled after 0s

一条通道一张令牌,机器人把消息 POST 到 /integrations/clawbot/{token},
正文按前缀落到任务或知识草稿:没有前缀就是任务,「知识:」「笔记:」记成知识,
「帮助」只回用法。首行当标题、其余当正文,`#标签` 摘出来挂到事项上。
同一个 msg_id 只处理一次——微信会重推,重推不能变成两条任务。

为什么不塞进 system_setting 的 JSON:一个工作区可能挂多个通道,还得能单独停用。
令牌库里只存 sha256,只在创建/重新生成时显示一次;收消息时还不知道是哪个工作区,
所以 bot_channel 的 RLS 额外放行 token_hash = current_setting('app.bot_token_hash'),
读到通道再 bind_tenant 进上下文,按通道创建者记账。

- 后端:models_integrations.py(bot_channel / bot_message)、迁移 0012(RLS)、
  api/integrations.py(收消息 + 管理接口)、错误码 9002/9003
- Web:/w/:wsId/integrations「接入」页——建通道、复制回调地址与令牌、最近消息、
  停用/换令牌/删除,只有 owner/admin 能改
- 测试:test_clawbot.py 10 例;web/e2e/integrations.mjs 7 项(冒充 clawbot 真推一条)

验证:pytest 150 passed,ruff check/format 全绿,tsc 0,vite build 通过,e2e 7 项全绿。
This commit is contained in:
Steven
2026-11-05 22:04:34 +08:00
parent 33a3fc03e0
commit 90b99f8253
16 changed files with 1606 additions and 4 deletions
@@ -0,0 +1,108 @@
"""微信(clawbot)接入通道
Revision ID: 0012
Revises: 0011
Create Date: 2026-11-05
两张表:
- `bot_channel`:一个工作区可以挂多个机器人通道,令牌只存 sha256。
- `bot_message`:进来的每条消息留痕,附带它变成了哪条任务/知识。
`bot_channel` 的 RLS 多放了一条口子:`token_hash = app_bot_token_hash()`。
原因很实在——**收到 webhook 时我们还不知道它是哪个工作区的**,而候选令牌
必须读过表才知道。没有这条口子,就只能要么关掉 RLS(不行),要么把令牌
放一张不挂租户的表里(等于把钥匙串放在门外)。现在的做法是:
请求先把 `app.bot_token_hash` 设成本次令牌的哈希,策略只允许命中那一行,
其余行照旧看不见。
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
from app.roles import APP_ROLE
revision: str = "0012"
down_revision: str = "0011"
branch_labels: Sequence[str] | None = None
depends_on: Sequence[str] | None = None
CREATE_STATEMENTS = (
"""
CREATE TABLE bot_channel (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
workspace_id uuid NOT NULL REFERENCES workspace(id) ON DELETE CASCADE,
kind varchar(32) NOT NULL DEFAULT 'wechat_clawbot',
name varchar(64) NOT NULL DEFAULT '',
token_hash varchar(64) NOT NULL,
status varchar(16) NOT NULL DEFAULT 'active',
created_by uuid REFERENCES app_user(id) ON DELETE SET NULL,
last_seen_at timestamptz,
last_error text,
version integer NOT NULL DEFAULT 1,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
)
""",
"""
CREATE TABLE bot_message (
id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
workspace_id uuid NOT NULL REFERENCES workspace(id) ON DELETE CASCADE,
channel_id uuid NOT NULL REFERENCES bot_channel(id) ON DELETE CASCADE,
external_id varchar(128),
sender varchar(128),
text text NOT NULL DEFAULT '',
action varchar(24) NOT NULL DEFAULT 'ignored',
item_id uuid REFERENCES item(id) ON DELETE SET NULL,
reply text NOT NULL DEFAULT '',
raw jsonb NOT NULL DEFAULT '{}'::jsonb,
version integer NOT NULL DEFAULT 1,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
)
""",
)
INDEX_STATEMENTS = (
"CREATE INDEX ix_bot_channel_ws ON bot_channel (workspace_id)",
"CREATE UNIQUE INDEX uq_bot_channel_token ON bot_channel (token_hash)",
"CREATE INDEX ix_bot_message_channel ON bot_message (channel_id, created_at)",
"CREATE UNIQUE INDEX uq_bot_message_external ON bot_message (channel_id, external_id) "
"WHERE external_id IS NOT NULL",
)
POLICY_SQL = (
"ALTER TABLE bot_channel ENABLE ROW LEVEL SECURITY",
"CREATE POLICY bot_channel_isolation ON bot_channel "
"USING (workspace_id = app_ws_id() OR token_hash = current_setting('app.bot_token_hash', true)) "
"WITH CHECK (workspace_id = app_ws_id())",
"ALTER TABLE bot_message ENABLE ROW LEVEL SECURITY",
"CREATE POLICY bot_message_isolation ON bot_message "
"USING (workspace_id = app_ws_id()) WITH CHECK (workspace_id = app_ws_id())",
f"GRANT SELECT, INSERT, UPDATE, DELETE ON bot_channel TO {APP_ROLE}",
f"GRANT SELECT, INSERT, UPDATE, DELETE ON bot_message TO {APP_ROLE}",
)
COMMENT_SQL = (
"COMMENT ON TABLE bot_channel IS '外部接入通道(微信 clawbot 等),令牌只存 sha256'",
"COMMENT ON TABLE bot_message IS '外部接入收到并处理过的消息,用于支线排查与去重'",
)
def upgrade() -> None:
for statement in CREATE_STATEMENTS:
op.execute(statement)
for statement in INDEX_STATEMENTS:
op.execute(statement)
for statement in POLICY_SQL:
op.execute(statement)
for statement in COMMENT_SQL:
op.execute(statement)
def downgrade() -> None:
op.execute("DROP TABLE IF EXISTS bot_message")
op.execute("DROP TABLE IF EXISTS bot_channel")
+562
View File
@@ -0,0 +1,562 @@
"""外部接入:微信(clawbot)发来的消息变成任务 / 知识。
## 一条消息是怎么变成事项的
微信里发一句话给机器人,clawbot 把消息 POST 到
POST /api/v1/integrations/clawbot/{token}
令牌在 URL 里,**不带登录态**:机器人不会登我们的账号,它只会照着配置好的
地址发。令牌是 32 字节随机串,库里只存 sha256。
正文按前缀决定去哪个池子:
知识:xxx / 笔记:xxx / note: xxx → 知识草稿
任务:xxx / todo: xxx → 任务(默认也是任务)
?、帮助、help → 回一段用法说明
首行是标题,其余行是正文。首行里的 `#标签` 会被摘出来挂到事项上
(标签是知识的主要检索入口,手机上敲#比打开网页加标签省事)。
## 为什么每条消息都留痕
微信会重推(网络抖动、机器人重启)。没有留痕就只能靠猜"那条任务到底进没进"。
`bot_message.external_id` 上有个部分唯一索引:同一个外部消息 id 第二次到达时
直接回上次的答案,不会变成两条任务。
## 安全边界
- 令牌错 → 9002(对外一律"通道不存在",不区分"错令牌"和"这个工作区没接")。
- 通道被停用 → 9003。
- 消息里不认人:机器人只知道"这是这个工作区的一条消息",落库按通道创建者记账,
不会因为消息内容去猜是谁发的。
"""
from __future__ import annotations
import hashlib
import secrets
import uuid
from datetime import datetime, timezone
from typing import Annotated, Any
from fastapi import APIRouter, Body, Query, Request
from fastapi.responses import PlainTextResponse
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from ..db import bind_tenant
from ..deps import AdminDep, SessionDep
from ..errors import AppError, ErrorCode
from ..models import Member, Workspace
from ..models_integrations import BotChannel, BotMessage
from ..models_knowledge import KnowledgeDoc
from ..models_task import Item
from ..schemas import ok
from .items import _log, _set_tags
webhook_router = APIRouter(prefix="/integrations", tags=["外部接入"])
admin_router = APIRouter(prefix="/workspaces", tags=["外部接入"])
# 标题在 item.title 上是 varchar(200),留点余量给"…"之类的后缀。
MAX_TITLE = 180
MAX_BODY = 20000
KNOWLEDGE_PREFIXES = ("知识", "笔记", "note", "knowledge", "kb")
TASK_PREFIXES = ("任务", "待办", "todo", "task")
HELP_PREFIXES = ("?", "?", "帮助", "help", "用法")
HELP_TEXT = (
"发什么都能用,写法:\n"
"· 直接说 → 记成任务\n"
"· 知识:内容 → 记成知识草稿(也可以写「笔记:」)\n"
"· 任务:内容 → 明确记成任务\n"
"首行当标题,后面几行当正文;标题里带 #标签 会自动加上。"
)
def hash_token(token: str) -> str:
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def new_token() -> str:
return secrets.token_urlsafe(32)
def _pick(data: dict[str, Any], *keys: str) -> str | None:
"""从任意形状的 payload 里挖字段(clawbot 的消息壳子没定死,宽松一点)。"""
for key in keys:
value = data.get(key)
if isinstance(value, str) and value.strip():
return value
if isinstance(value, dict):
nested = _pick(value, "text", "content", "body", "message")
if nested:
return nested
return None
def extract_text(payload: dict[str, Any]) -> str:
text = _pick(
payload,
"text",
"content",
"msg",
"message",
"query",
"body",
"data",
"payload",
)
return (text or "").strip()
def extract_external_id(payload: dict[str, Any]) -> str | None:
value = _pick(payload, "msg_id", "message_id", "msgid", "id", "uuid", "event_id")
return value[:128] if value else None
def extract_sender(payload: dict[str, Any]) -> str | None:
value = _pick(payload, "from", "sender", "from_user", "user", "uid", "sender_id")
return value[:128] if value else None
def split_command(text: str) -> tuple[str, str]:
"""返回 (kind, 去掉前缀后的正文)。"""
head, _, rest = text.partition("\n")
probe = head.strip().lower()
for prefix in HELP_PREFIXES:
if probe == prefix or probe.startswith(f"{prefix}:") or probe.startswith(f"{prefix}:"):
return "help", rest
for prefix in KNOWLEDGE_PREFIXES:
for sep in (":", ":", " ", "\u3000"):
if probe.startswith(f"{prefix}{sep}"):
offset = len(prefix) + len(sep)
return "knowledge", (head.strip()[offset:] + ("\n" + rest if rest else "")).strip()
for prefix in TASK_PREFIXES:
for sep in (":", ":", " ", "\u3000"):
if probe.startswith(f"{prefix}{sep}"):
offset = len(prefix) + len(sep)
return "task", (head.strip()[offset:] + ("\n" + rest if rest else "")).strip()
return "task", text
def split_tags(body: str) -> tuple[str, list[str]]:
"""把 `#标签` 摘出来。只认行首和空白后的 #,免得把 "C#" 吃掉。"""
tags: list[str] = []
kept: list[str] = []
for line in body.splitlines():
words = line.split(" ")
rest: list[str] = []
for word in words:
if word.startswith("#") and len(word) > 1 and not word[1:].isdigit():
name = word[1:].strip(",,。.;;")
if name and name.lower() not in {t.lower() for t in tags}:
tags.append(name)
continue
rest.append(word)
kept.append(" ".join(rest).rstrip())
return "\n".join(kept).strip(), tags
def split_title(body: str) -> tuple[str, str]:
lines = body.splitlines()
if not lines:
return "", ""
title = lines[0].strip()
rest = "\n".join(lines[1:]).strip()
return title[:MAX_TITLE], rest[:MAX_BODY]
async def _creator_id(session: SessionDep, channel: BotChannel) -> uuid.UUID:
"""消息记在谁名下:通道创建者;人没了就退回工作区 owner。"""
if channel.created_by is not None:
row = await session.execute(
select(Member.id).where(
Member.workspace_id == channel.workspace_id,
Member.user_id == channel.created_by,
Member.status == "active",
)
)
if row.first() is not None:
return channel.created_by
workspace = await session.get(Workspace, channel.workspace_id)
if workspace is None: # 理论上不可能:FK 是 CASCADE
raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在")
return workspace.owner_id
async def _create_task(
session: SessionDep,
*,
ws_id: uuid.UUID,
creator_id: uuid.UUID,
title: str,
body: str,
tags: list[str],
) -> Item:
item = Item(
workspace_id=ws_id,
kind="task",
title=title,
body_md=body,
status="inbox",
priority=0,
creator_id=creator_id,
)
session.add(item)
await session.flush()
if tags:
await _set_tags(session, ws_id, item, tags)
await _log(session, item, creator_id, "created", {"via": "wechat_clawbot"})
return item
async def _create_knowledge(
session: SessionDep,
*,
ws_id: uuid.UUID,
creator_id: uuid.UUID,
title: str,
body: str,
tags: list[str],
) -> Item:
item = Item(
workspace_id=ws_id,
kind="knowledge",
title=title,
body_md=body,
status="draft",
creator_id=creator_id,
)
session.add(item)
await session.flush()
doc = KnowledgeDoc(
item_id=item.id,
workspace_id=ws_id,
author_id=creator_id,
source_item_id=None,
source_kind="manual",
doc_type="note",
status_hint="draft",
generated_by="human",
)
session.add(doc)
await session.flush()
if tags:
await _set_tags(session, ws_id, item, tags)
await _log(session, item, creator_id, "knowledge_created", {"via": "wechat_clawbot"})
return item
def _link(request: Request, ws_id: uuid.UUID, item: Item) -> str:
base = str(request.base_url).rstrip("/")
# base_url 形如 http://host:8200/api/v1/,而页面在站点根上
base = base.split("/api/")[0]
if item.kind == "knowledge":
return f"{base}/w/{ws_id}/knowledge/{item.id}"
return f"{base}/w/{ws_id}?open={item.id}"
async def _channel_by_token(session: SessionDep, token: str) -> BotChannel:
from sqlalchemy import text as sql_text
# 策略要靠这个值才放行那一行,必须先设。
await session.execute(
sql_text("SELECT set_config('app.bot_token_hash', :h, true)"), {"h": hash_token(token)}
)
row = (
await session.execute(select(BotChannel).where(BotChannel.token_hash == hash_token(token)))
).scalar_one_or_none()
if row is None:
raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在或令牌已失效")
return row
@webhook_router.get("/clawbot/{token}")
async def clawbot_verify(
token: str, session: SessionDep, echostr: str | None = Query(default=None)
):
"""clawbot 后台通常先 GET 一次确认地址是活的:有 echostr 就原样回。"""
channel = await _channel_by_token(session, token)
if channel.status != "active":
raise AppError(ErrorCode.BOT_DISABLED, "这个通道已停用")
if echostr is not None:
return PlainTextResponse(echostr)
return ok({"channel": channel.name, "status": channel.status})
@webhook_router.post("/clawbot/{token}")
async def clawbot_message(
token: str,
request: Request,
session: SessionDep,
payload: Annotated[dict[str, Any] | None, Body()] = None,
):
channel = await _channel_by_token(session, token)
if channel.status != "active":
raise AppError(ErrorCode.BOT_DISABLED, "这个通道已停用")
raw: dict[str, Any] = payload if isinstance(payload, dict) else {}
text = extract_text(raw)
external_id = extract_external_id(raw)
sender = extract_sender(raw)
await bind_tenant(session, channel.workspace_id, channel.created_by)
# 重推:同一个外部消息 id 再来一次,回上次那句,不再建第二条。
if external_id:
previous = (
await session.execute(
select(BotMessage).where(
BotMessage.channel_id == channel.id,
BotMessage.external_id == external_id,
)
)
).scalar_one_or_none()
if previous is not None:
channel.last_seen_at = datetime.now(timezone.utc)
return ok({"reply": previous.reply, "action": previous.action, "duplicate": True})
message = BotMessage(
workspace_id=channel.workspace_id,
channel_id=channel.id,
external_id=external_id,
sender=sender,
text=text[:MAX_BODY],
action="ignored",
reply="",
raw=raw if isinstance(raw, dict) else {"value": raw},
)
if not text:
message.action = "ignored"
message.reply = "没收到正文,只收到一个空壳。"
session.add(message)
return ok({"reply": message.reply, "action": "ignored"})
kind, body = split_command(text)
creator_id = await _creator_id(session, channel)
if kind == "help":
message.action = "help"
message.reply = HELP_TEXT
else:
body, tags = split_tags(body)
title, rest = split_title(body)
if not title:
message.action = "ignored"
message.reply = "只有标签没有内容,这条没记。"
elif kind == "knowledge":
item = await _create_knowledge(
session,
ws_id=channel.workspace_id,
creator_id=creator_id,
title=title,
body=rest,
tags=tags,
)
message.action = "knowledge"
message.item_id = item.id
link = _link(request, channel.workspace_id, item)
message.reply = f"已记成知识草稿:《{item.title}》。去核对:{link}"
else:
item = await _create_task(
session,
ws_id=channel.workspace_id,
creator_id=creator_id,
title=title,
body=rest,
tags=tags,
)
message.action = "task"
message.item_id = item.id
message.reply = (
f"已记成任务:《{item.title}》。打开:{_link(request, channel.workspace_id, item)}"
)
channel.last_seen_at = datetime.now(timezone.utc)
channel.last_error = None
try:
# 用 SAVEPOINT 兜住唯一索引冲突:另一个并发请求刚插了同一条消息,
# 这里只要认输回同一句就行,不能把整笔事务打挂。
async with session.begin_nested():
session.add(message)
await session.flush()
except IntegrityError:
if external_id:
previous = (
await session.execute(
select(BotMessage).where(
BotMessage.channel_id == channel.id,
BotMessage.external_id == external_id,
)
)
).scalar_one_or_none()
if previous is not None:
return ok({"reply": previous.reply, "action": previous.action, "duplicate": True})
raise AppError(ErrorCode.BOT_TOKEN_INVALID, "同一条消息重复到达,已忽略") from None
return ok({"reply": message.reply, "action": message.action, "duplicate": False})
# ── 工作区里的管理接口(登录 + admin 以上) ─────────────────────────────
def _channel_out(channel: BotChannel) -> dict[str, Any]:
return {
"id": str(channel.id),
"kind": channel.kind,
"name": channel.name,
"status": channel.status,
"created_by": str(channel.created_by) if channel.created_by else None,
"last_seen_at": channel.last_seen_at,
"last_error": channel.last_error,
"created_at": channel.created_at,
}
def _webhook_url(request: Request, token: str) -> str:
base = str(request.base_url).rstrip("/")
return f"{base}/integrations/clawbot/{token}"
@admin_router.get("/{ws_id}/integrations")
async def list_integrations(ws_id: uuid.UUID, ctx: AdminDep, session: SessionDep):
rows = (
(
await session.execute(
select(BotChannel)
.where(BotChannel.workspace_id == ws_id)
.order_by(BotChannel.created_at.desc())
)
)
.scalars()
.all()
)
return ok({"items": [_channel_out(row) for row in rows]})
@admin_router.post("/{ws_id}/integrations")
async def create_integration(
ws_id: uuid.UUID,
ctx: AdminDep,
session: SessionDep,
request: Request,
body: Annotated[dict[str, Any] | None, Body()] = None,
):
name = str((body or {}).get("name") or "").strip()[:64] or "微信"
token = new_token()
channel = BotChannel(
workspace_id=ws_id,
kind="wechat_clawbot",
name=name,
token_hash=hash_token(token),
status="active",
created_by=ctx.user.id,
)
session.add(channel)
await session.flush()
return ok(
{
"channel": _channel_out(channel),
# 令牌只在这里出现一次,之后只能"重新生成"。
"token": token,
"webhook_url": _webhook_url(request, token),
}
)
@admin_router.patch("/{ws_id}/integrations/{channel_id}")
async def update_integration(
ws_id: uuid.UUID,
channel_id: uuid.UUID,
ctx: AdminDep,
session: SessionDep,
body: Annotated[dict[str, Any] | None, Body()] = None,
):
channel = await session.get(BotChannel, channel_id)
if channel is None or channel.workspace_id != ws_id:
raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在")
data = body or {}
if isinstance(data.get("name"), str):
channel.name = data["name"].strip()[:64]
if isinstance(data.get("status"), str):
if data["status"] not in ("active", "disabled"):
raise AppError(ErrorCode.VALIDATION_ERROR, "状态只能是 active 或 disabled")
channel.status = data["status"]
return ok({"channel": _channel_out(channel)})
@admin_router.post("/{ws_id}/integrations/{channel_id}/rotate")
async def rotate_integration(
ws_id: uuid.UUID,
channel_id: uuid.UUID,
ctx: AdminDep,
session: SessionDep,
request: Request,
):
channel = await session.get(BotChannel, channel_id)
if channel is None or channel.workspace_id != ws_id:
raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在")
token = new_token()
channel.token_hash = hash_token(token)
return ok(
{
"channel": _channel_out(channel),
"token": token,
"webhook_url": _webhook_url(request, token),
}
)
@admin_router.delete("/{ws_id}/integrations/{channel_id}")
async def delete_integration(
ws_id: uuid.UUID, channel_id: uuid.UUID, ctx: AdminDep, session: SessionDep
):
channel = await session.get(BotChannel, channel_id)
if channel is None or channel.workspace_id != ws_id:
raise AppError(ErrorCode.BOT_TOKEN_INVALID, "接入通道不存在")
await session.delete(channel)
return ok({"deleted": True})
@admin_router.get("/{ws_id}/integrations/{channel_id}/messages")
async def list_messages(
ws_id: uuid.UUID,
channel_id: uuid.UUID,
ctx: AdminDep,
session: SessionDep,
limit: int = Query(default=20, ge=1, le=100),
):
rows = (
(
await session.execute(
select(BotMessage)
.where(BotMessage.workspace_id == ws_id, BotMessage.channel_id == channel_id)
.order_by(BotMessage.created_at.desc())
.limit(limit)
)
)
.scalars()
.all()
)
return ok(
{
"items": [
{
"id": str(row.id),
"text": row.text,
"action": row.action,
"item_id": str(row.item_id) if row.item_id else None,
"reply": row.reply,
"sender": row.sender,
"created_at": row.created_at,
}
for row in rows
]
}
)
__all__ = ["webhook_router", "admin_router"]
+6
View File
@@ -79,6 +79,10 @@ class ErrorCode(IntEnum):
# 9xxx 客户端发布(OTA)
APP_RELEASE_NOT_FOUND = 9001
# 9xxx 外部接入(微信 clawbot 等)
BOT_TOKEN_INVALID = 9002
BOT_DISABLED = 9003
# 默认 HTTP 状态:业务码自己带语义,HTTP 只是传输层
_HTTP_STATUS: dict[ErrorCode, int] = {
@@ -135,6 +139,8 @@ _HTTP_STATUS: dict[ErrorCode, int] = {
ErrorCode.ATTACHMENT_FORBIDDEN: 403,
ErrorCode.CONFIRM_REQUIRED: 409,
ErrorCode.APP_RELEASE_NOT_FOUND: 404,
ErrorCode.BOT_TOKEN_INVALID: 404,
ErrorCode.BOT_DISABLED: 403,
}
+12 -1
View File
@@ -16,7 +16,16 @@ from starlette.responses import Response
from starlette.staticfiles import StaticFiles
from starlette.types import Scope
from .api import app_release, attachments, auth, items, knowledge, tickets, workspaces
from .api import (
app_release,
attachments,
auth,
integrations,
items,
knowledge,
tickets,
workspaces,
)
from .api import settings as settings_api
from .config import get_settings
from .db import SessionFactory, current_app_role, engine
@@ -118,6 +127,8 @@ app.include_router(knowledge.router, prefix=settings.api_prefix)
app.include_router(attachments.router, prefix=settings.api_prefix)
app.include_router(app_release.router, prefix=settings.api_prefix)
app.include_router(settings_api.router, prefix=settings.api_prefix)
app.include_router(integrations.admin_router, prefix=settings.api_prefix)
app.include_router(integrations.webhook_router, prefix=settings.api_prefix)
@app.get("/health", tags=["运维"], summary="存活探针")
+89
View File
@@ -0,0 +1,89 @@
"""外部接入:微信(clawbot)这一类的机器人通道。
为什么单开一张表而不是塞进 `system_setting`:
1. 一个工作区可以挂多个通道(自己的号、公司的号),也随时可能换;
塞进 setting 的 JSON 里就没法"停用这一个、保留那一个"。
2. 通道是**租户数据**:谁能往里发消息,取决于它属于哪个工作区。
租户表就得有 `workspace_id NOT NULL` + RLS(AGENTS.md 第 2.5 条)。
令牌只存 sha256:数据库被看见也拿不到能用的令牌;令牌只在创建时回显一次。
"""
from __future__ import annotations
import uuid
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Index, String, Text, text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import UUID as PgUUID
from sqlalchemy.orm import Mapped, mapped_column
from .models import Base, Timestamps, new_uuid
BOT_KINDS = ("wechat_clawbot",)
BOT_STATUSES = ("active", "disabled")
INTEGRATION_TENANT_TABLES = ("bot_channel", "bot_message")
class BotChannel(Base, Timestamps):
"""一条接入通道:微信里那个机器人 ⇄ 本工作区。"""
__tablename__ = "bot_channel"
__table_args__ = (
Index("ix_bot_channel_ws", "workspace_id"),
Index("uq_bot_channel_token", "token_hash", unique=True),
)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
kind: Mapped[str] = mapped_column(String(32), nullable=False, default="wechat_clawbot")
name: Mapped[str] = mapped_column(String(64), nullable=False, default="")
# hex(sha256(令牌))。令牌本身只在创建响应里出现一次。
token_hash: Mapped[str] = mapped_column(String(64), nullable=False)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="active")
# 消息落在谁名下:用创建这条通道的人;人没了就退回工作区 owner。
created_by: Mapped[uuid.UUID | None] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="SET NULL")
)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
last_error: Mapped[str | None] = mapped_column(Text)
class BotMessage(Base, Timestamps):
"""收到的每条消息都留痕:出问题时能回答"那条任务到底有没有进去"。"""
__tablename__ = "bot_message"
__table_args__ = (
Index("ix_bot_message_channel", "channel_id", "created_at"),
# 同一个外部消息 id 只处理一次:微信会重推,重推不能变成两条任务。
Index(
"uq_bot_message_external",
"channel_id",
"external_id",
unique=True,
postgresql_where=text("external_id IS NOT NULL"),
),
)
id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
workspace_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
)
channel_id: Mapped[uuid.UUID] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("bot_channel.id", ondelete="CASCADE"), nullable=False
)
external_id: Mapped[str | None] = mapped_column(String(128))
sender: Mapped[str | None] = mapped_column(String(128))
text: Mapped[str] = mapped_column(Text, nullable=False, default="")
# task / knowledge / help / ignored / error
action: Mapped[str] = mapped_column(String(24), nullable=False, default="ignored")
item_id: Mapped[uuid.UUID | None] = mapped_column(
PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="SET NULL")
)
reply: Mapped[str] = mapped_column(Text, nullable=False, default="")
raw: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+3 -1
View File
@@ -39,7 +39,9 @@ from app.stores import reset_store_for_tests
settings = get_settings()
DEV_CODE = settings.sms_dev_code
_TABLES = "system_setting, audit_log, invitation, member, workspace, app_user"
_TABLES = (
"bot_message, bot_channel, system_setting, audit_log, invitation, member, workspace, app_user"
)
def _run_migrations() -> None:
+199
View File
@@ -0,0 +1,199 @@
"""微信(clawbot)接入:一条消息进来,变成任务或知识。
测的是**整条链路**:令牌 → 租户上下文 → 建事项 → 留痕 → 重复推送幂等。
webhook 不带登录态,所以这里故意不用 `auth()` 头,只有管理接口才带。
"""
from __future__ import annotations
from typing import Any
from httpx import AsyncClient
from .conftest import create_workspace, login, unwrap
async def _channel(client: AsyncClient, user: dict[str, Any], ws: dict[str, Any]) -> dict:
return unwrap(
await client.post(
f"/api/v1/workspaces/{ws['id']}/integrations",
json={"name": "我的微信"},
headers=user["headers"],
)
)
async def _post(client: AsyncClient, token: str, payload: dict[str, Any]):
return await client.post(f"/api/v1/integrations/clawbot/{token}", json=payload)
async def test_plain_message_becomes_task(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信接入")
channel = await _channel(client, user, ws)
resp = await _post(
client,
channel["token"],
{"msg_id": "wx-1", "from": "steven", "text": "给客户回电话\n顺便确认下报价"},
)
data = unwrap(resp)
assert data["action"] == "task"
assert data["duplicate"] is False
assert "给客户回电话" in data["reply"]
items = unwrap(
await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"])
)["items"]
task = next(i for i in items if i["title"] == "给客户回电话")
assert task["kind"] == "task"
assert task["status"] == "inbox"
assert "顺便确认下报价" in task["body_md"]
async def test_knowledge_prefix_creates_draft_with_tags(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信知识")
channel = await _channel(client, user, ws)
data = unwrap(
await _post(
client,
channel["token"],
{"text": "知识:回滚前先停灰度 #发布 #回滚\n1. 停流量\n2. 回滚镜像"},
)
)
assert data["action"] == "knowledge"
docs = unwrap(
await client.get(
f"/api/v1/workspaces/{ws['id']}/knowledge?status=draft", headers=user["headers"]
)
)["items"]
doc = next(d for d in docs if d["title"] == "回滚前先停灰度")
assert doc["knowledge"]["status_hint"] == "draft"
assert doc["knowledge"]["doc_type"] == "note"
assert {t["name"] for t in doc["tags"]} == {"发布", "回滚"}
# 标签从标题摘出去以后,正文里不该再留 # 号
assert "#发布" not in doc["body_md"]
async def test_help_reply_creates_nothing(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信帮助")
channel = await _channel(client, user, ws)
data = unwrap(await _post(client, channel["token"], {"text": "帮助"}))
assert data["action"] == "help"
assert "记成任务" in data["reply"]
items = unwrap(
await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"])
)["items"]
assert items == []
async def test_unknown_token_is_404(client: AsyncClient) -> None:
resp = await client.post("/api/v1/integrations/clawbot/nope", json={"text": "x"})
assert resp.status_code == 404
assert resp.json()["code"] == 9002
async def test_disabled_channel_refuses_messages(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信停用")
channel = await _channel(client, user, ws)
await client.patch(
f"/api/v1/workspaces/{ws['id']}/integrations/{channel['channel']['id']}",
json={"status": "disabled"},
headers=user["headers"],
)
resp = await _post(client, channel["token"], {"text": "这条不该进去"})
assert resp.status_code == 403
assert resp.json()["code"] == 9003
async def test_same_message_id_is_idempotent(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信重推")
channel = await _channel(client, user, ws)
payload = {"msg_id": "dup-1", "text": "只该有一条"}
first = unwrap(await _post(client, channel["token"], payload))
second = unwrap(await _post(client, channel["token"], payload))
assert second["duplicate"] is True
assert second["reply"] == first["reply"]
items = unwrap(
await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"])
)["items"]
assert [i["title"] for i in items] == ["只该有一条"]
async def test_get_verify_echoes_challenge(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信校验")
channel = await _channel(client, user, ws)
resp = await client.get(f"/api/v1/integrations/clawbot/{channel['token']}?echostr=abc123")
assert resp.status_code == 200
assert resp.text == "abc123"
async def test_token_belongs_to_one_workspace(client: AsyncClient) -> None:
user = await login(client)
ws_a = await create_workspace(client, user, "A 区")
ws_b = await create_workspace(client, user, "B 区")
channel = await _channel(client, user, ws_a)
await _post(client, channel["token"], {"text": "只该进 A 区"})
items_a = unwrap(
await client.get(f"/api/v1/workspaces/{ws_a['id']}/items", headers=user["headers"])
)["items"]
items_b = unwrap(
await client.get(f"/api/v1/workspaces/{ws_b['id']}/items", headers=user["headers"])
)["items"]
assert [i["title"] for i in items_a] == ["只该进 A 区"]
assert items_b == []
async def test_rotate_invalidates_old_token(client: AsyncClient) -> None:
user = await login(client)
ws = await create_workspace(client, user, "微信换令牌")
channel = await _channel(client, user, ws)
rotated = unwrap(
await client.post(
f"/api/v1/workspaces/{ws['id']}/integrations/{channel['channel']['id']}/rotate",
headers=user["headers"],
)
)
assert rotated["token"] != channel["token"]
old = await _post(client, channel["token"], {"text": "旧令牌"})
assert old.status_code == 404
new = unwrap(await _post(client, rotated["token"], {"text": "新令牌能用"}))
assert new["action"] == "task"
async def test_member_cannot_manage_channels(client: AsyncClient) -> None:
owner = await login(client)
ws = await create_workspace(client, owner, "微信权限")
member = await login(client)
code = unwrap(
await client.post(
f"/api/v1/workspaces/{ws['id']}/invitations",
json={"role": "member"},
headers=owner["headers"],
)
)
await client.post(
f"/api/v1/workspaces/{ws['id']}/join",
json={"code": code["code"]},
headers=member["headers"],
)
resp = await client.post(
f"/api/v1/workspaces/{ws['id']}/integrations",
json={"name": "偷加的通道"},
headers=member["headers"],
)
assert resp.status_code == 403
assert resp.json()["code"] in (3004, 3003)