from __future__ import print_function
import json
import pandas as pd
import sys,datetime,time
import threading
# import paramiko
import inspect
import requests


def div_list(ls,n):
    if not isinstance(ls,list) or not isinstance(n,int):
        return []
    ls_len = len(ls)
    if n<=0 or 0==ls_len:
        return []
    if n > ls_len:
        return []
    elif n == ls_len:
        return [[i] for i in ls]
    else:
        j = int(ls_len/n)
        k = int(ls_len%n)
        ### j,j,j,...(前面有n-1个j),j+k
        #步长j,次数n-1
        ls_return = []
        for i in range(0,(n-1)*j,j):
            ls_return.append(ls[i:i+j])
        #算上末尾的j+k
        ls_return.append(ls[(n-1)*j:])
        return ls_return


def get_message_threading(cnt,index,sql,sql_con,list_id = None):
    df_list = []
    df_result = []
    df = pd.DataFrame()
    class MyThread(threading.Thread):
        def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None):
            threading.Thread.__init__(self, group, target, name, args, kwargs, daemon=daemon)
            self._return = None

        def run(self):
            if self._target is not None:
                self._return = self._target(*self._args, **self._kwargs)

        def get_return(self):
            try:
                return self._return
            except Exception:
                return None

        def join(self):
            threading.Thread.join(self)
            return self._return

    def case_even(n,values,sql,sql_con):

        x = values[0]
        y = values[1]
        z_id = values[2]
        if z_id :
            sql_feature_new = sql.format(str(tuple(z_id)).replace(',)',')'))
        else:
            sql_feature_new = sql+' limit {0},{1}'.format(x,y)
        df = pd.read_sql(sql_feature_new,con=sql_con)
        semlock.release()

        return df


    maxconnections = 20 #最大并发数
    semlock = threading.BoundedSemaphore(maxconnections)

    list_Thread=[]
    index_y = 0
    if index > cnt :
        cnt = index
    num = cnt/index
    if int(str(num).split('.')[1]) > 0:
        num = int(str(num).split('.')[0]) + 1
    else:
        num = int(str(num).split('.')[0])

    run_values=[]
    list_id_split = div_list(list_id,num)
    for i in range(num): #执行中的总数
        if cnt - index <= 0:
            index = cnt
        if list_id:
            run_values.append((index_y,index,list_id_split[i]))
        else:
            run_values.append((index_y,index,None))
        index_y = index_y + index

    for i in range(num):

        semlock.acquire()
        t = MyThread( target = case_even, args=(i,run_values[i],sql,sql_con,))
        list_Thread.append(t)
        t.start()

    for i in range(num):
        list_Thread[i].join()
        df_list.append(list_Thread[i].get_return())

    df = pd.concat(df_list)
    return df


def get_message_id(start,cnt,index,sql,sql_con):
    """
        cnt : 查询总数
        index : 步长
        sql:sql语句
        sql_con:sql连接
    """
    start = start - cnt
    df = pd.DataFrame()
    df_list = []
    index_y = 0

    if index > cnt :
        cnt = index
    num = cnt/index
    if int(str(num).split('.')[1]) > 0:

        num = int(str(num).split('.')[0]) + 1
    else:
        num = int(str(num).split('.')[0])

    for i in range(1,num+1):

        if cnt - index <= 0:
            index = cnt

        sql_feature_new = sql.format(start+index)+' limit 0,{1}'.format(index_y,index)
        start = start + index
        index_y = index_y + index
        df_temp = pd.read_sql(sql_feature_new,con=sql_con)
        df_list.append(df_temp)
        cnt = cnt - index
    df = pd.concat(df_list,axis = 0)
    return df

def get_message(cnt,index,sql,sql_con):
    """
        cnt : 查询总数
        index : 步长
        sql:sql语句
        sql_con:sql连接
    """

    df = pd.DataFrame()
    df_list = []
    index_y = 0

    if index > cnt :
        cnt = index
    num = cnt/index
    if int(str(num).split('.')[1]) > 0:

        num = int(str(num).split('.')[0]) + 1
    else:
        num = int(str(num).split('.')[0])

    for i in range(1,num+1):

        if cnt - index <= 0:
            index = cnt

        sql_feature_new = sql+' limit {0},{1}'.format(index_y,index)
        index_y = index_y + index
        #         print(sql_feature_new)
        df_temp = pd.read_sql(sql_feature_new,con=sql_con)
        df_list.append(df_temp)
        cnt = cnt - index
    df = pd.concat(df_list,axis = 0)
    return df

