Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-10-06 12:20:00 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

update 0.0.6.2

Browse Source
This commit is contained in:
HibiKi
2021-11-23 21:44:59 +08:00
parent c802dfda22
commit 3de487fb24
251 changed files with 14413 additions and 14002 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/genshin/__init__.py
Regular → Executable
+1
View File
@@ -1,3 +1,4 @@
import nonebot
nonebot.load_plugins("plugins/genshin")
plugins/genshin/almanac/__init__.py
Regular → Executable
+14 -10
View File
@@ -38,12 +38,15 @@ ALC_PATH.mkdir(parents=True, exist_ok=True)
@almanac.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
alc_img = await get_alc_image(ALC_PATH)
mes = alc_img + "\n ※ 黄历数据来源于 genshin.pub"
await almanac.send(mes)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 发送查看原神黄历"
)
if alc_img:
mes = alc_img + "\n ※ 黄历数据来源于 genshin.pub"
await almanac.send(mes)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 发送查看原神黄历"
)
else:
await almanac.send("黄历图片下载失败...")
@scheduler.scheduled_job(
@@ -58,7 +61,8 @@ async def _():
gl = await bot.get_group_list()
gl = [g["group_id"] for g in gl]
alc_img = await get_alc_image(ALC_PATH)
mes = alc_img + "\n ※ 黄历数据来源于 genshin.pub"
for gid in gl:
if await group_manager.check_group_task_status(gid, "genshin_alc"):
await bot.send_group_msg(group_id=int(gid), message=mes)
if alc_img:
mes = alc_img + "\n ※ 黄历数据来源于 genshin.pub"
for gid in gl:
if await group_manager.check_group_task_status(gid, "genshin_alc"):
await bot.send_group_msg(group_id=int(gid), message=mes)
plugins/genshin/almanac/data_source.py
Regular → Executable
+26 -43
View File
@@ -1,43 +1,26 @@
from utils.browser import get_browser
from utils.message_builder import image
from datetime import datetime
from services.log import logger
from pathlib import Path
import os
url = "https://genshin.pub"
async def get_alc_image(path: Path):
date = datetime.now().date()
for file in os.listdir(path):
if f'{date}.png' != file:
file = path / file
file.unlink()
if f'{date}.png' in os.listdir(path):
return image(f'{date}.png', 'genshin/alc')
page = None
try:
browser = await get_browser()
page = await browser.new_page()
await page.goto(url, wait_until="networkidle", timeout=10000)
await page.set_viewport_size({"width": 2560, "height": 1080})
card = await page.query_selector('.GSAlmanacs_gs_almanacs__3qT_A')
await card.screenshot(path=path / f'{date}.png', timeout=100000)
except Exception as e:
logger.error(f'获取原神黄历发生错误..{type(e)}: {e}')
finally:
if page:
await page.close()
return image(f'{date}.png', 'genshin/alc')
from utils.message_builder import image
from datetime import datetime
from pathlib import Path
from utils.http_utils import AsyncPlaywright
from nonebot.adapters.cqhttp import MessageSegment
from typing import Optional
import os
url = "https://genshin.pub"
async def get_alc_image(path: Path) -> Optional[MessageSegment]:
"""
截取黄历
:param path: 存储路径
"""
date = datetime.now().date()
for file in os.listdir(path):
if f"{date}.png" != file:
file = path / file
file.unlink()
if f"{date}.png" in os.listdir(path):
return image(f"{date}.png", "genshin/alc")
return await AsyncPlaywright.screenshot(
url, path / f"{date}.png", ".GSAlmanacs_gs_almanacs__3qT_A"
)
plugins/genshin/material_remind/__init__.py
Regular → Executable
+2 -2
View File
@@ -1,6 +1,6 @@
from nonebot import on_command, Driver
from nonebot.typing import T_State
from nonebot.adapters.cqhttp import Bot, MessageEvent, Message
from nonebot.adapters.cqhttp import Bot, MessageEvent, Message, GroupMessageEvent
from utils.message_builder import image
from utils.image_utils import CreateImg
from utils.browser import get_browser
@@ -59,7 +59,7 @@ async def _(bot: Bot, event: MessageEvent, state: T_State):
)
)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 发送查看今日素材"
)
plugins/genshin/qiu_qiu_translation/__init__.py
-46
View File
@@ -1,46 +0,0 @@
from .qiu_translation import qiu_qiu_word_translation, qiu_qiu_phrase_translation
from nonebot.adapters.cqhttp import Bot, MessageEvent, GroupMessageEvent
from nonebot.typing import T_State
from nonebot import on_command
from utils.utils import get_message_text
from services.log import logger
__zx_plugin_name__ = "丘丘语翻译"
__plugin_usage__ = """
usage：
异世界旅游小助手，仅支持丘丘语翻译至中文
指令：
丘丘语翻译/丘丘一下 [文本]
""".strip()
__plugin_des__ = "其实我听得懂丘丘人讲话"
__plugin_cmd__ = ["丘丘语翻译/丘丘一下 [文本]"]
__plugin_type__ = ("原神相关",)
__plugin_version__ = 0.1
__plugin_author__ = "HibiKier"
__plugin_settings__ = {
"level": 5,
"default_status": True,
"limit_superuser": False,
"cmd": ["丘丘语翻译", "丘丘一下"],
}
qiuqiu = on_command("丘丘语翻译", aliases={"丘丘一下", "丘丘翻译"}, priority=5, block=True)
suffix = "\n※ 只能从丘丘语翻译为中文，不能反向翻译\n" "※ 注意空格，不要加入任何标点符号\n" "※ 翻译数据来源于 米游社论坛"
@qiuqiu.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
txt = get_message_text(event.json()).lower()
if txt == "":
return
mes = qiu_qiu_phrase_translation(txt)
if not mes:
mes = qiu_qiu_word_translation(txt)
mes += suffix
# print(mes)
await qiuqiu.send(mes, at_sender=True)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 发送丘丘翻译:" + txt
)
plugins/genshin/qiu_qiu_translation/qiu_qiu_dictionary.json
-127
View File
@@ -1,127 +0,0 @@
{
"word": {
"a": "啊",
"beru": "做",
"biadam": "找死啊",
"biat": "暴揍",
"buka": "肚子",
"celi": "元素",
"dada": "厉害",
"dala": "什么",
"domu": "跳舞",
"eleka": "此时此刻",
"guru": "咕",
"gusha": "草",
"ika": "敌人",
"kundela": "活着",
"kuzi": "强大",
"kucha": "吃草",
"la": "啦~",
"lata": "水元素",
"mani": "给予",
"mi": "我",
"mimi": "我们",
"mita": "肉类",
"mosi": "吃",
"movo": "带来",
"muhe": "胜利",
"ni": "风",
"nini": "风暴",
"nunu": "睡觉",
"nye": "不",
"odomu": "朋友",
"olah": "你好",
"plata": "盾牌",
"pupu": "噗噗",
"sada": "唱歌",
"shato": "伴随",
"si": "什么",
"tomo": "感激",
"todo": "放心",
"tiga": "矿石",
"upa": "聚集",
"unu": "乌努",
"valo": "谢谢你",
"vin": "酒",
"ya": "人",
"yaya": "人们",
"ye": "家伙",
"yeye": "这些家伙",
"yo": "你",
"yoyo": "你们",
"zido": "杀死"
},
"phrase": {
"beru si": "做什么",
"biadam": "可恶啊，找死啊，愤怒语气",
"biat": "打爆，暴揍；可恶的",
"dada": " 极好的，厉害的",
"dala": "表示疑问，不明白的集合，翻译为哪个，什么",
"guru-guru": "拟声词，形容肚子咕咕叫的声音，饥饿",
"guru guru": "拟声词，形容肚子咕咕叫的声音，饥饿",
"gusha": "植物类食物，不怎么好的，难过的",
"kucha pupu": "拟声词，吃草时候的拟声词，咯吱噗噗，形容吃草吃得香，贬义",
"mita": "肉类，好的，好东西，有时代丘丘人，丘丘人部落，好肉族",
"mosi mita": "吃肉，表示开心",
"mosi gusha": "吃草，表示难过",
"movo": "带来，搬运；可引申为收获，丰收",
"muhe": "胜利，战胜；成功的",
"nini": "大量风元素，风暴，狂风",
"nye": "表示否定，没有了，不是",
"pupu":"拟声词，噗噗，表示嘲讽",
"sada": "唱歌，歌颂",
"shato": "乘着，伴随着",
"upa": "凝聚，聚集，集合；聚落，部落",
"unu": "乌努,丘丘人所崇拜的神灵",
"valo": "谢谢你，不客气/再见",
"vin": "酒的，喝酒有关的，酒桶盖子",
"ya": "特指人类，一个人类",
"yaya": "人类的复数 ，大群人类",
"ye": "你这个家伙，蔑称；或用于地位高者对地位低者的称呼",
"yeye": "你们这些家伙，蔑称，ye的复数",
"yo": "你，友善称呼",
"yoyo": "你们，yo的复数，友善称呼ye",
"nye mita da ye mosi zido": "再见，一路平安",
"vin plata dada": "你的酒桶盾牌真不错",
"kucha pupu gucha ye": "你咯吱噗噗的吃草的样子真的好搞笑",
"mani nini biaodomu": "愿风暴给予你死亡",
"celi dada mimi nunu": "赞美元素，我们睡觉吧",
"muhe ye": "你们是不可战胜的",
"ye dada": "你们可真棒",
"ye yika": "你们是敌人",
"nini zido": "愿风暴杀死你"
}
}
plugins/genshin/qiu_qiu_translation/qiu_translation.py
-73
View File
@@ -1,73 +0,0 @@
import json
import os
FILE_PATH = os.path.dirname(__file__)
QIU_QIU_WORD = {}
QIU_QIU_PHRASE = {}
with open(
os.path.join(FILE_PATH, "qiu_qiu_dictionary.json"), "r", encoding="UTF-8"
) as f:
data = json.load(f)
QIU_QIU_WORD = data["word"]
QIU_QIU_PHRASE = data["phrase"]
def compare_words(word):
# 比对word库是否有匹配的单词，有的话返回翻译，没有返回原词
if word in QIU_QIU_WORD:
return QIU_QIU_WORD[word]
return word
def compare_phrase(phrase):
# 比对phrase库是否有匹配的单词，有的话返回翻译，没有的话匹配word库，都没有返回原词
if phrase in QIU_QIU_PHRASE:
return QIU_QIU_PHRASE[phrase]
if phrase in QIU_QIU_WORD:
return QIU_QIU_WORD[phrase]
return phrase
def qiu_qiu_word_translation(txt: str):
# 对语句按空格分隔替换单词翻译
txt_list = txt.split(" ")
mes = "你查询的的丘丘语意思为:\n"
for word in txt_list:
tra_word = compare_words(word)
if tra_word == word:
# 如果是原词表示没有翻译，前后加空格接回语句里
if not mes[-1] == " ":
mes += " "
mes += tra_word
mes += " "
else:
mes += tra_word
mes += "\n"
return mes
def qiu_qiu_phrase_translation(phrase):
# 语句翻译，先看phrase库是不是有匹配的语句
# 没有的话把单词拆开返回单词的意思
tra_phrase = compare_phrase(phrase)
if tra_phrase != phrase:
return f"\n翻译丘丘语意思为:\n【{tra_phrase}】\n"
txt_list = phrase.split(" ")
mes = "没有查到这句丘丘语,以下是单词的翻译\n"
for word in txt_list:
if word == " ":
continue
tra_word = compare_phrase(word)
if tra_word == word:
mes += f"{word} : 没有这个词的翻译\n"
else:
mes += f"{word} : {tra_word}\n"
return mes
plugins/genshin/query_resource_points/__init__.py
Regular → Executable
View File
plugins/genshin/query_resource_points/map.py
Regular → Executable
+266 -266
View File
@@ -1,266 +1,266 @@
from pathlib import Path
from configs.path_config import IMAGE_PATH, TEXT_PATH
from utils.image_utils import CreateImg
from typing import Tuple, List
from math import sqrt, pow
import random
try:
import ujson as json
except ModuleNotFoundError:
import json
icon_path = Path(IMAGE_PATH) / "genshin" / "genshin_icon"
map_path = Path(IMAGE_PATH) / "genshin" / "map" / "map.png"
resource_label_file = Path(TEXT_PATH) / "genshin" / "resource_label_file.json"
resource_point_file = Path(TEXT_PATH) / "genshin" / "resource_point_file.json"
class Map:
"""
原神资源生成类
"""
def __init__(
self,
resource_name: str,
center_point: Tuple[int, int],
deviation: Tuple[int, int] = (25, 51),
padding: int = 100,
planning_route: bool = False,
ratio: float = 1,
):
"""
参数：
:param resource_name: 资源名称
:param center_point: 中心点
:param deviation: 坐标误差
:param padding: 截图外边距
:param planning_route: 是否规划最佳线路
:param ratio: 压缩比率
"""
self.map = CreateImg(0, 0, background=map_path)
self.resource_name = resource_name
self.center_x = center_point[0]
self.center_y = center_point[1]
self.deviation = deviation
self.padding = int(padding * ratio)
self.planning_route = planning_route
self.ratio = ratio
self.deviation = (
int(self.deviation[0] * ratio),
int(self.deviation[1] * ratio),
)
data = json.load(open(resource_label_file, "r", encoding="utf8"))
# 资源 id
self.resource_id = [
data[x]["id"]
for x in data
if x != "CENTER_POINT" and data[x]["name"] == resource_name
][0]
# 传送锚点 id
self.teleport_anchor_id = [
data[x]["id"]
for x in data
if x != "CENTER_POINT" and data[x]["name"] == "传送锚点"
][0]
# 神像 id
self.teleport_god_id = [
data[x]["id"]
for x in data
if x != "CENTER_POINT" and data[x]["name"] == "七天神像"
][0]
# 资源坐标
data = json.load(open(resource_point_file, "r", encoding="utf8"))
self.resource_point = [
Resources(
int((self.center_x + data[x]["x_pos"]) * ratio),
int((self.center_y + data[x]["y_pos"]) * ratio),
)
for x in data
if x != "CENTER_POINT" and data[x]["label_id"] == self.resource_id
]
# 传送锚点坐标
self.teleport_anchor_point = [
Resources(
int((self.center_x + data[x]["x_pos"]) * ratio),
int((self.center_y + data[x]["y_pos"]) * ratio),
)
for x in data
if x != "CENTER_POINT" and data[x]["label_id"] == self.teleport_anchor_id
]
# 神像坐标
self.teleport_god_point = [
Resources(
int((self.center_x + data[x]["x_pos"]) * ratio),
int((self.center_y + data[x]["y_pos"]) * ratio),
)
for x in data
if x != "CENTER_POINT" and data[x]["label_id"] == self.teleport_god_id
]
# 将地图上生成资源图标
def generate_resource_icon_in_map(self) -> int:
x_list = [x.x for x in self.resource_point]
y_list = [x.y for x in self.resource_point]
min_width = min(x_list) - self.padding
max_width = max(x_list) + self.padding
min_height = min(y_list) - self.padding
max_height = max(y_list) + self.padding
self._generate_transfer_icon((min_width, min_height, max_width, max_height))
for res in self.resource_point:
icon = self._get_icon_image(self.resource_id)
self.map.paste(
icon, (res.x - self.deviation[0], res.y - self.deviation[1]), True
)
if self.planning_route:
self._generate_best_route()
self.map.crop((min_width, min_height, max_width, max_height))
rand = random.randint(1, 10000)
self.map.save(f"{IMAGE_PATH}/temp/genshin_map_{rand}.png")
return rand
# 资源数量
def get_resource_count(self) -> int:
return len(self.resource_point)
# 生成传送锚点和神像
def _generate_transfer_icon(self, box: Tuple[int, int, int, int]):
min_width, min_height, max_width, max_height = box
for resources in [self.teleport_anchor_point, self.teleport_god_point]:
id_ = (
self.teleport_anchor_id
if resources == self.teleport_anchor_point
else self.teleport_god_id
)
for res in resources:
if min_width < res.x < max_width and min_height < res.y < max_height:
icon = self._get_icon_image(id_)
self.map.paste(
icon,
(res.x - self.deviation[0], res.y - self.deviation[1]),
True,
)
# 生成最优路线（说是最优其实就是直线最短）
def _generate_best_route(self):
line_points = []
teleport_list = self.teleport_anchor_point + self.teleport_god_point
for teleport in teleport_list:
current_res, res_min_distance = teleport.get_resource_distance(self.resource_point)
current_teleport, teleport_min_distance = current_res.get_resource_distance(teleport_list)
if current_teleport == teleport:
self.map.line(
(current_teleport.x, current_teleport.y, current_res.x, current_res.y), (255, 0, 0), width=1
)
is_used_res_points = []
for res in self.resource_point:
if res in is_used_res_points:
continue
current_teleport, teleport_min_distance = res.get_resource_distance(teleport_list)
current_res, res_min_distance = res.get_resource_distance(self.resource_point)
if teleport_min_distance < res_min_distance:
self.map.line(
(current_teleport.x, current_teleport.y, res.x, res.y), (255, 0, 0), width=1
)
else:
is_used_res_points.append(current_res)
self.map.line(
(current_res.x, current_res.y, res.x, res.y), (255, 0, 0), width=1
)
res_cp = self.resource_point[:]
res_cp.remove(current_res)
# for _ in res_cp:
current_teleport_, teleport_min_distance = res.get_resource_distance(teleport_list)
current_res, res_min_distance = res.get_resource_distance(res_cp)
if teleport_min_distance < res_min_distance:
self.map.line(
(current_teleport.x, current_teleport.y, res.x, res.y), (255, 0, 0), width=1
)
else:
self.map.line(
(current_res.x, current_res.y, res.x, res.y), (255, 0, 0), width=1
)
is_used_res_points.append(current_res)
is_used_res_points.append(res)
# resources_route = []
# # 先连上最近的资源路径
# for res in self.resource_point:
# # 拿到最近的资源
# current_res, _ = res.get_resource_distance(
# self.resource_point
# + self.teleport_anchor_point
# + self.teleport_god_point
# )
# self.map.line(
# (current_res.x, current_res.y, res.x, res.y), (255, 0, 0), width=1
# )
# resources_route.append((current_res, res))
# teleport_list = self.teleport_anchor_point + self.teleport_god_point
# for res1, res2 in resources_route:
# point_list = [x for x in resources_route if res1 in x or res2 in x]
# if not list(set(point_list).intersection(set(teleport_list))):
# if res1 not in teleport_list and res2 not in teleport_list:
# # while True:
# # tmp = [x for x in point_list]
# # break
# teleport1, distance1 = res1.get_resource_distance(teleport_list)
# teleport2, distance2 = res2.get_resource_distance(teleport_list)
# if distance1 > distance2:
# self.map.line(
# (teleport1.x, teleport1.y, res1.x, res1.y),
# (255, 0, 0),
# width=1,
# )
# else:
# self.map.line(
# (teleport2.x, teleport2.y, res2.x, res2.y),
# (255, 0, 0),
# width=1,
# )
# self.map.line(xy, (255, 0, 0), width=3)
# 获取资源图标
def _get_icon_image(self, id_: int) -> "CreateImg":
icon = icon_path / f"{id_}.png"
if icon.exists():
return CreateImg(
int(50 * self.ratio), int(50 * self.ratio), background=icon
)
return CreateImg(
int(50 * self.ratio),
int(50 * self.ratio),
background=f"{icon_path}/box.png",
)
# def _get_shortest_path(self, res: 'Resources', res_2: 'Resources'):
# 资源类
class Resources:
def __init__(self, x: int, y: int):
self.x = x
self.y = y
def get_distance(self, x: int, y: int):
return int(sqrt(pow(abs(self.x - x), 2) + pow(abs(self.y - y), 2)))
# 拿到资源在该列表中的最短路径
def get_resource_distance(self, resources: List["Resources"]) -> "Resources, int":
current_res = None
min_distance = 999999
for res in resources:
distance = self.get_distance(res.x, res.y)
if distance < min_distance and res != self:
current_res = res
min_distance = distance
return current_res, min_distance
from pathlib import Path
from configs.path_config import IMAGE_PATH, TEXT_PATH
from utils.image_utils import CreateImg
from typing import Tuple, List
from math import sqrt, pow
import random
try:
import ujson as json
except ModuleNotFoundError:
import json
icon_path = Path(IMAGE_PATH) / "genshin" / "genshin_icon"
map_path = Path(IMAGE_PATH) / "genshin" / "map" / "map.png"
resource_label_file = Path(TEXT_PATH) / "genshin" / "resource_label_file.json"
resource_point_file = Path(TEXT_PATH) / "genshin" / "resource_point_file.json"
class Map:
"""
原神资源生成类
"""
def __init__(
self,
resource_name: str,
center_point: Tuple[int, int],
deviation: Tuple[int, int] = (25, 51),
padding: int = 100,
planning_route: bool = False,
ratio: float = 1,
):
"""
参数：
:param resource_name: 资源名称
:param center_point: 中心点
:param deviation: 坐标误差
:param padding: 截图外边距
:param planning_route: 是否规划最佳线路
:param ratio: 压缩比率
"""
self.map = CreateImg(0, 0, background=map_path)
self.resource_name = resource_name
self.center_x = center_point[0]
self.center_y = center_point[1]
self.deviation = deviation
self.padding = int(padding * ratio)
self.planning_route = planning_route
self.ratio = ratio
self.deviation = (
int(self.deviation[0] * ratio),
int(self.deviation[1] * ratio),
)
data = json.load(open(resource_label_file, "r", encoding="utf8"))
# 资源 id
self.resource_id = [
data[x]["id"]
for x in data
if x != "CENTER_POINT" and data[x]["name"] == resource_name
][0]
# 传送锚点 id
self.teleport_anchor_id = [
data[x]["id"]
for x in data
if x != "CENTER_POINT" and data[x]["name"] == "传送锚点"
][0]
# 神像 id
self.teleport_god_id = [
data[x]["id"]
for x in data
if x != "CENTER_POINT" and data[x]["name"] == "七天神像"
][0]
# 资源坐标
data = json.load(open(resource_point_file, "r", encoding="utf8"))
self.resource_point = [
Resources(
int((self.center_x + data[x]["x_pos"]) * ratio),
int((self.center_y + data[x]["y_pos"]) * ratio),
)
for x in data
if x != "CENTER_POINT" and data[x]["label_id"] == self.resource_id
]
# 传送锚点坐标
self.teleport_anchor_point = [
Resources(
int((self.center_x + data[x]["x_pos"]) * ratio),
int((self.center_y + data[x]["y_pos"]) * ratio),
)
for x in data
if x != "CENTER_POINT" and data[x]["label_id"] == self.teleport_anchor_id
]
# 神像坐标
self.teleport_god_point = [
Resources(
int((self.center_x + data[x]["x_pos"]) * ratio),
int((self.center_y + data[x]["y_pos"]) * ratio),
)
for x in data
if x != "CENTER_POINT" and data[x]["label_id"] == self.teleport_god_id
]
# 将地图上生成资源图标
def generate_resource_icon_in_map(self) -> int:
x_list = [x.x for x in self.resource_point]
y_list = [x.y for x in self.resource_point]
min_width = min(x_list) - self.padding
max_width = max(x_list) + self.padding
min_height = min(y_list) - self.padding
max_height = max(y_list) + self.padding
self._generate_transfer_icon((min_width, min_height, max_width, max_height))
for res in self.resource_point:
icon = self._get_icon_image(self.resource_id)
self.map.paste(
icon, (res.x - self.deviation[0], res.y - self.deviation[1]), True
)
if self.planning_route:
self._generate_best_route()
self.map.crop((min_width, min_height, max_width, max_height))
rand = random.randint(1, 10000)
self.map.save(f"{IMAGE_PATH}/temp/genshin_map_{rand}.png")
return rand
# 资源数量
def get_resource_count(self) -> int:
return len(self.resource_point)
# 生成传送锚点和神像
def _generate_transfer_icon(self, box: Tuple[int, int, int, int]):
min_width, min_height, max_width, max_height = box
for resources in [self.teleport_anchor_point, self.teleport_god_point]:
id_ = (
self.teleport_anchor_id
if resources == self.teleport_anchor_point
else self.teleport_god_id
)
for res in resources:
if min_width < res.x < max_width and min_height < res.y < max_height:
icon = self._get_icon_image(id_)
self.map.paste(
icon,
(res.x - self.deviation[0], res.y - self.deviation[1]),
True,
)
# 生成最优路线（说是最优其实就是直线最短）
def _generate_best_route(self):
line_points = []
teleport_list = self.teleport_anchor_point + self.teleport_god_point
for teleport in teleport_list:
current_res, res_min_distance = teleport.get_resource_distance(self.resource_point)
current_teleport, teleport_min_distance = current_res.get_resource_distance(teleport_list)
if current_teleport == teleport:
self.map.line(
(current_teleport.x, current_teleport.y, current_res.x, current_res.y), (255, 0, 0), width=1
)
is_used_res_points = []
for res in self.resource_point:
if res in is_used_res_points:
continue
current_teleport, teleport_min_distance = res.get_resource_distance(teleport_list)
current_res, res_min_distance = res.get_resource_distance(self.resource_point)
if teleport_min_distance < res_min_distance:
self.map.line(
(current_teleport.x, current_teleport.y, res.x, res.y), (255, 0, 0), width=1
)
else:
is_used_res_points.append(current_res)
self.map.line(
(current_res.x, current_res.y, res.x, res.y), (255, 0, 0), width=1
)
res_cp = self.resource_point[:]
res_cp.remove(current_res)
# for _ in res_cp:
current_teleport_, teleport_min_distance = res.get_resource_distance(teleport_list)
current_res, res_min_distance = res.get_resource_distance(res_cp)
if teleport_min_distance < res_min_distance:
self.map.line(
(current_teleport.x, current_teleport.y, res.x, res.y), (255, 0, 0), width=1
)
else:
self.map.line(
(current_res.x, current_res.y, res.x, res.y), (255, 0, 0), width=1
)
is_used_res_points.append(current_res)
is_used_res_points.append(res)
# resources_route = []
# # 先连上最近的资源路径
# for res in self.resource_point:
# # 拿到最近的资源
# current_res, _ = res.get_resource_distance(
# self.resource_point
# + self.teleport_anchor_point
# + self.teleport_god_point
# )
# self.map.line(
# (current_res.x, current_res.y, res.x, res.y), (255, 0, 0), width=1
# )
# resources_route.append((current_res, res))
# teleport_list = self.teleport_anchor_point + self.teleport_god_point
# for res1, res2 in resources_route:
# point_list = [x for x in resources_route if res1 in x or res2 in x]
# if not list(set(point_list).intersection(set(teleport_list))):
# if res1 not in teleport_list and res2 not in teleport_list:
# # while True:
# # tmp = [x for x in point_list]
# # break
# teleport1, distance1 = res1.get_resource_distance(teleport_list)
# teleport2, distance2 = res2.get_resource_distance(teleport_list)
# if distance1 > distance2:
# self.map.line(
# (teleport1.x, teleport1.y, res1.x, res1.y),
# (255, 0, 0),
# width=1,
# )
# else:
# self.map.line(
# (teleport2.x, teleport2.y, res2.x, res2.y),
# (255, 0, 0),
# width=1,
# )
# self.map.line(xy, (255, 0, 0), width=3)
# 获取资源图标
def _get_icon_image(self, id_: int) -> "CreateImg":
icon = icon_path / f"{id_}.png"
if icon.exists():
return CreateImg(
int(50 * self.ratio), int(50 * self.ratio), background=icon
)
return CreateImg(
int(50 * self.ratio),
int(50 * self.ratio),
background=f"{icon_path}/box.png",
)
# def _get_shortest_path(self, res: 'Resources', res_2: 'Resources'):
# 资源类
class Resources:
def __init__(self, x: int, y: int):
self.x = x
self.y = y
def get_distance(self, x: int, y: int):
return int(sqrt(pow(abs(self.x - x), 2) + pow(abs(self.y - y), 2)))
# 拿到资源在该列表中的最短路径
def get_resource_distance(self, resources: List["Resources"]) -> "Resources, int":
current_res = None
min_distance = 999999
for res in resources:
distance = self.get_distance(res.x, res.y)
if distance < min_distance and res != self:
current_res = res
min_distance = distance
return current_res, min_distance
plugins/genshin/query_resource_points/query_resource.py
Regular → Executable
+100 -111
View File
@@ -3,18 +3,15 @@ from configs.path_config import IMAGE_PATH, TEXT_PATH
from PIL.Image import UnidentifiedImageError
from utils.message_builder import image
from services.log import logger
from .map import Map
from utils.image_utils import CreateImg
import asyncio
from pathlib import Path
from asyncio.exceptions import TimeoutError
from asyncio import Semaphore
from aiohttp.client import ClientSession
from utils.user_agent import get_user_agent
from utils.image_utils import is_valid
from utils.http_utils import AsyncHttpx
from pathlib import Path
from .map import Map
import asyncio
import nonebot
import aiohttp
import aiofiles
import os
try:
@@ -96,130 +93,126 @@ async def init(flag: bool = False):
global CENTER_POINT, resource_name_list
try:
semaphore = asyncio.Semaphore(10)
async with aiohttp.ClientSession(headers=get_user_agent()) as session:
await download_map_init(session, semaphore, MAP_RATIO, flag)
await download_resource_data(session, semaphore)
await download_resource_type(session)
if not CENTER_POINT:
CENTER_POINT = json.load(open(resource_label_file, "r", encoding="utf8"))[
"CENTER_POINT"
]
with open(resource_type_file, "r", encoding="utf8") as f:
data = json.load(f)
for id_ in data:
for x in data[id_]["children"]:
resource_name_list.append(x["name"])
await download_map_init(semaphore, flag)
await download_resource_data(semaphore)
await download_resource_type()
if not CENTER_POINT:
CENTER_POINT = json.load(open(resource_label_file, "r", encoding="utf8"))[
"CENTER_POINT"
]
with open(resource_type_file, "r", encoding="utf8") as f:
data = json.load(f)
for id_ in data:
for x in data[id_]["children"]:
resource_name_list.append(x["name"])
except TimeoutError:
logger.warning('原神资源查询信息初始化超时....')
pass
# 图标及位置资源
async def download_resource_data(session: ClientSession, semaphore: Semaphore):
async def download_resource_data(semaphore: Semaphore):
icon_path.mkdir(parents=True, exist_ok=True)
resource_label_file.parent.mkdir(parents=True, exist_ok=True)
try:
async with session.get(POINT_LIST_URL, timeout=5) as response:
if response.status == 200:
data = await response.json()
if data["message"] == "OK":
data = data["data"]
for lst in ["label_list", "point_list"]:
resource_data = {"CENTER_POINT": CENTER_POINT}
tasks = []
file = (
resource_label_file
if lst == "label_list"
else resource_point_file
)
for x in data[lst]:
id_ = x["id"]
if lst == "label_list":
img_url = x["icon"]
tasks.append(
asyncio.ensure_future(
download_image(
img_url,
f"{icon_path}/{id_}.png",
session,
semaphore,
True,
)
response = await AsyncHttpx.get(POINT_LIST_URL)
if response.status_code == 200:
data = response.json()
if data["message"] == "OK":
data = data["data"]
for lst in ["label_list", "point_list"]:
resource_data = {"CENTER_POINT": CENTER_POINT}
tasks = []
file = (
resource_label_file
if lst == "label_list"
else resource_point_file
)
for x in data[lst]:
id_ = x["id"]
if lst == "label_list":
img_url = x["icon"]
tasks.append(
asyncio.ensure_future(
download_image(
img_url,
f"{icon_path}/{id_}.png",
semaphore,
True,
)
)
resource_data[id_] = x
await asyncio.gather(*tasks)
with open(file, "w", encoding="utf8") as f:
json.dump(resource_data, f, ensure_ascii=False, indent=4)
else:
logger.warning(f'获取原神资源失败 msg: {data["message"]}')
)
resource_data[id_] = x
await asyncio.gather(*tasks)
with open(file, "w", encoding="utf8") as f:
json.dump(resource_data, f, ensure_ascii=False, indent=4)
else:
logger.warning(f"获取原神资源失败 code：{response.status}")
logger.warning(f'获取原神资源失败 msg: {data["message"]}')
else:
logger.warning(f"获取原神资源失败 code：{response.status_code}")
except TimeoutError:
logger.warning("获取原神资源数据超时...已再次尝试...")
await download_resource_data(session, semaphore)
await download_resource_data(semaphore)
# 下载原神地图并拼图
async def download_map_init(
session: ClientSession, semaphore: Semaphore, ratio: float = 1, flag: bool = False
semaphore: Semaphore, flag: bool = False
):
global CENTER_POINT, MAP_RATIO
map_path.mkdir(exist_ok=True, parents=True)
_map = map_path / "map.png"
if _map.exists() and os.path.getsize(_map) > 1024 * 1024 * 30:
_map.unlink()
async with session.get(MAP_URL, timeout=5) as response:
if response.status == 200:
data = await response.json()
if data["message"] == "OK":
data = json.loads(data["data"]["info"]["detail"])
CENTER_POINT = (data["origin"][0], data["origin"][1])
if not _map.exists():
# padding_w, padding_h = data['padding']
data = data["slices"]
idx = 0
for _map_data in data[0]:
map_url = _map_data['url']
await download_image(
map_url,
f"{map_path}/{idx}.png",
session,
semaphore,
force_flag=flag,
)
idx += 1
_w, h = CreateImg(0, 0, background=f"{map_path}/0.png", ratio=MAP_RATIO).size
w = _w * len(os.listdir(map_path))
map_file = CreateImg(w, h, _w, h, ratio=MAP_RATIO)
for i in range(idx):
map_file.paste(CreateImg(0, 0, background=f"{map_path}/{i}.png", ratio=MAP_RATIO))
map_file.save(f"{map_path}/map.png")
else:
logger.warning(f'获取原神地图失败 msg: {data["message"]}')
response = await AsyncHttpx.get(MAP_URL)
if response.status_code == 200:
data = response.json()
if data["message"] == "OK":
data = json.loads(data["data"]["info"]["detail"])
CENTER_POINT = (data["origin"][0], data["origin"][1])
if not _map.exists():
data = data["slices"]
idx = 0
for _map_data in data[0]:
map_url = _map_data['url']
await download_image(
map_url,
f"{map_path}/{idx}.png",
semaphore,
force_flag=flag,
)
idx += 1
_w, h = CreateImg(0, 0, background=f"{map_path}/0.png", ratio=MAP_RATIO).size
w = _w * len(os.listdir(map_path))
map_file = CreateImg(w, h, _w, h, ratio=MAP_RATIO)
for i in range(idx):
map_file.paste(CreateImg(0, 0, background=f"{map_path}/{i}.png", ratio=MAP_RATIO))
map_file.save(f"{map_path}/map.png")
else:
logger.warning(f"获取原神地图失败 code：{response.status}")
logger.warning(f'获取原神地图失败 msg: {data["message"]}')
else:
logger.warning(f"获取原神地图失败 code：{response.status}")
# 下载资源类型数据
async def download_resource_type(session: ClientSession):
async def download_resource_type():
resource_type_file.parent.mkdir(parents=True, exist_ok=True)
async with session.get(LABEL_URL, timeout=5) as response:
if response.status == 200:
data = await response.json()
if data["message"] == "OK":
data = data["data"]["tree"]
resource_data = {}
for x in data:
id_ = x["id"]
resource_data[id_] = x
with open(resource_type_file, "w", encoding="utf8") as f:
json.dump(resource_data, f, ensure_ascii=False, indent=4)
logger.info(f"更新原神资源类型成功...")
else:
logger.warning(f'获取原神资源类型失败 msg: {data["message"]}')
response = await AsyncHttpx.get(LABEL_URL)
if response.status_code == 200:
data = response.json()
if data["message"] == "OK":
data = data["data"]["tree"]
resource_data = {}
for x in data:
id_ = x["id"]
resource_data[id_] = x
with open(resource_type_file, "w", encoding="utf8") as f:
json.dump(resource_data, f, ensure_ascii=False, indent=4)
logger.info(f"更新原神资源类型成功...")
else:
logger.warning(f"获取原神资源类型失败 code：{response.status}")
logger.warning(f'获取原神资源类型失败 msg: {data["message"]}')
else:
logger.warning(f"获取原神资源类型失败 code：{response.status_code}")
# 初始化资源图标
@@ -239,7 +232,6 @@ def gen_icon(icon: str):
async def download_image(
img_url: str,
path: str,
session: ClientSession,
semaphore: Semaphore,
gen_flag: bool = False,
force_flag: bool = False,
@@ -247,15 +239,12 @@ async def download_image(
async with semaphore:
try:
if not os.path.exists(path) or not is_valid or force_flag:
async with session.get(img_url, timeout=5) as response:
async with aiofiles.open(path, "wb") as f:
await f.write(await response.read())
logger.info(f"下载原神资源图标：{img_url}")
if gen_flag:
gen_icon(path)
except TimeoutError:
logger.warning("下载原神资源图片超时...已再次尝试...")
await download_image(img_url, path, session, semaphore, gen_flag)
if await AsyncHttpx.download_file(img_url, path):
logger.info(f"下载原神资源图标：{img_url}")
if gen_flag:
gen_icon(path)
else:
logger.info(f"下载原神资源图标：{img_url} 失败，等待下次更新...")
except UnidentifiedImageError:
logger.warning(f"原神图片打开错误..已删除，等待下次更新... file: {path}")
if os.path.exists(path):