2026/2/7 5:39:46
网站建设
项目流程
给个网站带颜色,wordpress+park主题,为什么网页在线视频会卡,wordpress for sae图床QiWe开放平台 个人名片 API驱动企微自动化#xff0c;让开发更高效 核心能力#xff1a;为开发者提供标准化接口、快速集成工具#xff0c;助力产品高效拓展功能场景 官方站点#xff1a;https://www.qiweapi.com 团队定位#xff1a;专注企微API生态的技术服务团队 对接…QiWe开放平台 · 个人名片API驱动企微自动化让开发更高效核心能力为开发者提供标准化接口、快速集成工具助力产品高效拓展功能场景官方站点https://www.qiweapi.com团队定位专注企微API生态的技术服务团队对接通道搜「QiWe 开放平台」联系客服核心理念合规赋能让企微开发更简单、更高效本文聚焦 Python 语言极简拆解 RPA 模式下企微第三方 API 外部群主动调用的完整实现流程从环境搭建、核心代码编写到测试落地全程实操导向无冗余理论适配 Python 技术栈团队快速落地覆盖文本推送、群聊查询核心场景。一、前置准备3 步完成极简配置1. 环境准备服务器Windows Server 2019/2022需开启桌面交互、安装基础显卡驱动软件企微官方稳定版关闭自动更新 / 弹窗、Python 3.9核心依赖终端直接安装pip install requests pyautogui pywinauto redis pymysql2. 基础配置企微登录专用运营账号固定窗口位置建议最大化锁定群聊列表布局中间件部署 Redis缓存 token、群聊定位参数、MySQL存储操作日志开启本地访问配置简单密码权限服务器创建专用运行账号赋予程序 / 日志 / 数据盘读写权限关闭无关权限。3. 核心前置规避踩坑提前获取企微群聊名称用于定位、企业 ID 应用密钥用于鉴权禁用企微自动升级、消息弹窗、文件预览避免干扰 RPA 操作测试 Python 环境能否正常调用企微客户端通过 pywinauto 获取窗口句柄验证。二、核心代码实现分模块编写可直接复用整体分为 4 个核心模块代码极简注释清晰可直接复制修改配置后使用聚焦外部群主动调用核心能力。1. 基础工具模块鉴权 缓存 日志全局复用import redis import pymysql import requests import time # 1. Redis缓存初始化用于缓存token、群聊定位参数 redis_client redis.Redis(hostlocalhost, port6379, db0, passwordyour_redis_pwd) # 2. MySQL初始化用于存储操作日志 mysql_conn pymysql.connect( hostlocalhost, userroot, passwordyour_mysql_pwd, dbqw_rpa_api, charsetutf8 ) cursor mysql_conn.cursor() # 3. 鉴权获取access_token缓存7200s自动刷新 def get_access_token(): cache_key qw_rpa:access_token token redis_client.get(cache_key) if token: return token.decode() # 调用鉴权接口复用之前封装的鉴权逻辑 res requests.post( http://localhost:8080/auth, json{corp_id: your_corp_id, app_secret: your_app_secret} ) token res.json()[data][access_token] redis_client.setex(cache_key, 7200, token) return token # 4. 日志记录简化版仅记录核心信息 def record_log(task_id, chat_id, operate_type, result): sql INSERT INTO operate_log (task_id, chat_id, operate_type, result, operate_time) VALUES (%s, %s, %s, %s, %s) cursor.execute(sql, (task_id, chat_id, operate_type, result, time.strftime(%Y-%m-%d %H:%M:%S))) mysql_conn.commit()2. RPA 定位与操作模块核心实现群聊定位 消息推送from pywinauto import Application import pyautogui # 1. 连接企微客户端获取窗口句柄 def connect_qw(): try: # 通过企微进程名连接无需坐标更稳定 app Application(backenduia).connect(process12345) # 替换为你的企微进程ID qw_window app.window(title_re企业微信) # 匹配企微窗口标题 qw_window.set_focus() # 激活企微窗口 return qw_window except Exception as e: print(f企微连接失败{str(e)}) return None # 2. 群聊定位文本匹配控件辅助适配群聊列表 def locate_group_chat(qw_window, chat_name): cache_key fqw_rpa:group:{chat_name} # 先查缓存避免重复定位 if redis_client.exists(cache_key): return True try: # 定位群聊列表点击搜索框 search_box qw_window.child_window(title搜索, control_typeEdit).click_input() pyautogui.typewrite(chat_name, interval0.05) # 输入群聊名称模拟人工输入 time.sleep(1) # 等待搜索结果 # 定位目标群聊并点击进入 group_chat qw_window.child_window(titlechat_name, control_typeListItem).click_input() time.sleep(0.5) # 缓存定位结果有效期7天 redis_client.setex(cache_key, 604800, success) return True except Exception as e: print(f群聊定位失败{str(e)}) return False # 3. 文本消息推送核心操作标准化执行 def push_text_message(chat_name, content): task_id ftask_{int(time.time())} # 生成唯一任务ID幂等性控制 qw_window connect_qw() if not qw_window: record_log(task_id, chat_name, push_text, 企微未连接) return {code: 500, msg: 企微未连接, task_id: task_id} # 定位群聊 if not locate_group_chat(qw_window, chat_name): record_log(task_id, chat_name, push_text, 群聊定位失败) return {code: 500, msg: 群聊定位失败, task_id: task_id} try: # 定位输入框输入消息 input_box qw_window.child_window(title输入框, control_typeEdit).click_input() pyautogui.typewrite(content, interval0.05) time.sleep(0.2) # 点击发送按钮 qw_window.child_window(title发送, control_typeButton).click_input() time.sleep(0.5) # 记录成功日志 record_log(task_id, chat_name, push_text, success) return {code: 200, msg: 推送成功, task_id: task_id} except Exception as e: record_log(task_id, chat_name, push_text, f失败{str(e)}) return {code: 500, msg: f推送失败{str(e)}, task_id: task_id}3. 群聊查询模块获取企业外部群列表简化版def get_external_group_list(): 获取企业名下有效外部群列表RPA模拟查询简化版 qw_window connect_qw() if not qw_window: return {code: 500, msg: 企微未连接, data: []} try: # 定位外部群入口并点击 qw_window.child_window(title外部群, control_typeListItem).click_input() time.sleep(1) # 遍历群聊列表获取有效群聊名称简化版仅获取可见群聊 group_list [] group_items qw_window.child_window(control_typeList).children() for item in group_items: group_name item.window_text() if group_name and 已解散 not in group_name: # 剔除已解散群聊 group_list.append(group_name) return {code: 200, msg: 查询成功, data: group_list[:20]} # 限制返回数量避免过多 except Exception as e: return {code: 500, msg: f查询失败{str(e)}, data: []}4. API 封装模块FastAPI提供标准化调用入口from fastapi import FastAPI app FastAPI(titlePythonRPA 企微第三方API, version1.0) # 1. 消息推送接口外部调用核心接口 app.post(/group/push/text) def group_push_text(chat_name: str, content: str): return push_text_message(chat_name, content) # 2. 群聊查询接口 app.get(/group/list) def group_list(): return get_external_group_list() # 3. 日志查询接口简化版 app.get(/log/query) def log_query(task_id: str): sql SELECT * FROM operate_log WHERE task_id %s cursor.execute(sql, (task_id,)) log cursor.fetchone() return {code: 200, data: log if log else None} # 启动服务终端执行该脚本即可 if __name__ __main__: import uvicorn uvicorn.run(app__main__:app, host0.0.0.0, port8080, reloadFalse)三、测试落地3 步验证快速上线1. 本地测试启动 Redis、MySQL创建对应的数据库和日志表运行 API 脚本启动 FastAPI 服务用 Apifox 调用/group/push/text接口传入群聊名称和文本内容验证推送是否成功。2. 核心优化规避踩坑提升稳定性操作速率所有输入 / 点击添加延时输入 0.05s / 字符、点击≥200ms模拟人工异常兜底添加企微掉线、定位失败重试逻辑最多 2 次失败则记录日志并提醒缓存优化群聊定位参数缓存 7 天token 缓存 7200s减少重复操作提升效率。3. 生产部署用 PyInstaller 将脚本打包为可执行文件上传至服务器设置开机自启配置防火墙开放 8080 端口启动服务后调用接口做最终验证确认功能正常、日志完整。四、核心总结PythonRPA 开发企微第三方 API核心优势是轻量、易上手、成本低无需复杂协议开发依托 Python 丰富的 RPA 生态pyautogui、pywinauto可快速实现外部群主动调用。本文实现的核心能力的是文本推送、群聊查询可基于此扩展图片 / 文件推送、群成员管理等功能代码模块化设计复用性强。落地关键是控制操作速率、做好缓存与异常兜底规避企微风控确保长期稳定运行完全适配中小团队 Python 技术栈的落地需求。