Python连接DB2操作的一个小例子

QQ交流群：127591054 作者：JackChiang 作者QQ：595696297

Python版本3.5

这个小项目，是解决数据库查看的时候一些错误的，查看。每天都会跑数据，但是第一天的数据和第二天的名字可以是一样的，我第二天还可以在跑一遍。

涉及到的知识：
1、日期获取，例如获取2012-02-21~2013-08-21之间所有的日期列表，需要一天一天查询遍历。
2、正则匹配名字，由于一个调度会生成4条数据，需要找到这4条数据的顺序必须固定才为正确否则就是不对的。
3、然后把查询出来有问题的数据写入到文件。

代码有详细的注解，有问题可以练习我QQ学习哦

# -*- coding: utf-8 -*-
import time
import sys,os #要重新载入sys。因为 Python 初始化后会删除 sys.setdefaultencoding 这个方 法
import ibm_db
import re
import datetime
from datetime import *

#1、首先获取db2数据库连接，进行连接
#2、读取用户要操作的日期，可以输入日期范围和固定日期
    #2.1、如果输入指定范围日期格式为2018-09-12~2019-09-21
        #如果用户输入的是范围，这时候需要一天一天的进行处理，方法和2.2一样
    #2.2、如果输入指定日期为：2019-09-02
#3、可以查询当前日志所有时间范围数据检查


#获取当前脚本所在路径
def cur_file_dir():
    path = sys.path[0] #获取脚本路径
    if os.path.isdir(path):#判断脚本是文件还是编译后的文件，如果是脚本返回脚本目录，如果是编译文件，返回编译文件路径
        return path
    elif os.path.isfile(path):
        return os.path.dirname(path)

#添加配置文件
def OneFile():
    dic = {}
    str = cur_file_dir()
    file_input = open(str+'\\DB2_Protory.txt','r')
    list_file_line = file_input.readlines()
    for list1 in list_file_line:
        print(list1.strip('\n'))
        list1 = list1.strip('\n')
        key = (list1.split('='))[0].upper()
        value = (list1.split('='))[1]
        dic[key]=value
        file_input.flush()
    file_input.close()
    return dic

#首先获取db2数据库连接，进行连接
def ConDB(database,hostname,port,protocol,uid,pwd):
    dburl = "DATABASE=%s;HOSTNAME=%s;PORT=%s;PROTOCOL=%s;UID=%s;PWD=%s;"%(database,hostname,port,protocol,uid,pwd)
    print(dburl)
    conn = ibm_db.connect(dburl, "", "")
    if conn:
        sql = "SELECT service_level, fixpack_num FROM TABLE(sysproc.env_get_inst_info())as INSTANCEINFO"
        stmt = ibm_db.exec_immediate(conn,sql)
        result = ibm_db.fetch_both(stmt)

        print('----------------------连接成功---------------------')
        print('---地址：%s 数据库：%s 用户：%s'%(hostname,database,uid))
        print('DB2  version: ',result[0])

        return conn
    else:
        return

#用来关闭数据库连接
def CloseClo(conn):
    ibm_db.close(conn)
#日期处理
def CurDate(str):
    #首先判断当前日期是连续的还是单一天数的。
    str_Date = re.split(r'~',str)
    #判断用户输入日期是不是有效日期
    flog = 1
    try:
        for i in range(len(str_Date)):
            datetime.strptime(str_Date[i],"%Y-%m-%d")
    except:
        flog = 0
    # 代表有两个日期，或者就是只有一个日期
    if len(str_Date) > 1 and flog == 1:
        #判断后面输入的日期是否大于前面的日期，验证用户输入数据是否正确
        if datetime.strptime(str_Date[0],"%Y-%m-%d").date() > datetime.strptime(str_Date[1],"%Y-%m-%d").date():
            #插入0代表数据错误
            str_Date.insert(0,0)
            return str_Date
        else:
            #否则插入2代表有两个日期
            str_Date.insert(0,2)
        return str_Date
    elif len(str_Date) == 1 and flog == 1:
        str_Date.insert(0,1)
    else:
        str_Date.insert(0,0)
    return str_Date

#获取两个日期中间的日期列表
def gen_dates(b_date, days):
    day = timedelta(days=1)
    for i in range(days):
        yield b_date + day*i


