2024 年最新使用 ntwork 框架搭建企业微信机器人详细教程

ops/2024/10/10 23:26:20/

NTWORK 概述

基于 PC 企业微信的 api 接口，支持收发文本、群@、名片、图片、文件、视频、链接卡片等。

下载安装 ntwork

pip install ntwork

国内源安装

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple ntwork

企业微信版本下载

官方下载：https://dldir1.qq.com/wework/work_weixin/WeCom_4.0.8.6027.exe

简单基础案例

import sys
import ntworkwework = ntwork.WeWork()# 打开pc企业微信, smart: 是否管理已经登录的企业微信`在这里插入代码片`
wework.open(smart=True)# 等待登录
wework.wait_login()# 向文件助手发送一条消息
wework.send_text(conversation_id="FILEASSIST", content="hello, NtWork")try:while True:pass
except KeyboardInterrupt:ntwork.exit_()sys.exit()

获取联系人列表、群列表

# -*- coding: utf-8 -*-
import sys
import ntworkwework = ntwork.WeWork()# 打开pc企业微信, smart: 是否管理已经登录的微信
wework.open(smart=True)# 等待登录
wework.wait_login()# 获取内部(同事)列表并输出
contacts = wework.get_inner_contacts()
print("内部(同事)联系人列表: ")
print(contacts)# 获取外部(客户)列表并输出
contacts = wework.get_external_contacts()
print("外部(客户)联系人列表: ")
print(contacts)# 获取群列表并输出
rooms = wework.get_rooms()
print("群列表: ")
print(rooms)try:while True:pass
except KeyboardInterrupt:ntwork.exit_()sys.exit()

监听消息自动回复

import sys
import ntworkwework = ntwork.WeWork()# 打开pc企业微信, smart: 是否管理已经登录的微信
wework.open(smart=True)# 注册消息回调
@wework.msg_register(ntwork.MT_RECV_TEXT_MSG)
def on_recv_text_msg(wework_instance: ntwork.WeWork, message):data = message["data"]sender_user_id = data["sender"]self_user_id = wework_instance.get_login_info()["user_id"]conversation_id: str = data["conversation_id"]# 判断消息不是自己发的并且不是群消息时，回复对方if sender_user_id != self_user_id and not conversation_id.startswith("R:"):wework_instance.send_text(conversation_id=conversation_id, content=f"你发送的消息是: {data['content']}")try:while True:pass
except KeyboardInterrupt:ntwork.exit_()sys.exit()

使用 fastapi 框架

通过 fastapi 的 swagger 在线文档可以很方便的管理 ntwork 接口：案例

常见问题解决方案

版本不匹配异常

如果出现ntwork.exception.WeWorkVersionNotMatchError异常异常, 请确认是否安装github上指定的企业微信版本，如果确认已经安装，还是报错，可以在代码中添加以下代码，跳过微信版本检测

import ntwork
ntwork.set_wework_exe_path(wework_version='4.0.8.6027')

账号克隆登录

新建一些 ntwork.wework 实例，然后调用 open 方法。

import ntwork# 多开3个微信
for i in range(3):wework = ntwork.WeWork()wework.open(smart=False)

更完善的多实例管理查看 fastapi_example例子

如何监听输出所有的消息

# 注册监听所有消息回调
import ntwork@wework.msg_register(ntwork.MT_ALL)
def on_recv_text_msg(wework_instance: ntwork.WeWork, message):print("########################")print(message)

完全例子查看examples/msg_register_all.py

如何关闭 ntwork 日志

os.environ['NTWORK_LOG'] = "ERROR" 要在import ntwork前执行

# -*- coding: utf-8 -*-
import os
import sys
import time
os.environ['NTWORK_LOG'] = "ERROR"import ntwork

如何关闭 cmd 窗口

先使用pip install pywin32 安装pywin32模块, 然后在代码中添加以下代码， 完整例子查看examples/cmd_close_event.py

import sys
import ntwork
import win32apidef on_exit(sig, func=None):ntwork.exit_()sys.exit()# 当关闭cmd窗口时
win32api.SetConsoleCtrlHandler(on_exit, True)

Pyinstaller 打包 exe

使用 pyinstaller 打包 ntwork 项目，需要添加 --collect-data=ntwork 选项

