使用python把氚云数据同步至本地MYSQL做数据分析

注册氚云社区，学习低代码知识，与更多氚友互动交流

您需要登录才可以下载或查看，没有帐号？立即注册

x

本帖最后由志伟于 2020-12-30 16:54 编辑

公司的数据比较多，需要生成复杂的中国式报表，使用氚云的报表功能暂时未能满足，只能把数据先同步到MYSQL再使用其他报表功能生成适合公司使用的报表。程序已经运行一年，同步数据100W以上，在这里把代码分享出来，方便有同样需求的朋友。qzw本表单字段比较多，没有做表单字段增减，源码复制。

1、创建conf.py设置基础信息（sql连接、平台信息、需要同步的字段等等）

# -*- coding: utf-8 -*-
"""
Created on Wed Dec 30 14:55:38 2020
@author: qzw
"""
import pymysql
from DBUtils.PooledDB import PooledDB
def conn_db():
pool = PooledDB(pymysql, 3, host="localhost", user="root", passwd="XXX", db="test", port=3306) # 3为连接池里的最少连接数
conn = pool.connection()
cursor = conn.cursor()
return conn, cursor
# 表的所有字段后期修改表，修改这里的字段即可， columns不要换行！！！！！！！
columns="""ObjectId,Name,CreatedById,CreatedByName,OwnerId,OwnerName,OwnerDeptId,OwnerDeptName,F0000030Id,F0000030Name,F0000004Id,F0000004Name,F0000005Id,F0000005Name,F0000007Id,F0000007Name,CreatedTime,ModifiedBy,ModifiedTime,WorkflowInstanceId,Status,SeqNo,F0000032,F0000052,F0000008,F0000046,F0000137,F0000044,F0000045,F0000053,F0000103,F0000055,F0000104,F0000057,F0000105,F0000059,F0000060,F0000061,F0000062,F0000063,F0000064,F0000106,F0000069,F0000101,F0000065,F0000138,F0000066,F0000136,F0000070,F0000067,F0000139,F0000068,F0000112,F0000071,F0000072,F0000140,F0000073,F0000100,F0000074,F0000075,F0000141,F0000076,F0000099,F0000077,F0000078,F0000142,F0000079,F0000098,F0000081,F0000102,F0000082,F0000143,F0000083,F0000084,F0000080,F0000085,F0000086,F0000144,F0000145,F0000087,F0000088,F0000089,F0000090,F0000131,F0000130,F0000091,F0000092,F0000146,F0000147,F0000093,F0000094,F0000095,F0000096,F0000132,F0000133,F0000107,F0000148,F0000109,F0000110,F0000111,F0000108,F0000113,F0000116,F0000114,F0000149,F0000115,F0000118,F0000117,F0000119,F0000120,F0000150,F0000121,F0000123,F0000122,F0000124,F0000125,F0000126,F0000151,F0000128,F0000127,F0000135,F0000134"""
# 默认为0的字段
default_0 = ['F0000103', 'F0000104', 'F0000105', 'F0000059', 'F0000060', 'F0000061', 'F0000064', 'F0000106',
'F0000101', 'F0000066', 'F0000136', 'F0000068', 'F0000112', 'F0000073', 'F0000100', 'F0000076',
'F0000099', 'F0000077', 'F0000079', 'F0000098', 'F0000102', 'F0000083', 'F0000084', 'F0000080',
'F0000086', 'F0000087', 'F0000088', 'F0000089', 'F0000090', 'F0000130', 'F0000092', 'F0000093',
'F0000094', 'F0000095', 'F0000096', 'F0000133', 'F0000109', 'F0000110', 'F0000111', 'F0000108',
'F0000113', 'F0000116', 'F0000115', 'F0000117', 'F0000119', 'F0000121', 'F0000122', 'F0000124',
'F0000125', 'F0000128', 'F0000127', 'F0000135', 'F0000134', 'F0000103']
table_name = "test" # 数据表名字
ActionName = "LoadBizObjects"
SchemaCode = "表单ID"
EngineCode = "你的EngineCode "
EngineSecret = "你的EngineSecret "

复制代码

2、主程序

