# -*- coding: utf-8 -*-
#
import random
import string
import os
import sys
import re
import subprocess
import time
import datetime
from configparser import ConfigParser
import ast

import cx_Oracle
import mysql.connector
import threading

count = 0
def f_run_insert_tab(table=""):
    print("%s [INFO] function %s start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), f_run_insert_tab.__name__))
    print("当前子线程:%s, 当前输入参数,table:%s " % (threading.current_thread().name, table))
    # print("当前子线程:{}".format(threading.current_thread().name))
    table_schema = table.split('.')[0]
    table_name = table.split('.')[1]

    insert_starttime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    # time.sleep(random.randint(1,10))

    source_schema = "%s_raw" %(table_schema)
    target_schema = table_schema
    stmt_select = "select count(1) from  %s.%s " % (target_schema, table_name)
    stmt_delete = "truncate table %s.%s " % (target_schema, table_name)
    stmt_insert = "insert into %s.%s select * from %s.%s" % (target_schema, table_name, source_schema, table_name)

    cnx_o = cx_Oracle.connect(user="",
                              password="",
                              dsn="127.0.0.1:1521/oratidb")
    cursor_o = cnx_o.cursor()

    cnx = mysql.connector.connect(**db_conn_dict)
    cursor = cnx.cursor()

    rows_delete = 0
    rows_insert = 0
    run_failed = 0

    try:
        cursor.execute(stmt_delete)
        rows_delete = cursor.rowcount

        insert_starttime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        cursor.execute(stmt_insert)
        rows_insert = cursor.rowcount

        print("%s [SUCC] execute insert finish, %s %s(delete) %s(insert) rows affect, the i table" % (
            time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), table_name, rows_delete, rows_insert))
        cnx.commit()
        repl_status = 'insert_succ'
    except Exception as e:
        print("%s [ERROR] insert failed,exit %s, Error:%s" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), stmt_insert, e))
        run_failed = run_failed + 1
        repl_status = 'insert_fail'
        # exit()

    stmt_merge_log = """update test.test_parallel 
                        set batchid = '%s',
                            repl_status = '%s',
                            delete_count = '%s',
                            source_count = '%s',
                            target_count = 0 ,
                            insert_starttime = '%s',
                            insert_endtime = now() ,
                            repl_starttime = now() ,
                            repl_endtime = null ,
                            remark = '%s'
                        where table_schema='%s' and table_name='%s'
                     """ % (batchid, repl_status, rows_delete, rows_insert, insert_starttime, threading.current_thread().name, target_schema, table_name)

    cursor.execute(stmt_merge_log)
    cnx.commit()

    if rows_insert == 0:
        return
    elif rows_insert > 50000:
        time.sleep(3)
    else:
        time.sleep(1)

    target_rows = 0
    repl_status = 'replicate_timeout'

    ##check Oracle Replicate speed
    for loop_i in range(60 + int(rows_insert / 1000)):
        try:
            cursor_o.execute(stmt_select)
        except cx_Oracle.DatabaseError as e:
            if error.code == 8103:
                # select return "ORA-08103: object no longer exists" when truncate
                print('Database error: %s. retry...'.format(e))
                time.sleep(1)
                continue
            else:
                print('Database error: %s'.format(e))

        for row in cursor_o:
            target_rows = row[0]
        if target_rows == rows_insert:
            repl_status = 'replicate_succ'
            print("%s [SUCC] target insert finish, %s %s(replicate) rows affect" % (
                time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), table_name, target_rows))
            break
        time.sleep(1)

    # update replicate time
    stmt_merge_log = """update test.test_parallel 
                        set repl_status = '%s',
                            target_count = '%s',
                            repl_endtime = now() ,
                            repl_duration = TIMESTAMPDIFF(SECOND,repl_starttime, now())
                        where table_schema='%s' and table_name='%s'
                     """ % (repl_status, target_rows, target_schema, table_name)

    cursor.execute(stmt_merge_log)
    cnx.commit()

    cursor.close()
    cnx.close()

    cursor_o.close()
    cnx_o.close()

def f_run_insert_ctl(wrokerid=0, worker_count=10):
    print("%s [INFO] function %s start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), f_run_insert_ctl.__name__))
    print("输入参数,wrokerid:%s worker_count:%s" %(wrokerid, worker_count))
    print("当前子线程:{}".format(threading.current_thread().name))
    # f_check_drainer()

    # global cnx
    # global cursor
    # cnx = mysql.connector.connect(**db_conn_dict)
    # cursor = cnx.cursor()

    i = 0
    for tab in tables_lst:
        i = i + 1
        drainer_isok = f_check_drainer()
        if drainer_isok != 0:
            print("%s [ERRO] drainer process is down, exit %s " % (
                time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()),threading.current_thread().name))
            break

        if i%worker_count == wrokerid:
            print("%s, %s:%s:%s" %(time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()),threading.current_thread().name,i,tab))
            f_run_insert_tab(tab)
            time.sleep(0.1)
        else:
            continue
        drainer_isok = 1

    # cursor.close()
    # cnx.close()