打包成单个 exe 程序

pyinstaller -F --collect-data=ntwork main.py

将所有的依赖文件打包到一个目录中

pyinstaller -y --collect-data=ntwork main.py

打包 fastapi_example 示例，需要添加 --paths=. --collect-data=ntchat

pyinstaller -F --paths=. --collect-data=ntchat main.py

登录日志预览

2024-04-12 00:45:21,949 - WeWorkManager - INFO - initialize wework, version: 4.0.8.6027
2024-04-12 00:45:21,949 - WeWorkManager - DEBUG - new wework instance
2024-04-12 00:45:21,982 - WeWorkInstance - INFO - open wework pid: 57024
2024-04-12 00:45:23,671 - WeWorkManager - DEBUG - accept client_id: 1
2024-04-12 00:45:26,225 - WeWorkInstance - DEBUG - on recv message: {'data': {'account': '', 'acctid': 'WangPaiGeGe', 'avatar': 'https://wework.qpic.cn/wwpic3az/457956_UJkq5TJkTfCnA5x_1712341231/0', 'corp_id': '1970325052533916', 'document_root': 'C:\\Users\\Administrator\\Documents\\WXWork\\1688856738534700', 'email': '', 'job_name': '', 'mobile': '19561983930', 'nickname': '', 'pid': 57024, 'position': '', 'sex': 1, 'user_id': '1688856738534700', 'username': '王牌哥哥'}, 'type': 11026}
2024-04-12 00:45:26,241 - WeWorkInstance - INFO - login success, name: 王牌哥哥

私聊自动回复案例

import sys
import ntworkwework = ntwork.WeWork()# 打开 pc 企业微信, smart: 是否管理已经登录的微信
wework.open(smart=True)# 注册消息回调
@wework.msg_register(ntwork.MT_RECV_TEXT_MSG)
def on_recv_text_msg(wework_instance: ntwork.WeWork, message):data = message["data"]sender_user_id = data["sender"]self_user_id = wework_instance.get_login_info()["user_id"]conversation_id: str = data["conversation_id"]# 判断消息不是自己发的并且不是群消息时，回复对方if sender_user_id != self_user_id and not conversation_id.startswith("R:"):wework_instance.send_text(conversation_id=conversation_id, content=f"你发送的消息是: {data['content']}")try:while True:pass
except KeyboardInterrupt:ntwork.exit_()sys.exit()

消息属性

私聊消息属性说明

{'appinfo': 'CAQQiZrgsAYYpMv0mZmAgAMgyIKKUw==','at_list': [],'content': '我喜欢你','content_type': 0,'conversation_id': 'S:1688856625489316_1688856738534700','is_pc': 0,'local_id': '34','receiver': '1688856738534700','send_time': '1712852233','sender': '1688856625489316','sender_name': '唤醒手腕','server_id': '1000235'
}

群聊消息属性说明

{'appinfo': 'CAQQiKTgsAYYpMv0mZmAgAMg99ak/Ak=','at_list': [],'content': '我喜欢你','content_type': 0,'conversation_id': 'R:1970325052533916','is_pc': 0,'local_id': '41','receiver': '1688856738534700','send_time': '1712853512','sender': '1688856625489316','sender_name': '唤醒手腕','server_id': '1000258'
}

案例：故障离线表格统计

读取故障数据源码

import re# 假设这是你的消息体字符串
test_message = """  
【故障位置】化州西收费站97车道
【故障情况】不能缴费
【填报人】唤醒手腕
"""def analysis_repair_data(message):# 定义正则表达式模式，用于匹配每个字段pattern = r"【故障位置】(.*)\n【故障情况】(.*)\n【填报人】(.*)"# 使用re.search来查找匹配项match = re.search(pattern, message, re.DOTALL)if match:fault_location = match.group(1).strip()  # 提取故障位置fault_situation = match.group(2).strip()  # 提取故障情况reporter = match.group(3).strip()  # 提取填报人return {"fault_location": fault_location,"fault_situation": fault_situation,"reporter": reporter}else:return False

读取维修数据源码

