Diuse/main_plugin_farm
1
0
Fork 0
mirror of https://github.com/Shu-Ying/main_plugin_farm.git synced 2026-10-28 13:10:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

🎨 程序启动不再读取全部json, 改为首次使用时读取

Browse Source
This commit is contained in:
Diuse
2025-10-24 01:26:17 +08:00
parent 2f0023d468
commit a7ffaf9418
9 changed files with 361 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+30 -3
View File
@@ -7,6 +7,7 @@ from zhenxun.services.log import logger
from zhenxun.utils.message import MessageUtils
from .command import diuse_farm, diuse_register, reclamation
from .core.activity.sign_in import g_pSignInManager
from .core.database.database import g_pSqlManager
from .core.dbService import g_pDBService
from .core.farm import g_pFarmManager
@@ -14,6 +15,7 @@ from .core.help import g_pHelpManager
from .core.player.playerPool import g_pUserPool
from .core.shop import g_pShopManager
from .event.event import g_pEventManager
from .utils.config import g_sResourcePath, g_sTranslation
from .utils.json import g_pJsonManager
from .utils.request import g_pRequestManager
@@ -54,6 +56,12 @@ __plugin_meta__ = PluginMetadata(
help="我的农场返回图片的清晰度, [low, medium, hight, original]",
default_value="low",
),
RegisterConfig(
key="签到图片样式",
value="text",
help="签到图片的样式, [text, html]",
default_value="text",
),
RegisterConfig(
key="兑换倍数",
value="2",
@@ -90,9 +98,6 @@ async def start():
# 数据库加载
await g_pSqlManager.init()
# 初始化读取Json
await g_pJsonManager.init()
# 初始化数据库变量 和 加载作物数据库
await g_pDBService.init()
@@ -101,6 +106,28 @@ async def start():
await g_pHelpManager.createHelpImage()
# todayRewards = [
# {
# "imagePath": "../plant/ATTomato/icon.png",
# "name": "金币",
# "quantity": 100,
# },
# ]
# logger.info(f"{todayRewards[0]}")
# cumulativeRewards = [
# {
# "imagePath": g_sResourcePath / "/plant/ATTomato/icon.png",
# "name": "金币",
# "quantity": 200,
# "description": "累计签到1次",
# "achieved": True,
# },
# ]
# await g_pSignInManager.createSignImage(todayRewards, cumulativeRewards, 1, 1, 1)
# 析构函数
@driver.on_shutdown
command.py
+3 -46
View File
@@ -24,11 +24,11 @@ from zhenxun.services.log import logger
from zhenxun.utils._build_image import BuildImage
from zhenxun.utils.message import MessageUtils
from .core.activity.sign_in import g_pSignInManager
from .core.dbService import g_pDBService
from .core.farm import g_pFarmManager
from .core.shop import g_pShopManager
from .utils.config import g_bSignStatus, g_sTranslation
from .utils.json import g_pJsonManager
from .utils.tool import g_pToolManager
diuse_register = on_alconna(
@@ -45,7 +45,7 @@ async def handle_register(session: Uninfo):
uid = str(session.user.id)
player = await g_pToolManager.getPlayerByUid(uid)
if player is not None and await player.isRegistered():
await MessageUtils.build_message(g_sTranslation["register"]["already"]).send(
await MessageUtils.build_message(g_sTranslation["register"]["repeat"]).send(
reply_to=True
)
return
@@ -560,52 +560,9 @@ async def _(session: Uninfo):
# 判断签到是否正常加载
if not g_bSignStatus:
await MessageUtils.build_message(g_sTranslation["signIn"]["error"]).send()
return
toDay = g_pToolManager.dateTime().date().today()
message = ""
status = await g_pDBService.userSign.sign(uid, toDay.strftime("%Y-%m-%d"))
# 如果完成签到
if status == 1 or status == 2:
# 获取签到总天数
signDay = await g_pDBService.userSign.getUserSignCountByDate(
uid, toDay.strftime("%Y-%m")
)
exp, point = await g_pDBService.userSign.getUserSignRewardByDate(
uid, toDay.strftime("%Y-%m-%d")
)
message += g_sTranslation["signIn"]["success"].format(
day=signDay, exp=exp, num=point
)
reward = g_pJsonManager.m_pSign["continuou"].get(f"{signDay}", None)
if reward:
extraPoint = reward.get("point", 0)
extraExp = reward.get("exp", 0)
plant = reward.get("plant", {})
message += g_sTranslation["signIn"]["grandTotal"].format(
exp=extraExp, num=extraPoint
)
vipPoint = reward.get("vipPoint", 0)
if vipPoint > 0:
message += g_sTranslation["signIn"]["grandTotal1"].format(num=vipPoint)
if plant:
for key, value in plant.items():
message += g_sTranslation["signIn"]["grandTotal2"].format(
name=key, num=value
)
else:
message = g_sTranslation["signIn"]["error1"]
message = await g_pSignInManager.signInByUid(uid)
await MessageUtils.build_message(message).send()
core/activity/sign_in.py
+154
View File
@@ -0,0 +1,154 @@
from datetime import datetime
from typing import Any
from zhenxun.configs.config import Config
from zhenxun.services.log import logger
from ...utils.config import DATA_PATH, g_sResourcePath, g_sTranslation
from ...utils.json import g_pJsonManager
from ...utils.tool import g_pToolManager
from ..dbService import g_pDBService
class CSignInManager:
async def signInByUid(self, uid: str) -> str:
toDay = g_pToolManager.dateTime().date().today()
message = ""
status = await g_pDBService.userSign.sign(uid, toDay.strftime("%Y-%m-%d"))
# 如果完成签到
if status == 1 or status == 2:
# 获取签到总天数
signDay = await g_pDBService.userSign.getUserSignCountByDate(
uid, toDay.strftime("%Y-%m")
)
exp, point = await g_pDBService.userSign.getUserSignRewardByDate(
uid, toDay.strftime("%Y-%m-%d")
)
message += g_sTranslation["signIn"]["success"].format(
day=signDay, exp=exp, num=point
)
jSign = await g_pJsonManager.getSign()
reward = jSign["continuou"].get(f"{signDay}", None)
signType = Config.get_config("zhenxun_plugin_farm", "签到图片样式")
if reward:
extraPoint = reward.get("point", 0)
extraExp = reward.get("exp", 0)
plant = reward.get("plant", {})
message += g_sTranslation["signIn"]["grandTotal"].format(
exp=extraExp, num=extraPoint
)
vipPoint = reward.get("vipPoint", 0)
if vipPoint > 0:
message += g_sTranslation["signIn"]["grandTotal1"].format(
num=vipPoint
)
if plant:
for key, value in plant.items():
message += g_sTranslation["signIn"]["grandTotal2"].format(
name=key, num=value
)
else:
message = g_sTranslation["signIn"]["error1"]
pass
return message
def generateSignData(
self,
todayRewards: list[dict[str, Any]],
cumulativeRewards: list[dict[str, Any]],
currentProgress: int,
totalMilestones: int,
consecutiveDays: int = 0,
) -> dict[str, Any]:
"""生成签到页面所需的数据
Args:
todayRewards (list[dict[str, Any]]): 今日奖励列表
cumulativeRewards (list[dict[str, Any]]): 累计奖励列表
currentProgress (int): 当前进度
totalMilestones (int): 总里程碑数
consecutiveDays (int, optional): 连续签到天数
Returns:
dict[str, Any]: 签到数据字典
"""
# 计算进度百分比
progressPercentage = min(100, int((currentProgress / totalMilestones) * 100))
# 计算已达成里程碑数量
achievedMilestones = 0
for reward in cumulativeRewards:
if reward.get("achieved", False):
achievedMilestones += 1
return {
"currentDate": datetime.now().strftime("%Y年%m月%d日"),
"todayRewards": todayRewards,
"cumulativeRewards": cumulativeRewards,
"currentProgress": currentProgress,
"totalMilestones": totalMilestones,
"progressPercentage": progressPercentage,
"achievedMilestones": achievedMilestones,
"consecutiveDays": consecutiveDays,
}
async def createSignImage(
self,
todayRewards: list[dict[str, Any]],
cumulativeRewards: list[dict[str, Any]],
currentProgress: int,
totalMilestones: int,
consecutiveDays: int = 0,
):
"""创建签到图片的完整流程
Args:
todayRewards (list[dict[str, Any]]): 今日奖励列表
cumulativeRewards (list[dict[str, Any]]): 累计奖励列表
currentProgress (int): 当前进度
totalMilestones (int): 总里程碑数
consecutiveDays (int, optional): 连续签到天数
"""
try:
templatePath = g_sResourcePath / "html/sign_in.html"
outputPath = g_sResourcePath / "temp_html/sign_in.html"
savePath = DATA_PATH / "farm_res/html/sign_in.png"
# 1. 生成渲染数据
context = self.generateSignData(
todayRewards=todayRewards,
cumulativeRewards=cumulativeRewards,
currentProgress=currentProgress,
totalMilestones=totalMilestones,
consecutiveDays=consecutiveDays,
)
# 2. 渲染HTML文件
g_pToolManager.renderHtmlToFile(templatePath, context, outputPath)
# 3. 生成图片字节数据
# imageBytes = await g_pToolManager.screenshotHtmlToBytes(outputPath)
# 4. 保存图片文件
await g_pToolManager.screenshotSave(
str(outputPath), str(savePath), 600, 800
)
except Exception as e:
logger.error(f"❌ 生成签到图片时出错: {e}")
raise
g_pSignInManager = CSignInManager()
core/database/userSign.py
+7 -5
View File
@@ -121,7 +121,7 @@ class CUserSignDB(CSqlManager):
signDate (str): 日期字符串 'YYYY-MM-DD' 不传默认当前系统日期
Returns:
bool: 0: 签到失败 1: 签到成功 2: 重复签到
int: 0: 签到失败 1: 签到成功 2: 重复签到
"""
try:
player = await g_pToolManager.getPlayerByUid(uid)
@@ -136,9 +136,10 @@ class CUserSignDB(CSqlManager):
todayStr = g_pToolManager.dateTime().date().today().strftime("%Y-%m-%d")
isSupplement = 0 if signDate == todayStr else 1
sign = await g_pJsonManager.getSign()
expMax, expMin, pointMax, pointMin = [
g_pJsonManager.m_pSign.get(key, default)
sign.get(key, default)
for key, default in (
("exp_max", 50),
("exp_min", 5),
@@ -169,13 +170,14 @@ class CUserSignDB(CSqlManager):
if row["currentMonth"] == currentMonth
else 1
)
lastDate = row["lastSignDate"]
prevDate = (
g_pToolManager.dateTime().strptime(signDate, "%Y-%m-%d")
- timedelta(days=1)
).strftime("%Y-%m-%d")
continuousDays = (
row["continuousDays"] + 1 if lastDate == prevDate else 1
row["continuousDays"] + 1
if row["lastSignDate"] == prevDate
else 1
)
supplementCount = (
row["supplementCount"] + 1
@@ -222,7 +224,7 @@ class CUserSignDB(CSqlManager):
)
# 计算累签奖励
reward = g_pJsonManager.m_pSign["continuou"].get(f"{monthSignDays}", None)
reward = sign["continuou"].get(f"{monthSignDays}", None)
if reward:
point += reward.get("point", 0)
exp += reward.get("exp", 0)
core/farm.py
+11 -6
View File
@@ -69,12 +69,13 @@ class CFarmManager:
"""
img = BuildImage(background=g_sResourcePath / "background/background.jpg")
soilSize = g_pJsonManager.m_pSoil["size"]
soil = await g_pJsonManager.getSoil()
soilSize = soil["size"]
soilPos = soil["soil"]
grass = BuildImage(background=g_sResourcePath / "soil/草土地.png")
await grass.resize(0, soilSize[0], soilSize[1])
soilPos = g_pJsonManager.m_pSoil["soil"]
player = await g_pToolManager.getPlayerByUid(uid)
if not player:
return img.pic2bytes()
@@ -987,7 +988,8 @@ class CFarmManager:
Returns:
str: 返回条件文本信息
"""
rec = g_pJsonManager.m_pLevel["reclamation"]
jLevel = await g_pJsonManager.getLevel()
rec = jLevel["reclamation"]
player = await g_pToolManager.getPlayerByUid(uid)
if not player:
return g_sTranslation["basic"]["error"]
@@ -1031,7 +1033,8 @@ class CFarmManager:
return g_sTranslation["basic"]["error"]
level = await player.getUserLevel()
rec = g_pJsonManager.m_pLevel["reclamation"]
jsonLevel = await g_pJsonManager.getLevel()
rec = jsonLevel["reclamation"]
try:
if player.user["soil"] >= 30:
@@ -1084,7 +1087,8 @@ class CFarmManager:
# 获取升级所需
soilLevelText = await g_pDBService.userSoil.getSoilLevel(soilLevel)
fileter = g_pJsonManager.m_pSoil["upgrade"][soilLevelText][countSoil]
jSoil = await g_pJsonManager.getSoil()
fileter = jSoil["upgrade"][soilLevelText][countSoil]
nextLevel = await g_pDBService.userSoil.getSoilLevelText(soilLevel)
@@ -1134,7 +1138,8 @@ class CFarmManager:
countSoil = await g_pDBService.userSoil.countSoilByLevel(uid, soilLevel)
soilLevelText = await g_pDBService.userSoil.getSoilLevel(soilLevel)
fileter = g_pJsonManager.m_pSoil["upgrade"][soilLevelText][countSoil]
soil = await g_pJsonManager.getSoil()
fileter = soil["upgrade"][soilLevelText][countSoil]
getters = {
"level": (await player.getUserLevel())[0],
core/help.py
+5 -89
View File
@@ -1,97 +1,11 @@
from pathlib import Path
from jinja2 import Template
from playwright.async_api import async_playwright
from zhenxun.configs.path_config import DATA_PATH
from zhenxun.services.log import logger
from ..utils.config import g_sResourcePath
from ..utils.tool import g_pToolManager
class CHelpManager:
@classmethod
def renderHtmlToFile(
cls, path: Path | str, context: dict, output: Path | str
) -> None:
"""
使用 Jinja2 渲染 HTML 模板并保存到指定文件，会自动创建父目录
Args:
path (str): 模板 HTML 路径
context (dict): 用于渲染的上下文字典
output (str): 输出 HTML 文件路径
"""
templatePath = str(path)
outputPath = str(output)
templateStr = Path(templatePath).read_text(encoding="utf-8")
template = Template(templateStr)
rendered = template.render(**context)
# 自动创建目录
Path(outputPath).parent.mkdir(parents=True, exist_ok=True)
Path(outputPath).write_text(rendered, encoding="utf-8")
@classmethod
async def screenshotHtmlToBytes(cls, path: str) -> bytes:
"""
使用 Playwright 截图本地 HTML 文件并返回 PNG 图片字节数据
Args:
path (str): 本地 HTML 文件路径
Returns:
bytes: PNG 图片的原始字节内容
"""
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page(
viewport={"width": 1200, "height": 900}, device_scale_factor=1
)
file_url = Path(path).resolve().as_uri()
await page.goto(file_url, wait_until="networkidle")
await page.evaluate("""() => {
return new Promise(r => setTimeout(r, 200));
}""")
image_bytes = await page.screenshot(full_page=True)
await browser.close()
return image_bytes
@classmethod
async def screenshotSave(
cls, path: str, save: str, width: int, height: int
) -> None:
"""
使用 Playwright 渲染本地 HTML 并将截图保存到指定路径
Args:
path (str): HTML 文件路径
save (str): PNG 保存路径（如 output/image.png）
width (int): 图片宽度
height (int): 图片高度
"""
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page(
viewport={"width": width, "height": height}, device_scale_factor=1
)
file_url = Path(path).resolve().as_uri()
await page.goto(file_url, wait_until="networkidle")
await page.evaluate("""() => {
return new Promise(r => setTimeout(r, 200));
}""")
# 确保保存目录存在
Path(save).parent.mkdir(parents=True, exist_ok=True)
# 截图并保存到本地文件
await page.screenshot(path=save, full_page=True)
await browser.close()
@classmethod
async def createHelpImage(cls) -> bool:
templatePath = g_sResourcePath / "html/help.html"
@@ -110,9 +24,11 @@ class CHelpManager:
}
try:
cls.renderHtmlToFile(templatePath, context, outputPath)
g_pToolManager.renderHtmlToFile(templatePath, context, outputPath)
bytes = await cls.screenshotSave(str(outputPath), str(savePath), 1500, 2300)
bytes = await g_pToolManager.screenshotSave(
str(outputPath), str(savePath), 1500, 2300
)
except Exception as e:
logger.warning("绘制农场帮助菜单失败", e=e)
return False
resource/background/background.png
BIN
View File
Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 5.5 MiB

utils/json.py
+65 -23
View File
@@ -8,22 +8,59 @@ from .request import g_pRequestManager
class CJsonManager:
def __init__(self):
self.m_pItem = {}
self.m_pLevel = {}
self.m_pSoil = {}
self.m_pSign = {}
self._m_pItem = {}
self._m_pLevel = {}
self._m_pSoil = {}
self._m_pSign = {}
async def init(self) -> bool:
if not await self.initItem():
return False
self._m_bItemLoaded = False
self._m_bLevelLoaded = False
self._m_bSoilLoaded = False
self._m_bSignLoaded = False
if not await self.initLevel():
return False
async def getItem(self) -> dict:
if not self._m_bItemLoaded:
success = await self.initItem()
if not success:
return {}
return self._m_pItem
if not await self.initSoil():
return False
async def getLevel(self) -> dict:
if not self._m_bLevelLoaded:
success = await self.initLevel()
if not success:
return {}
return self._m_pLevel
return await self.initSignInFile()
async def getSoil(self) -> dict:
if not self._m_bSoilLoaded:
success = await self.initSoil()
if not success:
return {}
return self._m_pSoil
async def getSign(self) -> dict:
if not self._m_bSignLoaded:
success = await self.initSign()
if not success:
return {}
return self._m_pSign
async def reloadItem(self) -> bool:
self._m_bItemLoaded = False
return await self.initItem()
async def reloadLevel(self) -> bool:
self._m_bLevelLoaded = False
return await self.initLevel()
async def reloadSoil(self) -> bool:
self._m_bSoilLoaded = False
return await self.initSoil()
async def reloadSign(self) -> bool:
self._m_bSignLoaded = False
return await self.initSign()
async def initItem(self) -> bool:
try:
@@ -31,14 +68,16 @@ class CJsonManager:
config.g_sConfigPath / "item.json",
encoding="utf-8",
) as file:
self.m_pItem = json.load(file)
self._m_pItem = json.load(file)
self._m_bItemLoaded = True
return True
except FileNotFoundError:
logger.warning("item.json 打开失败")
self._m_pItem = {}
return False
except json.JSONDecodeError as e:
logger.warning(f"item.json JSON格式错误: {e}")
self._m_pItem = {}
return False
async def initLevel(self) -> bool:
@@ -47,14 +86,16 @@ class CJsonManager:
config.g_sConfigPath / "level.json",
encoding="utf-8",
) as file:
self.m_pLevel = json.load(file)
self._m_pLevel = json.load(file)
self._m_bLevelLoaded = True
return True
except FileNotFoundError:
logger.warning("level.json 打开失败")
self._m_pLevel = {}
return False
except json.JSONDecodeError as e:
logger.warning(f"level.json JSON格式错误: {e}")
self._m_pLevel = {}
return False
async def initSoil(self) -> bool:
@@ -63,26 +104,25 @@ class CJsonManager:
config.g_sConfigPath / "soil.json",
encoding="utf-8",
) as file:
self.m_pSoil = json.load(file)
self._m_pSoil = json.load(file)
self._m_bSoilLoaded = True
return True
except FileNotFoundError:
logger.warning("soil.json 打开失败")
self._m_pSoil = {}
return False
except json.JSONDecodeError as e:
logger.warning(f"soil.json JSON格式错误: {e}")
self._m_pSoil = {}
return False
async def initSignInFile(self) -> bool:
if not await g_pRequestManager.initSignInFile():
config.g_bSignStatus = False
return False
else:
result = await self.initSign()
config.g_bSignStatus = result
return result
async def initSign(self) -> bool:
@@ -91,14 +131,16 @@ class CJsonManager:
config.g_sSignInPath,
encoding="utf-8",
) as file:
self.m_pSign = json.load(file)
self._m_pSign = json.load(file)
self._m_bSignLoaded = True
return True
except FileNotFoundError:
logger.warning("sign_in.json 打开失败")
self._m_pSign = {}
return False
except json.JSONDecodeError as e:
logger.warning(f"sign_in.json JSON格式错误: {e}")
self._m_pSign = {}
return False
utils/tool.py
+86
View File
@@ -1,7 +1,11 @@
from datetime import datetime
import os
from pathlib import Path
from zoneinfo import ZoneInfo
from jinja2 import Template
from playwright.async_api import async_playwright
from zhenxun.services.log import logger
from zhenxun.utils.message import MessageUtils
@@ -107,5 +111,87 @@ class CToolManager:
tz = ZoneInfo("Asia/Shanghai")
return datetime.now(tz)
@classmethod
def renderHtmlToFile(
cls, path: Path | str, context: dict, output: Path | str
) -> None:
"""
使用 Jinja2 渲染 HTML 模板并保存到指定文件，会自动创建父目录
Args:
path (str): 模板 HTML 路径
context (dict): 用于渲染的上下文字典
output (str): 输出 HTML 文件路径
"""
templatePath = str(path)
outputPath = str(output)
templateStr = Path(templatePath).read_text(encoding="utf-8")
template = Template(templateStr)
rendered = template.render(**context)
# 自动创建目录
Path(outputPath).parent.mkdir(parents=True, exist_ok=True)
Path(outputPath).write_text(rendered, encoding="utf-8")
@classmethod
async def screenshotHtmlToBytes(cls, path: str) -> bytes:
"""
使用 Playwright 截图本地 HTML 文件并返回 PNG 图片字节数据
Args:
path (str): 本地 HTML 文件路径
Returns:
bytes: PNG 图片的原始字节内容
"""
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page(
viewport={"width": 1200, "height": 900}, device_scale_factor=1
)
file_url = Path(path).resolve().as_uri()
await page.goto(file_url, wait_until="networkidle")
await page.evaluate("""() => {
return new Promise(r => setTimeout(r, 200));
}""")
image_bytes = await page.screenshot(full_page=True)
await browser.close()
return image_bytes
@classmethod
async def screenshotSave(
cls, path: str, save: str, width: int, height: int
) -> None:
"""
使用 Playwright 渲染本地 HTML 并将截图保存到指定路径
Args:
path (str): HTML 文件路径
save (str): PNG 保存路径（如 output/image.png）
width (int): 图片宽度
height (int): 图片高度
"""
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page(
viewport={"width": width, "height": height}, device_scale_factor=1
)
file_url = Path(path).resolve().as_uri()
await page.goto(file_url, wait_until="networkidle")
await page.evaluate("""() => {
return new Promise(r => setTimeout(r, 200));
}""")
# 确保保存目录存在
Path(save).parent.mkdir(parents=True, exist_ok=True)
# 截图并保存到本地文件
await page.screenshot(path=save, full_page=True)
await browser.close()
g_pToolManager = CToolManager()