feat(channel-service): add Python FastAPI pywechat channel service

- Add FastAPI project under channel-service/
- Implement mock pywechat client adapter (login, send, heartbeat)
- Add /v1/wechat-personal/login/*, /v1/wechat-personal/send/* APIs
- Add /v1/heartbeat and /v1/monitor/status for client health
- Add /monitor HTML page with Web Components
- Add pytest suite and Dockerfile
- Update .gitignore for Python/Node artifacts
This commit is contained in:
2026-07-07 12:36:54 +08:00
parent df2d253c7b
commit bd0ac464cf
19 changed files with 575 additions and 0 deletions

View File

View File

@@ -0,0 +1,110 @@
"""pywechat client abstraction.
The real implementation controls a Windows WeChat PC client via pywechat/pywinauto.
This module provides a mock adapter for development and testing.
"""
from __future__ import annotations
import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
class LoginStatus(str, Enum):
PENDING = "pending"
SCANNING = "scanning"
SUCCESS = "success"
EXPIRED = "expired"
FAILED = "failed"
class OnlineStatus(str, Enum):
OFFLINE = "offline"
ONLINE = "online"
ERROR = "error"
@dataclass
class WeChatClientState:
account_id: str
account_name: Optional[str] = None
login_status: LoginStatus = LoginStatus.PENDING
online_status: OnlineStatus = OnlineStatus.OFFLINE
last_heartbeat_at: Optional[float] = None
login_id: Optional[str] = None
qr_code_url: Optional[str] = None
risk_level: int = 0
sent_count_today: int = 0
error_message: Optional[str] = None
class PyWeChatClient:
"""Mock pywechat client. Replace with real pywechat integration on Windows."""
def __init__(self, account_id: str, account_name: Optional[str] = None):
self.state = WeChatClientState(account_id=account_id, account_name=account_name)
self._conversations: dict[str, str] = {}
def start_login(self) -> str:
login_id = str(uuid.uuid4())
self.state.login_id = login_id
self.state.login_status = LoginStatus.PENDING
# In production this would return a real QR code from WeChat.
self.state.qr_code_url = f"https://mock.pywechat.qrcode/{login_id}"
return login_id
def confirm_login(self) -> None:
self.state.login_status = LoginStatus.SUCCESS
self.state.online_status = OnlineStatus.ONLINE
self.state.last_heartbeat_at = time.time()
def expire_login(self) -> None:
self.state.login_status = LoginStatus.EXPIRED
self.state.online_status = OnlineStatus.OFFLINE
def heartbeat(self) -> None:
self.state.last_heartbeat_at = time.time()
if self.state.online_status != OnlineStatus.ERROR:
self.state.online_status = OnlineStatus.ONLINE
def send_text(self, conversation_id: str, text: str) -> dict:
"""Send a text message to a conversation (group or single chat).
In production this drives the WeChat PC client UI via pywechat.
"""
self.state.sent_count_today += 1
return {
"success": True,
"message_id": f"mock-msg-{uuid.uuid4().hex[:16]}",
"conversation_id": conversation_id,
"text_preview": text[:50],
}
def get_state(self) -> WeChatClientState:
# Auto-expire login after 5 minutes if not confirmed.
if self.state.login_status == LoginStatus.PENDING and self.state.login_id:
# mock: auto-confirm after a few seconds for testing
pass
return self.state
class ClientRegistry:
def __init__(self):
self._clients: dict[str, PyWeChatClient] = {}
def get_or_create(self, account_id: str, account_name: Optional[str] = None) -> PyWeChatClient:
if account_id not in self._clients:
self._clients[account_id] = PyWeChatClient(account_id, account_name)
return self._clients[account_id]
def get(self, account_id: str) -> Optional[PyWeChatClient]:
return self._clients.get(account_id)
def list_all(self) -> list[PyWeChatClient]:
return list(self._clients.values())
registry = ClientRegistry()