def parse_json(x):
    feature = json.loads(x)
    feature.update(feature['data'])
    del feature['data']
    return feature

def dict_generator(indict, pre=None):
    pre = pre[:] if pre else []
    if isinstance(indict, dict):
        for key, value in indict.items():
            if isinstance(value, dict):
                if len(value) == 0:
                    yield pre+[key, '{}']
                else:
                    for d in dict_generator(value, pre + [key]):
                        yield d
            elif isinstance(value, list):
                if len(value) == 0:
                    yield pre+[key, '[]']
                else:
                    for v in value:
                        for d in dict_generator(v, pre + [key]):
                            yield d
            elif isinstance(value, tuple):
                if len(value) == 0:
                    yield pre+[key, '()']
                else:
                    for v in value:
                        for d in dict_generator(v, pre + [key]):
                            yield d
            else:
                yield pre + [key, value]
    else:
        yield indict



def parse_json_new(x,join = '.'):
    newJson = {}
    if isinstance(x,dict):
        x = x
    elif isinstance(x,str):
        try:
            x = json.loads(x)
        except:
            print()
    for i in dict_generator(x):
        newJson[str(join).join(i[0:-1])] =  i[-1]
    return newJson



def parse_json_tongdun(x,join = '.'):
    newJson = {}
    x = x[0]
    if isinstance(x,dict):
        x = x
    elif isinstance(x,str):
        try:
            x = json.loads(x)
        except:
            print()
    for i in dict_generator_tongdun(x):
        # newJson_new[str(join).join(i[0:-1])].append( i[-1])
        key = str(join).join(i[0:-1])
        value = i[-1]
        if key in newJson.keys():
            if isinstance(newJson[key],list) == False:
                newJson[key] = [newJson[key]]
            newJson[key].append(value)
        else:
            newJson[key] = value
    return newJson
def dict_generator_tongdun(indict, pre=None):
    pre = pre[:] if pre else []
    if isinstance(indict, dict):
        for key, value in indict.items():
            if isinstance(value, dict):
                if len(value) == 0:
                    yield pre+[key, '{}']
                else:
                    for d in dict_generator_tongdun(value, pre + [key]):
                        yield d
            elif isinstance(value, list):
                if len(value) == 0:
                    yield pre+[key, '[]']
                else:
                    for v in value:
                        if isinstance(v,dict):
                            for d in dict_generator_tongdun(v, pre + [key]):
                                yield d
                        # if key == 'list':
                        # # elif isinstance(v,list):
                        #     yield pre+[key, value]
            elif isinstance(value, tuple):
                if len(value) == 0:
                    yield pre+[key, '()']
                else:
                    for v in value:
                        for d in dict_generator_tongdun(v, pre + [key]):
                            yield d
            else:
                yield pre + [key, value]
    else:
        yield indict



def del_diff_columns(old_columns,new_columns,df):
    """并修改名称,返回修改后的名称"""
    rename_columns = {}
    for i in range (len(old_columns)):
        rename_columns[old_columns[i]] = new_columns[i]
    diff_columns = list(set(old_columns).difference(set(df.columns.tolist())))
    for columns in diff_columns:
        del rename_columns[columns]
    return rename_columns

# def rename_Columns(name_old,name_new,rename=[]):
#     """ 支持部分columns_name修改
#         name_old : 读取的原始名称
#        name_new : 要替换的前缀
#        rename :  要替换哪些名称
#         """
#     try:
#         name_old = name_old.split('.')[1]
#     except:
#         name_old = name_old

#     if name_old in rename:
#         name_new = name_new+name_old
#         return name_new
#     return name_old

def getlen(x):
    x = str(x)
    if len(x) < 26:
        x = '0'
    return x

def bytesToStr(x):
    if x == None:
        x = ''
    else:
        try :
            x = str(x,encoding='utf-8')
        except:
            x = x
    return x


def str_connect(x,y):
    return str(x)+'_'+str(y)
