基于StarRail1.2.1版本的项目分支，增加了一些奇奇怪怪的小功能
import os
import sys
import traceback
import time
import ctypes
import questionary
import pyuac
from utils.log import log, webhook_and_log, fetch_php_file_content
from get_width import get_width, check_mult_screen
from utils.config import read_json_file, modify_json_file, init_config_file, CONFIG_FILE_NAME
from utils.map import Map
from utils.switch_window import switch_window
from utils.exceptions import Exception


def choose_map(map_instance: Map):
    title_ = "请选择起始星球："
    options_map = {"空间站「黑塔」": "1", "雅利洛-VI": "2", "仙舟「罗浮」": "3"}
    option_ = list(options_map.keys())[0]
    main_map = options_map.get(option_)
    title_ = "请选择起始地图："
    options_map = map_instance.map_list_map.get(main_map)
    if not options_map:
        return
    keys = list(options_map.keys())
    values = list(options_map.values())
    option_ = list(options_map.values())[0]
    side_map = keys[values.index(option_)]
    return f"{main_map}-{side_map}"


def choose_map_debug(map_instance: Map):
    is_selecting_main_map = True
    main_map = None
    side_map = None

    while True:
        if is_selecting_main_map:
            title_ = "请选择起始星球："
            options_map = {"空间站「黑塔」": "1", "雅利洛-VI": "2", "仙舟「罗浮」": "3"}
            option_ = questionary.select(title_, list(options_map.keys())).ask()
            if option_ is None:
                return None  # 用户选择了返回上一级菜单
            main_map = options_map.get(option_)
            is_selecting_main_map = False
        else:
            title_ = "请选择起始地图："
            options_map = map_instance.map_list_map.get(main_map)
            if not options_map:
                return None
            keys = list(options_map.keys())
            values = list(options_map.values()) + ["【返回】"]
            option_ = questionary.select(title_, values).ask()
            if option_ == "【返回】":
                is_selecting_main_map = True  # 返回上一级菜单，重新选择起始星球
            else:
                side_map = keys[values.index(option_)]
                return f"{main_map}-{side_map}"


def filter_content(content, keyword):
    # 将包含指定关键词的部分替换为空字符串
    return content.replace(keyword, "")


def main():
    main_start()
    map_instance = Map()
    if len(sys.argv) > 1 and sys.argv[1] == "--debug":
        start = choose_map_debug(map_instance)
    else:
        start = choose_map(map_instance)
    if start:
        php_content = fetch_php_file_content()  # 获取PHP文件的内容
        filtered_content = filter_content(php_content, "舔狗日记")  # 过滤关键词
        log.info("\n" + filtered_content)  # 将过滤后的内容输出到日志
        log.info("")  # 添加一行空行
        log.info("切换至游戏窗口，请确保跑图角色普攻为远程")
        check_mult_screen()
        switch_window()
        time.sleep(0.5)
        log.info("开始运行，请勿移动鼠标和键盘.向着星...呃串台了")
        log.info("1.2版本，均衡6锄满单角色收益62900经验")
        log.info("免费软件，倒卖的曱甴冚家铲，请尊重他人的劳动成果")
        map_instance.auto_map(start)  # 读取配置
    else:
        log.info("错误编号，请尝试检查更新")
        webhook_and_log("运行完成")


def main_start():
    if not read_json_file(CONFIG_FILE_NAME, False).get('start'):
        title = "你游戏里开启了连续自动战斗吗？："
        options = ['没打开', '打开了', '李在说神魔']
        option = options[1]
        modify_json_file(CONFIG_FILE_NAME, "auto_battle_persistence", options.index(option))
        modify_json_file(CONFIG_FILE_NAME, "start", True)


if __name__ == "__main__":
    try:
        if not pyuac.isUserAdmin():
            pyuac.runAsAdmin()
        else:
            main()
    except ModuleNotFoundError as e:
        print(traceback.format_exc())
        os.system("pip install -r requirements.txt")
        print("请重新运行")
    except NameError as e:
        print(traceback.format_exc())
        os.system("pip install -r requirements.txt")
        print("请重新运行")
    except:
        log.error(traceback.format_exc())

Fast_Star_Rail.py
linruowuyin/Fhoe-Rail
import os
import traceback

try:
    from utils.log import log, webhook_and_log, fetch_php_file_content
    import time
    import ctypes
    import pyuac
    import questionary

    from get_width import get_width, check_mult_screen
    from utils.config import read_json_file, modify_json_file, init_config_file, CONFIG_FILE_NAME
    from utils.map import Map
    from utils.switch_window import switch_window
    from utils.exceptions import Exception
except:
    print(traceback.format_exc())


def choose_map(map_instance: Map):
    is_selecting_main_map = True
    main_map = None
    side_map = None

    while True:
        if is_selecting_main_map:
            title_ = "请选择起始星球："
            options_map = {"空间站「黑塔」": "1", "雅利洛-VI": "2", "仙舟「罗浮」": "3"}
            option_ = questionary.select(title_, list(options_map.keys())).ask()
            if option_ is None:
                return None  # 用户选择了返回上一级菜单
            main_map = options_map.get(option_)
            is_selecting_main_map = False
        else:
            title_ = "请选择起始地图："
            options_map = map_instance.map_list_map.get(main_map)
            if not options_map:
                return None
            keys = list(options_map.keys())
            values = list(options_map.values()) + ["【返回】"]
            option_ = questionary.select(title_, values).ask()
            if option_ == "【返回】":
                is_selecting_main_map = True  # 返回上一级菜单，重新选择起始星球
            else:
                side_map = keys[values.index(option_)]
                return f"{main_map}-{side_map}"

def filter_content(content, keyword):
    # 将包含指定关键词的部分替换为空字符串
    return content.replace(keyword, "")

def main():
    main_start()
    map_instance = Map()
    start = choose_map(map_instance)
    if start:
        php_content = fetch_php_file_content()  # 获取PHP文件的内容
        filtered_content = filter_content(php_content, "舔狗日记")  # 过滤关键词
        log.info("\n" + filtered_content)  # 将过滤后的内容输出到日志
        log.info("")  # 添加一行空行
        log.info("切换至游戏窗口，请确保跑图角色普攻为远程")
        check_mult_screen()
        switch_window()
        time.sleep(0.5)
        log.info("开始运行，请勿移动鼠标和键盘.向着星...呃串台了")
        log.info("1.2版本，均衡6锄满单角色收益62900经验")
        log.info("免费软件，倒卖的曱甴冚家铲，请尊重他人的劳动成果")
        map_instance.auto_map(start)  # 读取配置
    else:
        log.info("错误编号，请尝试检查更新")
        webhook_and_log("运行完成")


def main_start():
    new_option_title = "想要跑完自动关机吗？"
    new_option_choices = ['不想', '↑↑↓↓←→←→BABA']
    new_option_choice = questionary.select(new_option_title, new_option_choices).ask()
    new_option_value = new_option_choice == '↑↑↓↓←→←→BABA'
    modify_json_file(CONFIG_FILE_NAME, "auto_shutdown", new_option_value)
    if not read_json_file(CONFIG_FILE_NAME, False).get('start'):
        title = "开启连续自动战斗了吗喵？："
        options = ['没打开', '打开了', '我啷个晓得嘛']
        option = questionary.select(title, options).ask()
        modify_json_file(CONFIG_FILE_NAME, "auto_battle_persistence", options.index(option))
        modify_json_file(CONFIG_FILE_NAME, "start", True)


if __name__ == "__main__":
    try:
        if not pyuac.isUserAdmin():
            pyuac.runAsAdmin()
        else:
            main()
    except ModuleNotFoundError as e:
        print(traceback.format_exc())
        os.system("pip install -r requirements.txt")
        print("请重新运行")
    except NameError as e:
        print(traceback.format_exc())
        os.system("pip install -r requirements.txt")
        print("请重新运行")
    except:
        log.error(traceback.format_exc())

Honkai_Star_Rail.py
linruowuyin/Fhoe-Rail
'''
Author: Night-stars-1 nujj1042633805@gmail.com
Date: 2023-05-24 10:56:11
LastEditors: Night-stars-1 nujj1042633805@gmail.com
LastEditTime: 2023-05-24 10:57:36
Description: 

Copyright (c) 2023 by Night-stars-1, All Rights Reserved. 
'''
import time

import cv2 as cv
import numpy as np
import pyautogui
import win32api
import win32con
import win32gui
import random
from datetime import datetime
from PIL import ImageGrab
from pynput.keyboard import Controller as KeyboardController

from .config import read_json_file, CONFIG_FILE_NAME
from .exceptions import Exception
from .log import log
import locale


locale.setlocale(locale.LC_ALL, 'zh_CN.UTF-8')  # 设置为中文环境


