ETL生成相关报表后, 自动生成Excel或者其他格式的数据文件, 自动上传到钉盘, 通知用户查看
在 钉盘-团队文件 下新建 数据共享/report 文件夹
需要 pyqueen>=1.2.5
pip install pyqueen -i https://pypi.tuna.tsinghua.edu.cn/simple --upgrade from pyqueen import DataSource, TimeKit# from settings import DATABASESDATABASES = {'1001':{'conn_type': 'mysql','host': '', 'username': 'root', 'password': '', 'port': '3306', 'db_name': 'dw' }}ds = DataSource(**DATABASES['1001'])sql_detail = '''SELECT '2026-05-12' as '日期', 12333.5 as '销售金额', 0.3514 as '毛利率'union allSELECT '2026-05-13' as '日期', 12333.5 as '销售金额', 0.3514 as '毛利率''''sql_sumary = '''SELECT 202605 as '月份', 12333.5 as '销售金额', 0.3514 as '毛利率''''df_detail = ds.read_sql(sql_detail)df_summary = ds.read_sql(sql_sumary)fmt = {'月份':'###0','销售金额': '#,##0.0','毛利率': '0.00%'}file_path = ds.to_excel(sheet_list=[[df_summary,'汇总'],[df_detail,'明细']], file_path='销售日报.xlsx', fmt=fmt)from pyqueen import DataSource, TimeKit# from settings import DATABASESDATABASES = {'1001':{'conn_type': 'mysql','host': '', 'username': 'root', 'password': '', 'port': '3306', 'db_name': 'dw' }}ds = DataSource(**DATABASES['1001'])sql = '''SELECT '2026-05-12' as '日期', 12333.5 as '销售金额', 0.3514 as '毛利率'union allSELECT '2026-05-13' as '日期', 12333.5 as '销售金额', 0.3514 as '毛利率''''df = ds.read_sql(sql_detail)'''- v2c: 将指定值(v)填入指定单元格(c)- df2c_v: 将DataFrame的值填入Excel(仅数据,不包含表头),从指定单元格(c)开始- df2c_hv: 将DataFrame填入Excel(包含表头和数据),从指定单元格(c)开始- update: 刷新Excel中所有公式 - sheet (可选): 指定工作表名称,只刷新该工作表的公式 - r (可选): 指定单元格或范围,如 "B3" 或 "A3:B6",只刷新指定范围的公式'''script = [ {'action': 'v2c','sheet':'月报', 'v':'3月报表','c':'A1'}, {'action': 'df2c_v','sheet':'月报', 'df': df,'c':'A2'}, {'action': 'df2c_hv','sheet':'月报', 'df': df,'c':'A5'}, {"action": "update", "sheet": "月报"}]ds.update_excel(file_path='销售日报.xlsx',script=script)from pyqueen import DingtalkApp# from settings import DINGAPP# 联系钉钉管理员获取应用凭证DINGAPP = {'6001':{'app_key' = "",'app_secret' = "",'agent_id' = "" }}da = DingtalkApp(**DINGAPP['6001'])user_id = ""da.set_user_id(user_id=user_id)# 如果不知道如何获取 user_id # 根据手机号查询后配置即可# print(da.mobile2uid('1xxxxxxxxx'))# 也可以直接用手机号, 每次会多请求一次接口# da.set_user_id(moble='1xxxxxxxxx') folder_path = "数据共享/report"file_path = "/data/月报.xlsx"da.upload_file(remote_folder=folder_path, local_file_path=file_path)from pyqueen import Dingtalk# from settings import ROBOTSROBOTS = {'2001': {'access_token' : '','secret' : '' }}ding = Dingtalk(**ROBOTS['2001'])ding.send_markdown(title='数据更新', text='月报数据已经更新, 请前往钉盘查收')from pyqueen import DingtalkApp# from settings import DINGAPP# 联系钉钉管理员获取应用凭证DINGAPP = {'6001':{'app_key': "",'app_secret' : "",'agent_id' : "" }}da = DingtalkApp(**DINGAPP['6001'])# 目标用户列表# 用 print(da.mobile2uid('1xxxxxxxxx')) 获取用户user_id_list = '11111,33333'da.send_work_msg( userid_list=user_id_list, msg={"msgtype": "text","text": {"content": "月报数据已经更新, 请前往钉盘查收" } })# 消息也支持 markdownmd_text = '''标题# 一级标题## 二级标题### 三级标题#### 四级标题##### 五级标题###### 六级标题引用> A man who stands for nothing will fall for anything.文字加粗、斜体**bold***italic*链接[this is a link](http://name.com)'''da.send_work_msg( userid_list=user_id_list, msg={"msgtype": "markdown","markdown": {"title": "数据更新","text": md_text } })