from collections import defaultdict
reDict = defaultdict(int)  # 返回错误计数，用于出测试报告图表


def assert_diff(x,y,col_name,code=None):
    global reDict
    try:
        if isinstance(eval(x),list) and isinstance(eval(y),list):
            diff = list(set(a) ^ set(b))
            if diff:
                if code:
                    reDict['error'][code]+=1
                    return False
                else:
                    reDict[col_name]+=1
                    return col_name+' : '+str(x)+'_'+str(y) + '\n'
            else:
                if code:
                    reDict['error'][code]+=0
                return
    except:
        pass

    try:
        if abs(float(x) - float(y)) > 1/100000:
            if code:
                reDict[code]['error']+=1
                reDict[code]['sum']+=1
                return False
            else:
                reDict[col_name]+=1
                return col_name+' : '+str(x)+'_'+str(y) + '\n'
        else:
            if code:
                reDict[code]['pass']+=1
                reDict[code]['sum']+=1
            return
    except Exception as e:
        pass

    try:
        if str(x) != str(y):

            if code:
                reDict[code]['error']+=1
                reDict[code]['sum']+=1
                return False
            else:
                #                 print(col_name,'---',str(x) ,'    ----    ',str(y))
                reDict[col_name]+=1
                return col_name+' : '+str(x)+'_'+str(y) + '\n'
        else:
            if code:
                reDict[code]['pass']+=1
                reDict[code]['sum']+=1
            return
    except:
        pass

def assert_columns(df,columns,not_columns=[],is_AssertFilter=True,is_reDict = False,is_DelColumnsXY = True,code=None):
    """ 校验Dataframe columns 的 x 和 y 列
        not_columns : 不参与校验的列
        is_AssertFilter: 默认返回错误的数据，过滤正确值
        is_reDict: 默认不返回总结信息
        is_DelColumnsXY : 默认删除 columns 的 x和y
        code : 提取某列作为错误统计
    """
    df.fillna(False,inplace=True)
    df['测试结论'] = 'False'
    df['测试描述'] = ""
    global reDict
    reDict = defaultdict(int)

    if len(df)>0:
        if code:
            code = df.loc[df.duplicated('code',keep='first')==False]['code'].tolist()
            for i in code:
                reDict[i]={}
                reDict[i]['sum'] = 0
                reDict[i]['error'] = 0
                reDict[i]['pass'] = 0
        for cls in columns:
            if cls not in ['测试结论'] + not_columns :
                try:
                    if code:
                        _assert = df.apply(lambda x : assert_diff(x[cls+'_x'],x[cls+'_y'],cls,x['code']),axis=1)
                        df['测试描述']+=_assert.fillna('').astype(str)
                    else:
                        _assert = df.apply(lambda x : assert_diff(x[cls+'_x'],x[cls+'_y'],cls),axis=1)
                        df['测试描述'] += _assert.fillna('').astype(str)
                    df[cls] = df.apply(lambda x : str_connect(x[cls+'_x'],x[cls+'_y']),axis=1)
                    if is_DelColumnsXY:
                        del df[cls+'_x']
                        del df[cls+'_y']
                except:
                    pass
        if is_AssertFilter:
            df = df.loc[df['测试结论']!=True]
        df['测试描述'] = df['测试描述'].apply(lambda x : x.strip('\n').strip(' '))
        df.loc[df['测试描述'] == '','测试结论'] = True
        if is_reDict:
            return df,dict(reDict)
        else:return df