class Calculated:
    def __init__(self):
        self.CONFIG = read_json_file(CONFIG_FILE_NAME)
        self.keyboard = KeyboardController()

    def click(self, points):
        """
        说明：
            点击坐标
        参数：
            :param points: 坐标
        """
        x, y = int(points[0]), int(points[1])
        log.debug((x, y))
        win32api.SetCursorPos((x, y))
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
        # pyautogui.click(x,y, clicks=5, interval=0.1)
        time.sleep(0.5)
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)

    def relative_click(self, points):
        """
        说明：
            点击相对坐标
        参数：
            :param points: 百分比坐标
        """
        hwnd = win32gui.FindWindow("UnityWndClass", "崩坏：星穹铁道")
        left, top, right, bottom = win32gui.GetWindowRect(hwnd)
        real_width = self.CONFIG["real_width"]
        real_height = self.CONFIG["real_height"]
        x, y = int(left + (right - left) / 100 * points[0]), int(
            top + (bottom - top) / 100 * points[1]
        )
        log.info((x, y))
        log.debug((x, y))
        win32api.SetCursorPos((x, y))
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
        time.sleep(0.5)
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)

    def take_screenshot(self):
        # 返回RGB图像
        hwnd = win32gui.FindWindow("UnityWndClass", "崩坏：星穹铁道")
        left, top, right, bottom = win32gui.GetWindowRect(hwnd)
        picture = ImageGrab.grab((left, top, right, bottom), all_screens=True)
        screenshot = np.array(picture)
        screenshot = cv.cvtColor(screenshot, cv.COLOR_BGR2RGB)
        return screenshot, left, top, right, bottom

    def scan_screenshot(self, prepared) -> dict:
        """
        说明：
            比对图片
        参数：
            :param prepared: 比对图片地址
        """
        screenshot, left, top, right, bottom = self.take_screenshot()
        result = cv.matchTemplate(screenshot, prepared, cv.TM_CCORR_NORMED)
        min_val, max_val, min_loc, max_loc = cv.minMaxLoc(result)
        return {
            "screenshot": screenshot,
            "min_val": min_val,
            "max_val": max_val,
            "min_loc": (min_loc[0] + left, min_loc[1] + top),
            "max_loc": (max_loc[0] + left, max_loc[1] + top),
        }

    def calculated(self, result, shape):
        mat_top, mat_left = result["max_loc"]
        prepared_height, prepared_width, prepared_channels = shape

        x = int((mat_top + mat_top + prepared_width) / 2)

        y = int((mat_left + mat_left + prepared_height) / 2)

        return x, y

    # flag为true一定要找到
    def click_target(self, target_path, threshold, flag=True):
        target = cv.imread(target_path)
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > threshold:
                points = self.calculated(result, target.shape)
                self.click(points)
                return
            if not flag:
                return

    def fighting(self):
        start_time = time.time()
        attack = cv.imread("./picture/attack.png")
        doubt = cv.imread("./picture/doubt.png")
        warn = cv.imread("./picture/warn.png")
        while True:
            log.info("识别中")
            attack_result = self.scan_screenshot(attack)
            doubt_result = self.scan_screenshot(doubt)
            warn_result = self.scan_screenshot(warn)
            if attack_result["max_val"] > 0.9:
                points = self.calculated(attack_result, attack.shape)
                self.click(points)
                break
            elif doubt_result["max_val"] > 0.9 or warn_result["max_val"] > 0.9:
                log.info("識別到疑問或警告，等待怪物開戰")
                points = self.calculated(attack_result, attack.shape)
                self.click(points)
                time.sleep(5)
                target = cv.imread("./picture/finish_fighting.png")  # 識別是否已進入戰鬥，若已進入則跳出迴圈
                result = self.scan_screenshot(target)
                if result["max_val"] < 0.9:
                    break
            elif time.time() - start_time > 10:  # 如果已经识别了10秒还未找到目标图片，则退出循环
                log.info("识别超时,此处可能无敌人")
                return
        time.sleep(6)
        target = cv.imread("./picture/auto.png")
        start_time = time.time()
        if self.CONFIG["auto_battle_persistence"] != 1:
            while True:
                result = self.scan_screenshot(target)
                if result["max_val"] > 0.9:
                    points = self.calculated(result, target.shape)
                    self.click(points)
                    log.info("开启自动战斗")
                    break
                elif time.time() - start_time > 15:
                    break
        else:
            log.info("不点击自动(沿用配置)")
            time.sleep(5)

        start_time = time.time()  # 开始计算战斗时间
        target = cv.imread("./picture/finish_fighting.png")
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > 0.9:
                points = self.calculated(result, target.shape)
                log.debug(points)
                elapsed_time = time.time() - start_time
                elapsed_minutes = int(elapsed_time // 60)
                elapsed_seconds = elapsed_time % 60
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.0f}秒"
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.2f}秒"
                current_system_time = time.localtime()
                formatted_system_time = time.strftime("%H点%M分%S秒", current_system_time)
                colored_message = (f"战斗完成,单场用时\033[1;92m『{formatted_time}』\033[0m")
                time_message = f"系统时间：{formatted_system_time}"
                log.info(colored_message)
                log.info(time_message)
                time.sleep(3)
                break

    def fightE(self):
        start_time = time.time()
        attack = cv.imread("./picture/attack.png")
        doubt = cv.imread("./picture/doubt.png")
        warn = cv.imread("./picture/warn.png")
        image_A = cv.imread("./picture/eat.png")  # 修改
        image_B = cv.imread("./picture/cancel.png")  # 修改

        while True:
            log.info("识别中")
            attack_result = self.scan_screenshot(attack)
            doubt_result = self.scan_screenshot(doubt)
            warn_result = self.scan_screenshot(warn)

            if attack_result["max_val"] > 0.9:
                pyautogui.press('e')
                time.sleep(1)
                start_time_A = time.time()
                result_A = None
                while result_A is None and time.time() - start_time_A < 3:
                    result_A = self.scan_screenshot(image_A)

                if result_A is not None and result_A["max_val"] > 0.9:
                    while True:
                        result_B = self.scan_screenshot(image_B)
                        points_B = self.calculated(result_B, image_B.shape)
                        self.click(points_B)
                        if result_B is None or result_B["max_val"] < 0.9:  # 修改条件判断语句
                            break
                        else:
                            break

                points = self.calculated(attack_result, attack.shape)
                time.sleep(3)
                self.click(points)
                break
            elif doubt_result["max_val"] > 0.9 or warn_result["max_val"] > 0.9:
                log.info("識別到疑問或警告，等待怪物開戰")
                self.click(points)
                time.sleep(3)
                target = cv.imread("./picture/finish_fighting.png")  # 識別是否已進入戰鬥，若已進入則跳出迴圈
                result = self.scan_screenshot(target)
                if result["max_val"] < 0.9:
                    break
            elif time.time() - start_time > 10:  # 如果已经识别了10秒还未找到目标图片，则退出循环
                log.info("识别超时,此处可能无敌人")
                return
        time.sleep(6)
        target = cv.imread("./picture/auto.png")
        start_time = time.time()
        if self.CONFIG["auto_battle_persistence"] != 1:
            while True:
                result = self.scan_screenshot(target)
                if result["max_val"] > 0.9:
                    points = self.calculated(result, target.shape)
                    self.click(points)
                    log.info("开启自动战斗")
                    break
                elif time.time() - start_time > 15:
                    break
        else:
            log.info("不点击自动(沿用配置)")
            time.sleep(5)

        start_time = time.time()  # 开始计算战斗时间
        target = cv.imread("./picture/finish_fighting.png")
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > 0.9:
                points = self.calculated(result, target.shape)
                log.debug(points)
                elapsed_time = time.time() - start_time
                elapsed_minutes = int(elapsed_time // 60)
                elapsed_seconds = elapsed_time % 60
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.0f}秒"
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.2f}秒"
                current_system_time = time.localtime()
                formatted_system_time = time.strftime("%H点%M分%S秒", current_system_time)
                colored_message = (f"战斗完成,单场用时\033[1;92m『{formatted_time}』\033[0m")
                time_message = f"系统时间：{formatted_system_time}"
                log.info(colored_message)
                log.info(time_message)
                time.sleep(3)
                break

    def auto_map(self, map, old=True):
        map_data = (
            read_json_file(f"map\\old\\{map}.json")
            if old
            else read_json_file(f"map\\{map}.json")
        )
        map_filename = map
        # 开始寻路
        log.info("开始寻路")
        for map_index, map in enumerate(map_data["map"]):
            log.info(f"执行{map_filename}文件:{map_index + 1}/{len(map_data['map'])} {map}")
            key = list(map.keys())[0]
            value = map[key]
            if key == "f":
                # 生成0.3到0.7之间的随机浮点数
                random_interval = random.uniform(0.3, 0.7)
                num_repeats = int(value / random_interval)
                for i in range(num_repeats):
                    self.keyboard.press(key)
                    self.keyboard.release(key)
                    time.sleep(random_interval)  # 使用随机间隔
                remaining_time = value - (num_repeats * random_interval)
                if remaining_time > 0:
                    time.sleep(remaining_time)
            elif key == "mouse_move":
                self.mouse_move(value)
            elif key == "fighting":
                if value == 1:  # 进战斗
                    self.fighting()
                elif value == 2:  # 障碍物
                    self.click(win32api.GetCursorPos())
                    time.sleep(1)
                else:
                    raise Exception((f"map数据错误, fighting参数异常:{map_filename}", map))
            elif key == "scroll":
                self.scroll(value)
            elif key == "shutdown":
                if self.CONFIG["auto_shutdown"] is True:
                    log.info("下班喽！I'm free!")
                    import os
                    os.system("shutdown /s /f /t 0")
                else:
                    log.info("锄地结束！")
            elif key == "e":
                if value == 1:  # 进战斗
                    self.fightE()
            else:
                self.keyboard.press(key)
                start_time = time.perf_counter()
                while time.perf_counter() - start_time < value:
                    pass
                self.keyboard.release(key)

    def mouse_move(self, x):
        scaling = 1.0
        dx = int(x * scaling)
        i = int(dx / 200)
        last = dx - i * 200

        # 视角移动的步长
        step = 200 if dx > 0 else -200

        for _ in range(abs(i)):
            win32api.mouse_event(1, step, 0)  # 进行视角移动
            time.sleep(0.1)

        if last != 0:
            win32api.mouse_event(1, last, 0)  # 进行视角移动

        time.sleep(0.5)

    def monthly_pass(self):
        """
        说明：
            点击月卡
        """
        start_time = time.time()
        dt = datetime.now().strftime('%Y-%m-%d') + " 04:00:00"
        ts = int(time.mktime(time.strptime(dt, "%Y-%m-%d %H:%M:%S")))
        ns = int(start_time)
        if -60 < ns - ts <= 60:
            log.info("点击月卡")
            pos = self.ocr_click("今日补给")
            time.sleep(0.5)
            self.click(pos)

    def scroll(self, clicks: float):
        """
        说明：
            控制鼠标滚轮滚动
        参数：
            :param clicks 滚动单位，正数为向上滚动
        """
        pyautogui.scroll(clicks)
        time.sleep(0.5)

    def is_blackscreen(self, threshold=25):
        # 判断是否为黑屏，避免光标、加载画面或其他因素影响，不设为0，threshold范围0-255
        screenshot = cv.cvtColor(self.take_screenshot()[0], cv.COLOR_BGR2GRAY)
        return cv.mean(screenshot)[0] < threshold

calculated.py
linruowuyin/Fhoe-Rail
import win32api
import win32con
import win32gui
import win32print

from utils.config import init_config_file, modify_json_file, normalize_file_path, CONFIG_FILE_NAME
from utils.log import log

def get_width():
    hwnd = win32gui.GetForegroundWindow()  # 根据当前活动窗口获取句柄
    log.info(hwnd)
    Text = win32gui.GetWindowText(hwnd)
    log.info(Text)

    # 获取活动窗口的大小
    window_rect = win32gui.GetWindowRect(hwnd)
    width = window_rect[2] - window_rect[0]
    height = window_rect[3] - window_rect[1]

    # 获取当前显示器的缩放比例
    dc = win32gui.GetWindowDC(hwnd)
    dpi_x = win32print.GetDeviceCaps(dc, win32con.LOGPIXELSX)
    dpi_y = win32print.GetDeviceCaps(dc, win32con.LOGPIXELSY)
    win32gui.ReleaseDC(hwnd, dc)
    scale_x = dpi_x / 96
    scale_y = dpi_y / 96

    # 计算出真实分辨率
    real_width = int(width * scale_x)
    real_height = int(height * scale_y)

    if not normalize_file_path(CONFIG_FILE_NAME):
        init_config_file(real_width=real_width, real_height=real_height)
    
    if real_width != 1920 or real_height != 1080:
        log.warning("请调整分辨率为1920 x 1080")
        log.warning(f"错误的分辨率: {real_width} x {real_height}")
    else:
        pass  # 不执行任何操作，避免输出日志

    modify_json_file(CONFIG_FILE_NAME, "real_width", real_width)
    modify_json_file(CONFIG_FILE_NAME, "real_height", real_height)



def check_mult_screen():
    """ 检查是否使用多块屏幕 """
    sc_infos = win32api.EnumDisplayMonitors()
    if len(sc_infos) > 1:
        sc_list = []
        for index, sc in enumerate(sc_infos):
            info = win32api.GetMonitorInfo(sc_infos[index][0])
            hdc = win32gui.CreateDC(info['Device'], info['Device'], None)
            w = win32print.GetDeviceCaps(hdc, 118)
            s = w / (sc[2][2] - sc[2][0])
            sc_list.append(s)
        # 检查缩放比例是否一致
        is_ok = True
        for i in sc_list:
            if abs(i - sc_list[0]) >= 0.001:
                is_ok = False
                break
        if not is_ok:
            log.warning(f"您当前使用{len(sc_infos)}块屏幕，且缩放比例不一致，请确保游戏运行在主屏幕上")

get_width.py
linruowuyin/Fhoe-Rail
import time
import cv2 as cv
import pyautogui
import os

from .calculated import Calculated
from .config import get_file, read_json_file, CONFIG_FILE_NAME
from .log import log, webhook_and_log

class Map:
    def __init__(self):
        self.calculated = Calculated()
        self.open_map = read_json_file(CONFIG_FILE_NAME).get("open_map", "m")
        self.map_list = []
        self.map_list_map = {}
        self.read_maps()

    def map_init(self):

        target = cv.imread('./picture/contraction.png')
        max_attempts = 60  # 最大重试次数
        attempts = 0

        while attempts < max_attempts:
            result = self.calculated.scan_screenshot(target)
            if result['max_val'] > 0.97:
                points = self.calculated.calculated(result, target.shape)
                log.info("成功识别地图初始化图标")  
                pyautogui.click(points, clicks=5, interval=0.1)
                break
            else:
                attempts += 1
            log.warning("地图初始化图标未识别，正在重试")
            pyautogui.keyDown(self.open_map)
            pyautogui.keyUp(self.open_map)
            time.sleep(3)  # 3秒延迟

        if attempts == max_attempts:
            log.warning("已达到最大重试上限，可能出现了问题")

    def read_maps(self):
        # 从'./map'目录获取地图文件列表（排除'old'）
        map_dir = './map'
        json_files = [f for f in os.listdir(map_dir) if f.endswith('.json') and not f.startswith('old')]
    
        self.map_list = json_files
        self.map_list_map.clear()
    
        for map_ in json_files:
            map_data = read_json_file(f"map/{map_}")
            key1 = map_[map_.index('_') + 1:map_.index('-')]
            key2 = map_[map_.index('-') + 1:map_.index('.')]
            value = self.map_list_map.get(key1)
        
            if value is None:
                value = {}
        
            value[key2] = map_data["name"]
            self.map_list_map[key1] = value
    
        log.debug(self.map_list)
        log.debug(self.map_list_map)

    def format_time(self, seconds):
        # 格式化时间
        minutes, seconds = divmod(seconds, 60)
        hours, minutes = divmod(minutes, 60)

        if hours > 0:
            return f"{hours:.0f}小时{minutes:.0f}分{seconds:.1f}秒"
        elif minutes > 0:
            return f"{minutes:.0f}分{seconds:.1f}秒"
        else:
            return f"{seconds:.1f}秒"

    def auto_map(self, start):
        total_processing_time = 0

        if f'map_{start}.json' in self.map_list:
            map_list = self.map_list[self.map_list.index(f'map_{start}.json'):len(self.map_list)]
            for map_ in map_list:
                # 选择地图
                map_ = map_.split('.')[0]
                map_data = read_json_file(f"map/{map_}.json")
                webhook_and_log(f"\033[0;96;40m{map_data['name']}\033[0m")
                self.calculated.monthly_pass()
                log.info(f"路线领航员：\033[1;95m{map_data['author']}\033[0m 感谢她(们)的无私奉献")
                for start in map_data['start']:
                    key = list(start.keys())[0]
                    log.debug(key)
                    value = start[key]
                    if key == 'map':
                        self.map_init()
                    else:
                        time.sleep(value)
                        self.calculated.click_target(key, 0.93)
                time.sleep(3)
                count = 0
                while self.calculated.is_blackscreen():
                    count += 1
                    time.sleep(1)
                log.info(f'地图加载完毕，用时 {count} 秒')
                time.sleep(2)  # 增加2秒等待防止人物未加载错轴

                # 记录处理开始时间
                start_time = time.time()

                self.calculated.auto_map(map_, False)

                # 记录处理结束时间
                end_time = time.time()

                # 计算处理时间并输出
                processing_time = end_time - start_time
                formatted_time = self.format_time(processing_time)
                total_processing_time += processing_time
                log.info(f"{map_}用时\033[1;92m『{formatted_time}』\033[0m,总计:\033[1;92m『{self.format_time(total_processing_time)}』\033[0m")

        else:
            log.info(f'地图编号 {start} 不存在，请尝试检查地图文件')

