目录
企微群机器人使用
APScheduler概要使用说明
cron方式设置
场景设置案例
表结构
代码
场景：对于周期性重复需要做的事情，每次都需要人工去提醒，容易忘记，而企业微信群可以添加群机器人，群机器人提供接口能力，按照接口格式说明把消息内容发到群里以及@相关人，达到提醒的目的。群机器人只是能让我们发送消息，而定时发送的实现这里采用python的APScheduler库实现。
正好，企业微信群有一个机器人，正可以实现这一功能。
刚开始只是在测试团队内部使用，也是写着玩玩，后来项目内使用的人多了就做成了数据库配置化，目前这套代码实际使用一年了，过程也遇到不同的场景也不断的优化。
简要设计思路：根据使用场景、机器人的接口文档，设置适合的数据库表结构，使用Python编写逻辑代码，利用APScheduler模块实现定时提醒的功能。
企微群机器人使用
机器人如何添加以及机器人的接口说明请看官方的信息文档，这里就不复制了~
使用说明：如何设置群机器人 -帮助中心-企业微信
接口使用：
注意：Markdown模式无法@所有人，微信开放社区官方也做了说明。
机器人发送MarkDown消息无法@所有人 | 微信开放社区
APScheduler概要使用说明
APScheduler 官网文档
这里不做详细说明，主要概要的记录，大家具体学习可以另行百度哈~
APScheduler 四个组件分别为：触发器(trigger)，作业存储(job store)，执行器(executor)，调度器(scheduler)。
触发器(trigger)
包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置意外，触发器完全是无状态的，APScheduler 有三种内建的 trigger：
date: 特定的时间点触发interval: 固定时间间隔触发
cron: 在特定时间周期性地触发
作业存储(job store)
存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。
APScheduler 默认使用 MemoryJobStore，可以修改使用 DB 存储方案
执行器(executor)
处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。最常用的 executor 有两种：
ProcessPoolExecutorThreadPoolExecutor
调度器(scheduler)
通常在应用中只有一个调度器，应用的开发者通常不会直接处理作业存储、调度器和触发器，相反，调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成，例如添加、修改和移除作业。
cron方式设置
以下只对cron方式设置定时任务进行简单使用说明。
(int|str) 表示参数既可以是int类型，也可以是str类型(datetime | str) 表示参数既可以是datetime类型，也可以是str类型year (int|str) – 4-digit year -（表示四位数的年份，如）month (int|str) – month (1-12) -（表示取值范围为1-12月）day (int|str) – day of the (1-31) -（表示取值范围为1-31日）week (int|str) – ISO week (1-53) -（格里历12月31日可以写成-W52-7（扩展形式）或W527（紧凑形式））day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) - （表示一周中的第几天，既可以用0-6表示也可以用其英语缩写表示）hour (int|str) – hour (0-23) - （表示取值范围为0-23时）minute (int|str) – minute (0-59) - （表示取值范围为0-59分）second (int|str) – second (0-59) - （表示取值范围为0-59秒）start_date (datetime|str) – earliest possible date/time to trigger on (inclusive) - （表示开始时间）end_date (datetime|str) – latest possible date/time to trigger on (inclusive) - （表示结束时间）timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone) -（表示时区取值）
表达式
场景设置案例
#表示3月22日17时19分07秒执行该程序sched.add_job(my_job, 'cron', year=,month = 03,day = 22,hour = 17,minute = 19,second = 07)#表示任务在6,7,8,11,12月份的第三个星期五的00:00,01:00,02:00,03:00 执行该程序sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')#表示从星期一到星期五5:30（AM）直到-05-30 00:00:00sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='-05-30')#表示每5秒执行该程序一次，相当于interval 间隔调度中seconds = 5sched.add_job(my_job, 'cron',second = '*/5')# 装饰器# 每分钟执行一次@scheduler.scheduled_job('cron',minute = '*/1')# 每两秒执行一次@scheduler.scheduled_job('cron',second = '*/2')
最小使用示例
每两秒执行一次
from apscheduler.schedulers.blocking import BlockingSchedulerscheduler = BlockingScheduler(timezone='Asia/Shanghai')@scheduler.scheduled_job('cron',second = '*/2')def mytest():print("每两秒执行一次")scheduler.start()
机器人配置化表结构
robot表，配置机器人的主要内容
CREATE TABLE `robot` (`id` int(11) NOT NULL AUTO_INCREMENT,`group_chat` varchar(40) DEFAULT NULL COMMENT '群',`robot_key` varchar(64) NOT NULL COMMENT '机器人key',`robot_content` varchar(1000) NOT NULL COMMENT '消息内容',`data_type` varchar(64) NOT NULL COMMENT '消息类型，text/markdown',`mentioned_list` varchar(200) DEFAULT NULL COMMENT '企业微信id,提醒群中的指定成员(@某个成员),all表示提醒所有人;如果不知道微信id,可以使用mentioned_mobile_list',`mentioned_mobile_list` varchar(200) DEFAULT NULL COMMENT '手机号列表，提醒手机号对应的群成员(@某个成员),markdown类型不支持',`dict_scope` varchar(50) DEFAULT NULL COMMENT '不是固定@的人则通过,dict表查询',`day_of_week` varchar(30) DEFAULT NULL COMMENT '周几提醒,星期一至周日:1,2,3,4,5,6,7',`hour_minute` varchar(64) DEFAULT NULL COMMENT '提示的时、分，24小时制，精确到分钟',`chinese_calendar` varchar(20) NOT NULL COMMENT '执行时间，workday：工作日才执行;natural_day：自然日执行;not_legal_holiday：法定节日不执行',`person_in_charge` varchar(64) NOT NULL COMMENT '机器人需求者',`delete` varchar(50) DEFAULT NULL COMMENT '1：不执行,0：执行',`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=29 DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='群机器人配置表';
这里给些robot表列子（有手动删除了一些关键信息，插入应该没有误删错些啥吧。。哈哈）
INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (9, '测试一组', '74744e03-fc46-4d59-bc2d-xxxxx', '### <@${wxID}>是本周迭代负责人哟🌟~>点击查看负责人的 [待办事项](/6) ', 'markdown', null, null, 'tester_rotation_1', '1', '09:45;', 'workday', 'kevin', '0', '-08-28 13:09:13', '-07-31 15:52:47');INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (10, '测试一组', '74744e03-fc46-4d59-bc2d-xxxxx', '1、为方便值日人员拉取测试报告，请大家及时更新需求状态、任务状态、BUG状态。2、非发版验证人员填写验证点：https://doc.weixinAaKACY', 'text', 'kevinli;falinechen;', null, null, '5', '17:00;', 'workday', 'kevin', '0', '-08-28 13:09:13', '-05-05 13:04:35');INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (11, '测试一组', '74744e03-fc46-4d59-bc2d-xxxxx', '请查看自己已评审的需求是否已经预估工时，以及已经归属迭代的需求是否已经排期，以便项目经理进行需求迭代排期及人力统计~', 'text', 'kevinli;falinechen;', null, null, '1;2;3', '17:00;', 'workday', 'kevin', '1', '-08-28 13:09:13', '-05-05 13:04:35');INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (12, 'sit和uat环境沟通专门群', 'd7af2748-23fc-4xxx-xxxxx', '### 整点发版~ 点击可快速跳转喔 👉 [流水线](/ec) ', 'markdown', null, null, 'tester_rotation_1', '1;2;3;4;6;7', '10:00;11:00;12:00;14:00;15:00;16:00;17:00;18:00;19:00', 'workday', 'kevin', '0', '-08-28 13:09:13', '-05-05 13:04:35');INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (13, '产品一组', '918ee4e8-1a94-4520-a061-xxxx', '请各位产品尽快评审需求，评审后记得将需求放入【待规划迭代】中。', 'text', 'fafahao;terry', '18576400xxx', null, '1;2;3', '11:00;16:00', 'workday', '', '0', '-08-28 13:09:13', '-06-20 09:19:08');INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (14, '产品一组', '918ee4e8-1a94-4520-a061-xxx', '### 每周五两件事:>1：<font>在tita更新自己的OKR进度和下周计划</font>>2：<font>更新保险科技IT需求中自己的需求状态</font>', 'markdown', 'fafahao;terrychen;', '18576400xxx', null, '5', '17:00;', 'workday', 'Josie', '1', '-08-28 13:09:13', '-06-20 09:19:08');INSERT INTO robot (id, group_chat, robot_key, robot_content, data_type, mentioned_list, mentioned_mobile_list, dict_scope, day_of_week, `hour_minute`, chinese_calendar, person_in_charge, `delete`, created_at, updated_at) VALUES (15, '产品一组', '918ee4e8-1a94-4520-a061-xx5', '各位产品，请记得将写完的需求放入【待评审的需求】中。', 'text', 'fafahao;terrychen;', '18576400xxx', null, '4;5', '17:00;', 'workday', 'Josie', '0', '-08-28 13:09:13', '-06-20 09:19:08');
dict表，对于周期性变化@不同的人，可以把对应的信息配置到这个表。比如，按周进行轮班，每周通过读取数据库的配置信息进行@不同的人。
CREATE TABLE `dict` (`id` int(11) NOT NULL AUTO_INCREMENT,`scope` varchar(50) DEFAULT NULL COMMENT '类型',`attribute` varchar(225) DEFAULT NULL COMMENT '字典名称',`value` varchar(500) DEFAULT NULL COMMENT '字典值',`created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',`remark` varchar(510) DEFAULT NULL COMMENT '备注',PRIMARY KEY (`id`)) ENGINE=InnoDB AUTO_INCREMENT=35 DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='字典项表';
执行代码
代码概要讲解：
总体思路也是比较简单，就是利用APScheduler每分钟通过SQL查询有没有符合条件的机器人配置信息，有符合的就拿去request请求就好了。
# -*- coding: utf-8 -*-# mon,tue,wed,thu,fri,sat,sunfrom apscheduler.schedulers.blocking import BlockingSchedulerimport requestsimport pymysqlfrom datetime import datetimefrom chinese_calendar import is_workdayimport chinese_calendarimport timescheduler = BlockingScheduler()class DB:def __init__(self):self.conn = pymysql.connect(host='****',port=3307,user='root',passwd='abc123456',db='db_tester')self.cur = self.conn.cursor()def __del__(self): # 析构函数，实例删除时触发self.cur.close()self.conn.close()def db_select(self, sql):self.cur.execute(sql)return [dict(zip([key[0] for key in self.cur.description], values)) for values in self.cur.fetchall()]class Request():def __init__(self):self.wx_url = 'https://qyapi./cgi-bin/webhook/send?key='self.headers = {'Content-Type': 'application/json'}self.data_markdown = {'msgtype': 'markdown', 'markdown': {'content': ''}}self.data_text = {'msgtype': 'text', 'text': {'content': "", 'mentioned_list': [], 'mentioned_mobile_list': []}}def request_to_run(self,key,data):r=requests.post(url=self.wx_url+key,headers=self.headers,json=data)print(self.wx_url + key, data)def robot_run(robot_data):# 每个循环重新定义类，做初始初始化数据request_rotbot = Request()robot_key = robot_data['robot_key']mentioned_list_value = ""if robot_data["data_type"] == "text":request_rotbot.data_text['text']['content'] = robot_data['robot_content']if robot_data['mentioned_list'] is None:robot_data['mentioned_list'] = ''if robot_data['mentioned_mobile_list'] is None:robot_data['mentioned_mobile_list'] = ''if robot_data['value'] is None:robot_data['value'] = ''mentioned_list_value = robot_data['mentioned_list'] + ";" + robot_data['mentioned_mobile_list'] + ";" + robot_data['value']for mentioned in mentioned_list_value.split(";"):if mentioned != "":# 如果是手机号if mentioned.isdigit() and len(mentioned) == 11:request_rotbot.data_text['text']['mentioned_mobile_list'].append(mentioned)elif mentioned != "all":request_rotbot.data_text['text']['mentioned_list'].append(mentioned)elif mentioned == "all":mentioned = "@" + mentionedrequest_rotbot.data_text['text']['mentioned_list'].append(mentioned)# 最终mentioned_list或者mentioned_mobile_list不为空才会发请求if request_rotbot.data_text['text']['mentioned_list'] or request_rotbot.data_text['text']['mentioned_mobile_list']:print(request_rotbot.data_text)request_rotbot.request_to_run(key=robot_key, data=request_rotbot.data_text)time.sleep(0.3)if robot_data["data_type"] == "markdown":if robot_data['mentioned_list'] is None:robot_data['mentioned_list'] = ""if robot_data['value'] is None:robot_data['value'] = ""request_rotbot.data_markdown['markdown']['content'] = robot_data['robot_content']mentioned_list_value = robot_data['mentioned_list'] + ";" + robot_data['value']for mentioned in mentioned_list_value.split(";"):# markdown 格式下，过滤手机号 和 @allif mentioned != "" and len(mentioned) != 11 and mentioned != "all" and "${wxID}" not in robot_data['robot_content']:request_rotbot.data_markdown['markdown']['content'] += "<@" + mentioned + ">"if "${wxID}" in robot_data['robot_content']:request_rotbot.data_markdown['markdown']['content'] = robot_data['robot_content'].replace("${wxID}",mentioned)print(request_rotbot.data_markdown)request_rotbot.request_to_run(key=robot_key, data=request_rotbot.data_markdown)time.sleep(0.3)#@scheduler.scheduled_job('cron',second = '*/2')@scheduler.scheduled_job('cron',minute = '*/1')def tester_group_on_duty_1():db = DB()now_time = datetime.now()#第几周number_of_weeks = now_time.isocalendar()[1] +1#星期几：1-7 对应星期一到星期日dayOfWeek = now_time.isoweekday()#时分 18:47datetime_now = now_time.strftime("%H:%M")# 获取当前的时间,格式-05-05date_is_workday = now_time.date()#是否时节假日（周末也是节假日），假日名称（英文名）on_holiday, holiday_name = chinese_calendar.get_holiday_detail(date_is_workday)data_list_sql = '''select r.robot_key,r.robot_content,r.data_type,r.mentioned_list,r.mentioned_mobile_list,d.`value`,r.chinese_calendarfrom robot r LEFT JOIN dict d ON r.dict_scope=d.scopewhere r.dict_scope is null and `delete`=0 and r.day_of_week like "%{day_of_week}%" and `hour_minute` like "%{hour_minute}%"UNIONselect r.robot_key,r.robot_content,r.data_type,r.mentioned_list,r.mentioned_mobile_list,d.`value`,r.chinese_calendarfrom robot r LEFT JOIN dict d ON r.dict_scope=d.scopewhere r.dict_scope is not null and `delete`=0 and r.day_of_week like "%{day_of_week}%" and `hour_minute` like "%{hour_minute}%" and d.attribute = "{number_of_weeks}";'''.format(day_of_week=dayOfWeek,hour_minute=datetime_now,number_of_weeks=number_of_weeks)# 调试机器人# data_list_sql = '''select r.robot_key,r.robot_content,r.data_type,r.mentioned_list,r.mentioned_mobile_list,d.`value`# from robot r LEFT JOIN dict d ON r.dict_scope=d.scope# where r.dict_scope is null# UNION# select r.robot_key,r.robot_content,r.data_type,r.mentioned_list,r.mentioned_mobile_list,d.`value`# from robot r LEFT JOIN dict d ON r.dict_scope=d.scope# where r.dict_scope is not null and d.attribute = 35;# '''# data_list_sql = '''# select r.robot_key,r.robot_content,r.data_type,r.mentioned_list,r.mentioned_mobile_list,d.`value`,r.chinese_calendar# from robot r LEFT JOIN dict d ON r.dict_scope=d.scope# where r.id=25# '''print("sql语句：",data_list_sql)data_list = db.db_select(data_list_sql)print("slq查询到的数据数据：",data_list)try:for data_data in data_list:# 工作日才执行if is_workday(date_is_workday) and data_data["chinese_calendar"] == "workday" :robot_run(robot_data=data_data)# 不是法定节假日时执行elif holiday_name == None and data_data["chinese_calendar"] == "not_legal_holiday":robot_run(robot_data=data_data)# 自然日执行elif data_data["chinese_calendar"] == "natural_day":robot_run(robot_data=data_data)else:raise NameError('数据库中填写的chinese_calendar值找不到！')except Exception as msg:tester_abnormal_alarm_sql = '''select r.robot_key,r.robot_content,r.data_type,r.mentioned_list,r.mentioned_mobile_list,d.`value` from robot r LEFT JOIN dict d ON r.dict_scope=d.scope where r.dict_scope ="tester_abnormal_alarm" and `delete`=0;'''tester_abnormal_alarm = db.db_select(tester_abnormal_alarm_sql)request_rotbot1 = Request()#测试一组的keyrobot_key = tester_abnormal_alarm[0]['robot_key']request_rotbot1.data_text['text']['mentioned_list'].append(tester_abnormal_alarm[0]['mentioned_list'])request_rotbot1.data_text['text']['content'] = "线上执行有异常，请检查！\nsql语句：${data_list_sql}\nsql查询到的内容：${data_list}\n异常信息：{msg}".format(data_list_sql=data_list_sql,data_list=data_list,msg=msg)request_rotbot1.request_to_run(key=robot_key, data=request_rotbot1.data_text)print("执行有异常：",msg) # 用msg接收异常scheduler.start()
如果觉得《企业微信群：机器人定时提醒功能数据库配置化》对你有帮助，请点赞、收藏，并留下你的观点哦！
