125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
"""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
|
|
|
|
def set_login_status(self, status: LoginStatus, error_message: Optional[str] = None) -> None:
|
|
old = self.state.login_status
|
|
self.state.login_status = status
|
|
if status == LoginStatus.SUCCESS:
|
|
self.state.online_status = OnlineStatus.ONLINE
|
|
self.state.last_heartbeat_at = time.time()
|
|
elif status in (LoginStatus.EXPIRED, LoginStatus.FAILED):
|
|
self.state.online_status = OnlineStatus.OFFLINE
|
|
if error_message:
|
|
self.state.error_message = error_message
|
|
|
|
def set_online_status(self, status: OnlineStatus) -> None:
|
|
self.state.online_status = status
|
|
|
|
|
|
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()
|