103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
import logging
|
||
import random
|
||
import time
|
||
from typing import Optional
|
||
|
||
from wcferry import Wcf
|
||
|
||
|
||
class MessageUtil:
|
||
"""
|
||
消息发送工具类，封装了发送文本和文件的方法
|
||
"""
|
||
|
||
# 修改 MessageUtil 类的初始化方法，接受联系人管理器而不是联系人字典
|
||
def __init__(self, wcf, contact_manager):
|
||
self.wcf = wcf
|
||
self.contact_manager = contact_manager
|
||
self.LOG = logging.getLogger("MessageUtil")
|
||
|
||
def send_text_msg(self, msg: str, receiver: str, at_list: str = "") -> None:
|
||
"""
|
||
发送文本消息
|
||
|
||
:param msg: 消息字符串
|
||
:param receiver: 接收人wxid或者群id
|
||
:param at_list: 要@的wxid, @所有人的wxid为：notify@all
|
||
"""
|
||
# 风控处理，随机延迟发送，解决群消息高频发送导致的微信风险
|
||
time.sleep(random.uniform(0.3, 1.0))
|
||
|
||
ats = ""
|
||
if at_list:
|
||
if at_list == "notify@all": # @所有人
|
||
ats = " @所有人"
|
||
else:
|
||
wxids = at_list.split(",")
|
||
for wxid in wxids:
|
||
# 根据 wxid 查找群昵称
|
||
ats += f" @{self.wcf.get_alias_in_chatroom(wxid, receiver)}"
|
||
|
||
# {msg}{ats} 表示要发送的消息内容后面紧跟@，例如 北京天气情况为：xxx @张三
|
||
if ats == "":
|
||
self.LOG.info(f"To {receiver}: {msg}")
|
||
self.wcf.send_text(f"{msg}", receiver, at_list)
|
||
else:
|
||
self.LOG.info(f"To {receiver}: {ats}\r{msg}")
|
||
self.wcf.send_text(f"{ats}\n{msg}", receiver, at_list)
|
||
|
||
def send_file(self, file_path: str, receiver: str) -> None:
|
||
"""
|
||
发送文件消息
|
||
|
||
:param file_path: 文件路径
|
||
:param receiver: 接收人wxid或者群id
|
||
"""
|
||
# 风控处理，随机延迟发送，解决群消息高频发送导致的微信风险
|
||
time.sleep(random.uniform(0.5, 1.5))
|
||
|
||
self.LOG.info(f"Sending file to {receiver}: {file_path}")
|
||
self.wcf.send_file(file_path, receiver)
|
||
|
||
def send_rich_text(self, name: str, account: str, title: str, digest: str, url: str, thumburl: str, receiver: str) -> int:
|
||
"""
|
||
发送富文本消息
|
||
|
||
卡片样式：
|
||
|-------------------------------------|
|
||
|title, 最长两行 |
|
||
|(长标题, 标题短的话这行没有) |
|
||
|digest, 最多三行，会占位 |--------|
|
||
|digest, 最多三行，会占位 |thumburl|
|
||
|digest, 最多三行，会占位 |--------|
|
||
|(account logo) name |
|
||
|-------------------------------------|
|
||
|
||
:param name: 左下显示的名字
|
||
:param account: 填公众号 id 可以显示对应的头像（gh_ 开头的）
|
||
:param title: 标题，最多两行
|
||
:param digest: 摘要，三行
|
||
:param url: 点击后跳转的链接
|
||
:param thumburl: 缩略图的链接
|
||
:param receiver: 接收人, wxid 或者 roomid
|
||
:return: 0 为成功，其他失败
|
||
"""
|
||
# 风控处理，随机延迟发送，解决群消息高频发送导致的微信风险
|
||
time.sleep(random.uniform(0.5, 1.5))
|
||
|
||
self.LOG.info(f"Sending rich text to {receiver}: {title}")
|
||
return self.wcf.send_rich_text(name, account, title, digest, url, thumburl, receiver)
|
||
|
||
def update_contacts(self, contacts: dict) -> None:
|
||
"""
|
||
更新联系人字典
|
||
|
||
:param contacts: 联系人字典，格式为 {"wxid": "NickName"}
|
||
"""
|
||
self.contacts.update(contacts)
|
||
|
||
# 修改使用 allContacts 的地方，改为使用 contact_manager
|
||
# 例如：
|
||
# 原来的代码: nickname = self.allContacts.get(wxid, wxid)
|
||
# 修改为: nickname = self.contact_manager.get_nickname(wxid) |