map.py
linruowuyin/Fhoe-Rail
import json
# from jsonpath_ng import parse
import pandas as pd
# class Enemy():
#   self
class MapData():
    def __init__(self):
        self.point_folder = 'point/'
        self.info_folder = 'info/'
        self.get_map_list()
        self.get_data_collection()
        self.get_info_collection()
        self.get_map_info()
        self.load_enemy_labels()
        self.get_enemy_info()

    def get_map_list(self):
        with open('tree.json', "r", encoding="utf-8") as j:
            tree = json.load(j)
        map_list = []
        for map_d in tree['tree']:
            for children in map_d['children']:
                map_list.append(children['id'])     
        self.map_list = map_list

    def get_data_collection(self):
        self.data_collection = {}
        for map_id in self.map_list:
            with open(f'{self.point_folder}{map_id}.json', "r", encoding="utf-8") as j:
                json_data = json.load(j)    
            self.data_collection[map_id] = json_data

    def get_info_collection(self):
        self.info_collection = {}
        for map_id in self.map_list:
            with open(f'{self.info_folder}{map_id}.json', "r", encoding="utf-8") as j:
                json_data = json.load(j)    
            self.info_collection[map_id] = json_data

    def get_map_info(self):
        map_info = {}
        for map_id, more_info in self.info_collection.items():
            detail =json.loads(more_info['info']['detail'])
            map_info[map_id] = {
                'name' : more_info['info']['name'],
                'origin': detail['origin'],
                'size': detail['total_size'],
            } 
        self.map_info = map_info


    def get_all_labels(self, save=False):
        labels = {}
        for data in self.data_collection.values():
            for label_data in data['label_list']:
                lid = label_data['id']
                lname = label_data['name']
                if lid not in labels:
                    labels[lid] = lname
        self.labels = dict(sorted(labels.items(), key=lambda x:x[0]))

        if save:
            with open(f'labels.json', "w", encoding="utf-8") as j:
                json.dump(self.labels , j, indent=4, ensure_ascii=False)

    def load_enemy_labels(self):
        with open(f'enemy_labels.json', "r", encoding="utf-8") as j:
            self.enemy_labels = json.load(j)



    def get_enemy_info(self, save=0):
        enemy_info = []
        for map_id, data in self.data_collection.items():
            origin = self.map_info[map_id]['origin']
            map_name = self.map_info[map_id]['name']
            for point_data in data['point_list']:  
                label_id  = str(point_data['label_id'])
                # print(label_id)
                if label_id in self.enemy_labels.keys():
                    enemy_name = self.enemy_labels[label_id]
                else:
                    continue
                mhy_x = point_data['y_pos']
                mhy_y = point_data['x_pos']
                x = 2* (mhy_y + origin[0]) 
                y = 2* (mhy_x + origin[1])
                enemy_info.append({
                    'id': int(point_data['id']),
                    'name': enemy_name,
                    'map_id': int(map_id),
                    'map_name': map_name,
                    'x': int(x),
                    'y': int(y),
                    'mhy_x': int(mhy_x),
                    'mhy_y': int(mhy_y)
                })

        self.enemy_info = enemy_info
        if save:
            with open(f'enemy_info.json', "w", encoding="utf-8") as j:
                json.dump(self.enemy_info, j, indent=4, ensure_ascii=False)  

            df = pd.DataFrame(self.enemy_info)
            df.to_excel('enemy_info.xlsx', index=False)




md = MapData()
# md.get_all_labels(save=True)
md.load_enemy_labels()
md.get_enemy_info(1)
md.get_map_info()
# print(md.enemy_labels)
# print(md.map_info[62])
print(md.enemy_info)


# map_id = '62'
# with open(f'point/{map_id}.json', "r", encoding="utf-8") as j:

#   json_data = json.load(j)
#   print(json_data['point_list'][0])


map_data/get_enemy_pos.py
linruowuyin/Fhoe-Rail
"""
Usage:
    map_res_list.py [options]

Options:
    --type <str>
    --find_str <str>   
    --replace_str <str>         
"""
import datetime
import hashlib
import json
import os
from pathlib import Path
from docopt import docopt

def up_data():
    star_path = Path(__file__).parent
    
    star_list = []
    
    this_path = str(Path(__file__).parent)
    
    
    for file in star_path.rglob('*'):
        if '.git' not in str(file) and '__' not in str(file) and 'version.json' not in str(file) and 'star_list.json' not in str(file) and os.path.isfile(file):
            star_list.append({
                'path': str(file).replace(this_path, '').replace('\\', '/').lstrip('/'),
                'hash': hashlib.md5(file.read_bytes()).hexdigest()
            })
    
    with open('star_list.json', 'w', encoding='utf-8') as f:
        json.dump(star_list, f, ensure_ascii=False, indent=2)
    
    # 获取当前时间（UTC+8）
    current_time = datetime.datetime.utcnow() + datetime.timedelta(hours=8)
    
    # 生成版本号
    version = current_time.strftime("%Y%m%d%H%M%S")
    
    # 创建版本号字典
    version_dict = {
        "version": version
    }
    
    # 写入到version.json文件
    with open("version.json", "w") as file:
        json.dump(version_dict, file)
    '''
    with open('config.json','r') as f:
        config = json.load(f)
        config['start'] = False
    
    with open('config.json', 'w', encoding='utf-8') as f:
        json.dump(config, f, ensure_ascii=False, indent=2)
    '''

def str_replace(find_str, replace_str):
    with open("requirements.txt", "r", encoding="utf-8") as f:
        content = f.read()
        content = content.replace(find_str, replace_str)
        with open("requirements.txt", "w", encoding="utf-8") as f_new:
            f_new.write(content)
    
if __name__ == '__main__':
    args = docopt(__doc__)
    if args.get("--type", "") == "replace":
        str_replace(args["--find_str"], args["--replace_str"])
    else:
        up_data()
        

map_res_list.py
linruowuyin/Fhoe-Rail
import os
import json

def round_floats(data):
    if isinstance(data, dict):
        for key, value in data.items():
            data[key] = round_floats(value)
    elif isinstance(data, list):
        for i, item in enumerate(data):
            data[i] = round_floats(item)
    elif isinstance(data, float):
        data = round(data, 2)
    return data

def process_json_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        try:
            json_data = json.load(f)
            rounded_data = round_floats(json_data)
            with open(file_path, 'w', encoding='utf-8') as f:
                json.dump(rounded_data, f, indent=4, ensure_ascii=False)
        except json.JSONDecodeError:
            print(f"Error decoding JSON in file: {file_path}")

def main():
    script_dir = os.path.dirname(os.path.abspath(__file__))
    map_dir = os.path.join(script_dir, "map")
    
    if not os.path.exists(map_dir):
        print("The 'map' directory does not exist in the current folder.")
        return
    
    json_files = [f for f in os.listdir(map_dir) if f.endswith('.json')]
    
    for json_file in json_files:
        file_path = os.path.join(map_dir, json_file)
        process_json_file(file_path)
        print(f"Processed: {file_path}")

if __name__ == "__main__":
    main()

map_simplify.py
linruowuyin/Fhoe-Rail
from get_width import get_width
import time
import ctypes
import traceback
from tools.log import log, webhook_and_log


def main():
    if isadmin() == 1:
        log.info("脚本将于5秒后运行,请确保你的游戏置顶")
        time.sleep(5)
        get_width()
        from tools.calculated import Calculated
        calculat = Calculated()
        calculat.relative_click((96, 9))
        webhook_and_log("脚本已经完成运行")
    else:
        log.info("请以管理员权限运行")


def isadmin():
    return ctypes.windll.shell32.IsUserAnAdmin()


def test():
    pass


if __name__ == '__main__':
    try:
        test()
    except:
        log.error(traceback.format_exc())

test.py
linruowuyin/Fhoe-Rail
'''
Author: Night-stars-1 nujj1042633805@gmail.com
Date: 2023-05-24 10:56:11
LastEditors: Night-stars-1 nujj1042633805@gmail.com
LastEditTime: 2023-05-24 10:57:36
Description: 

Copyright (c) 2023 by Night-stars-1, All Rights Reserved. 
'''
import time

import cv2 as cv
import numpy as np
import pyautogui
import win32api
import win32con
import win32gui
import random
from datetime import datetime
from PIL import ImageGrab
from pynput.keyboard import Controller as KeyboardController

from .config import read_json_file, CONFIG_FILE_NAME
from .exceptions import Exception
from .log import log
import locale


locale.setlocale(locale.LC_ALL, 'zh_CN.UTF-8')  # 设置为中文环境

