diff --git a/patch_installer.py b/patch_installer.py new file mode 100644 index 0000000..39948c9 --- /dev/null +++ b/patch_installer.py @@ -0,0 +1,249 @@ +import io +import json +import os +from pathlib import Path +import shutil +import tempfile +import time +import zipfile +import requests +from tqdm import tqdm + +# 모드팩 및 Git 서버 설정 +MODPACK_NAME = "Society Sunlit Valley_JGL_Fixed" +DOMAIN = "https://git.serin.one" +OWNER_REPO = "abellelus7074/jgl_sunlit_patch" +BRANCH = "main" + +# 관리 및 동기화할 Target 디렉터리 구분 +COMPARE_TARGETS = ["mods"] +CLEAR_TARGETS = ["config", "kubejs"] + +# CurseForge 및 로컬 패처 설정 경로 +CURSEFORGE_INST_PATH = ( + Path(os.getenv("APPDATA")) + / "CurseForge/agent/GameInstances/MinecraftGameInstance.json" +) +LOCAL_VERSION_INST_PATH = ( + Path(os.getenv("APPDATA")) + / "CurseForge/agent/GameInstances/patcher_config.json" +) + + +def get_local_commit_sha() -> str | None: + """로컬에 저장된 마지막 패치 완료 커밋 SHA 로드""" + if not LOCAL_VERSION_INST_PATH.exists(): + return None + try: + with open(LOCAL_VERSION_INST_PATH, "r", encoding="utf-8") as f: + data = json.load(f) + return data.get("commit_sha") + except Exception: + return None + + +def update_local_commit_sha(commit_sha: str): + """패치 완료 후 최신 커밋 SHA 정보를 로컬에 저장""" + LOCAL_VERSION_INST_PATH.parent.mkdir(parents=True, exist_ok=True) + with open(LOCAL_VERSION_INST_PATH, "w", encoding="utf-8") as f: + json.dump({"commit_sha": commit_sha}, f, ensure_ascii=False, indent=4) + + +def get_latest_remote_commit_sha() -> str | None: + """Git API를 통해 최신 브랜치 커밋 SHA 가져오기""" + url = f"{DOMAIN}/api/v1/repos/{OWNER_REPO}/branches/{BRANCH}" + try: + r = requests.get(url, timeout=10) + r.raise_for_status() + return r.json().get("commit", {}).get("id") + except Exception as e: + print(f"[ERROR] 원격 서버 최신 커밋 조회 실패: {e}") + return None + + +def download_folder_zip_bytes(target_dir_name: str) -> io.BytesIO | None: + """Gitea Archive API를 이용하여 특정 폴더 단위로 ZIP 파일 다운로드 (메모리 버퍼 반환)""" + clean_target = target_dir_name.strip("/\\") + zip_url = f"{DOMAIN}/{OWNER_REPO}/archive/{BRANCH}.zip?path={clean_target}" + print(f"[{clean_target}] 폴더 아카이브 다운로드 중...") + print(f"다운로드 경로 : {zip_url}") + + try: + response = requests.get(zip_url, stream=True, timeout=60) + response.raise_for_status() + + total_size = int(response.headers.get("content-length", 0)) + bytes_io = io.BytesIO() + + with tqdm( + total=total_size, + unit="B", + unit_scale=True, + unit_divisor=1024, + desc=f"[{clean_target}] 다운로드", + ncols=80, + ) as pbar: + for chunk in response.iter_content(chunk_size=1024 * 64): + if chunk: + bytes_io.write(chunk) + pbar.update(len(chunk)) + + bytes_io.seek(0) + return bytes_io + except Exception as e: + print(f"[ERROR] [{clean_target}] 아카이브 다운로드 실패: {e}") + return None + + +def extract_zip_to_directory(zip_bytes: io.BytesIO, target_dir_name: str, extract_path: Path): + """ZIP 메타데이터 최상위 접두사를 제거하고 지정된 경로에 압축 풀기""" + clean_target = target_dir_name.strip("/\\") + with zipfile.ZipFile(zip_bytes) as zip_file: + namelist = zip_file.namelist() + if not namelist: + return + + # Gitea ZIP 내부 최상위 루트 디렉터리 명칭 추출 + root_prefix = namelist[0].split("/")[0] + "/" + + for member in namelist: + if member.endswith("/"): + continue + + rel_path = member[len(root_prefix):] if member.startswith(root_prefix) else member + + if not rel_path.startswith(f"{clean_target}/"): + rel_path = f"{clean_target}/{rel_path}" + + # target_dir_name 내부 상대 경로 구하기 + sub_rel_path = rel_path[len(clean_target):].lstrip("/\\") + dest_file = extract_path / sub_rel_path + dest_file.parent.mkdir(parents=True, exist_ok=True) + + with zip_file.open(member) as source, open(dest_file, "wb") as target_file: + shutil.copyfileobj(source, target_file) + + +def sync_clear_target(target_dir_name: str, modpack_path: Path): + """삭제 로직: 기존 폴더 완전 삭제 후 압축 해제""" + clean_target = target_dir_name.strip("/\\") + local_dir = modpack_path / clean_target + print(f"삭제 로직 경로 : {local_dir}") + print(f"[{clean_target}] 로컬 폴더 초기화(삭제) 중...") + if local_dir.exists(): + shutil.rmtree(local_dir, ignore_errors=True) + local_dir.mkdir(parents=True, exist_ok=True) + + zip_bytes = download_folder_zip_bytes(clean_target) + if zip_bytes: + print(f"[{clean_target}] 압축 해제 중...") + extract_zip_to_directory(zip_bytes, clean_target, local_dir) + + +def sync_compare_target(target_dir_name: str, modpack_path: Path): + """비교 로직: 임시 폴더에 압축 해제 후 다운받은 파일과 비교하여 삭제 및 추가/덮어쓰기""" + clean_target = target_dir_name.strip("/\\") + local_dir = modpack_path / clean_target + print(f"비교 로직 경로 : {local_dir}") + local_dir.mkdir(parents=True, exist_ok=True) + + zip_bytes = download_folder_zip_bytes(clean_target) + if not zip_bytes: + return + + # 임시 디렉터리에 다운로드한 압축 해제 + with tempfile.TemporaryDirectory() as temp_dir_str: + temp_dir = Path(temp_dir_str) + print(f"[{clean_target}] 임시 폴더에 압축 해제 중...") + extract_zip_to_directory(zip_bytes, clean_target, temp_dir) + + # 1. 원격(다운로드된 파일) 목록 세트 생성 (소문자 상대 경로) + remote_files = set() + for downloaded_file in temp_dir.rglob("*"): + if downloaded_file.is_file(): + rel_p = downloaded_file.relative_to(temp_dir).as_posix().lower() + remote_files.add(rel_p) + + # 2. 로컬 파일 대조 및 다운로드 목록에 없는 파일 삭제 + print(f"[{clean_target}] 구버전/유실 파일 삭제 비교 중...") + for local_file in local_dir.rglob("*"): + if local_file.is_file(): + rel_p = local_file.relative_to(local_dir).as_posix().lower() + if rel_p not in remote_files: + print(f"삭제 중 (원격에 없음): {local_file.relative_to(modpack_path)}") + local_file.unlink(missing_ok=True) + + # 3. 임시 폴더의 파일들을 로컬 폴더로 복사 (신규 추가 및 덮어쓰기) + print(f"[{clean_target}] 최신 파일 적용(복사/덮어쓰기) 중...") + for downloaded_file in temp_dir.rglob("*"): + if downloaded_file.is_file(): + rel_path = downloaded_file.relative_to(temp_dir) + dest_file = local_dir / rel_path + dest_file.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(downloaded_file, dest_file) + + +def get_modpack_path(): + """CurseForge 설정 파일에서 지정된 모드팩 설치 경로 추출""" + if not CURSEFORGE_INST_PATH.exists(): + return "NOT_INSTALLED" + + with open(CURSEFORGE_INST_PATH, "r", encoding="utf-8") as f: + data = json.load(f) + + instances = data if isinstance(data, list) else data.get("instances", []) + return next( + ( + item.get("installPath") + for item in instances + if item.get("name") == MODPACK_NAME + ), + "NULL_MODPACK", + ) + + +if __name__ == "__main__": + modpack_path_str = get_modpack_path() + if modpack_path_str in ["NOT_INSTALLED", "NULL_MODPACK"]: + print("[ERROR] CurseForge 또는 지정된 모드팩이 설치되지 않았습니다.") + print("10초 후 종료합니다.") + time.sleep(10) + raise SystemExit + + modpack_path = Path(modpack_path_str) + + remote_commit_sha = get_latest_remote_commit_sha() + if not remote_commit_sha: + print("[ERROR] 서버 커밋 정보를 가져오지 못했습니다.") + print("10초 후 종료합니다.") + time.sleep(10) + raise SystemExit + + local_commit_sha = get_local_commit_sha() + + # 원격 커밋과 로컬 커밋 대조 + if local_commit_sha and local_commit_sha == remote_commit_sha: + print("최신 버전입니다. 업데이트할 변경 사항이 없습니다.") + print("10초 후 종료합니다.") + time.sleep(10) + raise SystemExit + + print(f"\n==========================================") + print(f"[INFO] 모드팩 설치 경로: {modpack_path}") + print(f"[INFO] 이전 커밋 SHA: {local_commit_sha or '없음 (신규 패치)'}") + print(f"[INFO] 최신 커밋 SHA: {remote_commit_sha}") + print(f"==========================================\n") + + # 1. CLEAR_TARGETS 처리 (삭제 로직: 완전 삭제 후 폴더 ZIP 해제) + for target in CLEAR_TARGETS: + sync_clear_target(target, modpack_path) + + # 2. COMPARE_TARGETS 처리 (비교 로직: 임시 해제 후 비교 삭제/추가) + for target in COMPARE_TARGETS: + sync_compare_target(target, modpack_path) + + # 패치 완료 시 로컬에 최신 커밋 SHA 기록 + update_local_commit_sha(remote_commit_sha) + print("\n[성공] 패치가 성공적으로 완료되었습니다.") + print("10초 후 종료합니다.") + time.sleep(10) \ No newline at end of file