# -*- coding: utf-8 -*-
#
"""
multiprocessing
"""
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
from multiprocessing import Pool


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('%s [INFO] I am child process (%s) and my parent is %s.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), os.getpid(), os.getppid()))
    pidremark = "%s-%s" %(os.getppid(), os.getpid())
    # return
    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)

    #exec delete if contain pk
    if table in pktables_lst and not table_name.startswith('X'):
        stmt_delete = "delete from %s.%s " % (target_schema, table_name)
        print('%s [INFO] delete sql: %s ' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), stmt_delete))

    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, pidremark, target_schema, table_name)

    cursor.execute(stmt_merge_log)
    cnx.commit()

    if rows_insert == 0:
        return
    elif rows_insert > 50000:
        time.sleep(random.randint(2, 5))
    else:
        time.sleep(random.randint(1, 2))

    #wait delete run
    if rows_delete > 10000:
        time.sleep(int(rows_delete / 10000))
    elif rows_delete > 100:
        time.sleep(random.randint(1, 2))

    target_rows = 0
    repl_status = 'replicate_timeout'

    ##check Oracle Replicate speed
    for loop_i in range(300 + 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(2)
                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(2)

    # 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)

    stmt_hislog = """insert into test.test_parallel_his 
                     select t.*,now() 
                     from test.test_parallel t
                     where table_schema='%s' and table_name='%s'
                  """ % (target_schema, table_name)
    cursor.execute(stmt_hislog)

    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("当前子线程:%s, 当前输入参数,wrokerid:%s worker_count:%s " % (threading.current_thread().name, wrokerid, worker_count))
    print('%s [INFO] I am child process (%s) and my parent is %s.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), os.getpid(), os.getppid()))
    pidremark = "%s-%s" % (os.getppid(), os.getpid())

    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()), pidremark))
            break

        if i%worker_count == wrokerid:
            print("%s [INFO] %s:%s:%s" %(time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), pidremark, 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))
    print('%s [INFO] I am child process (%s) and my parent is %s.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), os.getpid(), os.getppid()))

    global tables_lst
    global pktables_lst
    tables_lst = []
    pktables_lst = []
    #stmt_query = "select concat(table_schema,'.',table_name) from test.test_parallel t where repl_status='waiting' order by table_name limit 20000"
    stmt_query = "select concat(table_schema,'.',table_name),ifnull(contain_pk,''),source_count,ifnull(contain_uk,''),ifnull(contain_datatypes,'') from test.test_parallel t where repl_status='waiting' order by table_name limit 20000"

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

    for row in cursor:
        tables_lst.append(row[0])
        if row[2] < 100000 and re.search('heronsdb', row[0]):
            pktables_lst.append(row[0])
        elif len(row[1]) > 3 and row[2] < 10000 and not re.search('text|blob|bit|binary|float|double', row[1]):
            pktables_lst.append(row[0])
        elif len(row[3]) > 3 and row[2] < 10000 and not re.search('text|blob|bit|binary|float|double|enum', row[4]):
            pktables_lst.append(row[0])

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

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))
    print('%s [INFO] I am child process (%s) and my parent is %s.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), os.getpid(), os.getppid()))
    # return
    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)

        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()
        cursor_o.close()
        cnx_o.close()

        #
        print("%s [INFO] start drainer background " % (
            time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))
        os.system('cd /data2/autotest/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('%s [INFO] I am child process (%s) and my parent is %s.' % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), os.getpid(), os.getppid()))
    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()

    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)

    p_num = 5
    p = Pool(p_num)
    for i in range(p_num):
        p.apply_async(f_run_insert_ctl, args=(i, p_num))

    print('%s [INFO] Waiting for all subprocesses done...'% (
        time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))
    p.close()
    p.join()
    print('%s [INFO]All subprocesses done.'% (
        time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime())))


    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)

    #analyze result
    from analyze_replicate_result import f_get_replicate_data
    input_batchid = "'%s'"%(batchid)
    f_get_replicate_data(db_conn_str, input_batchid)

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