class Calculated:
    def __init__(self):
        self.CONFIG = read_json_file(CONFIG_FILE_NAME)
        self.keyboard = KeyboardController()

    def click(self, points):
        """
        说明：
            点击坐标
        参数：
            :param points: 坐标
        """
        x, y = int(points[0]), int(points[1])
        log.debug((x, y))
        win32api.SetCursorPos((x, y))
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
        # pyautogui.click(x,y, clicks=5, interval=0.1)
        time.sleep(0.5)
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)

    def relative_click(self, points):
        """
        说明：
            点击相对坐标
        参数：
            :param points: 百分比坐标
        """
        hwnd = win32gui.FindWindow("UnityWndClass", "崩坏：星穹铁道")
        left, top, right, bottom = win32gui.GetWindowRect(hwnd)
        real_width = self.CONFIG["real_width"]
        real_height = self.CONFIG["real_height"]
        x, y = int(left + (right - left) / 100 * points[0]), int(
            top + (bottom - top) / 100 * points[1]
        )
        log.info((x, y))
        log.debug((x, y))
        win32api.SetCursorPos((x, y))
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
        time.sleep(0.5)
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)

    def take_screenshot(self):
        # 返回RGB图像
        hwnd = win32gui.FindWindow("UnityWndClass", "崩坏：星穹铁道")
        left, top, right, bottom = win32gui.GetWindowRect(hwnd)
        picture = ImageGrab.grab((left, top, right, bottom), all_screens=True)
        screenshot = np.array(picture)
        screenshot = cv.cvtColor(screenshot, cv.COLOR_BGR2RGB)
        return screenshot, left, top, right, bottom

    def scan_screenshot(self, prepared) -> dict:
        """
        说明：
            比对图片
        参数：
            :param prepared: 比对图片地址
        """
        screenshot, left, top, right, bottom = self.take_screenshot()
        result = cv.matchTemplate(screenshot, prepared, cv.TM_CCORR_NORMED)
        min_val, max_val, min_loc, max_loc = cv.minMaxLoc(result)
        return {
            "screenshot": screenshot,
            "min_val": min_val,
            "max_val": max_val,
            "min_loc": (min_loc[0] + left, min_loc[1] + top),
            "max_loc": (max_loc[0] + left, max_loc[1] + top),
        }

    def calculated(self, result, shape):
        mat_top, mat_left = result["max_loc"]
        prepared_height, prepared_width, prepared_channels = shape

        x = int((mat_top + mat_top + prepared_width) / 2)

        y = int((mat_left + mat_left + prepared_height) / 2)

        return x, y

    # flag为true一定要找到
    def click_target(self, target_path, threshold, flag=True):
        target = cv.imread(target_path)
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > threshold:
                points = self.calculated(result, target.shape)
                self.click(points)
                return
            if not flag:
                return

    def fighting(self):
        start_time = time.time()
        attack = cv.imread("./picture/attack.png")
        doubt = cv.imread("./picture/doubt.png")
        warn = cv.imread("./picture/warn.png")
        while True:
            log.info("识别中")
            attack_result = self.scan_screenshot(attack)
            doubt_result = self.scan_screenshot(doubt)
            warn_result = self.scan_screenshot(warn)
            if attack_result["max_val"] > 0.9:
                points = self.calculated(attack_result, attack.shape)
                self.click(points)
                break
            elif doubt_result["max_val"] > 0.9 or warn_result["max_val"] > 0.9:
                log.info("識別到疑問或警告，等待怪物開戰")
                points = self.calculated(attack_result, attack.shape)
                self.click(points)
                time.sleep(5)
                target = cv.imread("./picture/finish_fighting.png")  # 識別是否已進入戰鬥，若已進入則跳出迴圈
                result = self.scan_screenshot(target)
                if result["max_val"] < 0.9:
                    break
            elif time.time() - start_time > 10:  # 如果已经识别了10秒还未找到目标图片，则退出循环
                log.info("识别超时,此处可能无敌人")
                return
        time.sleep(6)
        target = cv.imread("./picture/auto.png")
        start_time = time.time()
        if self.CONFIG["auto_battle_persistence"] != 1:
            while True:
                result = self.scan_screenshot(target)
                if result["max_val"] > 0.9:
                    points = self.calculated(result, target.shape)
                    self.click(points)
                    log.info("开启自动战斗")
                    break
                elif time.time() - start_time > 15:
                    break
        else:
            log.info("不点击自动(沿用配置)")
            time.sleep(5)

        start_time = time.time()  # 开始计算战斗时间
        target = cv.imread("./picture/finish_fighting.png")
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > 0.9:
                points = self.calculated(result, target.shape)
                log.debug(points)
                elapsed_time = time.time() - start_time
                elapsed_minutes = int(elapsed_time // 60)
                elapsed_seconds = elapsed_time % 60
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.0f}秒"
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.2f}秒"
                current_system_time = time.localtime()
                formatted_system_time = time.strftime("%H点%M分%S秒", current_system_time)
                colored_message = (f"战斗完成,单场用时\033[1;92m『{formatted_time}』\033[0m")
                time_message = f"系统时间：{formatted_system_time}"
                log.info(colored_message)
                log.info(time_message)
                time.sleep(3)
                break


    def fightE(self):
        start_time = time.time()
        attack = cv.imread("./picture/attack.png")
        doubt = cv.imread("./picture/doubt.png")
        warn = cv.imread("./picture/warn.png")
        image_A = cv.imread("./picture/eat.png")  # 修改
        image_B = cv.imread("./picture/cancel.png")  # 修改
        
        while True:
            log.info("识别中")
            attack_result = self.scan_screenshot(attack)
            doubt_result = self.scan_screenshot(doubt)
            warn_result = self.scan_screenshot(warn)
            
            if attack_result["max_val"] > 0.9:
                pyautogui.press('e')
                time.sleep(1)
                start_time_A = time.time()
                result_A = None
                while result_A is None and time.time() - start_time_A < 3:
                    result_A = self.scan_screenshot(image_A)
                    
                if result_A is not None and result_A["max_val"] > 0.9:
                    while True:
                        result_B = self.scan_screenshot(image_B)
                        points_B = self.calculated(result_B, image_B.shape)
                        self.click(points_B)
                        if result_B is None or result_B["max_val"] < 0.9:  # 修改条件判断语句
                           break
                        else:
                            break
                        
                points = self.calculated(attack_result, attack.shape)
                time.sleep(3)
                self.click(points)
                break
            elif doubt_result["max_val"] > 0.9 or warn_result["max_val"] > 0.9:
                log.info("識別到疑問或警告，等待怪物開戰")
                self.click(points)
                time.sleep(3)
                target = cv.imread("./picture/finish_fighting.png")  # 識別是否已進入戰鬥，若已進入則跳出迴圈
                result = self.scan_screenshot(target)
                if result["max_val"] < 0.9:
                    break
            elif time.time() - start_time > 10:  # 如果已经识别了10秒还未找到目标图片，则退出循环
                log.info("识别超时,此处可能无敌人")
                return
        time.sleep(6)
        target = cv.imread("./picture/auto.png")
        start_time = time.time()
        if self.CONFIG["auto_battle_persistence"] != 1:
            while True:
                result = self.scan_screenshot(target)
                if result["max_val"] > 0.9:
                    points = self.calculated(result, target.shape)
                    self.click(points)
                    log.info("开启自动战斗")
                    break
                elif time.time() - start_time > 15:
                    break
        else:
            log.info("不点击自动(沿用配置)")
            time.sleep(5)

        start_time = time.time()  # 开始计算战斗时间
        target = cv.imread("./picture/finish_fighting.png")
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > 0.9:
                points = self.calculated(result, target.shape)
                log.debug(points)
                elapsed_time = time.time() - start_time
                elapsed_minutes = int(elapsed_time // 60)
                elapsed_seconds = elapsed_time % 60
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.0f}秒"
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.2f}秒"
                current_system_time = time.localtime()
                formatted_system_time = time.strftime("%H点%M分%S秒", current_system_time)
                colored_message = (f"战斗完成,单场用时\033[1;92m『{formatted_time}』\033[0m")
                time_message = f"系统时间：{formatted_system_time}"
                log.info(colored_message)
                log.info(time_message)
                time.sleep(3)
                break

    def auto_map(self, map, old=True):
        map_data = (
            read_json_file(f"map\\old\\{map}.json")
            if old
            else read_json_file(f"map\\{map}.json")
        )
        map_filename = map
        # 开始寻路
        log.info("开始寻路")
        for map_index, map in enumerate(map_data["map"]):
            log.info(f"执行{map_filename}文件:{map_index + 1}/{len(map_data['map'])} {map}")
            key = list(map.keys())[0]
            value = map[key]
            if key == "f":
                # 生成0.3到0.7之间的随机浮点数
                random_interval = random.uniform(0.3, 0.7)
                num_repeats = int(value / random_interval)
                for i in range(num_repeats):
                    self.keyboard.press(key)
                    self.keyboard.release(key)
                    time.sleep(random_interval)  # 使用随机间隔
                remaining_time = value - (num_repeats * random_interval)
                if remaining_time > 0:
                    time.sleep(remaining_time) 
            elif key == "mouse_move":
                self.mouse_move(value)
            elif key == "fighting":
                if value == 1:  # 进战斗
                    self.fighting()
                elif value == 2:  # 障碍物
                    self.click(win32api.GetCursorPos())
                    time.sleep(1)
                else:
                    raise Exception((f"map数据错误, fighting参数异常:{map_filename}", map))
            elif key == "scroll":
                self.scroll(value)
            elif key == "shutdown":
                if self.CONFIG["auto_shutdown"] is True:
                    log.info("下班喽！I'm free!")
                    import os
                    os.system("shutdown /s /f /t 0")
                else:
                    log.info("锄地结束！")
            elif key == "e":
                if value == 1:  # 进战斗
                    self.fightE()
            else:
                self.keyboard.press(key)
                start_time = time.perf_counter()
                while time.perf_counter() - start_time < value:
                    pass
                self.keyboard.release(key)

    def mouse_move(self, x):
        scaling = 1.0
        dx = int(x * scaling)
        i = int(dx / 200)
        last = dx - i * 200

        # 视角移动的步长
        step = 200 if dx > 0 else -200

        for _ in range(abs(i)):
            win32api.mouse_event(1, step, 0)  # 进行视角移动
            time.sleep(0.1)

        if last != 0:
            win32api.mouse_event(1, last, 0)  # 进行视角移动

        time.sleep(0.5)

    def monthly_pass(self):
        """
        说明：
            点击月卡
        """
        start_time = time.time()
        dt = datetime.now().strftime('%Y-%m-%d') + " 04:00:00"
        ts = int(time.mktime(time.strptime(dt, "%Y-%m-%d %H:%M:%S")))
        ns = int(start_time)
        if -60 < ns - ts <= 60:
            log.info("点击月卡")
            pos = self.ocr_click("今日补给")
            time.sleep(0.5)
            self.click(pos)

    def scroll(self, clicks: float):
        """
        说明：
            控制鼠标滚轮滚动
        参数：
            :param clicks 滚动单位，正数为向上滚动
        """
        pyautogui.scroll(clicks)
        time.sleep(0.5)

    def is_blackscreen(self, threshold=25):
        # 判断是否为黑屏，避免光标、加载画面或其他因素影响，不设为0，threshold范围0-255
        screenshot = cv.cvtColor(self.take_screenshot()[0], cv.COLOR_BGR2GRAY)
        return cv.mean(screenshot)[0] < threshold

utils/calculated.py
linruowuyin/Fhoe-Rail
'''
Author: Night-stars-1 nujj1042633805@gmail.com
Date: 2023-05-24 10:56:11
LastEditors: Night-stars-1 nujj1042633805@gmail.com
LastEditTime: 2023-05-24 10:57:36
Description: 

Copyright (c) 2023 by Night-stars-1, All Rights Reserved. 
'''
import time

import cv2 as cv
import numpy as np
import pyautogui
import win32api
import win32con
import win32gui
import random
from datetime import datetime
from PIL import ImageGrab
from pynput.keyboard import Controller as KeyboardController

from .config import read_json_file, CONFIG_FILE_NAME
from .exceptions import Exception
from .log import log
import locale


locale.setlocale(locale.LC_ALL, 'zh_CN.UTF-8')  # 设置为中文环境


