249 lines
9.6 KiB
Python
249 lines
9.6 KiB
Python
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) |