811 lines
28 KiB
Python
811 lines
28 KiB
Python
"""wechatauto 顶层 API —— 兼容当前微信 4.x 客户端。
|
||
|
||
实现说明
|
||
========
|
||
早期版本基于 UIAutomation（``mmui::*`` 控件树）驱动微信。当前 4.1.x 客户端
|
||
冷启动时 UIA 树只暴露 ``Qt51514QWindowIcon`` + ``MMUIRenderSubWindow*`` 空壳
|
||
（原 wxauto UI 方案因此失效）；通过热激活 Qt accessibility gate（见
|
||
:mod:`wechatauto.uia_driver`）后可物化 ``mmui::*`` 完整控件树。
|
||
|
||
本模块把 :class:`WeChat` / :class:`Chat` 的公共 API 重新实现为
|
||
「UIA 优先（:class:`wechatauto.uia_driver.WeChatUIA`）+ 坐标/OCR
|
||
（:class:`wechatauto.guia.WeChatGUI`）+ 本地数据库
|
||
（:class:`wechatauto.db.WeChatDB`）」混合技术栈，**保持方法签名不变**，
|
||
原有调用方代码无需改动即可运行。
|
||
|
||
:class:`Listener` 抽象类保留仅为向后兼容（已由 :mod:`wechatauto.db` 的
|
||
``Listener`` 取代）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import os
|
||
import re
|
||
import threading
|
||
import time
|
||
from abc import ABC, abstractmethod
|
||
from typing import (
|
||
Callable,
|
||
TYPE_CHECKING,
|
||
Union,
|
||
List,
|
||
Dict,
|
||
Literal,
|
||
Optional,
|
||
)
|
||
|
||
from wechatauto.param import WxParam, WxResponse, PROJECT_NAME
|
||
from wechatauto.logger import wxlog
|
||
from wechatauto.utils.lock import uilock
|
||
|
||
if TYPE_CHECKING:
|
||
from wechatauto.msgs.base import Message
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 兼容占位：UIA 时代的监听器抽象基类（保留导出，不再使用）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class Listener(ABC):
|
||
"""监听器抽象基类（兼容保留）。
|
||
|
||
当前版本请使用 :class:`wechatauto.db.Listener`。
|
||
"""
|
||
|
||
@abstractmethod
|
||
def _get_listen_messages(self):
|
||
...
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DB 消息 → Message 对象适配
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class _FakeRect:
|
||
"""伪矩形，供现有 Message 类计算 hash 使用。"""
|
||
|
||
def __init__(self):
|
||
self.top = self.left = self.bottom = self.right = 0
|
||
|
||
def height(self):
|
||
return self.bottom - self.top
|
||
|
||
def width(self):
|
||
return self.right - self.left
|
||
|
||
|
||
class _DBMessageControl:
|
||
"""让 DB 消息复用现有 Message 子类的轻量伪控件。
|
||
|
||
仅提供 ``Name`` / ``runtimeid`` / ``BoundingRectangle`` / ``Exists``
|
||
等只读接口；交互类操作（点击/滚动）因 DB 消息无对应控件而明确报错。
|
||
"""
|
||
|
||
def __init__(self, content: str, msg_id):
|
||
self.Name = content or ''
|
||
self.AutomationId = None
|
||
self.ClassName = "mmui::ChatTextItemView"
|
||
self.runtimeid = str(msg_id)
|
||
self._rect = _FakeRect()
|
||
|
||
@property
|
||
def BoundingRectangle(self):
|
||
return self._rect
|
||
|
||
def Exists(self, timeout=0) -> bool:
|
||
return True
|
||
|
||
def GetChildren(self):
|
||
return []
|
||
|
||
def Click(self, *args, **kwargs):
|
||
raise NotImplementedError('DB 消息不支持点击操作')
|
||
|
||
def RightClick(self, *args, **kwargs):
|
||
raise NotImplementedError('DB 消息不支持右键操作')
|
||
|
||
|
||
class _DBMessageParent:
|
||
"""Message 所需的 parent 占位（root 指向 Chat）。"""
|
||
|
||
def __init__(self, chat):
|
||
self.root = chat
|
||
self.msgbox = None
|
||
|
||
|
||
class _AllMessageChat:
|
||
"""AddListenAll 使用的轻量 Chat 占位（仅含 .who，不触发 GUI 初始化）。"""
|
||
|
||
def __init__(self, username: str):
|
||
self.who = username
|
||
self._wxid = username
|
||
|
||
|
||
def _extract_group_sender(content) -> str:
|
||
"""群消息内容形如 ``wxid_xxx:\\n正文``，提取发送者 wxid。"""
|
||
if isinstance(content, bytes):
|
||
content = content.decode('utf-8', errors='ignore')
|
||
m = re.match(r'^(wxid_[0-9a-zA-Z_]+):\s*\n', content or '')
|
||
return m.group(1) if m else ''
|
||
|
||
|
||
def _pick_msg_class(is_self: bool, mtype: Optional[str], content: str):
|
||
from wechatauto.msgs import friend as friendmsg
|
||
from wechatauto.msgs import self as selfmsg
|
||
|
||
mod = selfmsg if is_self else friendmsg
|
||
|
||
def get(name):
|
||
return getattr(mod, name)
|
||
|
||
if mtype == '文本':
|
||
return get('SelfTextMessage' if is_self else 'FriendTextMessage')
|
||
if mtype == '图片':
|
||
return get('SelfImageMessage' if is_self else 'FriendImageMessage')
|
||
if mtype == '语音':
|
||
return get('SelfVoiceMessage' if is_self else 'FriendVoiceMessage')
|
||
if mtype == '视频':
|
||
return get('SelfVideoMessage' if is_self else 'FriendVideoMessage')
|
||
if mtype == '位置':
|
||
return get('SelfLocationMessage' if is_self else 'FriendLocationMessage')
|
||
if mtype == '文件/链接/卡片':
|
||
head = (content or '')[:8]
|
||
if '[链接' in head or head.startswith('链接'):
|
||
return get('SelfLinkMessage' if is_self else 'FriendLinkMessage')
|
||
if head.startswith('文件') or '[文件' in head:
|
||
return get('SelfFileMessage' if is_self else 'FriendFileMessage')
|
||
if head.startswith('位置') or head.startswith('[位置'):
|
||
return get('SelfLocationMessage' if is_self else 'FriendLocationMessage')
|
||
if '[个人名片' in head or '[名片' in head:
|
||
return get('SelfPersonalCardMessage' if is_self else 'FriendPersonalCardMessage')
|
||
return get('SelfOtherMessage' if is_self else 'FriendOtherMessage')
|
||
if mtype == '动画表情':
|
||
return get('SelfEmojiMessage' if is_self else 'FriendEmojiMessage')
|
||
return get('SelfOtherMessage' if is_self else 'FriendOtherMessage')
|
||
|
||
|
||
def _db_row_to_message(row: dict, chat: 'Chat', self_wxid: str = None) -> 'Message':
|
||
"""把 db.py 的消息行转换为现有 Message 子类实例。
|
||
|
||
direction 判定：``sender_id == 2`` 视为自己（与 guia 发送校验一致），
|
||
也可用 self_wxid 比对兜底。
|
||
"""
|
||
from wechatauto.db import WeChatDB
|
||
from wechatauto.msgs.mattr import SystemMessage
|
||
|
||
mtype = row.get('type')
|
||
if mtype is None and row.get('local_type') is not None:
|
||
mtype = WeChatDB._msg_type_name(row.get('local_type'))
|
||
content = row.get('content') or ''
|
||
if isinstance(content, bytes):
|
||
content = WeChatDB._friendly_content(content, mtype)
|
||
sender_id = row.get('sender_id')
|
||
is_self = sender_id == 2 or bool(self_wxid and str(sender_id) == str(self_wxid))
|
||
|
||
ctrl = _DBMessageControl(content, row.get('local_id'))
|
||
parent = _DBMessageParent(chat)
|
||
|
||
if mtype == '系统消息':
|
||
msg = SystemMessage(ctrl, parent)
|
||
else:
|
||
cls = _pick_msg_class(is_self, mtype, content)
|
||
msg = cls(ctrl, parent)
|
||
|
||
# 附加 DB 元数据
|
||
msg.local_id = row.get('local_id')
|
||
msg.sort_seq = row.get('sort_seq')
|
||
msg.create_time = row.get('create_time')
|
||
msg.wxid = sender_id
|
||
msg.attr = 'self' if is_self else 'friend'
|
||
sender = _extract_group_sender(content) or getattr(chat, 'who', '')
|
||
msg.sender = sender or getattr(chat, 'who', '')
|
||
msg.sender_remark = msg.sender
|
||
return msg
|
||
|
||
|
||
class SessionItem:
|
||
"""会话列表条目（兼容 SessionElement 常用字段）。"""
|
||
|
||
def __init__(self, name: str, unread: int = 0, summary: str = '',
|
||
last_time: int = 0, username: str = ''):
|
||
self.name = name
|
||
self.unread = unread
|
||
self.summary = summary
|
||
self.last_time = last_time
|
||
self.username = username
|
||
|
||
def __repr__(self):
|
||
return f'<{PROJECT_NAME} - {self.__class__.__name__}("{self.name}")>'
|
||
|
||
|
||
def _resolve_wxid(db, name: str) -> str:
|
||
"""把会话显示名解析为数据库 wxid；文件传输助手/未知则原样返回。"""
|
||
if name in ('filehelper', '文件传输助手'):
|
||
return 'filehelper'
|
||
try:
|
||
for hit in db.search_contact(name):
|
||
if name in (hit.get('nick_name'), hit.get('remark')):
|
||
return hit['username']
|
||
except Exception:
|
||
pass
|
||
return name
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Chat
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class Chat:
|
||
"""聊天窗口实例（基于 GUI + 本地数据库）。"""
|
||
|
||
def __init__(self, who: str = None, gui=None, db=None):
|
||
from wechatauto.guia import WeChatGUI
|
||
from wechatauto.db import WeChatDB
|
||
|
||
self.who = who or ''
|
||
self._gui = gui or WeChatGUI()
|
||
self._db = db or WeChatDB()
|
||
self._wxid = _resolve_wxid(self._db, self.who)
|
||
self._last_seq: Optional[int] = None
|
||
|
||
def __repr__(self):
|
||
return f'<{PROJECT_NAME} - {self.__class__.__name__} object("{self.who}")>'
|
||
|
||
def __str__(self):
|
||
return self.who or self.nickname
|
||
|
||
def __add__(self, other):
|
||
return (self.who or '') + other
|
||
|
||
def __radd__(self, other):
|
||
return other + (self.who or '')
|
||
|
||
# -- 展示 -------------------------------------------------------------
|
||
|
||
def Show(self):
|
||
"""打开该会话的聊天窗口并置前。"""
|
||
self._gui.open_chat(self.who)
|
||
|
||
def Close(self) -> None:
|
||
"""关闭聊天（GUI 模式下无独立窗口，置前即可）。"""
|
||
self._gui.bring_to_front()
|
||
|
||
@uilock
|
||
def VoiceCall(self, who: str = None, video: bool = False) -> WxResponse:
|
||
"""发起语音/视频通话。
|
||
|
||
Args:
|
||
who: 通话对象，不指定则使用当前聊天对象
|
||
video: True 尝试视频通话（当前版本未暴露视频按钮，通常失败）
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
target = who or self.who
|
||
uia = self._gui._get_uia()
|
||
if uia is None:
|
||
return WxResponse.failure('UIA 驱动不可用，无法发起通话')
|
||
if not uia.voice_call(target, video=video):
|
||
return WxResponse.failure('通话发起失败（可能未打开会话或控件不可用）')
|
||
return WxResponse.success(f'已发起通话：{target}')
|
||
|
||
@uilock
|
||
def Poke(self, who: str = None) -> WxResponse:
|
||
"""对联系人发起「拍一拍」（右键头像 → 点击拍一拍）。
|
||
|
||
Args:
|
||
who: 拍一拍对象，不指定则使用当前聊天对象
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
target = who or self.who
|
||
uia = self._gui._get_uia()
|
||
if uia is None:
|
||
return WxResponse.failure('UIA 驱动不可用，无法发起拍一拍')
|
||
if not uia.poke(target):
|
||
return WxResponse.failure('拍一拍失败（未找到对方消息或菜单不可识别）')
|
||
return WxResponse.success(f'已对 {target} 拍一拍')
|
||
|
||
@uilock
|
||
def RecallLastMessage(self, who: str = None) -> WxResponse:
|
||
"""撤回当前会话最近一条自己发送的消息。
|
||
|
||
Args:
|
||
who: 会话对象，不指定则使用当前聊天对象
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
target = who or self.who
|
||
uia = self._gui._get_uia()
|
||
if uia is None:
|
||
return WxResponse.failure('UIA 驱动不可用，无法撤回消息')
|
||
if not uia.recall_last_message(target):
|
||
return WxResponse.failure('撤回失败（消息已过期或控件不可识别）')
|
||
return WxResponse.success(f'已撤回对 {target} 发送的最近一条消息')
|
||
|
||
@uilock
|
||
def ForwardVoiceMessage(
|
||
self,
|
||
who: str = None,
|
||
target: str = None,
|
||
save_dir: str = None,
|
||
) -> WxResponse:
|
||
"""转发语音消息（从本地媒体库提取 SILK 文件发送给目标）。
|
||
|
||
微信不支持右键直接转发语音，故实现为「找到本地语音文件 → 以文件
|
||
消息发送」。默认转发本会话最近一条语音到 target（不指定则发给
|
||
本会话对象自身）。
|
||
|
||
Args:
|
||
who: 语音所在会话，不指定则用当前会话
|
||
target: 转发目标联系人，不指定则转发给 who 本身
|
||
save_dir: 语音文件临时保存目录
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
chat = Chat(who or self.who, self._gui, self._db) if who else self
|
||
msgs = chat.GetAllMessage()
|
||
for m in msgs:
|
||
if getattr(m, 'type', None) == 'voice':
|
||
return m.forward_to(target or chat.who, save_dir=save_dir)
|
||
return WxResponse.failure(f'会话「{chat.who}」最近 50 条中没有语音消息')
|
||
|
||
# -- 信息 -------------------------------------------------------------
|
||
|
||
def ChatInfo(self) -> Dict[str, str]:
|
||
"""获取聊天窗口信息。"""
|
||
info = {'chat_name': self.who, 'chat_type': 'friend'}
|
||
if self._wxid and self._wxid.endswith('@chatroom'):
|
||
info['chat_type'] = 'group'
|
||
return info
|
||
|
||
# -- 发送 -------------------------------------------------------------
|
||
|
||
@uilock
|
||
def SendMsg(
|
||
self,
|
||
msg: str,
|
||
who: str = None,
|
||
clear: bool = True,
|
||
at: Union[str, List[str]] = None,
|
||
exact: bool = False,
|
||
) -> WxResponse:
|
||
"""发送消息。
|
||
|
||
Args:
|
||
msg: 消息内容
|
||
who: 发送对象，不指定则发送给当前聊天对象
|
||
clear: 是否发送前清空编辑框（GUI 路径恒清理）
|
||
at: @对象（支持 str 或 list）
|
||
exact: 是否精确匹配会话名
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
target = who or self.who
|
||
if at:
|
||
return self._gui.at_member(at, msg, target)
|
||
return self._gui.send_msg(msg, target)
|
||
|
||
@uilock
|
||
def SendFiles(
|
||
self,
|
||
filepath,
|
||
who=None,
|
||
exact=False
|
||
) -> WxResponse:
|
||
"""向当前聊天窗口发送文件/图片。
|
||
|
||
Args:
|
||
filepath: 文件绝对路径（str 或 list）
|
||
who: 发送对象，不指定则发送给当前聊天对象
|
||
exact: 是否精确匹配会话名
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
target = who or self.who
|
||
if isinstance(filepath, (list, tuple)):
|
||
result = None
|
||
for p in filepath:
|
||
result = self._gui.send_file(p, target)
|
||
return result or WxResponse.failure('文件列表为空')
|
||
return self._gui.send_file(filepath, target)
|
||
|
||
# -- 读取 -------------------------------------------------------------
|
||
|
||
def GetAllMessage(self) -> List['Message']:
|
||
"""获取当前聊天窗口最近 50 条消息。"""
|
||
rows = self._db.get_messages(self._wxid, limit=50)
|
||
self_wxid = self._db.get_self_info()['username']
|
||
return [_db_row_to_message(r, self, self_wxid) for r in rows]
|
||
|
||
def GetNewMessage(self) -> List['Message']:
|
||
"""获取新消息（首次调用仅建立基线，返回空列表）。"""
|
||
latest = self._db.get_messages(self._wxid, limit=1)
|
||
current = latest[0]['sort_seq'] if latest else 0
|
||
if self._last_seq is None:
|
||
self._last_seq = current
|
||
return []
|
||
if current <= self._last_seq:
|
||
return []
|
||
rows = self._db.get_new_messages(self._wxid, since_seq=self._last_seq)
|
||
self._last_seq = current
|
||
self_wxid = self._db.get_self_info()['username']
|
||
return [_db_row_to_message(r, self, self_wxid) for r in rows]
|
||
|
||
def GetMessageById(self, msg_id) -> Optional['Message']:
|
||
"""根据消息 local_id 获取消息实例。"""
|
||
try:
|
||
local_id = int(str(msg_id).replace('db-', ''))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
row = self._db.get_message_row(self._wxid, local_id)
|
||
if not row:
|
||
return None
|
||
return _db_row_to_message(row, self)
|
||
|
||
def GetMessageByHash(self, msg_hash: str) -> Optional['Message']:
|
||
"""根据消息哈希值获取消息实例。"""
|
||
if not msg_hash:
|
||
return None
|
||
self_wxid = self._db.get_self_info()['username']
|
||
for row in self._db.get_messages(self._wxid, limit=200):
|
||
m = _db_row_to_message(row, self, self_wxid)
|
||
if m.hash == msg_hash or getattr(m, 'hash_text', None) == msg_hash:
|
||
return m
|
||
return None
|
||
|
||
def GetLastMessage(self) -> Optional['Message']:
|
||
"""获取当前聊天窗口的最后一条消息。"""
|
||
rows = self._db.get_messages(self._wxid, limit=1)
|
||
if not rows:
|
||
return None
|
||
return _db_row_to_message(rows[0], self)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# WeChat
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class WeChat(Chat, Listener):
|
||
"""微信主窗口实例（兼容 API）。"""
|
||
|
||
def __init__(
|
||
self,
|
||
nickname: str = None,
|
||
start_listener: bool = False,
|
||
debug: bool = False,
|
||
**kwargs
|
||
):
|
||
from wechatauto.guia import WeChatGUI
|
||
from wechatauto.db import WeChatDB
|
||
|
||
self._gui = WeChatGUI()
|
||
self._db = WeChatDB()
|
||
info = self._db.get_self_info()
|
||
self.nickname = nickname or info.get('nick_name') or info.get('username') or ''
|
||
self.who = self.nickname
|
||
self._wxid = info.get('username') or ''
|
||
self.listen: Dict[str, tuple] = {}
|
||
self._listener = None
|
||
self._listen_wrappers: Dict[str, Callable] = {}
|
||
self._listener_is_listening = False
|
||
self._listener_stop_event = threading.Event()
|
||
self._current_chat: Optional['Chat'] = None
|
||
self._listen_all_active = False
|
||
self._listen_all_callback: Optional[Callable] = None
|
||
|
||
if start_listener:
|
||
self._listener_start()
|
||
if debug:
|
||
wxlog.set_debug(True)
|
||
wxlog.debug('Debug mode is on')
|
||
|
||
# -- 监听（基于 db.Listener）------------------------------------------
|
||
|
||
def _listener_start(self):
|
||
from wechatauto.db import Listener as DBListener
|
||
if self._listener is not None:
|
||
if self._listener._thread and self._listener._thread.is_alive():
|
||
return
|
||
self._listener = None
|
||
self._listener = DBListener(self._db, interval=WxParam.LISTEN_INTERVAL)
|
||
for name, (chat, _cb) in self.listen.items():
|
||
wrapper = self._make_listen_cb(chat, _cb)
|
||
self._listen_wrappers[name] = wrapper
|
||
self._listener.add_listener(chat._wxid, wrapper)
|
||
self._listener.start()
|
||
self._listener_is_listening = True
|
||
self._listener_stop_event.clear()
|
||
|
||
def _listener_stop(self):
|
||
if self._listener is not None:
|
||
self._listener.stop()
|
||
self._listener_is_listening = False
|
||
self._listener_stop_event.set()
|
||
|
||
def _make_listen_cb(self, chat: 'Chat', callback: Callable) -> Callable:
|
||
self_wxid = self._db.get_self_info()['username']
|
||
|
||
def _wrapper(row: dict, listener) -> None:
|
||
try:
|
||
msg = _db_row_to_message(row, chat, self_wxid)
|
||
callback(msg, chat)
|
||
except Exception:
|
||
import traceback
|
||
wxlog.debug(f'监听消息回调发生错误：{traceback.format_exc()}')
|
||
|
||
return _wrapper
|
||
|
||
def _get_listen_messages(self):
|
||
"""兼容占位：实际监听由 db.Listener 完成。"""
|
||
return
|
||
|
||
@uilock
|
||
def AddListenChat(
|
||
self,
|
||
nickname: str,
|
||
callback: Callable[['Message', 'Chat'], None],
|
||
) -> WxResponse:
|
||
"""添加监听聊天。
|
||
|
||
Args:
|
||
nickname: 要监听的聊天对象（显示名）
|
||
callback: 回调函数，参数为 (Message 对象, Chat 对象)
|
||
|
||
Returns:
|
||
Chat 对象（监听成功后返回）
|
||
"""
|
||
if not self._listener_is_listening:
|
||
wxlog.debug('检测到未开启监听器，开启监听器')
|
||
self._listener_start()
|
||
if nickname in self.listen:
|
||
return WxResponse.failure('该聊天已监听')
|
||
chat = Chat(nickname, self._gui, self._db)
|
||
if self._db.get_messages(chat._wxid, limit=1) == [] and not chat._wxid:
|
||
return WxResponse.failure('找不到聊天窗口')
|
||
self.listen[nickname] = (chat, callback)
|
||
wrapper = self._make_listen_cb(chat, callback)
|
||
self._listen_wrappers[nickname] = wrapper
|
||
if self._listener is not None:
|
||
self._listener.add_listener(chat._wxid, wrapper)
|
||
return chat
|
||
|
||
def AddListenAll(
|
||
self,
|
||
callback: Callable[['Message', 'Chat'], None],
|
||
discover: bool = True,
|
||
) -> WxResponse:
|
||
"""监听所有会话的新消息（包括好友、群聊、文件传输助手等）。
|
||
|
||
Args:
|
||
callback: 回调函数，参数为 (Message 对象, Chat-like 对象)。
|
||
Chat-like 对象的 .who 属性为会话原始 username。
|
||
discover: 为 True 时自动发现新出现的会话（如新群聊）并注册
|
||
回调，无需重复调用。默认 True。
|
||
|
||
Returns:
|
||
WxResponse
|
||
|
||
示例::
|
||
|
||
wx = WeChat()
|
||
def on_all(msg, chat):
|
||
print(f'[{chat.who}] {msg.content}')
|
||
wx.AddListenAll(on_all)
|
||
wx.StartListening()
|
||
"""
|
||
if not self._listener_is_listening:
|
||
wxlog.debug('检测到未开启监听器，开启监听器')
|
||
self._listener_start()
|
||
if getattr(self, '_listen_all_active', False):
|
||
return WxResponse.failure('已开启全局监听')
|
||
self_wxid = self._db.get_self_info()['username']
|
||
|
||
def _wrap(row: dict, listener) -> None:
|
||
try:
|
||
username = row.get('username', '')
|
||
fake_chat = _AllMessageChat(username)
|
||
msg = _db_row_to_message(row, fake_chat, self_wxid)
|
||
callback(msg, fake_chat)
|
||
except Exception:
|
||
import traceback
|
||
wxlog.debug(f'全局监听回调发生错误：{traceback.format_exc()}')
|
||
|
||
self._listen_all_callback = callback
|
||
self._listen_all_active = True
|
||
if self._listener is not None:
|
||
self._listener.add_all(_wrap, discover=discover)
|
||
return WxResponse.success('已开启全局监听')
|
||
|
||
def RemoveListenAll(self) -> WxResponse:
|
||
"""停止全局监听。"""
|
||
if not getattr(self, '_listen_all_active', False):
|
||
return WxResponse.failure('未开启全局监听')
|
||
self._listen_all_active = False
|
||
self._listen_all_callback = None
|
||
if self._listener is not None:
|
||
self._listener._discover_new = False
|
||
self._listener._all_callback = None
|
||
return WxResponse.success('已停止全局监听')
|
||
|
||
def StartListening(self) -> None:
|
||
"""启动监听。"""
|
||
self._listener_start()
|
||
|
||
def StopListening(self, remove: bool = True) -> None:
|
||
"""停止监听。
|
||
|
||
Args:
|
||
remove: 是否同时移除所有监听对象
|
||
"""
|
||
self._listener_stop()
|
||
if remove:
|
||
self.listen.clear()
|
||
self._listen_wrappers.clear()
|
||
self._listen_all_active = False
|
||
self._listen_all_callback = None
|
||
|
||
@uilock
|
||
def RemoveListenChat(
|
||
self,
|
||
nickname: str,
|
||
close_window: bool = True
|
||
) -> WxResponse:
|
||
"""移除监听聊天。
|
||
|
||
Args:
|
||
nickname: 要移除监听的聊天对象
|
||
close_window: 是否关闭聊天窗口（GUI 模式忽略）
|
||
|
||
Returns:
|
||
WxResponse
|
||
"""
|
||
if nickname not in self.listen:
|
||
return WxResponse.failure('未找到监听对象')
|
||
chat, _cb = self.listen[nickname]
|
||
if self._listener is not None:
|
||
wrapper = self._listen_wrappers.pop(nickname, None)
|
||
if wrapper is not None:
|
||
self._listener.remove_listener(chat._wxid, wrapper)
|
||
del self.listen[nickname]
|
||
return WxResponse.success()
|
||
|
||
def KeepRunning(self):
|
||
"""阻塞主线程直到手动停止监听。"""
|
||
while not self._listener_stop_event.is_set():
|
||
try:
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
wxlog.debug(f'wechatauto("{self.nickname}") shutdown')
|
||
self.StopListening(True)
|
||
break
|
||
|
||
# -- 会话 -------------------------------------------------------------
|
||
|
||
def GetSession(self) -> List['SessionItem']:
|
||
"""获取当前会话列表。"""
|
||
sessions = []
|
||
for row in self._db.get_sessions(limit=50):
|
||
username = row.get('username') or ''
|
||
name = row.get('last_sender') or username
|
||
if not name or name == username:
|
||
try:
|
||
nick = self._db.get_nickname(username)
|
||
name = nick or username
|
||
except Exception:
|
||
name = username
|
||
sessions.append(SessionItem(
|
||
name=name,
|
||
unread=row.get('unread', 0),
|
||
summary=row.get('summary', ''),
|
||
last_time=row.get('last_time', 0),
|
||
username=username,
|
||
))
|
||
return sessions
|
||
|
||
@uilock
|
||
def ChatWith(
|
||
self,
|
||
who: str,
|
||
exact: bool = True,
|
||
force: bool = False,
|
||
force_wait: Union[float, int] = 0.5
|
||
):
|
||
"""打开聊天窗口。
|
||
|
||
Args:
|
||
who: 要聊天的对象
|
||
exact: 搜索会话时是否精确匹配
|
||
force: 忽略（兼容保留）
|
||
force_wait: 忽略（兼容保留）
|
||
|
||
Returns:
|
||
str: 成功时返回会话显示名，失败返回 None
|
||
"""
|
||
chat = Chat(who, self._gui, self._db)
|
||
self._gui.open_chat(chat.who)
|
||
if self._gui.get_input_box():
|
||
self._current_chat = chat
|
||
self.who = chat.who
|
||
self._wxid = chat._wxid
|
||
return chat.who
|
||
self._gui.open_chat(chat.who)
|
||
if self._gui.get_input_box():
|
||
self._current_chat = chat
|
||
self.who = chat.who
|
||
self._wxid = chat._wxid
|
||
return chat.who
|
||
return None
|
||
|
||
# -- 消息读取（委托给当前打开的会话）----------------------------------
|
||
|
||
def _cur(self) -> 'Chat':
|
||
return self._current_chat if self._current_chat is not None else self
|
||
|
||
def GetAllMessage(self) -> List['Message']:
|
||
"""获取当前打开会话的最近 50 条消息。"""
|
||
return self._cur().GetAllMessage()
|
||
|
||
def GetNewMessage(self) -> List['Message']:
|
||
"""获取当前打开会话的新消息。"""
|
||
return self._cur().GetNewMessage()
|
||
|
||
def GetMessageById(self, msg_id) -> Optional['Message']:
|
||
"""根据消息 local_id 获取消息实例。"""
|
||
return self._cur().GetMessageById(msg_id)
|
||
|
||
def GetMessageByHash(self, msg_hash: str) -> Optional['Message']:
|
||
"""根据消息哈希值获取消息实例。"""
|
||
return self._cur().GetMessageByHash(msg_hash)
|
||
|
||
def GetLastMessage(self) -> Optional['Message']:
|
||
"""获取当前打开会话的最后一条消息。"""
|
||
return self._cur().GetLastMessage()
|
||
|
||
def GetSubWindow(self, nickname: str) -> Optional['Chat']:
|
||
"""获取子窗口实例（GUI 模式下返回对应 Chat 对象）。"""
|
||
chat = Chat(nickname, self._gui, self._db)
|
||
try:
|
||
hits = self._db.search_contact(nickname)
|
||
except Exception:
|
||
hits = []
|
||
if hits or nickname in ('filehelper', '文件传输助手'):
|
||
return chat
|
||
return None
|
||
|
||
def GetAllSubWindow(self) -> List['Chat']:
|
||
"""获取所有子窗口实例（GUI 模式下无独立子窗口，返回空列表）。"""
|
||
return []
|
||
|
||
# -- 路径 / 生命周期 ---------------------------------------------------
|
||
|
||
@property
|
||
def path(self):
|
||
from wechatauto.utils.win32 import GetPathByHwnd
|
||
return GetPathByHwnd(self._gui.main_hwnd)
|
||
|
||
@property
|
||
def dir(self):
|
||
wxdir = self.path
|
||
if not wxdir:
|
||
return None
|
||
wxdir = os.path.dirname(wxdir)
|
||
for d in os.listdir(wxdir):
|
||
if re.match(r'\d+\.\d+\.\d+\.\d+', d):
|
||
return os.path.join(wxdir, d)
|
||
return None
|
||
|
||
def ShutDown(self):
|
||
"""强制退出微信进程。"""
|
||
pid = ctypes.c_ulong()
|
||
ctypes.windll.user32.GetWindowThreadProcessId(
|
||
self._gui.main_hwnd, ctypes.byref(pid))
|
||
if pid.value:
|
||
os.system(f'taskkill /f /pid {pid.value}')
|