import re# 假设这是你的消息体字符串
test_message = """  
【故障编号】 
【维修情况】已修复  
【维修人】xxx  
"""def analysis_solve_data(message):# 定义正则表达式模式，用于匹配每个字段pattern = r"""  【故障编号】(.*)  【维修情况】(.*)  【维修人】(.*)  """pattern = re.compile(pattern, re.DOTALL | re.VERBOSE)# 使用re.search来查找匹配项match = pattern.search(message)if match:serial = match.group(1).strip()  # 提取故障编号repair_situation = match.group(2).strip()  # 提取维修情况repair_person = match.group(3).strip()  # 提取维修人return {"serial": serial,"repair_situation": repair_situation,"repair_person": repair_person}else:return False

更新离线 Excel 表格源码

from datetime import datetimefrom openpyxl.reader.excel import load_workbook# 加载工作簿
wb = load_workbook(filename='故障维修登记表.xlsx')
# 选择活动工作表或者通过名字选择工作表
ws = wb.active  # 使用活动工作表def insert_new_note(repair_data):# 获取行数row_count = ws.max_rownow = datetime.now()# 使用f-string格式化时间formatted_time = f"{now:%Y%m%d-%H%M%S}"data = {"index": row_count,"serial": formatted_time,"fault_location": repair_data['fault_location'],"fault_situation": repair_data['fault_situation'],"reporter": repair_data['reporter'],"repair_situation": "","repair_person": ""}row_num = row_count + 1ws.cell(row=row_num, column=1).value = data['index']ws.cell(row=row_num, column=2).value = data['serial']ws.cell(row=row_num, column=3).value = data['fault_location']ws.cell(row=row_num, column=4).value = data['fault_situation']ws.cell(row=row_num, column=5).value = data['reporter']ws.cell(row=row_num, column=6).value = data['repair_situation']ws.cell(row=row_num, column=7).value = data['repair_person']# 保存工作簿到文件wb.save("故障维修登记表.xlsx")return data['serial']def finish_old_note(solve_data):target_name = solve_data['serial']# 遍历所有行，查找目标值state = Falsefor index, row in enumerate(ws.iter_rows(values_only=True)):if row[1] == "编号":continueif row[1] == target_name:  # 假设"Name"在第一列# 找到目标行后，修改该行的数据# 这里假设我们要修改的是第二列的数据，将其改为"New Value"ws.cell(row=(index + 1), column=6).value = solve_data['repair_situation']ws.cell(row=(index + 1), column=7).value = solve_data['repair_person']state = Truewb.save("故障维修登记表.xlsx")break  # 找到后就退出循环if state:return solve_data['serial']else:return False

主程序 app.py 源码

# -*- coding: utf-8 -*-
import sys
from datetime import datetimeimport ntworkfrom read_repair_data import analysis_repair_data
from read_solve_data import analysis_solve_data
from update_excel import insert_new_note, finish_old_notewework = ntwork.WeWork()# 打开 pc 企业微信, smart: 是否管理已经登录的微信
wework.open(smart=True)# 注册消息回调
@wework.msg_register(ntwork.MT_RECV_TEXT_MSG)
def on_recv_text_msg(wework_instance: ntwork.WeWork, message):data = message["data"]if data['content'][0:1] != "【":returnsender_user_id = data["sender"]self_user_id = wework_instance.get_login_info()["user_id"]conversation_id: str = data["conversation_id"]# 判断消息不是自己发的并且不是群消息时，回复对方if sender_user_id != self_user_id and conversation_id.startswith("R:"):insert_data = analysis_repair_data(data['content'])if insert_data:serial = insert_new_note(insert_data)now = datetime.now()# 使用f-string格式化时间formatted_time = f"{now:%Y-%m-%d %H:%M:%S}"print(f"【反馈结果】【{formatted_time}】请注意存在新的反馈：{serial}")wework_instance.send_text(conversation_id=conversation_id, content=f"【反馈结果】反馈成功\n【故障编号】{serial}")insert_data = analysis_solve_data(data['content'])if insert_data:serial = finish_old_note(insert_data)if serial:# 使用f-string格式化时间now = datetime.now()formatted_time = f"{now:%Y-%m-%d %H:%M:%S}"print(f"【维修结果】【{formatted_time}】请注意存在新的维修：{serial}")wework_instance.send_text(conversation_id=conversation_id, content=f"【维修结果】维修成功\n【故障编号】{str(serial)}")else:wework_instance.send_text(conversation_id=conversation_id, content=f"【操作警告】故障编号不存在！")try:while True:pass
except KeyboardInterrupt:ntwork.exit_()sys.exit()

