yuxing/Onekey
1
0
Fork 0
mirror of https://github.com/ikunshare/Onekey.git synced 2026-01-12 16:25:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
ikun0014
2025-02-27 19:15:20 +08:00
parent 7ef7297119
commit 09a9e48f7e
1 changed files with 50 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
main.py
View File

@@ -1,8 +1,8 @@
DEFAULT_CONFIG = {
"Github_Personal_Token": "",
"Custom_Steam_Path": "",
"QA1": "Github Personal Token可在GitHub设置的Developer settings中生成",
"教程": "https://ikunshare.com/Onekey_tutorial",
"QA1": "Github Personal Token可在GitHub设置的Developer settings中生成",
"教程": "https://ikunshare.com/Onekey_tutorial",
}
import platform
@@ -22,16 +22,16 @@ from enum import Enum
class RepoChoice(Enum):
IKUN = ("ikun0014/ManifestHub", "已断更的旧仓库")
AUIOWU = ("Auiowu/ManifestAutoUpdate", "未知维护状态的仓库")
STEAM_AUTO = ("SteamAutoCracks/ManifestHub", "官方推荐仓库")
IKUN = ("ikun0014/ManifestHub", "已断更的旧仓库")
AUIOWU = ("Auiowu/ManifestAutoUpdate", "未知维护状态的仓库")
STEAM_AUTO = ("SteamAutoCracks/ManifestHub", "官方推荐仓库")
DEFAULT_CONFIG = {
"Github_Personal_Token": "",
"Custom_Steam_Path": "",
"QA1": "Github Personal Token可在GitHub设置的Developer settings中生成",
"教程": "https://ikunshare.com/Onekey_tutorial",
"QA1": "Github Personal Token可在GitHub设置的Developer settings中生成",
"教程": "https://ikunshare.com/Onekey_tutorial",
}
DEFAULT_REPO = RepoChoice.STEAM_AUTO
@@ -45,7 +45,7 @@ log = logzero.setup_logger("Onekey")
def init() -> None:
"""初始化控制台输出"""
"""初始化控制台输出"""
banner = r"""
_____ __ _ _____ _ _ _____ __ __
/ _ \ | \ | | | ____| | | / / | ____| \ \ / /
@@ -55,55 +55,55 @@ def init() -> None:
\_____/ |_| \_| |_____| |_| \_\ |_____| /_/
"""
print(banner)
print("作者: ikun0014 | 版本: 1.3.7 | 官网: ikunshare.com")
print("项目仓库: GitHub: https://github.com/ikunshare/Onekey")
print("提示: 请确保已安装最新版Windows 10/11并正确配置Steam")
print("作者: ikun0014 | 版本: 1.3.7 | 官网: ikunshare.com")
print("项目仓库: GitHub: https://github.com/ikunshare/Onekey")
print("提示: 请确保已安装最新版Windows 10/11并正确配置Steam")
def validate_windows_version() -> None:
"""验证Windows版本"""
"""验证Windows版本"""
if platform.system() != "Windows":
log.error("仅支持Windows操作系统")
log.error("仅支持Windows操作系统")
sys.exit(1)
release = platform.uname().release
if release not in WINDOWS_VERSIONS:
log.error(f"需要Windows 10/11，当前版本: Windows {release}")
log.error(f"需要Windows 10/11，当前版本: Windows {release}")
sys.exit(1)
async def load_config() -> Dict[str, Any]:
"""异步加载配置文件"""
"""异步加载配置文件"""
if not CONFIG_PATH.exists():
await generate_config()
log.info("请填写配置文件后重新运行程序")
log.info("请填写配置文件后重新运行程序")
sys.exit(0)
try:
async with aiofiles.open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.loads(await f.read())
except json.JSONDecodeError:
log.error("配置文件损坏，正在重新生成...")
log.error("配置文件损坏，正在重新生成...")
await generate_config()
sys.exit(1)
except Exception as e:
log.error(f"配置加载失败: {str(e)}")
log.error(f"配置加载失败: {str(e)}")
sys.exit(1)
async def generate_config() -> None:
"""生成默认配置文件"""
"""生成默认配置文件"""
try:
async with aiofiles.open(CONFIG_PATH, "w", encoding="utf-8") as f:
await f.write(json.dumps(DEFAULT_CONFIG, indent=2, ensure_ascii=False))
log.info("配置文件已生成")
log.info("配置文件已生成")
except IOError as e:
log.error(f"配置文件创建失败: {str(e)}")
log.error(f"配置文件创建失败: {str(e)}")
sys.exit(1)
def get_steam_path(config: Dict) -> Path:
"""获取Steam安装路径"""
"""获取Steam安装路径"""
try:
if custom_path := config.get("Custom_Steam_Path"):
return Path(custom_path)
@@ -111,23 +111,23 @@ def get_steam_path(config: Dict) -> Path:
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, STEAM_REG_PATH) as key:
return Path(winreg.QueryValueEx(key, "SteamPath")[0])
except Exception as e:
log.error(f"Steam路径获取失败: {str(e)}")
log.error(f"Steam路径获取失败: {str(e)}")
sys.exit(1)
async def download_key(file_path: str, repo: str, sha: str) -> bytes:
"""下载密钥文件"""
"""下载密钥文件"""
try:
return await fetch_from_cdn(file_path, repo, sha)
except Exception as e:
log.error(f"密钥下载失败: {str(e)}")
log.error(f"密钥下载失败: {str(e)}")
raise
async def handle_depot_files(
repo: str, app_id: str, steam_path: Path
) -> List[Tuple[str, str]]:
"""处理清单文件和密钥"""
"""处理清单文件和密钥"""
collected = []
try:
async with httpx.AsyncClient() as client:
@@ -154,27 +154,27 @@ async def handle_depot_files(
)
collected.extend(parse_key_vdf(key_content))
except httpx.HTTPStatusError as e:
log.error(f"HTTP错误: {e.response.status_code}")
log.error(f"HTTP错误: {e.response.status_code}")
except Exception as e:
log.error(f"文件处理失败: {str(e)}")
log.error(f"文件处理失败: {str(e)}")
return collected
async def download_manifest(path: str, save_dir: Path, repo: str, sha: str) -> None:
"""下载清单文件"""
"""下载清单文件"""
save_path = save_dir / path
if save_path.exists():
log.warning(f"清单已存在: {path}")
log.warning(f"清单已存在: {path}")
return
content = await fetch_from_cdn(path, repo, sha)
async with aiofiles.open(save_path, "wb") as f:
await f.write(content)
log.info(f"清单下载成功: {path}")
log.info(f"清单下载成功: {path}")
async def fetch_from_cdn(path: str, repo: str, sha: str) -> bytes:
"""从CDN获取资源"""
"""从CDN获取资源"""
mirrors = (
[
f"https://jsdelivr.pai233.top/gh/{repo}@{sha}/{path}",
@@ -192,36 +192,36 @@ async def fetch_from_cdn(path: str, repo: str, sha: str) -> bytes:
return res.content
except httpx.HTTPError:
continue
raise Exception("所有镜像源均不可用")
raise Exception("所有镜像源均不可用")
def parse_key_vdf(content: bytes) -> List[Tuple[str, str]]:
"""解析密钥文件"""
"""解析密钥文件"""
try:
depots = vdf.loads(content.decode("utf-8"))["depots"]
return [(d_id, d_info["DecryptionKey"]) for d_id, d_info in depots.items()]
except Exception as e:
log.error(f"密钥解析失败: {str(e)}")
log.error(f"密钥解析失败: {str(e)}")
return []
async def setup_unlock_tool(
config: Dict, depot_data: List[Tuple[str, str]], app_id: str, tool_choice: int
) -> bool:
"""配置解锁工具"""
"""配置解锁工具"""
if tool_choice == 1:
return await setup_steamtools(config, depot_data, app_id)
elif tool_choice == 2:
return await setup_greenluma(config, depot_data)
else:
log.error("无效的工具选择")
log.error("无效的工具选择")
return False
async def setup_steamtools(
config: Dict, depot_data: List[Tuple[str, str]], app_id: str
) -> bool:
"""配置SteamTools"""
"""配置SteamTools"""
steam_path = (
Path(config["Custom_Steam_Path"])
if config.get("Custom_Steam_Path")
@@ -247,13 +247,13 @@ async def setup_steamtools(
await proc.wait()
if proc.returncode != 0:
log.error(f"Lua编译失败: {await proc.stderr.read()}")
log.error(f"Lua编译失败: {await proc.stderr.read()}")
return False
return True
async def setup_greenluma(config: Dict, depot_data: List[Tuple[str, str]]) -> bool:
"""配置GreenLuma"""
"""配置GreenLuma"""
steam_path = (
Path(config["Custom_Steam_Path"])
if config.get("Custom_Steam_Path")
@@ -280,39 +280,39 @@ async def setup_greenluma(config: Dict, depot_data: List[Tuple[str, str]]) -> bo
async def main_flow():
"""主流程控制"""
"""主流程控制"""
validate_windows_version()
init()
try:
app_id = input("请输入游戏AppID: ").strip()
app_id = input("请输入游戏AppID: ").strip()
if not app_id.isdigit():
raise ValueError("无效的AppID")
raise ValueError("无效的AppID")
print(
"\n".join(
[f"{idx+1}. {item.value[1]}" for idx, item in enumerate(RepoChoice)]
)
)
repo_choice = int(input("请选择清单仓库 (默认3): ") or 3)
repo_choice = int(input("请选择清单仓库 (默认3): ") or 3)
selected_repo = list(RepoChoice)[repo_choice - 1].value[0]
tool_choice = int(input("请选择解锁工具 (1.SteamTools 2.GreenLuma): "))
tool_choice = int(input("请选择解锁工具 (1.SteamTools 2.GreenLuma): "))
config = await load_config()
steam_path = get_steam_path(config)
depot_data = await handle_depot_files(selected_repo, app_id, steam_path)
if await setup_unlock_tool(config, depot_data, app_id, tool_choice):
log.info("游戏解锁配置成功！")
log.info("游戏解锁配置成功！")
if tool_choice == 1:
log.info("请重启SteamTools生效")
log.info("请重启SteamTools生效")
elif tool_choice == 2:
log.info("请重启GreenLuma生效")
log.info("请重启GreenLuma生效")
else:
log.error("配置失败，请检查日志")
log.error("配置失败，请检查日志")
except Exception as e:
log.error(f"运行错误: {str(e)}")
log.error(f"运行错误: {str(e)}")
log.debug(traceback.format_exc())
finally:
await CLIENT.aclose()