|
上次假期，制作了一个学生请假系统，并写了一下实施思路，大家比较喜欢，一直想写技术细节，现在暑假又过了一半，对系统也做了修正，把写技术细节的愿望实现一下。下面的链接是上次的文章
企业微信+微信+简道云+API：创建贴心实用的学生请假系统
接下来是本次书写的技术方案。
01 学生的信息库建立
1.从企业微信家校通讯录导出学生的信息，获取student_userId。点击下图位置导出家校通讯录。
2.打开导出的通讯录，可以看到下图中的学生账号。
3.在简道云中建立数据表单，方便查看和修改。数据库中含学生的身份证号码，供以后家长绑订学生时使用。
4.建立mysql数据表单
5.把导出的student_userId和学生的学号信息精准对应，分别导入到简道云和mysql数据库中
6.简道云学生数据表单，新建数据推送，和mysql数据库表单实现实时同步
02 云家校主页
1.主页预览
2.表单属性设置及主页选项卡
3.打卡记录选项卡
4.请假记录选项卡
5.表单外链样式
03 家长绑订学生
1.微信绑订学生表单
（1）界面
（2）表单字段
（3）表单添加的智能助手
2.微信绑订学生数据库表单
（1）界面
（2）字段
3.删除我绑订的学生
（1）界面
（2）字段
（3）添加的智能助手
04 学生请假流程表单
1.界面
2.流程设定
3.新建数据推送
06 学校返校前上传核酸证明
1.界面
2.表单字段
3.添加智能助手
06 发送家长通知的代码
1.接收请假表单数据推送的网络路径代码
@jiandaoyun.route('/app1/', methods = ['POST'])
# 这一段接收数据的语句不用修改
def hello_world():
# 从简道云表单根据A表单动作触发获取值
sj = json.loads(request.data)
# print(sj)
op = sj['op']
data = sj['data']
if op == "data_test": # 测试触发的这个语句
print("测试成功")
return '200'
else:
# 多线程进行数据处理程序
threading.Thread(target = Processing_data().from_to_defname, args = (op, data,)).start()
return 'success', 200
2.处理接收到学生请假表单推送的数据代码
class Processing_data(object):
def create_tables(self):
pass
def from_to_defname(self, op, data):
appId_entryId = data['appId'] + data['entryId']
# 学习调试—智能助手案例-A表
if appId_entryId == "5c44742d930c2878a14fcb885ec227cdd67f73000676c419":
self.main1(op, data)
def main1(self,op,data):
print("疫情防控——学生请假")
print("op:",op)
print("data:", data)
banzhuren_yijian=data["banzhuren_yijian"]
banzhuren_name=data["banzhuren_name"]
xuehao_xingming=data["xuehao_xingming"]
qingjia_yuanyin=data["qingjia_yuanyin"]
print(qingjia_yuanyin)
hehaizi_guanxi=data["hehaizi_guanxi"]
print(hehaizi_guanxi)
jiazhang_shouji=data["jiazhang_shouji"]
print(jiazhang_shouji)
start_time=data["start_time"]
start_wubie=data["start_wubie"]
end_time=data["end_time"]
end_wubie=data["end_wubie"]
shifou_hesuan = data["shifou_hesuan"]
fanxiao_shijian = data["fanxiao_shijian"]
print(start_time)
Jiaxiaogoutong().send_xuexiaotongzhi_xueshengqingjia(xuehao_xingming,op,qingjia_yuanyin,hehaizi_guanxi,jiazhang_shouji,start_time,start_wubie,end_time,end_wubie,banzhuren_name,banzhuren_yijian,shifou_hesuan,fanxiao_shijian)
class Jiaxiaogoutong(): # 这个是家校沟通发送消息的功能
def send_xuexiaotongzhi_xueshengkaoqin(self,xuehaoxingming_list,wubie,chuqin_jilu,shangbaoren_T):
riqi = datetime.datetime.today().strftime("%Y-%m-%d")
sql_sel_stu_userid = "SELECT stu_userid FROM student_info WHERE xuehao_xingming = %s LIMIT 1"
print("xuehaoxingming_list:",xuehaoxingming_list)
for xuehao_xingming in xuehaoxingming_list:
userids=MySQLUtil().select_one_none(sql_sel_stu_userid, xuehao_xingming)
print("userids:",userids)
if userids is not None:
xingming = xuehao_xingming[8:]
WeChat().msg_text_school(xingming,userids,riqi,wubie,chuqin_jilu,shangbaoren_T)
def send_xuexiaotongzhi_xueshengqingjia(self,xuehao_xingming,op,qingjia_yuanyin,hehaizi_guanxi,jiazhang_shouji,start_time,start_wubie,end_time,end_wubie,banzhuren_name,banzhuren_yijian,shifou_hesuan,fanxiao_shijian):
riqi = datetime.datetime.today().strftime("%Y-%m-%d")
start_time=JdyFuctionUtils().get_local_time(start_time)
print(start_time)
start_time=start_time.strftime("%Y-%m-%d")+start_wubie
end_time=JdyFuctionUtils().get_local_time(end_time)
end_time=end_time.strftime("%Y-%m-%d")+end_wubie
sql_sel_stu_userid = "SELECT stu_userid FROM student_info WHERE xuehao_xingming = %s LIMIT 1"
userids=MySQLUtil().select_one_none(sql_sel_stu_userid, xuehao_xingming)
print("userids:",userids)
print(fanxiao_shijian)
if op == "data_create":
if userids is not None:
xingming = xuehao_xingming[8:]
WeChat().msg_text_school_qingjia_satrt(xingming,userids,riqi,qingjia_yuanyin,hehaizi_guanxi,jiazhang_shouji,start_time,end_time)
elif op == "data_update" and fanxiao_shijian is None:
if userids is not None:
xingming = xuehao_xingming[8:]
WeChat().msg_text_school_qingjia_end(xingming,userids,riqi,qingjia_yuanyin,hehaizi_guanxi,jiazhang_shouji,start_time,end_time,banzhuren_name,banzhuren_yijian,shifou_hesuan)
3.根据相关流程，发送家长通知的代码
# !/usr/bin/env python
# -*- coding: utf-8 -*-
'''
1.这是处理学生请假，给家长发送通知的文件
'''
from flask import Flask, request, render_template
import http.client
import datetime,time
from datetime import timedelta
from apscheduler.schedulers.blocking import BlockingScheduler
import requests
import json
import threading
import math
from wx_util.yiqingdaka import WX_ARG
import os
import sys
current_directory = os.path.dirname(os.path.abspath(__file__))
sys.path.append("../")
'''这是企业微信的类'''
class WeChat(object):
'''
下面是构造函数
1.成员函数的函数名可以自定义，但是，构造函数的函数名是固定的__init__
2.成员函数需要被手动调用，但是，构造函数在创建对象的过程中是自动被调用的对于同一个对象而言
3.成员函数可以被调用多次，但是，构造函数只能被调用一次
4.构造函数的特点：创建对象；给对象的成员变量赋值
5.构造函数也被称为构造器，当创建对象的时候第一个被自动调用的函数
'''
def __init__(self):
self.CORPID = WX_ARG.corpid # 企业ID，在管理后台获取
self.CORPSECRET = WX_ARG.corpsecret # 自建应用的Secret，每个自建应用里都有单独的secret
self.TOUSER = "" # 接收者用户名,多个用户用|分割
self.CORPNAME = WX_ARG.corpname # 给应用起一个名字，方便在后面写入access_token时调用
self.access_token_filename=self.CORPNAME+'_access_token.conf'
'''获取我的企业微信中家校通讯录的Access Token'''
def _get_access_token(self):
url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'
values = {'corpid': self.CORPID,
'corpsecret': self.CORPSECRET,
}
req = requests.post(url, params=values)
data = json.loads(req.text)#转换为json格式
# print(data)
return data["access_token"]#返回data中的"access_token"
def get_access_token(self):
cur_time = time.time()#获取现在的时间
# print('执行的foxtable access_token')
try:
# print('执行获取access_token的try,读取文件内的Access_token,进行判断')
with open(current_directory +'/temp/'+self.access_token_filename+'', 'r') as f:#读取文件中的t即上次获取access token的时间和access token的值
t, access_token = f.read().split()#以空格为分格符，分别给t和access_token赋值
# print(t)
# print(access_token)
if 0 < cur_time - float(t) < 7260:#判断access token是否在有效期内，如果是，则继续使用access token
# print('access_token在有效范围内')
# print(access_token)
return access_token
else:
with open(current_directory +'/temp/'+self.access_token_filename+'', 'w') as f:#否则，先打开文件，权限可以编辑
access_token = self._get_access_token()#获取access token
f.write('\t'.join([str(cur_time), access_token]))#把获取到的当前时间和Access token写入文件
return access_token#返回access token的值
except:#如果是第一次运行，文件中为空值，则运行此语句，获取access token的值
with open(current_directory +'/temp/'+self.access_token_filename+'', 'w') as f:
access_token = self._get_access_token()
cur_time = time.time()
f.write('\t'.join([str(cur_time), access_token]))
# print("执行获取access_token的except")
return access_token
def send_school(self,data_str): # 发送学校通知
return self.post_request_send(WX_ARG.url_send_xuexiaotongzhi.format(self.get_access_token()), data_str)
def get_rquest(self,url):#所有的get请求
try:
conn = http.client.HTTPSConnection(WX_ARG.url)
conn.request('GET', url)
response = conn.getresponse()
# print(response)
arg = eval(response.read().decode())
# print(arg)
return arg
except:
# print(sys.exc_info()[0])
return None
finally:
conn.close()
def post_request(self,url, data_str):#所有的post请求
# print("进入post环节")
try:
# print("进入try")
conn = http.client.HTTPSConnection(WX_ARG.url)
conn.request('POST', url, data_str)
response = conn.getresponse()
# print('response:', response)
arg = eval(response.read().decode())
# print('arg:', arg)
return arg
except:
# print("进入except")
print(sys.exc_info()[0])
return None
finally:
conn.close()
def post_request_send(self,url, data_str):#所有的post请求
try:
conn = http.client.HTTPSConnection(WX_ARG.url)
print("进入尝试发送环节")
conn.request('POST', url, data_str)
response = conn.getresponse() #值为
print("发送后的返回值",response)
arg = eval(response.read().decode()) #转换后为{'errcode': 0, 'errmsg': 'ok', 'invaliduser': ''}
print(arg["errmsg"])
return arg["errmsg"]
except:
print(sys.exc_info()[0])#正常情况下为(None, None, None)0的值也为None
return None
finally:
conn.close()
def msg_text_school(self, xingming,userids,riqi,wubie,chuqin_jilu,shangbaoren_T):
send_values = {
"to_student_userid": [userids],
"msgtype" : "text",
"agentid": WX_ARG.corpagentld,
"text" : {
"content" : "您宝贝的课后服务考勤\n姓名："+ xingming + "\n日期："+ riqi + "\n午别："+ wubie + "\n考勤："+ chuqin_jilu + "\n教师："+ shangbaoren_T +"\n\n点击可查看考勤详情\n"
}
}
# print("进入发送环节")
send_msges = (bytes(json.dumps(send_values), 'utf-8'))
self.send_school(send_msges)
def msg_text_school_qingjia_satrt(self,xingming,userids,riqi,qingjia_yuanyin,hehaizi_guanxi,jiazhang_shouji,start_time,end_time):
send_values = {
"to_student_userid": [userids],
"msgtype": "text",
"agentid": WX_ARG.corpagentld,
"text": {
"content":"" + xingming + "的家长发起了请假申请\n学生姓名："+ xingming + "\n开始时间："+ start_time + "\n结束时间："+ end_time + "\n请假原因："+ qingjia_yuanyin + "\n发起人："+ xingming + "" + hehaizi_guanxi + "\n发起人手机："+ jiazhang_shouji +"\n查看更多\n"
}
}
# print("进入发送环节")
send_msges = (bytes(json.dumps(send_values), 'utf-8'))
self.send_school(send_msges)
def msg_text_school_qingjia_end(self,xingming,userids,riqi,qingjia_yuanyin,hehaizi_guanxi,jiazhang_shouji,start_time,end_time,banzhuren_name,banzhuren_yijian,shifou_hesuan):
if shifou_hesuan == "是":
send_values = {
"to_student_userid": [userids],
"msgtype": "text",
"agentid": WX_ARG.corpagentld,
"text": {
"content": ""+ xingming + "的请假审批意见\n班主任："+ banzhuren_name + "\n班主任意见："+ banzhuren_yijian +"，您的学生返校前需要点击击本链接，提交核酸检测阴性证明\n学生姓名："+ xingming + "\n假期开始："+ start_time + "\n假期结束："+ end_time + "\n请假原因："+ qingjia_yuanyin + "\n发起人："+ xingming + "" + hehaizi_guanxi + "\n发起人手机："+ jiazhang_shouji +"\n"
}
}
else:
send_values = {
"to_student_userid": [userids],
"msgtype": "text",
"agentid": WX_ARG.corpagentld,
"text": {
"content": ""+ xingming + "的请假审批意见\n班主任：" + banzhuren_name + "\n班主任意见：" + banzhuren_yijian + "\n\n学生姓名：" + xingming + "\n假期开始：" + start_time + "\n假期结束：" + end_time + "\n请假原因：" + qingjia_yuanyin + "\n发起人：" + xingming +""+ hehaizi_guanxi + "\n发起人手机：" + jiazhang_shouji + "\n查看更多\n"
}
}
# print("进入发送环节")
send_msges = (bytes(json.dumps(send_values), 'utf-8'))
self.send_school(send_msges)
WX_ARG.py文件内容
# -*- coding: utf-8 -*-
'''
请假审批的应用
修改需要的值，就在这个地方进行修改
'''
corpid = '*******'
corpsecret = '******'
corpname = '******'
corpagentld = '*******' # 疫情打卡的id号
url = 'qyapi.weixin.qq.com'
'''base最基本的链接'''
url_send_xuexiaotongzhi = '/cgi-bin/externalcontact/message/send?access_token={}'
'''发送学校通知'''
|
给您点赞