填报故障真机演示

在这里插入图片描述

维修故障真机演示

在这里插入图片描述

Excel 表格情况预览

在这里插入图片描述


http://www.ppmy.cn/ops/39150.html

相关文章

JCR一区 | Matlab实现1D-2D-GASF-CNN-GRU-MATT的多通道输入数据分类预测

JCR一区 | Matlab实现1D-2D-GASF-CNN-GRU-MATT的多通道输入数据分类预测

JCR一区 | Matlab实现1D-2D-GASF-CNN-GRU-MATT的多通道输入数据分类预测 目录 JCR一区 | Matlab实现1D-2D-GASF-CNN-GRU-MATT的多通道输入数据分类预测分类效果基本介绍程序设计参考资料 分类效果 基本介绍 基本介绍 Matlab实现1D-2D-GASF-CNN-GRU-MATT的多通道输入数据分类预…
阅读更多...
c++ 线程实验

c++ 线程实验

在C中&#xff0c;你可以使用C11或更高版本的线程库来创建和管理线程。以下是一个简单的C线程实验示例&#xff0c;该示例展示了如何创建和启动一个线程&#xff0c;以及如何在主线程和子线程中执行不同的任务。 #include <iostream> #include <thread> #include …
阅读更多...
hyper-v启动centos7虚拟机不能联网

hyper-v启动centos7虚拟机不能联网

虚拟网卡要和之前虚拟机里面设置的GATEWAY一致。 安装CentOS遇到Error setting up base repository换url 或者换镜像包iso(这个有用&#xff09; 没掌握摸Yu的精髓 好累啊
阅读更多...
Python写了for i in range(10)却只打印一遍?

Python写了for i in range(10)却只打印一遍?

题目&#xff1a;定义一个两个参数的重复打印函数&#xff0c;第一个参数指定要打印的字符串&#xff0c;第二个参数指定要重复打印的次数&#xff0c;在主程序中调用该函数&#xff0c;打印10遍你的学号姓名。 为什么调用函数后结果只打印了一遍? 看了题目感觉就很诡异&#…
阅读更多...
您的文件和驱动器上的“密码保护”有多安全？

您的文件和驱动器上的“密码保护”有多安全？

某些行业&#xff08;例如医疗保健、法律和公司&#xff09;的人们在通过电子邮件发送文件时通常依赖密码保护&#xff0c;认为它可以提供足够的安全性来防止窥探。然而&#xff0c;对 PDF 或 Excel 文件进行简单的密码保护并不像看起来那样万无一失。 使用密码保护文件而不加…
阅读更多...
Git详解之五：分布式Git

Git详解之五：分布式Git

为了便于项目中的所有开发者分享代码&#xff0c;我们准备好了一台服务器存放远程 Git 仓库。经过前面几章的学习&#xff0c;我们已经学会了一些基本的本地工作流程中所需用到的命令。接下来&#xff0c;我们要学习下如何利用 Git 来组织和完成分布式工作流程。 特别是&#…
阅读更多...
FreeRTOS标准库例程代码

FreeRTOS标准库例程代码

1.设备STM32F103C8T6 2.工程模板 单片机: 部分单片机的程序例程 - Gitee.comhttps://gitee.com/lovefoolnotme/singlechip/tree/master/STM32_FREERTOS/1.%E5%B7%A5%E7%A8%8B%E6%A8%A1%E6%9D%BF 3.代码 1-FreeRTOS移植模板 #include "system.h" #include "…
阅读更多...
典型相关分析

典型相关分析

典型相关分析是一种用于探索两组变量之间关系的多变量统计方法。以下是典型相关分析的一般操作步骤&#xff1a; 1. **数据准备**&#xff1a; - 收集两组变量的数据&#xff0c;通常每组变量都包含多个变量。 - 确保数据满足典型相关分析的假设&#xff0c;例如变量之间…
阅读更多...
最新文章

Copyright @ 2022~2023