zayac/tools-pyqt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

结构调整

Browse Source
This commit is contained in:
zayac 2024-01-21 17:04:13 +08:00
parent a0a03b61d0
commit c1a143aa8e
16 changed files with 356 additions and 332 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
README.md
View File

@ -1,12 +1,14 @@
# tools-ui
图形化界面版本的小工具
$env:PLAYWRIGHT_BROWSERS_PATH="0"
# tools-ui
图形化界面版本的小工具
$env:PLAYWRIGHT_BROWSERS_PATH="0"
pyinstaller -F -w -i .\src\ui\icon\icon.ico .\src\app.py --hidden-import plyer.platforms.win.notification --add-data "./src/ui/icon;ui/icon/" --add-data "./src/ui/style.qss;ui/"
```
tools_pyqt
├─ .git
│ ├─ HEAD

17
requirements.txt
View File

@ -1 +1,16 @@
aiohttp==3.9.1 loguru==0.7.2 pika==1.3.2 playwright==1.40.0 pyperclip==1.8.2 PyQt6==6.6.1 PyQt6_sip==13.6.0 python_dateutil==2.8.2 QDarkStyle==3.2.3 Requests==2.31.0 schedule==1.2.1 SQLAlchemy==2.0.25 typing_extensions==4.9.0
aiohttp==3.9.1
cachetools==5.3.2
loguru==0.7.2
pika==1.3.2
playwright==1.40.0
PyQt6==6.6.1
PyQt6_sip==13.6.0
pyTelegramBotAPI==4.15.2
python_dateutil==2.8.2
qdarkstyle==3.2.3
qtpy==2.4.1
requests==2.31.0
schedule==1.2.1
SQLAlchemy==2.0.25
typing_extensions==4.9.0
mysql-connector-python==8.3.0

22
src/bot/bot.py
View File

@ -2,10 +2,13 @@ import schedule
import telebot
import time
import threading
from loguru import logger
from telebot import types
from src.core.报数 import text_count_by_telegram_id, get_net_win_by_telegram_id
from src.core.查询存款失败用户 import get_pay_failed_by_telegram_id
from src.entity.finance import get_net_win_by_telegram_id
from src.entity.visual_list import text_count_by_telegram_id
from src.entity.pay_record import get_pay_failed_by_telegram_id
from src.entity.user import get_all_users
TOKEN = '6013830443:AAGzq1Tgtr_ZejU7bv0mab14xOwi0_64d0w'
@ -103,5 +106,16 @@ def handle_message(message):
bot.send_message(message.chat.id, response)
schedule_user_updates() # Schedule the user updates
bot.polling() # Start polling
def start_polling():
while True:
try:
bot.polling(none_stop=True)
except Exception as e:
logger.error(f"长轮询异常: {e}")
# 在重新尝试前暂停一段时间
time.sleep(15)
if __name__ == "__main__":
schedule_user_updates() # 定时更新用户
start_polling() # 开始长轮询

2
src/change_url/change_url_bot.py
View File

@ -45,4 +45,4 @@ def handle_forwarded_message(message):
bot.reply_to(message, msg)
bot.polling()
bot.polling(none_stop=True)

62
src/core/salary.py
View File

@ -1,62 +0,0 @@
from decimal import Decimal
from src.core.util import (get_first_day_by_str, get_first_day_of_last_month,
get_last_day_of_last_month)
from src.entity.database import db
from src.entity.finance import get_finance
from src.entity.user import User
def calculate_commission(profit, employee_type, target_completion):
# 定义提成点位
commission_rates = {
"1": [(0, 0.03), (100001, 0.08), (300001, 0.10), (500001, 0.11), (700001, 0.12), (1000001, 0.13)],
"2": [(0, 0.02), (100001, 0.05), (300001, 0.07), (500001, 0.08), (700001, 0.09), (1000001, 0.10)],
"3": [(0, 0.02), (100001, 0.03), (300001, 0.04), (500001, 0.05), (700001, 0.06), (1000001, 0.07)]
}
# 根据负盈利选择正确的提成点位
rates = commission_rates[employee_type]
rate = 0
for r in rates:
if profit >= r[0]:
rate = r[1]
else:
break
# 计算提成
commission = profit * target_completion * Decimal(str(rate))
return commission
def calculate_salary(employee_type, agent_profit):
if employee_type == "3":
base_salary = 30000
else:
base_salary = 12000
total_salary = 0
for profit in agent_profit:
total_salary += calculate_commission(profit, employee_type, 1)
if total_salary > 70000:
base_salary = 0
return total_salary + base_salary
def get_salary(user: User, date: str):
profits = []
start_date = get_first_day_by_str(date)
for account in user.accounts:
finance = get_finance(account, start_date, date)
print(f'{finance.name}: {finance.netProfit}')
profits.append(int(float(finance.netProfit)))
return f'方式一：{calculate_salary("1", profits)}\n方式二：{calculate_salary("2", profits)}\n方式三：{calculate_salary("3", profits)}'
def get_last_month_salary(user: User):
profits = []
for account in user.accounts:
finance = get_finance(account, get_first_day_of_last_month(), get_last_day_of_last_month())
print(f'{finance.name}: {finance.netProfit}')
profits.append(int(float(finance.netProfit)))
return f'方式一：{calculate_salary("1", profits)}\n方式二：{calculate_salary("2", profits)}\n方式三：{calculate_salary("3", profits)}'

64
src/core/喜报.py
View File

@ -1,64 +0,0 @@
import time
from concurrent.futures.thread import ThreadPoolExecutor
from typing import List
from loguru import logger
from util import get_curr_day
from src.core.constant import BOT_TOKEN, GROUP_ID
from src.core.message_client import send_message
from src.entity.account import Account
from src.entity.banner_info import BannerInfo, get_banner_info
from src.entity.database import db
from src.entity.member import get_today_new_member_list
from src.entity.pay_record import get_latest_deposit_user
from src.entity.user import User
def query_banner_info(account: Account):
last_banner_info = get_banner_info(account)
while True:
try:
date = get_curr_day()
banner_info = get_banner_info(account)
logger.debug(f'{account.name}请求成功：{banner_info}')
logger.info(
f'{time.strftime("%Y-%m-%d %H:%M:%S")} {account.name}：注册：{banner_info.registerMembers},首存：{banner_info.firstDepositNum},负盈利：{banner_info.netWinLose},有效：{banner_info.effectiveNew},活跃：{banner_info.activeMembers}')
if banner_info.registerMembers > last_banner_info.registerMembers:
register_count = banner_info.registerMembers - last_banner_info.registerMembers
logger.debug(f'新注册用户数为 {register_count}')
members = get_today_new_member_list(account, register_count)
if members is not None:
names = ','.join([f'`{member.name}`' for member in members])
else:
names = 'unknown'
msg = f'👏 {account.name} 注册:{register_count} 用户: {names} 总数: {banner_info.registerMembers}'
send_message(BOT_TOKEN, GROUP_ID, msg)
logger.info(f'发送的消息: {msg}')
last_banner_info = banner_info
if banner_info.firstDepositNum > last_banner_info.firstDepositNum:
count = banner_info.firstDepositNum - last_banner_info.firstDepositNum
member_details_list = get_latest_deposit_user(account, count)
msg = '\n'.join(
[f"用户: `{member_detail.name}`, 首存金额: *{member_detail.deposit}*" for member_detail in
member_details_list])
send_message(BOT_TOKEN, GROUP_ID,
f'🎉 {account.name} 首存:{count} {msg} 总数:*{banner_info.firstDepositNum}*')
logger.info(f'发送的消息: {msg}')
last_banner_info = banner_info
time.sleep(60)
except Exception as e:
send_message(BOT_TOKEN, GROUP_ID, str(e))
logger.exception(f'发生未知错误:{e} ')
time.sleep(10)
return query_banner_info(account)
def get_banner_info_by_user(user: User) -> List[BannerInfo]:
with ThreadPoolExecutor(max_workers=len(user.accounts)) as executor:
futures = [executor.submit(get_banner_info, account) for account in user.accounts]
return [future.result() for future in futures]

9
src/core/定时任务.py
View File

@ -2,12 +2,15 @@ import time
import schedule
from loguru import logger
from 报数 import get_net_win_by_user, text_count_by_user
from 查询存款失败用户 import get_pay_failed_by_user
from src.core.util import get_curr_day
from src.entity.finance import get_net_win_by_user
from src.entity.pay_record import get_pay_failed_by_user
from src.core.constant import BOT_TOKEN, COUNT_GROUP_ID
from src.core.message_client import send_message
from src.entity.user import get_user_by_username_and_password
from src.entity.visual_list import text_count_by_user
def job_count(username, password):
@ -27,7 +30,7 @@ def query_failed_deposit(username, password):
def query_net_win(username, password) -> None:
logger.info(f'Running query_net_win for username: {username}')
user = get_user_by_username_and_password(username, password)
send_message(BOT_TOKEN, COUNT_GROUP_ID, get_net_win_by_user(user))
send_message(BOT_TOKEN, COUNT_GROUP_ID, get_net_win_by_user(user,date=get_curr_day()))
logger.info(f'Finished query_net_win for username: {username}')

72
src/core/报数.py
View File

@ -1,72 +0,0 @@
from concurrent.futures import ThreadPoolExecutor
from typing import List
import pyperclip
from src import logger
from src.core import util
from src.core.util import get_curr_day
from src.entity.database import db
from src.entity.finance import Finance, get_finance
from src.entity.user import User, get_user_by_telegram_id
from src.entity.visual_list import VisualInfo, get_visual_list
def get_statics(account, date=util.get_curr_day()) -> VisualInfo:
params = {"monthDate": util.get_curr_month()}
data = get_visual_list(account, params)
# 合并列表并创建日期到数据的映射
date_map = {item.staticsDate: item for item in data.curData + data.lastData}
# 直接通过日期获取数据
return date_map.get(date)
def count_by_user(user: User, date: str):
accounts = user.accounts
with ThreadPoolExecutor(max_workers=len(accounts)) as t:
futures = [t.submit(get_statics, account, date) for account in accounts]
return [future.result() for future in futures]
def text_count_by_user(user: User, date: str) -> str:
visual_list = count_by_user(user, date)
text = '\n\n'.join(
f'{result.agentName}\n注册：{result.isNew}\n首存：{result.firstCount}\n日活：{int(result.countBets)}\n流水：{int(result.bets)}'
for result in visual_list
)
logger.info(f'Generated text: {text}')
return text
def text_count_by_telegram_id(telegram_id: int) -> str:
visual_list = count_by_user(get_user_by_telegram_id(telegram_id), get_curr_day())
text = '\n\n'.join(
f'{result.agentName}\n注册：{result.isNew}\n首存：{result.firstCount}\n日活：{int(result.countBets)}\n流水：{int(result.bets)}'
for result in visual_list
)
logger.info(f'Generated text: {text}')
return text
def get_finances_by_user(user: User, date) -> List[Finance]:
accounts = user.accounts
start_date = util.get_first_day_by_str(date)
with ThreadPoolExecutor(max_workers=len(accounts)) as t:
futures = [t.submit(get_finance, account, start_date, date) for account in accounts]
return [future.result() for future in futures]
def get_net_win_by_user(user: User, date: str) -> str:
finances = get_finances_by_user(user, date)
finance_strings = [f"{finance.name}: {finance.netProfit}" for finance in finances]
logger.info(f'Finance strings: {finance_strings}')
return '\n'.join(finance_strings)
def get_net_win_by_telegram_id(telegram_id: int) -> str:
user = get_user_by_telegram_id(telegram_id)
finances = get_finances_by_user(user, get_curr_day())
finance_strings = [f"{finance.name}: {finance.netProfit}" for finance in finances]
logger.info(f'Finance strings: {finance_strings}')
return '\n'.join(finance_strings)

103
src/core/查询存款失败用户.py
View File

@ -1,103 +0,0 @@
from concurrent.futures import ThreadPoolExecutor
from typing import Dict, List, Optional
from loguru import logger
from src.core import api_request
from src.core.constant import PAY_RECORD_LIST_URL
from src.core.util import get_first_day_month, get_curr_day
from src.entity.account import Account
from src.entity.member import MemberList, get_member_list
from src.entity.user import User, get_user_by_telegram_id
def get_pay_record_list(account: Account, date: str) -> Dict[str, List[str]]:
logger.info(f'Getting pay record list for account: {account.name} and date: {date}')
_names = {'name': account.name, 'names': []}
params = {
"pageNum": 1,
"pageSize": 100,
"registerSort": 1,
"drawSort": -1,
"depositSort": -1,
"lastLoginTimeSort": -1,
"name": "",
"minPay": None,
"maxPay": None,
"startDate": get_first_day_month(),
"registerStartDate": date,
"endDate": date,
"registerEndDate": date,
"firstPayStartTime": "",
"firstPayEndTime": "",
"isBet": "0",
"tagsFlag": "1"
}
member_list = get_member_list(account, params)
if member_list is not None and len(member_list) > 0:
with ThreadPoolExecutor(max_workers=len(member_list)) as executor:
futures = [executor.submit(get_pay_record, account, member, date) for member in member_list]
for future in futures:
result = future.result()
if result:
_names['names'].append(result)
logger.info(f'Finished getting pay record list for account: {account.name} and date: {date}')
return _names
def get_pay_record(account: Account, member: MemberList, date: str) -> Optional[str]:
logger.info(f'Getting pay record for account: {account.name}, member: {member.name}, and date: {date}')
params = {
"pageNum": 1,
"pageSize": 15,
"id": member.id,
"startDate": get_first_day_month(),
"endDate": date
}
res = api_request.account_post(PAY_RECORD_LIST_URL, account=account, params=params)
if int(res.data['orderAmountTotal']) > 0 and int(res.data['scoreAmountTotal']) == 0:
return member.name
return ""
def get_pay_failed_by_user(user: User, date: str) -> Optional[str]:
logger.info(f'Getting pay failed by user: {user.username}')
with ThreadPoolExecutor(max_workers=len(user.accounts)) as executor:
futures = [executor.submit(get_pay_record_list, account, date) for account in user.accounts]
# 使用列表推导式构建结果字符串
text_lines = [
"{}\n{}".format(res['name'], '\n'.join(res['names']))
for future in futures if (res := future.result())['names']
]
text = '\n'.join(text_lines)
if not text:
logger.info('无存款失败用户')
return '无存款失败用户'
logger.info(text)
return text
def get_pay_failed_by_telegram_id(telegram_id: int) -> Optional[str]:
user = get_user_by_telegram_id(telegram_id)
with ThreadPoolExecutor(max_workers=len(user.accounts)) as executor:
futures = [executor.submit(get_pay_record_list, account, get_curr_day()) for account in user.accounts]
# 使用列表推导式构建结果字符串
text_lines = [
"{}\n{}".format(res['name'], '\n'.join(res['names']))
for future in futures if (res := future.result())['names']
]
text = '\n'.join(text_lines)
if not text:
logger.info('无存款失败用户')
return '无存款失败用户'
logger.info(text)
return text

2
src/entity/__init__.py
View File

@ -1,3 +1 @@
from src import logger
from . import account, user

1
src/entity/account.py
View File

@ -1,4 +1,3 @@
import json
from dataclasses import dataclass
from enum import Enum

60
src/entity/banner_info.py
View File

@ -1,8 +1,17 @@
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import List
from src.core.api_request import account_post
from src.core.constant import BANNER_URL
from src.core.constant import BANNER_URL, BOT_TOKEN, GROUP_ID
from src.core.message_client import send_message
from src.core.util import get_curr_day
from src.entity.account import Account
import time
from src.entity.member import get_today_new_member_list
from src.entity.pay_record import get_latest_deposit_user
from src.entity.user import User
@dataclass
@ -30,3 +39,52 @@ def get_banner_info(account: Account) -> BannerInfo:
banner_info = BannerInfo(**api_response.data)
banner_info.agentCode = account.username
return banner_info
def query_banner_info(account: Account):
last_banner_info = get_banner_info(account)
while True:
try:
date = get_curr_day()
banner_info = get_banner_info(account)
from loguru import logger
logger.debug(f'{account.name}请求成功：{banner_info}')
logger.info(
f'{time.strftime("%Y-%m-%d %H:%M:%S")} {account.name}：注册：{banner_info.registerMembers},首存：{banner_info.firstDepositNum},负盈利：{banner_info.netWinLose},有效：{banner_info.effectiveNew},活跃：{banner_info.activeMembers}')
if banner_info.registerMembers > last_banner_info.registerMembers:
register_count = banner_info.registerMembers - last_banner_info.registerMembers
logger.debug(f'新注册用户数为 {register_count}')
members = get_today_new_member_list(account, register_count)
if members is not None:
names = ','.join([f'`{member.name}`' for member in members])
else:
names = 'unknown'
msg = f'👏 {account.name} 注册:{register_count} 用户: {names} 总数: {banner_info.registerMembers}'
send_message(BOT_TOKEN, GROUP_ID, msg)
logger.info(f'发送的消息: {msg}')
last_banner_info = banner_info
if banner_info.firstDepositNum > last_banner_info.firstDepositNum:
count = banner_info.firstDepositNum - last_banner_info.firstDepositNum
member_details_list = get_latest_deposit_user(account, count)
msg = '\n'.join(
[f"用户: `{member_detail.name}`, 首存金额: *{member_detail.deposit}*" for member_detail in
member_details_list])
send_message(BOT_TOKEN, GROUP_ID,
f'🎉 {account.name} 首存:{count} {msg} 总数:*{banner_info.firstDepositNum}*')
logger.info(f'发送的消息: {msg}')
last_banner_info = banner_info
time.sleep(60)
except Exception as e:
send_message(BOT_TOKEN, GROUP_ID, str(e))
logger.exception(f'发生未知错误:{e} ')
time.sleep(10)
return query_banner_info(account)
def get_banner_info_by_user(user: User) -> List[BannerInfo]:
with ThreadPoolExecutor(max_workers=len(user.accounts)) as executor:
futures = [executor.submit(get_banner_info, account) for account in user.accounts]
return [future.result() for future in futures]

101
src/entity/finance.py
View File

@ -1,10 +1,16 @@
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from decimal import Decimal
from typing import List
from loguru import logger
from src.core import util
from src.core.api_request import account_post
from src.core.constant import FINANCE_URL
from src.core.util import get_curr_day, get_first_day_by_str
from src.entity.account import Account
from src.entity.user import User, get_user_by_telegram_id
'''
财务报表
@ -45,3 +51,98 @@ def get_finance(account: Account, start_date=util.get_first_day_month(), end_dat
finance = Finance(**api_response.data)
finance.name = account.name
return finance
def get_finances_by_user(user: User, date) -> List[Finance]:
accounts = user.accounts
start_date = util.get_first_day_by_str(date)
with ThreadPoolExecutor(max_workers=len(accounts)) as t:
futures = [t.submit(get_finance, account, start_date, date) for account in accounts]
return [future.result() for future in futures]
def get_net_win_by_user(user: User, date: str) -> str:
finances = get_finances_by_user(user, date)
finance_strings = [f"{finance.name}: {finance.netProfit}" for finance in finances]
logger.info(f'Finance strings: {finance_strings}')
return '\n'.join(finance_strings)
def get_net_win_by_telegram_id(telegram_id: int) -> str:
user = get_user_by_telegram_id(telegram_id)
finances = get_finances_by_user(user, get_curr_day())
finance_strings = [f"{finance.name}: {finance.netProfit}" for finance in finances]
logger.info(f'Finance strings: {finance_strings}')
return '\n'.join(finance_strings)
def calculate_commission(profit, employee_type, target_completion):
# 定义提成点位
commission_rates = {
"1": [(0, 0.02), (100001, 0.05), (300001, 0.07), (500001, 0.08), (700001, 0.09), (1000001, 0.10)],
"2": [(0, 0.02), (100001, 0.03), (300001, 0.04), (500001, 0.05), (700001, 0.06), (1000001, 0.07)]
}
# 根据负盈利选择正确的提成点位
rates = commission_rates[employee_type]
rate = 0
for r in rates:
if profit >= r[0]:
rate = r[1]
else:
break
# 计算提成
commission = profit * target_completion * Decimal(str(rate))
return commission
def adjust_profits(profits):
# 新增 adjust_profits 函数处理销售额的负数抵扣逻辑
positive_profits = [p for p in profits if p > 0]
negative_profits = [p for p in profits if p < 0]
for neg_profit in negative_profits:
for i, pos_profit in enumerate(positive_profits):
if pos_profit + neg_profit > 0:
positive_profits[i] += neg_profit
neg_profit = 0
break
else:
neg_profit += pos_profit
positive_profits[i] = 0
if neg_profit != 0:
# 当所有账户都是负数时
return [0]
return positive_profits + negative_profits
def calculate_adjusted_salary(user: User, date: str, employee_type):
profits = []
start_date = get_first_day_by_str(date)
for account in user.accounts:
finance = get_finance(account, start_date, date)
print(f'{finance.name}: {finance.netProfit}')
profits.append(int(float(finance.netProfit)))
adjusted_profits = adjust_profits(profits)
return calculate_salary(employee_type, adjusted_profits)
def calculate_salary(employee_type, agent_profit):
if employee_type == "2":
base_salary = 10000
else:
base_salary = 12000
total_salary = 0
for profit in agent_profit:
total_salary += calculate_commission(profit, employee_type, 1)
if total_salary > 70000:
base_salary = 0
return total_salary + base_salary
def get_adjusted_salary(user: User, date: str):
return f'方式一：{calculate_adjusted_salary(user, date, "1")}\n方式二：{calculate_adjusted_salary(user, date, "2")}'

104
src/entity/pay_record.py
View File

@ -2,14 +2,16 @@ import asyncio
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List, Optional
from src import logger
from src.core import api_request, util
from src.core.constant import PAY_RECORD_URL
from src.core.util import get_curr_day
from src.core.constant import PAY_RECORD_URL, PAY_RECORD_LIST_URL
from src.core.util import get_curr_day, get_first_day_month
from src.entity.account import Account
from src.entity.member import (async_get_member_detail_by_name,
get_member_by_name)
get_member_by_name, get_member_list, MemberList)
from src.entity.user import User, get_user_by_telegram_id
@dataclass
@ -32,7 +34,7 @@ class PayRecord(object):
# 根据用户查询最新存款信息
def get_pay_record_list(account: Account):
def get_pay_record(account: Account):
logger.info(f'Getting pay record list for account: {account.name}')
# 获取当前成功存款的用户
params = {
@ -51,7 +53,7 @@ def get_pay_record_list(account: Account):
def get_latest_deposit_user(account: Account, count: int):
logger.info(f'Getting latest deposit user for account: {account.name} and count: {count}')
pay_record_list = get_pay_record_list(account)
pay_record_list = get_pay_record(account)
# 提取所有用户名
names = []
seen = set()
@ -133,3 +135,95 @@ async def async_get_latest_deposit_user(account: Account, count: int):
detail.deposit = record_dict[detail.firstPayAt]
logger.info(f'Finished async getting latest deposit user for account: {account.name} and count: {count}')
return details
def get_pay_record_list(account: Account, date: str) -> Dict[str, List[str]]:
logger.info(f'Getting pay record list for account: {account.username} and date: {date}')
_names = {'name': account.username, 'names': []}
params = {
"pageNum": 1,
"pageSize": 100,
"registerSort": 1,
"drawSort": -1,
"depositSort": -1,
"lastLoginTimeSort": -1,
"name": "",
"minPay": None,
"maxPay": None,
"startDate": get_first_day_month(),
"registerStartDate": date,
"endDate": date,
"registerEndDate": date,
"firstPayStartTime": "",
"firstPayEndTime": "",
"isBet": "0",
"tagsFlag": "1"
}
member_list = get_member_list(account, params)
if member_list is not None and len(member_list) > 0:
with ThreadPoolExecutor(max_workers=len(member_list)) as executor:
futures = [executor.submit(get_pay_record_detail, account, member, date) for member in member_list]
for future in futures:
result = future.result()
if result:
_names['names'].append(result)
logger.info(f'Finished getting pay record list for account: {account.name} and date: {date}')
return _names
def get_pay_record_detail(account: Account, member: MemberList, date: str) -> Optional[str]:
logger.info(f'Getting pay record for account: {account.name}, member: {member.name}, and date: {date}')
params = {
"pageNum": 1,
"pageSize": 15,
"id": member.id,
"startDate": get_first_day_month(),
"endDate": date
}
res = api_request.account_post(PAY_RECORD_LIST_URL, account=account, params=params)
if int(res.data['orderAmountTotal']) > 0 and int(res.data['scoreAmountTotal']) == 0:
return member.name
return ""
def get_pay_failed_by_user(user: User, date: str) -> Optional[str]:
logger.info(f'Getting pay failed by user: {user.username}')
with ThreadPoolExecutor(max_workers=len(user.accounts)) as executor:
futures = [executor.submit(get_pay_record_list, account, date) for account in user.accounts]
# 使用列表推导式构建结果字符串
text_lines = [
"{}\n{}".format(res['name'], '\n'.join(res['names']))
for future in futures if (res := future.result())['names']
]
text = '\n'.join(text_lines)
if not text:
logger.info('无存款失败用户')
return '无存款失败用户'
logger.info(text)
return text
def get_pay_failed_by_telegram_id(telegram_id: int) -> Optional[str]:
user = get_user_by_telegram_id(telegram_id)
with ThreadPoolExecutor(max_workers=len(user.accounts)) as executor:
futures = [executor.submit(get_pay_record_list, account, get_curr_day()) for account in user.accounts]
# 使用列表推导式构建结果字符串
text_lines = [
"{}\n{}".format(res['name'], '\n'.join(res['names']))
for future in futures if (res := future.result())['names']
]
text = '\n'.join(text_lines)
if not text:
logger.info('无存款失败用户')
return '无存款失败用户'
logger.info(text)
return text

42
src/entity/visual_list.py
View File

@ -1,9 +1,14 @@
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from loguru import logger
from src.core import api_request
from src.core.constant import VISUAL_LIST_URL
from src.core.util import get_curr_day, get_curr_month
from src.entity.account import Account
from src.entity.user import User, get_user_by_telegram_id
# 视图列表对象 对应界面上的图表
@ -83,3 +88,40 @@ async def async_get_curr_data(account: Account, params: dict) -> list[VisualInfo
res = await api_request.async_account_post(url=VISUAL_LIST_URL, account=account,
params=params)
return [VisualInfo(**item) for item in res.data['curData']]
def get_statics(account, date=get_curr_day()) -> VisualInfo:
params = {"monthDate": get_curr_month()}
data = get_visual_list(account, params)
# 合并列表并创建日期到数据的映射
date_map = {item.staticsDate: item for item in data.curData + data.lastData}
# 直接通过日期获取数据
return date_map.get(date)
def count_by_user(user: User, date: str):
accounts = user.accounts
with ThreadPoolExecutor(max_workers=len(accounts)) as t:
futures = [t.submit(get_statics, account, date) for account in accounts]
return [future.result() for future in futures]
def text_count_by_user(user: User, date: str) -> str:
visual_list = count_by_user(user, date)
text = '\n\n'.join(
f'{result.agentName}\n注册：{result.isNew}\n首存：{result.firstCount}\n日活：{int(result.countBets)}\n流水：{int(result.bets)}'
for result in visual_list
)
logger.info(f'Generated text: {text}')
return text
def text_count_by_telegram_id(telegram_id: int) -> str:
visual_list = count_by_user(get_user_by_telegram_id(telegram_id), get_curr_day())
text = '\n\n'.join(
f'{result.agentName}\n注册：{result.isNew}\n首存：{result.firstCount}\n日活：{int(result.countBets)}\n流水：{int(result.bets)}'
for result in visual_list
)
logger.info(f'Generated text: {text}')
return text

11
src/ui/data_query.py
View File

@ -1,11 +1,10 @@
import time
from PyQt6.QtCore import QObject, QRunnable, QThread, pyqtSignal
from src.core.salary import get_salary
from src.core.报数 import get_net_win_by_user, text_count_by_user
from src.core.查询存款失败用户 import get_pay_failed_by_user
from PyQt6.QtCore import QObject, QRunnable, pyqtSignal
from src.entity.banner_info import get_banner_info
from src.entity.finance import get_net_win_by_user, get_adjusted_salary
from src.entity.pay_record import get_pay_failed_by_user
from src.entity.visual_list import text_count_by_user
class TaskSignals(QObject):
@ -56,7 +55,7 @@ class ButtonTask(QRunnable):
elif self.query_type == '负盈利':
result = get_net_win_by_user(self.user, self.selected_date_str)
elif self.query_type == '薪资':
result = get_salary(self.user, self.selected_date_str)
result = get_adjusted_salary(self.user, self.selected_date_str)
# 数据查询完成，发出信号
self.signals.query_completed.emit(result, auto_clipboard, notify)