完成基础框架

This commit is contained in:
evilbeast 2022-08-23 12:05:50 +08:00
commit 311a281d82
15 changed files with 539 additions and 0 deletions

26
.gitignore vendored Normal file
View File

@ -0,0 +1,26 @@
# If you need to exclude files such as those generated by an IDE, use
# $GIT_DIR/info/exclude or the mm.excludesFile configuration variable as
# described in https://git-scm.com/docs/gitignore
*.egg-info
*.pot
*.py[co]
.tox/
__pycache__
MANIFEST
dist/
docs/_build/
docs/locale/
node_modules/
tests/coverage_html/
tests/.coverage
build/
tests/report/
venv
.idea
log/
*.c
main.spec
build/
config.ini
ntchat/wc/*.pyd

7
ntchat/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from .conf import VERSION
from .core.wechat import WeChat
from .wc import wcprobe
__version__ = VERSION
exit_ = wcprobe.exit

1
ntchat/conf/__init__.py Normal file
View File

@ -0,0 +1 @@
VERSION = '0.1.0'

0
ntchat/const/__init__.py Normal file
View File

37
ntchat/const/wx_type.py Normal file
View File

@ -0,0 +1,37 @@
MT_READY_MSG = 11024
MT_USER_LOGIN_MSG = 11025
MT_USER_LOGOUT_MSG = 11026
MT_GET_SELF_MSG = 11028
MT_GET_CONTACTS_MSG = 11030
MT_GET_ROOMS_MSG = 11031
MT_GET_ROOM_MEMBERS_MSG = 11032
MT_GET_CONTACT_DETAIL_MSG = 11034
# 发送消息
MT_SEND_TEXT_MSG = 11036
MT_SEND_ROOM_AT_MSG = 11037
MT_SEND_CARD_MSG = 11038
MT_SEND_LINK_MSG = 11039
MT_SEND_IMAGE_MSG = 11040
MT_SEND_FILE_MSG = 11041
MT_SEND_VIDEO_MSG = 11042
MT_SEND_GIF_MSG = 11043
# 接收消息类
MT_RECV_TEXT_MSG = 11046
MT_RECV_PICTURE_MSG = 11047
MT_RECV_VOICE_MSG = 11048
MT_RECV_FRIEND_MSG = 11049
MT_RECV_CARD_MSG = 11050
MT_RECV_VIDEO_MSG = 11051
MT_RECV_EMOJI_MSG = 11052
MT_RECV_LOCATION_MSG = 11053
MT_RECV_LINK_MSG = 11054
MT_RECV_FILE_MSG = 11055
MT_RECV_MINIAPP_MSG = 11056
MT_RECV_WCPAY_MSG = 11057
MT_RECV_SYSTEM_MSG = 11058
MT_RECV_REVOKE_MSG = 11059
MT_RECV_OTHER_MSG = 11060
MT_RECV_OTHER_APP_MSG = 11061

0
ntchat/core/__init__.py Normal file
View File

74
ntchat/core/mgr.py Normal file
View File

@ -0,0 +1,74 @@
import json
import os.path
from ntchat.wc import wcprobe
from ntchat.utils.xdg import get_helper_file
from ntchat.exception import WeChatVersionNotMatchError, WeChatBindError
from ntchat.utils.singleton import Singleton
from ntchat.const import wx_type
from ntchat.utils.logger import get_logger
log = get_logger("WeChatManager")
class WeChatMgr(metaclass=Singleton):
__instance_list = []
__instance_map = {}
def __init__(self, wechat_exe_path=None, wechat_version=None):
self.set_wechat_exe_path(wechat_exe_path, wechat_version)
# init callbacks
wcprobe.init_callback(self.__on_accept, self.__on_recv, self.__on_close)
def set_wechat_exe_path(self, wechat_exe_path=None, wechat_version=None):
exe_path = ''
if wechat_exe_path is not None:
exe_path = wechat_exe_path
if wechat_version is None:
version = wcprobe.get_install_wechat_version()
else:
version = wechat_version
helper_file = get_helper_file(version)
if not os.path.exists(helper_file):
raise WeChatVersionNotMatchError()
log.info("initialize wechat, version: %s", version)
# init env
wcprobe.init_env(helper_file, exe_path)
def append_instance(self, instance):
log.debug("new wechat instance")
self.__instance_list.append(instance)
def __bind_wechat(self, client_id, pid):
bind_instance = None
if client_id not in self.__instance_map:
for instance in self.__instance_list:
if instance.pid == pid:
instance.client_id = client_id
instance.status = True
self.__instance_map[client_id] = instance
bind_instance = instance
break
if bind_instance is None:
raise WeChatBindError()
self.__instance_list.remove(bind_instance)
def __on_accept(self, client_id):
log.debug("accept client_id: %d", client_id)
def __on_recv(self, client_id, data):
message = json.loads(data)
if message["type"] == wx_type.MT_READY_MSG:
self.__bind_wechat(client_id, message["data"]["pid"])
else:
self.__instance_map[client_id].on_recv(message)
def __on_close(self, client_id):
log.debug("close client_id: %d", client_id)
if client_id in self.__instance_map:
self.__instance_map[client_id].login_status = False
self.__instance_map[client_id].status = False

256
ntchat/core/wechat.py Normal file
View File

@ -0,0 +1,256 @@
import pyee
import json
from ntchat.core.mgr import WeChatMgr
from ntchat.const import wx_type
from threading import Event
from ntchat.wc import wcprobe
from ntchat.utils import generate_guid
from ntchat.utils import logger
from ntchat.exception import WeChatNotLoginError
from functools import wraps
from typing import (
List,
Union,
Tuple
)
log = logger.get_logger("WeChatInstance")
class ReqData:
__response_message = None
msg_type: int = 0
request_data = None
def __init__(self, msg_type, data):
self.msg_type = msg_type
self.request_data = data
self.__wait_event = Event()
def wait_response(self, timeout=None):
self.__wait_event.wait(timeout)
return self.get_response_data()
def on_response(self, message):
self.__response_message = message
self.__wait_event.set()
def get_response_data(self):
if self.__response_message is None:
return None
return self.__response_message["data"]
class WeChat:
client_id: int = 0
pid: int = 0
status: bool = False
login_status: bool = False
def __init__(self):
WeChatMgr().append_instance(self)
self.__wait_login_event = Event()
self.__req_data_cache = {}
self.__msg_event_emitter = pyee.EventEmitter()
self.__login_info = {}
def on(self, msg_type, f):
return self.__msg_event_emitter.on(str(msg_type), f)
def msg_register(self, msg_type: Union[int, List[int], Tuple[int]]):
if not (isinstance(msg_type, list) or isinstance(msg_type, tuple)):
msg_type = [msg_type]
def wrapper(f):
wraps(f)
for event in msg_type:
self.on(event, f)
return f
return wrapper
def on_recv(self, message):
msg_type = message["type"]
extend = message.get("extend", None)
if msg_type == wx_type.MT_USER_LOGIN_MSG:
self.login_status = False
self.__wait_login_event.set()
self.__login_info = message.get("data", {})
elif msg_type == wx_type.MT_USER_LOGOUT_MSG:
self.login_status = False
if extend is not None and extend in self.__req_data_cache:
req_data = self.__req_data_cache[extend]
req_data.on_response(message)
del self.__req_data_cache[extend]
else:
self.__msg_event_emitter.emit(str(msg_type), self, message)
def wait_login(self, timeout=None):
log.info("wait login...")
self.__wait_login_event.wait(timeout)
def open(self, smart=False):
self.pid = wcprobe.open(smart)
log.info("open wechat pid: %d", self.pid)
return self.pid != 0
def attach(self, pid: int):
self.pid = pid
log.info("attach wechat pid: %d", self.pid)
return wcprobe.attach(pid)
def detach(self):
log.info("detach wechat pid: %d", self.pid)
return wcprobe.detach(self.pid)
def __send(self, msg_type, data=None, extend=None):
if not self.login_status:
raise WeChatNotLoginError()
message = {
'type': msg_type,
'data': {} if data is None else data,
}
if extend is not None:
message["extend"] = extend
message_json = json.dumps(message)
log.debug("communicate wechat pid:%d, data: %s", self.pid, message)
return wcprobe.send(self.client_id, message_json)
def __send_sync(self, msg_type, data=None, timeout=10):
req_data = ReqData(msg_type, data)
extend = self.__new_extend()
self.__req_data_cache[extend] = req_data
self.__send(msg_type, data, extend)
return req_data.wait_response(timeout)
def __new_extend(self):
while True:
guid = generate_guid("req")
if guid not in self.__req_data_cache:
return guid
def __repr__(self):
return f"WeChatInstance(pid: {self.pid}, client_id: {self.client_id})"
def get_login_info(self):
"""
获取登录信息
"""
return self.__login_info
def get_self_info(self):
"""
获取自己个人信息跟登录信息类似
"""
return self.__send_sync(wx_type.MT_GET_SELF_MSG)
def get_contacts(self):
"""
获取联系人列表
"""
return self.__send_sync(wx_type.MT_GET_CONTACTS_MSG)
def get_contact_detail(self, wxid):
data = {
"wxid": wxid
}
return self.__send_sync(wx_type.MT_GET_CONTACT_DETAIL_MSG, data)
def get_rooms(self):
"""
获取群列表
"""
return self.__send_sync(wx_type.MT_GET_ROOMS_MSG)
def get_room_members(self, room_wxid: str):
"""
获取群成员列表
"""
data = {
"room_wxid": room_wxid
}
return self.__send_sync(wx_type.MT_GET_ROOM_MEMBERS_MSG, data)
def send_text(self, to_wxid: str, content: str):
"""
发送文本消息
"""
data = {
"to_wxid": to_wxid,
"content": content
}
return self.__send(wx_type.MT_SEND_TEXT_MSG, data)
def send_room_at_msg(self, to_wxid: str, content: str, at_list: List[str]):
"""
发送群@消息
"""
data = {
'to_wxid': to_wxid,
'content': content,
'at_list': at_list
}
return self.__send(wx_type.MT_SEND_ROOM_AT_MSG, data)
def send_card(self, to_wxid: str, card_wxid: str):
"""
发送名片
"""
data = {
'to_wxid': to_wxid,
'card_wxid': card_wxid
}
return self.__send(wx_type.MT_SEND_CARD_MSG, data)
def send_link_card(self, to_wxid: str, title: str, desc: str, url: str, image_url: str):
"""
发送链接卡片
"""
data = {
'to_wxid': to_wxid,
'title': title,
'desc': desc,
'url': url,
'image_url': image_url
}
return self.__send(wx_type.MT_SEND_LINK_MSG, data)
def send_image(self, to_wxid: str, file_path: str):
"""
发送图片
"""
data = {
'to_wxid': to_wxid,
'file': file_path
}
return self.__send(wx_type.MT_SEND_IMAGE_MSG, data)
def send_file(self, to_wxid: str, file_path: str):
"""
发送文件
"""
data = {
'to_wxid': to_wxid,
'file': file_path
}
return self.__send(wx_type.MT_SEND_FILE_MSG, data)
#
def send_video(self, to_wxid: str, file_path: str):
"""
发送视频
"""
data = {
'to_wxid': to_wxid,
'file': file_path
}
return self.__send(wx_type.MT_SEND_VIDEO_MSG, data)
# 发送gif
def send_gif(self, to_wxid, file):
data = {
'to_wxid': to_wxid,
'file': file
}
return self.__send(wx_type.MT_SEND_GIF_MSG, data)

View File

@ -0,0 +1,10 @@
class WeChatVersionNotMatchError(Exception):
pass
class WeChatBindError(Exception):
pass
class WeChatNotLoginError(Exception):
pass

24
ntchat/utils/__init__.py Normal file
View File

@ -0,0 +1,24 @@
import uuid
import time
from typing import (
Any,
Dict
)
class ObjectDict(Dict[str, Any]):
"""Makes a dictionary behave like an object, with attribute-style access.
"""
def __getattr__(self, name: str) -> Any:
try:
return self[name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name: str, value: Any) -> None:
self[name] = value
def generate_guid(prefix=''):
return str(uuid.uuid3(uuid.NAMESPACE_URL, prefix + str(time.time())))

61
ntchat/utils/logger.py Normal file
View File

@ -0,0 +1,61 @@
import logging
import os
import configparser
from datetime import datetime
from .xdg import get_log_dir, get_exec_dir
NTCHAT_LOG_KEY = 'NTCHAT_LOG'
NTCHAT_LOG_FILE_KEY = 'NTCHAT_LOG_FILE'
config_file = os.path.join(get_exec_dir(), "config.ini")
CONFIG_DEBUG_LEVEL = ''
if os.path.exists(config_file):
config = configparser.ConfigParser()
config.read(config_file)
CONFIG_DEBUG_LEVEL = config.get('Config', 'LogLevel', fallback=CONFIG_DEBUG_LEVEL)
def get_logger(name: str) -> logging.Logger:
"""
configured Loggers
"""
NTCHAT_LOG = os.environ.get(NTCHAT_LOG_KEY, 'DEBUG')
log_formatter = logging.Formatter(
fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
if CONFIG_DEBUG_LEVEL:
NTCHAT_LOG = CONFIG_DEBUG_LEVEL
# create logger and set level to debug
logger = logging.getLogger(name)
logger.handlers = []
logger.setLevel(NTCHAT_LOG)
logger.propagate = False
# create file handler and set level to debug
if NTCHAT_LOG_FILE_KEY in os.environ:
filepath = os.environ[NTCHAT_LOG_FILE_KEY]
else:
base_dir = get_log_dir()
if not os.path.exists(base_dir):
os.mkdir(base_dir)
time_now = datetime.now()
time_format = '%Y-%m-%d-%H-%M'
filepath = f'{base_dir}/log-{time_now.strftime(time_format)}.txt'
file_handler = logging.FileHandler(filepath, 'a')
file_handler.setLevel(NTCHAT_LOG)
file_handler.setFormatter(log_formatter)
logger.addHandler(file_handler)
# create console handler and set level to info
console_handler = logging.StreamHandler()
console_handler.setLevel(NTCHAT_LOG)
console_handler.setFormatter(log_formatter)
logger.addHandler(console_handler)
return logger

View File

@ -0,0 +1,9 @@
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]

34
ntchat/utils/xdg.py Normal file
View File

@ -0,0 +1,34 @@
import os
import sys
import os.path
def get_exec_dir():
return os.path.dirname(sys.argv[0])
def get_log_dir():
log_dir = os.path.join(os.path.dirname(sys.argv[0]), 'log')
if not os.path.isdir(log_dir):
os.makedirs(log_dir)
return log_dir
def get_root_dir():
return os.path.dirname(os.path.dirname(__file__))
def get_wc_dir():
return os.path.join(get_root_dir(), "wc")
def get_helper_file(version):
return os.path.join(get_wc_dir(), f"helper_{version}.pyd")
def get_support_download_url():
return 'https://webcdn.m.qq.com/spcmgr/download/WeChat3.6.0.18.exe'
if __name__ == '__main__':
print(get_helper_file('3.6.0.18'))

0
ntchat/wc/__init__.py Normal file
View File

0
setup.py Normal file
View File