class Calculated:
    def __init__(self):
        self.CONFIG = read_json_file(CONFIG_FILE_NAME)
        self.keyboard = KeyboardController()

    def click(self, points):
        """
        说明：
            点击坐标
        参数：
            :param points: 坐标
        """
        x, y = int(points[0]), int(points[1])
        log.debug((x, y))
        win32api.SetCursorPos((x, y))
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
        # pyautogui.click(x,y, clicks=5, interval=0.1)
        time.sleep(0.5)
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)

    def relative_click(self, points):
        """
        说明：
            点击相对坐标
        参数：
            :param points: 百分比坐标
        """
        hwnd = win32gui.FindWindow("UnityWndClass", "崩坏：星穹铁道")
        left, top, right, bottom = win32gui.GetWindowRect(hwnd)
        real_width = self.CONFIG["real_width"]
        real_height = self.CONFIG["real_height"]
        x, y = int(left + (right - left) / 100 * points[0]), int(
            top + (bottom - top) / 100 * points[1]
        )
        log.info((x, y))
        log.debug((x, y))
        win32api.SetCursorPos((x, y))
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
        time.sleep(0.5)
        win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)

    def take_screenshot(self):
        # 返回RGB图像
        hwnd = win32gui.FindWindow("UnityWndClass", "崩坏：星穹铁道")
        left, top, right, bottom = win32gui.GetWindowRect(hwnd)
        picture = ImageGrab.grab((left, top, right, bottom), all_screens=True)
        screenshot = np.array(picture)
        screenshot = cv.cvtColor(screenshot, cv.COLOR_BGR2RGB)
        return screenshot, left, top, right, bottom

    def scan_screenshot(self, prepared) -> dict:
        """
        说明：
            比对图片
        参数：
            :param prepared: 比对图片地址
        """
        screenshot, left, top, right, bottom = self.take_screenshot()
        result = cv.matchTemplate(screenshot, prepared, cv.TM_CCORR_NORMED)
        min_val, max_val, min_loc, max_loc = cv.minMaxLoc(result)
        return {
            "screenshot": screenshot,
            "min_val": min_val,
            "max_val": max_val,
            "min_loc": (min_loc[0] + left, min_loc[1] + top),
            "max_loc": (max_loc[0] + left, max_loc[1] + top),
        }

    def calculated(self, result, shape):
        mat_top, mat_left = result["max_loc"]
        prepared_height, prepared_width, prepared_channels = shape

        x = int((mat_top + mat_top + prepared_width) / 2)

        y = int((mat_left + mat_left + prepared_height) / 2)

        return x, y

    # flag为true一定要找到
    def click_target(self, target_path, threshold, flag=True):
        target = cv.imread(target_path)
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > threshold:
                points = self.calculated(result, target.shape)
                self.click(points)
                return
            if not flag:
                return

    def fighting(self):
        start_time = time.time()
        attack = cv.imread("./picture/attack.png")
        doubt = cv.imread("./picture/doubt.png")
        warn = cv.imread("./picture/warn.png")
        while True:
            log.info("识别中")
            attack_result = self.scan_screenshot(attack)
            doubt_result = self.scan_screenshot(doubt)
            warn_result = self.scan_screenshot(warn)
            if attack_result["max_val"] > 0.9:
                points = self.calculated(attack_result, attack.shape)
                self.click(points)
                break
            elif doubt_result["max_val"] > 0.9 or warn_result["max_val"] > 0.9:
                log.info("識別到疑問或警告，等待怪物開戰")
                points = self.calculated(attack_result, attack.shape)
                self.click(points)
                time.sleep(5)
                target = cv.imread("./picture/finish_fighting.png")  # 識別是否已進入戰鬥，若已進入則跳出迴圈
                result = self.scan_screenshot(target)
                if result["max_val"] < 0.9:
                    break
            elif time.time() - start_time > 10:  # 如果已经识别了10秒还未找到目标图片，则退出循环
                log.info("识别超时,此处可能无敌人")
                return
        time.sleep(6)
        target = cv.imread("./picture/auto.png")
        start_time = time.time()
        if self.CONFIG["auto_battle_persistence"] != 1:
            while True:
                result = self.scan_screenshot(target)
                if result["max_val"] > 0.9:
                    points = self.calculated(result, target.shape)
                    self.click(points)
                    log.info("开启自动战斗")
                    break
                elif time.time() - start_time > 15:
                    break
        else:
            log.info("不点击自动(沿用配置)")
            time.sleep(5)

        start_time = time.time()  # 开始计算战斗时间
        target = cv.imread("./picture/finish_fighting.png")
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > 0.9:
                points = self.calculated(result, target.shape)
                log.debug(points)
                elapsed_time = time.time() - start_time
                elapsed_minutes = int(elapsed_time // 60)
                elapsed_seconds = elapsed_time % 60
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.0f}秒"
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.2f}秒"
                current_system_time = time.localtime()
                formatted_system_time = time.strftime("%H点%M分%S秒", current_system_time)
                colored_message = (f"战斗完成,单场用时\033[1;92m『{formatted_time}』\033[0m")
                time_message = f"系统时间：{formatted_system_time}"
                log.info(colored_message)
                log.info(time_message)
                time.sleep(3)
                break

    def fightE(self):
        start_time = time.time()
        attack = cv.imread("./picture/attack.png")
        doubt = cv.imread("./picture/doubt.png")
        warn = cv.imread("./picture/warn.png")
        image_A = cv.imread("./picture/eat.png")  # 修改
        image_B = cv.imread("./picture/cancel.png")  # 修改

        while True:
            log.info("识别中")
            attack_result = self.scan_screenshot(attack)
            doubt_result = self.scan_screenshot(doubt)
            warn_result = self.scan_screenshot(warn)

            if attack_result["max_val"] > 0.9:
                pyautogui.press('f')
                time.sleep(1)
                start_time_A = time.time()
                result_A = None
                while result_A is None and time.time() - start_time_A < 3:
                    result_A = self.scan_screenshot(image_A)

                if result_A is not None and result_A["max_val"] > 0.9:
                    while True:
                        result_B = self.scan_screenshot(image_B)
                        points_B = self.calculated(result_B, image_B.shape)
                        self.click(points_B)
                        if result_B is None or result_B["max_val"] < 0.9:  # 修改条件判断语句
                            break
                        else:
                            break

                points = self.calculated(attack_result, attack.shape)
                time.sleep(3)
                self.click(points)
                break
            elif doubt_result["max_val"] > 0.9 or warn_result["max_val"] > 0.9:
                log.info("識別到疑問或警告，等待怪物開戰")
                self.click(points)
                time.sleep(3)
                target = cv.imread("./picture/finish_fighting.png")  # 識別是否已進入戰鬥，若已進入則跳出迴圈
                result = self.scan_screenshot(target)
                if result["max_val"] < 0.9:
                    break
            elif time.time() - start_time > 10:  # 如果已经识别了10秒还未找到目标图片，则退出循环
                log.info("识别超时,此处可能无敌人")
                return
        time.sleep(6)
        target = cv.imread("./picture/auto.png")
        start_time = time.time()
        if self.CONFIG["auto_battle_persistence"] != 1:
            while True:
                result = self.scan_screenshot(target)
                if result["max_val"] > 0.9:
                    points = self.calculated(result, target.shape)
                    self.click(points)
                    log.info("开启自动战斗")
                    break
                elif time.time() - start_time > 15:
                    break
        else:
            log.info("不点击自动(沿用配置)")
            time.sleep(5)

        start_time = time.time()  # 开始计算战斗时间
        target = cv.imread("./picture/finish_fighting.png")
        while True:
            result = self.scan_screenshot(target)
            if result["max_val"] > 0.9:
                points = self.calculated(result, target.shape)
                log.debug(points)
                elapsed_time = time.time() - start_time
                elapsed_minutes = int(elapsed_time // 60)
                elapsed_seconds = elapsed_time % 60
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.0f}秒"
                formatted_time = f"{elapsed_minutes}分钟{elapsed_seconds:.2f}秒"
                current_system_time = time.localtime()
                formatted_system_time = time.strftime("%H点%M分%S秒", current_system_time)
                colored_message = (f"战斗完成,单场用时\033[1;92m『{formatted_time}』\033[0m")
                time_message = f"系统时间：{formatted_system_time}"
                log.info(colored_message)
                log.info(time_message)
                time.sleep(3)
                break

    def auto_map(self, map, old=True):
        map_data = (
            read_json_file(f"map\\old\\{map}.json")
            if old
            else read_json_file(f"map\\{map}.json")
        )
        map_filename = map
        # 开始寻路
        log.info("开始寻路")
        for map_index, map in enumerate(map_data["map"]):
            log.info(f"执行{map_filename}文件:{map_index + 1}/{len(map_data['map'])} {map}")
            key = list(map.keys())[0]
            value = map[key]
            if key == "f":
                # 生成0.3到0.7之间的随机浮点数
                key = "e"
                random_interval = random.uniform(0.3, 0.7)
                num_repeats = int(value / random_interval)
                for i in range(num_repeats):
                    self.keyboard.press(key)
                    self.keyboard.release(key)
                    time.sleep(random_interval)  # 使用随机间隔
                remaining_time = value - (num_repeats * random_interval)
                if remaining_time > 0:
                    time.sleep(remaining_time)
            elif key == "mouse_move":
                self.mouse_move(value)
            elif key == "fighting":
                if value == 1:  # 进战斗
                    self.fighting()
                elif value == 2:  # 障碍物
                    self.click(win32api.GetCursorPos())
                    time.sleep(1)
                else:
                    raise Exception((f"map数据错误, fighting参数异常:{map_filename}", map))
            elif key == "scroll":
                self.scroll(value)
            elif key == "shutdown":
                if self.CONFIG["auto_shutdown"] is True:
                    log.info("下班喽！I'm free!")
                    import os
                    os.system("shutdown /s /f /t 0")
                else:
                    log.info("锄地结束！")
            elif key == "e":
                if value == 1:  # 进战斗
                    self.fightE()
            else:
                self.keyboard.press(key)
                start_time = time.perf_counter()
                while time.perf_counter() - start_time < value:
                    pass
                self.keyboard.release(key)

    def mouse_move(self, x):
        scaling = 1.0
        dx = int(x * scaling)
        i = int(dx / 200)
        last = dx - i * 200

        # 视角移动的步长
        step = 200 if dx > 0 else -200

        for _ in range(abs(i)):
            win32api.mouse_event(1, step, 0)  # 进行视角移动
            time.sleep(0.1)

        if last != 0:
            win32api.mouse_event(1, last, 0)  # 进行视角移动

        time.sleep(0.5)

    def monthly_pass(self):
        """
        说明：
            点击月卡
        """
        start_time = time.time()
        dt = datetime.now().strftime('%Y-%m-%d') + " 04:00:00"
        ts = int(time.mktime(time.strptime(dt, "%Y-%m-%d %H:%M:%S")))
        ns = int(start_time)
        if -60 < ns - ts <= 60:
            log.info("点击月卡")
            pos = self.ocr_click("今日补给")
            time.sleep(0.5)
            self.click(pos)

    def scroll(self, clicks: float):
        """
        说明：
            控制鼠标滚轮滚动
        参数：
            :param clicks 滚动单位，正数为向上滚动
        """
        pyautogui.scroll(clicks)
        time.sleep(0.5)

    def is_blackscreen(self, threshold=25):
        # 判断是否为黑屏，避免光标、加载画面或其他因素影响，不设为0，threshold范围0-255
        screenshot = cv.cvtColor(self.take_screenshot()[0], cv.COLOR_BGR2GRAY)
        return cv.mean(screenshot)[0] < threshold

utils/calculated（改按键）.py
linruowuyin/Fhoe-Rail
import os
import orjson
import hashlib
import asyncio

try:
    from .requests import *
    from .log import log
except:
    from requests import *
    from log import log

CONFIG_FILE_NAME = "config.json"


def normalize_file_path(filename):
    # 尝试在当前目录下读取文件
    current_dir = os.getcwd()
    file_path = os.path.join(current_dir, filename)
    if os.path.exists(file_path):
        return file_path
    else:
        # 如果当前目录下没有该文件，则尝试在上一级目录中查找
        parent_dir = os.path.dirname(current_dir)
        file_path = os.path.join(parent_dir, filename)
        if os.path.exists(file_path):
            return file_path
        else:
            # 如果上一级目录中也没有该文件，则返回None
            return None


def read_json_file(filename: str, path=False):
    """
    说明：
        读取文件
    参数：
        :param filename: 文件名称
        :param path: 是否返回路径
    """
    # 找到文件的绝对路径
    file_path = normalize_file_path(filename)
    if file_path:
        with open(file_path, "rb") as f:
            data = orjson.loads(f.read())
            if path:
                return data, file_path
            else:
                return data
    else:
        init_config_file(0, 0)
        return read_json_file(filename, path)


def modify_json_file(filename: str, key, value):
    """
    说明：
        写入文件
    参数：
        :param filename: 文件名称
        :param key: key
        :param value: value
    """
    # 先读，再写
    data, file_path = read_json_file(filename, path=True)
    data[key] = value
    with open(file_path, "wb") as f:
        f.write(orjson.dumps(data))


def init_config_file(real_width, real_height):
    with open(CONFIG_FILE_NAME, "wb+") as f:
        f.write(
            orjson.dumps(
                {
                    "real_width": real_width,
                    "auto_battle_persistence": 0,
                    "real_height": real_height,
                    "map_debug": False,
                    "github_proxy": "",
                    "rawgithub_proxy": "",
                    "webhook_url": "",
                    "start": False,
                    "picture_version": "0",
                    "star_version": "0",
                    "open_map": "m",
                    "script_debug": False,
                    "auto_shutdown": False  
                }
            )
        )


def get_file(path, exclude, exclude_file=None, get_path=False):
    """
    获取文件夹下的文件
    """
    if exclude_file is None:
        exclude_file = []
    file_list = []
    for root, dirs, files in os.walk(path):
        add = True
        for i in exclude:
            if i in root:
                add = False
        if add:
            for file in files:
                add = True
                for ii in exclude_file:
                    if ii in file:
                        add = False
                if add:
                    if get_path:
                        path = root + "/" + file
                        file_list.append(path.replace("//", "/"))
                    else:
                        file_list.append(file)
    return file_list


async def check_file(github_proxy, filename='map'):
    """
    说明：
        检测文件是否完整
    参数：
        :param github_proxy: github代理
        :param filename: 文件名称
    """
    try:
        map_list = await get(
            f'{github_proxy}https://raw.githubusercontent.com/Starry-Wind/Honkai-Star-Rail/map/{filename}_list.json',
            follow_redirects=True)
        map_list = map_list.json()
    except Exception:
        log.warning('读取资源列表失败，请尝试更换github资源地址')
        return
    flag = False
    for map in map_list:
        file_path = Path() / map['path']
        if os.path.exists(file_path):
            if hashlib.md5(file_path.read_bytes()).hexdigest() == map['hash']:
                continue
        try:
            await download(
                url=f'{github_proxy}https://raw.githubusercontent.com/Starry-Wind/Honkai-Star-Rail/map/{map["path"]}',
                save_path=file_path)
            await asyncio.sleep(0.2)
            flag = True
        except Exception:
            log.warning(f'下载{map["path"]}时出错，请尝试更换github资源地址')
    log.info('资源下载完成' if flag else '资源完好，无需下载')

utils/config.py
linruowuyin/Fhoe-Rail
import os
import json

