feat: Initialize wxauto WeChat automation project with job extraction tools

- Add wxauto package with WeChat UI automation and message handling capabilities
- Implement job_extractor.py for automated job posting extraction from WeChat groups
- Add job_extractor_gui.py providing graphical interface for job extraction tool
- Create comprehensive documentation in Chinese covering GUI usage, multi-group support, and quick start guides
- Add build configuration files (build_exe.py, build_exe.spec) for packaging as standalone executable
- Include utility scripts for WeChat interaction (auto_send_msg.py, get_history.py, receive_file_transfer.py)
- Add project configuration files (pyproject.toml, setup.cfg, requirements.txt)
- Include test files (test_api.py, test_com_fix.py) for API and compatibility validation
- Add Apache 2.0 LICENSE and comprehensive README documentation
- Configure .gitignore to exclude build artifacts, logs, and temporary files
This commit is contained in:
2026-02-11 14:49:38 +08:00
commit b66bac7ca8
52 changed files with 15318 additions and 0 deletions

16
wxauto/__init__.py Normal file
View File

@@ -0,0 +1,16 @@
from .wx import (
WeChat,
Chat,
WeChatLogin
)
from .param import WxParam
import pythoncom
pythoncom.CoInitialize()
__all__ = [
'WeChat',
'Chat',
'WeChatLogin',
'WxParam'
]

5
wxauto/exceptions.py Normal file
View File

@@ -0,0 +1,5 @@
class WxautoOCRError(Exception):
...
class NetWorkError(Exception):
...

294
wxauto/languages.py Normal file
View File

@@ -0,0 +1,294 @@
WECHAT_MAIN = {
'新的朋友': {'cn': "新的朋友", 'cn_t': '', 'en': ""},
"添加朋友": {'cn': "添加朋友", 'cn_t': '', 'en': ""},
"搜索结果": {'cn': "搜索:", 'cn_t': '', 'en': ""},
"找不到相关账号或内容": {'cn': "找不到相关账号或内容", 'cn_t': '', 'en': ""},
}
WECHAT_CHAT_BOX = {
"查看更多消息": {'cn': "查看更多消息", 'cn_t': '', 'en': ""},
"消息": {'cn': "消息", 'cn_t': '', 'en': ""},
"表情" : {'cn': "表情(Alt+E)", 'cn_t': '', 'en': ""},
"发送文件": {'cn': "发送文件", 'cn_t': '', 'en': ""},
"截图": {'cn': "截图", 'cn_t': '', 'en': ""},
"聊天记录": {'cn': "聊天记录", 'cn_t': '', 'en': ""},
"语音聊天": {'cn': "语音聊天", 'cn_t': '', 'en': ""},
"视频聊天": {'cn': "视频聊天", 'cn_t': '', 'en': ""},
'聊天信息': {'cn': "聊天信息", 'cn_t': '', 'en': ""},
"发送": {'cn': "发送(S)", 'cn_t': '', 'en': ""},
"置顶": {'cn': "置顶", 'cn_t': '', 'en': ""},
"最小化": {'cn': "最小化", 'cn_t': '', 'en': ""},
"最大化": {'cn': "最大化", 'cn_t': '', 'en': ""},
"关闭": {'cn': "关闭", 'cn_t': '', 'en': ""},
'以下为新消息': {'cn': "以下为新消息", 'cn_t': '', 'en': ""},
're_新消息按钮': {'cn': '.*?条新消息', 'cn_t': '', 'en': ""},
}
WECHAT_SESSION_BOX = {
# 聊天页面
"聊天记录": {'cn': "聊天记录", 'cn_t': '', 'en': ""},
"会话": {'cn': "会话", 'cn_t': '', 'en': ""},
"已置顶": {'cn': "已置顶", 'cn_t': '', 'en': ""},
"文件传输助手": {'cn': "文件传输助手", 'cn_t': '', 'en': ""},
"折叠的群聊": {'cn': "折叠的群聊", 'cn_t': '', 'en': ""},
"发起群聊": {'cn': "发起群聊", 'cn_t': '', 'en': ""},
"搜索": {'cn': "搜索", 'cn_t': '', 'en': ""},
"re_条数": {'cn': r'\[\d+条\]', 'cn_t': '', 'en': ""},
# 联系人页面
"添加朋友": {'cn': "添加朋友", 'cn_t': '', 'en': ""},
"联系人": {'cn': "联系人", 'cn_t': '', 'en': ""},
"通讯录管理": {'cn': "通讯录管理", 'cn_t': '', 'en': ""},
"新的朋友": {'cn': "新的朋友", 'cn_t': '', 'en': ""},
"公众号": {'cn': "公众号", 'cn_t': '', 'en': ""},
"企业号": {'cn': "企业号", 'cn_t': '', 'en': ""},
"群聊": {'cn': "群聊", 'cn_t': '', 'en': ""},
# 收藏页面
"分类": {'cn': "分类", 'cn_t': '', 'en': ""},
"新建笔记": {'cn': "新建笔记", 'cn_t': '', 'en': ""},
"全部收藏": {'cn': "全部收藏", 'cn_t': '', 'en': ""},
"最近使用": {'cn': "最近使用", 'cn_t': '', 'en': ""},
"链接": {'cn': "链接", 'cn_t': '', 'en': ""},
"图片与视频": {'cn': "图片与视频", 'cn_t': '', 'en': ""},
"笔记": {'cn': "笔记", 'cn_t': '', 'en': ""},
"文件": {'cn': "文件", 'cn_t': '', 'en': ""},
"聊天记录": {'cn': "聊天记录", 'cn_t': '', 'en': ""},
"分割线": {'cn': "分割线", 'cn_t': '', 'en': ""},
"展开标签": {'cn': "展开标签", 'cn_t': '', 'en': ""},
"折叠标签": {'cn': "折叠标签", 'cn_t': '', 'en': ""},
"标签": {'cn': "标签", 'cn_t': '', 'en': ""},
}
WECHAT_NAVIGATION_BOX = {
'聊天': {'cn': "聊天", 'cn_t': '', 'en': ""},
'通讯录': {'cn': "通讯录", 'cn_t': '', 'en': ""},
'收藏': {'cn': "收藏", 'cn_t': '', 'en': ""},
'聊天文件': {'cn': "聊天文件", 'cn_t': '', 'en': ""},
'朋友圈': {'cn': "朋友圈", 'cn_t': '', 'en': ""},
'搜一搜': {'cn': "搜一搜", 'cn_t': '', 'en': ""},
"视频号": {'cn': "视频号", 'cn_t': '', 'en': ""},
"看一看": {'cn': "看一看", 'cn_t': '', 'en': ""},
"小程序面板": {'cn': "小程序面板", 'cn_t': '', 'en': ""},
"手机": {'cn': "手机", 'cn_t': '', 'en': ""},
"设置及其他": {'cn': "设置及其他", 'cn_t': '', 'en': ""},
}
EMOTION_WINDOW = {
"添加的单个表情": {'cn': "添加的单个表情", 'cn_t': '', 'en': ""},
}
MOMENT_PRIVACY = {
'谁可以看': {'cn': "谁可以看", 'cn_t': '', 'en': ""},
"公开": {'cn': "公开", 'cn_t': '', 'en': ""},
"所有朋友可见": {'cn': "所有朋友可见", 'cn_t': '', 'en': ""},
"私密": {'cn': "私密", 'cn_t': '', 'en': ""},
"仅自己可见": {'cn': "仅自己可见", 'cn_t': '', 'en': ""},
"白名单": {'cn': "选中的标签或朋友可见", 'cn_t': '', 'en': ""},
"黑名单": {'cn': "选中的标签或朋友不可见", 'cn_t': '', 'en': ""},
"完成": {'cn': "完成", 'cn_t': '', 'en': ""},
"确定": {'cn': "确定", 'cn_t': '', 'en': ""},
"取消": {'cn': "取消", 'cn_t': '', 'en': ""}
}
PROFILE_CARD = {
'微信号': {'cn': "微信号:", 'cn_t': '', 'en': ""},
'昵称': {'cn': "昵称:", 'cn_t': '', 'en': ""},
'备注': {'cn': "备注", 'cn_t': '', 'en': ""},
'地区': {'cn': "地区:", 'cn_t': '', 'en': ""},
'标签': {'cn': "标签", 'cn_t': '', 'en': ""},
'共同群聊': {'cn': "共同群聊", 'cn_t': '', 'en': ""},
'来源': {'cn': "来源", 'cn_t': '', 'en': ""},
'发消息': {'cn': "发消息", 'cn_t': '', 'en': ""},
'语音聊天': {'cn': "语音聊天", 'cn_t': '', 'en': ""},
'视频聊天': {'cn': "视频聊天", 'cn_t': '', 'en': ""},
'更多': {'cn': "更多", 'cn_t': '', 'en': ""},
"设置备注和标签": {'cn': "设置备注和标签", 'cn_t': '', 'en': ""},
'确定': {'cn': "确定", 'cn_t': '', 'en': ""},
'输入标签': {'cn': "输入标签", 'cn_t': '', 'en': ""},
'备注名': {'cn': "备注名", 'cn_t': '', 'en': ""},
}
MESSAGES = {
'[图片]': {'cn': "[图片]", 'cn_t': '', 'en': ""},
'[视频]': {'cn': "[视频]", 'cn_t': '', 'en': ""},
'[语音]': {'cn': "[语音]", 'cn_t': '', 'en': ""},
'[音乐]': {'cn': "[音乐]", 'cn_t': '', 'en': ""},
'[位置]': {'cn': "[位置]", 'cn_t': '', 'en': ""},
'[链接]': {'cn': "[链接]", 'cn_t': '', 'en': ""},
'[文件]': {'cn': "[文件]", 'cn_t': '', 'en': ""},
'[名片]': {'cn': "[名片]", 'cn_t': '', 'en': ""},
'[笔记]': {'cn': "[笔记]", 'cn_t': '', 'en': ""},
'[视频号]': {'cn': "[视频号]", 'cn_t': '', 'en': ""},
'[动画表情]': {'cn': "[动画表情]", 'cn_t': '', 'en': ""},
'[聊天记录]': {'cn': "[聊天记录]", 'cn_t': '', 'en': ""},
'微信转账': {'cn': "微信转账", 'cn_t': '', 'en': ""},
'接收中': {'cn': "接收中", 'cn_t': '', 'en': ""},
're_语音': {'cn': "^\[语音\]\d+秒(,未播放)?$", 'cn_t': '', 'en': ""},
're_引用消息': {'cn': "(^.+)\n引用.*?的消息 : (.+$)", 'cn_t': '', 'en': ""},
're_拍一拍': {'cn': "^.+拍了拍.+$", 'cn_t': '', 'en': ""},
}
CHATROOM_DETAIL_WINDOW = {
"聊天信息": {'cn': "聊天信息", 'cn_t': '', 'en': ""},
"查看更多": {'cn': "查看更多", 'cn_t': '', 'en': ""},
'群聊名称': {'cn': "群聊名称", 'cn_t': '', 'en': ""},
'仅群主或管理员可以修改': {'cn': "仅群主或管理员可以修改", 'cn_t': '', 'en': ""},
'我在本群的昵称': {'cn': "我在本群的昵称", 'cn_t': '', 'en': ""},
"仅群主和管理员可编辑": {'cn': "仅群主和管理员可编辑", 'cn_t': '', 'en': ""},
'点击编辑群公告': {'cn': "点击编辑群公告", 'cn_t': '', 'en': ""},
'编辑': {'cn': "编辑", 'cn_t': '', 'en': ""},
'备注': {'cn': "备注", 'cn_t': '', 'en': ""},
'群公告': {'cn': "群公告", 'cn_t': '', 'en': ""},
'分隔线': {'cn': "分隔线", 'cn_t': '', 'en': ""},
'完成': {'cn': "完成", 'cn_t': '', 'en': ""},
'发布': {'cn': "发布", 'cn_t': '', 'en': ""},
'退出群聊': {'cn': "退出群聊", 'cn_t': '', 'en': ""},
'退出': {'cn': "退出", 'cn_t': '', 'en': ""},
'聊天成员': {'cn': "聊天成员", 'cn_t': '', 'en': ""},
"添加": {'cn': "添加", 'cn_t': '', 'en': ""},
"移出": {'cn': "移出", 'cn_t': '', 'en': ""},
"re_退出群聊": {'cn': "将退出群聊“.*?”", 'cn_t': '', 'en': ""},
}
PROFILE_WINDOW = {
"微信号": {'cn': "微信号:", 'cn_t': '', 'en': ""},
"昵称": {'cn': "昵称:", 'cn_t': '', 'en': ""},
"地区": {'cn': "地区:", 'cn_t': '', 'en': ""},
"个性签名": {'cn': "个性签名", 'cn_t': '', 'en': ""},
"来源": {'cn': "来源", 'cn_t': '', 'en': ""},
"备注": {'cn': "备注", 'cn_t': '', 'en': ""},
"共同群聊": {'cn': "共同群聊", 'cn_t': '', 'en': ""},
'添加到通讯录': {'cn': "添加到通讯录", 'cn_t': '', 'en': ""},
'更多': {'cn': "更多", 'cn_t': '', 'en': ""},
}
ADD_NEW_FRIEND_WINDOW = {
'标签': {'cn': "标签", 'cn_t': '', 'en': ""},
'确定': {'cn': "确定", 'cn_t': '', 'en': ""},
'备注名': {'cn': "备注名", 'cn_t': '', 'en': ""},
'朋友圈': {'cn': "朋友圈", 'cn_t': '', 'en': ""},
'仅聊天': {'cn': "仅聊天", 'cn_t': '', 'en': ""},
'聊天、朋友圈、微信运动等': {
'cn': "聊天、朋友圈、微信运动等",
'cn_t': '',
'en': ""
},
"你的联系人较多,添加新的朋友时需选择权限": {
'cn': "你的联系人较多,添加新的朋友时需选择权限",
'cn_t': '',
'en': ""
},
"发送添加朋友申请": {
'cn': "发送添加朋友申请",
'cn_t': '',
'en': ""
}
}
ADD_GROUP_MEMBER_WINDOW = {
'搜索': {'cn': "搜索", 'cn_t': '', 'en': ""},
'确定': {'cn': "确定", 'cn_t': '', 'en': ""},
'完成': {'cn': "完成", 'cn_t': '', 'en': ""},
'发送': {'cn': "发送", 'cn_t': '', 'en': ""},
'已选择联系人': {'cn': "已选择联系人", 'cn_t': '', 'en': ""},
'请勾选需要添加的联系人': {
'cn': "请勾选需要添加的联系人",
'cn_t': '',
'en': ""
}
}
IMAGE_WINDOW = {
'上一张': {'cn': '上一张', 'cn_t': '上一張', 'en': 'Previous'},
'下一张': {'cn': '下一张', 'cn_t': '下一張', 'en': 'Next'},
'预览': {'cn': '预览', 'cn_t': '預覽', 'en': 'Preview'},
'放大': {'cn': '放大', 'cn_t': '放大', 'en': 'Zoom'},
'缩小': {'cn': '缩小', 'cn_t': '縮小', 'en': 'Shrink'},
'图片原始大小': {'cn': '图片原始大小', 'cn_t': '圖片原始大小', 'en': 'Original image size'},
'旋转': {'cn': '旋转', 'cn_t': '旋轉', 'en': 'Rotate'},
'编辑': {'cn': '编辑', 'cn_t': '編輯', 'en': 'Edit'},
'翻译': {'cn': '翻译', 'cn_t': '翻譯', 'en': 'Translate'},
'提取文字': {'cn': '提取文字', 'cn_t': '提取文字', 'en': 'Extract Text'},
'识别图中二维码': {'cn': '识别图中二维码', 'cn_t': '識别圖中QR Code', 'en': 'Extract QR Code'},
'另存为': {'cn': '另存为...', 'cn_t': '另存爲...', 'en': 'Save as…'},
'更多': {'cn': '更多', 'cn_t': '更多', 'en': 'More'},
'复制': {'cn': '复制', 'cn_t': '複製', 'en': 'Copy'},
'最小化': {'cn': '最小化', 'cn_t': '最小化', 'en': 'Minimize'},
'最大化': {'cn': '最大化', 'cn_t': '最大化', 'en': 'Maximize'},
'关闭': {'cn': '关闭', 'cn_t': '關閉', 'en': 'Close'},
'': {'cn': '', 'cn_t': '', 'en': ''}
}
MENU_OPTIONS = {
# session
'置顶': {'cn': '置顶', 'cn_t': '', 'en': ''},
'取消置顶': {'cn': '取消置顶', 'cn_t': '', 'en': ''},
'标为未读': {'cn': '标为未读', 'cn_t': '', 'en': ''},
'消息免打扰': {'cn': '消息免打扰', 'cn_t': '', 'en': ''},
'在独立窗口打开': {'cn': '在独立窗口打开', 'cn_t': '', 'en': ''},
'不显示聊天': {'cn': '不显示聊天', 'cn_t': '', 'en': ''},
'删除聊天': {'cn': '删除聊天', 'cn_t': '', 'en': ''},
# msg
'撤回': {'cn': '撤回', 'cn_t': '', 'en': ''},
'复制': {'cn': '复制', 'cn_t': '', 'en': ''},
'放大阅读': {'cn': '放大阅读', 'cn_t': '', 'en': ''},
'翻译': {'cn': '翻译', 'cn_t': '', 'en': ''},
'转发': {'cn': '转发...', 'cn_t': '', 'en': ''},
'收藏': {'cn': '收藏', 'cn_t': '', 'en': ''},
'多选': {'cn': '多选', 'cn_t': '', 'en': ''},
'引用': {'cn': '引用', 'cn_t': '', 'en': ''},
'搜一搜': {'cn': '搜一搜', 'cn_t': '', 'en': ''},
'删除': {'cn': '删除', 'cn_t': '', 'en': ''},
'编辑': {'cn': '编辑', 'cn_t': '', 'en': ''},
'另存为': {'cn': '另存为...', 'cn_t': '', 'en': ''},
'语音转文字': {'cn': '语音转文字', 'cn_t': '', 'en': ''},
'在文件夹中显示': {'cn': '在文件夹中显示', 'cn_t': '', 'en': ''},
# edit
'剪切': {'cn': '剪切', 'cn_t': '', 'en': ''},
'粘贴': {'cn': '粘贴', 'cn_t': '', 'en': ''},
}
MOMENTS = {
'朋友圈': {'cn': '朋友圈', 'cn_t': '', 'en': ''},
'刷新': {'cn': '刷新', 'cn_t': '', 'en': ''},
'评论': {'cn': '评论', 'cn_t': '', 'en': ''},
'广告': {'cn': '广告', 'cn_t': '', 'en': ''},
'': {'cn': '', 'cn_t': '', 'en': ''},
'取消': {'cn': '取消', 'cn_t': '', 'en': ''},
'发送': {'cn': '发送', 'cn_t': '', 'en': ''},
'分隔符_点赞': {'cn': '', 'cn_t': '', 'en': ''},
're_图片数': {'cn': '包含\d+张图片', 'cn_t': '', 'en': ''},
}
NEW_FRIEND_ELEMENT = {
'新的朋友': {'cn': '新的朋友', 'cn_t': '', 'en': ''},
'回复': {'cn': '回复', 'cn_t': '', 'en': ''},
'发送': {'cn': '发送', 'cn_t': '', 'en': ''},
'朋友圈': {'cn': '朋友圈', 'cn_t': '', 'en': ''},
'仅聊天': {'cn': '仅聊天', 'cn_t': '', 'en': ''},
'聊天、朋友圈、微信运动等': {'cn': '聊天、朋友圈、微信运动等', 'cn_t': '', 'en': ''},
'备注名': {'cn': '备注名', 'cn_t': '', 'en': ''},
'标签': {'cn': '标签', 'cn_t': '', 'en': ''},
}
WECHAT_BROWSER = {
"关闭": {'cn': '关闭', 'cn_t': '', 'en': ''},
"更多": {'cn': '更多', 'cn_t': '', 'en': ''},
'地址和搜索栏': {'cn': '地址和搜索栏', 'cn_t': '', 'en': ''},
'转发给朋友': {'cn': '转发给朋友', 'cn_t': '', 'en': ''},
'复制链接': {'cn': '复制链接', 'cn_t': '', 'en': ''},
}
WECHAT_LOGINWND = {
"进入微信": {'cn': '进入微信', 'cn_t': '', 'en': ''},
"切换账号": {'cn': '切换账号', 'cn_t': '', 'en': ''},
"仅传输文件": {'cn': '仅传输文件', 'cn_t': '', 'en': ''},
"二维码": {'cn': '仅传输文件', 'cn_t': '', 'en': ''},
"提示": {'cn': '提示', 'cn_t': '', 'en': ''},
"确定": {'cn': '确定', 'cn_t': '', 'en': ''},
}

128
wxauto/logger.py Normal file
View File

@@ -0,0 +1,128 @@
from wxauto.param import WxParam
import logging
import colorama
from pathlib import Path
from datetime import datetime
import sys
import io
# # 初始化 colorama
colorama.init()
if hasattr(sys.stdout, 'buffer'):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='ignore')
LOG_COLORS = {
'DEBUG': colorama.Fore.CYAN,
'INFO': colorama.Fore.GREEN,
'WARNING': colorama.Fore.YELLOW,
'ERROR': colorama.Fore.RED,
'CRITICAL': colorama.Fore.MAGENTA
}
class ColoredFormatter(logging.Formatter):
def format(self, record):
levelname = record.levelname
message = super().format(record)
return f"{LOG_COLORS[levelname]}{message}{colorama.Style.RESET_ALL}"
class WxautoLogger:
name: str = 'wxauto'
def __init__(self):
self.logger = self.setup_logger()
self.file_handler = None # 先不创建文件处理器
self.set_debug(False)
def setup_logger(self) -> logging.Logger:
"""设置日志记录器"""
# 配置根记录器
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
# 添加asyncio日志过滤
logging.getLogger('asyncio').setLevel(logging.WARNING)
# 设置第三方库的日志级别
logging.getLogger('comtypes').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
logging.getLogger('requests').setLevel(logging.WARNING)
# 清除现有处理器
root_logger.handlers.clear()
# 格式
fmt = '%(asctime)s [%(name)s] [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s'
# 控制台处理器(带颜色)
self.console_handler = logging.StreamHandler()
console_formatter = ColoredFormatter(
fmt=fmt,
datefmt="%Y-%m-%d %H:%M:%S"
)
self.console_handler.setFormatter(console_formatter)
self.console_handler.setLevel(logging.DEBUG)
root_logger.addHandler(self.console_handler)
return logging.getLogger(self.name)
def setup_file_logger(self):
"""根据WxParam.ENABLE_FILE_LOGGER决定是否创建文件日志处理器"""
if not WxParam.ENABLE_FILE_LOGGER or self.file_handler is not None:
return
# 文件处理器(无颜色)
log_dir = Path("wxauto_logs")
log_dir.mkdir(parents=True, exist_ok=True)
# 使用当前时间创建日志文件
current_time = datetime.now().strftime("%Y%m%d")
log_file = log_dir / f"app_{current_time}.log"
self.file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_formatter = logging.Formatter(
'%(asctime)s [%(name)s] [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s',
datefmt="%Y-%m-%d %H:%M:%S"
)
self.file_handler.setFormatter(file_formatter)
self.file_handler.setLevel(logging.DEBUG)
# 将文件处理器添加到日志记录器
logging.getLogger().addHandler(self.file_handler)
def set_debug(self, debug=False):
"""动态设置日志级别"""
if debug:
self.console_handler.setLevel(logging.DEBUG)
else:
self.console_handler.setLevel(logging.INFO)
def _ensure_file_logger(self):
"""确保文件日志处理器被初始化"""
if WxParam.ENABLE_FILE_LOGGER and self.file_handler is None:
self.setup_file_logger()
def debug(self, msg: str, stacklevel=2, *args, **kwargs):
self._ensure_file_logger() # 确保文件日志初始化
self.logger.debug(msg, *args, stacklevel=stacklevel, **kwargs)
def info(self, msg: str, stacklevel=2, *args, **kwargs):
self._ensure_file_logger() # 确保文件日志初始化
self.logger.info(msg, *args, stacklevel=stacklevel, **kwargs)
def warning(self, msg: str, stacklevel=2, *args, **kwargs):
self._ensure_file_logger() # 确保文件日志初始化
self.logger.warning(msg, *args, stacklevel=stacklevel, **kwargs)
def error(self, msg: str, stacklevel=2, *args, **kwargs):
self._ensure_file_logger() # 确保文件日志初始化
self.logger.error(msg, *args, stacklevel=stacklevel, **kwargs)
def critical(self, msg: str, stacklevel=2, *args, **kwargs):
self._ensure_file_logger() # 确保文件日志初始化
self.logger.critical(msg, *args, stacklevel=stacklevel, **kwargs)
wxlog = WxautoLogger()

75
wxauto/msgs/__init__.py Normal file
View File

@@ -0,0 +1,75 @@
from .msg import parse_msg
from .base import (
BaseMessage,
HumanMessage
)
from .attr import (
SystemMessage,
TickleMessage,
TimeMessage,
FriendMessage,
SelfMessage
)
from .type import (
TextMessage,
ImageMessage,
VoiceMessage,
VideoMessage,
FileMessage,
LinkMessage,
OtherMessage
)
from .self import (
SelfMessage,
SelfTextMessage,
SelfVoiceMessage,
SelfImageMessage,
SelfVideoMessage,
SelfFileMessage,
SelfLinkMessage,
SelfOtherMessage,
)
from .friend import (
FriendMessage,
FriendTextMessage,
FriendVoiceMessage,
FriendImageMessage,
FriendVideoMessage,
FriendFileMessage,
FriendLinkMessage,
FriendOtherMessage,
)
__all__ = [
'parse_msg',
'BaseMessage',
'HumanMessage',
'SystemMessage',
'TickleMessage',
'TimeMessage',
'FriendMessage',
'SelfMessage',
'TextMessage',
'ImageMessage',
'VoiceMessage',
'VideoMessage',
'FileMessage',
'LinkMessage',
'OtherMessage',
'SelfMessage',
'SelfTextMessage',
'SelfVoiceMessage',
'SelfImageMessage',
'SelfVideoMessage',
'SelfFileMessage',
'SelfLinkMessage',
'SelfOtherMessage',
'FriendMessage',
'FriendTextMessage',
'FriendVoiceMessage',
'FriendImageMessage',
'FriendVideoMessage',
'FriendFileMessage',
'FriendLinkMessage',
'FriendOtherMessage',
]

85
wxauto/msgs/attr.py Normal file
View File

@@ -0,0 +1,85 @@
from .base import *
from wxauto.utils.tools import (
parse_wechat_time
)
class SystemMessage(BaseMessage):
attr = 'system'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
self.sender = 'system'
self.sender_remark = 'system'
class TickleMessage(SystemMessage):
attr = 'tickle'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
self.tickle_list = [
i.Name for i in
control.ListItemControl().GetParentControl().GetChildren()
]
self.content = f"[{len(self.tickle_list)}条]{self.tickle_list[0]}"
class TimeMessage(SystemMessage):
attr = 'time'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
self.time = parse_wechat_time(self.content)
class FriendMessage(HumanMessage):
attr = 'friend'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
self.head_control = self.control.ButtonControl(RegexName='.*?')
self.sender = self.head_control.Name
if (
(remark_control := self.control.TextControl()).Exists(0)
and remark_control.BoundingRectangle.top < self.head_control.BoundingRectangle.top
):
self.sender_remark = remark_control.Name
else:
self.sender_remark = self.sender
@property
def _xbias(self):
if WxParam.FORCE_MESSAGE_XBIAS:
return int(self.head_control.BoundingRectangle.width()*1.5)
return WxParam.DEFAULT_MESSAGE_XBIAS
class SelfMessage(HumanMessage):
attr = 'self'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
@property
def _xbias(self):
if WxParam.FORCE_MESSAGE_XBIAS:
return -int(self.head_control.BoundingRectangle.width()*1.5)
return -WxParam.DEFAULT_MESSAGE_XBIAS

185
wxauto/msgs/base.py Normal file
View File

