ZhengZhou103/target/classes/Py/LoadJson.py

import os
import datetime
from urllib import request
# pip install pandas
# pip install sqlalchemy
# pip install requests
# pip install pymysql
import pandas as pd
import requests

from sqlalchemy import create_engine

import json

import pymysql
from dbutils.pooled_db import PooledDB


class MysqlClient(object):
    __pool = None

    def __init__(self, mincached=10, maxcached=20, maxshared=10, maxconnections=200, blocking=True,
                 maxusage=100, setsession=None, reset=True,
                 host="gtzz.dsideal.com", port=22066, db="gtzz_base_db",
                 user="root", passwd="DsideaL147258369", charset='utf8mb4'):
        """
        :param mincached:连接池中空闲连接的初始数量
        :param maxcached:连接池中空闲连接的最大数量
        :param maxshared:共享连接的最大数量
        :param maxconnections:创建连接池的最大数量
        :param blocking:超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理
        :param maxusage:单个连接的最大重复使用次数
        :param setsession:optional list of SQL commands that may serve to prepare
            the session, e.g. ["set datestyle to ...", "set time zone ..."]
        :param reset:how connections should be reset when returned to the pool
            (False or None to rollback transcations started with begin(),
            True to always issue a rollback for safety's sake)
        :param host:数据库ip地址
        :param port:数据库端口
        :param db:库名
        :param user:用户名
        :param passwd:密码
        :param charset:字符编码
        """

        if not self.__pool:
            self.__class__.__pool = PooledDB(pymysql,
                                             mincached, maxcached,
                                             maxshared, maxconnections, blocking,
                                             maxusage, setsession, reset,
                                             host=host, port=port, db=db,
                                             user=user, passwd=passwd,
                                             charset=charset,
                                             cursorclass=pymysql.cursors.DictCursor
                                             )
        self._conn = None
        self._cursor = None
        self.__get_conn()

    def __get_conn(self):
        self._conn = self.__pool.connection()
        self._cursor = self._conn.cursor()

    def close(self):
        try:
            self._cursor.close()
            self._conn.close()
        except Exception as e:
            print(e)

    def __execute(self, sql, param=()):
        count = self._cursor.execute(sql, param)
        # print(count)
        self._conn.commit()
        return count

    @staticmethod
    def __dict_datetime_obj_to_str(result_dict):
        """把字典里面的datatime对象转成字符串，使json转换不出错"""
        if result_dict:
            result_replace = {k: v.__str__() for k, v in result_dict.items() if isinstance(v, datetime.datetime)}
            result_dict.update(result_replace)
        return result_dict

    def find(self, sql, param=()):
        """查询单个结果"""
        count = self.__execute(sql, param)
        result = self._cursor.fetchone()
        """:type result:dict"""
        result = self.__dict_datetime_obj_to_str(result)
        return count, result

    def findList(self, sql, param=()):
        """
        查询多个结果
        :param sql: qsl语句
        :param param: sql参数
        :return: 结果数量和查询结果集
        """
        count = self.__execute(sql, param)
        result = self._cursor.fetchall()
        """:type result:list"""
        [self.__dict_datetime_obj_to_str(row_dict) for row_dict in result]
        return count, result

    def execute(self, sql, param=()):
        count = self.__execute(sql, param)
        return count

    def begin(self):
        """开启事务"""
        self._conn.autocommit(0)

    def end(self, option='commit'):
        """结束事务"""
        if option == 'commit':
            self._conn.autocommit()
        else:
            self._conn.rollback()


def printf(str):
    now_time = datetime.datetime.now()
    str_time = now_time.strftime("%Y-%m-%d %X")  # 格式化时间字符串
    print(str_time + " " + str)