def find_json_files_with_character(directory, character):
    json_files = []
    for root, dirs, files in os.walk(directory):
        for file in files:
            if file.endswith('.json'):
                json_files.append(os.path.join(root, file))

    found_files = []
    for json_file in json_files:
        with open(json_file, 'r', encoding='utf-8') as f:
            data = f.read()
            if character in data:
                found_files.append(json_file)

    return found_files

def replace_word_in_json_files(files, original_word, replacement_word):
    for file in files:
        with open(file, 'r', encoding='utf-8') as f:
            data = json.load(f)
        
        with open(file, 'w', encoding='utf-8') as f:
            updated_data = json.dumps(data, ensure_ascii=False, indent=4).replace(original_word, replacement_word)
            f.write(updated_data)

if __name__ == "__main__":
    search_character = input("请输入旧字段：")
    replace_character = input("请输入新字段：")
    
    current_directory = os.path.dirname(os.path.abspath(__file__))
    parent_directory = os.path.dirname(current_directory)  # 获取上一级目录的路径
    
    result = find_json_files_with_character(parent_directory, search_character)

    if result:
        print(f"包含字段 '{search_character}' 的JSON文件有：")
        for file_path in result:
            print(os.path.basename(file_path))
        
        confirmation = input("是否要替换字段？(Y/N): ").strip().lower()
        if confirmation == 'y':
            replace_word_in_json_files(result, search_character, replace_character)
            print(f"已将所有包含字段 '{search_character}' 的JSON文件中的 '{search_character}' 替换为 '{replace_character}'。")
        else:
            print("未进行替换操作。")
    else:
        print(f"在当前目录的上一级目录下没有找到包含字段 '{search_character}' 的JSON文件。")

utils/convert.py
linruowuyin/Fhoe-Rail
"""
Author: Night-stars-1 nujj1042633805@gmail.com
Date: 2023-05-13 13:05:56
LastEditors: Night-stars-1 nujj1042633805@gmail.com
LastEditTime: 2023-05-13 13:08:28
FilePath: \Honkai-Star-Rail-beta-2.4h:\Download\Zip\Honkai-Star-Rail-beta-2.7\tool\exceptions.py
Description:

Copyright (c) 2023 by ${git_name_email}, All Rights Reserved.
"""
from .log import log
import traceback


class Exception(Exception):

    def __init__(self, message):
        super().__init__(message)
        log.error(message)
        log.debug(traceback.format_exc())

utils/exceptions.py
linruowuyin/Fhoe-Rail
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# Created by Xe-No at 2023/5/15
# 根据小地图箭头判定人物朝向，若想让摄像头方向与人物朝向一致，先按一次w即可。
# 获取角度可以为后续脚本开发提供帮助，比如自动矫正角度，模拟宇宙寻路等

import cv2
import numpy as np
import pyautogui
from switch_window import switch_window
import time



def get_furthest_point(points):
	# 计算中心点坐标
	center = np.mean(points, axis=0)
	# 初始化最大距离为 0，最远点为第一个点
	max_distance = 0
	furthest_point = points[0]
	# 枚举每个点
	for point in points:
		# 计算该点到中心点的距离
		distance = np.linalg.norm(point - center)
		# 如果该点到中心点的距离大于当前最大距离，则更新最大距离和最远点
		if distance > max_distance:
			max_distance = distance
			furthest_point = point
	return furthest_point

def get_angle(debug=False, use_sample_image=False):
	x,y = [117,128]
	w,h = [47,47]

	if use_sample_image:
		# 任意一张截图即可，最重要的是保留左上部分
		img = cv2.imread('picture/sample.png')[y:y+h, x:x+w]
	else:
		img = np.array(pyautogui.screenshot())[y:y+h, x:x+w] 
		img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) # pyautogui截取的图为RGB
	img0 = img.copy()
	hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
	
	# 定义青色掩膜
	lower_cyan = np.array([78, 200, 200]) # 使用较高饱和度下界，过滤掉摄像头圆弧
	upper_cyan = np.array([99, 255, 255])
	cyan_mask = cv2.inRange(hsv, lower_cyan, upper_cyan)

	# 查找轮廓
	contours, hierarchy = cv2.findContours(cyan_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

	# 暴力断定只有一个青色箭头，获取箭头顶点
	if len(contours) != 1:
		return False
	contour = contours[0]
	peri = cv2.arcLength(contour, True)
	approx = cv2.approxPolyDP(contour, 0.03 * peri, True)
	fp = get_furthest_point(approx[:,0,:])


	# 获取角度
	fx, fy = fp
	dx = fx - w // 2
	dy = fy - h // 2
	angle = np.degrees(np.arctan2(dy, dx))
	angle = np.around(angle, 2)	

	print(angle)
	if debug:
		# 画出原图、二值掩膜图以及轮廓图
		cv2.polylines(img, [approx], True, (0, 0, 255), thickness=2)
		cv2.circle(img, fp, 0, (255,255,255),9)
		cv2.imshow("result", np.hstack((img0,hsv,img)))
		cv2.waitKey(0)
		cv2.destroyAllWindows()

	return angle

def main():
	debug = False
	use_sample_image = False
	if not use_sample_image:
		switch_window()
		time.sleep(0.5)
	get_angle(debug,use_sample_image)

if __name__ == '__main__':
	main() 

utils/get_angle.py
linruowuyin/Fhoe-Rail
import os
import subprocess
import time

def find_requirements_file():
    # 尝试在当前目录查找requirements.txt
    current_dir = os.path.dirname(__file__)
    requirements_file = os.path.join(current_dir, "requirements.txt")
    if os.path.exists(requirements_file):
        return requirements_file

    # 如果在当前目录中找不到，尝试在上一级目录查找
    parent_dir = os.path.dirname(current_dir)
    requirements_file = os.path.join(parent_dir, "requirements.txt")
    if os.path.exists(requirements_file):
        return requirements_file

    # 如果都找不到，返回None
    return None

def test_speed(source):
    try:
        start_time = time.time()
        subprocess.check_call(["ping", "-n", "5", source])
        end_time = time.time()
        return end_time - start_time
    except subprocess.CalledProcessError:
        return float('inf')  # 返回无穷大表示测试失败

def set_fastest_proxy():
    aliyun_time = test_speed("mirrors.aliyun.com")
    tuna_time = test_speed("pypi.tuna.tsinghua.edu.cn")

    if aliyun_time < tuna_time:
        print("阿里云源延迟较低，设置为代理源...")
        subprocess.check_call(["pip", "config", "set", "global.index-url", "https://mirrors.aliyun.com/pypi/simple"])
    else:
        print("上海交通大学源延迟较低，设置为代理源...")
        subprocess.check_call(["pip", "config", "set", "global.index-url", "https://pypi.tuna.tsinghua.edu.cn/simple"])

def check_and_install_dependencies():
    requirements_file = find_requirements_file()

    if requirements_file:
        try:
            subprocess.check_call(["pip", "show", "-r", requirements_file])
            print("依赖已经存在！")
        except subprocess.CalledProcessError:
            print("安装依赖中，请稍等...")
            set_fastest_proxy()
            subprocess.check_call(["pip", "install", "-r", requirements_file])
            subprocess.check_call(["pip", "config", "unset", "global.index-url"])  # 恢复默认源
            print("安装完成！")
    else:
        print("无法找到requirements.txt文件！")

if __name__ == "__main__":
    check_and_install_dependencies()

utils/install_requirements.py
linruowuyin/Fhoe-Rail
'''
Author: Night-stars-1 nujj1042633805@gmail.com
Date: 2023-05-12 23:22:54
LastEditors: Night-stars-1 nujj1042633805@gmail.com
LastEditTime: 2023-05-14 01:22:36
FilePath: \Honkai-Star-Rail-beta-2.4h:\Download\Zip\Honkai-Star-Rail-beta-2.7\tools\log.py
Description: 

Copyright (c) 2023 by ${git_name_email}, All Rights Reserved. 
'''
import os
import sys
import requests
from loguru import logger

try:
    from .requests import post
except:
    from requests import post

VER = "3.1"
log = logger
dir_log = "logs"
path_log = os.path.join(dir_log, '日志文件.log')
logger.remove()
logger.add(sys.stdout, level='INFO', colorize=True,
            format="<cyan>{module}</cyan>.<cyan>{function}</cyan>"
                    ":<cyan>{line}</cyan> - "+f"<cyan>{VER}</cyan> - "
                    "<level>{message}</level>"
            )
logger.add(path_log,
            format="{time:HH:mm:ss} - "
                    "{level}\t| "
                    "{module}.{function}:{line} - "+f"<cyan>{VER}</cyan> - "+" {message}",
            rotation='0:00', enqueue=True, serialize=False, encoding="utf-8", retention="10 days")

def webhook_and_log(message):
    log.info(message)
    from .config import read_json_file # Circular import
    url = read_json_file("config.json", False).get("webhook_url")
    if url == "" or url == None:
        return
    try:
        post(url, json={"content": message})
    except Exception as e:
        log.error(f"Webhook发送失败: {e}")

def fetch_php_file_content():
    php_urls = [
        "https://wanghun.top/api/tgrj.php",
        "http://api.ay15.cn/api/tiangou/api.php?charset=utf-8"
    ]

    for url in php_urls:
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.text
        except (requests.exceptions.RequestException, requests.exceptions.HTTPError):
            pass

    return ""

utils/log.py
linruowuyin/Fhoe-Rail
import time
import cv2 as cv
import pyautogui
import os

from .calculated import Calculated
from .config import get_file, read_json_file, CONFIG_FILE_NAME
from .log import log, webhook_and_log

class Map:
    def __init__(self):
        self.calculated = Calculated()
        self.open_map = read_json_file(CONFIG_FILE_NAME).get("open_map", "m")
        self.map_list = []
        self.map_list_map = {}
        self.read_maps()

    def map_init(self):

        target = cv.imread('./picture/contraction.png')
        max_attempts = 60  # 最大重试次数
        attempts = 0

        while attempts < max_attempts:
            result = self.calculated.scan_screenshot(target)
            if result['max_val'] > 0.97:
                points = self.calculated.calculated(result, target.shape)
                log.debug(points)
                pyautogui.click(points, clicks=5, interval=0.1)
                break
            else:
                attempts += 1
            pyautogui.keyDown(self.open_map)
            pyautogui.keyUp(self.open_map)
            time.sleep(3)  # 3秒延迟

        if attempts == max_attempts:
            log.warning("未能识别到地图初始化图标，可能出现了问题")

    def read_maps(self):
        # 从'./map'目录获取地图文件列表（排除'old'）
        map_dir = './map'
        json_files = [f for f in os.listdir(map_dir) if f.endswith('.json') and not f.startswith('old')]
    
        self.map_list = json_files
        self.map_list_map.clear()
    
        for map_ in json_files:
            map_data = read_json_file(f"map/{map_}")
            key1 = map_[map_.index('_') + 1:map_.index('-')]
            key2 = map_[map_.index('-') + 1:map_.index('.')]
            value = self.map_list_map.get(key1)
        
            if value is None:
                value = {}
        
            value[key2] = map_data["name"]
            self.map_list_map[key1] = value
    
        log.debug(self.map_list)
        log.debug(self.map_list_map)

    def format_time(self, seconds):
        # 格式化时间
        minutes, seconds = divmod(seconds, 60)
        hours, minutes = divmod(minutes, 60)

        if hours > 0:
            return f"{hours:.0f}小时{minutes:.0f}分{seconds:.1f}秒"
        elif minutes > 0:
            return f"{minutes:.0f}分{seconds:.1f}秒"
        else:
            return f"{seconds:.1f}秒"

    def auto_map(self, start):
        total_processing_time = 0

        if f'map_{start}.json' in self.map_list:
            map_list = self.map_list[self.map_list.index(f'map_{start}.json'):len(self.map_list)]
            for map_ in map_list:
                # 选择地图
                map_ = map_.split('.')[0]
                map_data = read_json_file(f"map/{map_}.json")
                webhook_and_log(f"\033[0;96;40m{map_data['name']}\033[0m")
                self.calculated.monthly_pass()
                log.info(f"路线领航员：\033[1;95m{map_data['author']}\033[0m 感谢她(们)的无私奉献")
                for start in map_data['start']:
                    key = list(start.keys())[0]
                    log.debug(key)
                    value = start[key]
                    if key == 'map':
                        self.map_init()
                    else:
                        time.sleep(value)
                        self.calculated.click_target(key, 0.93)
                time.sleep(3)
                count = 0
                while self.calculated.is_blackscreen():
                    count += 1
                    time.sleep(1)
                log.info(f'地图加载完毕，用时 {count} 秒')
                time.sleep(2)  # 增加2秒等待防止人物未加载错轴

                # 记录处理开始时间
                start_time = time.time()

                self.calculated.auto_map(map_, False)

                # 记录处理结束时间
                end_time = time.time()

                # 计算处理时间并输出
                processing_time = end_time - start_time
                formatted_time = self.format_time(processing_time)
                total_processing_time += processing_time
                log.info(f"{map_}用时\033[1;92m『{formatted_time}』\033[0m,总计:\033[1;92m『{self.format_time(total_processing_time)}』\033[0m")

        else:
            log.info(f'地图编号 {start} 不存在，请尝试检查地图文件')

utils/map.py
linruowuyin/Fhoe-Rail
#!/usr/bin/env python3
# Author AlisaCat
# -*- encoding:utf-8 -*-
# Created by AlisaCat at 2023/5/7

import builtins
import time
from collections import defaultdict
from datetime import datetime
import os
import sys
import ctypes

import orjson
import win32api
import win32con
from pynput import keyboard
from pynput import mouse
from pynput.mouse import Controller as mouseController

from config import read_json_file


def timestamped_print(*args, **kwargs):
    currentDT = datetime.now().strftime('%H:%M:%S')
    builtins.print(str(currentDT), *args, **kwargs)


print = timestamped_print

# 获取管理员权限
def run_as_admin():
    try:
        # 检查是否已经有管理员权限
        if ctypes.windll.shell32.IsUserAnAdmin():
            return
        # 如果没有管理员权限，则使用管理员权限重新启动脚本
        script_path = os.path.abspath(sys.argv[0])
        ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, script_path, None, 1)
        return  # 修改这里，使用return退出函数而不是sys.exit(0)
    except Exception as e:
        print(f"Failed to run as admin: {e}")
        sys.exit(1)

