From 1641553d3113d3b24cc9bbe1cfbbd6b18d0b70d1 Mon Sep 17 00:00:00 2001 From: marsal Date: Fri, 10 Jul 2026 12:18:54 +0800 Subject: [PATCH] feat(send): async acceptance and send_result event --- channel-service/app/api/wechat_personal.py | 20 +++- channel-service/app/services/send_service.py | 115 ++++++++++++++++--- channel-service/tests/test_api.py | 10 +- channel-service/tests/test_send_async.py | 62 ++++++++++ 4 files changed, 185 insertions(+), 22 deletions(-) create mode 100644 channel-service/tests/test_send_async.py diff --git a/channel-service/app/api/wechat_personal.py b/channel-service/app/api/wechat_personal.py index 3be3eec..2fefe15 100644 --- a/channel-service/app/api/wechat_personal.py +++ b/channel-service/app/api/wechat_personal.py @@ -25,12 +25,16 @@ class SendTextRequest(BaseModel): account_id: str conversation_id: str = Field(..., description="目标会话/群 ID") text: str = Field(..., min_length=1, description="消息内容") + record_id: str = Field(..., description="mci-server 发送记录 ID") + tenant_code: str = Field("default", description="业务系统租户编码") class SendImageRequest(BaseModel): account_id: str conversation_id: str image_url: str + record_id: str = Field(..., description="mci-server 发送记录 ID") + tenant_code: str = Field("default", description="业务系统租户编码") @router.post("/login/start") @@ -55,9 +59,21 @@ def expire_login(account_id: str, request: LoginConfirmRequest): @router.post("/send/text") def send_text(request: SendTextRequest): - return send_service.send_text(request.account_id, request.conversation_id, request.text) + return send_service.send_text( + request.account_id, + request.conversation_id, + request.text, + request.record_id, + request.tenant_code, + ) @router.post("/send/image") def send_image(request: SendImageRequest): - return send_service.send_image(request.account_id, request.conversation_id, request.image_url) + return send_service.send_image( + request.account_id, + request.conversation_id, + request.image_url, + request.record_id, + request.tenant_code, + ) diff --git a/channel-service/app/services/send_service.py b/channel-service/app/services/send_service.py index 4bc7447..0c93560 100644 --- a/channel-service/app/services/send_service.py +++ b/channel-service/app/services/send_service.py @@ -1,33 +1,116 @@ from __future__ import annotations +import uuid +from typing import Optional + from app.clients.pywechat_client import ClientRegistry, OnlineStatus, registry +from app.models.event import EventType, OutboundEvent +from app.repositories.event_store import EventStore, event_store class SendService: - def __init__(self, registry: ClientRegistry): + def __init__(self, registry: ClientRegistry, store: EventStore): self.registry = registry + self.store = store - def send_text(self, account_id: str, conversation_id: str, text: str) -> dict: + def send_text( + self, + account_id: str, + conversation_id: str, + text: str, + record_id: str, + tenant_code: str = "default", + ) -> dict: client = self.registry.get(account_id) if client is None: - return {"success": False, "error": f"account {account_id} not registered"} + return self._fail(account_id, record_id, tenant_code, conversation_id, f"account {account_id} not registered") if client.state.online_status != OnlineStatus.ONLINE: - return {"success": False, "error": "account not online"} - return client.send_text(conversation_id, text) + return self._fail(account_id, record_id, tenant_code, conversation_id, "account not online") + if client.state.risk_level >= 80: + return self._fail( + account_id, + record_id, + tenant_code, + conversation_id, + "account risk level too high", + error_code="RISK_CONTROL_BLOCKED", + ) - def send_image(self, account_id: str, conversation_id: str, image_url: str) -> dict: - # Placeholder for image sending via pywechat. + message_id = f"mock-msg-{uuid.uuid4().hex[:16]}" + channel_result = client.send_text(conversation_id, text) + self._emit_send_result(account_id, record_id, tenant_code, conversation_id, True, message_id, channel_result) + return {"accepted": True, "message_id": message_id} + + def send_image( + self, + account_id: str, + conversation_id: str, + image_url: str, + record_id: str, + tenant_code: str = "default", + ) -> dict: client = self.registry.get(account_id) if client is None: - return {"success": False, "error": f"account {account_id} not registered"} + return self._fail(account_id, record_id, tenant_code, conversation_id, f"account {account_id} not registered") if client.state.online_status != OnlineStatus.ONLINE: - return {"success": False, "error": "account not online"} - return { - "success": True, - "message_id": f"mock-img-{image_url.split('/')[-1]}", - "conversation_id": conversation_id, - "type": "image", - } + return self._fail(account_id, record_id, tenant_code, conversation_id, "account not online") + + message_id = f"mock-img-{image_url.split('/')[-1]}" + self._emit_send_result( + account_id, + record_id, + tenant_code, + conversation_id, + True, + message_id, + {"image_url": image_url}, + ) + return {"accepted": True, "message_id": message_id} + + def _fail( + self, + account_id: str, + record_id: str, + tenant_code: str, + conversation_id: str, + error: str, + error_code: str = "SEND_FAILED", + ) -> dict: + self._emit_send_result( + account_id, + record_id, + tenant_code, + conversation_id, + False, + None, + {"error": error, "error_code": error_code}, + ) + return {"accepted": False, "error": error} + + def _emit_send_result( + self, + account_id: str, + record_id: str, + tenant_code: str, + conversation_id: str, + success: bool, + message_id: Optional[str], + channel_result: dict, + ) -> None: + event = OutboundEvent( + event_type=EventType.SEND_RESULT, + account_id=account_id, + tenant_code=tenant_code, + data={ + "record_id": record_id, + "conversation_id": conversation_id, + "success": success, + "message_id": message_id, + "channel_result": channel_result, + "error": channel_result.get("error"), + }, + ) + self.store.save(event) -send_service = SendService(registry) +send_service = SendService(registry, event_store) diff --git a/channel-service/tests/test_api.py b/channel-service/tests/test_api.py index 325f09e..e828965 100644 --- a/channel-service/tests/test_api.py +++ b/channel-service/tests/test_api.py @@ -40,10 +40,11 @@ def test_send_after_login(): r = client.post("/v1/wechat-personal/send/text", json={ "account_id": "bot-002", "conversation_id": "room-001", - "text": "hello" + "text": "hello", + "record_id": "rec-001", }) assert r.status_code == 200 - assert r.json()["success"] is True + assert r.json()["accepted"] is True assert r.json()["message_id"] @@ -51,10 +52,11 @@ def test_send_without_login_fails(): r = client.post("/v1/wechat-personal/send/text", json={ "account_id": "bot-offline", "conversation_id": "room-001", - "text": "hello" + "text": "hello", + "record_id": "rec-offline", }) assert r.status_code == 200 - assert r.json()["success"] is False + assert r.json()["accepted"] is False def test_heartbeat_and_monitor(): diff --git a/channel-service/tests/test_send_async.py b/channel-service/tests/test_send_async.py new file mode 100644 index 0000000..fbc8ac8 --- /dev/null +++ b/channel-service/tests/test_send_async.py @@ -0,0 +1,62 @@ +from fastapi.testclient import TestClient + +from app.main import app +from app.models.event import EventType +from app.repositories.event_store import event_store + +client = TestClient(app) + + +def _truncate_store() -> None: + """Mark all pending events as delivered and remove old events. + + Tests share the global SQLite-backed event store, so we clean it up + before each test to avoid cross-test contamination. + """ + for event in event_store.fetch_pending(limit=10000): + event_store.mark_delivered(event.event_id) + event_store.cleanup(0) + + +def test_send_text_creates_send_result_event(): + _truncate_store() + client.post("/v1/wechat-personal/login/start", json={"account_id": "send-bot", "tenant_code": "rescue"}) + client.post("/v1/wechat-personal/login/send-bot/confirm", json={"tenant_code": "rescue"}) + + r = client.post("/v1/wechat-personal/send/text", json={ + "account_id": "send-bot", + "conversation_id": "room-001", + "text": "hello", + "record_id": "rec-123", + "tenant_code": "rescue", + }) + assert r.status_code == 200 + assert r.json()["accepted"] is True + + pending = event_store.fetch_pending(limit=100) + assert any(e.event_type == EventType.SEND_RESULT and e.data.get("record_id") == "rec-123" for e in pending) + + +def test_send_text_blocked_by_risk_control(): + _truncate_store() + client.post("/v1/wechat-personal/login/start", json={"account_id": "risk-bot", "tenant_code": "rescue"}) + client.post("/v1/wechat-personal/login/risk-bot/confirm", json={"tenant_code": "rescue"}) + client.post("/v1/heartbeat", json={"account_id": "risk-bot", "tenant_code": "rescue", "risk_level": 90}) + + r = client.post("/v1/wechat-personal/send/text", json={ + "account_id": "risk-bot", + "conversation_id": "room-001", + "text": "hello", + "record_id": "rec-risk", + "tenant_code": "rescue", + }) + assert r.status_code == 200 + assert r.json()["accepted"] is False + + pending = event_store.fetch_pending(limit=100) + event = next( + (e for e in pending if e.event_type == EventType.SEND_RESULT and e.data.get("record_id") == "rec-risk"), + None, + ) + assert event is not None + assert event.data["channel_result"]["error_code"] == "RISK_CONTROL_BLOCKED"