@@ -0,0 +1,185 @@
from wxauto import uiautomation as uia
from wxauto.logger import wxlog
from wxauto.param import (
WxResponse,
WxParam,
PROJECT_NAME
)
from wxauto.ui.component import (
CMenuWnd,
SelectContactWnd
)
from wxauto.utils.tools import roll_into_view
from wxauto.languages import *
from typing import (
Dict,
List,
Union,
TYPE_CHECKING
)
from hashlib import md5
import time
if TYPE_CHECKING:
from wxauto.ui.chatbox import ChatBox
def truncate_string(s: str, n: int=8) -> str:
s = s.replace('\n', '').strip()
return s if len(s) <= n else s[:n] + '...'
class Message:...
class BaseMessage(Message):
type: str = 'base'
attr: str = 'base'
control: uia.Control
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
self.control = control
self.parent = parent
self.root = parent.root
self.content = self.control.Name
self.id = self.control.runtimeid
self.sender = self.attr
self.sender_remark = self.attr
def __repr__(self):
cls_name = self.__class__.__name__
content = truncate_string(self.content)
return f"<{PROJECT_NAME} - {cls_name}({content}) at {hex(id(self))}>"
@property
def message_type_name(self) -> str:
return self.__class__.__name__
def chat_info(self) -> Dict:
if self.control.Exists(0):
return self.parent.get_info()
def _lang(self, text: str) -> str:
return MESSAGES.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def roll_into_view(self) -> WxResponse:
if roll_into_view(self.control.GetParentControl(), self.control, equal=True) == 'not exist':
wxlog.warning('消息目标控件不存在,无法滚动至显示窗口')
return WxResponse.failure('消息目标控件不存在,无法滚动至显示窗口')
return WxResponse.success('成功')
@property
def info(self) -> Dict:
_info = self.parent.get_info().copy()
_info['class'] = self.message_type_name
_info['id'] = self.id
_info['type'] = self.type
_info['attr'] = self.attr
_info['content'] = self.content
return _info
class HumanMessage(BaseMessage):
attr = 'human'
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
self.head_control = self.control.ButtonControl(searchDepth=2)
def roll_into_view(self) -> WxResponse:
if roll_into_view(self.control.GetParentControl(), self.head_control, equal=True) == 'not exist':
return WxResponse.failure('消息目标控件不存在,无法滚动至显示窗口')
return WxResponse.success('成功')
def click(self):
self.roll_into_view()
self.head_control.Click(x=self._xbias)
def right_click(self):
self.roll_into_view()
self.head_control.Click(x=-self._xbias)
self.head_control.RightClick(x=self._xbias)
def select_option(self, option: str, timeout=None) -> WxResponse:
self.root._show()
def _select_option(self, option):
if not (roll_result := self.roll_into_view()):
return roll_result
self.right_click()
menu = CMenuWnd(self.root)
return menu.select(item=option)
if timeout:
t0 = time.time()
while True:
if (time.time() - t0) > timeout:
return WxResponse(False, '引用消息超时')
if quote_result := _select_option(self, option):
return quote_result
else:
return _select_option(self, option)
def quote(
self, text: str,
at: Union[List[str], str] = None,
timeout: int = 3
) -> WxResponse:
"""引用消息
Args:
text (str): 引用内容
at (List[str], optional): @用户列表
timeout (int, optional): 超时时间单位为秒若为None则不启用超时设置
Returns:
WxResponse: 调用结果
"""
if not self.select_option('引用', timeout=timeout):
wxlog.debug(f"当前消息无法引用:{self.content}")
return WxResponse(False, '当前消息无法引用')
if at:
self.parent.input_at(at)
return self.parent.send_text(text)
def reply(
self, text: str,
at: Union[List[str], str] = None
) -> WxResponse:
"""引用消息
Args:
text (str): 回复内容
at (List[str], optional): @用户列表
timeout (int, optional): 超时时间单位为秒若为None则不启用超时设置
Returns:
WxResponse: 调用结果
"""
if at:
self.parent.input_at(at)
return self.parent.send_text(text)
def forward(self, targets: Union[List[str], str], timeout: int = 3) -> WxResponse:
"""转发消息
Args:
targets (Union[List[str], str]): 目标用户列表
timeout (int, optional): 超时时间单位为秒若为None则不启用超时设置
Returns:
WxResponse: 调用结果
"""
if not self.select_option('转发', timeout=timeout):
return WxResponse(False, '当前消息无法转发')
select_wnd = SelectContactWnd(self)
return select_wnd.send(targets)

73
wxauto/msgs/friend.py Normal file
View File