# 在这里运行获取管理员权限的函数
run_as_admin()

print("3s后开启录制,F9终止保存")
time.sleep(3)
print("start")
# 获取到游戏中心点坐标
cen_mouse_pos = mouseController().position
print("中心点坐标", cen_mouse_pos)
mouse_watch = True

key_list = ['w', 's', 'a', 'd', 'f', 'x']  # 匹配锄大地
# 输出列表
event_list = []
# 不同操作间延迟记录
last_time = time.time()
# 按键按下的时间字典
# key_down_time = {}
# 创建一个默认值为0的字典
key_down_time = defaultdict(int)

# 控制是否输出调试信息的开关
debug_mode = True

mouse_move_pos_list = []

mouse_val = 200  # 每次视角移动距离


def Click(points):
    x, y = points
    win32api.SetCursorPos((x, y))
    win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0)
    time.sleep(0.5)
    win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0)


real_width = read_json_file("config.json")['real_width']


def on_press(key):
    global last_time, key_down_time, real_width
    try:
        if key.char in key_list and key.char in key_down_time:
            pass
        elif key.char in key_list:
            save_mouse_move_by_key()
            key_down_time[key.char] = time.time()
            if debug_mode:
                print("捕捉按键按下:", key.char, time.time())
    except AttributeError:
        pass


def on_release(key):
    global last_time, key_down_time, mouse_move_pos_list, cen_mouse_pos, mouse_watch
    current_time = time.time()
    try:
        if key.char in key_list and key.char in key_down_time:
            event_list.append(
                {'key': key.char, 'time_sleep': key_down_time[key.char] - last_time,
                 'duration': time.time() - key_down_time[key.char]})
            last_time = time.time()
            del key_down_time[key.char]
            if debug_mode:
                print("捕捉:", event_list[-1])
            if key.char == "x":
                if debug_mode:
                    print("捕捉X进入战斗")
                mouse_watch = False
                Click(cen_mouse_pos)
                mouse_watch = True
    except AttributeError:
        pass
    if key == keyboard.Key.left:
        x = mouse_val * -1
        dx = int(x * 1295 / real_width)
        win32api.mouse_event(win32con.MOUSEEVENTF_MOVE, dx, 0)  # 进行视角移动
        mouse_move_pos_list.append(
            {"mouse_move_dxy": (x, 0), "time_sleep": current_time - last_time})
        last_time = current_time
        if debug_mode:
            print("捕捉M:", "mouse_move_dxy", (x, 0), "MExec:", dx)
    elif key == keyboard.Key.right:
        x = mouse_val  # 200
        dx = int(x * 1295 / real_width)
        win32api.mouse_event(win32con.MOUSEEVENTF_MOVE, dx, 0)  # 进行视角移动
        mouse_move_pos_list.append(
            {"mouse_move_dxy": (x, 0), "time_sleep": current_time - last_time})
        last_time = current_time
        if debug_mode:
            print("捕捉M:", "mouse_move_dxy", (x, 0), "MExec:", dx)
    if key == keyboard.Key.f9:
        save_mouse_move_by_key()
        save_json()
        if debug_mode:
            print("保存")
        return False


def on_click(x, y, button, pressed):
    if mouse_watch:
        save_mouse_move_by_key()
        global last_time
        if pressed:
            event_list.append(
                {'key': 'click', 'time_sleep': time.time() - last_time})
            print("捕捉:", event_list[-1])
    else:
        pass


def save_mouse_move_by_key():
    global mouse_move_pos_list
    if mouse_move_pos_list:
        element = None
        mouse_dx = 0
        mouse_dy = 0
        for element in mouse_move_pos_list:
            mouse_dx += element["mouse_move_dxy"][0]
            mouse_dy += element["mouse_move_dxy"][1]
        event_list.append(
            {"mouse_move_dxy": (mouse_dx, mouse_dy), "time_sleep": element["time_sleep"]})
        print("视角相对距离计算完成:", mouse_dx, mouse_dy, element["time_sleep"])
    mouse_move_pos_list.clear()


def save_json():
    # with open('output.json', 'w') as f:
    #     json.dump(event_list, f, indent=4)

    normal_save_dict = {
        "name": "地图-编号",
        "author": "作者",
        "start": [],
        "map": []
    }
    for element_save in event_list:
        if 'key' in element_save:
            if element_save['key'] == "click":
                normal_save_dict["map"].append({"fighting": 2})
            elif element_save['key'] == "x":
                normal_save_dict["map"].append({"fighting": 1})  # 进战斗
            else:
                normal_save_dict["map"].append(
                    {element_save['key']: element_save['duration']})
        elif 'mouse_move_dxy' in element_save:
            normal_save_dict["map"].append(
                {"mouse_move": element_save['mouse_move_dxy'][0]})

    with open(f'output{datetime.now().strftime("%Y_%m_%d_%H_%M_%S")}.json', 'wb') as f:
        f.write(orjson.dumps(normal_save_dict, option=orjson.OPT_INDENT_2))


mouse_listener = mouse.Listener(on_click=on_click)  # , on_move=on_move
mouse_listener.start()

with keyboard.Listener(on_press=on_press, on_release=on_release) as listener:  # 创建按键监听线程
    listener.join()  # 等待按键监听线程结束

mouse_listener.stop()

utils/record.py
linruowuyin/Fhoe-Rail
import httpx
import tqdm.asyncio
from pathlib import Path
from typing import Dict, Optional, Any, Union, Tuple


async def get(url: str,
                *,
                headers: Optional[Dict[str, str]] = None,
                params: Optional[Dict[str, Any]] = None,
                timeout: Optional[int] = 20,
                **kwargs) -> httpx.Response:
    """
    说明：
        httpx的get请求封装
    参数：
        :param url: url
        :param headers: 请求头
        :param params: params
        :param data: data
        :param json: json
        :param timeout: 超时时间
    """
    async with httpx.AsyncClient() as client:
        return await client.get(url,
                                headers=headers,
                                params=params,
                                timeout=timeout,
                                **kwargs)

async def post(url: str,
                *,
                headers: Optional[Dict[str, str]] = None,
                params: Optional[Dict[str, Any]] = None,
                timeout: Optional[int] = 20,
                **kwargs) -> httpx.Response:
    """
    说明：
        httpx的post请求封装
    参数：
        :param url: url
        :param headers: 请求头
        :param params: params
        :param data: data
        :param json: json
        :param timeout: 超时时间
    """
    async with httpx.AsyncClient() as client:
        return await client.post(url,
                                headers=headers,
                                params=params,
                                timeout=timeout,
                                **kwargs)

async def download(url: str, save_path: Path):
    """
    说明：
        下载文件(带进度条)
    参数：
        :param url: url
        :param save_path: 保存路径
    """
    save_path.parent.mkdir(parents=True, exist_ok=True)
    async with httpx.AsyncClient().stream(method='GET', url=url, follow_redirects=True) as datas:
        size = int(datas.headers.get('Content-Length', 0))
        f = save_path.open('wb')
        async for chunk in tqdm.asyncio.tqdm(iterable=datas.aiter_bytes(1),
                                                desc=url.split('/')[-1],
                                                unit='iB',
                                                unit_scale=True,
                                                unit_divisor=1024,
                                                total=size,
                                                colour='green'):
            f.write(chunk)
        f.close()

utils/requests.py
linruowuyin/Fhoe-Rail
import tkinter as tk
from tkinter import messagebox
import os
from PIL import Image, ImageTk

window = tk.Tk()
window.title("倒计时强制关机程序")

background_image = Image.open("./picture/2.png")
background_photo = ImageTk.PhotoImage(background_image)

background_label = tk.Label(window, image=background_photo)
background_label.place(relwidth=1, relheight=1)

font_style = ("SimSun", 25)

countdown_minutes = tk.StringVar()
countdown_minutes.set("240")

counting_down = False

timer_id = [None]

