Commit 9d877115 authored by 郑磊's avatar 郑磊

更新到最新代码

parent f6b4831e
*-cache
*.bak
**/__pycache__
*.pyc
/logs/
/test/
*.~*
*-workspace
*.db
fs-env/
.idea/
static/
bin/
\ No newline at end of file
/static/*
!/static/DejaVuSans-BoldOblique.ttf
/core/config/*.yaml
import json
import math
import threading
import time
from concurrent.futures.thread import ThreadPoolExecutor
import pandas as pd
from sqlalchemy.orm import Session
from core.config.env import env
from core.config.env import env, red
from libs import dlc
from libs.business import TYPE_NAME, GUILD_NAME
from libs.db_link import LinkMysql
from libs.functions import get_now_timestamp, time_str_to_timestamp, \
......@@ -169,8 +172,8 @@ def create_account(param):
# account_uuid = result['data']['result']['data']['uuid']
if result['status']:
account_uuid = result['data']['uuid']
sql = f"insert into fi_account(name, unique_tag, config_key, description, uuid, income, output, create_time) " \
f"values('{param.name}', '{param.unique_tag}', '{param.config_key}', '{param.remark}', '{account_uuid}', '{income}', '{output}', {get_now_timestamp()});"
sql = f"insert into fi_account(name, unique_tag, beneficiary, config_key, description, uuid, income, output, create_time) " \
f"values('{param.name}', '{param.unique_tag}', '{param.beneficiary}','{param.config_key}', '{param.remark}', '{account_uuid}', '{income}', '{output}', {get_now_timestamp()});"
LinkMysql(env.DB_3YV2).perform_mysql(sql)
except Exception as e:
Logger(40).logger.error(f"创建新账户失败,原因:{str(e)}")
......@@ -186,7 +189,7 @@ def update_account_info(old_data):
output = ','.join(map(str, old_data.output))
try:
sql = f"update fi_account set name='{old_data.name}',income='{income}', output='{output}', " \
f"description='{old_data.remark}' where id = {old_data.id}"
f"description='{old_data.remark}',beneficiary='{old_data.beneficiary}' where id = {old_data.id}"
LinkMysql(env.DB_3YV2).perform_mysql(sql)
except Exception as e:
Logger(40).logger.error(f"修改账户sql:{sql},原因:{str(e)}")
......@@ -199,6 +202,32 @@ def get_finance_group_by(date, condition):
return result
def special_data_handle(old_data, special_data):
"""处理可消费,可提现数据"""
if not special_data:
return old_data
bk = pd.DataFrame(special_data)
for old in old_data:
df = bk[(bk["create_time"] == old['create_time']) & (bk["amount_type"] == 2)] # 可消费
serializer_info = df.to_dict(orient='records')
if serializer_info:
old['con_income'] = serializer_info[0]['income']
old['con_outcome'] = serializer_info[0]['outcome']
else:
old['con_income'] = 0
old['con_outcome'] = 0
df = bk[(bk["create_time"] == old['create_time']) & (bk["amount_type"] == 3)] # 可提现
serializer_info = df.to_dict(orient='records')
if serializer_info:
old['with_income'] = serializer_info[0]['income']
old['with_outcome'] = serializer_info[0]['outcome']
else:
old['with_income'] = 0
old['with_outcome'] = 0
return old_data
def get_finance_info(unique_tag, id, page, size, start_time, end_time, is_list=None):
"""账户财务信息"""
finance_condition = []
......@@ -209,56 +238,51 @@ def get_finance_info(unique_tag, id, page, size, start_time, end_time, is_list=N
finance_condition.append(f"create_time < {time_str_to_timestamp(end_time + ' 23:59:59')} ")
if unique_tag in ['guild_account', 'user_account', 'knapsack_account', 'pledgeDeduction', 'anchor_account']:
if unique_tag == 'guild_account':
if finance_condition:
finance_condition.append('amount_type=1')
count_sql = f"select create_time from v3_guild_account_statistics_copy where {' and '.join(finance_condition)} GROUP BY create_time"
data_sql = f"select id,initial_money as balance,income,outcome,create_time from v3_guild_account_statistics_copy where {' and '.join(finance_condition)} GROUP BY create_time order by create_time DESC limit {(int(page) - 1) * size},{size}"
else:
count_sql = f"select create_time from v3_guild_account_statistics_copy GROUP BY create_time"
data_sql = f"select id,initial_money as balance,income,outcome,create_time from v3_guild_account_statistics_copy GROUP BY create_time order by create_time DESC limit {(int(page) - 1) * size},{size}"
data_sql = f"select id,initial_money as balance,income,outcome,create_time from v3_guild_account_statistics_copy where {' and '.join(finance_condition)} GROUP BY create_time order by create_time limit {(int(page) - 1) * size},{size}"
other_sql = f"select initial_money as balance,income,outcome,create_time,amount_type from v3_guild_account_statistics_copy where amount_type != 1"
if unique_tag == 'anchor_account':
if finance_condition:
count_sql = f"select date as create_time from v3_user_account_statistics where {' and '.join(finance_condition)} GROUP BY date"
data_sql = f"select id,initial_money as balance,income,outcome,date as create_time from v3_user_account_statistics where {' and '.join(finance_condition)} GROUP BY date order by date limit {(int(page) - 1) * size},{size}"
conditions = [i.replace('create_time', 'date') for i in finance_condition]
count_sql = f"select date as create_time from v3_user_account_statistics where {' and '.join(conditions)} GROUP BY date"
data_sql = f"select id,initial_money as balance,income,pay as outcome,date as create_time from v3_user_account_statistics where {' and '.join(conditions)} GROUP BY date order by date limit {(int(page) - 1) * size},{size}"
else:
count_sql = f"select date as create_time from v3_user_account_statistics GROUP BY date"
data_sql = f"select id,initial_money as balance,income,pay as outcome,date as create_time from v3_user_account_statistics GROUP BY date order by date limit {(int(page) - 1) * size},{size}"
if unique_tag == 'user_account':
if finance_condition:
finance_condition.append('type=1')
condition = [i.replace('create_time', 'calculation_time') for i in finance_condition]
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where type=1 and {' and '.join(condition)} GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=1 and {' and '.join(condition)} GROUP BY calculation_time order by calculation_time DESC limit {(int(page) - 1) * size},{size}"
else:
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where type=1 GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=1 GROUP BY calculation_time order by calculation_time DESC limit {(int(page) - 1) * size},{size}"
other_sql = f"select initial_money as balance,income,outcome,calculation_time as create_time,amount_type from finance_data_calculation_sum_copy where {' and '.join(condition)} and amount_type != 1"
condition.append('amount_type=1')
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where {' and '.join(condition)} GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where {' and '.join(condition)} GROUP BY calculation_time order by calculation_time limit {(int(page) - 1) * size},{size}"
if unique_tag == 'knapsack_account':
if finance_condition:
finance_condition.append('amount_type=1')
condition = [i.replace('create_time', 'calculation_time') for i in finance_condition]
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where type=4 and {' and '.join(condition)} GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=4 and {' and '.join(condition)} GROUP BY calculation_time order by calculation_time DESC limit {(int(page) - 1) * size},{size}"
else:
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where type=4 GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=4 GROUP BY calculation_time order by calculation_time DESC limit {(int(page) - 1) * size},{size}"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=4 and {' and '.join(condition)} GROUP BY calculation_time order by calculation_time limit {(int(page) - 1) * size},{size}"
if unique_tag == 'pledgeDeduction':
if finance_condition:
finance_condition.append('amount_type=1')
condition = [i.replace('create_time', 'calculation_time') for i in finance_condition]
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where type=5 and {' and '.join(condition)} GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=5 and {' and '.join(condition)} GROUP BY calculation_time order by calculation_time DESC limit {(int(page) - 1) * size},{size}"
else:
count_sql = f"select calculation_time from finance_data_calculation_sum_copy where type=5 GROUP BY calculation_time"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=5 GROUP BY calculation_time order by calculation_time DESC limit {(int(page) - 1) * size},{size}"
data_sql = f"select id,initial_money as balance,income,outcome,calculation_time as create_time from finance_data_calculation_sum_copy where type=5 and {' and '.join(condition)} GROUP BY calculation_time order by calculation_time limit {(int(page) - 1) * size},{size}"
with ThreadPoolExecutor(max_workers=2) as pool:
future1 = pool.submit(LinkMysql(env.DB_3YV2).query_mysql, count_sql)
future2 = pool.submit(LinkMysql(env.DB_3YV2).query_mysql, data_sql)
count = future1.result()
res = future2.result()
else:
if finance_condition:
sys_sql = f"select initial_money as balance,income,outcome,create_time from v2_system_account_statistics_copy where fi_account_id={id} and {' and '.join(finance_condition)} ORDER BY create_time DESC"
else:
sys_sql = f"select initial_money as balance,income,outcome,create_time from v2_system_account_statistics_copy where fi_account_id={id} ORDER BY create_time DESC"
finance_condition.append('amount_type=1')
sys_sql = f"select initial_money as balance,income,outcome,create_time from v2_system_account_statistics_copy where fi_account_id={id} and {' and '.join(finance_condition)} ORDER BY create_time "
result = LinkMysql(env.DB_3YV2).query_mysql(sys_sql)
res = result[int(page - 1) * size: page * size]
count = result
if unique_tag in ['guild_account', 'user_account']:
other = LinkMysql(env.DB_3YV2).query_mysql(other_sql)
res = special_data_handle(res, other)
# 判断是列表还是导出接口
if is_list:
return res, len(count)
......@@ -356,7 +380,7 @@ def query_account_data():
class AccountStatistics(object):
"""账户列表,查询"""
def __init__(self, page, size, uuid, user_id, start_time, end_time, type, gift_type, unique):
def __init__(self, page, size, uuid, user_id, start_time, end_time, type, gift_type, unique, amount_type):
self.page = page
self.size = size
self.uuid = uuid
......@@ -366,6 +390,7 @@ class AccountStatistics(object):
self.type = type
self.gift_type = gift_type
self.unique = unique
self.amount_type = amount_type
self.guild_dict = {}
self.user_list = []
......@@ -375,18 +400,49 @@ class AccountStatistics(object):
if self.user_list:
user_uuid = self.user_list[0]
condition.append(f" uuid='{user_uuid.get('uuid')}'")
u_sql = f"select uuid,type,SUM(cast(amount as decimal(20,6)))/1000 as amount,FROM_UNIXTIME(create_time,'%Y%c%d') as create_time from {date} WHERE {' and '.join(condition)} GROUP BY uuid,type,FROM_UNIXTIME(create_time,'%Y%c%d')"
u_sql = f"select uuid,type,amount_type,SUM(cast(amount as decimal(20,6)))/1000 as amount,FROM_UNIXTIME(create_time,'%Y%c%d') as create_time from {date} WHERE {' and '.join(condition)} GROUP BY uuid,type,FROM_UNIXTIME(create_time,'%Y%c%d')"
else:
if self.uuid:
condition.append(f" uuid='{self.uuid}'")
if condition:
u_sql = f"select id,uuid,order_number,type,reference_type,amount/1000 as amount,reference_number,create_time from {date} WHERE {' and '.join(condition)} ORDER BY create_time DESC"
u_sql = f"select id,uuid,order_number,amount_type,type,reference_type,amount/1000 as amount,reference_number,create_time from {date} WHERE {' and '.join(condition)} ORDER BY create_time DESC"
else:
u_sql = f"select id,uuid,order_number,type,reference_type,amount/1000 as amount,reference_number,create_time from {date} ORDER BY create_time DESC"
u_sql = f"select id,uuid,order_number,amount_type,type,reference_type,amount/1000 as amount,reference_number,create_time from {date} ORDER BY create_time DESC"
Logger().logger.info(f"查询sql:{u_sql}")
result = LinkMysql(env.DB_HISTORY).query_mysql(u_sql)
return result
def special_handle(self, old_data, special_data):
"""处理可消费,可提现数据"""
bk = pd.DataFrame(special_data)
for old in old_data:
if not old.get('guild_id'):
Logger().logger.info("错误的公会数据:", old)
continue
try:
df = bk[(bk["guild_id"] == old['guild_id']) & (bk["create_time"] == old['create_time']) & (bk["amount_type"] == 2)] # 可消费
except Exception as e:
Logger().logger.info(f"处理可消费可提现报错:{e}")
continue
serializer_info = df.to_dict(orient='records')
if serializer_info:
old['con_income'] = serializer_info[0]['income']
old['con_outcome'] = serializer_info[0]['outcome']
else:
old['con_income'] = 0
old['con_outcome'] = 0
df = bk[(bk["guild_id"] == old['guild_id']) & (bk["create_time"] == old['create_time']) & (bk["amount_type"] == 3)] # 可提现
serializer_info = df.to_dict(orient='records')
if serializer_info:
old['with_income'] = serializer_info[0]['income']
old['with_outcome'] = serializer_info[0]['outcome']
else:
old['with_income'] = 0
old['with_outcome'] = 0
return old_data
def user_guild_query(self, data):
guild_sql = f"select uuid from guild"
......@@ -431,6 +487,12 @@ class AccountStatistics(object):
public_list.append(f" amount_type='backpack'")
if self.unique == 'user_account':
public_list.append(f" amount_type in('consumable','withdrawable')")
if self.unique == 'user_account' and self.amount_type == 'consumable':
public_list.append(f" amount_type in('consumable')")
if self.unique == 'user_account' and self.amount_type == 'withdrawable':
public_list.append(f" amount_type in('withdrawable')")
if self.amount_type and self.unique != 'user_account':
public_list.append(f" amount_type='{self.amount_type}'")
return public_list
def get_finance_details(self, is_list=None):
......@@ -502,14 +564,23 @@ class AccountStatistics(object):
guild_cond_list = []
start_time = time_str_to_timestamp(self.start_time + ' 00:00:00')
end_time = time_str_to_timestamp(self.end_time + ' 23:59:59')
guild_cond_list.append(f" (income>0 or outcome>0 or initial_money>0) ")
guild_cond_list.append(f" create_time >={start_time} and create_time<= {end_time}")
# guild_cond_list.append(f" (income>0 or outcome>0 or initial_money>0) ")
guild_cond_list.append(f" create_time >={start_time} and create_time< {end_time}")
if self.user_id:
guild_cond_list.append(f" guild_id={self.user_id}")
sql = f"select guild_id,initial_money,income,outcome,create_time from v3_guild_account_detail_copy where {' and '.join(guild_cond_list)} order by create_time DESC"
mysql_data = LinkMysql(env.DB_3YV2).query_mysql(sql)
other_sql = f"select guild_id,initial_money as balance,income,outcome,create_time,amount_type from v3_guild_account_detail_copy where {' and '.join(guild_cond_list)} and amount_type != 1"
guild_cond_list.append("amount_type=1")
sql = f"select guild_id,initial_money,income,outcome,amount_type,create_time from v3_guild_account_detail_copy where {' and '.join(guild_cond_list)} order by create_time DESC"
with ThreadPoolExecutor(max_workers=2) as pool:
future1 = pool.submit(LinkMysql(env.DB_3YV2).query_mysql, other_sql)
future2 = pool.submit(LinkMysql(env.DB_3YV2).query_mysql, sql)
other_data = future1.result()
mysql_data = future2.result()
Logger(20).logger.info(f"other_sql:::{other_sql}")
Logger(20).logger.info(f"sql:::{sql}")
for info in mysql_data:
info['guild_name'] = self.guild_dict.get(info['guild_id'])
mysql_data = self.special_handle(mysql_data, other_data)
return mysql_data[int(self.page - 1) * self.size: self.page * self.size], len(mysql_data)
def guild_calculation_pledge(self):
......@@ -519,7 +590,7 @@ class AccountStatistics(object):
end_time = time_str_to_timestamp(self.end_time + ' 23:59:59')
month_tuple = get_date_list(self.start_time, self.end_time)
cp_cond.append(f" create_time >={start_time} and create_time<= {end_time}")
cp_cond.append(" (income != 0 or outcome !=0 or initial_money !=0) ")
cp_cond.append(" (income != 0 or outcome !=0 or initial_money !=0) and amount_type=1")
if self.user_id:
guild_sql = f"select ice_uuid from guild where id={self.user_id}"
res_data = LinkMysql(env.DB_3YV2).query_mysql(guild_sql)
......@@ -577,11 +648,18 @@ def delete_specify_field(data, unique_tag):
def delete_guild_specify_field(data):
res_list = []
for i in data:
if not i.get('guild_id'):
Logger(20).logger.info(f'错误数据:{i}')
continue
structure = {
"name": GUILD_NAME[i.get('guild_id')],
"guild_id":i.get('guild_id', ''),
"income": i.get('income', ''),
"outcome": i.get('outcome', ''),
"con_income": i.get('con_income', ''),
"con_outcome": i.get('con_outcome', ''),
"with_income": i.get('with_income', ''),
"with_outcome": i.get('with_outcome', ''),
"initial_money": i.get('initial_money', ''),
"create_time": i.get('create_time', '')
}
......@@ -592,7 +670,7 @@ def delete_guild_specify_field(data):
class SpecificAccountQuery(object):
"""账户用户,背包,公会第四层查询"""
def __init__(self, page, size, uuid, time, type, reference_type, unique_tag, guild_id):
def __init__(self, page, size, uuid, time, type, reference_type, unique_tag, guild_id, amount_type):
self.page = page
self.size = size
self.uuid = uuid
......@@ -602,6 +680,7 @@ class SpecificAccountQuery(object):
self._type = type
self.reference_type = reference_type
self.unique_tag = unique_tag
self.amount_type = amount_type
self.total_list = []
def condition_query(self, date, cond_list):
......@@ -617,6 +696,8 @@ class SpecificAccountQuery(object):
if not guild_data:
return [], 0, 0
self.uuid = guild_data[0]['uuid']
if self.amount_type:
condition.append(f" amount_type='{self.amount_type}'")
condition.append(f" uuid='{self.uuid}'")
condition.append(f" create_time >= {time_str_to_timestamp(self.start_time + ' 00:00:00')}")
end_time = time_str_to_timestamp(self.end_time + ' 00:00:00') + 86400 # 结束时间那天包含在内,固小于第二天凌晨
......@@ -650,6 +731,13 @@ class SpecificAccountQuery(object):
for i in res:
i['reference_name'] = TYPE_NAME[i['reference_type']] if TYPE_NAME.get(i['reference_type']) else i[
'reference_type']
if i.get('amount_type'):
if i['amount_type'] == 'consumable':
i['amount_type'] = '可消费账户'
elif i['amount_type'] == 'withdrawable':
i['amount_type'] = '可提现账户'
else:
i['amount_type'] = '背包账户'
data_pd = pd.DataFrame(self.total_list)
amount_total = data_pd['amount'].sum()
return res, total, float(amount_total)
......@@ -657,15 +745,25 @@ class SpecificAccountQuery(object):
class HomePageDisplay(object):
def __init__(self, date, unique_tag):
self.date = 'assets_log_' + date
self.date = date
self.unique_tag = unique_tag
self.account = []
self.guild = []
self.bus_data = []
self.income = 0
self.outcome = 0
def th_task(self, sql):
res_ads = LinkMysql(env.DB_HISTORY).query_mysql(sql)
if res_ads:
df = pd.DataFrame(res_ads)
df_data = df.groupby(['reference_type', 'type'], as_index=False).sum()
child_thread_data = df_data.to_dict(orient='records')
Logger(20).logger.info(f'子线程{threading.Thread().getName()}查询结束!!!共{len(child_thread_data)}条')
self.bus_data += child_thread_data
def get_month_data(self):
Logger(20).logger.info("开始查询!!")
acc_sql = "select unique_tag,uuid from fi_account"
guild_sql = "select uuid from guild"
with ThreadPoolExecutor(max_workers=2) as pool:
......@@ -675,6 +773,7 @@ class HomePageDisplay(object):
guild_data = future2.result()
account = [i['uuid'] for i in acc_data]
guild = [i['uuid'] for i in guild_data]
Logger(20).logger.info("系统,公会账户查询完毕!!")
assets_cond = []
if self.unique_tag == 'guild_account':
assets_cond.append(f" uuid in{tuple(guild)}")
......@@ -690,26 +789,30 @@ class HomePageDisplay(object):
Logger(20).logger.info('没找到系统账户')
return [], 0, 0
assets_cond.append(f" uuid='{acc_uuid[0]}'")
assets_sql = f"select reference_type,type,sum(amount) as amount from {self.date} where {' and '.join(assets_cond)} GROUP BY reference_type,type"
Logger(20).logger.info("查询数据中!!!!")
assets_sql = f"select reference_type,type,sum(cast(amount as decimal(20,6))) as amount from assets_log_{self.date} where {' and '.join(assets_cond)} GROUP BY reference_type,type"
# total_data = dlc.tx_query_list("clearing_center", assets_sql, ["reference_type", "type", "amount"])
total_data = LinkMysql(env.DB_HISTORY).query_mysql(assets_sql)
# 数据分类
income = []
outcome = []
Logger(20).logger.info("数据查询完成!!!!")
for i in total_data:
op = {}
if TYPE_NAME.get(i['reference_type']):
op['name'] = TYPE_NAME.get(i['reference_type'])
else:
op['name'] = i['reference_type']
op['money'] = round(i['amount'] / 1000, 3)
op['money'] = i['amount'] / 1000
if i['type'] == 0:
op['type'] = '出账'
self.outcome = self.outcome + i['amount']
outcome.append(op)
else:
elif i['type'] == 1:
op['type'] = '入账'
self.income = self.income + i['amount']
income.append(op)
Logger(20).logger.info("返回!!!!")
res_list = outcome + income
return res_list, self.outcome/1000, self.income/1000
......
......@@ -16,6 +16,7 @@ class AccountCreate(BaseModel):
remark: Optional[str] = ''
unique_tag: Optional[str] = ''
config_key: Optional[str] = ''
beneficiary: Optional[str] = ''
income: Optional[list] = []
output: Optional[list] = []
......@@ -32,6 +33,7 @@ class AccountUpdate(BaseModel):
remark: str
income: list
output: list
beneficiary: Optional[str] = ''
class FinanceInfo(PublicModel):
......@@ -45,6 +47,7 @@ class FinanceDetails(PublicModel):
uuid: str
user_id: Optional[int] = None
unique_tag: Optional[str] = ""
amount_type: Optional[str] = ""
class FixTable(BaseModel):
......
......@@ -7,10 +7,7 @@ from app.api.account import schemas, crud
from app.api.account.crud import AccountStatistics, SpecificAccountQuery, HomePageDisplay
from app.api.statement import crud as statement_crud
from libs import functions
from libs.business import GUILD_NAME
from libs.export import Export, TableToFile
from libs.functions import get_date_list
from libs.log_utils import Logger
from libs.result_format import HttpResultResponse, HttpMessage
from libs.token_verify import login_required
......@@ -75,7 +72,13 @@ def finance_info_excel(data: schemas.FinanceInfo, request: Request,
headers = request.get("headers")
statement_list = crud.get_finance_info(data.unique_tag, data.id, data.page, 99999999, data.start_time, data.end_time)
df = ['账户余额', '入账', '出账', '时间']
return statement_crud.data_to_file(db, statement_list, "财务信息", headers, df)
if data.unique_tag in ['guild_account', 'user_account']:
df = ['id', '账户余额', '入账', '出账', '时间', '可消费入账', '可消费出账', '可提现入账', '可提现出账']
if data.unique_tag in ['anchor_account', 'knapsack_account', 'pledgeDeduction']:
df.insert(0, 'id')
# return statement_crud.data_to_file(db, statement_list, "财务信息", headers, df)
url = TableToFile(db, statement_list, data.unique_tag, headers, df).main_method()
return HttpResultResponse(data=url)
@router.get("/finance/details")
......@@ -88,12 +91,13 @@ def finance_details(page: int,
type: Optional[int] = None,
gift_type: Optional[str] = "",
unique_tag: Optional[str] = "",
amount_type: Optional[str] = "",
token=Depends(login_required)):
"""账户财务明细列表"""
if not start_time and not end_time:
return HttpResultResponse(code=500, msg="请输入你要查询的时间段")
res, total, count = AccountStatistics(page, size, uuid, user_id, start_time, end_time, type, gift_type,
unique_tag).get_finance_details(is_list=1)
unique_tag, amount_type).get_finance_details(is_list=1)
return HttpResultResponse(total=total, data=res, count=count)
......@@ -103,19 +107,21 @@ def finance_info_excel(data: schemas.FinanceDetails, request: Request,
"""账户财务明细导出"""
headers = request.get("headers")
statement_list = AccountStatistics(data.page, 99999999, data.uuid, data.user_id, data.start_time, data.end_time, data.type,
data.gift_type, data.unique_tag).get_finance_details()
data.gift_type, data.unique_tag, data.amount_type).get_finance_details()
if data.unique_tag in ["knapsack_account", "user_account", "guild_account", "pledgeDeduction"]:
field_head = ['uuid', '入账', '出账', '时间']
statement_list = statement_list[0]
if data.unique_tag == 'guild_account':
field_head = ['公会名', '公会id', '入账', '出账', '余额', '时间']
field_head = ['公会名', '公会id', '总入账', '总出账', '可消费入账', '可消费出账', '可提现入账', '可提现出账', '余额', '时间']
data = crud.delete_guild_specify_field(statement_list)
return TableToFile(db, data, "财务明细", headers, field_head).main_method()
url = TableToFile(db, data, "财务明细", headers, field_head).main_method()
return HttpResultResponse(data=url)
else:
field_head = ['订单号', '出入账方式', '礼物类型', '金额', '时间']
data = crud.delete_specify_field(statement_list, data.unique_tag)
# return statement_crud.data_to_file(db, data, "财务明细", headers, field_head)
return TableToFile(db, data, "财务明细", headers, field_head).main_method()
url = TableToFile(db, data, "财务明细", headers, field_head).main_method()
return HttpResultResponse(data=url)
@router.get("/finance/fourth/details")
......@@ -125,14 +131,15 @@ def finance_fourth_info(page: int,
guild_id: Optional[str] = "", # 针对公会账户,没有uuid,传公会id过来,再查uuid。
time: Optional[str] = "",
type: Optional[int] = None,
reference_type: Optional[str] = "",
gift_type: Optional[str] = "",
unique_tag: Optional[str] = "",
amount_type: Optional[str] = "",
token=Depends(login_required)):
"""账户财务明细 第四层"""
if not all([time, unique_tag]):
return HttpResultResponse(code=500, msg="缺少必传参数")
res, total, count = SpecificAccountQuery(page, size, uuid, time, type, reference_type, unique_tag,
guild_id).business_logic()
res, total, count = SpecificAccountQuery(page, size, uuid, time, type, gift_type, unique_tag,
guild_id, amount_type).business_logic()
return HttpResultResponse(total=total, data=res, count=count)
......@@ -203,8 +210,9 @@ def query_guilds_info(uuid: str, token=Depends(login_required)):
def read_account(date: Optional[str] = "",
unique_tag: Optional[str] = "",
account_type: Optional[str] = "",
token=Depends(login_required)):
"""月,业务类型,消费类型,出入账目统计"""
token=Depends(login_required)
):
"""月,业务类型,消费类型,出入账目统计(账户类型汇总)"""
if not date or (not unique_tag and not account_type):
return HttpResultResponse(code=500, msg=HttpMessage.MISSING_PARAMETER)
if account_type and not unique_tag:
......
......@@ -82,12 +82,12 @@ class CalculationMonthlyBill(object):
self.structure_key = []
self.query_data = []
def thead_task(self, as_list, page, size):
if as_list:
q_sql = f"SELECT reference_type, type, sum(cast(amount as decimal(20,3))) as money FROM (select reference_type,type,amount FROM {self.date} where {' and '.join(as_list)} limit {page},{size}) as a GROUP BY reference_type,type ORDER BY reference_type"
else:
def thead_task(self, page, size):
q_sql = f"SELECT reference_type, type, sum(cast(amount as decimal(20,3))) as money FROM (select reference_type,type,amount FROM {self.date} limit {page},{size}) as a GROUP BY reference_type, type ORDER BY reference_type"
count_data = LinkMysql(env.DB_HISTORY).query_mysql(q_sql)
if not count_data:
count_data = []
self.query_data = self.query_data + count_data
def data_deal_with(self):
......@@ -135,6 +135,9 @@ class CalculationMonthlyBill(object):
def search_red_data(self, red_str):
"""redis缓存数据筛选"""
res_list = []
if isinstance(red_str, list):
red_data_list = red_str
else:
red_data_list = list(eval(red_str))
for reds in red_data_list:
if self.name and not self.key_type:
......@@ -171,7 +174,7 @@ class CalculationMonthlyBill(object):
# 创建线程
for i in range(10):
ths.append(threading.Thread(target=self.thead_task,
args=[assert_list, num*i, num]))
args=[num*i, num]))
# 启动线程
for i in range(10):
ths[i].start()
......@@ -186,6 +189,10 @@ class CalculationMonthlyBill(object):
res_all_data.append(v)
# 存入redis
red.set('business_type_sum-' + str(self.date), str(res_all_data), 1800)
if assert_list:
res_all_data = self.search_red_data(res_all_data)
else:
res_all_data = self.date
else:
if assert_list:
res_all_data = self.search_red_data(business_type_sum_data)
......
......@@ -112,6 +112,7 @@ def reference_type_total(date: str, type: str):
def abnormal_total(date: str, type: str, page: Optional[int] = None, size: Optional[int] = None,
token=Depends(login_required)):
"""异常数据详情"""
# 我们的第二个情人节,也陪你走过了春夏秋冬,从陌生的彼此走到了一起,完成了爱你所爱,喜你所喜,弃你所恶这样的转变
if not all([date, type]):
return HttpResultResponse(code=500, msg='缺少必传参数')
result, tota = crud.AbnormalDataDetails(date, type, page, size).abnormal_task()
......
......@@ -96,6 +96,7 @@ class GuildMargin(object):
}
res = send_json_rpc_request(request_data, 'Server.UserExecute.Recharge')
Logger().logger.info(f"清算recharge_user结果:{str(res)}")
res['_sql'] = []
insert_sql = f"insert into all_record_table(user_id, type, status, uuid, reference_number, money, amount_type, money_data, is_add, create_time, errmsg) " \
f"values({guild_id}, '{reference_type}', 2, '{ice_uuid}', {reference_number}, {money * 10}, 1, '保证金充值',1,{timestamp}, '{json.dumps(res)}');"
if res['status'] == 9:
......@@ -104,7 +105,7 @@ class GuildMargin(object):
if res['status'] == False:
insert_sql = f"insert into all_record_table(user_id, type, status, uuid, reference_number, money, amount_type, money_data, is_add, create_time, errmsg) " \
f"values({guild_id}, '{reference_type}', 3, '{ice_uuid}', {reference_number}, {money * 10}, 1, '保证金充值',1,{timestamp}, '{json.dumps(res)}');"
print(f'保证金错误,订单号:{reference_number}')
Logger().logger.info(f'保证金错误,订单号:{reference_number}')
LinkMysql(env.DB_3YV2).perform_mysql(insert_sql)
def guild_authority(self, guild_id, amount, guild_info):
......
import json
import math
import threading
import time
from concurrent.futures.thread import ThreadPoolExecutor
from datetime import datetime
import xlsxwriter
import openpyxl
import pandas as pd
from fastapi import Depends
from sqlalchemy import func, and_
from app import get_db
from app.api.statement import schemas
from app.api.statement.guild import query_token
from starlette.responses import StreamingResponse
from sqlalchemy.orm import Session
from app.api.export import crud
from core.config.env import env
from libs.db_link import LinkMysql
from core.config.env import env
from libs import functions
from libs.db_link import LinkMysql
from libs.functions import time_str_to_timestamp, timestamp_to_time_str, get_month_last_month, get_date_list
from libs.log_utils import Logger
from libs.orm import QueryAllData
from models.recharge import Recharge, UserWC, GuildWC, FinanceFixLog
from models.menuconfig import Menuconfig
from models.users import User
from decimal import Decimal
locka = threading.Lock()
......@@ -106,13 +103,82 @@ class RechargeStatement(object):
def __init__(self):
self.linkmysql = LinkMysql(env.DB_HISTORY)
# self.count=[]
self.order_id = []
# self.once_res=[]
# self.moeny_data=[]
def user_order_data(self, query_data):
if not query_data:
return []
Logger().logger.info("获取order_id")
for bus in query_data:
bus.pop('id')
if bus['reference_type'] == 'userRecharge':
try:
ref_type, o_id = bus['reference_number'].split('_')
except:
Logger().logger.info(str(bus))
continue
if o_id not in self.order_id:
self.order_id.append(o_id)
Logger().logger.info("获取筛选条件")
pay_discount = [i['uuid'] for i in query_data]
pay_discount.append('_') # 防止列表长度为1,sql查询tuple()报错
nick_sql = f"SELECT uuid,nick_name FROM v2_user where uuid in{tuple(pay_discount)}"
nick_data = LinkMysql(env.DB_3YV2).query_mysql(nick_sql)
nick_dict = {i['uuid']: i['nick_name'] for i in nick_data}
if not self.order_id:
for ni in query_data:
ni['nick_name'] = nick_dict[ni['uuid']]
ni['income_money'] = ni['amount']
ni['status'] = 1
ni['amount'] = 0
return query_data
if len(self.order_id) == 1:
o_sql = f"SELECT o.id,o.money,o.userid,v.nick_name,o.status,o.paychannel,o.sid,o.current FROM orders as o LEFT JOIN v2_user as v on o.userid=v.user_id where o.id={self.order_id[0]}"
else:
o_sql = f"SELECT o.id,o.money,o.userid,v.nick_name,o.status,o.paychannel,o.sid,o.current FROM orders as o LEFT JOIN v2_user as v on o.userid=v.user_id where o.id in{tuple(self.order_id)}"
order_data = LinkMysql(env.DB_3YV2).query_mysql(o_sql)
Logger().logger.info("查询支付配置")
p_sql = "SELECT id,name FROM pay_config"
pay_data = LinkMysql(env.DB_3YV2).query_mysql(p_sql)
pay_dict = {}
for i in pay_data:
pay_dict[i['id']] = i['name']
for x in query_data:
if x['reference_type'] == 'userRecharge':
ref_type, o_id = x['reference_number'].split('_')
else:
o_id = '1'
for y in order_data:
if o_id == str(y['id']):
x['income_money'] = x['amount']
x['user_id'] = y['userid']
x['nick_name'] = y['nick_name']
x['amount'] = y['money']
x['status'] = y['status']
x['paychannel'] = pay_dict.get(y['paychannel'], y['paychannel'])
x['sid'] = y['sid']
else:
x['nick_name'] = nick_dict.get(x['uuid']) if nick_dict.get(x['uuid']) else ""
x['status'] = 1
Logger().logger.info(f"拼接的数据:{query_data[0]}")
return query_data
def query_sum_money(self, start_time, end_time):
start_stamp = time_str_to_timestamp(start_time + ' 00:00:00')
end_stamp = time_str_to_timestamp(end_time + ' 23:59:59')
order_sql = f"SELECT sum(money) as money FROM orders WHERE status =1 and lastupdate>={start_stamp} and lastupdate<{end_stamp}"
order_res = LinkMysql(env.DB_3YV2).query_mysql(order_sql)
if order_res:
sum_money = order_res[0]['money']
return sum_money if sum_money else 0
return 0
def query_data(self, db, page, size, order_number, uuid, sid, start_time, end_time, type, menu_id, month_type,export_status):
"""列表"""
query = []
query.append("type=1")
if order_number:
query.append(f"reference_number='{order_number}'")
if uuid:
......@@ -143,23 +209,25 @@ class RechargeStatement(object):
if start_time:
query.append(f" create_time >= {time_str_to_timestamp(start_time + ' 00:00:00')} ")
if end_time:
query.append(f" create_time < {time_str_to_timestamp(end_time + ' 23:59:59')} ")
query.append(f" create_time <= {time_str_to_timestamp(end_time + ' 23:59:59')} ")
query = ' and '.join(query)
now_month = get_month_last_month(month_type,start_time,end_time)
count, once_res, moeny_data = self.statistics_data(month_type, query, page, size, now_month[1],export_status)
else:
query_data = self.query_add_time(start_time, end_time, query)
count, once_res, moeny_data = self.thread_data(month_type, query_data, page, size,start_time,end_time,export_status)
return count, once_res, moeny_data
incone_sum_money = self.query_sum_money(start_time, end_time)
result = self.user_order_data(once_res)
return count, result, moeny_data, incone_sum_money
def query_add_time(self, start_time, end_time, old_query):
data = []
query = []
query1 = []
query.append(f" create_time >= {time_str_to_timestamp(start_time + ' 00:00:00')} ")
query.append(f" create_time < {time_str_to_timestamp(end_time + ' 23:59:59')} ")
query.append(f" create_time <= {time_str_to_timestamp(end_time + ' 23:59:59')} ")
query1.append(f" create_time >= {time_str_to_timestamp(start_time + ' 00:00:00')} ")
query1.append(f" create_time < {time_str_to_timestamp(end_time + ' 23:59:59')} ")
query1.append(f" create_time <= {time_str_to_timestamp(end_time + ' 23:59:59')} ")
if old_query:
query = query + old_query
query1 = query1 + old_query
......@@ -209,38 +277,38 @@ class RechargeStatement(object):
count, once_res, moeny_data = self.statistics_data(month_type, query, page, size, now_month,export_status)
return count, once_res, moeny_data
def get_statements(self, db, data):
def get_statements(self, data):
"""导出"""
query = []
query.append("o.status=1")
if data.order_number:
query.append(f"id={data.order_number} ")
if data.user_id:
query.append(f"uuid='{data.user_id}' ")
if data.sid:
query.append(f"sid='{data.sid}' ")
if data.menu_id:
querydata = db.query(Menuconfig).filter(Menuconfig.id.in_(tuple(data.menu_id)))
reference_data = [
QueryAllData.serialization(item, remove={'menu_name', 'remark', 'menu_type', 'create_time', 'id'}) for
item in querydata]
if len([item.get("menu_label") for item in reference_data]) > 1:
reference_type = tuple([item.get("menu_label") for item in reference_data])
else:
reference_type = "('" + [item.get("menu_label") for item in reference_data][0] + "')"
query.append(f"reference_type in {reference_type}")
if data.month_type == 1:
Logger().logger.info(f"查询reference_number是:{data.order_number}!!!!!")
_, number = data.order_number.split('_')
query.append(f"o.id={number}")
if data.uuid:
u_sql = f"SELECT user_id FROM v2_user WHERE uuid='{data.uuid}'"
user_info = LinkMysql(env.DB_3YV2).query_mysql(u_sql)
if user_info:
if user_info[0]['user_id']:
query.append(f" o.userid={user_info[0]['user_id']}")
if data.start_time:
query.append(f" create_time >= {time_str_to_timestamp(data.start_time + ' 00:00:00')} ")
query.append(f" o.lastupdate >= {time_str_to_timestamp(data.start_time + ' 00:00:00')} ")
if data.end_time:
query.append(f" create_time < {time_str_to_timestamp(data.end_time + ' 23:59:59')} ")
query = ' and '.join(query)
now_month = get_month_last_month(data.month_type,data.start_time)
once_res = self.data_delcy(data.month_type, query, now_month[1])
else:
query_data = self.query_add_time(query,data.start_time, data.end_time)
now_month = get_month_last_month(data.month_type)
once_res = self.data_delcy(data.month_type, query_data, now_month)
return once_res
query.append(f" o.lastupdate <= {time_str_to_timestamp(data.end_time + ' 23:59:59')} ")
Logger().logger.info("查询支付配置")
p_sql = "SELECT id,name FROM pay_config"
pay_data = LinkMysql(env.DB_3YV2).query_mysql(p_sql)
pay_dict = {}
for i in pay_data:
pay_dict[i['id']] = i['name']
Logger().logger.info("查询orders表!!!!!")
or_sql = f"SELECT o.userid,v.nick_name,o.money,o.status,o.paychannel,o.sid,o.lastupdate FROM orders as o LEFT JOIN v2_user as v on o.userid=v.user_id WHERE {' and '.join(query)}"
order_info = LinkMysql(env.DB_3YV2).query_mysql(or_sql)
for order in order_info:
order['paychannel'] = pay_dict.get(order['paychannel'], order['paychannel'])
order['status'] = '成功' if order['status'] else '失败'
order['lastupdate'] = timestamp_to_time_str(order['lastupdate'])
return order_info
def data_delcy(self, month_type, query, now_month):
if month_type == 1:
......@@ -258,25 +326,66 @@ class WithdrawStatement(object):
"""提现报表"""
def __init__(self):
self.derive_user_list = []
self.linkmysql = LinkMysql(env.DB_HISTORY)
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_add_time(self,query,start_time, end_time):
query.append(f" create_time >= {time_str_to_timestamp(start_time + ' 00:00:00')} ")
query.append(f" create_time < {time_str_to_timestamp(end_time + ' 23:59:59')} ")
return query
def user_data_structure(self, user_data, st_time, en_time):
if not user_data:
return user_data
Logger().logger.info("通过assets_log业务数据查用户uuid")
user_uuid_list = []
for user in user_data:
if user['uuid'] not in user_uuid_list:
user_uuid_list.append(user['uuid'])
if len(user_uuid_list) == 1:
user_uuid_list.append(datetime.now().strftime('%Y%m%d%H%M%S'))
Logger().logger.info("用户uuid查用户id")
u_sql = f"select user_id,uuid from v2_user where uuid in{tuple(user_uuid_list)}"
user_id_info = LinkMysql(env.DB_3YV2).query_mysql(u_sql)
user_ids = []
user_uuid_id = {}
user_id_uuid = {}
for i in user_id_info:
user_ids.append(i['user_id'])
user_uuid_id[i['uuid']] = i['user_id']
user_id_uuid[i['user_id']] = i['uuid']
if len(user_ids) == 1:
user_ids.append(datetime.now().strftime('%Y%m%d%H%M%S'))
Logger().logger.info("用户id查用户所属公会")
condition = []
if st_time:
condition.append(f" t.update_time>={time_str_to_timestamp(st_time + ' 00:00:00')}")
if en_time:
condition.append(f" t.update_time<{time_str_to_timestamp(en_time + ' 23:59:59')}")
if condition:
gu_sql = f"select t.userid,t.usernumber,t.nickname,t.current,t.update_time,t.money,t.final_money,g.guild_name from tixian_order as t LEFT JOIN guild as g on t.guild_id=g.id where t.userid in{tuple(user_ids)} and {(' and '.join(condition))}"
else:
gu_sql = f"select t.userid,t.usernumber,t.nickname,t.current,t.update_time,t.money,t.final_money,g.guild_name from tixian_order as t LEFT JOIN guild as g on t.guild_id=g.id where t.userid in{tuple(user_ids)}"
guild_user_info = LinkMysql(env.DB_3YV2).query_mysql(gu_sql)
user_to_guild = {}
for ug in guild_user_info:
ug['uuid'] = user_id_uuid[ug['userid']]
user_to_guild[ug['userid']] = ug.get('guild_name','')
Logger().logger.info("拼接数据!!!!")
for us in user_data:
for tx in guild_user_info:
amount = Decimal(us['amount']).quantize(Decimal("0.00"))
if tx['uuid'] == us['uuid'] and int(amount) == tx['money']:
user_id = user_uuid_id.get(us['uuid'])
us['guild_name'] = user_to_guild.get(user_id)
us['dec_money'] = tx['money'] * 0.05
us['withdrawal_time'] = time_str_to_timestamp(str(tx['current']))
us['update_time'] = tx['update_time']
us['final_money'] = Decimal(tx['final_money']).quantize(Decimal("0.00"))
us['usernumber'] = tx['usernumber']
us['nickname'] = tx['nickname']
us['money'] = tx['money']
return user_data
def get_user_withdraw_cash(self, db, page, size, uuid, status, start_time, end_time, month_type, menu_id):
query = []
if uuid:
......@@ -317,7 +426,8 @@ class WithdrawStatement(object):
query_data = self.query_add_time(query,start_time, end_time)
now_month = get_month_last_month(month_type,start_time,end_time)
count, once_res, moeny_data,reality_moeny = self.dispose_user(month_type, query_data, page, size, now_month)
return count, once_res, moeny_data/1000,reality_moeny/1000
res_user_data = self.user_data_structure(once_res, start_time, end_time)
return count, res_user_data, moeny_data/1000,reality_moeny/1000
def dispose_user(self, month_type, query, page, size, now_month):
if month_type == 1:
......@@ -349,6 +459,33 @@ class WithdrawStatement(object):
reality_moeny=moeny_data-abs(moeny_income)
return count, once_res, moeny_data,reality_moeny
def structure_data(self, guild_data, s_time, e_time):
guild_query = []
Logger().logger.info("拼接数据!!!")
if s_time:
guild_query.append(f" t.update_time>={time_str_to_timestamp(s_time + ' 00:00:00')} ")
if e_time:
guild_query.append(f" t.update_time<{time_str_to_timestamp(e_time + ' 23:59:59')} ")
if guild_query:
gui_sql = f"SELECT g.id,g.guild_name,g.uuid,t.withdrawal_time,t.update_time,t.money,t.dec_money,t.finalMoney from guild_ti_xian as t LEFT JOIN guild as g on t.guild_id=g.id where {(' and '.join(guild_query))}"
else:
gui_sql = "SELECT g.id,g.guild_name,g.uuid,t.withdrawal_time,t.update_time,t.money,t.dec_money,t.finalMoney from guild_ti_xian as t LEFT JOIN guild as g on t.guild_id=g.id"
guild_info = LinkMysql(env.DB_3YV2).query_mysql(gui_sql)
for guild in guild_data:
for i in guild_info:
amount = Decimal(guild['amount']).quantize(Decimal("0.00"))
if guild['uuid'] == i['uuid'] and str(amount) == i['finalMoney']:
guild['withdrawal_time'] = i['withdrawal_time']
guild['update_time'] = i['update_time']
guild['guild_id'] = i['id']
guild['guild_name'] = i['guild_name']
guild['money'] = i['money']
guild['dec_money'] = i['dec_money']
guild['finalMoney'] = Decimal(i['finalMoney']).quantize(Decimal("0.00"))
Logger().logger.info("完成!!!")
return guild_data
def get_guild_withdraw_cash(self, db, page, size, name, status, start_time, end_time, month_type, menu_id):
"""公会提现"""
query =[]
......@@ -390,7 +527,9 @@ class WithdrawStatement(object):
query_data = self.query_add_time(query,start_time, end_time)
now_month = get_month_last_month(month_type,start_time,end_time)
count, once_res, moeny_data,reality_moeny = self.dispose_user(month_type, query_data, page, size, now_month)
return count, once_res, moeny_data/1000,reality_moeny/1000
# 数据构造
res_list = self.structure_data(once_res, start_time, end_time)
return count, res_list, moeny_data/1000, reality_moeny/1000
class FinanceFix(object):
......@@ -426,6 +565,90 @@ class FinanceFix(object):
return [], 0
class BlindBoxRecharge(object):
"""盲盒充值列表"""
def __init__(self, params):
self.page = int(params.get('page'))
self.size = int(params.get('size'))
self.start_time = params.get('start_time')
self.end_time = params.get('end_time')
self.reference_number = params.get('reference_number')
self.uuid = params.get('uuid')
self.uuid_list = []
def _conditions(self):
cond_list = []
cond_list.append("reference_type='blind_box_mall_account_recharge'")
if self.start_time:
cond_list.append(f"create_time >= {time_str_to_timestamp(self.start_time + ' 00:00:00')} ")
if self.end_time:
cond_list.append(f"create_time < {time_str_to_timestamp(self.end_time + ' 23:59:59')} ")
if self.reference_number:
cond_list.append(f"reference_number='{self.reference_number}'")
if self.uuid:
cond_list.append(f"uuid='{self.uuid}'")
return cond_list
@staticmethod
def time_month(y_month, num):
now_time = datetime.now().strftime('%Y%m')
if num == 2:
if int(y_month) > int(now_time):
return now_time
if num == 1:
if int(y_month) < int(now_time):
return now_time
return y_month
@staticmethod
def _recharge_data(query_data):
if not query_data:
return []
bk = pd.DataFrame(query_data)
bk['nick_name'] = "商城礼物盲盒"
bk['status'] = 1
pd_to_dict = bk.to_dict(orient='records')
Logger().logger.info(f"拼接的数据:{pd_to_dict[0]}")
return pd_to_dict
def box_recharge(self, is_export=None):
conditions = self._conditions()
start_month = self.start_time.replace('-', '')[:-2]
end_month = self.end_time.replace('-', '')[:-2]
if is_export:
self.size = 999999999
# 时间处理,看查询一个月还是两个月
if start_month == end_month:
year_month = self.time_month(start_month, 1)
time_list = [year_month]
else:
eny_month = self.time_month(end_month, 2)
time_list = [start_month, eny_month]
time_list = list(set(time_list))
Logger().logger.info(f"开始查询数据!!!!日期:{time_list}, 条件:{conditions}")
if len(time_list) == 1:
data_sql = f"SELECT uuid,amount,FROM_UNIXTIME(create_time) as payment_time,reference_type,reference_number,channel,business_number FROM recharge_log_{time_list[0]} where {' and '.join(conditions)} ORDER BY payment_time desc limit {(int(self.page) - 1) * self.size},{self.size}"
money_sql = f"SELECT sum(amount) as m_sum FROM recharge_log_{time_list[0]} where {' and '.join(conditions)}"
count_sql = f"SELECT count(id) as c_sum FROM recharge_log_{time_list[0]} where {' and '.join(conditions)}"
else:
data_sql = f"SELECT uuid,amount,FROM_UNIXTIME(create_time) as payment_time,reference_type,reference_number,channel,business_number FROM recharge_log_{time_list[0]} where {' and '.join(conditions)} UNION ALL SELECT uuid,amount,FROM_UNIXTIME(create_time) as payment_time,reference_type,reference_number,channel,business_number FROM recharge_log_{time_list[1]} where {' and '.join(conditions)} ORDER BY payment_time desc limit {(self.page - 1) * self.size},{self.size}"
money_sql = f"SELECT sum(a.b) as m_sum FROM ("f"SELECT sum(amount) as b FROM recharge_log_{int(time_list[0])} where {' and '.join(conditions)} UNION ALL SELECT sum(amount)FROM recharge_log_{int(time_list[1])} where {' and '.join(conditions)}) AS a "
count_sql = "SELECT sum(a.b) as c_sum FROM ("f"SELECT count(id) as b FROM recharge_log_{int(time_list[0])} where {' and '.join(conditions)} UNION ALL SELECT count(id) FROM recharge_log_{int(time_list[1])} where {' and '.join(conditions)}) AS a "
with ThreadPoolExecutor(max_workers=3) as pool:
future1 = pool.submit(LinkMysql(env.DB_HISTORY).query_mysql, data_sql)
future2 = pool.submit(LinkMysql(env.DB_HISTORY).query_mysql, money_sql)
future3 = pool.submit(LinkMysql(env.DB_HISTORY).query_mysql, count_sql)
res_data = future1.result()
res_money = future2.result()
res_count = future3.result()
Logger().logger.info("查询数据over!!!!")
res_data_to = self._recharge_data(res_data)
if res_data_to:
return res_data_to, res_money[0].get('m_sum'), res_count[0].get('c_sum')
else:
return [], 0, 0
def create_menu(db: Session, menu: schemas.MenuAdd):
try:
db_menu = Menuconfig(menu_name=menu.menu_name, menu_label=menu.menu_label, menu_type=menu.menu_type,
......@@ -492,3 +715,53 @@ def get_menu_config(db: Session, menu_type):
Menuconfig.id.desc())
data = [QueryAllData.serialization(item, remove={'create_time', 'remark'}) for item in querydata]
return data
def user_data_handle(data):
"""用户提现数据处理"""
for_mat = []
for i in data:
ting = {}
ting['reference_number'] = i.get('reference_number')
ting['money'] = i.get('money')
ting['reference_type'] = i.get('reference_type')
ting['uuid'] = i.get('uuid')
ting['amount_type'] = i.get('amount_type')
ting['payment_time'] = i.get('payment_time')
for_mat.append(ting)
return for_mat
def guild_data_handle(data):
"""公会提现数据处理"""
guild_mat = []
for i in data:
mat = {}
w_time = functions.timestamp_to_time_str(i.get('withdrawal_time'))
mat['withdrawal_time'] = w_time if w_time else ''
mat['payment_time'] = i.get('payment_time')
mat['guild_id'] = i.get('guild_id')
mat['guild_name'] = i.get('guild_name')
mat['money'] = i.get('money')
mat['dec_money'] = i.get('dec_money')
mat['finalMoney'] = i.get('finalMoney')
guild_mat.append(mat)
return guild_mat
def blind_box_handle(data):
"""盲盒充值数据处理"""
box_mat = []
for i in data:
box = {}
box['reference_number'] = i['reference_number']
box['sid'] = i.get('sid')
box['paychannel'] = i.get('paychannel')
box['uuid'] = i.get('uuid')
box['nick_name'] = i.get('nick_name')
box['status'] = '成功' if i['status'] else '失败'
box['reference_type'] = '盲盒充值'
box['income_money'] = i.get('income_money')
box['payment_time'] = i.get('payment_time')
box_mat.append(box)
return box_mat
......@@ -7,10 +7,12 @@ from app.api.account import schemas as acc_schemas
from app import get_db
from fastapi import Depends, APIRouter, File, Request
from sqlalchemy.orm import Session
from app.api.statement.crud import RechargeStatement, WithdrawStatement, get_menu_list, get_menu_config
from app.api.statement.crud import RechargeStatement, WithdrawStatement, get_menu_list, get_menu_config, \
BlindBoxRecharge
from app.api.statement.guild import GuildSet, paymentset_guild_data, outon_account_data, accout_list_data, \
query_uuid_or_user_number, account_money, transfer_money, transfer_query, GuildSettlementAdd, GuildSettlementmodify
from app.api.statement.schemas import PaymentWithdrawalList, PaymentAdd, PaymentAccountlList, UserNumber, CreateBill
from libs.export import TableToFile
from libs.img_code import new_upload_file, random_number
from libs.result_format import HttpResultResponse, HttpMessage
from libs.token_verify import login_required
......@@ -27,20 +29,18 @@ def statement_recharge_list(request: Request,db: Session = Depends(get_db),page:
return HttpResultResponse(code=500, msg='时间为必传参数')
query_params = request.query_params
menu_id=query_params.getlist("menu_id[]")
total,statement_list,money= RechargeStatement().query_data(db,page,size,order_number,uuid,sid,start_time,end_time,types,menu_id,month_type,'')
return HttpResultResponse(total=total,count=float(money),data=statement_list)
total,statement_list,money, sun_m = RechargeStatement().query_data(db,page,size,order_number,uuid,sid,start_time,end_time,types,menu_id,month_type,'')
return HttpResultResponse(total=total,count=float(money),data=statement_list, sum_money=sun_m)
@router.post("/derive/excel")
def statement_derive_excel(request:Request,data: schemas.StatementList,db: Session = Depends(get_db),token=Depends(login_required)):
"""充值报表导出"""
header_list = request.get("headers")
export_status =1
total,statement_list,money= RechargeStatement().query_data(db, data.page, data.size, data.order_number, data.uuid,
data.sid, data.start_time, data.end_time, data.types,
data.menu_id, data.month_type,export_status)
field_list = ["id", "uuid", "充值金额(元)", "支付时间", "类型", "订单号"]
return crud.data_to_file(db, statement_list, "充值报表", header_list, field_list)
statement_list = RechargeStatement().get_statements(data)
field_list = ["用户Id", "昵称", "充值金额(元)", "充值状态", "渠道", "商户订单号", "充值时间"]
url = TableToFile(db, statement_list, "充值报表", header_list, field_list).main_method()
return HttpResultResponse(data=url)
@router.get("/userWithdrawal/list")
......@@ -218,7 +218,7 @@ def recovery_fix(data: acc_schemas.RecoveryupdateTable, token=Depends(login_requ
@router.post("/menu/add")
def menu_add(data: schemas.MenuAdd, db: Session = Depends(get_db), token=Depends(login_required)):
"""新增财务菜单配置"""
"""新增财务菜单配置"""
db_user = crud.get_menu_name(db,menu_name=data.menu_name)
if db_user:
return HttpResultResponse(code=400, msg=HttpMessage.TYPE_NAME)
......@@ -240,7 +240,7 @@ def menu_edit(data: schemas.MenuEdit,db: Session = Depends(get_db),page: Optiona
"""修改配置列表"""
db_menu = crud.get_menu_id(db, id=data.id)
if db_menu:
db_menu = crud.get_menu_update(db, data)
db_menu = crud.get_menu_update(db,data)
else:
return HttpResultResponse(code=400, msg=HttpMessage.MENU_NOT_EXIST)
return HttpResultResponse(data=db_menu)
......@@ -256,7 +256,7 @@ def menu_delte(id: Optional[int] = '',db: Session = Depends(get_db),page: Option
@router.get("/menu/config")
def menu_list(db: Session = Depends(get_db),menu_type: Optional[int] = ""):
"""菜单配置下拉"""
menu_list = get_menu_config(db, menu_type)
menu_list = get_menu_config(db,menu_type)
return HttpResultResponse(data=menu_list)
......@@ -272,3 +272,74 @@ def guild_modify(db: Session = Depends(get_db)):
"""公会结算同步"""
code, data = GuildSettlementmodify(db)
return HttpResultResponse(code=code, msg=HttpMessage.SUCCESS)
@router.get("/userWithdrawal/excel")
def user_withdrawal_excel(request: Request,
db: Session = Depends(get_db),
page: Optional[int] = 1,
size: Optional[int] = 9999999,
uuid: Optional[str] = '',
status: Optional[int] = '',
start_time: Optional[str] = '',
end_time: Optional[str] = "",
month_type: Optional[int] = "",
token=Depends(login_required)):
"""用户提现列表导出"""
query_params = request.query_params
menu_id = query_params.getlist("menu_id[]")
header_list = request.get("headers")
to, statement_list, mo, re = WithdrawStatement().get_user_withdraw_cash(db, page, 9999999, uuid, status, start_time,
end_time, month_type, menu_id)
field_list = ["订单号", "提现金额", "业务类型", "uuid", "账户类型", "提现时间"]
res_data = crud.user_data_handle(statement_list)
url = TableToFile(db, res_data, "用户提现", header_list, field_list).main_method()
return HttpResultResponse(data=url)
@router.get("/guildWithdrawal/excel")
def guild_withdrawal_excel(request: Request,
db: Session = Depends(get_db),
page: Optional[int] = 1,
size: Optional[int] = 9999999,
guild_id: Optional[int] = '',
status: Optional[int] = '',
start_time: Optional[str] = '',
end_time: Optional[str] = "",
month_type: Optional[int] = "",
token=Depends(login_required)):
"""公会提现导出"""
query_params = request.query_params
menu_id=query_params.getlist("menu_id[]")
header_list = request.get("headers")
res = WithdrawStatement().get_guild_withdraw_cash(db,page,9999999,guild_id,status,start_time,end_time,month_type,menu_id)
field_list = ["提现时间", "处理时间", "公会ID", "公会名称", "提现金额", "扣除金额", "实际到账金额"]
res_data = crud.guild_data_handle(res[1])
url = TableToFile(db, res_data, "公会提现", header_list, field_list).main_method()
return HttpResultResponse(data=url)
@router.get("/blind_box/recharge")
def blind_box_recharge(request: Request, token=Depends(login_required)):
"""盲盒充值"""
query_params = request.query_params
if not all([query_params.get('start_time'), query_params.get('end_time')]):
return HttpResultResponse(code=500, msg='时间为必传参数')
data, money, count = BlindBoxRecharge(query_params).box_recharge()
return HttpResultResponse(total=count, money=money, data=data)
@router.get("/blind_box/excel")
def blind_box_excel(request: Request,
db: Session = Depends(get_db),
token=Depends(login_required)):
"""盲盒充值导出"""
query_params = request.query_params
header_list = request.get("headers")
if not all([query_params.get('start_time'), query_params.get('end_time')]):
return HttpResultResponse(code=500, msg='时间为必传参数')
data, _, _ = BlindBoxRecharge(query_params).box_recharge(1)
field_list = ["订单号", "商户订单号", "渠道", "UUID", "昵称", "充值状态", "业务类型", "充值金额(元)", "充值时间"]
res_data = crud.blind_box_handle(data)
url = TableToFile(db, res_data, "盲盒充值", header_list, field_list).main_method()
return HttpResultResponse(data=url)
......@@ -20,7 +20,7 @@ def get_user(db, username: str):
try:
result = db.query(users.User).filter(users.User.username == username).first()
except Exception as e:
result = db.query(users.User).filter(users.User.username == username).first()
return None
return result
......@@ -91,6 +91,8 @@ def authenticate_pwd(db: Session, form_data: GoogleLogin):
"""只验证密码"""
user_data = get_user(db=db, username=form_data.username)
# 如果密码不正确,也是返回False
if not user_data:
return {"result": False, "msg": "密码错误"}
md5_password = md5(form_data.password)
if md5_password != user_data.hashed_password:
return {"result": False, "msg": "密码错误"}
......
......@@ -34,6 +34,7 @@ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Toke
client = CosS3Client(config)
COS_PATH = 'https://fj-dc-test-1256890024.cos.ap-guangzhou.myqcloud.com' # 测试
COS_RERURN_PATH = '/images/'
tencent = apo.get("TencentCloud")
class Env(BaseSettings):
......@@ -57,13 +58,13 @@ class TestingEnv(Env):
NACOS_URL = YAML_DATA.get('config_url')
NACOSCONFIG = "show=all&dataId=fj-finance-test&group=DEFAULT_GROUP&tenant=cw-test&namespaceId=cw-test"
NACOS_NAME = YAML_DATA.get('name')
NACOS_PWD = YAML_DATA.get('password')
NACOS_PWD = YAML_DATA.get('pwd')
DB_HISTORY = apo.get('history')
DB_3YV2 = apo.get('business')
Redis = apo.get('redis')
SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM: str = "HS256"
PASSWORD: str = "fj123456"
PASSWORD: str = "fj147sy258.#"
oss_url = 'http://oss.3yakj.com/application_static_data'
CLEARING_CENTER_URL: str = 'http://106.55.103.148:6464/'
CLEARING_CENTER_HOST: str = '106.55.103.148'
......@@ -72,6 +73,7 @@ class TestingEnv(Env):
PHP_URL = "http://106.55.103.148:8787"
class ProdEnv(Env):
"""生产环境配置"""
......@@ -80,7 +82,7 @@ class ProdEnv(Env):
PEM_PATH = os.path.join(SITE_ROOT_YAML, "config", "")
YAML_DATA = apo.get('yaml')
NACOS_NAME = YAML_DATA.get('name')
NACOS_PWD = YAML_DATA.get('password')
NACOS_PWD = YAML_DATA.get('pwd')
LOGIN_URL = YAML_DATA.get('login_url')
NACOS_URL = YAML_DATA.get('config_url')
NACOSCONFIG = "dataId=fj-finance&group=DEFAULT_GROUP&namespaceId=cw-pro&tenant=cw-pro&show=all&username=fj_finance"
......@@ -89,7 +91,7 @@ class ProdEnv(Env):
Redis = apo.get('redis')
SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM: str = "HS256"
PASSWORD: str = "fj123456"
PASSWORD: str = "fj147sy258.#"
CLEARING_CENTER_URL: str = 'http://219.152.95.226:5454/'
CLEARING_CENTER_HOST: str = '219.152.95.226'
CLEARING_CENTER_PORT: int = 5454
......@@ -97,8 +99,9 @@ class ProdEnv(Env):
PHP_URL = "http://219.152.95.226:6750"
# env = TestingEnv() # 开发环境
env = ProdEnv() # 生产环境
env = ProdEnv() # 生产环境docke
redis_data = env.Redis
pool = redis.ConnectionPool(host=redis_data.get("host"), port=redis_data.get("port"), password=redis_data.get("password"),
......
......@@ -21,9 +21,9 @@ SQLALCHEMY_DATABASE_URL = f'sqlite:///{modul_path}/sql_app.db'
# SQLALCHEMY_DATABASE_URL, encoding='utf-8', echo=True, connect_args={'check_same_thread': False}
# )
#数据中心测试服
engine = create_engine('mysql+mysqldb://data_center:KCMBfAjeJhbJXsSe@43.138.132.9:3398/finance')
engine = create_engine('mysql+mysqldb://data_center:KCMBfAjeJhbJXsSe@43.138.132.9:3398/finance', pool_size=20)
#财务原数据数据库
engine_3yakj = create_engine('mysql+mysqldb://root:c1ea602311a369f6@106.55.103.148:3398/3yakj_v2')
engine_3yakj = create_engine('mysql+mysqldb://root:c1ea602311a369f6@106.55.103.148:3398/3yakj_v2', pool_size=20)
# 数据库 session 类,用于创建 session 实例
# autoflush 是指发送数据库语句到数据库,但数据库不一定执行写入到磁盘
......
#version: "3"
#services:
# app:
# restart: always
# container_name: financial-system
# build: .
# ports:
# - "8001:8001"
# volumes:
# - /www/wwwroot/financial-system/:/financial-system/
# stdin_open: true
# command: python main.py
version: "3"
version: '3'
services:
financial-system:
app:
restart: always
container_name: financial-system
build:
context: ./../_base/python-38/
image: python-38
build: .
ports:
- "8009:8009"
deploy:
resources:
limits:
cpus: "4.00"
memory: 16G
reservations:
memory: 500M
- '8009:8009'
volumes:
- /www/python-38/site-packages/financial-system/:/usr/local/lib/python3.8/site-packages
- /www/wwwroot/financial-system/:/var/www/
logging:
driver: json-file
options:
max-size: "20m"
max-file: "10"
stdin_open: true
command: >
sh -c "python -m ensurepip &&
python -m pip install --upgrade pip &&
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple &&
python main.py"
\ No newline at end of file
- .:/financial-system
command: python main.py
......@@ -60,3 +60,31 @@ class LinkMysql(object):
# rb_info = pika.PlainCredentials(rb.get('username'), rb.get('password'))
# self.connection = pika.BlockingConnection(parameters=pika.ConnectionParameters(rb.get('host'), rb.get('port'), rb.get('vhost'), rb_info))
# self.channel = self.connection.channel()
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
# 连接服务器
class LinkMonitor(object):
def __init__(self):
conn = CosConfig(Region=env.TX_REGION, SecretId=env.TX_SECRET_ID, SecretKey=env.TX_SECRET_KEY,
Token=None, Scheme='https')
self.client = CosS3Client(conn)
# 断点上传
def upload_block(self, path, key, folder=""):
# 正常情况日志级别使用INFO,需要定位时可以修改为DEBUG,此时SDK会打印和服务端的通信信息
# logging.basicConfig(level=logging.INFO, stream=sys.stdout)
token = None
scheme = 'https' # 指定使用 http/https 协议来访问 COS,默认为 https,可不填
response = self.client.upload_file(
Bucket=env.TX_BUCKET,
Key=folder + key + '.xlsx',
LocalFilePath=path,
EnableMD5=False,
progress_callback=None
)
import datetime
import json
import time
import requests
from core.config.env import env
from libs import functions
from libs.log_utils import Logger
def retry(number=1):
"""
最外层传递装饰器参数(这一层可以不写)
中间层传递被装饰器装饰的函数,这里相当于get()
内层传递被装饰器装饰函数的参数,这里是get()的参数a
装饰器作用:限制request重复请求的次数
"""
def outer(func):
def inner(database_name, mon_sql, field_list):
num = 1
res_data = 0
for i in range(number):
try:
res_data = func(database_name, mon_sql, field_list)
break
except Exception as e:
Logger(21).logger.info(f"------------JDBC超时或者失败{i + 1}次-----------")
time.sleep(1)
num += 1
if num == 3:
now_time = functions.get_now_datetime()
environment = "正式环境"
if 'dev' in env.TX_URL:
environment = "测试环境"
msg = {
"error_time": now_time,
"error_env": environment,
"error_line": str(e.__traceback__.tb_frame.f_globals["__file__"]) + ':' + str(
e.__traceback__.tb_lineno),
"error_info": str(e),
"query_sql": str(mon_sql)
}
Logger(21).logger.error("腾讯数据库查询异常:" + str(msg))
return res_data
return res_data
return inner
return outer
@retry(number=2)
def tx_query_num(database_name, mon_sql, field_list):
"""
查询腾讯sql总数语句
:param database_name 数据库
:param mon_sql 查询sql
:param field_list 要返回的字段
:return 0
"""
data = {
"sql_sentence": mon_sql,
"database_name": database_name,
"expiretime": 10,
"fields": field_list
}
start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
res = requests.post(url=env.TX_URL, json=data)
end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
if res.status_code != 200:
raise Exception(res.text)
content = json.loads(res.text)
Logger(21).logger.info(f"数据库查询sql:{str(mon_sql)},结果:{str(content)},开始时间:{str(start_time)},结束时间:{str(end_time)}")
if content.get("code") == 200:
row = content.get("rows")[0]
amount = row.get("amount")
if not amount:
amount = 0
else:
amount = 0
return amount
@retry(number=2)
def tx_query_list(database_name, mon_sql, field_list):
"""
查询sql语句
:param database_name
:param mon_sql
:param field_list
:return []
"""
data = {
"sql_sentence": mon_sql,
"database_name": database_name,
"expiretime": 10,
"fields": field_list
}
start_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
res = requests.post(url=env.TX_URL, json=data)
end_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
if res.status_code != 200:
raise Exception(res.text)
content = json.loads(res.text)
Logger(21).logger.info(f"---结果:{content}--")
Logger(21).logger.info(
f"数据库查询sql:{database_name}---结果:{content.get('code')}---,开始时间:{start_time},结束时间:{end_time}")
if res.status_code == 200:
row = content.get("rows")
if not row:
row = []
else:
row = []
return row
import math
import os
import random
import time
import openpyxl
import threading
import pandas as pd
from fastapi import Response
from app.api.statement.guild import query_token
from starlette.responses import StreamingResponse
from datetime import datetime
from app.api.export import crud
from core.config.env import env
from libs.db_link import LinkMonitor
from libs.log_utils import Logger
......@@ -101,6 +107,137 @@ class TableToFile(object):
bk.to_excel(writer, sheet_name=sheet_name, index=False)
self.lock.release()
def th_task(self, branch_data, f_name, num):
try:
bk = pd.DataFrame(branch_data)
if branch_data[0].get('create_time'):
if isinstance(branch_data[0]['create_time'], int):
bk['create_time'] = bk['create_time'].apply(
lambda x: time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(x)))
bk.columns = self.field_list # 修改pandas头
with pd.ExcelWriter(f'static/{f_name}/{self.name}-{num}.xlsx') as writer:
bk.to_excel(writer, sheet_name='Sheet1', index=False)
except Exception as e:
Logger(40).logger.error(f'导出线程{threading.Thread().getName()}失败,原因:{e}')
def th_number(self, total):
if total <= 500000:
strip = 5
data = math.ceil(total/5)
elif 500000 < total <= 1000000:
strip = 7
data = math.ceil(total/strip)
else:
# 当数量超过100w条时,每个xlsx只写10w条数据
data = 100000
strip = math.ceil(total/data)
return strip, data
def compress_folder(self, name):
import zipfile
# 定义需要压缩的文件夹路径和名称
directory_name = f"static/{name}"
zip_file_name = f"static/{self.name}.zip"
# 创建 ZipFile 对象,用于写入压缩文件
with zipfile.ZipFile(zip_file_name, 'w', compression=zipfile.ZIP_DEFLATED) as zip_file:
# 遍历需要压缩的文件夹中的所有子目录和文件
for root, dirs, files in os.walk(directory_name):
for file in files:
# 构造文件的完整路径
file_path = os.path.join(root, file)
# 在压缩文件中添加文件
zip_file.write(file_path)
return zip_file_name
# def main_method(self):
# """主函数"""
# Logger().logger.info('开始导出')
# user = query_token(self.db, self.header)
# params = {"source": self.name, "method": "data_to_file", "status": 1}
# if len(self.data) == 0:
# params["status"] = 3
# crud.create_export_data(self.db, params, user)
# Logger().logger.info(f'导出没有数据')
# return None
# folder_name = datetime.now().strftime('%m%d%H%M%S')
# try:
# os.mkdir(f"static/{folder_name}")
# Logger().logger.info("文件夹已创建!")
# except OSError as error:
# uid = random.randint(1, 1000)
# Logger().logger.info(f"无法创建目录:{folder_name},原因:{error},重新创建随机文件夹")
# folder_name = folder_name + str(uid)
# os.mkdir(f"static/{folder_name}")
# # 判断多少条线程
# number, count = self.th_number(len(self.data))
# Logger().logger.info(f"开启线程:{number}, 每个数量:{count}")
# # 起线程
# ths = []
# for x in range(number):
# ths.append(threading.Thread(target=self.th_task,
# args=[self.data[x * count:(1 + x) * count], folder_name, x]))
# # 启动线程
# for y in range(number):
# ths[y].start()
# # 等待所有线程完成
# for z in range(number):
# ths[z].join()
# Logger().logger.info(f"线程结束,压缩文件!!!")
# zip_folder = self.compress_folder(folder_name)
# # 记录导出
# crud.create_export_data(self.db, params, user)
# with open(zip_folder, 'rb') as f:
# data = f.read()
# response = Response(content=data)
# response.headers["Content-Disposition"] = "attachment; filename=example.zip"
# Logger().logger.info(f"返回压缩文件!!!")
# return response
# def main_method(self):
# """主函数"""
# Logger().logger.info('开始导出')
# user = query_token(self.db, self.header)
# params = {"source": self.name, "method": "data_to_file", "status": 1}
# if len(self.data) == 0:
# params["status"] = 3
# crud.create_export_data(self.db, params, user)
# Logger().logger.info(f'导出没有数据')
# return None
# try:
# bk = pd.DataFrame(self.data)
# if self.data[0].get('create_time'):
# if isinstance(self.data[0]['create_time'], int):
# bk['create_time'] = bk['create_time'].apply(
# lambda x: time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(x)))
# bk.columns = self.field_list # 修改pandas头
# write_data = bk.to_dict(orient='records')
# with pd.ExcelWriter(f'static/{self.name}.xlsx') as writer:
# # bk.to_excel(writer, sheet_name='Sheet1', index=False)
# threads = []
# rows_per_thread = math.ceil(len(write_data) / 5)
# for i in range(5):
# sheet_name = 'sheet' + str(i + 1)
# threads.append(threading.Thread(target=self.thread_task,
# args=[bk.iloc[i * rows_per_thread: rows_per_thread * (i+1)], writer, sheet_name]))
# # 启动线程
# for y in threads:
# y.start()
# # 等待所有线程完成
# for z in threads:
# z.join()
# # 记录导出
# crud.create_export_data(self.db, params, user)
# import zipfile
# with zipfile.ZipFile('ceshi.zip', 'w') as zip:
# # 将指定文件添加到压缩文件中
# zip.write(f"static/{self.name}.xlsx")
# return
# except Exception as e:
# Logger().logger.info(f'导出异常:{str(e)}')
# params["status"] = 2
# crud.create_export_data(self.db, params, user)
def main_method(self):
"""主函数"""
Logger().logger.info('开始导出')
......@@ -121,6 +258,9 @@ class TableToFile(object):
write_data = bk.to_dict(orient='records')
with pd.ExcelWriter(f'static/{self.name}.xlsx') as writer:
# bk.to_excel(writer, sheet_name='Sheet1', index=False)
if len(self.data) < 500:
bk.to_excel(writer, sheet_name='sheet', index=False)
else:
threads = []
rows_per_thread = math.ceil(len(write_data) / 5)
for i in range(5):
......@@ -133,13 +273,14 @@ class TableToFile(object):
# 等待所有线程完成
for z in threads:
z.join()
file = open(writer, 'rb')
SITE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LinkMonitor().upload_block(f'{SITE_ROOT}/static/{self.name}.xlsx', self.name, 'finance/')
# 记录导出
crud.create_export_data(self.db, params, user)
return StreamingResponse(file,
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
except Exception as e:
Logger().logger.info(f'导出异常:{str(e)}')
params["status"] = 2
crud.create_export_data(self.db, params, user)
return None
return f"https://{env.TX_BUCKET}.cos.ap-guangzhou.myqcloud.com/finance/{self.name}.xlsx"
......@@ -266,7 +266,8 @@ def AES_Decrypt(data):
Logger(40).logger.error(f"php数据解密异常:{str(e)},数据:{plaintext}")
coding_data = str(plaintext, encoding="utf-8")
num = coding_data.index(']')
return list(eval(coding_data[:num + 1]))
missing_value = coding_data[:num + 1]
return list(eval(missing_value.replace('null', '""')))
return res_data
......
......@@ -13,20 +13,20 @@ session = requests.session()
class imageCode():
'''
"""
验证码处理
'''
"""
def rndColor(self):
'''随机颜色'''
"""随机颜色"""
return (random.randint(32, 127), random.randint(32, 127), random.randint(32, 127))
def geneText(self):
'''生成4位验证码'''
"""生成4位验证码"""
return ''.join(random.sample(string.ascii_letters + string.digits, 4)) # ascii_letters是生成所有字母 digits是生成所有数字0-9
def drawLines(self, draw, num, width, height):
'''划线'''
"""划线"""
for num in range(num):
x1 = random.randint(0, width / 2)
y1 = random.randint(0, height / 2)
......@@ -35,7 +35,7 @@ class imageCode():
draw.line(((x1, y1), (x2, y2)), fill='black', width=1)
def getVerifyCode(self):
'''生成验证码图形'''
"""生成验证码图形"""
code = self.geneText()
# 图片大小120×50
width, height = 120, 50
......@@ -67,9 +67,9 @@ class imageCode():
return img.decode('utf-8')
def new_upload_file(file_object,filename):
'''图片上传cos'''
filename = COS_RERURN_PATH+filename+'.png'
def new_upload_file(file_object, filename):
"""图片上传cos"""
filename = COS_RERURN_PATH + filename + '.png'
try:
response = client.put_object(
Bucket=Bucket,
......
......@@ -23,31 +23,10 @@ app.add_middleware(
allow_headers=['*']) # 允许跨域的headers,可以用来鉴别来源等作用。
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
hs = request.headers
token = hs.get("authorization")
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
url_list = ['/api/users/imgCode', '/api/users/login', '/api/users/goodleCode', '/api/users/googleLogin']
if request.url.path in url_list:
return response
if token:
try:
payload = jwt.decode(token.replace('Bearer','').replace(' ',''), env.SECRET_KEY, algorithms=[env.ALGORITHM])
except Exception as e:
print(e)
return response
timestamp = payload.get("exp")
access_token_expires = timedelta(hours=time_format(timestamp))
create_access_token({'username':payload.get("xup"),'password':payload.get("password")},expires_delta=access_token_expires) #更新token时间
return response
app.include_router(api_router, prefix="/api") # 路由
if __name__ == '__main__':
create_yaml()
uvicorn.run(app=app, host="0.0.0.0", port=8009)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment