main.py

import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import shutil
import sys
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET
import zipfile
from pathlib import Path


LOADER_CONFIG = {
	"forge": {
		"metadata_url": "https://maven.minecraftforge.net/net/minecraftforge/forge/maven-metadata.xml",
		"base_url": "https://maven.minecraftforge.net/net/minecraftforge/forge",
		"artifact": "forge",
		"output_root": "forge",
	},
	"neoforge": {
		"metadata_url": "https://maven.neoforged.net/api/maven/versions/releases/net%2Fneoforged%2Fneoforge",
		"metadata_format": "json",
		"base_url": "https://maven.neoforged.net/releases/net/neoforged/neoforge",
		"artifact": "neoforge",
		"output_root": "neoforge",
	},
}

KEEP_ITEMS = {"version.json", "install_profile.json", "maven", "data"}
MAX_WORKERS = 16


def fetch_versions(metadata_url: str, metadata_format: str = "xml") -> list[str]:
	req = urllib.request.Request(
		metadata_url,
		headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
	)
	with urllib.request.urlopen(req) as response:
		data = response.read()

	if metadata_format == "json":
		payload = json.loads(data.decode("utf-8"))
		versions = payload.get("versions", [])
		return [version for version in versions if isinstance(version, str)]

	root = ET.fromstring(data)
	versions = [node.text for node in root.findall("./versioning/versions/version") if node.text]
	return versions


def normalize_requested_version(loader: str, version: str) -> str:
	resolved = version.strip()
	if loader != "neoforge":
		return resolved

	parts = resolved.split(".")
	if len(parts) >= 2 and parts[0] == "1" and all(part.isdigit() for part in parts[1:]):
		major = str(int(parts[1]))
		minor = str(int(parts[2])) if len(parts) >= 3 else "0"
		return f"{major}.{minor}"

	return resolved


def filter_versions(versions: list[str], version_prefix: str, loader: str) -> list[str]:
	if loader == "forge":
		prefix = f"{version_prefix}-"
		return [version for version in versions if version.startswith(prefix)]

	return [
		version
		for version in versions
		if version == version_prefix or version.startswith(f"{version_prefix}.") or version.startswith(f"{version_prefix}-")
	]


def download_installer(base_url: str, artifact: str, version: str, cache_dir: Path) -> Path | None:
	filename = f"{artifact}-{version}-installer.jar"
	url = f"{base_url}/{version}/{filename}"
	destination = cache_dir / filename

	if destination.exists():
		print(f"[cache] {filename}")
		return destination

	try:
		print(f"[download] {url}")
		req = urllib.request.Request(
			url,
			headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"},
		)
		with urllib.request.urlopen(req) as response, destination.open("wb") as output:
			shutil.copyfileobj(response, output)
		return destination
	except urllib.error.HTTPError as exc:
		print(f"[skip] {filename} HTTP {exc.code}")
	except urllib.error.URLError as exc:
		print(f"[skip] {filename} URL error: {exc.reason}")

	return None


def extract_installer(installer_jar: Path, target_dir: Path) -> None:
	if target_dir.exists():
		shutil.rmtree(target_dir)
	target_dir.mkdir(parents=True, exist_ok=True)

	with zipfile.ZipFile(installer_jar, "r") as zip_ref:
		zip_ref.extractall(target_dir)


def cleanup_target(target_dir: Path) -> None:
	for item in target_dir.iterdir():
		if item.name in KEEP_ITEMS:
			continue

		if item.is_dir():
			shutil.rmtree(item)
		else:
			item.unlink()


def write_metadata_index(loader: str, versions_newest_first: list[str], output_root: Path) -> None:
	metadata_dir = Path("metadata")
	metadata_dir.mkdir(parents=True, exist_ok=True)
	metadata_file = metadata_dir / f"{loader}.json"

	# 扫描已存在的版本目录
	disk_versions = []
	if output_root.exists():
		disk_versions = [item.name for item in output_root.iterdir() if item.is_dir()]

	# 合并：本次匹配版本 + 磁盘已有版本，去重并保持最新在前
	seen = set()
	merged = []
	for version in versions_newest_first + disk_versions:
		if version not in seen:
			seen.add(version)
			merged.append(version)

	payload = {
		"latest": merged[0] if merged else "",
		"versions": merged,
	}
	metadata_file.write_text(json.dumps(payload, ensure_ascii=False, indent=4), encoding="utf-8")


def process_single_version(config: dict[str, str], cache_dir: Path, output_root: Path, version: str) -> str:
	target_dir = output_root / version
	if target_dir.exists():
		print(f"[skip] 已存在目录: {target_dir}")
		return "skipped_existing"

	installer = download_installer(config["base_url"], config["artifact"], version, cache_dir)
	if installer is None:
		return "failed"

	print(f"[extract] {installer.name} -> {target_dir}")
	try:
		extract_installer(installer, target_dir)
		cleanup_target(target_dir)
	except zipfile.BadZipFile:
		print(f"[skip] 无法解压: {installer}")
		return "failed"
	except Exception as exc:
		print(f"[skip] 处理失败 {version}: {exc}")
		return "failed"

	return "processed"


def process(loader: str, mc_version: str) -> int:
	if loader not in LOADER_CONFIG:
		print(f"不支持的loader: {loader}")
		return 1

	config = LOADER_CONFIG[loader]
	cache_dir = Path("cache")
	cache_dir.mkdir(parents=True, exist_ok=True)

	try:
		versions = fetch_versions(config["metadata_url"], config.get("metadata_format", "xml"))
	except urllib.error.URLError as exc:
		print(f"拉取元数据失败: {exc}")
		return 1
	except json.JSONDecodeError as exc:
		print(f"解析元数据JSON失败: {exc}")
		return 1
	except ET.ParseError as exc:
		print(f"解析元数据失败: {exc}")
		return 1

	resolved_version = normalize_requested_version(loader, mc_version)
	if resolved_version != mc_version:
		print(f"版本映射: {mc_version} -> {resolved_version}")

	matched_versions = versions if resolved_version.lower() == "all" else filter_versions(versions, resolved_version, loader)
	if not matched_versions:
		print(f"未找到 {loader} {resolved_version} 对应的版本")
		return 1

	output_root = Path(config["output_root"])
	output_root.mkdir(parents=True, exist_ok=True)

	versions_newest_first = list(reversed(matched_versions))
	write_metadata_index(loader, versions_newest_first, output_root)

	print(f"匹配到 {len(matched_versions)} 个版本")

	processed = 0
	skipped_existing = 0
	failed = 0

	with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
		futures = [
			executor.submit(process_single_version, config, cache_dir, output_root, version)
			for version in matched_versions
		]
		for future in as_completed(futures):
			result = future.result()
			if result == "processed":
				processed += 1
			elif result == "skipped_existing":
				skipped_existing += 1
			else:
				failed += 1

	print(
		f"完成: 新增处理 {processed} 个，已存在跳过 {skipped_existing} 个，失败 {failed} 个，总匹配 {len(matched_versions)} 个"
	)
	return 0


def parse_args() -> argparse.Namespace:
	parser = argparse.ArgumentParser(description="批量下载并处理 Forge/NeoForge installer")
	parser.add_argument("loader", choices=["forge", "neoforge"], help="loader类型: forge 或 neoforge")
	parser.add_argument("version", help="Minecraft主版本前缀，例如: 1.21；或 all 表示全部版本")
	return parser.parse_args()


def main() -> int:
	args = parse_args()
	return process(args.loader, args.version)


if __name__ == "__main__":
	sys.exit(main())