dataCreating/bin/DBConfig.py

# -*- coding:utf-8 -*-
import faker_data

from pathlib import Path
from datetime import datetime


# 定义公共的部分
path = Path(__file__)
etc_dir = path.parent.parent / "etc"


# def get_now():
#     now = datetime.now()
#     #now_day = now.strftime("%Y-%m-%d")
#     now_hours = now.strftime("%Y%m%d%H")
#     return(now_hours)


def mysql_config():
    mysql_info_dict={}
    mysql_info_dict['driver_name'] =  "com.mysql.jdbc.Driver"
    mysql_info_dict['driver_jar'] =  "mysql-connector-java-8.0.29.jar"
    driver_jar_path = etc_dir / "driver" / mysql_info_dict['driver_jar']
    mysql_info_dict['driver_jar_path'] =  str(driver_jar_path)
    #这里要连接到具体的库
    mysql_info_dict['jdbc_url'] =  "jdbc:mysql://10.10.29.9:3306/test2"
    mysql_info_dict['db_user'] =  "root"
    mysql_info_dict['db_password'] =  "Unary@2023"

    #create table sql
    # now_hours = get_now()
    #TODO(MH):根据时间生成不同的表
    table_name = "mysql_test_table"
    create_sql_string = "CREATE TABLE IF NOT EXISTS `"
    create_sql_string += table_name
    create_sql_string += "` ("
    create_sql_string += "`uuid` varchar(50) DEFAULT NULL,"
    create_sql_string += "`id` bigint DEFAULT NULL,"
    create_sql_string += "`name` varchar(50) DEFAULT NULL,"
    create_sql_string += "`mobile` varchar(50) DEFAULT NULL,"
    create_sql_string += "`ssn` varchar(50) DEFAULT NULL,"
    create_sql_string += "`sex` int DEFAULT NULL,"
    create_sql_string += "`email` varchar(50) DEFAULT NULL,"
    create_sql_string += "`job` varchar(50) DEFAULT NULL,"
    create_sql_string += "`address` varchar(50) DEFAULT NULL,"
    create_sql_string += "`actime_time` timestamp NULL DEFAULT NULL"
    create_sql_string += ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;"
    mysql_info_dict['create_sql_string'] =  create_sql_string

    # insert into sql
    insert_list = []
    data = faker_data.faker_data(lines=2)[1:]
    for a in data:
       #拼接sql
       insert_sql_string = "INSERT INTO " + table_name
       insert_sql_string += "(uuid, id, name, mobile, ssn, sex, email, job, address, actime_time) VALUES(" 
       str_data = ','.join("'{0}'".format(x) for x in a)
       insert_sql_string += str_data + ");"

       insert_list.append(insert_sql_string)
    # 返回所有的插入语句
    mysql_info_dict['insert_list'] =  insert_list

    # delete sql
    del_sql = "delete FROM "+ table_name + " where 1=1 limit 100"
    mysql_info_dict['del_sql'] =  del_sql


    # uptata sql
    updata_sql = "UPDATE "+ table_name + " t SET t.job='测试' WHERE t.name like '李%' AND  sex = 1 AND  id = 1 "
    mysql_info_dict['updata_sql'] =  updata_sql

    return(mysql_info_dict)


def oracle_config():
    oracle_info_dict={}
    oracle_info_dict['driver_name'] =  "oracle.jdbc.OracleDriver"
    oracle_info_dict['driver_jar'] =  "ojdbc8-12.2.0.1.jar"
    driver_jar_path = etc_dir / "driver" / oracle_info_dict['driver_jar']
    oracle_info_dict['driver_jar_path'] =  str(driver_jar_path)
    #这里要连接到具体的实例
    oracle_info_dict['jdbc_url'] =  "jdbc:oracle:thin:@10.10.13.21:1521:a"
    oracle_info_dict['db_user'] =  "unary"
    oracle_info_dict['db_password'] =  "unary2023"
    
    #create table sql
    # now_hours = get_now()
    #TODO(MH):根据时间生成不同的表
    table_name = "ORACLE_TEST"
    create_sql_string = 'CREATE TABLE "'
    create_sql_string += table_name + '" '
    create_sql_string += '("UUID" VARCHAR2(50),'
    create_sql_string += '"ID" NUMBER(38,0), '
    create_sql_string += '"NAME" VARCHAR2(50),'
    create_sql_string += '"MOBILE" VARCHAR2(50),'
    create_sql_string += '"SSN" VARCHAR2(50),'
    create_sql_string += '"SEX" NUMBER(*,0), '
    create_sql_string += '"EMAIL" VARCHAR2(50), '
    create_sql_string += '"JOB" VARCHAR2(50),'
    create_sql_string += '"ADDRESS" VARCHAR2(100),'
    create_sql_string += '"ACTIME_TIME" VARCHAR2(50))'
    oracle_info_dict['create_sql_string'] =  create_sql_string


    # insert into sql
    insert_list = []
    data = faker_data.faker_data(lines=2)[1:]
    for a in data:
       #拼接sql
       insert_sql_string = "INSERT INTO " + table_name
       insert_sql_string += " t (uuid, id, name, mobile, ssn, sex, email, job, address, actime_time) VALUES("
       str_data = ','.join("'{0}'".format(x) for x in a)
       insert_sql_string += str_data + ")"

       insert_list.append(insert_sql_string)
    # 返回所有的插入语句
    oracle_info_dict['insert_list'] =  insert_list

    # delete sql
    del_sql = "delete FROM "+ table_name + " WHERE rownum<=100"
    oracle_info_dict['del_sql'] =  del_sql


    # uptata sql
    updata_sql = "UPDATE "+ table_name + " t SET t.job='测试' WHERE t.name like '李%' AND  sex = 1 AND  id = 1 "
    oracle_info_dict['updata_sql'] =  updata_sql

    return(oracle_info_dict)