def get_date_list(user_say_date):
    """
    获取日期列表
    :param start: 开始日期
    :param end: 结束日期
    :return:
    """
    data = []
    if user_say_date[0] == 1:
        date.append(user_say_date[1])
        return data
    elif user_say_date[0] == 2:
        start = datetime.strptime(user_say_date[1],"%Y-%m-%d").date()
        end = datetime.strptime(user_say_date[2], "%Y-%m-%d").date()
        for d in gen_dates(start, (end-start).days):
            data.append(d)
        #最后把日期加入
        date.append(user_say_date[2])
        return data
    else:
        data = []
        return data





#查询根据时间所判断有问题的数据
def SelSQl(success_date,conn):

    #开始处理
    #拼接处理SQL
    SQL= "这里写的是操作的代码"%success_date


    stmt = ibm_db.exec_immediate(conn, SQL)
    #一行一行的读取
    result = ibm_db.fetch_both(stmt)
    # 设置默认值
    flog = 0

    a = ['', '', '', '']
    seq_id = [0, 0, 0, 0]
    read_error = 0 #记录出错次数，4个为一次
    sum_date = 0 #记录总数

    #打开文件准备记录错误信息
    str = cur_file_dir()
    f1 = open(str+"\\ErrorSQL.SQL",'a')

    while (result):
        # 四条数据做一次判断
        sum_date = sum_date + 1
        if flog == 4:
            #列表数据达到四条，开始判断顺序
            a0 = r"^LD_.*?_INIT$" #第一行匹配格式
            a1 = r"^AP_.*?_INIT$"  # 第二行匹配格式
            a2 = r"^LD_.*?[^I]?[^N]?[^I]?[^T]{1}$"  # 第三行匹配格式
            a3 = r"^AP_.*?[^I]?[^N]?[^I]?[^T]{1}$"  # 第四行匹配格式
            print(a)
            if re.match(a0, a[0]) != None and re.match(a1, a[1]) != None and re.match(a2, a[2]) != None and re.match(a3,a[3]) != None:
                pass
            else:
                # 如果其中有一个匹配不正确，代表数据顺序有问题。
                # 所以要把错误信息放入到列表
                read_error = read_error + 1
                if read_error == 1:
                    print('---------------------------日期：%s----------------------------'%success_date)
                    f1.write('---------------------------日期：%s----------------------------'%success_date)
                    f1.write('\n')
                print('发现数据出现问题！')
                # 把出错信息写入到文件。
                ErrorSQL = "SELECT * FROM ETL.JOB_LOG A WHERE JOB_SEQ_ID IN(%d,%d,%d,%d) UNION ALL" % (
                seq_id[0], seq_id[1], seq_id[2], seq_id[3])
                f1.write(ErrorSQL)
                f1.write('\n')

            # 把每条时间放入指定列表位置，四条数据刷新一次
            flog = 0
            a = ['', '', '', '']
            seq_id = [0, 0, 0, 0]
        a[flog] = result[2]
        seq_id[flog] = result[0]
        flog = flog + 1
        result = ibm_db.fetch_both(stmt)
    #写完错误关掉文件
    if read_error > 0 :
        f1.write('总行数为：%s ，出错的信息有： %s 次'%(sum_date/4,read_error))
        f1.write('\n')
    f1.close()
    print('总行数为：%s ，出错的信息有： %s 次'%(sum_date/4,read_error))

#Main
dice = OneFile()
conn = None
try:
    #conn = ConDB(dice['DATABASE'],dice['HOSTNAME'],dice['PORT'],dice['PROTOCOL'],dice['UID'],dice['PWD'])
    user_say = input("请输入指定范围日期格式例如为2018-09-12~2019-09-21，或者输入指定一天日期例如：2019-09-21")
    user_say_date = CurDate(user_say)
    print(user_say_date)
    data = get_date_list(user_say_date)

    print(data)
    #获取到一个日期列表然后循环进行查询
    #for current_data in data:
    #   SelSQl(current_data,conn)
    CloseClo(conn)
except Exception as ex:
    CloseClo(conn)
    print(ex)
finally:
    CloseClo(conn)