401 lines
12 KiB
Python
401 lines
12 KiB
Python
from datetime import date, datetime, timedelta
|
||
from typing import List, Union
|
||
|
||
from database import CSqlManager
|
||
|
||
from zhenxun.services.log import logger
|
||
|
||
|
||
class CUserDB(CSqlManager):
|
||
@classmethod
|
||
async def initDB(cls):
|
||
"""初始化用户表结构，确保user表存在且字段完整"""
|
||
# 用户Uid
|
||
# 农场名称
|
||
# 经验值
|
||
# 金币
|
||
# 解锁土地数量
|
||
# 偷菜时间字符串
|
||
# 剩余偷菜次数
|
||
userInfo = {
|
||
"uid": "INTEGER PRIMARY KEY AUTOINCREMENT",
|
||
"name": "TEXT NOT NULL",
|
||
"exp": "INTEGER DEFAULT 0",
|
||
"point": "INTEGER DEFAULT 0",
|
||
"soil": "INTEGER DEFAULT 3",
|
||
"stealTime": "TEXT DEFAULT NULL",
|
||
"stealCount": "INTEGER DEFAULT 0"
|
||
}
|
||
await cls.ensureTableSchema("user", userInfo)
|
||
|
||
@classmethod
|
||
async def initUserInfoByUid(cls, uid: str, name: str = "", exp: int = 0, point: int = 500) -> Union[bool, str]:
|
||
"""初始化用户信息，包含初始偷菜时间字符串与次数
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
name (str): 农场名称
|
||
exp (int): 农场经验
|
||
point (int): 农场币
|
||
|
||
Returns:
|
||
Union[bool, str]: False 表示失败，字符串表示成功信息
|
||
"""
|
||
nowStr = date.today().strftime('%Y-%m-%d')
|
||
sql = (
|
||
f"INSERT INTO user (uid, name, exp, point, soil, stealTime, stealCount) "
|
||
f"VALUES ({uid}, '{name}', {exp}, {point}, 3, '{nowStr}', 5)"
|
||
)
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(sql)
|
||
return "开通农场成功"
|
||
except Exception as e:
|
||
logger.warning("initUserInfoByUid 事务执行失败！", e=e)
|
||
return False
|
||
|
||
@classmethod
|
||
async def getAllUsers(cls) -> List[str]:
|
||
"""获取所有用户UID列表
|
||
|
||
Returns:
|
||
List[str]: 用户UID列表
|
||
"""
|
||
cursor = await cls.m_pDB.execute("SELECT uid FROM user")
|
||
rows = await cursor.fetchall()
|
||
return [row[0] for row in rows]
|
||
|
||
@classmethod
|
||
async def getUserInfoByUid(cls, uid: str) -> dict:
|
||
"""获取指定用户完整信息
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
dict: 包含所有用户字段的字典
|
||
"""
|
||
if not uid:
|
||
return {}
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT * FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
async for row in cursor:
|
||
return {
|
||
"uid": row[0],
|
||
"name": row[1],
|
||
"exp": row[2],
|
||
"point": row[3],
|
||
"soil": row[4],
|
||
"stealTime": row[5] or "",
|
||
"stealCount": int(row[6])
|
||
}
|
||
return {}
|
||
except Exception as e:
|
||
logger.warning("getUserInfoByUid 查询失败！", e=e)
|
||
return {}
|
||
|
||
@classmethod
|
||
async def getUserNameByUid(cls, uid: str) -> str:
|
||
"""根据用户Uid获取用户名
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
str: 用户名，失败返回空字符串
|
||
"""
|
||
if not uid:
|
||
return ""
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT name FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
return row["name"] if row else ""
|
||
except Exception as e:
|
||
logger.warning("getUserNameByUid 查询失败！", e=e)
|
||
return ""
|
||
|
||
@classmethod
|
||
async def updateUserNameByUid(cls, uid: str, name: str) -> bool:
|
||
"""根据用户Uid更新用户名
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
name (str): 新用户名
|
||
|
||
Returns:
|
||
bool: 是否更新成功
|
||
"""
|
||
if not uid or not name:
|
||
return False
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(
|
||
"UPDATE user SET name = ? WHERE uid = ?", (name, uid)
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("updateUserNameByUid 事务执行失败！", e=e)
|
||
return False
|
||
|
||
@classmethod
|
||
async def getUserPointByUid(cls, uid: str) -> int:
|
||
"""获取指定用户农场币
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
int: 农场币数量，失败返回 -1
|
||
"""
|
||
if not uid:
|
||
return -1
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT point FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
return int(row[0]) if row and row[0] is not None else -1
|
||
except Exception as e:
|
||
logger.warning("getUserPointByUid 查询失败！", e=e)
|
||
return -1
|
||
|
||
@classmethod
|
||
async def updateUserPointByUid(cls, uid: str, point: int) -> bool:
|
||
"""根据用户Uid更新农场币数量
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
point (int): 新农场币数量
|
||
|
||
Returns:
|
||
bool: 是否更新成功
|
||
"""
|
||
if not uid or point < 0:
|
||
logger.warning("updateUserPointByUid 参数校验失败！")
|
||
return False
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(
|
||
"UPDATE user SET point = ? WHERE uid = ?", (point, uid)
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.error("updateUserPointByUid 事务执行失败！", e=e)
|
||
return False
|
||
|
||
@classmethod
|
||
async def getUserExpByUid(cls, uid: str) -> int:
|
||
"""获取指定用户经验值
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
int: 经验值，失败返回 -1
|
||
"""
|
||
if not uid:
|
||
return -1
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT exp FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
return int(row[0]) if row and row[0] is not None else -1
|
||
except Exception as e:
|
||
logger.warning("getUserExpByUid 查询失败！", e=e)
|
||
return -1
|
||
|
||
@classmethod
|
||
async def updateUserExpByUid(cls, uid: str, exp: int) -> bool:
|
||
"""根据用户Uid更新经验值
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
exp (int): 新经验值
|
||
|
||
Returns:
|
||
bool: 是否更新成功
|
||
"""
|
||
if not uid:
|
||
return False
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(
|
||
"UPDATE user SET exp = ? WHERE uid = ?", (exp, uid)
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("updateUserExpByUid 事务执行失败！", e=e)
|
||
return False
|
||
|
||
@classmethod
|
||
async def getUserLevelByUid(cls, uid: str) -> tuple[int, int, int]:
|
||
"""获取用户等级信息
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
tuple[int, int, int]: (当前等级, 下级所需经验, 当前等级剩余经验)，失败返回(-1, -1, -1)
|
||
"""
|
||
if not uid:
|
||
return -1, -1, -1
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT exp FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
if row and row[0] is not None:
|
||
expVal = int(row[0])
|
||
level = expVal // 200
|
||
nextLevelExp = 200 * (level + 1)
|
||
currentLevelExp = level * 200
|
||
remainingExp = expVal - currentLevelExp
|
||
return level, nextLevelExp, remainingExp
|
||
return -1, -1, -1
|
||
except Exception as e:
|
||
logger.warning("getUserLevelByUid 查询失败！", e=e)
|
||
return -1, -1, -1
|
||
|
||
@classmethod
|
||
async def getUserSoilByUid(cls, uid: str) -> int:
|
||
"""获取解锁土地数量
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
int: 解锁土地块数，失败返回0
|
||
"""
|
||
if not uid:
|
||
return 0
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT soil FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
return int(row[0]) if row and row[0] is not None else 0
|
||
except Exception as e:
|
||
logger.warning("getUserSoilByUid 查询失败！", e=e)
|
||
return 0
|
||
|
||
@classmethod
|
||
async def updateUserSoilByUid(cls, uid: str, soil: int) -> bool:
|
||
"""更新指定用户解锁土地数量
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
soil (int): 新土地数量
|
||
|
||
Returns:
|
||
bool: 更新成功返回True，否则False
|
||
"""
|
||
if not uid or soil < 0:
|
||
return False
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(
|
||
"UPDATE user SET soil = ? WHERE uid = ?", (soil, uid)
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("updateUserSoilByUid 事务执行失败！", e=e)
|
||
return False
|
||
|
||
@classmethod
|
||
async def getStealTimeByUid(cls, uid: str) -> str:
|
||
"""根据用户Uid获取偷菜时间字符串
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
str: 偷菜时间字符串，失败返回空字符串
|
||
"""
|
||
if not uid:
|
||
return ""
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT stealTime FROM user WHERE uid = ?", (uid,
|
||
)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
return row[0] if row and row[0] else ""
|
||
except Exception as e:
|
||
logger.warning("getStealTimeByUid 查询失败！", e=e)
|
||
return ""
|
||
|
||
@classmethod
|
||
async def updateStealTimeByUid(cls, uid: str, stealTime: str) -> bool:
|
||
"""根据用户Uid更新偷菜时间字符串
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
stealTime (str): 新偷菜时间字符串
|
||
|
||
Returns:
|
||
bool: 是否更新成功
|
||
"""
|
||
if not uid or not stealTime:
|
||
logger.warning("updateStealTimeByUid 参数校验失败！")
|
||
return False
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(
|
||
"UPDATE user SET stealTime = ? WHERE uid = ?", (stealTime, uid)
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("updateStealTimeByUid 事务执行失败！", e=e)
|
||
return False
|
||
|
||
@classmethod
|
||
async def getStealCountByUid(cls, uid: str) -> int:
|
||
"""根据用户Uid获取剩余偷菜次数
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
|
||
Returns:
|
||
int: 剩余偷菜次数，失败返回 -1
|
||
"""
|
||
if not uid:
|
||
return -1
|
||
try:
|
||
async with cls.m_pDB.execute(
|
||
"SELECT stealCount FROM user WHERE uid = ?", (uid,)
|
||
) as cursor:
|
||
row = await cursor.fetchone()
|
||
return int(row[0]) if row and row[0] is not None else 0
|
||
except Exception as e:
|
||
logger.warning("getStealCountByUid 查询失败！", e=e)
|
||
return -1
|
||
|
||
@classmethod
|
||
async def updateStealCountByUid(cls, uid: str, stealCount: int) -> bool:
|
||
"""根据用户Uid更新剩余偷菜次数
|
||
|
||
Args:
|
||
uid (str): 用户Uid
|
||
stealCount (int): 新剩余偷菜次数
|
||
|
||
Returns:
|
||
bool: 是否更新成功
|
||
"""
|
||
if not uid or stealCount < 0:
|
||
logger.warning("updateStealCountByUid 参数校验失败！")
|
||
return False
|
||
try:
|
||
async with cls._transaction():
|
||
await cls.m_pDB.execute(
|
||
"UPDATE user SET stealCount = ? WHERE uid = ?", (stealCount, uid)
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("updateStealCountByUid 事务执行失败！", e=e)
|
||
return False
|