mirror of
https://github.com/Shu-Ying/main_plugin_farm.git
synced 2026-11-16 01:40:07 +08:00
update # 初始化仓库
This commit is contained in:
@@ -0,0 +1,41 @@
|
||||
from . import utils
|
||||
|
||||
|
||||
def getConfig():
|
||||
from . import config
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def getDrawFarm():
|
||||
from .darw.drawFarm import CDarwFarm
|
||||
|
||||
return CDarwFarm
|
||||
|
||||
|
||||
def getRequestManager():
|
||||
from .request import CRequestManager
|
||||
|
||||
return CRequestManager
|
||||
|
||||
|
||||
def getToolManager():
|
||||
from .tool import CToolManager
|
||||
|
||||
return CToolManager
|
||||
|
||||
|
||||
def getJsonManager():
|
||||
from .json import g_pJsonManager
|
||||
|
||||
return g_pJsonManager
|
||||
|
||||
|
||||
__all__ = [
|
||||
"getConfig",
|
||||
"getDrawFarm",
|
||||
"getJsonManager",
|
||||
"getRequestManager",
|
||||
"getToolManager",
|
||||
"utils",
|
||||
]
|
||||
@@ -0,0 +1,135 @@
|
||||
from pathlib import Path
|
||||
|
||||
from zhenxun.configs.path_config import DATA_PATH
|
||||
|
||||
# 签到状态
|
||||
g_bSignStatus = True
|
||||
|
||||
# 是否处于Debug模式
|
||||
g_bIsDebug = False
|
||||
|
||||
# 数据库文件目录
|
||||
g_sDBPath = DATA_PATH / "farm_db"
|
||||
|
||||
# 数据库文件路径
|
||||
g_sDBFilePath = DATA_PATH / "farm_db/farm.db"
|
||||
|
||||
# 农场资源文件目录
|
||||
g_sResourcePath = Path(__file__).resolve().parent / "../resource"
|
||||
|
||||
# 农场作物数据库
|
||||
g_sPlantPath = g_sResourcePath / "db/plant.db"
|
||||
|
||||
# 农场配置文件目录
|
||||
g_sConfigPath = Path(__file__).resolve().parent / "../config"
|
||||
|
||||
# 农场签到文件路径
|
||||
g_sSignInPath = g_sConfigPath / "sign_in.json"
|
||||
|
||||
# 土地等级上限
|
||||
g_iSoilLevelMax = 3
|
||||
|
||||
# 农场同一文本
|
||||
g_sTranslation = {
|
||||
"basic": {
|
||||
"notFarm": "尚未开通农场，快at我发送 开通农场 开通吧 🌱🚜",
|
||||
"point": "你的当前农场币为: {point} 🌾💰",
|
||||
"vipPoint": "你的当前点券为: {vipPoint} 🌾💰",
|
||||
"error": "❌ 农场功能异常，请稍后再试 💔",
|
||||
},
|
||||
"register": {
|
||||
"success": "✅ 农场开通成功！\n💼 初始资金：{point}农场币 🥳🎉",
|
||||
"repeat": "🎉 您已经开通农场啦~ 😄",
|
||||
"error": "⚠️ 开通失败，请稍后再试 💔",
|
||||
"already": "🚜 你已经开通农场啦~ 快去种田吧 🌱",
|
||||
},
|
||||
"buySeed": {
|
||||
"notSeed": "🌱 请在指令后跟需要购买的种子名称",
|
||||
"notNum": "❗️ 请输入购买数量！",
|
||||
"noLevel": "🔒 你的等级不够哦，努努力吧 💪",
|
||||
"noPoint": "💰 你的农场币不够哦~ 快速速氪金吧！💸",
|
||||
"noVipPoint": "💰 你的点券不够哦~ 快速速氪金吧！💸",
|
||||
"success": "✅ 成功购买{name}，花费{total}农场币，剩余{point}农场币 🌾",
|
||||
"vipSuccess": "✅ 成功购买{name}，花费{total}点券，剩余{point}点券 🌾",
|
||||
"errorSql": "❌ 购买失败，执行数据库错误！🛑",
|
||||
"error": "❌ 购买出错！请检查需购买的种子名称！🔍",
|
||||
},
|
||||
"sowing": {
|
||||
"notSeed": "🌱 请在指令后跟需要播种的种子名称",
|
||||
"noSeed": "❌ 没有在你的仓库发现{name}种子，快去买点吧！🛒",
|
||||
"noNum": "⚠️ 播种失败！仓库中的{name}种子数量不足，当前剩余{num}个种子 🍂",
|
||||
"success": "✅ 播种{name}成功！仓库剩余{num}个种子 🌱",
|
||||
"success2": "✅ 播种数量超出开垦土地数量，已将可播种土地成功播种{name}！仓库剩余{num}个种子 🌾",
|
||||
"error": "❌ 播种失败，请稍后重试！⏳",
|
||||
},
|
||||
"harvest": {
|
||||
"append": "🌾 收获作物：{name}，数量为：{num}，经验为：{exp} ✨",
|
||||
"exp": "✨ 累计获得经验：{exp} 📈",
|
||||
"no": "🤷♂️ 没有可收获的作物哦~ 不要试图拔苗助长 🚫",
|
||||
"error": "❌ 收获失败，请稍后重试！⏳",
|
||||
},
|
||||
"eradicate": {
|
||||
"success": "🗑️ 成功铲除荒废作物，累计获得经验：{exp} ✨",
|
||||
"error": "❌ 没有可以铲除的作物 🚜",
|
||||
},
|
||||
"reclamation": {
|
||||
"confirm": "⚠️ 回复“是”将执行开垦 ⛏️",
|
||||
"timeOut": "⏰ 等待开垦回复超时，请重试",
|
||||
"perfect": "🌟 你已经开垦了全部土地 🎉",
|
||||
"next": "🔜 下次开垦所需条件：等级：{level}，农场币：{num} 💰",
|
||||
"next2": "🔜 下次开垦所需条件：等级：{level}，农场币：{num}，物品：{item} 📦",
|
||||
"nextLevel": "📈 当前用户等级{level}，升级所需等级为{next} ⏳",
|
||||
"noNum": "💰 当前用户农场币不足，升级所需农场币为{num} 💸",
|
||||
"success": "✅ 开垦土地成功！🌱",
|
||||
"error": "❌ 获取开垦土地条件失败！",
|
||||
"error1": "❌ 执行开垦失败！",
|
||||
"error2": "❌ 未知错误{e}💥",
|
||||
},
|
||||
"sellPlant": {
|
||||
"no": "🤷♀️ 你仓库没有可以出售的作物 🌾",
|
||||
"success": "💰 成功出售所有作物，获得农场币：{point}，当前农场币：{num} 🎉",
|
||||
"success1": "💰 成功出售{name}，获得农场币：{point}，当前农场币：{num} 🥳",
|
||||
"error": "❌ 出售作物{name}出错：仓库中不存在该作物 🚫",
|
||||
"error1": "❌ 出售作物{name}出错：数量不足 ⚠️",
|
||||
},
|
||||
"stealing": {
|
||||
"noTarget": "🎯 请在指令后跟需要at的人 👤",
|
||||
"targetNotFarm": "🚜 目标尚未开通农场，快邀请ta开通吧 😉",
|
||||
"max": "❌ 你今天可偷次数到达上限啦，手下留情吧 🙏",
|
||||
"info": "🤫 成功偷到作物：{name}，数量为：{num} 🍒",
|
||||
"noPlant": "🌱 目标没有作物可以被偷 🌾",
|
||||
"repeat": "🚫 你已经偷过目标啦，请手下留情 🙏",
|
||||
},
|
||||
"changeName": {
|
||||
"noName": "✏️ 请在指令后跟需要更改的农场名",
|
||||
"success": "✅ 更新农场名成功！🎉",
|
||||
"error": "❌ 农场名不支持特殊符号！🚫",
|
||||
"error1": "❌ 更新农场名失败！💔",
|
||||
},
|
||||
"signIn": {
|
||||
"success": "📝 签到成功！累计签到天数：{day}\n🎁 获得经验{exp}，获得金币{num} 💰",
|
||||
"grandTotal": "\n🎉 成功领取累计签到奖励：\n✨ 额外获得经验{exp}，额外获得金币{num} 🥳",
|
||||
"grandTotal1": "，额外获得点券{num} 🎫",
|
||||
"grandTotal2": "\n🌱 获得{name}种子 * {num} 🌟",
|
||||
"error": "❗️ 签到功能异常！",
|
||||
"error1": "❌ 签到失败！未知错误 💔",
|
||||
},
|
||||
"soilInfo": {
|
||||
"success": "土地成功升级至{name}，效果为：{text}",
|
||||
"timeOut": "等待土地升级回复超时，请重试",
|
||||
"error": "土地信息尚未查询到",
|
||||
"error1": "该土地已经升至满级啦~",
|
||||
"red": "增产+10%",
|
||||
"black": "增产+20% 时间-20%",
|
||||
"gold": "增产+28% 经验+28% 时间-20%",
|
||||
"amethyst": "增产+30% 经验+30% 时间-25% 幸运+1%",
|
||||
"aquamarine": "增产+32% 经验+32% 时间-28% 幸运+1%",
|
||||
"blackcrystal": "增产+32% 经验+40% 时间-28% 幸运+2%",
|
||||
},
|
||||
"lockPlant": {
|
||||
"noPlant": "很抱歉，你的作物仓库中没有该作物",
|
||||
"lockPlant": "加锁成功，现在{name}不会被一键售卖了",
|
||||
"unlockPlant": "解锁成功，现在{name}会被一键售卖了",
|
||||
"error": "未知错误",
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
from .drawFarm import CDarwFarm
|
||||
|
||||
__all__ = ["CDarwFarm"]
|
||||
@@ -0,0 +1,491 @@
|
||||
from zhenxun.configs.config import Config
|
||||
from zhenxun.services.log import logger
|
||||
from zhenxun.utils._build_image import BuildImage
|
||||
from zhenxun.utils.image_utils import ImageTemplate
|
||||
from zhenxun.utils.platform import PlatformUtils
|
||||
|
||||
from ...core import CPlayer
|
||||
from ...core.dbService import g_pDBService
|
||||
from .. import config, getJsonManager, getToolManager, utils
|
||||
|
||||
|
||||
class CDarwFarm:
|
||||
@classmethod
|
||||
async def drawFarmByUid(cls, uid: str) -> bytes:
|
||||
"""绘制用户农场
|
||||
|
||||
Args:
|
||||
uid (str): 用户UID
|
||||
|
||||
Returns:
|
||||
bytes: 返回绘制结果
|
||||
"""
|
||||
dynamic = Config.get_config("zhenxun_plugin_farm", "动态背景")
|
||||
|
||||
img, soilSize, soilPos, grass = await cls.createBaseCanvas(dynamic)
|
||||
|
||||
player = await getToolManager().getPlayerByUid(uid)
|
||||
if not player:
|
||||
return img.pic2bytes()
|
||||
|
||||
soilUnlock = int(player.user.get("soil", 3))
|
||||
|
||||
isFirstExpansion = True # 首次添加扩建图片
|
||||
isFirstRipe = True
|
||||
|
||||
for index in range(0, 30):
|
||||
x = soilPos[str(index + 1)]["x"]
|
||||
y = soilPos[str(index + 1)]["y"]
|
||||
|
||||
# 已解锁格子
|
||||
if index < soilUnlock:
|
||||
isRipe = await cls.drawUnlockedTile(img, uid, index + 1, x, y, soilSize)
|
||||
if isRipe and isFirstRipe:
|
||||
await cls.pasteRipeOnce(img, x, y, soilSize)
|
||||
isFirstRipe = False
|
||||
else:
|
||||
addedExpansion = await cls.drawLockedTile(
|
||||
img, x, y, soilSize, grass, isFirstExpansion
|
||||
)
|
||||
if addedExpansion:
|
||||
isFirstExpansion = False
|
||||
|
||||
await cls.drawUserInfo(img, player, uid)
|
||||
|
||||
await cls.applyDefinitionResize(img)
|
||||
|
||||
return img.pic2bytes()
|
||||
|
||||
@classmethod
|
||||
async def createBaseCanvas(
|
||||
cls, dynamic: bool
|
||||
) -> tuple[BuildImage, tuple, dict, BuildImage]:
|
||||
"""创建基础画布并返回 (img, soilSize, soilPos, grass)
|
||||
|
||||
Returns:
|
||||
tuple: (主画布, 土地尺寸, 土地坐标映射, 草地图)
|
||||
"""
|
||||
|
||||
if dynamic:
|
||||
season = getToolManager().getSeason()
|
||||
|
||||
backgroundPath = config.g_sResourcePath / "background/decorateIcon"
|
||||
|
||||
# 节日优先级更高
|
||||
if season == utils.Season.SPRING:
|
||||
bgFile = backgroundPath / "decorate_spring_1.jpg"
|
||||
elif season == utils.Season.SUMMER:
|
||||
bgFile = backgroundPath / "decorate_summer_1.jpg"
|
||||
elif season == utils.Season.AUTUMN:
|
||||
bgFile = backgroundPath / "decorate_autumn_1.jpg"
|
||||
elif season == utils.Season.WINTER:
|
||||
bgFile = backgroundPath / "decorate_winter_1.jpg"
|
||||
else:
|
||||
bgFile = backgroundPath / "decorate_default_1.jpg"
|
||||
|
||||
img = BuildImage(background=bgFile)
|
||||
else:
|
||||
img = BuildImage(
|
||||
background=config.g_sResourcePath
|
||||
/ "background/decorateIcon/decorate_default_1jpg"
|
||||
)
|
||||
|
||||
soil = await getJsonManager().getSoil()
|
||||
soilSize = soil["size"]
|
||||
soilPos = soil["soil"]
|
||||
|
||||
grass = BuildImage(background=config.g_sResourcePath / "soil/草土地.png")
|
||||
await grass.resize(0, soilSize[0], soilSize[1])
|
||||
|
||||
return img, soilSize, soilPos, grass
|
||||
|
||||
@classmethod
|
||||
async def drawUnlockedTile(
|
||||
cls, img: BuildImage, uid: str, soilIndex: int, x: int, y: int, soilSize
|
||||
) -> bool:
|
||||
"""绘制已解锁土地格（包含土地贴图与作物），返回该格是否可收获
|
||||
|
||||
Args:
|
||||
img (BuildImage): 主画布
|
||||
uid (str): 用户UID
|
||||
soilIndex (int): 土地索引（从1开始）
|
||||
x (int), y (int): 贴图位置
|
||||
soilSize: 土地贴图尺寸（[w,h] 或 (w,h)）
|
||||
|
||||
Returns:
|
||||
bool: 是否有可收获作物（isRipe）
|
||||
"""
|
||||
soilUrl = "soil/普通土地.png"
|
||||
soilInfo = await g_pDBService.userSoil.getUserSoil(uid, soilIndex)
|
||||
|
||||
if soilInfo:
|
||||
soilLevel = soilInfo.get("soilLevel", 0)
|
||||
if soilLevel == 1:
|
||||
soilUrl = "soil/红土地.png"
|
||||
elif soilLevel == 2:
|
||||
soilUrl = "soil/黑土地.png"
|
||||
elif soilLevel == 3:
|
||||
soilUrl = "soil/金土地.png"
|
||||
|
||||
soil = BuildImage(background=config.g_sResourcePath / soilUrl)
|
||||
await soil.resize(0, soilSize[0], soilSize[1])
|
||||
await img.paste(soil, (x, y))
|
||||
|
||||
isPlant, plant, isRipe, offsetX, offsetY = await cls.drawSoilPlant(
|
||||
uid, soilIndex
|
||||
)
|
||||
if isPlant and plant is not None:
|
||||
await img.paste(
|
||||
plant,
|
||||
(
|
||||
x + soilSize[0] // 2 - plant.width // 2 + offsetX,
|
||||
y + soilSize[1] // 2 - plant.height // 2 + offsetY,
|
||||
),
|
||||
)
|
||||
|
||||
return bool(isRipe)
|
||||
|
||||
@classmethod
|
||||
async def drawLockedTile(
|
||||
cls,
|
||||
img: BuildImage,
|
||||
x: int,
|
||||
y: int,
|
||||
soilSize,
|
||||
grass: BuildImage,
|
||||
isFirstExpansion: bool,
|
||||
) -> bool:
|
||||
"""绘制未解锁格子并在首次出现时绘制扩建图标
|
||||
|
||||
Returns:
|
||||
bool: 是否在此格绘制了扩建图标（仅首次返回 True）
|
||||
"""
|
||||
await img.paste(grass, (x, y))
|
||||
|
||||
if isFirstExpansion:
|
||||
expansion = BuildImage(
|
||||
background=config.g_sResourcePath / "background/expansion.png"
|
||||
)
|
||||
await expansion.resize(0, 69, 69)
|
||||
await img.paste(
|
||||
expansion,
|
||||
(
|
||||
x + soilSize[0] // 2 - expansion.width // 2,
|
||||
y + soilSize[1] // 2 - expansion.height,
|
||||
),
|
||||
)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def pasteRipeOnce(cls, img: BuildImage, x: int, y: int, soilSize):
|
||||
"""首次发现成熟作物时贴上成熟提示图（ripe）"""
|
||||
ripe = BuildImage(background=config.g_sResourcePath / "background/ripe.png")
|
||||
await img.paste(
|
||||
ripe,
|
||||
(x + soilSize[0] // 2 - ripe.width // 2, y - ripe.height // 2),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
async def drawUserInfo(cls, img: BuildImage, player: CPlayer, uid: str):
|
||||
"""绘制左上角用户信息（头像、框、昵称、经验、等级、货币等）
|
||||
TODO 需要详细的绘制头像框
|
||||
|
||||
Args:
|
||||
img (BuildImage): 主画布
|
||||
player(CPlayer): 用户
|
||||
uid (str): 用户Uid
|
||||
"""
|
||||
# 背景板
|
||||
background = BuildImage(
|
||||
background=config.g_sResourcePath / "background/userInfo.png"
|
||||
)
|
||||
await img.paste(background, (100, 30))
|
||||
|
||||
# 头像
|
||||
image = await PlatformUtils.get_user_avatar(uid, "qq")
|
||||
if image:
|
||||
avatar = BuildImage(background=image)
|
||||
await avatar.resize(0, 140, 150)
|
||||
await img.paste(avatar, (125, 85))
|
||||
|
||||
# 头像框
|
||||
frame = BuildImage(
|
||||
background=config.g_sResourcePath / "background/frame/frame.png"
|
||||
)
|
||||
await img.paste(frame, (75, 44))
|
||||
|
||||
nameImg = await BuildImage.build_text_image(
|
||||
player.user["name"], size=24, font_color=(77, 35, 4)
|
||||
)
|
||||
await img.paste(nameImg, (300, 92))
|
||||
|
||||
level = await player.getUserLevel()
|
||||
beginX = 309
|
||||
endX = 627
|
||||
width = int((level[2] / level[1]) * (endX - beginX)) if level[1] else 0
|
||||
await img.rectangle((beginX, 188, beginX + width, 222), (171, 194, 41))
|
||||
|
||||
# 经验
|
||||
expImg = await BuildImage.build_text_image(
|
||||
f"{level[2]} / {level[1]}", size=24, font_color=(102, 120, 19)
|
||||
)
|
||||
await img.paste(expImg, (390, 193))
|
||||
|
||||
# 等级
|
||||
levelImg = await BuildImage.build_text_image(
|
||||
str(level[0]), size=32, font_color=(214, 111, 1)
|
||||
)
|
||||
await img.paste(levelImg, (660, 187))
|
||||
|
||||
# 农场币
|
||||
pointImg = await BuildImage.build_text_image(
|
||||
str(player.user.get("point", 0)), size=24, font_color=(253, 253, 253)
|
||||
)
|
||||
await img.paste(pointImg, (330, 255))
|
||||
|
||||
# 点券
|
||||
bondsImg = await BuildImage.build_text_image(
|
||||
"0", size=24, font_color=(253, 253, 253)
|
||||
)
|
||||
await img.paste(bondsImg, (570, 255))
|
||||
|
||||
@classmethod
|
||||
async def drawUserDecorateIcon(cls, img: BuildImage, player: CPlayer) -> bool:
|
||||
"""绘制用户装饰图标
|
||||
|
||||
Args:
|
||||
img (BuildImage): 主画布
|
||||
player (CPlayer): 用户
|
||||
|
||||
Returns:
|
||||
bool: 是否成功绘制
|
||||
"""
|
||||
|
||||
decorateIconId = player.user.get("decorateIcon", 0)
|
||||
if decorateIconId <= 0:
|
||||
return False
|
||||
|
||||
iconPath = (
|
||||
config.g_sResourcePath / f"decorateIcon/DecorateIcon{decorateIconId}.png"
|
||||
)
|
||||
if not iconPath.exists():
|
||||
return False
|
||||
|
||||
icon = BuildImage(background=iconPath)
|
||||
await icon.resize(0, 100, 100)
|
||||
await img.paste(icon, (x, y))
|
||||
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
async def applyDefinitionResize(cls, img: BuildImage):
|
||||
"""根据配置调整绘制清晰度（original/medium/hight/其他）"""
|
||||
definition = Config.get_config("zhenxun_plugin_farm", "绘制农场清晰度")
|
||||
if definition == "medium":
|
||||
await img.resize(0.6)
|
||||
elif definition == "hight":
|
||||
await img.resize(0.8)
|
||||
elif definition == "original":
|
||||
pass
|
||||
else:
|
||||
await img.resize(0.4)
|
||||
|
||||
@classmethod
|
||||
async def drawDetailFarmByUid(cls, uid: str) -> list:
|
||||
"""绘制用户详细信息
|
||||
|
||||
Args:
|
||||
uid (str): 用户Uid
|
||||
|
||||
Returns:
|
||||
list:
|
||||
"""
|
||||
|
||||
info = []
|
||||
farm = await cls.drawFarmByUid(uid)
|
||||
player = await getToolManager().getPlayerByUid(uid)
|
||||
if not player:
|
||||
return info
|
||||
|
||||
info.append(BuildImage.open(farm))
|
||||
|
||||
dataList = []
|
||||
columnName = [
|
||||
"-",
|
||||
"土地ID",
|
||||
"土地等级",
|
||||
"作物名称",
|
||||
"成熟时间",
|
||||
"土地状态",
|
||||
"被偷数量",
|
||||
"剩余产出",
|
||||
]
|
||||
|
||||
icon = ""
|
||||
soilNumber = player.user.get("soil", 3)
|
||||
|
||||
for i in range(1, soilNumber + 1):
|
||||
soilInfo = await g_pDBService.userSoil.getUserSoil(uid, i)
|
||||
|
||||
if not soilInfo:
|
||||
continue
|
||||
|
||||
match soilInfo.get("soilLevel", 0):
|
||||
case 1:
|
||||
name = "红土地.png"
|
||||
case 2:
|
||||
name = "黑土地.png"
|
||||
case 3:
|
||||
name = "金土地.png"
|
||||
case _:
|
||||
name = "普通土地.png"
|
||||
iconPath = config.g_sResourcePath / "soil" / name
|
||||
|
||||
if iconPath.exists():
|
||||
icon = (iconPath, 33, 33)
|
||||
|
||||
plantName = soilInfo.get("plantName", "-")
|
||||
|
||||
if plantName == "-":
|
||||
matureTime = "-"
|
||||
soilStatus = "-"
|
||||
totalNumber = "-"
|
||||
plantNumber = "-"
|
||||
else:
|
||||
matureTime = (
|
||||
getToolManager()
|
||||
.dateTime()
|
||||
.fromtimestamp(int(soilInfo.get("matureTime", 0)))
|
||||
.strftime("%Y-%m-%d %H:%M:%S")
|
||||
)
|
||||
soilStatus = await g_pDBService.userSoil.getUserSoilStatus(uid, i)
|
||||
|
||||
totalNumber = await g_pDBService.userSteal.getTotalStolenCount(uid, i)
|
||||
planInfo = await g_pDBService.plant.getPlantByName(plantName)
|
||||
|
||||
if not planInfo:
|
||||
plantNumber = "None"
|
||||
else:
|
||||
plantNumber = f"{planInfo['harvest'] - totalNumber}"
|
||||
|
||||
dataList.append(
|
||||
[
|
||||
icon,
|
||||
i,
|
||||
await g_pDBService.userSoil.getSoilLevelText(soilInfo["soilLevel"]),
|
||||
plantName,
|
||||
matureTime,
|
||||
soilStatus,
|
||||
totalNumber,
|
||||
plantNumber,
|
||||
]
|
||||
)
|
||||
|
||||
if len(dataList) >= 15:
|
||||
result = await ImageTemplate.table_page(
|
||||
"土地详细信息",
|
||||
"",
|
||||
columnName,
|
||||
dataList,
|
||||
)
|
||||
|
||||
info.append(result.copy())
|
||||
dataList.clear()
|
||||
|
||||
if i >= soilNumber:
|
||||
result = await ImageTemplate.table_page(
|
||||
"土地详细信息",
|
||||
"",
|
||||
columnName,
|
||||
dataList,
|
||||
)
|
||||
|
||||
info.append(result.copy())
|
||||
dataList.clear()
|
||||
|
||||
return info
|
||||
|
||||
@classmethod
|
||||
async def drawSoilPlant(
|
||||
cls, uid: str, soilIndex: int
|
||||
) -> tuple[bool, BuildImage, bool, int, int]:
|
||||
"""绘制植物资源
|
||||
|
||||
Args:
|
||||
uid (str): 用户Uid
|
||||
soilIndex (int): 土地索引 从1开始
|
||||
|
||||
Returns:
|
||||
tuple[bool, BuildImage]: [绘制是否成功，资源图片, 是否成熟]
|
||||
"""
|
||||
|
||||
plant = None
|
||||
soilInfo = await g_pDBService.userSoil.getUserSoil(uid, soilIndex)
|
||||
|
||||
if not soilInfo:
|
||||
return False, None, False, 0, 0 # type: ignore
|
||||
|
||||
# 是否枯萎
|
||||
if int(soilInfo.get("wiltStatus", 0)) == 1:
|
||||
plant = BuildImage(background=config.g_sResourcePath / "plant/basic/9.png")
|
||||
await plant.resize(0, 150, 212)
|
||||
return True, plant, False, 0, 0
|
||||
|
||||
# 获取作物详细信息
|
||||
plantInfo = await g_pDBService.plant.getPlantByName(soilInfo["plantName"])
|
||||
if not plantInfo:
|
||||
logger.error(f"绘制植物资源失败: {soilInfo['plantName']}")
|
||||
return False, None, False, 0, 0 # type: ignore
|
||||
|
||||
offsetX = plantInfo.get("officX", 0)
|
||||
offsetY = plantInfo.get("officY", 0)
|
||||
offsetW = plantInfo.get("officW", 0)
|
||||
offsetH = plantInfo.get("officH", 0)
|
||||
|
||||
currentTime = getToolManager().dateTime().now().timestamp()
|
||||
phaseList = await g_pDBService.plant.getPlantPhaseByName(soilInfo["plantName"])
|
||||
|
||||
# 如果当前时间大于成熟时间 说明作物成熟
|
||||
if currentTime >= soilInfo["matureTime"]:
|
||||
plant = BuildImage(
|
||||
background=config.g_sResourcePath
|
||||
/ f"plant/{soilInfo['plantName']}/{len(phaseList)}.png"
|
||||
)
|
||||
|
||||
return True, plant, True, offsetX, offsetY
|
||||
else:
|
||||
# 如果是多阶段作物 且没有成熟 #早期思路 多阶段作物 直接是倒数第二阶段图片
|
||||
# if soilInfo["harvestCount"] >= 1:
|
||||
# plant = BuildImage(
|
||||
# background=g_sResourcePath
|
||||
# / f"plant/{soilInfo['plantName']}/{plantInfo['phase'] - 1}.png"
|
||||
# )
|
||||
|
||||
# return True, plant, False, offsetX, offsetY
|
||||
|
||||
# 如果没有成熟 则根据当前阶段进行绘制
|
||||
elapsedTime = currentTime - soilInfo["plantTime"]
|
||||
currentStage = currentStage = sum(
|
||||
1 for thr in phaseList if elapsedTime >= thr
|
||||
)
|
||||
|
||||
if currentStage <= 0:
|
||||
if not plantInfo["general"]:
|
||||
plant = BuildImage(
|
||||
background=config.g_sResourcePath
|
||||
/ f"plant/{soilInfo['plantName']}/0.png"
|
||||
)
|
||||
else:
|
||||
plant = BuildImage(
|
||||
background=config.g_sResourcePath / "plant/basic/0.png"
|
||||
)
|
||||
await plant.resize(0, 35 + offsetW, 58 + offsetH)
|
||||
else:
|
||||
plant = BuildImage(
|
||||
background=config.g_sResourcePath
|
||||
/ f"plant/{soilInfo['plantName']}/{currentStage}.png"
|
||||
)
|
||||
|
||||
return True, plant, False, offsetX, offsetY
|
||||
@@ -0,0 +1,146 @@
|
||||
import json
|
||||
|
||||
from zhenxun.services.log import logger
|
||||
|
||||
from . import config, getRequestManager
|
||||
|
||||
|
||||
class CJsonManager:
|
||||
def __init__(self):
|
||||
self._m_pItem = {}
|
||||
self._m_pLevel = {}
|
||||
self._m_pSoil = {}
|
||||
self._m_pSign = {}
|
||||
|
||||
self._m_bItemLoaded = False
|
||||
self._m_bLevelLoaded = False
|
||||
self._m_bSoilLoaded = False
|
||||
self._m_bSignLoaded = False
|
||||
|
||||
async def getItem(self) -> dict:
|
||||
if not self._m_bItemLoaded:
|
||||
success = await self.initItem()
|
||||
if not success:
|
||||
return {}
|
||||
return self._m_pItem
|
||||
|
||||
async def getLevel(self) -> dict:
|
||||
if not self._m_bLevelLoaded:
|
||||
success = await self.initLevel()
|
||||
if not success:
|
||||
return {}
|
||||
return self._m_pLevel
|
||||
|
||||
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:
|
||||
with open(
|
||||
config.g_sConfigPath / "item.json",
|
||||
encoding="utf-8",
|
||||
) as 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:
|
||||
try:
|
||||
with open(
|
||||
config.g_sConfigPath / "level.json",
|
||||
encoding="utf-8",
|
||||
) as 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:
|
||||
try:
|
||||
with open(
|
||||
config.g_sConfigPath / "soil.json",
|
||||
encoding="utf-8",
|
||||
) as 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 getRequestManager().initSignInFile():
|
||||
config.g_bSignStatus = False
|
||||
return False
|
||||
else:
|
||||
result = await self.initSign()
|
||||
config.g_bSignStatus = result
|
||||
return result
|
||||
|
||||
async def initSign(self) -> bool:
|
||||
try:
|
||||
with open(
|
||||
config.g_sSignInPath,
|
||||
encoding="utf-8",
|
||||
) as 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
|
||||
|
||||
|
||||
g_pJsonManager = CJsonManager()
|
||||
@@ -0,0 +1,289 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
import httpx
|
||||
from rich.progress import (
|
||||
BarColumn,
|
||||
DownloadColumn,
|
||||
Progress,
|
||||
TextColumn,
|
||||
TimeRemainingColumn,
|
||||
TransferSpeedColumn,
|
||||
)
|
||||
|
||||
from zhenxun.configs.config import Config
|
||||
from zhenxun.services.log import logger
|
||||
|
||||
from ..core.dbService import g_pDBService
|
||||
from . import config, getToolManager
|
||||
|
||||
|
||||
class CRequestManager:
|
||||
m_sTokens = os.environ.get("FARM_SERVICE_TOKEN", "")
|
||||
|
||||
@classmethod
|
||||
async def download(
|
||||
cls,
|
||||
url: str,
|
||||
savePath: str,
|
||||
fileName: str,
|
||||
params: dict | None = None,
|
||||
jsonData: dict | None = None,
|
||||
) -> bool:
|
||||
"""下载文件到指定路径并覆盖已存在的文件
|
||||
|
||||
Args:
|
||||
url (str): 文件的下载链接
|
||||
savePath (str): 保存文件夹路径
|
||||
fileName (str): 保存后的文件名
|
||||
params (dict | None): 可选的 URL 查询参数
|
||||
jsonData (dict | None): 可选的 JSON 请求体
|
||||
|
||||
Returns:
|
||||
bool: 是否下载成功
|
||||
"""
|
||||
headers = {"token": cls.m_sTokens}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
requestArgs: dict = {"headers": headers}
|
||||
if params:
|
||||
requestArgs["params"] = params
|
||||
if jsonData:
|
||||
requestArgs["json"] = jsonData
|
||||
|
||||
response = await client.request(
|
||||
"GET", url, **requestArgs, follow_redirects=True
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
logger.warning(
|
||||
f"文件下载失败: HTTP {response.status_code} {response.text}"
|
||||
)
|
||||
return False
|
||||
|
||||
totalLength = int(response.headers.get("Content-Length", 0))
|
||||
fullPath = os.path.join(savePath, fileName)
|
||||
os.makedirs(os.path.dirname(fullPath), exist_ok=True)
|
||||
|
||||
with Progress(
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
BarColumn(),
|
||||
DownloadColumn(),
|
||||
TransferSpeedColumn(),
|
||||
TimeRemainingColumn(),
|
||||
transient=True,
|
||||
) as progress:
|
||||
task = progress.add_task(
|
||||
f"[green]【真寻农场】正在下载 {fileName}", total=totalLength
|
||||
)
|
||||
|
||||
with open(fullPath, "wb") as f:
|
||||
async for chunk in response.aiter_bytes(chunk_size=1024):
|
||||
f.write(chunk)
|
||||
progress.advance(task, len(chunk))
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"下载文件异常: {e}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def post(cls, endpoint: str, name: str = "", jsonData: dict = {}) -> dict:
|
||||
"""发送POST请求到指定接口，统一调用，仅支持JSON格式数据
|
||||
|
||||
Args:
|
||||
endpoint (str): 请求的接口路径
|
||||
name (str, optional): 操作名称用于日志记录
|
||||
jsonData (dict): 以JSON格式发送的数据
|
||||
|
||||
Raises:
|
||||
ValueError: 当jsonData未提供时抛出
|
||||
|
||||
Returns:
|
||||
dict: 返回请求结果的JSON数据
|
||||
"""
|
||||
baseUrl = Config.get_config("zhenxun_plugin_farm", "服务地址")
|
||||
url = f"{baseUrl.rstrip('/')}:8998/{endpoint.lstrip('/')}"
|
||||
headers = {"token": cls.m_sTokens}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
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}"
|
||||
)
|
||||
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 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('/')}:8998/{endpoint.lstrip('/')}"
|
||||
headers = {"token": cls.m_sTokens}
|
||||
|
||||
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 initSignInFile(cls) -> bool:
|
||||
if os.path.exists(config.g_sSignInPath):
|
||||
try:
|
||||
with open(config.g_sSignInPath, encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
sign = json.loads(content)
|
||||
|
||||
date = sign.get("date", "")
|
||||
yearMonth = getToolManager().dateTime().now().strftime("%Y%m")
|
||||
|
||||
if date == yearMonth:
|
||||
logger.debug("真寻农场签到文件检查完毕")
|
||||
return True
|
||||
else:
|
||||
logger.warning("真寻农场签到文件检查失败, 即将下载")
|
||||
return await cls.downloadSignInFile()
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("真寻农场签到文件格式错误, 即将下载")
|
||||
return await cls.downloadSignInFile()
|
||||
else:
|
||||
return await cls.downloadSignInFile()
|
||||
|
||||
@classmethod
|
||||
async def downloadSignInFile(cls) -> bool:
|
||||
"""下载签到文件，并重命名为 sign_in.json
|
||||
|
||||
Returns:
|
||||
bool: 是否下载成功
|
||||
"""
|
||||
try:
|
||||
baseUrl = Config.get_config("zhenxun_plugin_farm", "服务地址")
|
||||
url = f"{baseUrl.rstrip('/')}:8998/sign_in"
|
||||
path = str(config.g_sSignInPath.parent.resolve(strict=False))
|
||||
yearMonth = getToolManager().dateTime().now().strftime("%Y%m")
|
||||
|
||||
# 下载为 signTemp.json
|
||||
success = await cls.download(
|
||||
url=url,
|
||||
savePath=path,
|
||||
fileName="signTemp.json",
|
||||
jsonData={"date": yearMonth},
|
||||
)
|
||||
|
||||
if not success:
|
||||
return False
|
||||
|
||||
# 重命名为 sign_in.json
|
||||
getToolManager().renameFile(f"{path}/signTemp.json", "sign_in.json")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("下载签到文件失败", e=e)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def initPlantDBFile(cls) -> bool:
|
||||
"""检查本地 plant.db 版本，如远程版本更新则重新下载
|
||||
|
||||
Returns:
|
||||
bool: 是否为最新版或成功更新
|
||||
"""
|
||||
versionPath = os.path.join(os.path.dirname(config.g_sPlantPath), "version.json")
|
||||
|
||||
try:
|
||||
with open(versionPath, encoding="utf-8") as f:
|
||||
localVersion = json.load(f).get("version", 0)
|
||||
except Exception as e:
|
||||
logger.warning(f"读取本地版本失败，默认版本为0: {e}")
|
||||
localVersion = 0
|
||||
|
||||
remoteInfo = await cls.get("plant_version", name="版本检查")
|
||||
remoteVersion = remoteInfo.get("version")
|
||||
|
||||
if remoteVersion is None:
|
||||
logger.warning("获取远程版本失败")
|
||||
return False
|
||||
|
||||
if float(remoteVersion) <= float(localVersion):
|
||||
logger.debug("plant.db 已为最新版本")
|
||||
return True
|
||||
|
||||
logger.warning(
|
||||
f"发现新版本 plant.db（远程: {remoteVersion} / 本地: {localVersion}），开始更新..."
|
||||
)
|
||||
|
||||
# 先断开数据库连接
|
||||
await g_pDBService.cleanup()
|
||||
|
||||
return await cls.downloadPlantDBFile(remoteVersion)
|
||||
|
||||
@classmethod
|
||||
async def downloadPlantDBFile(cls, remoteVersion: float) -> bool:
|
||||
"""下载最新版 plant.db 并更新本地 version.json
|
||||
|
||||
Args:
|
||||
remoteVersion (float): 远程版本号
|
||||
|
||||
Returns:
|
||||
bool: 是否下载并更新成功
|
||||
"""
|
||||
baseUrl = Config.get_config("zhenxun_plugin_farm", "服务地址")
|
||||
|
||||
savePath = os.path.dirname(config.g_sPlantPath)
|
||||
success = await cls.download(
|
||||
url=f"{baseUrl.rstrip('/')}:8998/file/plant.db",
|
||||
savePath=savePath,
|
||||
fileName="plantTemp.db",
|
||||
)
|
||||
|
||||
if not success:
|
||||
return False
|
||||
|
||||
# 重命名为 sign_in.json
|
||||
getToolManager().renameFile(f"{savePath}/plantTemp.db", "plant.db")
|
||||
|
||||
versionPath = os.path.join(savePath, "version.json")
|
||||
try:
|
||||
with open(versionPath, "w", encoding="utf-8") as f:
|
||||
json.dump({"version": remoteVersion}, f)
|
||||
logger.debug("版本文件已更新")
|
||||
except Exception as e:
|
||||
logger.warning(f"写入版本文件失败: {e}")
|
||||
return False
|
||||
|
||||
await g_pDBService.plant.init()
|
||||
await g_pDBService.plant.downloadPlant()
|
||||
|
||||
return True
|
||||
@@ -0,0 +1,236 @@
|
||||
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 zhdate import ZhDate
|
||||
|
||||
from zhenxun.services.log import logger
|
||||
from zhenxun.utils.message import MessageUtils
|
||||
|
||||
from ..core import CPlayer, g_pUserPool
|
||||
from . import utils
|
||||
|
||||
|
||||
class CToolManager:
|
||||
@classmethod
|
||||
async def repeat(cls):
|
||||
await MessageUtils.build_message(
|
||||
"尚未开通农场，快at我发送 开通农场 开通吧"
|
||||
).send()
|
||||
|
||||
@classmethod
|
||||
async def getPlayerByUid(cls, uid: str) -> CPlayer | None:
|
||||
player = g_pUserPool.getUser(uid)
|
||||
if player is None:
|
||||
player = CPlayer()
|
||||
if not await player.init(uid):
|
||||
return None
|
||||
g_pUserPool.createUser(uid, player)
|
||||
|
||||
return player
|
||||
|
||||
@classmethod
|
||||
def sanitize_username(cls, username: str, max_length: int = 15) -> str:
|
||||
"""
|
||||
安全处理用户名
|
||||
功能：
|
||||
1. 移除首尾空白
|
||||
2. 过滤危险字符
|
||||
3. 转义单引号
|
||||
4. 处理空值
|
||||
5. 限制长度
|
||||
"""
|
||||
# 处理空值
|
||||
if not username:
|
||||
return "神秘农夫"
|
||||
|
||||
# 基础清洗
|
||||
cleaned = username.strip()
|
||||
|
||||
# 允许的字符白名单（可自定义扩展）
|
||||
# fmt: off
|
||||
safe_chars = {
|
||||
"_", "-", "!", "@", "#", "$", "%", "^", "&", "*", "(", ")",
|
||||
"+", "=", ".", ",", "~", "·", " ",
|
||||
"a","b","c","d","e","f","g","h","i","j","k","l","m",
|
||||
"n","o","p","q","r","s","t","u","v","w","x","y","z",
|
||||
"A","B","C","D","E","F","G","H","I","J","K","L","M",
|
||||
"N","O","P","Q","R","S","T","U","V","W","X","Y","Z",
|
||||
"0","1","2","3","4","5","6","7","8","9",
|
||||
}
|
||||
# fmt: on
|
||||
# 添加常用中文字符（Unicode范围）
|
||||
safe_chars.update(chr(c) for c in range(0x4E00, 0x9FFF + 1))
|
||||
|
||||
# 过滤危险字符
|
||||
filtered = [
|
||||
c if c in safe_chars or 0x4E00 <= ord(c) <= 0x9FFF else "" for c in cleaned
|
||||
]
|
||||
|
||||
# 合并处理结果
|
||||
safe_str = "".join(filtered)
|
||||
|
||||
# 转义单引号（双重保障）
|
||||
escaped = safe_str.replace("'", "''")
|
||||
|
||||
# 处理空结果
|
||||
if not escaped:
|
||||
return "神秘农夫"
|
||||
|
||||
# 长度限制
|
||||
return escaped[:max_length]
|
||||
|
||||
@classmethod
|
||||
def renameFile(cls, currentFilePath: str, newFileName: str) -> bool:
|
||||
"""重命名文件，如果目标文件名已存在则先删除再重命名
|
||||
|
||||
Args:
|
||||
currentFilePath (str): 当前文件的完整路径
|
||||
newFileName (str): 重命名后的文件名
|
||||
|
||||
Returns:
|
||||
bool: 重命名成功返回 True，否则返回 False
|
||||
"""
|
||||
try:
|
||||
dirPath = os.path.dirname(currentFilePath)
|
||||
newFilePath = os.path.join(dirPath, newFileName)
|
||||
|
||||
if os.path.exists(newFilePath):
|
||||
os.remove(newFilePath)
|
||||
|
||||
os.rename(currentFilePath, newFilePath)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning(f"文件重命名失败: {e}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def dateTime(cls) -> datetime:
|
||||
tz = ZoneInfo("Asia/Shanghai")
|
||||
return datetime.now(tz)
|
||||
|
||||
@classmethod
|
||||
def getSeason(cls) -> utils.Season:
|
||||
month = cls.dateTime().month
|
||||
|
||||
if month in [3, 4, 5]:
|
||||
return utils.Season.SPRING
|
||||
elif month in [6, 7, 8]:
|
||||
return utils.Season.SUMMER
|
||||
elif month in [9, 10, 11]:
|
||||
return utils.Season.AUTUMN
|
||||
elif month in [12, 1, 2]:
|
||||
return utils.Season.WINTER
|
||||
else:
|
||||
return utils.Season.NO
|
||||
|
||||
@classmethod
|
||||
def getFestival(cls) -> utils.Festival:
|
||||
now = cls.dateTime()
|
||||
|
||||
lunarDate = ZhDate.from_datetime(now)
|
||||
lunarMonth = lunarDate.lunar_month
|
||||
lunarDay = lunarDate.lunar_day
|
||||
|
||||
festivalChecks = [
|
||||
(1, lambda: lunarMonth == 8 and lunarDay == 15), # 中秋节
|
||||
(2, lambda: (now.month, now.day) == (10, 1)), # 国庆节
|
||||
(3, lambda: lunarMonth == 1 and lunarDay == 15), # 元宵节
|
||||
(4, lambda: lunarMonth == 5 and lunarDay == 5), # 端午节
|
||||
(5, lambda: lunarMonth == 1 and lunarDay == 1), # 春节
|
||||
(6, lambda: (now.month, now.day) == (10, 31)), # 万圣节
|
||||
(7, lambda: (now.month, now.day) == (12, 25)), # 圣诞节
|
||||
(8, lambda: (now.month, now.day) == (1, 1)), # 新年
|
||||
(9, lambda: (now.month, now.day) == (2, 14)), # 情人节
|
||||
]
|
||||
|
||||
for code, checkFunc in festivalChecks:
|
||||
if checkFunc():
|
||||
return utils.Festival(code)
|
||||
|
||||
return utils.Festival.NO
|
||||
|
||||
@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()
|
||||
@@ -0,0 +1,22 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class Festival(Enum):
|
||||
NO = 0 # 无节日
|
||||
MIDAUTUMN = 1 # 中秋节
|
||||
NATIONALDAY = 2 # 国庆节
|
||||
LANTERNFESTIVAL = 3 # 元宵节
|
||||
DRAGONBOAT = 4 # 端午节
|
||||
SPRINGFESTIVAL = 5 # 春节
|
||||
HALLOWEEN = 6 # 万圣节
|
||||
CHRISTMAS = 7 # 圣诞节
|
||||
NEWYEAR = 8 # 新年
|
||||
VALENTINE = 9 # 情人节
|
||||
|
||||
|
||||
class Season(Enum):
|
||||
NO = 0 # 无季节
|
||||
SPRING = 1 # 春季
|
||||
SUMMER = 2 # 夏季
|
||||
AUTUMN = 3 # 秋季
|
||||
WINTER = 4 # 冬季
|
||||
Reference in New Issue
Block a user