def f_get_tables():
    print("%s [INFO] function %s start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), f_get_tables.__name__))
    print("当前子线程:{}".format(threading.current_thread().name))

    global tables_lst
    tables_lst = []
    stmt_query = "select concat(table_schema,'.',table_name) from test.test_parallel t where repl_status='waiting' order by table_name limit 1000"

    cnx = mysql.connector.connect(**db_conn_dict)
    cursor = cnx.cursor()
    cursor.execute(stmt_query)

    for row in cursor:
        tables_lst.append(row[0])

    print("find %s tables to insert data" % (len(tables_lst)))
    cursor.close()
    cnx.close()
    time.sleep(2)

def f_check_drainer():
    global drainer_isok
    ps_str = os.popen("ps -ef | grep dr_ora | grep -v grep | wc -l").read()
    if int(ps_str) == 1:
        drainer_isok = 0
    else:
        drainer_isok = 1
    # print("%s,%s,%s" %(int(ps_str), drainer_isok, type(drainer_isok)))
    return drainer_isok

def f_restart_drainer():
    print("%s [INFO] function %s start" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), f_restart_drainer.__name__))
    print("当前子线程:{}".format(threading.current_thread().name))

    ps_str = os.popen("ps -ef | grep dr_ora | grep -v grep | wc -l").read()
    if int(ps_str) == 0:
        print("%s [ERRO] drainer process is down, try restart " % (
            time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))
        repl_status = 'replicate_exit'

        # delete checkpoint
        cnx_o = cx_Oracle.connect(user="",
                                  password="",
                                  dsn="127.0.0.1:1521/oratidb")
        cursor_o = cnx_o.cursor()
        delete_checkpoint = "delete from findpt_lj.TIDB_BINLOG_CHECKPOINT"
        cursor_o.execute(delete_checkpoint)
        cnx_o.commit()
        cursor_o.close()
        cnx_o.close()

        print("%s [INFO] clear findpt_lj.TIDB_BINLOG_CHECKPOINT, %s rows affect " % (
            time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), cursor_o.rowcount))
        cnx_o.commit()
        #
        print("%s [INFO] start drainer background " % (
            time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))
        os.system('cd /data1/tmp/localdrainer/oracle/; nohup ./drainer -config dr_ora.toml &')
        # wait drainer start work
        time.sleep(300)
    else:
        ps_str = os.popen("ps -ef | grep dr_ | grep -v grep ").read()
        ps_lst = re.split(r"[ ]+", ps_str)
        print("%s [INFO] drainer is running from %s" % (
            time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), ps_lst[4]))

def main():
    print("主线程:{}".format(threading.current_thread().name))

    # read config file
    global config
    global db_conn_dict
    config = ConfigParser()
    config.read('repl_dbcfg.ini')
    db_conn_str = config.get('tidb', 'connstr')
    db_conn_dict = ast.literal_eval(db_conn_str)
    print('db_conn_dict:',type(db_conn_dict), db_conn_dict)
    # conn_str = db_conn_dict
    # print(type(conn_str), conn_str)

    #gloabl variables
    global batchid
    batchid = time.strftime("%Y%m%d.%H%M%S", time.localtime())

    f_get_tables()

    lock = threading.Lock()

    if len(tables_lst) == 0:
        print("%s [INFO] all table replicate tasks are finished, exit" % (
        time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))
        sys.exit(0)

    thread_1 = threading.Thread(target=f_run_insert_ctl, args=(1,10))
    thread_2 = threading.Thread(target=f_run_insert_ctl, args=(2,10))
    thread_3 = threading.Thread(target=f_run_insert_ctl, args=(3,10))
    thread_4 = threading.Thread(target=f_run_insert_ctl, args=(4,10))
    thread_5 = threading.Thread(target=f_run_insert_ctl, args=(5,10))
    thread_6 = threading.Thread(target=f_run_insert_ctl, args=(6,10))
    thread_7 = threading.Thread(target=f_run_insert_ctl, args=(7,10))
    thread_8 = threading.Thread(target=f_run_insert_ctl, args=(8,10))
    thread_9 = threading.Thread(target=f_run_insert_ctl, args=(9,10))
    thread_0 = threading.Thread(target=f_run_insert_ctl, args=(0,10))
    #
    thread_1.start()
    thread_2.start()
    thread_3.start()
    thread_4.start()
    thread_5.start()
    thread_6.start()
    thread_7.start()
    thread_8.start()
    thread_9.start()
    thread_0.start()
    thread_1.join()
    thread_2.join()
    thread_3.join()
    thread_4.join()
    thread_5.join()
    thread_6.join()
    thread_7.join()
    thread_8.join()
    thread_9.join()
    thread_0.join()

    print("%s [INFO] check drainer status in main" % (
        time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))
    drainer_isok = f_check_drainer()
    print("drainer_isok:",drainer_isok)
    if drainer_isok != 0:
        f_restart_drainer()
        time.sleep(300)

if __name__ == '__main__':
    for i in range(100):
        main()