# -*- coding: utf-8 -*-
"""
Created on Wed Dec 30 14:55:38 2020
@author: qzw
"""
import requests
import json
from conf import conn_db, table_name, ActionName, SchemaCode, EngineCode, EngineSecret, columns
url = 'https://www.h3yun.com/OpenApi/Invoke'
headers = {
"Content-Type": "Application/json",
"EngineCode": EngineCode,
"EngineSecret": EngineSecret
}
def existed_data(sql):
conn, cursor = conn_db()
cursor.execute(sql)
try:
exist = [d[0] for d in cursor.fetchall()]
except Exception as e:
exist = []
return exist
def insert_data(data, sql):
if data:
conn, cursor = conn_db()
try:
print("开始插入数据...")
cursor.executemany(sql, data)
conn.commit()
print("数据插入成功")
except Exception as e:
print(e)
conn.rollback()
else:
print('无数据插入')
def get_chuanyun_data(value, index):
print('获取数据中...')
chuanyun = []
# for index in range(0, 1000000, 500):
Filter = {
"FromRowNum": index,
"RequireCount": False,
"ReturnItems": [],
"SortByCollection": [],
"ToRowNum": index + 500,
"Matcher": {
"Type": "And", "Matchers": [
{"Type": "And", "Matchers": [{"Type": "Item", "Name": "SeqNo", "Operator": 7, "Value": value}]}
]
}
}
params = {
"ActionName": ActionName,
"SchemaCode": SchemaCode,
"Filter": json.dumps(Filter)
}
response = requests.post(url, headers=headers, data=json.dumps(params))
data = json.loads(response.text)
try:
dt = data["ReturnData"]['BizObjectArray']
chuanyun += dt
except Exception as e:
print(e)
if index == 0:
return []
else:
index += 500
response = requests.post(url, headers=headers, data=json.dumps(params))
data = json.loads(response.text)
try:
dt = data["ReturnData"]['BizObjectArray']
chuanyun += dt
except Exception as e:
return []
print('获取数据成功')
return chuanyun
def format_data(data):
print('数据格式化中...')
ret_data = []
if data:
for dt in data:
t = []
cls = columns.split(',')
try:
dt.pop('CreatedBy')
except:
pass
try:
dt.pop('OwnerId')
except:
pass
try:
dt.pop('OwnerDeptId')
except:
pass
try:
dt.pop('F0000030')
except:
pass
try:
dt.pop('F0000004')
except:
pass
try:
dt.pop('F0000005')
except:
pass
try:
dt.pop('F0000007')
except:
pass
try:
CreatedTime = dt['CreatedTime']
CreatedTime = CreatedTime.split(' ')[0] if CreatedTime else ''
dt['CreatedTime'] = CreatedTime
except:
pass
try:
F0000008 = dt['F0000008']
F0000008 = '/'.join((F0000008.split(' ')[0]).split('/')[:2]) if F0000008 else ''
dt['F0000008'] = F0000008
except:
pass
try:
CreatedByObject = dt.pop('CreatedByObject')
dt['CreatedById'] = CreatedByObject.pop('ObjectId')
dt['CreatedByName'] = CreatedByObject.pop('Name')
except:
dt['CreatedById'] = ''
dt['CreatedByName'] = ''
try:
OwnerIdObject = dt.pop('OwnerIdObject')
dt['OwnerId'] = OwnerIdObject.pop('ObjectId')
dt['OwnerName'] = OwnerIdObject.pop('Name')
except:
dt['OwnerId'] = ''
dt['OwnerName'] = ''
try:
OwnerDeptIdObject = dt.pop('OwnerDeptIdObject')
dt['OwnerDeptId'] = OwnerDeptIdObject.pop('ObjectId')
dt['OwnerDeptName'] = OwnerDeptIdObject.pop('Name')
except:
dt['OwnerDeptId'] = ''
dt['OwnerDeptName'] = ''
try:
F0000030Object = dt.pop('F0000030Object')
dt['F0000030Id'] = F0000030Object.pop('ObjectId')
dt['F0000030Name'] = F0000030Object.pop('Name')
except:
dt['F0000030Id'] = ''
dt['F0000030Name'] = ''
try:
F0000004Object = dt.pop('F0000004Object')
dt['F0000004Id'] = F0000004Object.pop('ObjectId')
dt['F0000004Name'] = F0000004Object.pop('Name')
except:
dt['F0000004Id'] = ''
dt['F0000004Name'] = ''
try:
F0000005Object = dt.pop('F0000005Object')
dt['F0000005Id'] = F0000005Object.pop('ObjectId')
dt['F0000005Name'] = F0000005Object.pop('Name')
except:
dt['F0000005Id'] = ''
dt['F0000005Name'] = ''
try:
F0000007Object = dt.pop('F0000007Object')
dt['F0000007Id'] = F0000007Object.pop('ObjectId')
dt['F0000007Name'] = F0000007Object.pop('Name')
except:
dt['F0000007Id'] = ''
dt['F0000007Name'] = ''
for cl in cls:
value = dt[cl] if dt[cl] else ''
t.append(value)
ret_data.append(tuple(t))
print('数据格式化成功')
return tuple(ret_data)
def run():
sql = f"""select SeqNo from {table_name}"""
index = 0
while 1:
exd = existed_data(sql)
chuanyun_dt = get_chuanyun_data(exd, index)
if not chuanyun_dt:
break
new_data = format_data(chuanyun_dt)
insert_sql = f"""INSERT INTO {table_name}({columns}) VALUES ({','.join(['%s'] * len(columns.split(',')))})"""
insert_data(new_data, insert_sql)
index += 500
if __name__ == '__main__':
run()