Genshin-bots/gsuid_core
1
0
Fork 0
mirror of https://github.com/Genshin-bots/gsuid_core.git synced 2025-05-12 06:55:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

🔥 移除一部分get_sqla语法

Browse Source
This commit is contained in:
KimigaiiWuyi 2023-09-21 00:50:23 +08:00
parent 9ffa131699
commit d2724be13b
8 changed files with 109 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
gsuid_core/plugins/core_command/core_user/draw_user_card.py
View File

@ -3,10 +3,9 @@ from typing import Tuple, Union, Optional
from PIL import Image, ImageDraw
from gsuid_core.utils.database.api import DBSqla
from gsuid_core.utils.fonts.fonts import core_font
from gsuid_core.utils.database.models import GsPush
from gsuid_core.utils.image.convert import convert_img
from gsuid_core.utils.database.models import GsBind, GsPush, GsUser
from gsuid_core.utils.image.image_tools import (
get_color_bg,
get_qq_avatar,
@ -23,11 +22,9 @@ EN_MAP = {'coin': '宝钱', 'resin': '体力', 'go': '派遣', 'transform': '质
async def get_user_card(bot_id: str, user_id: str) -> Union[bytes, str]:
get_sqla = DBSqla().get_sqla
sqla = get_sqla(bot_id)
uid_list = await sqla.get_bind_uid_list(user_id)
sr_uid_list = await sqla.get_bind_sruid_list(user_id)
user_list = await sqla.select_user_all_data_by_user_id(user_id)
uid_list = await GsBind.get_uid_list_by_game(user_id, bot_id)
sr_uid_list = await GsBind.get_uid_list_by_game(user_id, bot_id, 'sr')
user_list = await GsUser.get_user_all_data_by_user_id(user_id)
if user_list is None:
return '你还没有绑定过UID和CK!\n(该功能须同时绑定CK和UID才能使用)'
@ -65,10 +62,25 @@ async def get_user_card(bot_id: str, user_id: str) -> Union[bytes, str]:
if user_data.uid is not None and user_data.uid != '0':
uid_text = f'原神UID {user_data.uid}'
user_push_data = await sqla.select_push_data(user_data.uid)
user_push_data = await GsPush.select_data_by_uid(user_data.uid)
if user_push_data is None:
await sqla.insert_push_data(user_data.uid)
user_push_data = await sqla.select_push_data(user_data.uid)
await GsPush.full_insert_data(
bot_id=bot_id,
uid=user_data.uid,
coin_push='off',
coin_value=2100,
coin_is_push='off',
resin_push='on',
resin_value=140,
resin_is_push='off',
go_push='off',
go_value=120,
go_is_push='off',
transform_push='off',
transform_value=140,
transform_is_push='off',
)
user_push_data = await GsPush.select_data_by_uid(user_data.uid)
else:
uid_text = '未发现原神UID'
user_push_data = GsPush(bot_id='TEMP')

11
gsuid_core/utils/api/mys/request.py
View File

@ -15,6 +15,7 @@ from aiohttp import TCPConnector, ClientSession, ContentTypeError
from gsuid_core.logger import logger
from gsuid_core.utils.database.api import DBSqla
from gsuid_core.utils.database.models import GsUser
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
from .api import _API
@ -316,9 +317,10 @@ class BaseMysApi:
# 针对1034做特殊处理
if retcode == 1034:
if uid and self.is_sr and _ == 0:
sqla = self.dbsqla.get_sqla('TEMP')
new_fp = await self.generate_fp_by_uid(uid)
await sqla.update_user_data(uid, {'fp': new_fp})
await GsUser.update_data_by_uid_without_bot_id(
uid, fp=new_fp
)
header['x-rpc-device_fp'] = new_fp
if isinstance(params, Dict):
header['DS'] = get_ds_token(
@ -330,9 +332,10 @@ class BaseMysApi:
ch = await self._upass(header)
self.chs[header['Cookie']] = ch
elif retcode == -10001 and uid:
sqla = self.dbsqla.get_sqla('TEMP')
new_fp = await self.generate_fp_by_uid(uid)
await sqla.update_user_data(uid, {'fp': new_fp})
await GsUser.update_data_by_uid_without_bot_id(
uid, fp=new_fp
)
header['x-rpc-device_fp'] = new_fp
elif retcode != 0:
return retcode

19
gsuid_core/utils/api/mys_api.py
View File

@ -1,6 +1,7 @@
from typing import Literal, Optional
from gsuid_core.utils.api.mys import MysApi
from gsuid_core.utils.database.models import GsUser
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
gsconfig = core_plugins_config
@ -14,29 +15,25 @@ class _MysApi(MysApi):
self, uid: str, mode: Literal['OWNER', 'RANDOM'] = 'RANDOM'
) -> Optional[str]:
if mode == 'RANDOM':
return await self.dbsqla.get_sqla('TEMP').get_random_cookie(uid)
return await GsUser.get_random_cookie(uid)
else:
return await self.dbsqla.get_sqla('TEMP').get_user_cookie(uid)
return await GsUser.get_user_cookie_by_uid(uid)
async def get_stoken(self, uid: str) -> Optional[str]:
return await self.dbsqla.get_sqla('TEMP').get_user_stoken(uid)
return await GsUser.get_user_stoken_by_uid(uid)
async def get_user_fp(self, uid: str) -> Optional[str]:
data = await self.dbsqla.get_sqla('TEMP').get_user_fp(uid)
data = await GsUser.get_user_attr_by_uid(uid, 'fp')
if data is None:
data = await self.generate_fp_by_uid(uid)
await self.dbsqla.get_sqla('TEMP').update_user_data(
uid, {'fp': data}
)
await GsUser.update_data_by_uid_without_bot_id(uid, fp=data)
return data
async def get_user_device_id(self, uid: str) -> Optional[str]:
data = await self.dbsqla.get_sqla('TEMP').get_user_device_id(uid)
data = await GsUser.get_user_attr_by_uid(uid, 'device_id')
if data is None:
data = self.get_device_id()
await self.dbsqla.get_sqla('TEMP').update_user_data(
uid, {'device_id': data}
)
await GsUser.update_data_by_uid_without_bot_id(uid, device_id=data)
return data

39
gsuid_core/utils/cookie_manager/add_ck.py
View File

@ -3,8 +3,8 @@ from typing import Dict, List
from http.cookies import SimpleCookie
from gsuid_core.utils.api.mys_api import mys_api
from gsuid_core.utils.database.api import DBSqla
from gsuid_core.utils.error_reply import UID_HINT
from gsuid_core.utils.database.models import GsBind, GsUser, GsCache
pic_path = Path(__file__).parent / 'pic'
id_list = [
@ -26,15 +26,12 @@ sk_list = ['stoken', 'stoken_v2']
ck_list = ['cookie_token', 'cookie_token_v2']
lt_list = ['login_ticket', 'login_ticket_v2']
get_sqla = DBSqla().get_sqla
async def get_ck_by_all_stoken(bot_id: str):
sqla = get_sqla(bot_id)
uid_list: List = await sqla.get_all_uid_list()
uid_list: List = await GsBind.get_all_uid_list_by_game(bot_id)
uid_dict = {}
for uid in uid_list:
user_data = await sqla.select_user_data(uid)
user_data = await GsUser.select_data_by_uid(uid)
if user_data:
uid_dict[uid] = user_data.user_id
im = await refresh_ck_by_uid_list(bot_id, uid_dict)
@ -42,8 +39,7 @@ async def get_ck_by_all_stoken(bot_id: str):
async def get_ck_by_stoken(bot_id: str, user_id: str):
sqla = get_sqla(bot_id)
uid_list = await sqla.get_bind_uid_list(user_id)
uid_list = await GsBind.get_uid_list_by_game(user_id, bot_id)
if uid_list is None:
return UID_HINT
uid_dict = {uid: user_id for uid in uid_list}
@ -52,7 +48,6 @@ async def get_ck_by_stoken(bot_id: str, user_id: str):
async def refresh_ck_by_uid_list(bot_id: str, uid_dict: Dict):
sqla = get_sqla(bot_id)
uid_num = len(uid_dict)
if uid_num == 0:
return '请先绑定一个UID噢~'
@ -60,7 +55,7 @@ async def refresh_ck_by_uid_list(bot_id: str, uid_dict: Dict):
skip_num = 0
error_num = 0
for uid in uid_dict:
stoken = await sqla.get_user_stoken(uid)
stoken = await GsUser.get_user_stoken_by_uid(uid)
if stoken is None:
skip_num += 1
error_num += 1
@ -117,10 +112,9 @@ async def get_account_id(simp_dict: SimpleCookie) -> str:
async def _deal_ck(bot_id: str, mes: str, user_id: str) -> str:
sqla = get_sqla(bot_id)
simp_dict = SimpleCookie(mes)
uid = await sqla.get_bind_uid(user_id)
sr_uid = await sqla.get_bind_sruid(user_id)
uid = await GsBind.get_uid_by_game(user_id, bot_id)
sr_uid = await GsBind.get_uid_by_game(user_id, bot_id, 'sr')
uid_bind = sr_uid_bind = None
if uid is None and sr_uid is None:
@ -228,9 +222,9 @@ async def _deal_ck(bot_id: str, mes: str, user_id: str) -> str:
pass
if uid_bind:
await sqla.refresh_cache(uid_bind)
await GsCache.refresh_cache(uid_bind)
if sr_uid_bind:
await sqla.refresh_cache(sr_uid_bind)
await GsCache.refresh_cache(sr_uid_bind, 'sr')
if is_add_stoken:
im_list.append(f'添加Stoken成功,stuid={account_id},stoken={stoken}')
@ -240,14 +234,15 @@ async def _deal_ck(bot_id: str, mes: str, user_id: str) -> str:
device_id = mys_api.get_device_id()
fp = await mys_api.generate_fp_by_uid(uid)
await sqla.insert_user_data(
await GsUser.insert_data(
user_id,
uid_bind,
sr_uid_bind,
account_cookie,
app_cookie,
fp,
device_id,
bot_id,
uid=uid_bind,
sr_uid=sr_uid_bind,
cookie=cookie_token,
stoken=app_cookie,
fp=fp,
device_id=device_id,
)
im_list.append(

16
gsuid_core/utils/cookie_manager/qrlogin.py
View File

@ -16,13 +16,11 @@ from gsuid_core.models import Event
from gsuid_core.logger import logger
from gsuid_core.segment import MessageSegment
from gsuid_core.utils.api.mys_api import mys_api
from gsuid_core.utils.database.api import DBSqla
get_sqla = DBSqla().get_sqla
from gsuid_core.utils.database.models import GsBind
async def get_qrcode_base64(url: str, path: Path, bot_id: str) -> bytes:
qr = qrcode.QRCode(
qr = qrcode.QRCode( # type: ignore
version=1,
error_correction=ERROR_CORRECT_L,
box_size=10,
@ -83,8 +81,6 @@ async def refresh(
async def qrcode_login(bot: Bot, ev: Event, user_id: str) -> str:
sqla = get_sqla(ev.bot_id)
async def send_msg(msg: str):
await bot.send(msg)
return ''
@ -157,8 +153,12 @@ async def qrcode_login(bot: Bot, ev: Event, user_id: str) -> str:
im = '[登录]请求失败, 请稍后再试...'
return await send_msg(im)
uid_bind_list = await sqla.get_bind_uid_list(user_id) or []
sruid_bind_list = await sqla.get_bind_sruid_list(user_id) or []
uid_bind_list = (
await GsBind.get_uid_list_by_game(user_id, ev.bot_id) or []
)
sruid_bind_list = (
await GsBind.get_uid_list_by_game(user_id, ev.bot_id, 'sr') or []
)
# 没有在gsuid绑定uid的情况
if not (uid_bind_list or sruid_bind_list):
logger.warning('[登录]game_token获取失败')

37
gsuid_core/utils/database/base_models.py
View File

@ -23,6 +23,7 @@ from gsuid_core.data_store import get_res_path
T_BaseModel = TypeVar('T_BaseModel', bound='BaseModel')
T_BaseIDModel = TypeVar('T_BaseIDModel', bound='BaseIDModel')
T_User = TypeVar('T_User', bound='User')
T_Push = TypeVar('T_Push', bound='Push')
P = ParamSpec("P")
R = TypeVar("R")
@ -80,6 +81,26 @@ class BaseIDModel(SQLModel):
class BaseBotIDModel(BaseIDModel):
bot_id: str = Field(title='平台')
@classmethod
@with_session
async def update_data_by_uid_without_bot_id(
cls,
session: AsyncSession,
uid: str,
game_name: Optional[str] = None,
**data,
) -> int:
sql = update(cls).where(
getattr(cls, cls.get_gameid_name(game_name)) == uid,
)
if data is not None:
query = sql.values(**data)
query.execution_options(synchronize_session='fetch')
await session.execute(query)
await session.commit()
return 0
return -1
@classmethod
@with_session
async def update_data_by_uid(
@ -660,4 +681,18 @@ class Cache(BaseIDModel):
class Push(BaseBotIDModel):
pass
@classmethod
@with_session
async def select_data_by_uid(
cls: Type[T_Push],
session: AsyncSession,
uid: str,
game_name: Optional[str] = None,
) -> Optional[T_Push]:
result = await session.execute(
select(cls).where(
getattr(cls, cls.get_gameid_name(game_name)) == uid,
)
)
data = result.scalars().all()
return data[0] if data else None

18
poetry.lock generated
View File

@ -2049,13 +2049,13 @@ reference = "mirrors"
[[package]]
name = "smmap"
version = "5.0.0"
version = "5.0.1"
description = "A pure Python implementation of a sliding window memory map manager"
optional = false
python-versions = ">=3.6"
python-versions = ">=3.7"
files = [
{file = "smmap-5.0.0-py3-none-any.whl", hash = "sha256:2aba19d6a040e78d8b09de5c57e96207b09ed71d8e55ce0959eeee6c8e190d94"},
{file = "smmap-5.0.0.tar.gz", hash = "sha256:c840e62059cd3be204b0c9c9f74be2c09d5648eddd4580d9314c3ecde0b30936"},
{file = "smmap-5.0.1-py3-none-any.whl", hash = "sha256:e6d8668fa5f93e706934a62d7b4db19c8d9eb8cf2adbb75ef1b675aa332b69da"},
{file = "smmap-5.0.1.tar.gz", hash = "sha256:dceeb6c0028fdb6734471eb07c0cd2aae706ccaecab45965ee83f11c8d3b1f62"},
]
[package.source]
@ -2383,13 +2383,13 @@ reference = "mirrors"
[[package]]
name = "typing-extensions"
version = "4.7.1"
description = "Backported and Experimental Type Hints for Python 3.7+"
version = "4.8.0"
description = "Backported and Experimental Type Hints for Python 3.8+"
optional = false
python-versions = ">=3.7"
python-versions = ">=3.8"
files = [
{file = "typing_extensions-4.7.1-py3-none-any.whl", hash = "sha256:440d5dd3af93b060174bf433bccd69b0babc3b15b1a8dca43789fd7f61514b36"},
{file = "typing_extensions-4.7.1.tar.gz", hash = "sha256:b75ddc264f0ba5615db7ba217daeb99701ad295353c45f9e95963337ceeeffb2"},
{file = "typing_extensions-4.8.0-py3-none-any.whl", hash = "sha256:8f92fc8806f9a6b641eaa5318da32b44d401efaac0f6678c9bc448ba3605faa0"},
{file = "typing_extensions-4.8.0.tar.gz", hash = "sha256:df8e4339e9cb77357558cbdbceca33c303714cf861d1eef15e1070055ae8b7ef"},
]
[package.source]

4
requirements.txt
View File

@ -45,7 +45,7 @@ python-multipart==0.0.6 ; python_full_version >= "3.8.1" and python_full_version
pytz==2023.3.post1 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
qrcode[pil]==7.4.2 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
six==1.16.0 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
smmap==5.0.0 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
smmap==5.0.1 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
sniffio==1.3.0 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
sortedcontainers==2.4.0 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
soupsieve==2.5 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
@ -56,7 +56,7 @@ sqlmodel==0.0.8 ; python_full_version >= "3.8.1" and python_full_version < "4.0.
sqlmodelx==0.0.5 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
starlette==0.27.0 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
trio==0.22.2 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
typing-extensions==4.7.1 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
typing-extensions==4.8.0 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
tzdata==2023.3 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0" and platform_system == "Windows"
tzlocal==5.0.1 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"
urllib3==1.26.16 ; python_full_version >= "3.8.1" and python_full_version < "4.0.0"