def SyncData():
    # 连接数据库的方法
    connectString = 'mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8' % (
        "root", "DsideaL147258369", "gtzz.dsideal.com", 22066, "gtzz_base_db")
    engine = create_engine(connectString)  # use sqlalchemy to build link-engine
    # 当前日期
    cnt = 0
    # 申请数据库连接
    db = MysqlClient()

    dateStr = datetime.datetime.now().strftime('%Y%m%d')
    # dateStr = '20211115'
    text_lines = ['https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_PARA.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_ATTACH.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_FORM_VALUE.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_DETAIL.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_FORM_ITEM.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_TARGET.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_ATTACH_TYPE.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_TASK.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_FORM.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_MESSAGE.json',
                  'https://dsideal.obs.cn-north-1.myhuaweicloud.com/down/ccsdjg/#(date)/T_PRO_CONFIG.json'
                  ]
    # 按行遍历
    for line in text_lines:
        cnt = cnt + 1
        url = line.replace('#(date)', dateStr)
        # 去掉最后的换行符
        url = url.rstrip("\n")
        # 表名
        table_name = os.path.basename(url).split(".")[0].lower()

        # 尝试下载
        html = requests.head(url)  # 用head方法去请求资源头部
        if html.status_code != 200:  # 状态码
            printf("没有找到指定的下载文件，程序无法继续！")
            exit(0)
        # 下载文件
        r = request.Request(url)
        response = request.urlopen(r)
        # 中文编码转换
        s = response.read()
        s = s.decode('gbk')
        s = s.encode('gbk', 'ignore').decode('gbk')
        s1 = json.loads(s)
        try:
            x = pd.DataFrame(s1["DATA"])
            # https://www.jianshu.com/p/d615699ff254
            # DataFrame 的处理逻辑是如果目标表存在，则删除重建，并且不能完整保留原表字段的数据类型
            x.to_sql(table_name, con=engine, if_exists='replace', index=False)
            printf("成功完成表%s的数据导入！" % table_name)
        except Exception as err:
            print(err)

    # 创建主键
    sql = "alter table `gtzz_base_db`.`t_pro_task` modify column `sn` varchar(128) character set utf8mb4 collate utf8mb4_general_ci not null first,add primary key (`sn`)"
    db.execute(sql)

    # 添加一列:subject_id
    sql = "alter table `gtzz_base_db`.`t_pro_task` add column `subject_id` int null default 0 comment '与t_gtzz_subject对应的主键' after `reiterate_task_sn`"
    db.execute(sql)

    # 添加组织机构码
    sql = "alter table `gtzz_base_db`.`t_pro_task` add column `org_code` varchar(36) null comment '组织机构码' after `subject_id`"
    db.execute(sql)

    # 更新组织机构码
    sql = "update t_pro_task as t1 ,t_base_organization as t2 set t1.org_code=t2.org_code where t1.send_user_name=t2.org_name"
    db.execute(sql)

    # 创建索引
    sql = "alter table `gtzz_base_db`.`t_pro_message` modify column `task_sn` varchar(128) character set utf8mb4 collate utf8mb4_bin null default null after `modify_time`"
    db.execute(sql)

    sql = "alter table `gtzz_base_db`.`t_pro_message` add index(`task_sn`)"
    db.execute(sql)
    # 写入同步日志
    if cnt == 11:
        message = '正常同步成功!'
    else:
        message = '财务系统未授权同步文件！'
    sql = "insert into t_gtzz_sync_log(file_count,message) values(%s,'%s')" % (cnt, message)
    db.execute(sql)

    # 回写subject_id,初始化时应该都是0，更新时才会回写
    sql = "update t_pro_task as t1 ,t_gtzz_subject as t2 set t1.subject_id=t2.subject_id where t1.sn=t2.task_sn"
    db.execute(sql)

    # 填充表t_gtzz_subject
    sql = "select t1.sn,(select itemvalue20 from t_pro_form_value where task_sn=t1.sn) as money,(select itemvalue11 from t_pro_form_value where task_sn=t1.sn) as xmjj ,t1.subject_id,t1.task_number,t1.task_title,t1.task_year,t1.ex_send_org_name,(select class_name from t_pro_detail  where task_sn=t1.sn order by class_name limit 1 ) as subject_type_name, t1.create_time from t_pro_task as t1    left join t_pro_form_value as t2 on t2.task_sn=t1.sn  where   t1.ex_send_org_name in (select org_name from t_gtzz_school)  and t1.use_flag=1 and t1.draft_flag=0 and t1.close_flag=1 and  t1.data_type=0  order by t1.create_time desc"
    l1 = db.findList(sql)
    # 遍历每一条数据
    c1 = 0
    c2 = 0
    for i in range(0, l1[0]):
        row = l1[1][i]
        # 项目代号
        subject_id = row['subject_id']
        # sn
        task_sn = row['sn']
        # 单位名称
        xmdw = row['ex_send_org_name']
        # 项目名称
        xmmc = row['task_title']
        # 申报年度
        sbnd = row['task_year']
        # 查询获取单位编号
        sql = "select org_code from t_base_organization where org_name='%s'" % xmdw
        l2 = db.find(sql)
        org_code = l2[1]["org_code"]
        # 立项时间
        lixiang_time = row["create_time"]
        # 项目编号
        task_number = row["task_number"]

        # 项目金额
        money = row["money"]

        # 项目简介
        xmjj = row["xmjj"]

        # 项目类型
        sql = "select class_name,parent_name from t_pro_detail  where task_sn='%s'" % task_sn
        l3 = db.findList(sql)
        # 声明空的两个
        l1_list = []
        l2_list = []

        for x in l3[1]:
            l1_list.append(x["parent_name"])
            l2_list.append(x["class_name"])
        # 去重
        l1_list = list({}.fromkeys(l1_list).keys())
        l2_list = list({}.fromkeys(l2_list).keys())
        # 大的项目类型名称
        if l1_list == None or len(l1_list) == 0:
            subject_type_name = '其他'
        else:
            subject_type_name = l1_list[0]

        # 如果存在subject_id,则update
        if subject_id > 0:
            finish_scan = 1
            sql = "update t_gtzz_subject set finish_scan=1,task_sn='%s',org_code='%s',xmdw='%s',xmmc='%s',sbnd='%s',lixiang_time='%s',subject_type_id='-1',subject_type_name='%s' where subject_id=%d" % \
                  (task_sn, org_code, xmdw, xmmc, sbnd, lixiang_time, subject_type_name,subject_id)
            db.execute(sql)
        else:
            # 如果不存在subject_id,则insert into
            finish_scan = 0
            sql = "insert into t_gtzz_subject(finish_scan,task_sn,org_code,xmdw,xmmc,sbnd,lixiang_time,subject_type_id,subject_type_name,money,xmjj,task_number) values(%s,'%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s')" % \
                  (finish_scan, task_sn, org_code, xmdw, xmmc, sbnd, lixiang_time, '-1', subject_type_name, money, xmjj,
                   task_number)
            # 执行insert 语句
            db.execute(sql)
            # 保存子类型名称
            for x in l2_list:
                sql = "insert into t_gtzz_subject_sub_type(task_sn,sub_type_name,year) values('%s','%s','%s')" % (
                    task_sn, x, sbnd)
                db.execute(sql)

        # 补充一下subject_type_id
        sql = 'update t_gtzz_subject as t1,t_gtzz_subject_type as t2  set t1.subject_type_id=t2.subject_type_id where t1.subject_type_name=t2.subject_type_name'
        db.execute(sql)
        # 更新t_gtzz_subject_sub_type表的subject_id
        sql = "update t_gtzz_subject_sub_type as t1,t_gtzz_subject as t2 set t1.subject_id=t2.subject_id where t1.task_sn=t2.task_sn"
        db.execute(sql)

        if finish_scan == 0:
            c1 = c1 + 1
            printf('成功插入%s条记录！' % c1)
        else:
            c2 = c2 + 1
            printf('成功更新%s条记录！' % c2)
    printf('恭喜，所有操作成功完成！')


if __name__ == '__main__':
    SyncData()