def matching_data(df,key):
    """
    feature 特征匹配规则
    1. 优先order_no  ->  loan_id  ->  uuid+created_at
    2. create_at 与 applied_at 相同，取同一时间
    3. create_at 与 applied_at 不相同，取一天内时间最近
    4. create_at 与 applied_at 相同并有重复，取特征 id 最大
    5. order_no 或者 loan_id 重复，按照 2，3，4 顺序进行匹配
    """
    df['seconds'] = abs(pd.to_datetime(df['created_at'],format='%Y-%mm-%dd %H:%M:%S')- pd.to_datetime(df['applied_at'],format='%Y-%mm-%dd %H:%M:%S')).dt.seconds
    df['days'] = abs(pd.to_datetime(df['created_at'],format='%Y-%mm-%dd %H:%M:%S')- pd.to_datetime(df['applied_at'],format='%Y-%mm-%dd %H:%M:%S')).dt.days
    # 第一轮筛选：一天之内并且的数据
    df = df.loc[(df['seconds']<= 86400) & (df['days']<= 0)]

    # 第二轮筛选：时间最近的数据
    df_g_time = df.groupby(key).agg({'seconds':'min'}).reset_index()
    df_time = pd.merge(df_g_time,df,how='left',on=key+['seconds'])
    #     print(df_time.loc[df_time['uuid'] == 'ed961be0-7fd2-4fca-b1d1-c4cc8e24a7bc'])

    #     print(df_time)
    # 第三轮筛选：有重复主键，筛选id最大的
    df_g_id = df_time.groupby(key).agg({'id':'max'}).reset_index()
    df_end = pd.merge(df_g_id,df_time,how='left',on=key+['id'])
    #     print(df_end.loc[df_end['uuid'] == 'ed961be0-7fd2-4fca-b1d1-c4cc8e24a7bc'])
    return df_end

def matching_data_new(df,key):
    """
    feature 特征匹配规则
    1. 优先order_no  ->  loan_id  ->  uuid+created_at
    2. create_at 与 applied_at 相同，取同一时间
    3. create_at 与 applied_at 不相同，取一天内时间最近
    4. create_at 与 applied_at 相同并有重复，取特征 id 最大
    5. order_no 或者 loan_id 重复，按照 2，3，4 顺序进行匹配
    """
    df['seconds'] = abs(pd.to_datetime(df['applied_at_x'],format='%Y-%mm-%dd %H:%M:%S')- pd.to_datetime(df['applied_at_y'],format='%Y-%mm-%dd %H:%M:%S')).dt.seconds
    df['days'] = abs(pd.to_datetime(df['applied_at_x'],format='%Y-%mm-%dd %H:%M:%S')- pd.to_datetime(df['applied_at_y'],format='%Y-%mm-%dd %H:%M:%S')).dt.days
    # 第一轮筛选：一天之内并且的数据
    df = df.loc[(df['seconds']<= 86400) & (df['days']<= 0)]

    # 第二轮筛选：时间最近的数据
    df_g_time = df.groupby(key).agg({'seconds':'min'}).reset_index()


    df_time = pd.merge(df_g_time,df,how='left',on=key+['seconds'])
    #     print(df_time.loc[df_time['uuid'] == 'ed961be0-7fd2-4fca-b1d1-c4cc8e24a7bc'])


    # 第三轮筛选：有重复主键，筛选id最大的
    df_g_id = df_time.groupby(key).agg({'id':'max'}).reset_index()
    df_end = pd.merge(df_g_id,df_time,how='left',on=key+['id'])
    #     print(df_end.loc[df_end['uuid'] == 'ed961be0-7fd2-4fca-b1d1-c4cc8e24a7bc'])
    return df_end

def ssh(sys_ip,username,password,cmds):
    try:
        #创建ssh客户端
        client = paramiko.SSHClient()
        #第一次ssh远程时会提示输入yes或者no
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        #密码方式远程连接
        client.connect(sys_ip, 22, username=username, password=password, timeout=20)
        #互信方式远程连接
        #key_file = paramiko.RSAKey.from_private_key_file("/root/.ssh/id_rsa")
        #ssh.connect(sys_ip, 22, username=username, pkey=key_file, timeout=20)
        #执行命令
        stdin, stdout, stderr = ssh.exec_command(cmds)
        #获取命令执行结果,返回的数据是一个list
        result = stdout.readlines()
        return result
    except Exception as e:
        print(e)
    finally:
        client.close()


def get_current_function_name():
    """获取方法名"""
    return inspect.stack()[1][3]


def return_LenValue(value,lens,types):
    if types == '大于等于':
        if len(str(value))>=lens:
            return value
        else:return ''
    elif types == "小于":
        if len(str(value))<lens:
            return value
        else:return ''

class MyThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None):
        threading.Thread.__init__(self, group, target, name, args, kwargs, daemon=daemon)
        self._return = None

    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args, **self._kwargs)

    def get_return(self):
        try:
            return self._return
        except Exception:
            return None

    def join(self):
        threading.Thread.join(self)
        return self._return

