패치 로직 코드 추가

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)