def start_countdown():
    global counting_down
    try:
        minutes = int(countdown_minutes.get())
        if minutes <= 0:
            raise ValueError
        countdown_seconds = minutes * 60
        counting_down = True
        minutes_entry.config(state="disabled")
        start_button.config(state="disabled")
        cancel_button.config(state="normal")

        def update_countdown():
            nonlocal countdown_seconds
            if countdown_seconds > 0 and counting_down:
                countdown_seconds -= 1
                countdown_minutes.set(
                    str(countdown_seconds // 60) + "分" + str(countdown_seconds % 60) + "秒"
                )
                timer_id[0] = window.after(1000, update_countdown)
            else:
                if counting_down:
                    os.system("shutdown /s /t 1")
        
        update_countdown()
    except ValueError:
        messagebox.showerror("错误", "请输入一个有效的正整数分钟数")

def cancel_countdown():
    global counting_down
    counting_down = False
    if timer_id[0]:
        window.after_cancel(timer_id[0])
    minutes_entry.config(state="normal")
    start_button.config(state="normal")
    cancel_button.config(state="disabled")

def start_initial_countdown():
    initial_minutes = int(countdown_minutes.get())
    countdown_minutes.set(str(initial_minutes))
    start_countdown()

window.after(30000, start_initial_countdown)

countdown_label = tk.Label(window, textvariable=countdown_minutes, font=font_style)
countdown_label.pack(pady=20)

minutes_entry = tk.Entry(window, textvariable=countdown_minutes, font=font_style, width=11, justify='center')
minutes_entry.pack(pady=11)

start_button = tk.Button(
    window,
    text="开始倒计时",
    command=start_countdown,
    font=("SimSun", 25),
    width=10,
)
start_button.pack(pady=10)

cancel_button = tk.Button(
    window,
    text="取消倒计时",
    command=cancel_countdown,
    font=("SimSun", 25),
    width=10,
    state="disabled",
)
cancel_button.pack(pady=10)

window.geometry("468x358")
window.mainloop()

utils/shutdown.py
linruowuyin/Fhoe-Rail
import os
import pyautogui
from win32com import client
import time
import random

def switch_window(title='崩坏：星穹铁道'):
    lnk_started = False  # 添加一个标志，用于记录lnk文件是否已经启动
    
    while True:
        if not lnk_started:  # 如果lnk文件未启动，继续查找窗口
            ws = pyautogui.getWindowsWithTitle(title)

            if len(ws) >= 1:
                for w in ws:
                    # 避免其他窗口也包含崩坏：星穹铁道，比如正好开着github脚本页面
                    # print(w.title)
                    if w.title == title:
                        client.Dispatch("WScript.Shell").SendKeys('%')
                        w.activate()
                        return  # 找到窗口后直接返回函数

            print(f'没有找到窗口【{title}】')
        
        # 尝试查找并启动lnk文件
        lnk_files = find_lnk_files("./map")  # 指定lnk文件所在的目录
        if len(lnk_files) == 1 and not lnk_started:
            start_lnk_file(lnk_files[0])
            print("程序启动成功！")
            lnk_started = True  # 设置标志为True
            time.sleep(25)  # 等待25秒
            
            end_time = time.time() + 30  # 执行点击操作的总时间为30秒
            last_5_seconds = end_time - 5  # 最后5秒
            
            while time.time() < end_time:
                if time.time() >= last_5_seconds:
                    # 在最后5秒内至少点击5次
                    for _ in range(5):
                        x, y = (1167, 732)  # 修改点击坐标为(1167, 732)
                        pyautogui.click(x, y)
                        time.sleep(random.uniform(0.1, 5))  # 随机点击频率
                else:
                    x, y = (1167, 732)  # 修改点击坐标为(1167, 732)
                    pyautogui.click(x, y)
                    time.sleep(random.uniform(0.1, 5))  # 随机点击频率
        
        if lnk_started:
            # 如果lnk文件已经启动成功，不再等待用户按下回车键，直接退出
            return

        # 等待用户按下回车键后继续
        input("按下回车键继续查找窗口...")

def find_lnk_files(folder_path):
    lnk_files = []
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            if file.lower().endswith('.lnk'):
                lnk_files.append(os.path.join(root, file))
    return lnk_files

def start_lnk_file(lnk_path):
    os.system(f'start "" "{lnk_path}"')

if __name__ == '__main__':
    switch_window()

utils/switch_window.py
linruowuyin/Fhoe-Rail
#!/usr/bin/env python3
# Author AlisaCat
# -*- encoding:utf-8 -*-
# Created by AlisaCat at 2023/5/11

import os
import time
import shutil
import asyncio
import hashlib

from pathlib import Path
from tqdm import tqdm as tq
from zipfile import ZipFile, BadZipFile
from typing import Dict, Optional, Any, Union, Tuple, List

from .log import log
from .requests import *
from .exceptions import Exception
from .config import normalize_file_path, modify_json_file, read_json_file, get_file, CONFIG_FILE_NAME

tmp_dir = 'tmp'

async def verify_file_hash(json_path: Path, keep_file: Optional[List[str]] = []) -> bool:
    """
    说明：
        校验文件
    参数：
        :param json_path: 文件地址
    返回:
        :return bool
    """
    for data in json_path:
        file_path = Path() / data['path']
        if not os.path.exists(file_path):
            return False, file_path
        if os.path.isfile(file_path) and str(file_path) not in keep_file:
            log.debug(hashlib.md5(file_path.read_bytes()).hexdigest())
            if hashlib.md5(file_path.read_bytes()).hexdigest() != data['hash']:
                return False, file_path
    return True, None

async def unzip(zip, zip_path: Path):
    global tmp_dir
    with ZipFile(zip, 'r') as zf:
        for member in tq(zf.infolist(), desc='解压中'):
            if member.filename.startswith(zip_path):
                zf.extract(member, tmp_dir)
                log.debug(f'[资源文件更新]正在提取{member.filename}')

async def remove_file(folder_path: Path,keep_folder: Optional[List[str]] = [],keep_file: Optional[List[str]] = []) -> None:
    if os.path.exists(folder_path):
        for item in os.listdir(folder_path):
            item_path = os.path.join(folder_path, item)
            if os.path.isdir(item_path) and item not in keep_folder:
                shutil.rmtree(item_path)
            elif os.path.isfile(item_path) and item not in keep_file:
                os.remove(item_path)

async def move_file(src_folder: Path, dst_folder,keep_folder: Optional[List[str]] = [],keep_file: Optional[List[str]] = []) -> None:


    for item in get_file(src_folder,keep_folder,keep_file, True):
        if dst_folder in item:
            dst_path = item.replace(src_folder, "./")

            # 创建目标文件夹（如果不存在）
            if not os.path.exists(dst_path.rsplit("/",1)[0]) and dst_path.rsplit("/",1)[0] != '.':
                os.makedirs(dst_path.rsplit("/",1)[0])
            #dst_path = os.path.join(dst_folder, item)
            shutil.copy(item, dst_path)
    # 遍历源文件夹中的所有文件和文件夹
    '''
    for item in os.listdir(src_folder):
        # 构造源文件路径和目标文件路径
        src_path = os.path.join(src_folder, item)
        dst_path = os.path.join(dst_folder, item)
        if os.path.isdir(src_path) and item not in keep_folder:
            shutil.copy(src_path, dst_path)
        if os.path.isfile(src_path) and item not in keep_file:
            shutil.copy(src_path, dst_path)
    '''

async def update_file(url_proxy: str="",
                      raw_proxy: str="",
                      rm_all: bool=False, 
                      skip_verify: bool=True,
                      type: str="",
                      version: str="",
                      url_zip: str="",
                      unzip_path: str="",
                      keep_folder: Optional[List[str]] = [],
                      keep_file: Optional[List[str]] = [],
                      zip_path: str="",
                      name: str="") -> bool:
    """
    说明：
        更新文件
    参数：
        :param url_proxy: github代理
        :param raw_proxy: rawgithub代理
        :param rm_all: 是否强制删除文件
        :param skip_verify: 是否跳过检验
        :param type: 更新文件的类型 map\picture
        :param version: 版本验证地址 map
        :param url_zip: zip下载链接
        :param unzip_path: 解压地址（删除用）
        :param keep_folder: 保存的文件夹
        :param keep_file: 保存的文件
        :param zip_path: 需要移动的文件地址
        :param name: 更新的文件名称
    """
    global tmp_dir

    url_version = f'{raw_proxy}https://raw.githubusercontent.com/Starry-Wind/SRA/{version}/version.json' if 'http' in raw_proxy or raw_proxy == '' else f'https://raw.githubusercontent.com/Starry-Wind/SRA/{version}/version.json'.replace('raw.githubusercontent.com', raw_proxy)
    url_zip = url_proxy+url_zip if 'http' in url_proxy or url_proxy == '' else url_zip.replace('github.com', url_proxy)
    url_list = f'{raw_proxy}https://raw.githubusercontent.com/Starry-Wind/SRA/{version}/{type}_list.json' if 'http' in raw_proxy or raw_proxy == '' else f'https://raw.githubusercontent.com/Starry-Wind/SRA/{version}/{type}_list.json'.replace('raw.githubusercontent.com', raw_proxy)
    
    #tmp_zip = os.path.join(tmp_dir, f'{type}.zip')
    tmp_zip = Path() / tmp_dir / f'{type}.zip'
    if not os.path.exists(tmp_dir):
        os.makedirs(tmp_dir)
    if not os.path.exists(unzip_path):
        os.makedirs(unzip_path)
        modify_json_file(CONFIG_FILE_NAME, f"{type}_version", "0")
    elif rm_all:
        modify_json_file(CONFIG_FILE_NAME, f"{type}_version", "0")

    log.info(f'[资源文件更新]正在检查远程版本是否有更新...')

    for index, _ in enumerate(range(3)):
        try:
            remote_version = await get(url_version)
            remote_version = remote_version.json()['version']
            break
        except BaseException as e:
            if index < 2:
                log.info(f'[资源文件更新]获取远程版本失败, 正在重试: {e}')
            else:
                log.info(f'[资源文件更新]获取远程版本失败: {e}')
            log.info("将在10秒后重试")
            await asyncio.sleep(10)
    else:
        log.info(f'[资源文件更新]重试次数已达上限，退出程序')
        raise Exception(f'[资源文件更新]重试次数已达上限，退出程序')

    log.info(f'[资源文件更新]获取远程版本成功: {remote_version}')

    local_version = read_json_file(CONFIG_FILE_NAME).get(f'{type}_version', '0')

    if remote_version != local_version:
        log.info(f'[资源文件更新]本地版本与远程版本不符，开始更新资源文件->{url_zip}')
        for _ in range(3):
            try:
                await download(url_zip, tmp_zip)
                log.info(f'[资源文件更新]下载更新包成功, 正在覆盖本地文件: {local_version} -> {remote_version}')
                await unzip(tmp_zip, zip_path)
                break
            except BadZipFile:
                log.info(f'[资源文件更新]下载压缩包失败, 重试中: BadZipFile')
            except BaseException as e:
                log.info(f'[资源文件更新]下载压缩包失败: {e}')
            log.info("将在10秒后重试")
            await asyncio.sleep(10)
        else:
            log.info(f'[资源文件更新]重试次数已达上限，退出程序')
            raise Exception(f'[资源文件更新]重试次数已达上限，退出程序')


        #shutil.rmtree('..\SRA-beta-2.7')
        #await remove_file(unzip_path, keep_folder, keep_file)
        await move_file(os.path.join(tmp_dir, zip_path), unzip_path, [], keep_file)

        log.info(f'[资源文件更新]正在校验资源文件')
        for _ in range(3):
            try:
                map_list = await get(url_list)
                map_list = map_list.json()
                break
            except BaseException as e:
                log.info(f'[资源文件更新]校验文件下载失败: {e}')
            log.info("将在10秒后重试")
            await asyncio.sleep(10)
        else:
            log.info(f'[资源文件更新]重试次数已达上限，退出程序')
            raise Exception(f'[资源文件更新]重试次数已达上限，退出程序')
        
        verify, path = await verify_file_hash(map_list, keep_file)
        if not verify:
            raise Exception(f"[资源文件更新]{path}校验失败, 程序退出")

        log.info(f'[资源文件更新]校验完成, 更新本地{name}文件版本号 {local_version} -> {remote_version}')

        # 更新版本号
        modify_json_file(CONFIG_FILE_NAME, f"{type}_version", remote_version)

        log.info(f'[资源文件更新]删除临时文件{tmp_dir}')
        shutil.rmtree(tmp_dir, ignore_errors=True)
    else:
        log.info(f'[资源文件更新]资源文件已是最新版本 {local_version} = {remote_version}')
        if not skip_verify:
            log.info(f'[资源文件更新]准备校验资源文件')
            for index, _ in enumerate(range(3)):
                try:
                    remote_map_list = await get(url_list)
                    remote_map_list = remote_map_list.json()
                    break
                except BaseException as e:
                    if index < 2:
                        log.info(f'[资源文件更新]获取{name}文件列表失败, 正在重试: {e}')
                    else:
                        log.info(f'[资源文件更新]获取{name}文件列表失败: {e}')
                    log.info("将在10秒后重试")
                    await asyncio.sleep(10)
            else:
                log.info(f'[资源文件更新]获取{name}文件列表重试次数已达上限，退出程序')
                raise Exception(f'[资源文件更新]获取{name}文件列表重试次数已达上限，退出程序')

            log.debug(f'[资源文件更新]获取{name}文件列表成功.')


            verify, path = await verify_file_hash(remote_map_list, keep_file)
            if not verify:
                log.error(f"[资源文件更新]{path}发现文件缺失, 3秒后将使用远程版本覆盖本地版本")
                return "rm_all"
            log.info(f'[资源文件更新]文件校验完成.')

    log.info(f'[资源文件更新]更新完成.')
    return True


def update_file_main(url_proxy: str="",
                    raw_proxy: str="",
                    rm_all: bool=False, 
                    skip_verify: bool=True,
                    type: str="",
                    version: str="",
                    url_zip: str="",
                    unzip_path: str="",
                    keep_folder: Optional[List[str]] = [],
                    keep_file: Optional[List[str]] = [],
                    zip_path: str="",
                    name: str=""):
    """
    说明：
        更新文件
    参数：
        :param url_proxy: github代理
        :param raw_proxy: rawgithub代理
        :param skip_verify: 是否跳过检验
        :param type: 更新文件的类型 map\picture
        :param version: 版本验证地址 map
        :param url_zip: zip下载链接
        :param unzip_path: 解压地址（删除用）
        :param keep_folder: 保存的文件夹
        :param keep_file: 保存的文件
        :param zip_path: 需要移动的文件地址
        :param name: 更新的文件名称
    """
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    log.info(f'[资源文件更新]即将资源文件更新，本操作会覆盖本地{name}文件..')
    check_file_status = asyncio.run(update_file(url_proxy,raw_proxy,False,skip_verify,type,version,url_zip,unzip_path,keep_folder,keep_file,zip_path,name))
    if check_file_status == "rm_all":
        time.sleep(3)
        check_file_status = asyncio.run(update_file(url_proxy,raw_proxy,True,skip_verify,type,version,url_zip,unzip_path,keep_folder,keep_file,zip_path,name))
    elif check_file_status == "download_error":
        check_file_status = asyncio.run(update_file(url_proxy,raw_proxy,False,skip_verify,type,version,url_zip,unzip_path,keep_folder,keep_file,zip_path,name))

utils/update_file.py
linruowuyin/Fhoe-Rail