Diuse/zhenxun_plugin_farm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

📝 继续完善数据库

Browse Source
This commit is contained in:
Art_Sakura 2025-04-29 18:11:09 +08:00
parent 6b99ec427c
commit 8243a1e1c8
12 changed files with 358 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
__init__.py
View File

@ -7,7 +7,8 @@ from zhenxun.utils.message import MessageUtils
from .command import diuse_farm, diuse_register, reclamation
from .config import g_pJsonManager
from .database import g_pSqlManager
from .database.database import g_pSqlManager
from .dbService import g_pDBService
from .farm.farm import g_pFarmManager
from .farm.shop import g_pShopManager
from .request import g_pRequestManager
@ -36,7 +37,7 @@ __plugin_meta__ = PluginMetadata(
""".strip(),
extra=PluginExtraData(
author="Art_Sakura",
version="1.1",
version="1.2",
commands=[Command(command="我的农场")],
menu_type="群内小游戏",
configs=[
@ -79,6 +80,8 @@ async def start():
# 初始化读取Json
await g_pJsonManager.init()
await g_pDBService.init()
# 析构函数
@driver.on_shutdown
async def shutdown():

19
command.py
View File

@ -1,6 +1,5 @@
from nonebot.adapters import Event, MessageTemplate
from nonebot.rule import to_me
from nonebot.typing import T_State
from nonebot_plugin_alconna import (Alconna, AlconnaMatch, AlconnaQuery, Args,
Arparma, At, Match, MultiVar, Option,
Query, Subcommand, on_alconna, store_true)
@ -11,15 +10,15 @@ from zhenxun.services.log import logger
from zhenxun.utils.depends import UserName
from zhenxun.utils.message import MessageUtils
from .database import g_pSqlManager
from .dbService import g_pDBService
from .farm.farm import g_pFarmManager
from .farm.shop import g_pShopManager
async def isRegisteredByUid(uid: str) -> bool:
point = await g_pSqlManager.getUserPointByUid(uid)
result = await g_pDBService.user.isUserExist(uid)
if point < 0:
if not result:
await MessageUtils.build_message("尚未开通农场，快at我发送 开通农场 开通吧").send()
return False
@ -36,7 +35,7 @@ diuse_register = on_alconna(
@diuse_register.handle()
async def handle_register(session: Uninfo):
uid = str(session.user.id)
user = await g_pSqlManager.getUserInfoByUid(uid)
user = await g_pDBService.user.getUserInfoByUid(uid)
if user:
await MessageUtils.build_message("🎉 您已经开通农场啦~").send(reply_to=True)
@ -48,7 +47,7 @@ async def handle_register(session: Uninfo):
safe_name = sanitize_username(raw_name)
# 初始化用户信息
success = await g_pSqlManager.initUserInfoByUid(
success = await g_pDBService.user.initUserInfoByUid(
uid=uid,
name=safe_name,
exp=0,
@ -160,7 +159,7 @@ diuse_farm.shortcut(
@diuse_farm.assign("my-point")
async def _(session: Uninfo):
uid = str(session.user.id)
point = await g_pSqlManager.getUserPointByUid(uid)
point = await g_pDBService.user.getUserPointByUid(uid)
if point < 0:
await MessageUtils.build_message("尚未开通农场，快at我发送 开通农场 开通吧").send()
@ -366,9 +365,9 @@ async def _(session: Uninfo, target: Match[At]):
await MessageUtils.build_message("请在指令后跟需要at的人").finish(reply_to=True)
tar = target.result
point = await g_pSqlManager.getUserPointByUid(tar.target)
result = await g_pDBService.user.isUserExist(tar.target)
if point < 0:
if not result:
await MessageUtils.build_message("目标尚未开通农场，快邀请ta开通吧").send()
return None
@ -419,7 +418,7 @@ async def _(session: Uninfo, name: Match[str]):
safeName = sanitize_username(name.result)
result = await g_pSqlManager.updateUserNameByUid(uid, safeName)
result = await g_pDBService.user.updateUserNameByUid(uid, safeName)
if result == True:
await MessageUtils.build_message("更新用户名成功！").send(reply_to=True)

3
database/database.py
View File

@ -8,7 +8,6 @@ import aiosqlite
from zhenxun.services.log import logger
from ..config import g_sDBFilePath, g_sDBPath
from ..dbService import g_pDBService
class CSqlManager:
@ -27,8 +26,6 @@ class CSqlManager:
cls.m_pDB = await aiosqlite.connect(g_sDBFilePath)
cls.m_pDB.row_factory = aiosqlite.Row
await g_pDBService.init()
return True
@classmethod

58
database/user.py
View File

@ -1,10 +1,11 @@
import math
from datetime import date, datetime, timedelta
from typing import List, Union
from database import CSqlManager
from zhenxun.services.log import logger
from .database import CSqlManager
class CUserDB(CSqlManager):
@classmethod
@ -65,6 +66,28 @@ class CUserDB(CSqlManager):
rows = await cursor.fetchall()
return [row[0] for row in rows]
@classmethod
async def isUserExist(cls, uid: str) -> bool:
"""判断用户是否存在
Args:
uid (str): 用户Uid
Returns:
bool: 如果用户存在返回True，否则返回False
"""
if not uid:
return False
try:
async with cls.m_pDB.execute(
"SELECT 1 FROM user WHERE uid = ?", (uid,)
) as cursor:
row = await cursor.fetchone()
return row is not None
except Exception as e:
logger.warning("isUserExist 查询失败！", e=e)
return False
@classmethod
async def getUserInfoByUid(cls, uid: str) -> dict:
"""获取指定用户完整信息
@ -240,22 +263,37 @@ class CUserDB(CSqlManager):
uid (str): 用户Uid
Returns:
tuple[int, int, int]: (当前等级, 下级所需经验, 当前等级剩余经验)，失败返回(-1, -1, -1)
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
if not row or row[0] is None:
return -1, -1, -1
expVal = int(row[0])
levelStep = 200 # 每级经验增量
discriminant = 1 + 8 * expVal / levelStep
level = int((-1 + math.sqrt(discriminant)) // 2)
if level < 0:
level = 0
def cumExp(k: int) -> int:
return levelStep * k * (k + 1) // 2
totalExpCurrentLevel = cumExp(level)
totalExpNextLevel = cumExp(level + 1)
currentExp = expVal - totalExpCurrentLevel
return level, totalExpNextLevel, currentExp
return -1, -1, -1
except Exception as e:
logger.warning("getUserLevelByUid 查询失败！", e=e)

14
database/userItem.py
View File

@ -1,9 +1,9 @@
from typing import Optional
from database import CSqlManager
from zhenxun.services.log import logger
from .database import CSqlManager
class CUserItemDB(CSqlManager):
@classmethod
@ -41,7 +41,7 @@ class CUserItemDB(CSqlManager):
row = await cursor.fetchone()
return row[0] if row else None
except Exception as e:
logger.warning(f"真寻农场getUserItemByName查询失败！", e=e)
logger.warning(f"getUserItemByName查询失败！", e=e)
return None
@classmethod
@ -64,7 +64,7 @@ class CUserItemDB(CSqlManager):
rows = await cursor.fetchall()
return {row["item"]: row["count"] for row in rows}
except Exception as e:
logger.warning(f"真寻农场getUserItemByUid查询失败！", e=e)
logger.warning(f"getUserItemByUid查询失败！", e=e)
return {}
@classmethod
@ -88,7 +88,7 @@ class CUserItemDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场deleteUserItemByName失败！", e=e)
logger.warning(f"deleteUserItemByName失败！", e=e)
return False
@classmethod
@ -119,7 +119,7 @@ class CUserItemDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场updateUserItemByName失败！", e=e)
logger.warning(f"updateUserItemByName失败！", e=e)
return False
@classmethod
@ -164,5 +164,5 @@ class CUserItemDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场addUserItemByUid失败！", e=e)
logger.warning(f"addUserItemByUid失败！", e=e)
return False

12
database/userPlant.py
View File

@ -1,9 +1,9 @@
from typing import Dict, Optional
from database import CSqlManager
from zhenxun.services.log import logger
from .database import CSqlManager
class CUserPlantDB(CSqlManager):
@classmethod
@ -56,7 +56,7 @@ class CUserPlantDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场addUserPlantByUid 失败！", e=e)
logger.warning(f"addUserPlantByUid 失败！", e=e)
return False
@classmethod
@ -95,7 +95,7 @@ class CUserPlantDB(CSqlManager):
row = await cursor.fetchone()
return row[0] if row else None
except Exception as e:
logger.warning(f"真寻农场getUserPlantByName 查询失败！", e=e)
logger.warning(f"getUserPlantByName 查询失败！", e=e)
return None
@classmethod
@ -121,7 +121,7 @@ class CUserPlantDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场updateUserPlantByName失败！", e=e)
logger.warning(f"updateUserPlantByName失败！", e=e)
return False
@classmethod
@ -143,5 +143,5 @@ class CUserPlantDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场deleteUserPlantByName 失败！", e=e)
logger.warning(f"deleteUserPlantByName 失败！", e=e)
return False

89
database/userSeed.py
View File

@ -1,9 +1,9 @@
from typing import Optional
from database import CSqlManager
from zhenxun.services.log import logger
from .database import CSqlManager
class CUserSeedDB(CSqlManager):
@classmethod
@ -22,7 +22,7 @@ class CUserSeedDB(CSqlManager):
@classmethod
async def addUserSeedByUid(cls, uid: str, seed: str, count: int = 1) -> bool:
"""根据用户uid添加种子信息
"""根据用户uid添加种子信息（事务版本）
Args:
uid (str): 用户uid
@ -34,7 +34,6 @@ class CUserSeedDB(CSqlManager):
"""
try:
async with cls._transaction():
#检查是否已存在该种子
async with cls.m_pDB.execute(
"SELECT count FROM userSeed WHERE uid = ? AND seed = ?",
(uid, seed)
@ -42,21 +41,18 @@ class CUserSeedDB(CSqlManager):
row = await cursor.fetchone()
if row:
#如果种子已存在，则更新数量
newCount = row[0] + count
await cls.m_pDB.execute(
"UPDATE userSeed SET count = ? WHERE uid = ? AND seed = ?",
(newCount, uid, seed)
)
else:
#如果种子不存在，则插入新记录
newCount = count
await cls.m_pDB.execute(
"INSERT INTO userSeed (uid, seed, count) VALUES (?, ?, ?)",
(uid, seed, count)
)
#如果种子数量为 0，删除记录
if newCount <= 0:
await cls.m_pDB.execute(
"DELETE FROM userSeed WHERE uid = ? AND seed = ?",
@ -64,9 +60,33 @@ class CUserSeedDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场addUserSeedByUid 失败！", e=e)
logger.warning(f"addUserSeedByUid 失败！", e=e)
return False
@classmethod
async def _addUserSeedByUid(cls, uid: str, seed: str, count: int = 1) -> bool:
"""根据用户uid添加种子信息（非事务版，复用其他非事务接口）"""
try:
existing = await cls.getUserSeedByName(uid, seed)
newCount = (existing or 0) + count
if existing is not None:
await cls._updateUserSeedByName(uid, seed, newCount)
else:
await cls.m_pDB.execute(
"INSERT INTO userSeed (uid, seed, count) VALUES (?, ?, ?)",
(uid, seed, newCount)
)
if newCount <= 0:
await cls._deleteUserSeedByName(uid, seed)
return True
except Exception as e:
logger.warning(f"_addUserSeedByUid 失败！", e=e)
return False
@classmethod
async def getUserSeedByName(cls, uid: str, seed: str) -> Optional[int]:
"""根据种子名称获取种子数量
@ -87,7 +107,7 @@ class CUserSeedDB(CSqlManager):
row = await cursor.fetchone()
return row[0] if row else None
except Exception as e:
logger.warning(f"真寻农场getUserSeedByName 查询失败！", e=e)
logger.warning(f"getUserSeedByName 查询失败！", e=e)
return None
@classmethod
@ -131,7 +151,33 @@ class CUserSeedDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场updateUserSeedByName失败！", e=e)
logger.warning(f"updateUserSeedByName失败！", e=e)
return False
@classmethod
async def _updateUserSeedByName(cls, uid: str, seed: str, count: int) -> bool:
"""根据种子名称更新种子数量
Args:
uid (str): 用户uid
seed (str): 种子名称
count (int): 种子数量
Returns:
bool: 是否成功
"""
try:
if count <= 0:
return await cls.deleteUserSeedByName(uid, seed)
async with cls._transaction():
await cls.m_pDB.execute(
"UPDATE userSeed SET count = ? WHERE uid = ? AND seed = ?",
(count, uid, seed)
)
return True
except Exception as e:
logger.warning(f"updateUserSeedByName失败！", e=e)
return False
@classmethod
@ -153,5 +199,26 @@ class CUserSeedDB(CSqlManager):
)
return True
except Exception as e:
logger.warning(f"真寻农场deleteUserSeedByName 删除失败！", e=e)
logger.warning(f"deleteUserSeedByName 删除失败！", e=e)
return False
@classmethod
async def _deleteUserSeedByName(cls, uid: str, seed: str) -> bool:
"""根据种子名称从种子仓库中删除种子
Args:
uid (str): 用户uid
seed (str): 种子名称
Returns:
bool: 是否成功
"""
try:
await cls.m_pDB.execute(
"DELETE FROM userSeed WHERE uid = ? AND seed = ?",
(uid, seed)
)
return True
except Exception as e:
logger.warning(f"deleteUserSeedByName 删除失败！", e=e)
return False

211
database/userSoil.py
View File

@ -1,28 +1,27 @@
from datetime import date, datetime, timedelta
from typing import Optional
from database import CSqlManager
from zhenxun.services.log import logger
from ..config import g_pJsonManager
from ..dbService import g_pDBService
from .database import CSqlManager
class CUserSoilDB(CSqlManager):
@classmethod
async def initDB(cls):
#用户Uid
#地块索引从1开始
#作物名称
#播种时间
#成熟时间
#土地等级 0=普通地，1=红土地，2=黑土地，3=金土地
#枯萎状态 0=未枯萎，1=枯萎
#施肥状态 0=未施肥，1=施肥
#虫害状态 0=无虫害，1=有虫害
#杂草状态 0=无杂草，1=有杂草
#缺水状态 0=不缺水，1=缺水
# 用户Uid
# 地块索引从1开始
# 作物名称
# 播种时间
# 成熟时间
# 土地等级 0=普通地，1=红土地，2=黑土地，3=金土地
# 枯萎状态 0=未枯萎，1=枯萎
# 施肥状态 0=未施肥，1=施肥
# 虫害状态 0=无虫害，1=有虫害
# 杂草状态 0=无杂草，1=有杂草
# 缺水状态 0=不缺水，1=缺水
userSoil = {
"uid": "TEXT NOT NULL",
"soilIndex": "INTEGER NOT NULL",
@ -35,7 +34,7 @@ class CUserSoilDB(CSqlManager):
"bugStatus": "INTEGER DEFAULT 0",
"weedStatus": "INTEGER DEFAULT 0",
"waterStatus": "INTEGER DEFAULT 0",
"PRIMARY KEY": "(uid, soilIndex)"
"PRIMARY KEY": "(uid, soilIndex)",
}
await cls.ensureTableSchema("userSoil", userSoil)
@ -52,61 +51,50 @@ class CUserSoilDB(CSqlManager):
"""
cursor = await cls.m_pDB.execute("SELECT * FROM soil WHERE uid = ?", (uid,))
row = await cursor.fetchone()
if not row:
return {}
columns = [description[0] for description in cursor.description]
return dict(zip(columns, row))
@classmethod
async def migrateOldFarmData(cls):
async def migrateOldFarmData(cls) -> bool:
"""迁移旧土地数据到新表 userSoil 并删除旧表
Raises:
aiosqlite.Error: 数据库操作失败时抛出
Returns:
None
bool: 如果旧表不存在则返回 False，否则迁移并删除后返回 True
"""
# 检查旧表是否存在
cursor = await cls.m_pDB.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='soil'"
)
if not await cursor.fetchone():
return False
async with cls._transaction():
users = await g_pDBService.user.getAllUsers()
for uid in users:
farmInfo = await cls.getUserFarmByUid(uid)
for i in range(1, 31):
soilName = f"soil{i}"
if soilName not in farmInfo:
key = f"soil{i}"
data = farmInfo.get(key)
if not data:
continue
soilData = farmInfo[soilName]
if not soilData:
parts = data.split(",")
if len(parts) < 3:
continue
name = parts[0]
pt = int(parts[1])
mt = int(parts[2])
fields = soilData.split(',')
if len(fields) < 6:
continue
await cls.m_pDB.execute(
"INSERT INTO userSoil (uid,soilIndex,plantName,plantTime,matureTime) VALUES (?,?,?,?,?)",
(uid, i, name, pt, mt),
)
plantName = fields[0]
plantTime = int(fields[1])
matureTime = int(fields[2])
soilLevel = int(fields[5])
# 构建新的土地数据
soilInfo = {
"uid": uid,
"soilIndex": i,
"plantName": plantName,
"plantTime": plantTime,
"matureTime": matureTime,
"soilLevel": soilLevel,
"wiltStatus":0,
"fertilizerStatus": 0,
"bugStatus": 0,
"weedStatus": 0,
"waterStatus": 0
}
await cls.insertUserSoil(soilInfo)
# 彻底清除旧表
await cls.m_pDB.execute("DROP TABLE soil")
return True
@classmethod
async def insertUserSoil(cls, soilInfo: dict):
@ -131,10 +119,36 @@ class CUserSoilDB(CSqlManager):
soilInfo.get("fertilizerStatus", 0),
soilInfo.get("bugStatus", 0),
soilInfo.get("weedStatus", 0),
soilInfo.get("waterStatus", 0)
)
soilInfo.get("waterStatus", 0),
),
)
@classmethod
async def _insertUserSoil(cls, soilInfo: dict):
"""插入一条新的 userSoil 记录
Args:
soilInfo (dict): 新土地数据
Returns:
None
"""
await cls.m_pDB.execute(
"INSERT INTO userSoil (uid, soilIndex, plantName, plantTime, matureTime, soilLevel, fertilizerStatus, bugStatus, weedStatus, waterStatus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
soilInfo["uid"],
soilInfo["soilIndex"],
soilInfo.get("plantName", ""),
soilInfo.get("plantTime", 0),
soilInfo.get("matureTime", 0),
soilInfo.get("soilLevel", 0),
soilInfo.get("fertilizerStatus", 0),
soilInfo.get("bugStatus", 0),
soilInfo.get("weedStatus", 0),
soilInfo.get("waterStatus", 0),
),
)
@classmethod
async def getUserSoil(cls, uid: str, soilIndex: int) -> Optional[dict]:
"""获取指定用户某块土地的详细信息
@ -149,7 +163,7 @@ class CUserSoilDB(CSqlManager):
async with cls._transaction():
cursor = await cls.m_pDB.execute(
"SELECT * FROM userSoil WHERE uid = ? AND soilIndex = ?",
(uid, soilIndex)
(uid, soilIndex),
)
row = await cursor.fetchone()
if not row:
@ -157,6 +171,27 @@ class CUserSoilDB(CSqlManager):
columns = [description[0] for description in cursor.description]
return dict(zip(columns, row))
@classmethod
async def _getUserSoil(cls, uid: str, soilIndex: int) -> Optional[dict]:
"""获取指定用户某块土地的详细信息
Args:
uid (str): 用户ID
soilIndex (int): 土地索引
Returns:
Optional[dict]: 记录存在返回字段-值字典，否则返回 None
"""
cursor = await cls.m_pDB.execute(
"SELECT * FROM userSoil WHERE uid = ? AND soilIndex = ?",
(uid, soilIndex),
)
row = await cursor.fetchone()
if not row:
return None
columns = [description[0] for description in cursor.description]
return dict(zip(columns, row))
@classmethod
async def updateUserSoil(cls, uid: str, soilIndex: int, field: str, value):
"""更新指定用户土地的单个字段
@ -173,9 +208,27 @@ class CUserSoilDB(CSqlManager):
async with cls._transaction():
await cls.m_pDB.execute(
f"UPDATE userSoil SET {field} = ? WHERE uid = ? AND soilIndex = ?",
(value, uid, soilIndex)
(value, uid, soilIndex),
)
@classmethod
async def _updateUserSoil(cls, uid: str, soilIndex: int, field: str, value):
"""更新指定用户土地的单个字段
Args:
uid (str): 用户ID
soilIndex (int): 土地索引
field (str): 需更新的字段名
value: 新值
Returns:
None
"""
await cls.m_pDB.execute(
f"UPDATE userSoil SET {field} = ? WHERE uid = ? AND soilIndex = ?",
(value, uid, soilIndex),
)
@classmethod
async def deleteUserSoil(cls, uid: str, soilIndex: int):
"""删除指定用户的土地记录
@ -189,10 +242,24 @@ class CUserSoilDB(CSqlManager):
"""
async with cls._transaction():
await cls.m_pDB.execute(
"DELETE FROM userSoil WHERE uid = ? AND soilIndex = ?",
(uid, soilIndex)
"DELETE FROM userSoil WHERE uid = ? AND soilIndex = ?", (uid, soilIndex)
)
@classmethod
async def _deleteUserSoil(cls, uid: str, soilIndex: int):
"""删除指定用户的土地记录
Args:
uid (str): 用户ID
soilIndex (int): 土地索引
Returns:
None
"""
await cls.m_pDB.execute(
"DELETE FROM userSoil WHERE uid = ? AND soilIndex = ?", (uid, soilIndex)
)
@classmethod
async def isSoilPlanted(cls, uid: str, soilIndex: int) -> bool:
"""判断指定用户的指定土地是否已种植
@ -240,22 +307,24 @@ class CUserSoilDB(CSqlManager):
async with cls._transaction():
# 复用原有记录字段，保留土壤状态等信息
prev = soilRecord or {}
await cls.deleteUserSoil(uid, soilIndex)
await cls.insertUserSoil({
"uid": uid,
"soilIndex": soilIndex,
"plantName": plantName,
"plantTime": nowTs,
"matureTime": matureTs,
# 保留之前的土壤等级和状态字段，避免数据丢失
"soilLevel": prev.get("soilLevel", 0),
"wiltStatus": prev.get("wiltStatus", 0),
"fertilizerStatus": prev.get("fertilizerStatus", 0),
"bugStatus": prev.get("bugStatus", 0),
"weedStatus": prev.get("weedStatus", 0),
"waterStatus": prev.get("waterStatus", 0)
})
await cls._deleteUserSoil(uid, soilIndex)
await cls._insertUserSoil(
{
"uid": uid,
"soilIndex": soilIndex,
"plantName": plantName,
"plantTime": nowTs,
"matureTime": matureTs,
# 保留之前的土壤等级和状态字段，避免数据丢失
"soilLevel": prev.get("soilLevel", 0),
"wiltStatus": prev.get("wiltStatus", 0),
"fertilizerStatus": prev.get("fertilizerStatus", 0),
"bugStatus": prev.get("bugStatus", 0),
"weedStatus": prev.get("weedStatus", 0),
"waterStatus": prev.get("waterStatus", 0),
}
)
return True
except Exception as e:
logger.error(f"真寻农场播种失败！", e=e)
logger.error(f"播种失败！", e=e)
return False

4
database/userSteal.py
View File

@ -1,7 +1,7 @@
from database import CSqlManager
from zhenxun.services.log import logger
from .database import CSqlManager
class CUserStealDB(CSqlManager):
@classmethod

36
dbService.py
View File

@ -1,29 +1,35 @@
from typing import Optional
from .database.user import CUserDB
from .database.userItem import CUserItemDB
from .database.userPlant import CUserPlantDB
from .database.userSeed import CUserSeedDB
from .database.userSoil import CUserSoilDB
from .database.userSteal import CUserStealDB
class CDBService:
def __init__(self):
user: Optional["CUserDB"] = None
userSoil: Optional["CUserSoilDB"] = None
userPlant: Optional["CUserPlantDB"] = None
userSeed: Optional["CUserSeedDB"] = None
userItem: Optional["CUserItemDB"] = None
userSteal: Optional["CUserStealDB"] = None
@classmethod
async def init(cls):
from .database.user import CUserDB
from .database.userItem import CUserItemDB
from .database.userPlant import CUserPlantDB
from .database.userSeed import CUserSeedDB
from .database.userSoil import CUserSoilDB
from .database.userSteal import CUserStealDB
cls.user = CUserDB()
await cls.user.initDB()
cls.userSoil = CUserSoilDB()
await cls.userSoil.initDB()
cls.userPlant = CUserPlantDB()
await cls.userPlant.initDB()
cls.userSeed = CUserSeedDB()
await cls.userSeed.initDB()
cls.userItem = CUserItemDB()
await cls.userItem.initDB()
cls.userSteal = CUserStealDB()
await cls.userSteal.initDB()
#迁移旧数据库
await cls.userSoil.migrateOldFarmData()
g_pDBService = CDBService()

2
farm/farm.py
View File

@ -389,7 +389,7 @@ class CFarmManager:
number = plantInfo['harvest']
#处理偷菜扣除数量
stealNum = g_pDBService.userSteal.getTotalStolenCount(uid, i)
stealNum = await g_pDBService.userSteal.getTotalStolenCount(uid, i)
number -= stealNum

56
request.py
View File

@ -37,47 +37,73 @@ class CRequestManager:
return False
@classmethod
async def post(cls, endpoint: str, name: str = "", jsonData: dict = None, formData: dict = None) -> dict:
"""发送POST请求到指定接口，供其他方法统一调用
async def post(cls, endpoint: str, name: str = "", jsonData: dict = None) -> dict:
"""发送POST请求到指定接口，统一调用，仅支持JSON格式数据
Args:
endpoint (str): 请求的接口路径
name (str, optional): 操作名称用于日志记录
jsonData (dict, optional): 以JSON格式发送的数据
formData (dict, optional): 以表单格式发送的数据
jsonData (dict): 以JSON格式发送的数据
Raises:
ValueError: 当jsonData和formData都未提供时抛出
ValueError: 当jsonData未提供时抛出
Returns:
dict: 返回请求结果的JSON数据
"""
if jsonData is None and formData is None:
raise ValueError("post请求必须提供jsonData或formData其中之一")
if jsonData is None:
raise ValueError("post请求必须提供jsonData")
baseUrl = Config.get_config("zhenxun_plugin_farm", "服务地址")
url = f"{baseUrl.rstrip('/')}/{endpoint.lstrip('/')}"
headers = {"token": "xZ%?z5LtWV7H:0-Xnwp+bNRNQ-jbfrxG"}
try:
async with httpx.AsyncClient(timeout=5.0) as client:
if jsonData is not None:
response = await client.post(url, json=jsonData)
else:
response = await client.post(url, data=formData)
response = await client.post(url, json=jsonData, headers=headers)
if response.status_code == 200:
return response.json()
else:
logger.warning(f"真寻农场{name}请求失败: HTTP {response.status_code} {response.text}")
logger.warning(f"{name}请求失败: HTTP {response.status_code} {response.text}")
return {}
except httpx.RequestError as e:
logger.warning(f"真寻农场{name}请求异常: {e}")
logger.warning(f"{name}请求异常", e=e)
return {}
except Exception as e:
logger.warning(f"真寻农场{name}处理异常: {e}")
logger.warning(f"{name}处理异常", e=e)
return {}
@classmethod
async def get(cls, endpoint: str, name: str = "") -> dict:
"""发送GET请求到指定接口，统一调用，仅支持无体的查询
Args:
endpoint (str): 请求的接口路径
name (str, optional): 操作名称用于日志记录
Returns:
dict: 返回请求结果的JSON数据
"""
baseUrl = Config.get_config("zhenxun_plugin_farm", "服务地址")
url = f"{baseUrl.rstrip('/')}/{endpoint.lstrip('/')}"
headers = {"token": "xZ%?z5LtWV7H:0-Xnwp+bNRNQ-jbfrxG"}
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(url, headers=headers)
if response.status_code == 200:
return response.json()
else:
logger.warning(f"{name}请求失败: HTTP {response.status_code} {response.text}")
return {}
except httpx.RequestError as e:
logger.warning(f"{name}请求异常", e=e)
return {}
except Exception as e:
logger.warning(f"{name}处理异常", e=e)
return {}
@classmethod
async def sign(cls, uid: str) -> str: