import math
import threading
import pandas as pd
from sqlalchemy import and_, or_, func

from core.config.env import env
from libs.db_link import LinkMysql
from starlette.responses import StreamingResponse
from sqlalchemy.orm import Session

from libs.functions import wrapper_out
from models.recharge import Recharge, UserWC, GuildWC, FinanceFixLog

locka = threading.Lock()


# 写入文件
def data_to_file(data, name):
    bk = pd.DataFrame(data)
    with pd.ExcelWriter(f'static/{name}.xlsx') as writer:
        bk.to_excel(writer, sheet_name='Sheet1', index=False)
    file = open(writer, 'rb')
    return StreamingResponse(file, media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')


class RechargeStatement(object):
    """充值报表"""
    def __init__(self):
        self.derive_list = []

    def check_out(self, db, num):
        info_list = []
        locka.acquire()  # 防止序号错误
        once_res = db.query(Recharge).filter().offset(num * 10).limit(10).all()
        locka.release()
        for i in once_res:
            info_dict = i.to_dict()
            info_list.append(info_dict)
        info_list.reverse()
        self.derive_list += info_list

    def query_all_data(self, db):

        msg_count = db.query(func.count(Recharge.id)).scalar()
        num = math.ceil(msg_count / 10)
        # 创建线程
        ths = []
        # 创建线程
        for x in range(num):
            ths.append(threading.Thread(target=self.check_out, args=[db, x]))
        # 启动线程
        for y in range(num):
            ths[y].start()
        # 等待子进程结束
        for z in range(num):
            ths[z].join()
        return self.derive_list

    def get_statements(self, db: Session, param, sp=None):
        """列表"""
        not_null_filters = []
        if param.order_number:
            not_null_filters.append(Recharge.order_number == param.order_number)
        if param.user_id:
            not_null_filters.append(Recharge.user_id == param.user_id)
        if param.sid:
            not_null_filters.append(Recharge.sid.like(param.sid))
        if param.paychannel:
            not_null_filters.append(Recharge.pay_channel.like(param.pay_channel))
        if param.start_time:
            not_null_filters.append(Recharge.current >= param.start_time)
        if param.end_time:
            not_null_filters.append(Recharge.current <= param.end_time)
        # 判断有无条件
        try:
            if len(not_null_filters) > 0:
                get_user_orm_sql = db.query(Recharge).filter(and_(*not_null_filters))
                condition_data = db.execute(get_user_orm_sql).fetchall()
                serializer_info = [i[0].to_dict() for i in condition_data]
                serializer_info.reverse()
            else:
                serializer_info = self.query_all_data(db)
        except Exception as e:
            print(e)
            return [], 0, 0 if sp else []
        # 判断是列表还是导出接口
        if sp:
            if serializer_info:
                total = len(serializer_info)
                df = pd.DataFrame(serializer_info)
                count = df['money'].apply(lambda x: x).sum()
                return serializer_info[(int(param.page) - 1) * param.size:param.size * param.page], total, count
            return [], 0, 0
        else:
            return serializer_info


class WithdrawStatement(object):
    """提现报表"""
    def __init__(self):
        self.derive_user_list = []
        self.derive_guild_list = []

    def thread_to_data(self, db, num):
        user_list = []
        locka.acquire()
        once_res = db.query(UserWC).filter().offset(num * 10).limit(10).all()
        locka.release()
        for i in once_res:
            info_dict = i.to_dict()
            user_list.append(info_dict)
        self.derive_user_list += user_list

    def query_user_all_data(self, db):
        msg_count = db.query(func.count(UserWC.id)).scalar()
        num = math.ceil(msg_count / 10)
        # 创建线程
        ths = []
        for x in range(num):
            ths.append(threading.Thread(target=self.thread_to_data, args=[db, x]))
        for y in range(num):
            ths[y].start()
        for z in range(num):
            ths[z].join()
        return self.derive_user_list

    def get_user_withdraw_cash(self, db: Session, param):
        """用户提现"""
        is_filters = []
        if param.user_id:
            is_filters.append(UserWC.user_id == param.user_id)
        if param.status or param.status == 0:
            is_filters.append(UserWC.status == param.status)
        if param.start_time:
            is_filters.append(UserWC.current >= param.start_time)
        if param.end_time:
            is_filters.append(UserWC.current <= param.end_time)
        # 判断有无条件
        try:
            if len(is_filters) > 0:
                get_user_orm_sql = db.query(UserWC).filter(and_(*is_filters))
                condition_data = db.execute(get_user_orm_sql).fetchall()
                user_info = [i[0].to_dict() for i in condition_data]
            else:
                user_info = self.query_user_all_data(db)
        except Exception as e:
            print(e)
            return [], 0, 0, 0
        # 判断是列表还是导出接口
        user_info.reverse()
        if user_info:
            total = len(user_info)
            df = pd.DataFrame(user_info)
            count = df['money'].apply(lambda x: x).sum()
            final_count = df['final_money'].apply(lambda x: float(x)).sum()
            return user_info[(int(param.page) - 1) * param.size:param.size * param.page], total, count, final_count
        return [], 0, 0, 0

    def dispose_guild_to_data(self, db, num):
        user_list = []
        locka.acquire()
        once_res = db.query(GuildWC).filter().offset(num * 10).limit(10).all()
        locka.release()
        for i in once_res:
            info_dict = i.to_dict()
            user_list.append(info_dict)
        self.derive_user_list += user_list

    def query_guild_all_data(self, db):
        msg_count = db.query(func.count(UserWC.id)).scalar()
        num = math.ceil(msg_count / 10)
        # 创建线程
        ths = []
        for x in range(num):
            ths.append(threading.Thread(target=self.dispose_guild_to_data, args=[db, x]))
        for y in range(num):
            ths[y].start()
        for z in range(num):
            ths[z].join()
        return self.derive_user_list

    def get_guild_withdraw_cash(self, db: Session, cond):
        """公会提现"""
        is_guild_filters = []
        if cond.guild_id:
            is_guild_filters.append(GuildWC.guild_id == cond.guild_id)
        if cond.status or cond.status == 0:
            is_guild_filters.append(GuildWC.status == cond.status)
        if cond.start_time:
            is_guild_filters.append(GuildWC.update_time >= cond.start_time)
        if cond.end_time:
            is_guild_filters.append(GuildWC.update_time <= cond.end_time)
        # 判断有无条件
        try:
            if len(is_guild_filters) > 0:
                get_user_orm_sql = db.query(GuildWC).filter(and_(*is_guild_filters))
                condition_data = db.execute(get_user_orm_sql).fetchall()
                guild_info = [i[0].to_dict() for i in condition_data]
            else:
                guild_info = self.query_guild_all_data(db)
        except Exception as e:
            print(e)
            return [], 0, 0, 0
        # 判断是列表还是导出接口
        guild_info.reverse()
        if guild_info:
            total = len(guild_info)
            df = pd.DataFrame(guild_info)
            count = df['money'].apply(lambda x: x).sum()
            final_count = df['final_money'].apply(lambda x: float(x)).sum()
            return guild_info[(int(cond.page) - 1) * cond.size:cond.size * cond.page], total, count, final_count
        return [], 0, 0, 0


class FinanceFix(object):

    @staticmethod
    @wrapper_out()
    def get_finance_fix_data(db: Session, data):
        """财务修复"""
        total = db.query(func.count(FinanceFixLog.id)).scalar()
        output = db.query(FinanceFixLog).limit(data.page * data.size).offset((data.page - 1) * data.size).all()
        return [i.to_dict() for i in output], total