@@ -0,0 +1,73 @@
from .type import *
from .attr import FriendMessage
import sys
class FriendTextMessage(FriendMessage, TextMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class FriendQuoteMessage(FriendMessage, QuoteMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class FriendImageMessage(FriendMessage, ImageMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class FriendFileMessage(FriendMessage, FileMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class FriendLinkMessage(FriendMessage, LinkMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class FriendVideoMessage(FriendMessage, VideoMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class FriendVoiceMessage(FriendMessage, VoiceMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class FriendOtherMessage(FriendMessage, OtherMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)

133
wxauto/msgs/msg.py Normal file
View File

@@ -0,0 +1,133 @@
from .attr import *
from .type import OtherMessage
from . import self as selfmsg
from . import friend as friendmsg
from wxauto.languages import *
from wxauto.param import WxParam
from wxauto import uiautomation as uia
from typing import Literal
import re
class MESSAGE_ATTRS:
SYS_TEXT_HEIGHT = 33
TIME_TEXT_HEIGHT = 34
CHAT_TEXT_HEIGHT = 52
FILE_MSG_HEIGHT = 115
LINK_MSG_HEIGHT = 115
VOICE_MSG_HEIGHT = 55
TEXT_MSG_CONTROL_NUM = (8, 9, 10, 11)
TIME_MSG_CONTROL_NUM = (1,)
SYS_MSG_CONTROL_NUM = (4,5,6)
IMG_MSG_CONTROL_NUM = (9, 10, 11, 12)
FILE_MSG_CONTROL_NUM = tuple(i for i in range(15, 30))
LINK_MSG_CONTROL_NUM = tuple(i for i in range(15, 30))
VOICE_MSG_CONTROL_NUM = tuple(i for i in range(10, 30))
VIDEO_MSG_CONTROL_NUM = (13, 14, 15, 16)
QUOTE_MSG_CONTROL_NUM = tuple(i for i in range(16, 30))
LINK_MSG_CONTROL_NUM = tuple(i for i in range(15, 30))
def _lang(text: str) -> str:
return MESSAGES.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
SEPICIAL_MSGS = [
_lang(i)
for i in [
'[图片]', # ImageMessage
'[视频]', # VideoMessage
'[语音]', # VoiceMessage
'[文件]', # FileMessage
'[链接]', # LinkMessage
]
]
def parse_msg_attr(
control: uia.Control,
parent,
):
msg_rect = control.BoundingRectangle
height = msg_rect.height()
mid = (msg_rect.left + msg_rect.right) / 2
for length, _ in enumerate(uia.WalkControl(control)):length += 1
# TimeMessage
if (
length in MESSAGE_ATTRS.TIME_MSG_CONTROL_NUM
):
return TimeMessage(control, parent)
# FriendMessage or SelfMessage
if (head_control := control.ButtonControl(searchDepth=2)).Exists(0):
head_rect = head_control.BoundingRectangle
if head_rect.left < mid:
return parse_msg_type(control, parent, 'Friend')
else:
return parse_msg_type(control, parent, 'Self')
# SystemMessage or TickleMessage
else:
if length in MESSAGE_ATTRS.SYS_MSG_CONTROL_NUM:
return SystemMessage(control, parent)
elif control.ListItemControl(RegexName=_lang('re_拍一拍')).Exists(0):
return TickleMessage(control, parent)
else:
return OtherMessage(control, parent)
def parse_msg_type(
control: uia.Control,
parent,
attr: Literal['Self', 'Friend']
):
for length, _ in enumerate(uia.WalkControl(control)):length += 1
content = control.Name
wxlog.debug(f'content: {content}, length: {length}')
if attr == 'Friend':
msgtype = friendmsg
else:
msgtype = selfmsg
# Special Message Type
if content in SEPICIAL_MSGS:
# ImageMessage
if content == _lang('[图片]') and length in MESSAGE_ATTRS.IMG_MSG_CONTROL_NUM:
return getattr(msgtype, f'{attr}ImageMessage')(control, parent)
# VideoMessage
elif content == _lang('[视频]') and length in MESSAGE_ATTRS.VIDEO_MSG_CONTROL_NUM:
return getattr(msgtype, f'{attr}VideoMessage')(control, parent)
# FileMessage
elif content == _lang('[文件]') and length in MESSAGE_ATTRS.FILE_MSG_CONTROL_NUM:
return getattr(msgtype, f'{attr}FileMessage')(control, parent)
# LinkMessage
elif content == _lang('[链接]') and length in MESSAGE_ATTRS.LINK_MSG_CONTROL_NUM:
return getattr(msgtype, f'{attr}LinkMessage')(control, parent)
# TextMessage
if length in MESSAGE_ATTRS.TEXT_MSG_CONTROL_NUM:
return getattr(msgtype, f'{attr}TextMessage')(control, parent)
# QuoteMessage
elif (
rematch := re.compile(_lang('re_引用消息'), re.DOTALL).match(content)
and length in MESSAGE_ATTRS.QUOTE_MSG_CONTROL_NUM
):
return getattr(msgtype, f'{attr}QuoteMessage')(control, parent)
# VoiceMessage
elif (
rematch := re.compile(_lang('re_语音')).match(content)
and length in MESSAGE_ATTRS.VOICE_MSG_CONTROL_NUM
):
return getattr(msgtype, f'{attr}VoiceMessage')(control, parent)
return getattr(msgtype, f'{attr}OtherMessage')(control, parent)
def parse_msg(
control: uia.Control,
parent
):
result = parse_msg_attr(control, parent)
return result

67
wxauto/msgs/self.py Normal file
View File

@@ -0,0 +1,67 @@
from .type import *
from .attr import SelfMessage
import sys
class SelfTextMessage(SelfMessage, TextMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class SelfQuoteMessage(SelfMessage, QuoteMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
class SelfImageMessage(SelfMessage, ImageMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class SelfFileMessage(SelfMessage, FileMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class SelfLinkMessage(SelfMessage, LinkMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class SelfVideoMessage(SelfMessage, VideoMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class SelfVoiceMessage(SelfMessage, VoiceMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class SelfOtherMessage(SelfMessage, OtherMessage):
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)

222
wxauto/msgs/type.py Normal file
View File

@@ -0,0 +1,222 @@
from wxauto.utils.tools import (
get_file_dir,
)
from wxauto.ui.component import (
CMenuWnd,
WeChatImage,
WeChatBrowser,
)
from wxauto.utils.win32 import (
ReadClipboardData,
SetClipboardText,
)
from .base import *
from typing import (
Union,
)
from pathlib import Path
import shutil
import re
class TextMessage(HumanMessage):
type = 'text'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class QuoteMessage(HumanMessage):
type = 'quote'
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
self.content, self.quote_content = \
re.findall(self._lang('re_引用消息'), self.content, re.DOTALL)[0]
class MediaMessage:
def download(
self,
dir_path: Union[str, Path] = None,
timeout: int = 10
) -> Path:
if dir_path is None:
dir_path = WxParam.DEFAULT_SAVE_PATH
if self.type == 'image':
filename = f"wxauto_{self.type}_{time.strftime('%Y%m%d%H%M%S')}.png"
elif self.type == 'video':
filename = f"wxauto_{self.type}_{time.strftime('%Y%m%d%H%M%S')}.mp4"
filepath = get_file_dir(dir_path) / filename
self.click()
t0 = time.time()
while True:
self.right_click()
menu = CMenuWnd(self)
if menu and menu.select('复制'):
try:
clipboard_data = ReadClipboardData()
cpath = clipboard_data['15'][0]
break
except:
pass
else:
menu.close()
if time.time() - t0 > timeout:
return WxResponse.failure(f'下载超时: {self.type}')
time.sleep(0.1)
shutil.copyfile(cpath, filepath)
SetClipboardText('')
if imagewnd := WeChatImage():
imagewnd.close()
return filepath
class ImageMessage(HumanMessage, MediaMessage):
type = 'image'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class VideoMessage(HumanMessage, MediaMessage):
type = 'video'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
class VoiceMessage(HumanMessage):
type = 'voice'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
def to_text(self):
"""语音转文字"""
if self.control.GetProgenyControl(8, 4):
return self.control.GetProgenyControl(8, 4).Name
voicecontrol = self.control.ButtonControl(Name='')
if not voicecontrol.Exists(0.5):
return WxResponse.failure('语音转文字失败')
self.right_click()
menu = CMenuWnd(self.parent)
menu.select('语音转文字')
text = ''
while True:
if not self.control.Exists(0):
return WxResponse.failure('消息已撤回')
text_control = self.control.GetProgenyControl(8, 4)
if text_control is not None:
if text_control.Name == text:
return text
text = text_control.Name
time.sleep(0.1)
class LinkMessage(HumanMessage):
type = 'link'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
def get_url(self) -> str:
self.click()
if webbrower := WeChatBrowser():
url = webbrower.get_url()
webbrower.close()
return url
else:
wxlog.debug(f'找不到浏览器窗口')
return None
class FileMessage(HumanMessage):
type = 'file'
def __init__(
self,
control: uia.Control,
parent: "ChatBox"
):
super().__init__(control, parent)
#self.filename = control.TextControl().Name
self.filename = control.GetProgenyControl(9, control_type='TextControl').Name
self.filesize = control.GetProgenyControl(10, control_type='TextControl').Name
def download(
self,
dir_path: Union[str, Path] = None,
force_click: bool = False,
timeout: int = 10
) -> Path:
"""下载文件"""
if dir_path is None:
dir_path = WxParam.DEFAULT_SAVE_PATH
filepath = get_file_dir(dir_path) / self.filename
t0 = time.time()
def open_file_menu():
while not (menu := CMenuWnd(self.parent)):
self.roll_into_view()
self.right_click()
return menu
if force_click:
self.click()
while True:
if time.time() - t0 > timeout:
return WxResponse.failure("文件下载超时")
try:
if self.control.TextControl(Name=self._lang('接收中')).Exists(0):
time.sleep(0.1)
continue
menu = open_file_menu()
if (option := self._lang('复制')) in menu.option_names:
menu.select(option)
temp_filepath = Path(ReadClipboardData().get('15')[0])
break
except:
time.sleep(0.1)
t0 = time.time()
while True:
if time.time() - t0 > 2:
return WxResponse.failure("文件下载超时")
try:
shutil.copyfile(temp_filepath, filepath)
SetClipboardText('')
return filepath
except:
time.sleep(0.01)
class OtherMessage(BaseMessage):
type = 'other'
def __init__(
self,
control: uia.Control,
parent: "ChatBox",
):
super().__init__(control, parent)
self.url = control.TextControl().Name

65
wxauto/param.py Normal file
View File

@@ -0,0 +1,65 @@
from typing import Literal
import os
PROJECT_NAME = 'wxauto'
class WxParam:
# 语言设置
LANGUAGE: Literal['cn', 'cn_t', 'en'] = 'cn'
# 是否启用日志文件
ENABLE_FILE_LOGGER: bool = True
# 下载文件/图片默认保存路径
DEFAULT_SAVE_PATH: str = os.path.join(os.getcwd(), 'wxauto文件下载')
# 是否启用消息哈希值用于辅助判断消息,开启后会稍微影响性能
MESSAGE_HASH: bool = False
# 头像到消息X偏移量用于消息定位点击消息等操作
DEFAULT_MESSAGE_XBIAS = 51
# 是否强制重新自动获取X偏移量如果设置为True则每次启动都会重新获取
FORCE_MESSAGE_XBIAS: bool = True
# 监听消息时间间隔,单位秒
LISTEN_INTERVAL: int = 1
# 搜索聊天对象超时时间
SEARCH_CHAT_TIMEOUT: int = 5
class WxResponse(dict):
def __init__(self, status: str, message: str, data: dict = None):
super().__init__(status=status, message=message, data=data)
def __str__(self):
return str(self.to_dict())
def __repr__(self):
return str(self.to_dict())
def to_dict(self):
return {
'status': self['status'],
'message': self['message'],
'data': self['data']
}
def __bool__(self):
return self.is_success
@property
def is_success(self):
return self['status'] == '成功'
@classmethod
def success(cls, message=None, data: dict = None):
return cls(status="成功", message=message, data=data)
@classmethod
def failure(cls, message: str, data: dict = None):
return cls(status="失败", message=message, data=data)
@classmethod
def error(cls, message: str, data: dict = None):
return cls(status="错误", message=message, data=data)

8
wxauto/ui/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
from .base import BaseUIWnd, BaseUISubWnd
from . import (
chatbox,
component,
main,
navigationbox,
sessionbox
)

56
wxauto/ui/base.py Normal file
View File

@@ -0,0 +1,56 @@
from wxauto import uiautomation as uia
from wxauto.param import PROJECT_NAME
from wxauto.logger import wxlog
from abc import ABC, abstractmethod
import win32gui
from typing import Union
import time
class BaseUIWnd(ABC):
_ui_cls_name: str = None
_ui_name: str = None
control: uia.Control
@abstractmethod
def _lang(self, text: str):pass
def __repr__(self):
return f"<{PROJECT_NAME} - {self.__class__.__name__} at {hex(id(self))}>"
def __eq__(self, other):
return self.control == other.control
def __bool__(self):
return self.exists()
def _show(self):
if hasattr(self, 'HWND'):
win32gui.ShowWindow(self.HWND, 1)
win32gui.SetWindowPos(self.HWND, -1, 0, 0, 0, 0, 3)
win32gui.SetWindowPos(self.HWND, -2, 0, 0, 0, 0, 3)
self.control.SwitchToThisWindow()
def close(self):
try:
self.control.SendKeys('{Esc}')
except:
pass
def exists(self, wait=0):
try:
result = self.control.Exists(wait)
return result
except:
return False
class BaseUISubWnd(BaseUIWnd):
root: BaseUIWnd
parent: None
def _lang(self, text: str):
if getattr(self, 'parent'):
return self.parent._lang(text)
else:
return self.root._lang(text)

455
wxauto/ui/chatbox.py Normal file
View File

@@ -0,0 +1,455 @@
from .base import BaseUISubWnd
from wxauto.ui.component import (
CMenuWnd,
)
from wxauto.param import (
WxParam,
WxResponse,
)
from wxauto.languages import *
from wxauto.utils import (
SetClipboardText,
SetClipboardFiles,
GetAllWindowExs,
)
from wxauto.msgs import parse_msg
from wxauto import uiautomation as uia
from wxauto.logger import wxlog
from wxauto.uiautomation import Control
from wxauto.utils.tools import roll_into_view
import time
import os
USED_MSG_IDS = {}
class ChatBox(BaseUISubWnd):
def __init__(self, control: uia.Control, parent):
self.control: Control = control
self.root = parent
self.parent = parent # `wx` or `chat`
self.init()
def init(self):
self.msgbox = self.control.ListControl(Name=self._lang("消息"))
# if not self.msgbox.Exists(0):
# return
self.editbox = self.control.EditControl()
self.sendbtn = self.control.ButtonControl(Name=self._lang('发送'))
self.tools = self.control.PaneControl().ToolBarControl()
# self.id = self.msgbox.runtimeid
self._empty = False # 用于记录是否为完全没有聊天记录的窗口,因为这种窗口之前有不会触发新消息判断的问题
if (cid := self.id) and cid not in USED_MSG_IDS:
USED_MSG_IDS[self.id] = tuple((i.runtimeid for i in self.msgbox.GetChildren()))
if not USED_MSG_IDS[cid]:
self._empty = True
def _lang(self, text: str) -> str:
return WECHAT_CHAT_BOX.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def _update_used_msg_ids(self):
USED_MSG_IDS[self.id] = tuple((i.runtimeid for i in self.msgbox.GetChildren()))
def _open_chat_more_info(self):
for chatinfo_control, depth in uia.WalkControl(self.control):
if chatinfo_control.Name == self._lang('聊天信息'):
chatinfo_control.Click()
break
else:
return WxResponse.failure('未找到聊天信息按钮')
return ChatRoomDetailWnd(self)
def _activate_editbox(self):
if not self.editbox.HasKeyboardFocus:
self.editbox.MiddleClick()
@property
def who(self):
if hasattr(self, '_who'):
return self._who
self._who = self.editbox.Name
return self._who
@property
def id(self):
if self.msgbox.Exists(0):
return self.msgbox.runtimeid
return None
@property
def used_msg_ids(self):
return USED_MSG_IDS[self.id]
def get_info(self):
chat_info = {}
walk = uia.WalkControl(self.control)
for chat_name_control, depth in walk:
if isinstance(chat_name_control, uia.TextControl):
break
if (
not isinstance(chat_name_control, uia.TextControl)
or depth < 8
):
return {}
# chat_name_control = self.control.GetProgenyControl(11)
chat_name_control_list = chat_name_control.GetParentControl().GetChildren()
chat_name_control_count = len(chat_name_control_list)
if chat_name_control_count == 1:
if self.control.ButtonControl(Name='公众号主页', searchDepth=9).Exists(0):
chat_info['chat_type'] = 'official'
else:
chat_info['chat_type'] = 'friend'
chat_info['chat_name'] = chat_name_control.Name
elif chat_name_control_count >= 2:
try:
second_text = chat_name_control_list[1].Name
if second_text.startswith('@'):
chat_info['company'] = second_text
chat_info['chat_type'] = 'service'
chat_info['chat_name'] = chat_name_control.Name
else:
chat_info['group_member_count'] =\
int(second_text.replace('(', '').replace(')', ''))
chat_info['chat_type'] = 'group'
chat_info['chat_name'] =\
chat_name_control.Name.replace(second_text, '')
except:
chat_info['chat_type'] = 'friend'
chat_info['chat_name'] = chat_name_control.Name
ori_chat_name_control =\
chat_name_control.GetParentControl().\
GetParentControl().TextControl(searchDepth=1)
if ori_chat_name_control.Exists(0):
chat_info['chat_remark'] = chat_info['chat_name']
chat_info['chat_name'] = ori_chat_name_control.Name
self._info = chat_info
return chat_info
def input_at(self, at_list):
self._show()
if isinstance(at_list, str):
at_list = [at_list]
self._activate_editbox()
for friend in at_list:
self.editbox.SendKeys('@'+friend.replace(' ', ''))
atmenu = AtMenu(self)
atmenu.select(friend)
def clear_edit(self):
self._show()
self.editbox.Click()
self.editbox.SendKeys('{Ctrl}a', waitTime=0)
self.editbox.SendKeys('{DELETE}')
def send_text(self, content: str):
self._show()
t0 = time.time()
while True:
if time.time() - t0 > 10:
return WxResponse.failure(f'Timeout --> {self.who} - {content}')
SetClipboardText(content)
self._activate_editbox()
self.editbox.SendKeys('{Ctrl}v')
if self.editbox.GetValuePattern().Value.replace('', '').strip():
break
self.editbox.SendKeys('{Ctrl}v')
if self.editbox.GetValuePattern().Value.replace('', '').strip():
break
self.editbox.RightClick()
menu = CMenuWnd(self)
menu.select('粘贴')
if self.editbox.GetValuePattern().Value.replace('', '').strip():
break
t0 = time.time()
while self.editbox.GetValuePattern().Value:
if time.time() - t0 > 10:
return WxResponse.failure(f'Timeout --> {self.who} - {content}')
self._activate_editbox()
self.sendbtn.Click()
if not self.editbox.GetValuePattern().Value:
return WxResponse.success(f"success")
elif not self.editbox.GetValuePattern().Value.replace('', '').strip():
return self.send_text(content)
def send_msg(self, content: str, clear: bool=True, at=None):
if not content and not at:
return WxResponse.failure(f"参数 `content` 和 `at` 不能同时为空")
if clear:
self.clear_edit()
if at:
self.input_at(at)
return self.send_text(content)
def send_file(self, file_path):
self._show()
if isinstance(file_path, str):
file_path = [file_path]
file_path = [os.path.abspath(f) for f in file_path]
SetClipboardFiles(file_path)
self._activate_editbox()
self.editbox.SendKeys('{Ctrl}v')
self.sendbtn.Click()
if self.editbox.GetValuePattern().Value:
return WxResponse.fail("发送失败,请重试")
return WxResponse.success()
def load_more(self, interval=0.3):
self._show()
msg_len = len(self.msgbox.GetChildren())
loadmore = self.msgbox.GetChildren()[0]
loadmore_top = loadmore.BoundingRectangle.top
while True:
if len(self.msgbox.GetChildren()) > msg_len:
isload = True
break
else:
msg_len = len(self.msgbox.GetChildren())
self.msgbox.WheelUp(wheelTimes=10)
time.sleep(interval)
if self.msgbox.GetChildren()[0].BoundingRectangle.top == loadmore_top\
and len(self.msgbox.GetChildren()) == msg_len:
isload = False
break
else:
loadmore_top = self.msgbox.GetChildren()[0].BoundingRectangle.top
self.msgbox.WheelUp(wheelTimes=1, waitTime=0.1)
if isload:
return WxResponse.success()
else:
return WxResponse.failure("没有更多消息了")
def get_msgs(self):
if self.msgbox.Exists(0):
return [
parse_msg(msg_control, self)
for msg_control
in self.msgbox.GetChildren()
if msg_control.ControlTypeName == 'ListItemControl'
]
return []
def get_new_msgs(self):
if not self.msgbox.Exists(0):
return []
msg_controls = self.msgbox.GetChildren()
now_msg_ids = tuple((i.runtimeid for i in msg_controls))
if not now_msg_ids: # 当前没有消息id
return []
if self._empty and self.used_msg_ids:
self._empty = False
if not self._empty and (
(not self.used_msg_ids and now_msg_ids) # 没有使用过的消息id但当前有消息id
or now_msg_ids[-1] == self.used_msg_ids[-1] # 当前最后一条消息id和上次一样
or not set(now_msg_ids)&set(self.used_msg_ids) # 当前消息id和上次没有交集
):
# wxlog.debug('没有新消息')
return []
used_msg_ids_set = set(self.used_msg_ids)
last_one_msgid = max(
(x for x in now_msg_ids if x in used_msg_ids_set),
key=self.used_msg_ids.index, default=None
)
new1 = [x for x in now_msg_ids if x not in used_msg_ids_set]
new2 = now_msg_ids[now_msg_ids.index(last_one_msgid) + 1 :]\
if last_one_msgid is not None else []
new = [i for i in new1 if i in new2] if new2 else new1
USED_MSG_IDS[self.id] = tuple(self.used_msg_ids + tuple(new))[-100:]
new_controls = [i for i in msg_controls if i.runtimeid in new]
self.msgbox.MiddleClick()
return [
parse_msg(msg_control, self)
for msg_control
in new_controls
if msg_control.ControlTypeName == 'ListItemControl'
]
def get_msg_by_id(self, msg_id: str):
if not self.msgbox.Exists(0):
return []
msg_controls = self.msgbox.GetChildren()
if control_list := [i for i in msg_controls if i.runtimeid == msg_id]:
return parse_msg(control_list[0], self)
def _get_tail_after_nth_match(self, msgs, last_msg, n):
matches = [
i for i, msg in reversed(list(enumerate(msgs)))
if msg.content == last_msg
]
if len(matches) >= n:
wxlog.debug(f'匹配到基准消息:{last_msg}')
else:
split_last_msg = last_msg.split('')
nickname = split_last_msg[0]
content = ''.join(split_last_msg[1:])
matches = [
i for i, msg in reversed(list(enumerate(msgs)))
if msg.content == content
and msg.sender_remark == nickname
]
if len(matches) >= n:
wxlog.debug(f'匹配到基准消息:<{nickname}> {content}')
else:
wxlog.debug(f"未匹配到基准消息,以最后一条消息为基准:{msgs[-1].content}")
matches = [
i for i, msg in reversed(list(enumerate(msgs)))
if msg.attr in ('self', 'friend')
]
try:
index = matches[n - 1]
return msgs[index:]
except IndexError:
wxlog.debug(f"未匹配到第{n}条消息,返回空列表")
return []
def get_next_new_msgs(self, count=None, last_msg=None):
# 1. 消息列表不存在,则返回空列表
if not self.msgbox.Exists(0):
wxlog.debug('消息列表不存在,返回空列表')
return []
# 2. 判断是否有新消息按钮,有的话点一下
load_new_button = self.control.ButtonControl(RegexName=self._lang('re_新消息按钮'))
if load_new_button.Exists(0):
self._show()
wxlog.debug('检测到新消息按钮,点击加载新消息')
load_new_button.Click()
time.sleep(0.5)
msg_controls = self.msgbox.GetChildren()
USED_MSG_IDS[self.id] = tuple((i.runtimeid for i in msg_controls))
msgs = [
parse_msg(msg_control, self)
for msg_control
in msg_controls
if msg_control.ControlTypeName == 'ListItemControl'
]
# 3. 如果有“以下是新消息”标志,则直接返回该标志下的所有消息即可
index = next((
i for i, msg in enumerate(msgs)
if self._lang('以下为新消息') == msg.content
), None)
if index is not None:
wxlog.debug('获取以下是新消息下的所有消息')
return msgs[index:]
# 4. 根据会话列表传入的消息数量和最后一条新消息内容来判断新消息
if count and last_msg:
# index = next((
# i for i, msg in enumerate(msgs[::-1])
# if last_msg == msg.content
# ), None)
# if index is not None:
wxlog.debug(f'获取{count}条新消息,基准消息内容为:{last_msg}')
return self._get_tail_after_nth_match(msgs, last_msg, count)
def get_group_members(self):
self._show()
roominfoWnd = self._open_chat_more_info()
return roominfoWnd.get_group_members()
class ChatRoomDetailWnd(BaseUISubWnd):
_ui_cls_name: str = 'SessionChatRoomDetailWnd'
def __init__(self, parent):
self.parent = parent
self.root = parent.root
self.control = self.root.control.Control(ClassName=self._ui_cls_name, searchDepth=1)
def _lang(self, text: str) -> str:
return CHATROOM_DETAIL_WINDOW.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def _edit(self, key, value):
wxlog.debug(f'修改{key}为`{value}`')
btn = self.control.TextControl(Name=key).GetParentControl().ButtonControl(Name=key)
if btn.Exists(0):
roll_into_view(self.control, btn)
btn.Click()
else:
wxlog.debug(f'当前非群聊,无法修改{key}')
return WxResponse.failure(f'Not a group chat, cannot modify `{key}`')
while True:
edit_hwnd_list = [
i[0]
for i in GetAllWindowExs(self.control.NativeWindowHandle)
if i[1] == 'EditWnd'
]
if edit_hwnd_list:
edit_hwnd = edit_hwnd_list[0]
break
btn.Click()
edit_win32 = uia.Win32(edit_hwnd)
edit_win32.shortcut_select_all()
edit_win32.send_keys_shortcut('{DELETE}')
edit_win32.input(value)
edit_win32.send_keys_shortcut('{ENTER}')
return WxResponse.success()
def get_group_members(self, control=False):
"""获取群成员"""
more = self.control.ButtonControl(Name=self._lang('查看更多'), searchDepth=8)
if more.Exists(0.5):
more.Click()
members = [i for i in self.control.ListControl(Name=self._lang('聊天成员')).GetChildren()]
while members[-1].Name in [self._lang('添加'), self._lang('移出')]:
members = members[:-1]
if control:
return members
member_names = [i.Name for i in members]
self.close()
return member_names
class GroupMemberElement:
def __init__(self, control, parent) -> None:
self.control = control
self.parent = parent
self.root = self.parent.root
self.nickname = self.control.Name
def __repr__(self) -> str:
return f"<wxauto Group Member Element at {hex(id(self))}>"
class AtMenu(BaseUISubWnd):
_ui_cls_name = 'ChatContactMenu'
def __init__(self, parent):
self.root = parent.root
self.control = parent.parent.control.PaneControl(ClassName='ChatContactMenu')
# self.control.Exists(1)
def clear(self, friend):
if self.exists():
self.control.SendKeys('{ESC}')
for _ in range(len(friend)+1):
self.root.chatbox.editbox.SendKeys('{BACK}')
def select(self, friend):
friend_ = friend.replace(' ', '')
if self.exists():
ateles = self.control.ListControl().GetChildren()
if len(ateles) == 1:
ateles[0].Click()
else:
atele = self.control.ListItemControl(Name=friend)
if atele.Exists(0):
roll_into_view(self.control, atele)
atele.Click()
else:
self.clear(friend_)
else:
self.clear(friend_)

407
wxauto/ui/component.py Normal file
View File

@@ -0,0 +1,407 @@
from wxauto.utils import (
FindWindow,
SetClipboardText,
ReadClipboardData,
GetAllWindows,
GetWindowRect,
capture
)
from wxauto.param import (
WxParam,
WxResponse,
)
from wxauto.languages import *
from .base import (
BaseUISubWnd
)
from wxauto.logger import wxlog
from wxauto.uiautomation import (
ControlFromHandle,
)
from wxauto.utils.tools import (
get_file_dir,
roll_into_view,
)
from PIL import Image
from wxauto import uiautomation as uia
import traceback
import shutil
import time
class EditBox:
...
class SelectContactWnd(BaseUISubWnd):
"""选择联系人窗口"""
_ui_cls_name = 'SelectContactWnd'
def __init__(self, parent):
self.parent = parent
self.root = parent.root
hwnd = FindWindow(self._ui_cls_name, timeout=1)
if hwnd:
self.control = ControlFromHandle(hwnd)
else:
self.control = parent.root.control.PaneControl(ClassName=self._ui_cls_name, searchDepth=1)
self.editbox = self.control.EditControl()
def send(self, target):
if isinstance(target, str):
SetClipboardText(target)
while not self.editbox.HasKeyboardFocus:
self.editbox.Click()
time.sleep(0.1)
self.editbox.SendKeys('{Ctrl}a')
self.editbox.SendKeys('{Ctrl}v')
checkbox = self.control.ListControl().CheckBoxControl()
if checkbox.Exists(1):
checkbox.Click()
self.control.ButtonControl(Name='发送').Click()
return WxResponse.success()
else:
self.control.SendKeys('{Esc}')
wxlog.debug(f'未找到好友:{target}')
return WxResponse.failure(f'未找到好友:{target}')
elif isinstance(target, list):
n = 0
fail = []
multiselect = self.control.ButtonControl(Name='多选')
if multiselect.Exists(0):
multiselect.Click()
for i in target:
SetClipboardText(i)
while not self.editbox.HasKeyboardFocus:
self.editbox.Click()
time.sleep(0.1)
self.editbox.SendKeys('{Ctrl}a')
self.editbox.SendKeys('{Ctrl}v')
checkbox = self.control.ListControl().CheckBoxControl()
if checkbox.Exists(1):
checkbox.Click()
n += 1
else:
fail.append(i)
wxlog.debug(f"未找到转发对象:{i}")
if n > 0:
self.control.ButtonControl(RegexName='分别发送(\d+').Click()
if n == len(target):
return WxResponse.success()
else:
return WxResponse.success('存在未转发成功名单', data=fail)
else:
self.control.SendKeys('{Esc}')
wxlog.debug(f'所有好友均未未找到:{target}')
return WxResponse.failure(f'所有好友均未未找到:{target}')
class CMenuWnd(BaseUISubWnd):
_ui_cls_name = 'CMenuWnd'
def __init__(self, parent):
self.parent = parent
self.root = parent.root
if menulist := [i for i in GetAllWindows() if 'CMenuWnd' in i]:
self.control = uia.ControlFromHandle(menulist[0][0])
else:
self.control = self.root.control.MenuControl(ClassName=self._ui_cls_name)
def _lang(self, text: str) -> str:
return MENU_OPTIONS.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
@property
def option_controls(self):
return self.control.ListControl().GetChildren()
@property
def option_names(self):
return [c.Name for c in self.option_controls]
def select(self, item):
if not self.exists(0):
return WxResponse.failure('菜单窗口不存在')
if isinstance(item, int):
self.option_controls[item].Click()
return WxResponse.success()
item = self._lang(item)
for c in self.option_controls:
if c.Name == item:
c.Click()
return WxResponse.success()
if self.exists(0):
self.close()
return WxResponse.failure(f'未找到选项:{item}')
def close(self):
try:
self.control.SendKeys('{ESC}')
except Exception as e:
pass
class NetErrInfoTipsBarWnd(BaseUISubWnd):
_ui_cls_name = 'NetErrInfoTipsBarWnd'
def __init__(self, parent):
self.control = parent.root.control.PaneControl(ClassName=self._ui_cls_name)
def __bool__(self):
return self.exists(0)
class WeChatImage(BaseUISubWnd):
_ui_cls_name = 'ImagePreviewWnd'
def __init__(self) -> None:
self.hwnd = FindWindow(classname=self._ui_cls_name)
if self.hwnd:
self.control = ControlFromHandle(self.hwnd)
self.type = 'image'
if self.control.PaneControl(ClassName='ImagePreviewLayerWnd').Exists(0):
self.type = 'video'
MainControl1 = [i for i in self.control.GetChildren() if not i.ClassName][0]
self.ToolsBox, self.PhotoBox = MainControl1.GetChildren()
# tools按钮
self.t_previous = self.ToolsBox.ButtonControl(Name=self._lang('上一张'))
self.t_next = self.ToolsBox.ButtonControl(Name=self._lang('下一张'))
self.t_zoom = self.ToolsBox.ButtonControl(Name=self._lang('放大'))
self.t_translate = self.ToolsBox.ButtonControl(Name=self._lang('翻译'))
self.t_ocr = self.ToolsBox.ButtonControl(Name=self._lang('提取文字'))
self.t_save = self.control.ButtonControl(Name=self._lang('另存为...'))
self.t_qrcode = self.ToolsBox.ButtonControl(Name=self._lang('识别图中二维码'))
def _lang(self, text: str) -> str:
return IMAGE_WINDOW.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def ocr(self, wait=10):
result = ''
ctrls = self.PhotoBox.GetChildren()
if len(ctrls) == 2:
self.t_ocr.Click()
t0 = time.time()
while time.time() - t0 < wait:
ctrls = self.PhotoBox.GetChildren()
if len(ctrls) == 3:
TranslateControl = ctrls[-1]
result = TranslateControl.TextControl().Name
if result:
return result
else:
self.t_ocr.Click()
time.sleep(0.1)
return result
def save(self, dir_path=None, timeout=10):
"""保存图片/视频
Args:
dir_path (str): 绝对路径,包括文件名和后缀,例如:"D:/Images/微信图片_xxxxxx.png"
timeout (int, optional): 保存超时时间默认10秒
Returns:
str: 文件保存路径即savepath
"""
if dir_path is None:
dir_path = WxParam.DEFAULT_SAVE_PATH
suffix = 'png' if self.type == 'image' else 'mp4'
filename = f"wxauto_{self.type}_{time.strftime('%Y%m%d%H%M%S')}.{suffix}"
filepath = get_file_dir(dir_path) / filename
t0 = time.time()
SetClipboardText('')
while True:
if time.time() - t0 > timeout:
if self.control.Exists(0):
self.control.SendKeys('{Esc}')
raise TimeoutError('下载超时')
try:
self.control.ButtonControl(Name=self._lang('更多')).Click()
menu = self.control.MenuControl(ClassName='CMenuWnd')
menu.MenuItemControl(Name=self._lang('复制')).Click()
clipboard_data = ReadClipboardData()
path = clipboard_data['15'][0]
wxlog.debug(f"读取到图片/视频路径:{path}")
break
except:
wxlog.debug(traceback.format_exc())
time.sleep(0.1)
shutil.copyfile(path, filepath)
SetClipboardText('')
if self.control.Exists(0):
wxlog.debug("关闭图片窗口")
self.control.SendKeys('{Esc}')
return filepath
class NewFriendElement:
def __init__(self, control, parent):
self.parent = parent
self.root = parent.root
self.control = control
self.name = self.control.Name
self.msg = self.control.GetFirstChildControl().PaneControl(SearchDepth=1).GetChildren()[-1].TextControl().Name
self.NewFriendsBox = self.root.chatbox.control.ListControl(Name='新的朋友').GetParentControl()
self.status = self.control.GetFirstChildControl().GetChildren()[-1]
self.acceptable = isinstance(self.status, uia.ButtonControl)
def __repr__(self) -> str:
return f"<wxauto New Friends Element at {hex(id(self))} ({self.name}: {self.msg})>"
def delete(self):
wxlog.info(f'删除好友请求: {self.name}')
roll_into_view(self.NewFriendsBox, self.control)
self.control.RightClick()
menu = CMenuWnd(self.root)
menu.select('删除')
def reply(self, text):
wxlog.debug(f'回复好友请求: {self.name}')
roll_into_view(self.NewFriendsBox, self.control)
self.control.Click()
self.root.ChatBox.ButtonControl(Name='回复').Click()
edit = self.root.ChatBox.EditControl()
edit.Click()
edit.SendKeys('{Ctrl}a')
SetClipboardText(text)
edit.SendKeys('{Ctrl}v')
time.sleep(0.1)
self.root.ChatBox.ButtonControl(Name='发送').Click()
dialog = self.root.UiaAPI.PaneControl(ClassName='WeUIDialog')
while edit.Exists(0):
if dialog.Exists(0):
systext = dialog.TextControl().Name
wxlog.debug(f'系统提示: {systext}')
dialog.SendKeys('{Esc}')
self.root.ChatBox.ButtonControl(Name='').Click()
return WxResponse.failure(msg=systext)
time.sleep(0.1)
self.root.ChatBox.ButtonControl(Name='').Click()
return WxResponse.success()
def accept(self, remark=None, tags=None, permission='朋友圈'):
"""接受好友请求
Args:
remark (str, optional): 备注名
tags (list, optional): 标签列表
permission (str, optional): 朋友圈权限, 可选值:'朋友圈', '仅聊天'
"""
if not self.acceptable:
wxlog.debug(f"当前好友状态无法接受好友请求:{self.name}")
return
wxlog.debug(f"接受好友请求:{self.name} 备注:{remark} 标签:{tags}")
self.root._show()
roll_into_view(self.NewFriendsBox, self.status)
self.status.Click()
NewFriendsWnd = self.root.control.WindowControl(ClassName='WeUIDialog')
tipscontrol = NewFriendsWnd.TextControl(Name="你的联系人较多,添加新的朋友时需选择权限")
permission_sns = NewFriendsWnd.CheckBoxControl(Name='聊天、朋友圈、微信运动等')
permission_chat = NewFriendsWnd.CheckBoxControl(Name='仅聊天')
if tipscontrol.Exists(0.5):
permission_sns = tipscontrol.GetParentControl().GetParentControl().TextControl(Name='朋友圈')
permission_chat = tipscontrol.GetParentControl().GetParentControl().TextControl(Name='仅聊天')
if remark:
remarkedit = NewFriendsWnd.TextControl(Name='备注名').GetParentControl().EditControl()
remarkedit.Click()
remarkedit.SendKeys('{Ctrl}a')
remarkedit.SendKeys(remark)
if tags:
tagedit = NewFriendsWnd.TextControl(Name='标签').GetParentControl().EditControl()
for tag in tags:
tagedit.Click()
tagedit.SendKeys(tag)
NewFriendsWnd.PaneControl(ClassName='DropdownWindow').TextControl().Click()
if permission == '朋友圈':
permission_sns.Click()
elif permission == '仅聊天':
permission_chat.Click()
NewFriendsWnd.ButtonControl(Name='确定').Click()
class WeChatLoginWnd(BaseUISubWnd):
_ui_cls_name = 'WeChatLoginWndForPC'
def __init__(self):
self.hwnd = FindWindow(classname=self._ui_cls_name)
if self.hwnd:
self.control = ControlFromHandle(self.hwnd)
def _lang(self, text: str) -> str:
return WECHAT_LOGINWND.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def clear_hint(self) -> bool:
self._show()
hint_dialog = self.control.PaneControl(Name=self._lang('提示'))
if hint_dialog.Exists(0):
dialog_button = self.control.ButtonControl(Name=self._lang('确定'))
dialog_button.Click()
return True
else:
return False
def login(self) -> bool:
self._show()
enter = self.control.ButtonControl(Name=self._lang('进入微信'))
if enter.Exists(0):
enter.Click()
return True
else:
return False
def get_qrcode(self) -> Image.Image:
self._show()
qrcode = self.control.ButtonControl(Name=self._lang('二维码'))
if qrcode.Exists(0):
window_rect = GetWindowRect(self.hwnd)
win_left, win_top, win_right, win_bottom = window_rect
bbox = win_left + 62, win_top + 88, win_left + 218, win_top + 245
return capture(self.hwnd, bbox)
else:
return None
class WeChatBrowser(BaseUISubWnd):
_ui_cls_name = 'Chrome_WidgetWin_0'
_ui_name = '微信'
def __init__(self):
self.hwnd = FindWindow(classname=self._ui_cls_name, name=self._ui_name)
if self.hwnd:
self.control = ControlFromHandle(self.hwnd)
def _lang(self, text: str) -> str:
return WECHAT_BROWSER.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def get_url(self) -> str:
self._show()
tab = self.control.TabItemControl()
if tab.Exists():
tab.RightClick()
copy_link_item = uia.MenuItemControl(Name=self._lang('复制链接'))
if copy_link_item.Exists():
copy_link_item.Click()
clipboard_data = ReadClipboardData()
url = (clipboard_data.get('13') or
clipboard_data.get('1') or
None)
SetClipboardText('')
return url
else:
wxlog.debug(f'找不到复制链接菜单项')
else:
wxlog.debug(f'找不到标签页')
def close(self):
close_button = self.control.ButtonControl(Name=self._lang('关闭'), foundIndex=1)
if close_button.Exists():
close_button.Click()
close_button = self.control.ButtonControl(Name=self._lang('关闭'), foundIndex=2)
if close_button.Exists():
close_button.Click()
close_button = self.control.ButtonControl(Name=self._lang('关闭'), foundIndex=3)
if close_button.Exists():
close_button.Click()

265
wxauto/ui/main.py Normal file
View File

@@ -0,0 +1,265 @@
from .base import (
BaseUIWnd,
BaseUISubWnd
)
from wxauto import uiautomation as uia
from wxauto.param import WxResponse
from .chatbox import ChatBox
from .sessionbox import SessionBox
from .navigationbox import NavigationBox
from wxauto.param import (
WxParam,
WxResponse,
PROJECT_NAME
)
from wxauto.languages import *
from wxauto.utils import (
GetAllWindows,
FindWindow,
)
from wxauto.logger import wxlog
from typing import (
Union,
List,
)
class WeChatSubWnd(BaseUISubWnd):
_ui_cls_name: str = 'ChatWnd'
chatbox: ChatBox = None
nickname: str = ''
def __init__(
self,
key: Union[str, int],
parent: 'WeChatMainWnd',
timeout: int = 3
):
self.root = self
self.parent = parent
if isinstance(key, str):
hwnd = FindWindow(classname=self._ui_cls_name, name=key, timeout=timeout)
else:
hwnd = key
self.control = uia.ControlFromHandle(hwnd)
if self.control is not None:
chatbox_control = self.control.PaneControl(ClassName='', searchDepth=1)
self.chatbox = ChatBox(chatbox_control, self)
self.nickname = self.control.Name
def __repr__(self):
return f'<{PROJECT_NAME} - {self.__class__.__name__} object("{self.nickname}")>'
@property
def pid(self):
if not hasattr(self, '_pid'):
self._pid = self.control.ProcessId
return self._pid
def _get_chatbox(
self,
nickname: str=None,
exact: bool=False
) -> ChatBox:
return self.chatbox
def chat_info(self):
return self.chatbox.get_info()
def load_more_message(self, interval=0.3) -> WxResponse:
return self.chatbox.load_more(interval)
def send_msg(
self,
msg: str,
who: str=None,
clear: bool=True,
at: Union[str, List[str]]=None,
exact: bool=False,
) -> WxResponse:
chatbox = self._get_chatbox(who, exact)
if not chatbox:
return WxResponse.failure(f'未找到会话: {who}')
return chatbox.send_msg(msg, clear, at)
def send_files(
self,
filepath,
who=None,
exact=False
) -> WxResponse:
chatbox = self._get_chatbox(who, exact)
if not chatbox:
return WxResponse.failure(f'未找到会话: {who}')
return chatbox.send_file(filepath)
def get_group_members(
self,
who: str=None,
exact: bool=False
) -> List[str]:
chatbox = self._get_chatbox(who, exact)
if not chatbox:
return WxResponse.failure(f'未找到会话: {who}')
return chatbox.get_group_members()
def get_msgs(self):
chatbox = self._get_chatbox()
if chatbox:
return chatbox.get_msgs()
return []
def get_new_msgs(self):
return self._get_chatbox().get_new_msgs()
def get_msg_by_id(self, msg_id: str):
return self._get_chatbox().get_msg_by_id(msg_id)
version_error_msg = """
错误:未找到可用的微信窗口
当前版本仅适用于3.9版本客户端如果您当前客户端版本为4.x请在官网下载3.9版本客户端
下载链接https://pc.weixin.qq.com
"""
class WeChatMainWnd(WeChatSubWnd):
_ui_cls_name: str = 'WeChatMainWndForPC'
_ui_name: str = '微信'
def __init__(self, hwnd: int = None):
self.root = self
self.parent = self
if hwnd:
self._setup_ui(hwnd)
else:
hwnd = FindWindow(classname=self._ui_cls_name)
if not hwnd:
raise Exception(version_error_msg)
self._setup_ui(hwnd)
print(f'初始化成功,获取到已登录窗口:{self.nickname}')
def _setup_ui(self, hwnd: int):
self.HWND = hwnd
self.control = uia.ControlFromHandle(hwnd)
MainControl1 = [i for i in self.control.GetChildren() if not i.ClassName][0]
MainControl2 = MainControl1.GetFirstChildControl()
navigation_control, sessionbox_control, chatbox_control = MainControl2.GetChildren()
self.navigation = NavigationBox(navigation_control, self)
self.sessionbox = SessionBox(sessionbox_control, self)
self.chatbox = ChatBox(chatbox_control, self)
self.nickname = self.navigation.my_icon.Name
self.NavigationBox = self.navigation.control
self.SessionBox = self.sessionbox.control
self.ChatBox = self.chatbox.control
def __repr__(self):
return f'<{PROJECT_NAME} - {self.__class__.__name__} object("{self.nickname}")>'
def _lang(self, text: str) -> str:
return WECHAT_MAIN.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def _get_chatbox(
self,
nickname: str=None,
exact: bool=False
) -> ChatBox:
if nickname and (chatbox := WeChatSubWnd(nickname, self, timeout=0)).control:
return chatbox.chatbox
else:
if nickname:
switch_result = self.sessionbox.switch_chat(keywords=nickname, exact=exact)
if not switch_result:
return None
if self.chatbox.msgbox.Exists(0.5):
return self.chatbox
def switch_chat(
self,
keywords: str,
exact: bool = False,
force: bool = False,
force_wait: Union[float, int] = 0.5
):
return self.sessionbox.switch_chat(keywords, exact, force, force_wait)
def get_all_sub_wnds(self):
sub_wxs = [i for i in GetAllWindows() if i[1] == WeChatSubWnd._ui_cls_name]
return [WeChatSubWnd(i[0], self) for i in sub_wxs]
def get_sub_wnd(self, who: str, timeout: int=0):
if hwnd := FindWindow(classname=WeChatSubWnd._ui_cls_name, name=who, timeout=timeout):
return WeChatSubWnd(hwnd, self)
def open_separate_window(self, keywords: str) -> WeChatSubWnd:
if subwin := self.get_sub_wnd(keywords):
wxlog.debug(f"{keywords} 获取到已存在的子窗口: {subwin}")
return subwin
self._show()
if nickname := self.sessionbox.switch_chat(keywords):
wxlog.debug(f"{keywords} 切换到聊天窗口: {nickname}")
if subwin := self.get_sub_wnd(nickname):
wxlog.debug(f"{nickname} 获取到已存在的子窗口: {subwin}")
return subwin
else:
keywords = nickname
if result := self.sessionbox.open_separate_window(keywords):
find_nickname = result['data'].get('nickname', keywords)
return WeChatSubWnd(find_nickname, self)
def _get_next_new_message(self, filter_mute: bool=False):
def get_new_message(session):
last_content = session.content
new_count = session.new_count
chat_name = session.name
session.click()
return self.chatbox.get_next_new_msgs(new_count, last_content)
def get_new_session(filter_mute):
sessions = self.sessionbox.get_session()
if sessions[0].name == self._lang('折叠的群聊'):
self.navigation.chat_icon.DoubleClick()
sessions = self.sessionbox.get_session()
new_sessions = [
i for i in sessions
if i.isnew
and i.name != self._lang('折叠的群聊')
]
if filter_mute:
new_sessions = [i for i in new_sessions if i.ismute == False]
return new_sessions
if new_msgs := self.chatbox.get_new_msgs():
wxlog.debug("获取当前页面新消息")
return new_msgs
elif new_sessions := get_new_session(filter_mute):
wxlog.debug("当前会话列表获取新消息")
return get_new_message(new_sessions[0])
else:
self.sessionbox.go_top()
if new_sessions := get_new_session(filter_mute):
wxlog.debug("当前会话列表获取新消息")
return get_new_message(new_sessions[0])
else:
self.navigation.chat_icon.DoubleClick()
if new_sessions := get_new_session(filter_mute):
wxlog.debug("翻页会话列表获取新消息")
return get_new_message(new_sessions[0])
else:
wxlog.debug("没有新消息")
return []
def get_next_new_message(self, filter_mute: bool=False):
if filter_mute and not self.navigation.has_new_message():
return {}
new_msgs = self._get_next_new_message(filter_mute)
if new_msgs:
chat_info = self.chatbox.get_info()
return {
'chat_name': chat_info['chat_name'],
'chat_type': chat_info['chat_type'],
'msg': new_msgs
}
else:
return {}

View File

@@ -0,0 +1,55 @@
from __future__ import annotations
from wxauto.param import (
WxParam,
)
from wxauto.languages import *
from wxauto.uiautomation import Control
class NavigationBox:
def __init__(self, control, parent):
self.control: Control = control
self.root = parent.root
self.parent = parent
self.top_control = control.GetTopLevelControl()
self.init()
def _lang(self, text: str) -> str:
return WECHAT_NAVIGATION_BOX.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def init(self):
self.my_icon = self.control.ButtonControl()
self.chat_icon = self.control.ButtonControl(Name=self._lang('聊天'))
self.contact_icon = self.control.ButtonControl(Name=self._lang('通讯录'))
self.favorites_icon = self.control.ButtonControl(Name=self._lang('收藏'))
self.files_icon = self.control.ButtonControl(Name=self._lang('聊天文件'))
self.moments_icon = self.control.ButtonControl(Name=self._lang('朋友圈'))
self.browser_icon = self.control.ButtonControl(Name=self._lang('搜一搜'))
self.video_icon = self.control.ButtonControl(Name=self._lang('视频号'))
self.stories_icon = self.control.ButtonControl(Name=self._lang('看一看'))
self.mini_program_icon = self.control.ButtonControl(Name=self._lang('小程序面板'))
self.phone_icon = self.control.ButtonControl(Name=self._lang('手机'))
self.settings_icon = self.control.ButtonControl(Name=self._lang('设置及其他'))
def switch_to_chat_page(self):
self.chat_icon.Click()
def switch_to_contact_page(self):
self.contact_icon.Click()
def switch_to_favorites_page(self):
self.favorites_icon.Click()
def switch_to_files_page(self):
self.files_icon.Click()
def switch_to_browser_page(self):
self.browser_icon.Click()
# 是否有新消息
def has_new_message(self):
from wxauto.utils.win32 import capture
rect = self.chat_icon.BoundingRectangle
bbox = rect.left, rect.top, rect.right, rect.bottom
img = capture(self.root.HWND, bbox)
return any(p[0] > p[1] and p[0] > p[2] for p in img.getdata())

265
wxauto/ui/sessionbox.py Normal file
View File

@@ -0,0 +1,265 @@
from __future__ import annotations
from wxauto.ui.component import (
CMenuWnd
)
from wxauto.param import (
WxParam,
WxResponse,
)
from wxauto.languages import *
from wxauto.utils import (
SetClipboardText,
)
from wxauto.logger import wxlog
from wxauto.uiautomation import Control
from wxauto.utils.tools import roll_into_view
from typing import (
List,
Union
)
import time
import re
class SessionBox:
def __init__(self, control, parent):
self.control: Control = control
self.root = parent.root
self.parent = parent
self.top_control = control.GetTopLevelControl()
self.init()
def _lang(self, text: str) -> str:
return WECHAT_SESSION_BOX.get(text, {WxParam.LANGUAGE: text}).get(WxParam.LANGUAGE)
def init(self):
self.searchbox = self.control.EditControl(Name=self._lang('搜索'))
self.session_list =\
self.control.ListControl(Name=self._lang('会话'), searchDepth=7)
self.archived_session_list =\
self.control.ListControl(Name=self._lang('折叠的群聊'), searchDepth=7)
def get_session(self) -> List[SessionElement]:
if self.session_list.Exists(0):
return [
SessionElement(i, self)
for i in self.session_list.GetChildren()
if i.Name != self._lang('折叠置顶聊天')
and not re.match(self._lang('re_置顶聊天'), i.Name)
]
elif self.archived_session_list.Exists(0):
return [SessionElement(i, self) for i in self.archived_session_list.GetChildren()]
else:
return []
def roll_up(self, n: int=5):
self.control.MiddleClick()
self.control.WheelUp(wheelTimes=n)
def roll_down(self, n: int=5):
self.control.MiddleClick()
self.control.WheelDown(wheelTimes=n)
def switch_chat(
self,
keywords: str,
exact: bool = False,
force: bool = False,
force_wait: Union[float, int] = 0.5
):
wxlog.debug(f"切换聊天窗口: {keywords}, {exact}, {force}, {force_wait}")
self.root._show()
sessions = self.get_session()
for session in sessions:
if (
keywords == session.name
and session.control.BoundingRectangle.height()
):
session.switch()
return keywords
self.searchbox.RightClick()
SetClipboardText(keywords)
menu = CMenuWnd(self)
menu.select('粘贴')
search_result = self.control.ListControl(RegexName='.*?IDS_FAV_SEARCH_RESULT.*?')
if force:
time.sleep(force_wait)
self.searchbox.SendKeys('{ENTER}')
return ''
t0 = time.time()
while time.time() -t0 < WxParam.SEARCH_CHAT_TIMEOUT:
results = []
search_result_items = search_result.GetChildren()
highlight_who = re.sub(r'(\s+)', r'</em>\1<em>', keywords)
for search_result_item in search_result_items:
item_name = search_result_item.Name
if (
search_result_item.ControlTypeName == 'PaneControl'
and search_result_item.TextControl(Name='聊天记录').Exists(0)
) or item_name == f'搜索 {keywords}':
break
elif (
search_result_item.ControlTypeName == 'ListItemControl'
and search_result_item.TextControl(Name=f"微信号: <em>{keywords}</em>").Exists(0)
):
wxlog.debug(f"{keywords} 匹配到微信号:{item_name}")
search_result_item.Click()
return item_name
elif (
search_result_item.ControlTypeName == 'ListItemControl'
and search_result_item.TextControl(Name=f"昵称: <em>{highlight_who}</em>").Exists(0)
):
wxlog.debug(f"{keywords} 匹配到昵称:{item_name}")
search_result_item.Click()
return item_name
elif (
search_result_item.ControlTypeName == 'ListItemControl'
and search_result_item.TextControl(Name=f"群聊名称: <em>{highlight_who}</em>").Exists(0)
):
wxlog.debug(f"{keywords} 匹配到群聊名称:{item_name}")
search_result_item.Click()
return item_name
elif (
search_result_item.ControlTypeName == 'ListItemControl'
and keywords == item_name
):
wxlog.debug(f"{keywords} 完整匹配")
search_result_item.Click()
return keywords
elif (
search_result_item.ControlTypeName == 'ListItemControl'
and keywords in item_name
):
results.append(search_result_item)
if exact:
wxlog.debug(f"{keywords} 未精准匹配返回None")
if search_result.Exists(0):
search_result.SendKeys('{Esc}')
return None
if results:
wxlog.debug(f"{keywords} 匹配到多个结果,返回第一个")
results[0].Click()
return results[0].Name
if search_result.Exists(0):
search_result.SendKeys('{Esc}')
def open_separate_window(self, name: str):
wxlog.debug(f"打开独立窗口: {name}")
sessions = self.get_session()
for session in sessions:
if session.name == name:
wxlog.debug(f"找到会话: {name}")
while session.control.BoundingRectangle.height():
try:
session.click()
session.double_click()
except:
pass
time.sleep(0.1)
else:
return WxResponse.success(data={'nickname': name})
wxlog.debug(f"未找到会话: {name}")
return WxResponse.failure('未找到会话')
def go_top(self):
wxlog.debug("回到会话列表顶部")
if self.archived_session_list.Exists(0):
self.control.ButtonControl(Name=self._lang('返回')).Click()
time.sleep(0.3)
first_session_name = self.session_list.GetChildren()[0].Name
while True:
self.control.WheelUp(wheelTimes=3)
time.sleep(0.1)
if self.session_list.GetChildren()[0].Name == first_session_name:
break
else:
first_session_name = self.session_list.GetChildren()[0].Name
class SessionElement:
def __init__(
self,
control: Control,
parent: SessionBox
):
self.root = parent.root
self.parent = parent
self.control = control
info_controls = [i for i in self.control.GetProgenyControl(3).GetChildren() if i.ControlTypeName=='TextControl']
self.name = info_controls[0].Name
self.time = info_controls[-1].Name
self.content = (
temp_control.Name
if (temp_control := control.GetProgenyControl(4, -1, control_type='TextControl'))
else None
)
self.ismute = (
True
if control.GetProgenyControl(4, 1, control_type='PaneControl')
else False
)
self.isnew = (new_tag_control := control.GetProgenyControl(2, 2)) is not None
self.new_count = 0
if self.isnew:
if new_tag_name := (new_tag_control.Name):
try:
self.new_count = int(new_tag_name)
self.ismute = False
except ValueError:
self.new_count = 999
else:
new_text = re.findall(self._lang('re_条数'), str(self.content))
if new_text:
try:
self.new_count = int(re.findall('\d+', new_text[0])[0])
except ValueError:
self.new_count = 999
self.content = self.content[len(new_text[0])+1:]
else:
self.new_count = 1
self.info = {
'name': self.name,
'time': self.time,
'content': self.content,
'isnew': self.isnew,
'new_count': self.new_count,
'ismute': self.ismute
}
def _lang(self, text: str) -> str:
return self.parent._lang(text)
def roll_into_view(self):
self.root._show()
roll_into_view(self.control.GetParentControl(), self.control)
def _click(self, right: bool=False, double: bool=False):
self.roll_into_view()
if right:
self.control.RightClick()
elif double:
self.control.DoubleClick()
else:
self.control.Click()
def click(self):
self._click()
def right_click(self):
self._click(right=True)
def double_click(self):
self._click(double=True)
def switch(self):
self.click()

8142
wxauto/uiautomation.py Normal file

File diff suppressed because it is too large Load Diff

2
wxauto/utils/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .win32 import *
from . import tools

102
wxauto/utils/tools.py Normal file
View File

@@ -0,0 +1,102 @@
from pathlib import Path
from datetime import datetime, timedelta
import time
import re
def get_file_dir(dir_path=None):
if dir_path is None:
dir_path = Path('.').absolute()
elif isinstance(dir_path, str):
dir_path = Path(dir_path)
dir_path.mkdir(parents=True, exist_ok=True)
return dir_path
def now_time(fmt='%Y%m%d%H%M%S%f'):
return datetime.now().strftime(fmt)
def parse_wechat_time(time_str):
"""
时间格式转换函数
Args:
time_str: 输入的时间字符串
Returns:
转换后的时间字符串
"""
time_str = time_str.replace('星期天', '星期日')
match = re.match(r'^(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})$', time_str)
if match:
month, day, hour, minute, second = match.groups()
current_year = datetime.now().year
return datetime(current_year, int(month), int(day), int(hour), int(minute), int(second)).strftime('%Y-%m-%d %H:%M:%S')
match = re.match(r'^(\d{1,2}):(\d{1,2})$', time_str)
if match:
hour, minute = match.groups()
return datetime.now().strftime('%Y-%m-%d') + f' {hour}:{minute}:00'
match = re.match(r'^昨天 (\d{1,2}):(\d{1,2})$', time_str)
if match:
hour, minute = match.groups()
yesterday = datetime.now() - timedelta(days=1)
return yesterday.strftime('%Y-%m-%d') + f' {hour}:{minute}:00'
match = re.match(r'^星期([一二三四五六日]) (\d{1,2}):(\d{1,2})$', time_str)
if match:
weekday, hour, minute = match.groups()
weekday_num = ['', '', '', '', '', '', ''].index(weekday)
today_weekday = datetime.now().weekday()
delta_days = (today_weekday - weekday_num) % 7
target_day = datetime.now() - timedelta(days=delta_days)
return target_day.strftime('%Y-%m-%d') + f' {hour}:{minute}:00'
match = re.match(r'^(\d{4})年(\d{1,2})月(\d{1,2})日 (\d{1,2}):(\d{1,2})$', time_str)
if match:
year, month, day, hour, minute = match.groups()
return datetime(*[int(i) for i in [year, month, day, hour, minute]]).strftime('%Y-%m-%d %H:%M:%S')
match = re.match(r'^(\d{2})-(\d{2}) (上午|下午) (\d{1,2}):(\d{2})$', time_str)
if match:
month, day, period, hour, minute = match.groups()
current_year = datetime.now().year
hour = int(hour)
if period == '下午' and hour != 12:
hour += 12
elif period == '上午' and hour == 12:
hour = 0
return datetime(current_year, int(month), int(day), hour, int(minute)).strftime('%Y-%m-%d %H:%M:%S')
return time_str
def roll_into_view(win, ele, equal=False, bias=0):
while ele.BoundingRectangle.ycenter() < win.BoundingRectangle.top + bias or ele.BoundingRectangle.ycenter() >= win.BoundingRectangle.bottom - bias:
if ele.BoundingRectangle.ycenter() < win.BoundingRectangle.top + bias:
# 上滚动
while True:
if not ele.Exists(0) or not ele.BoundingRectangle.height():
return 'not exist'
win.WheelUp(wheelTimes=1)
time.sleep(0.1)
if equal:
if ele.BoundingRectangle.ycenter() >= win.BoundingRectangle.top + bias:
break
else:
if ele.BoundingRectangle.ycenter() > win.BoundingRectangle.top + bias:
break
elif ele.BoundingRectangle.ycenter() >= win.BoundingRectangle.bottom - bias:
# 下滚动
while True:
if not ele.Exists(0) or not ele.BoundingRectangle.height():
return 'not exist'
win.WheelDown(wheelTimes=1)
time.sleep(0.1)
if equal:
if ele.BoundingRectangle.ycenter() <= win.BoundingRectangle.bottom - bias:
break
else:
if ele.BoundingRectangle.ycenter() < win.BoundingRectangle.bottom - bias:
break
time.sleep(0.3)

302
wxauto/utils/win32.py Normal file
View File

@@ -0,0 +1,302 @@
import os
import time
import shutil
import win32ui
import win32gui
import win32api
import win32con
import win32process
import win32clipboard
import pyperclip
import ctypes
from PIL import Image
from wxauto import uiautomation as uia
def GetAllWindows():
"""
获取所有窗口的信息,返回一个列表,每个元素包含 (窗口句柄, 类名, 窗口标题)
"""
windows = []
def enum_windows_proc(hwnd, extra):
class_name = win32gui.GetClassName(hwnd) # 获取窗口类名
window_title = win32gui.GetWindowText(hwnd) # 获取窗口标题
windows.append((hwnd, class_name, window_title))
win32gui.EnumWindows(enum_windows_proc, None)
return windows
def GetCursorWindow():
x, y = win32api.GetCursorPos()
hwnd = win32gui.WindowFromPoint((x, y))
window_title = win32gui.GetWindowText(hwnd)
class_name = win32gui.GetClassName(hwnd)
return hwnd, window_title, class_name
def set_cursor_pos(x, y):
win32api.SetCursorPos((x, y))
def Click(rect):
x = (rect.left + rect.right) // 2
y = (rect.top + rect.bottom) // 2
set_cursor_pos(x, y)
win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)
def GetPathByHwnd(hwnd):
"""通过窗口句柄获取进程可执行文件路径 - 使用pywin32"""
try:
thread_id, process_id = win32process.GetWindowThreadProcessId(hwnd)
# 获取进程句柄
PROCESS_QUERY_INFORMATION = 0x0400
PROCESS_VM_READ = 0x0010
process_handle = win32api.OpenProcess(
PROCESS_QUERY_INFORMATION | PROCESS_VM_READ,
False,
process_id
)
# 获取可执行文件路径
exe_path = win32process.GetModuleFileNameEx(process_handle, 0)
# 关闭句柄
win32api.CloseHandle(process_handle)
return exe_path
except Exception as e:
print(f"Error: {e}")
return None
def GetVersionByPath(file_path):
try:
info = win32api.GetFileVersionInfo(file_path, '\\')
version = "{}.{}.{}.{}".format(win32api.HIWORD(info['FileVersionMS']),
win32api.LOWORD(info['FileVersionMS']),
win32api.HIWORD(info['FileVersionLS']),
win32api.LOWORD(info['FileVersionLS']))
except:
version = None
return version
def GetWindowRect(hwnd):
return win32gui.GetWindowRect(hwnd)
def capture(hwnd, bbox):
# 获取窗口的屏幕坐标
window_rect = win32gui.GetWindowRect(hwnd)
win_left, win_top, win_right, win_bottom = window_rect
win_width = win_right - win_left
win_height = win_bottom - win_top
# 获取窗口的设备上下文
hwndDC = win32gui.GetWindowDC(hwnd)
mfcDC = win32ui.CreateDCFromHandle(hwndDC)
saveDC = mfcDC.CreateCompatibleDC()
# 创建位图对象保存整个窗口截图
saveBitMap = win32ui.CreateBitmap()
saveBitMap.CreateCompatibleBitmap(mfcDC, win_width, win_height)
saveDC.SelectObject(saveBitMap)
# 使用PrintWindow捕获整个窗口包括被遮挡或最小化的窗口
result = ctypes.windll.user32.PrintWindow(hwnd, saveDC.GetSafeHdc(), 3)
# 转换为PIL图像
bmpinfo = saveBitMap.GetInfo()
bmpstr = saveBitMap.GetBitmapBits(True)
im = Image.frombuffer(
'RGB',
(bmpinfo['bmWidth'], bmpinfo['bmHeight']),
bmpstr, 'raw', 'BGRX', 0, 1)
# 释放资源
win32gui.DeleteObject(saveBitMap.GetHandle())
saveDC.DeleteDC()
mfcDC.DeleteDC()
win32gui.ReleaseDC(hwnd, hwndDC)
# 计算bbox相对于窗口左上角的坐标
bbox_left, bbox_top, bbox_right, bbox_bottom = bbox
# 转换为截图图像中的相对坐标
crop_left = bbox_left - win_left
crop_top = bbox_top - win_top
crop_right = bbox_right - win_left
crop_bottom = bbox_bottom - win_top
# 裁剪目标区域
cropped_im = im.crop((crop_left, crop_top, crop_right, crop_bottom))
return cropped_im
def GetText(HWND):
length = win32gui.SendMessage(HWND, win32con.WM_GETTEXTLENGTH)*2
buffer = win32gui.PyMakeBuffer(length)
win32api.SendMessage(HWND, win32con.WM_GETTEXT, length, buffer)
address, length_ = win32gui.PyGetBufferAddressAndLen(buffer[:-1])
text = win32gui.PyGetString(address, length_)[:int(length/2)]
buffer.release()
return text
def GetAllWindowExs(HWND):
if not HWND:
return
handles = []
win32gui.EnumChildWindows(
HWND, lambda hwnd, param: param.append([hwnd, win32gui.GetClassName(hwnd), GetText(hwnd)]), handles)
return handles
def FindWindow(classname=None, name=None, timeout=0) -> int:
t0 = time.time()
while True:
HWND = win32gui.FindWindow(classname, name)
if HWND:
break
if time.time() - t0 > timeout:
break
time.sleep(0.01)
return HWND
def FindTopLevelControl(classname=None, name=None, timeout=3):
hwnd = FindWindow(classname, name, timeout)
if hwnd:
return uia.ControlFromHandle(hwnd)
else:
return None
def FindWinEx(HWND, classname=None, name=None) -> list:
hwnds_classname = []
hwnds_name = []
def find_classname(hwnd, classname):
classname_ = win32gui.GetClassName(hwnd)
if classname_ == classname:
if hwnd not in hwnds_classname:
hwnds_classname.append(hwnd)
def find_name(hwnd, name):
name_ = GetText(hwnd)
if name in name_:
if hwnd not in hwnds_name:
hwnds_name.append(hwnd)
if classname:
win32gui.EnumChildWindows(HWND, find_classname, classname)
if name:
win32gui.EnumChildWindows(HWND, find_name, name)
if classname and name:
hwnds = [hwnd for hwnd in hwnds_classname if hwnd in hwnds_name]
else:
hwnds = hwnds_classname + hwnds_name
return hwnds
def ClipboardFormats(unit=0, *units):
units = list(units)
retry_count = 5
while retry_count > 0:
try:
win32clipboard.OpenClipboard()
try:
u = win32clipboard.EnumClipboardFormats(unit)
finally:
win32clipboard.CloseClipboard()
break
except Exception as e:
retry_count -= 1
units.append(u)
if u:
units = ClipboardFormats(u, *units)
return units
def ReadClipboardData():
Dict = {}
formats = ClipboardFormats()
for i in formats:
if i == 0:
continue
retry_count = 5
while retry_count > 0:
try:
win32clipboard.OpenClipboard()
try:
data = win32clipboard.GetClipboardData(i)
Dict[str(i)] = data
finally:
win32clipboard.CloseClipboard()
break
except Exception as e:
retry_count -= 1
return Dict
def SetClipboardText(text: str):
pyperclip.copy(text)
class DROPFILES(ctypes.Structure):
_fields_ = [
("pFiles", ctypes.c_uint),
("x", ctypes.c_long),
("y", ctypes.c_long),
("fNC", ctypes.c_int),
("fWide", ctypes.c_bool),
]
pDropFiles = DROPFILES()
pDropFiles.pFiles = ctypes.sizeof(DROPFILES)
pDropFiles.fWide = True
matedata = bytes(pDropFiles)
def SetClipboardFiles(paths):
for file in paths:
if not os.path.exists(file):
raise FileNotFoundError(f"file ({file}) not exists!")
files = ("\0".join(paths)).replace("/", "\\")
data = files.encode("U16")[2:]+b"\0\0"
t0 = time.time()
while True:
if time.time() - t0 > 10:
raise TimeoutError(f"设置剪贴板文件超时! --> {paths}")
try:
win32clipboard.OpenClipboard()
win32clipboard.EmptyClipboard()
win32clipboard.SetClipboardData(win32clipboard.CF_HDROP, matedata+data)
break
except:
pass
finally:
try:
win32clipboard.CloseClipboard()
except:
pass
def PasteFile(folder):
folder = os.path.realpath(folder)
if not os.path.exists(folder):
os.makedirs(folder)
t0 = time.time()
while True:
if time.time() - t0 > 10:
raise TimeoutError(f"读取剪贴板文件超时!")
try:
win32clipboard.OpenClipboard()
if win32clipboard.IsClipboardFormatAvailable(win32clipboard.CF_HDROP):
files = win32clipboard.GetClipboardData(win32clipboard.CF_HDROP)
for file in files:
filename = os.path.basename(file)
dest_file = os.path.join(folder, filename)
shutil.copy2(file, dest_file)
return True
else:
print("剪贴板中没有文件")
return False
except:
pass
finally:
win32clipboard.CloseClipboard()
def IsRedPixel(uicontrol):
rect = uicontrol.BoundingRectangle
hwnd = uicontrol.GetAncestorControl(lambda x,y:x.ClassName=='WeChatMainWndForPC').NativeWindowHandle
bbox = (rect.left, rect.top, rect.right, rect.bottom)
img = capture(hwnd, bbox)
return any(p[0] > p[1] and p[0] > p[2] for p in img.getdata())

386
wxauto/wx.py Normal file
View File

@@ -0,0 +1,386 @@
from .ui.main import (
WeChatMainWnd,
WeChatSubWnd
)
from .ui.component import (
WeChatLoginWnd
)
from .param import (
WxResponse,
WxParam,
PROJECT_NAME
)
from .logger import wxlog
from typing import (
Union,
List,
Dict,
Callable,
TYPE_CHECKING
)
from PIL import Image
from abc import ABC, abstractmethod
import threading
import traceback
import time
import sys
if TYPE_CHECKING:
from wxauto.msgs.base import Message
from wxauto.ui.sessionbox import SessionElement
class Listener(ABC):
def _listener_start(self):
wxlog.debug('开始监听')
self._listener_is_listening = True
self._listener_messages = {}
self._lock = threading.RLock()
self._listener_stop_event = threading.Event()
self._listener_thread = threading.Thread(target=self._listener_listen, daemon=True)
self._listener_thread.start()
def _listener_listen(self):
if not hasattr(self, 'listen') or not self.listen:
self.listen = {}
while not self._listener_stop_event.is_set():
try:
self._get_listen_messages()
except:
wxlog.debug(f'监听消息失败:{traceback.format_exc()}')
time.sleep(WxParam.LISTEN_INTERVAL)
def _safe_callback(self, callback, msg, chat):
try:
with self._lock:
callback(msg, chat)
except Exception as e:
wxlog.debug(f"监听消息回调发生错误:{traceback.format_exc()}")
def _listener_stop(self):
self._listener_is_listening = False
self._listener_stop_event.set()
self._listener_thread.join()
@abstractmethod
def _get_listen_messages(self):
...
class Chat:
"""微信聊天窗口实例"""
def __init__(self, core: WeChatSubWnd=None):
self.core = core
self.who = self.core.nickname
def __repr__(self):
return f'<{PROJECT_NAME} - {self.__class__.__name__} object("{self.core.nickname}")>'
@property
def chat_type(self):
return self.core.chatbox.get_info().get('chat_type', None)
def Show(self):
"""显示窗口"""
self.core._show()
def ChatInfo(self) -> Dict[str, str]:
"""获取聊天窗口信息
Returns:
dict: 聊天窗口信息
"""
return self.core.chatbox.get_info()
def SendMsg(
self,
msg: str,
who: str=None,
clear: bool=True,
at: Union[str, List[str]]=None,
exact: bool=False,
) -> WxResponse:
"""发送消息
Args:
msg (str): 消息内容
who (str, optional): 发送对象,不指定则发送给当前聊天对象,**当子窗口时,该参数无效**
clear (bool, optional): 发送后是否清空编辑框.
at (Union[str, List[str]], optional): @对象,不指定则不@任何人
exact (bool, optional): 搜索who好友时是否精确匹配默认False**当子窗口时,该参数无效**
Returns:
WxResponse: 是否发送成功
"""
return self.core.send_msg(msg, who, clear, at, exact)
def SendFiles(
self,
filepath,
who=None,
exact=False
) -> WxResponse:
"""向当前聊天窗口发送文件
Args:
filepath (str|list): 要复制文件的绝对路径
who (str): 发送对象,不指定则发送给当前聊天对象,**当子窗口时,该参数无效**
exact (bool, optional): 搜索who好友时是否精确匹配默认False**当子窗口时,该参数无效**
Returns:
WxResponse: 是否发送成功
"""
return self.core.send_files(filepath, who, exact)
def LoadMoreMessage(self, interval: float=0.3) -> WxResponse:
"""加载更多消息
Args:
interval (float, optional): 滚动间隔单位秒默认0.3
"""
return self.core.load_more_message(interval)
def GetAllMessage(self) -> List['Message']:
"""获取当前聊天窗口的所有消息
Returns:
List[Message]: 当前聊天窗口的所有消息
"""
return self.core.get_msgs()
def GetNewMessage(self) -> List['Message']:
"""获取当前聊天窗口的新消息
Returns:
List[Message]: 当前聊天窗口的新消息
"""
if not hasattr(self, '_last_chat'):
self._last_chat = self.ChatInfo().get('chat_name')
if (_last_chat := self.ChatInfo().get('chat_name')) != self._last_chat:
self._last_chat = _last_chat
self.core.chatbox._update_used_msg_ids()
return []
return self.core.get_new_msgs()
def GetMessageById(self, msg_id: str) -> 'Message':
"""根据消息id获取消息
Args:
msg_id (str): 消息id
Returns:
Message: 消息对象
"""
return self.core.get_msg_by_id(msg_id)
def GetGroupMembers(self) -> List[str]:
"""获取当前聊天群成员
Returns:
list: 当前聊天群成员列表
"""
return self.core.get_group_members()
def Close(self) -> None:
"""关闭微信窗口"""
self.core.close()
class WeChat(Chat, Listener):
"""微信主窗口实例"""
def __init__(
self,
debug: bool=False,
**kwargs
):
hwnd = None
if 'hwnd' in kwargs:
hwnd = kwargs['hwnd']
self.core = WeChatMainWnd(hwnd)
self.nickname = self.core.nickname
self.SessionBox = self.core.sessionbox
self.listen = {}
if debug:
wxlog.set_debug(True)
wxlog.debug('Debug mode is on')
self._listener_start()
self.Show()
def _get_listen_messages(self):
try:
sys.stdout.flush()
except:
pass
temp_listen = self.listen.copy()
for who in temp_listen:
chat, callback = temp_listen.get(who, (None, None))
try:
if chat is None or not chat.core.exists():
wxlog.debug(f"窗口 {who} 已关闭,移除监听")
self.RemoveListenChat(who, close_window=False)
continue
except:
continue
with self._lock:
msgs = chat.GetNewMessage()
for msg in msgs:
wxlog.debug(f"[{msg.attr} {msg.type}]获取到新消息:{who} - {msg.content}")
chat.Show()
self._safe_callback(callback, msg, chat)
def GetSession(self) -> List['SessionElement']:
"""获取当前会话列表
Returns:
List[SessionElement]: 当前会话列表
"""
return self.core.sessionbox.get_session()
def ChatWith(
self,
who: str,
exact: bool=False,
force: bool=False,
force_wait: Union[float, int] = 0.5
):
"""打开聊天窗口
Args:
who (str): 要聊天的对象
exact (bool, optional): 搜索who好友时是否精确匹配默认False
force (bool, optional): 不论是否匹配到都强制切换若启用则exact参数无效默认False
> 注force原理为输入搜索关键字后在等待`force_wait`秒后不判断结果直接回车,谨慎使用
force_wait (Union[float, int], optional): 强制切换时等待时间默认0.5秒
"""
return self.core.switch_chat(who, exact, force, force_wait)
def AddListenChat(
self,
nickname: str,
callback: Callable[['Message', str], None],
) -> WxResponse:
"""添加监听聊天将聊天窗口独立出去形成Chat对象子窗口用于监听
Args:
nickname (str): 要监听的聊天对象
callback (Callable[[Message, str], None]): 回调函数,参数为(Message对象, 聊天名称)
"""
if nickname in self.listen:
return WxResponse.failure('该聊天已监听')
subwin = self.core.open_separate_window(nickname)
if subwin is None:
return WxResponse.failure('找不到聊天窗口')
name = subwin.nickname
chat = Chat(subwin)
self.listen[name] = (chat, callback)
return chat
def StopListening(self, remove: bool = True) -> None:
"""停止监听"""
while self._listener_thread.is_alive():
self._listener_stop()
if remove:
listen = self.listen.copy()
for who in listen:
self.RemoveListenChat(who)
def StartListening(self) -> None:
if not self._listener_thread.is_alive():
self._listener_start()
def RemoveListenChat(
self,
nickname: str,
close_window: bool = True
) -> WxResponse:
"""移除监听聊天
Args:
nickname (str): 要移除的监听聊天对象
close_window (bool, optional): 是否关闭聊天窗口. Defaults to True.
Returns:
WxResponse: 执行结果
"""
if nickname not in self.listen:
return WxResponse.failure('未找到监听对象')
chat, _ = self.listen[nickname]
if close_window:
chat.Close()
del self.listen[nickname]
return WxResponse.success()
def GetNextNewMessage(self, filter_mute=False) -> Dict[str, List['Message']]:
"""获取下一个新消息
Args:
filter_mute (bool, optional): 是否过滤掉免打扰消息. Defaults to False.
Returns:
Dict[str, List['Message']]: 消息列表
"""
return self.core.get_next_new_message(filter_mute)
def SwitchToChat(self) -> None:
"""切换到聊天页面"""
self.core.navigation.chat_icon.Click()
def SwitchToContact(self) -> None:
"""切换到联系人页面"""
self.core.navigation.contact_icon.Click()
def GetSubWindow(self, nickname: str) -> 'Chat':
"""获取子窗口实例
Args:
nickname (str): 要获取的子窗口的昵称
Returns:
Chat: 子窗口实例
"""
if subwin := self.core.get_sub_wnd(nickname):
return Chat(subwin)
def GetAllSubWindow(self) -> List['Chat']:
"""获取所有子窗口实例
Returns:
List[Chat]: 所有子窗口实例
"""
return [Chat(subwin) for subwin in self.core.get_all_sub_wnds()]
def KeepRunning(self):
"""保持运行"""
while not self._listener_stop_event.is_set():
try:
time.sleep(1)
except KeyboardInterrupt:
wxlog.debug(f'wxauto("{self.nickname}") shutdown')
self.StopListening(True)
break
class WeChatLogin:
def ClearHint(self):
if loginWnd := WeChatLoginWnd() :
return loginWnd.clear_hint()
else:
return False
def Login(self):
"""登录"""
if loginWnd := WeChatLoginWnd() :
return loginWnd.login()
else:
return False
def GetQRCode(self) -> Image.Image:
"""获取二维码"""
if loginWnd := WeChatLoginWnd() :
return loginWnd.get_qrcode()
else:
return None