wechat/wxhook
5
0
Fork 0
mirror of https://github.com/miloira/wxhook.git synced 2024-11-22 02:19:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

version 0.0.4

Browse Source
This commit is contained in:
张明明 2024-05-06 16:28:29 +08:00
parent 70b1663939
commit 95b6af8b79
6 changed files with 51 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
README.md
View File

@ -60,15 +60,13 @@ pip install wxhook
```python
# import os
# os.environ["WXHOOK_LOG_LEVEL"] = "INFO" # 修改日志输出级别
import time
from wxhook import Bot
from wxhook import events
from wxhook.model import Event
def on_login(bot: Bot):
bot.send_text("filehelper", "登录成功之后会触发这个函数")
print("登录成功之后会触发这个函数")
def on_start(bot: Bot):
@ -76,40 +74,33 @@ def on_start(bot: Bot):
def on_stop(bot: Bot):
bot.send_text("filehelper", "关闭微信客户端之前会触发这个函数")
time.sleep(1) # 防止客户端关闭太快导致消息发送失败
print("关闭微信客户端之前会触发这个函数")
def on_before_message(bot: Bot, event: Event):
print("消息事件处理之前")
def on_after_message(bot: Bot, event: Event):
print("消息事件处理之后")
bot = Bot(
# faked_version="3.9.10.19", # 解除微信低版本限制
on_login=on_login,
on_start=on_start,
on_stop=on_stop
on_stop=on_stop,
on_before_message=on_before_message,
on_after_message=on_after_message
)
# 消息回调地址
# bot.set_webhook_url("http://127.0.0.1:8000")
@bot.handle(events.TEXT_MESSAGE, once=True)
def on_message(bot: Bot, event: Event):
bot.send_text("filehelper", "这条消息只会发送一次哦")
@bot.handle(events.TEXT_MESSAGE)
def on_message(bot: Bot, event: Event):
if event.fromUser != bot.info.wxid:
bot.send_text(event.fromUser, event.content)
@bot.handle([events.IMAGE_MESSAGE, events.EMOJI_MESSAGE, events.VIDEO_MESSAGE])
def on_message(bot: Bot, event: Event):
if event.fromUser != bot.info.wxid:
if event.type == events.IMAGE_MESSAGE:
bot.send_text(event.fromUser, "图片消息")
elif event.type == events.EMOJI_MESSAGE:
bot.send_text(event.fromUser, "表情消息")
elif event.type == events.VIDEO_MESSAGE:
bot.send_text(event.fromUser, "视频消息")
bot.send_text("filehelper", "hello world!")
bot.run()

2
setup.py
View File

@ -18,7 +18,7 @@ URL = 'https://github.com/miloira/wxhook'
EMAIL = '690126048@qq.com'
AUTHOR = 'Msky'
REQUIRES_PYTHON = '>=3.8.0'
VERSION = '0.0.3'
VERSION = '0.0.4'
# What packages are required for this module to be executed?
REQUIRED = [

2
wxhook/__init__.py
View File

@ -1,3 +1,3 @@
from .core import Bot
version = "0.0.3"
version = "0.0.4"

26
wxhook/core.py
View File

@ -9,10 +9,10 @@ import psutil
import pyee
import requests
from .events import ALL_MESSAGE, SYSTEM_MESSAGE
from .events import ALL_MESSAGE
from .logger import logger
from .model import RawData, Event, Account, Contact, ContactDetail, Room, RoomMembers, Table, DB, Response
from .utils import WeChatManager, start_wechat_with_inject, fake_wechat_version, parse_event
from .model import Event, Account, Contact, ContactDetail, Room, RoomMembers, Table, DB, Response
from .utils import WeChatManager, start_wechat_with_inject, fake_wechat_version, get_pid, parse_event
class RequestHandler(socketserver.BaseRequestHandler):
@ -68,22 +68,25 @@ class Bot:
self.IMAGE_SAVE_PATH = None
self.VIDEO_SAVE_PATH = None
try:
code, output = start_wechat_with_inject(self.remote_port)
if code == 1:
raise Exception(output)
except Exception:
output = get_pid(self.remote_port)
self.process = psutil.Process(int(output))
if self.faked_version is not None:
if fake_wechat_version(self.process.pid, self.version, faked_version) == 0:
logger.info(f"wechat version faked: {self.version} -> {faked_version}")
logger.success(f"wechat version faked: {self.version} -> {faked_version}")
else:
logger.info(f"wechat version fake failed.")
logger.error(f"wechat version fake failed.")
logger.info(f"API Server at 0.0.0.0:{self.remote_port}")
self.wechat_manager.add(self.process.pid, self.remote_port, self.server_port)
self.call_hook_func(self.on_start, self)
self.handle(SYSTEM_MESSAGE, once=True)(self.init_bot)
self.handle(ALL_MESSAGE, once=True)(self.init_bot)
self.hook_sync_msg(self.server_host, self.server_port)
@staticmethod
@ -92,13 +95,12 @@ class Bot:
return func(*args, **kwargs)
def init_bot(self, bot: "Bot", event: Event) -> None:
if event.content["sysmsg"]["@type"] == "SafeModuleCfg":
self.DATA_SAVE_PATH = bot.info.dataSavePath
self.WXHELPER_PATH = os.path.join(self.DATA_SAVE_PATH, "wxhelper")
self.FILE_SAVE_PATH = os.path.join(self.WXHELPER_PATH, "file")
self.IMAGE_SAVE_PATH = os.path.join(self.WXHELPER_PATH, "image")
self.VIDEO_SAVE_PATH = os.path.join(self.WXHELPER_PATH, "video")
self.call_hook_func(self.on_login, bot)
self.call_hook_func(self.on_login, bot, event)
def set_webhook_url(self, webhook_url: str) -> None:
self.webhook_url = webhook_url
@ -409,7 +411,7 @@ class Bot:
def get_db_info(self) -> list[DB]:
"""获取数据库句柄"""
return [DB(databaseName=item["databaseName"], handle=item["handle"],
tables=[Table(**subitem) for subitem in item["tables"]]) for item in self.call_api("/api/getDBInfo")]
tables=[Table(**sub_item) for sub_item in item["tables"]]) for item in self.call_api("/api/getDBInfo")]
def exec_sql(self, db_handle: int, sql: str) -> Response:
"""执行SQL命令"""
@ -430,7 +432,7 @@ class Bot:
def on_event(self, raw_data: bytes):
try:
data = json.loads(raw_data)
event = Event(**parse_event(data), rawData=RawData(raw_data))
event = Event(**parse_event(data))
logger.debug(event)
self.call_hook_func(self.on_before_message, self, event)
self.event_emitter.emit(str(ALL_MESSAGE), self, event)
@ -460,7 +462,7 @@ class Bot:
try:
server = socketserver.ThreadingTCPServer((self.server_host, self.server_port), RequestHandler)
server.bot = self
logger.info(f"{self.server_host}:{self.server_port}")
logger.info(f"Listening Server at {self.server_host}:{self.server_port}")
server.serve_forever()
except (KeyboardInterrupt, SystemExit):
self.exit()

12
wxhook/model.py
View File

@ -64,17 +64,6 @@ class RoomMembers:
members: str # 聊天室成员的微信ID列表，各ID之间使用特定字符分隔
@dataclass
class RawData:
"""原始数据"""
data: bytes
def __repr__(self):
return "<RawData>"
__str__ = __repr__
@dataclass
class Event:
"""消息事件"""
@ -90,7 +79,6 @@ class Event:
signature: typing.Optional[str] = None # 消息签名，包含一系列的配置信息
toUser: typing.Optional[str] = None # 消息接收者的用户ID
type: typing.Optional[int] = None # 消息类型
rawData: typing.Optional[RawData] = None # 原始数据
@dataclass

19
wxhook/utils.py
View File

@ -24,7 +24,7 @@ def fake_wechat_version(pid: int, old_version: str, new_version: str):
return int(result.stdout)
def get_processes(process_name):
def get_processes(process_name: str):
processes = []
for process in psutil.process_iter():
if process.name().lower() == process_name.lower():
@ -32,11 +32,16 @@ def get_processes(process_name):
return processes
def parse_xml(xml):
def get_pid(port: int):
output = subprocess.run(f"netstat -ano | findStr \"{port}\"", capture_output=True, text=True, shell=True).stdout
return int(output.split("\n")[0].split("LISTENING")[-1])
def parse_xml(xml: str):
return xmltodict.parse(xml)
def parse_event(event, fields=None):
def parse_event(event: dict, fields=None):
for field in fields or ["content", "signature"]:
try:
if field in event:
@ -70,11 +75,11 @@ class WeChatManager:
data = json.load(file)
return data
def write(self, data):
def write(self, data: dict):
with open(self.filename, "w", encoding="utf-8") as file:
json.dump(data, file)
def refresh(self, pid_list):
def refresh(self, pid_list: list[int]):
data = self.read()
cleaned_data = []
remote_port_list = [19000]
@ -95,14 +100,14 @@ class WeChatManager:
data = self.read()
return data["increase_remote_port"] + 1
def get_listen_port(self, remote_port):
def get_listen_port(self, remote_port: int):
return 19000 - (remote_port - 19000)
def get_port(self):
remote_port = self.get_remote_port()
return remote_port, self.get_listen_port(remote_port)
def add(self, pid, remote_port, server_port):
def add(self, pid: int, remote_port: int, server_port: int):
data = self.read()
data["increase_remote_port"] = remote_port
data["wechat"].append({