
“钉钉和企业微信里的消息到底能不能用Python爬” 这是我在后台被问得最多的一类问题。很多人一听到“爬”第一反应就是抓包、模拟登录、抠接口但真要把办公数据自动化采集这件事落地最稳、最长久的路其实是官方开放平台。我自己带过好几个自动化采集项目从钉钉审批流到企业微信告警通知都跑过可以负责任地说通过官方接口拿数据不仅省心而且能覆盖90%以上的办公自动化需求。这篇文章我不讲那些灰色的东西只讲怎么用Python把钉钉和企业微信里和数据相关的消息、审批、考勤、通讯录规规矩矩地收进自己的系统里。适合正在做办公自动化、企业内部系统集成的开发、运维、数据同学参考。1. 为什么说“爬”钉钉和企业微信最稳的路是官方接口1.1 抓包模拟登录那套路子为什么越来越不好使放开场白里说的很多新手可能仍然对“模拟登录、抓包”抱有兴趣。我简单拆一下为什么这条路越来越窄第一钉钉和企业微信的接口做了大量签名和加密请求头里带了一堆动态参数今天能跑的脚本明天可能就全线失效。头部版本更新所有参数就变了。我见过有人花了一周逆向一个打卡数据的接口结果客户端一升级全白干。第二客户端通信走的是长连接和私有协议不是简单的HTTP。抓包抓到的东西大多是压缩和二进制的解起来要花很大力气。就算你解出来了后续还有风控策略等着你。第三也是最重要的账号安全风控非常敏锐。一旦检测到异常请求或非官方客户端行为轻则验证码弹窗重则账号受限甚至封禁。为了采集几条审批数据搭上办公账号这个代价不值得。1.2 官方接口能拿到的数据清单钉钉 VS 企业微信我把两边开放平台的能力做了个对比这里不完全按官方文档罗列而是按“日常采集项目里真正用得上”的维度来说能力领域钉钉开放平台企业微信API/开放平台身份与组织信息通讯录、部门、角色、员工信息通讯录、标签、部门、外部联系人消息类群机器人发消息、工作通知、Stream模式收消息群机器人发消息、应用消息推送、回调接收消息审批类审批实例、模板、状态流转、附件审批能力相对弱常需第三方应用补充考勤类打卡结果、排班、班次、请假套件打卡数据部分开放需限额申请日程会议类日程、会议、会议室会议、日程、直播回放这张表根据我的实际经验简化过。两边能力都有精细化权限具体以官方文档为准。但可以看出钉钉在“内部流程数据”采集上更厚企业微信在“消息和通讯录”上更顺手。做方案选型的时候先搞清楚你的数据主要在哪边再决定主攻方向。1.3 合规的底线自有数据、授权范围、最小权限这里必须说清楚。办公数据自动化采集不是灰色地带前提是你的数据来源范围合法只能采集本企业内部的、你有管理权限的数据。比如你自己所在组织下的审批实例、你管理范围内的部门成员。不能采集非授权用户的私聊内容。哪怕你觉得“只是统计关键词”未经授权采集聊天记录也涉及严重隐私问题。推荐采用最小权限原则。创建应用时只勾选你要用到的权限点不要图省事一次全开。一句话总结官方给什么拿什么授权给多少要多少。在这个前提下采集方案才是可持续的。我见过有人贪方便把权限全部拉满最后出问题的时候连他自己都说不清楚数据被谁看了。2. 钉钉侧落地企业内部应用拉取审批、考勤和消息2.1 创建企业内部应用拿到AppKey和AppSecret打开钉钉开放平台进入开发者后台选择“企业内部应用”创建一个应用。创建后你会在“凭证与基础信息”里看到AppKey和AppSecret这两个凭证相当于应用访问钉钉的身份证明。创建的时候要注意两件事。一是应用类型选择“企业内部应用”二是权限申请按需来。比如要做审批采集就在权限管理里搜索“审批”相关的接口权限并申请要做考勤就申请“考勤”权限。权限申请一旦通过AppKey和AppSecret就有了对应的调用能力。一个小提示很多权限申请需要企业管理员在管理后台审核。如果你是项目的实施人最好提前跟管理员打好招呼说明每个权限的用途审核会快很多也避免后续扯皮。2.2 用Python维护access_token获取与本地缓存钉钉老接口取token是GET /gettoken新接口统一走POST。新接口返回的是accessToken字段import requests def get_dd_token(app_key: str, app_secret: str) - str: url https://api.dingtalk.com/v1.0/oauth2/accessToken resp requests.post(url, json{appKey: app_key, appSecret: app_secret}) resp.raise_for_status() return resp.json()[accessToken]这里提个醒token有效期一般是7200秒。但你没必要每次请求都重新拿。更稳的做法是第一次拿完后存到本地文件或Redis带过期时间。过期时间要减去一个安全余量比如设置7100秒。我第一次做的时候偷懒每次调用都gettoken结果钉钉直接限流。那个项目一开始没跑几天就被打回重做。缓存token看似多写几行代码却能省掉大量限流麻烦。2.3 采集审批实例和打卡记录两个高频接口实战审批实例是办公数据里最常被采集的。比如你想统计某个流程的审批耗时就得把实例列表拉下来再逐条看状态def fetch_process_instances(token, process_code, start_ms, end_ms): url fhttps://oapi.dingtalk.com/topapi/processinstance/list?access_token{token} payload { process_code: process_code, start_time: start_ms, end_time: end_ms, size: 20, cursor: 0 } resp requests.post(url, jsonpayload) data resp.json() if data.get(errcode) ! 0: raise RuntimeError(data.get(errmsg)) return data.get(result, {})注意start_time和end_time是毫秒级时间戳。遍历分页时用返回的next_cursor继续拉直到没有更多数据。我习惯把“拉取一页-处理一页-写库一页”串成一个循环尽量不要一次性把全量数据load到内存里。打卡记录接口稍微有点不一样时间字段用的是字符串def fetch_attendance(token, user_ids, date_from, date_to): url fhttps://oapi.dingtalk.com/topapi/attendance/list?access_token{token} payload { workDateFrom: date_from, # 2024-01-01 00:00:00 workDateTo: date_to, # 2024-01-31 23:59:59 userIdList: user_ids, offset: 0, limit: 10 } resp requests.post(url, jsonpayload) return resp.json()这个接口返回的记录比较多注意分页offset和limit配合。建议按天或按周拉取不要一次性拉三个月既容易超时也容易触发限流。我一般会做成“按天循环”每天一个任务这样即使某天失败了重跑的成本也很低。2.4 主动收消息机器人Webhook和Stream模式钉钉机器人有两个常见方向。一个是往外发——你把数据结果通过Webhook推到群里实现“数据看板自动播报”。另一个是往里收——让Python程序接收群消息或单聊消息做关键词响应。发消息的Webhook很简单def send_dd_markdown(webhook_url, title, text): resp requests.post(webhook_url, json{ msgtype: markdown, markdown: {title: title, text: text} })往回收推荐Stream模式。钉钉官方提供了dingtalk-stream这个Python SDK它基于WebSocket长连接不需要暴露公网回调地址。用Stream可以实时订阅消息、机器人事件等对“消息采集”这个需求来说比HTTP回调更省事。我实际用Stream做过一个机器人员工在群里发特定指令它就把对应的数据查询结果推回来。部署很简单一台普通服务器保持长连接就行不用配公网IP也不用做域名校验。3. 企业微信侧落地群机器人推送与API拉取怎么配合3.1 群机器人Webhook一进去就能用适合做数据上报出口企业微信的群机器人是我用过的最顺手的告警通道。往任意群里添加一个自定义机器人复制Webhook地址就能开始推消息。它在很多项目里承担“数据出口”的角色你的采集任务跑完了把结构化结果格式化成一篇文章推到群里大家就看到了。def send_wx_markdown(webhook_url, content): resp requests.post(webhook_url, json{ msgtype: markdown, markdown: {content: content} })推送消息的时候注意企业微信的文本长度限制。markdown格式内容不要塞太多太长会被截断。我一般会把重点数据放前面细节放附件或链接里。比如推送日报时第一行放“今日新增审批37条”后面接一个简短的表格比堆一大段文字效果好得多。3.2 企微API拉取通讯录、应用消息与外部联系人如果只做推送机器人webhook够了。但如果你要“采集”数据就得用企微服务商API的“读取”能力。先取tokendef get_wx_token(corp_id, secret): url https://qyapi.weixin.qq.com/cgi-bin/gettoken resp requests.get(url, params{corpid: corp_id, corpsecret: secret}) data resp.json() if data.get(errcode) ! 0: raise RuntimeError(data.get(errmsg)) return data[access_token]成员读取def list_department_users(token, department_id1): url https://qyapi.weixin.qq.com/cgi-bin/user/list resp requests.get(url, params{ access_token: token, department_id: department_id, fetch_child: 1 }) data resp.json() return data.get(userlist, [])这个接口适合做组织架构同步。我有个项目就是要每天把企业微信组织架构同步到内部BI系统跑了一年都很稳。配合外部联系人接口还能把客户/合作伙伴的维度也纳进来。做同步的时候注意先拉部门树再逐部门拉成员不要直接翻全量容易超时。3.3 推送和拉取的取舍什么场景用哪条路场景推荐方式原因把采集结果发给群成员看群机器人Webhook无需单独开发直接push把企微里的组织信息同步到第三方系统API拉取数据主动权在自己手里实时接收员工发给应用的消息消息回调需要解密事件驱动延迟低采集历史审批/考勤数据API拉取优先钉钉数据完整可分页这个取舍的关键在于你要的是“数据出去”还是“数据进来”。采集类的需求主动权在自己手里更重要通知类的需求及时性更重要。在一套系统里两条路经常是并用的。4. 把采集脚本升级成自动流水线调度、存储和异常自愈4.1 三种调度方案对比别一上来就上服务器集群很多人代码写好了却挂在“怎么定时跑”上。我按规模从轻到重说本地crontab适合个人电脑或一台内网服务器。写一个cron表达式每天凌晨跑一次采集。缺点是没有跨机容错。APScheduler在Python进程里做定时任务。适合要传参、要做复杂任务链的场景。云函数/容器定时任务适合数据量上来以后想要多地域容灾和自动扩缩容。函数计算服务可以按调用次数收费成本很低。我的习惯是小项目APScheduler大项目云函数。一开始不建议堆K8s。我之前见过一个团队连采集脚本带调度都往K8s里塞结果光维护集群就花了三倍于开发的时间完全不值得。4.2 落库设计SQLite起步注意幂等和增量采集数据最终要落到库里。数据量不大的阶段SQLite完全够用。设计表的时候一定给业务唯一ID建唯一索引。比如审批实例在钉钉里有一个instance_id每次采到同样的ID就直接跳过这就实现了幂等。CREATE TABLE IF NOT EXISTS dingtalk_process_instance ( instance_id TEXT PRIMARY KEY, process_code TEXT, title TEXT, status TEXT, create_time DATETIME, finish_time DATETIME, raw_data TEXT, collected_at DATETIME DEFAULT CURRENT_TIMESTAMP );增量采集的核心是有个游标。每天只拉上次时间戳之后的新数据。你可以把最近一次成功采集的时间存到meta表里下次从那个时间开始。我第一次做的时候全量重采数据量大了以后接口频繁超时改成增量后压力瞬间小了很多。4.3 失败告警让机器人把采集异常推给你自动化采集最怕“静默失败”——任务挂了没人知道。我的做法是在每个采集任务外面包一层try/except一旦异常就调用群机器人Webhook把错误信息推到群里。这样当天早上就知道昨晚任务有没有成功。def safe_run(task_func, webhook_url, task_name): try: task_func() except Exception as e: send_wx_markdown(webhook_url, f采集任务{task_name}失败\n错误{e}) raise日志也别省。logging模块输出到文件加上RotatingFileHandler按大小滚动出错时有据可查。我一般会保留最近30天的日志排查历史问题的时候特别有用。5. 真实踩坑记录token、时区、加密和限流5.1 token缓存没做好被限流是小事关键影响全链路前面提过token过期时间是7200秒。但很多人不知道的是频繁获取token本身就会触发频率限制。缓存token的标准姿势不管用文件还是Redis都要把过期时间设置为比7200秒稍短比如7100秒。这样即使本地时钟和服务器有偏差也不会用上已过期token。我之前有个项目token缓存写好后没做过期时间判断结果有一次程序跑了一下午到晚上token实际已经失效了但本地文件里的“假token”还在被使用导致整晚采集全部失败。后来我学乖了每次调接口前都检查一下token的本地过期时间快到了就主动刷新宁可多拿一次也不用失效的。5.2 时间戳的“毫秒”和“秒”坑钉钉和企微的接口时间戳大多用毫秒。如果你拿Python的time.time()直接用那是秒差了1000倍接口会返回“无效时间”。一个通用小工具import time def now_ms(): return int(time.time() * 1000)审批单的start_time和end_time要用这个。企业微信写消息回调里的timestamp也用同样的逻辑。另外把毫秒时间戳转成可读时间时注意用本地时区还是UTC。我习惯把数据统一存成UTC时间展示时再转本地时区避免不同服务器之间出现时区偏差。5.3 企业微信回调消息的AES解密别自己造轮子企微的接收消息回调默认会把请求体做AES加密。官方提供了WXBizMsgCrypt的Python SDK示例。别自己在网上找野路子解密直接用官方库是最省事的。我踩过最大的坑是回调URL配置成功但消息始终解不出来后来发现是编码问题response里必须严格按照官方指定的JSON格式返回需要含encrypt字段。具体来说收到回调请求后先用官方SDK解出明文消息处理业务逻辑然后要返回一个加密的成功响应。如果你的响应格式不对企微会认为回调失败重试几次之后就不推了。这块调试的时候建议先打印原始请求的body和签名参数逐项核对。5.4 频率配额对照表哪些接口容易撞墙接口类型常见限制我的建议钉钉获取token按应用维度有限流必须本地缓存钉钉审批实例查询QPS不高分页拉取慢一点加sleep批量小步拉企微通讯录读取QPS限制明显全量同步控制在凌晨执行群机器人Webhook每分钟20条左右聚合消息别一条条打这个表里的具体数字以官方文档为准但方向是对的宁可慢不要触发全局限流。我之前在企微上做全量通讯录同步一开始循环不设sleep跑到一半就被限流了整个任务要等几分钟才能继续。后来改成每拉完一个部门就sleep 0.5秒一路顺畅。5.5 越权风险采集范围必须卡死在授权边界内最后说一个踩过阴影的坑。有一次我做企业微信通知机器人配置权限时没细看一个“读取通讯录”权限勾上去了。后来PM说“反正都通了顺便把全公司通讯录也同步了吧”。我细想后赶紧踩了刹车因为在未经充分授权和告知的前提下全量同步通讯录到内部系统风险远大于收益。正确的做法是每个数据项都要问一句“这个数据我们真的有权限拿吗拿来做什么存多久谁会看”。把范围卡死在授权边界内比任何技术方案都重要。我后来给自己定了个规矩所有采集项目的权限清单必须由管理员签字确认代码里也做成白名单配置文件不写死在业务代码里方便审计。从我自己的项目经验看用Python接钉钉和企业微信的官方接口做办公数据自动化采集最大的门槛其实不是代码而是“搞清楚你的数据边界”和“愿意耐心看文档”。动手之前把权限理清把token缓存想好把增量逻辑设计好后面跑起来就很顺。做出来之后你会发现这个小小的采集程序能帮整个团队省下大量的手工统计时间——审批核对、考勤汇总、告警播报全都可以自动跑。如果你也正打算搞一套建议先从一个最小的接口跑通闭环再加调度、加落库一步步来就好。这些坑我都替你们踩过了照着这个思路走能省不少时间。