server_pack_builder.py

#!/usr/bin/env python3
import argparse
import json
import logging
import os
import shutil
import sys
import tempfile
import threading
import zipfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, Optional, Tuple
from urllib.parse import urlparse

import requests

# specific import for TOML parsing
if sys.version_info >= (3, 11):
    import tomllib
else:
    try:
        import tomli as tomllib
    except ImportError:
        sys.exit("Error: Python 3.11+ or 'tomli' package is required for TOML parsing.")


def setup_logging(verbose: bool, handler: Optional[logging.Handler] = None):
    """Configures logging."""
    level = logging.DEBUG if verbose else logging.INFO
    root_logger = logging.getLogger()
    root_logger.setLevel(level)

    # Clear existing handlers
    if root_logger.hasHandlers():
        root_logger.handlers.clear()

    if handler:
        root_logger.addHandler(handler)
    else:
        # Default to stdout
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(logging.Formatter("%(message)s"))
        root_logger.addHandler(handler)

    # Silence noisy libraries
    logging.getLogger("urllib3").setLevel(logging.WARNING)
    logging.getLogger("requests").setLevel(logging.WARNING)


def get_default_worker_count() -> int:
    cpu_count = os.cpu_count() or 1
    return max(4, min(32, cpu_count * 4))


_thread_local = threading.local()


def get_requests_session() -> requests.Session:
    session = getattr(_thread_local, "session", None)
    if session is None:
        session = requests.Session()
        _thread_local.session = session
    return session


def is_fabric_client_only(jar_path: str) -> Tuple[Optional[bool], str]:
    """
    Checks if a jar is a Fabric mod and if it is client-side only.
    Returns: (is_client_only, reason_string)
    is_client_only: True if client-only, False if server/universal, None if not a Fabric mod.
    """
    try:
        with zipfile.ZipFile(jar_path, "r") as jar:
            if "fabric.mod.json" not in jar.namelist():
                return None, "Not a Fabric mod"

            with jar.open("fabric.mod.json") as f:
                try:
                    data = json.load(f)
                    env = data.get("environment", "*")

                    if env == "client":
                        return True, "Fabric environment set to 'client'"
                    elif env == "server":
                        return False, "Fabric environment set to 'server'"
                    else:
                        return False, f"Fabric environment set to '{env}' (Universal)"
                except json.JSONDecodeError:
                    return None, "Invalid fabric.mod.json"
    except zipfile.BadZipFile:
        return None, "Corrupt JAR file"
    except Exception as e:
        return None, f"Error reading Fabric metadata: {e}"


def is_forge_client_only(jar_path: str) -> Tuple[Optional[bool], str]:
    """
    Checks if a jar is a Forge mod and if it is client-side only.
    Returns: (is_client_only, reason_string)
    is_client_only: True if client-only, False if server/universal, None if not a Forge mod.
    """
    try:
        with zipfile.ZipFile(jar_path, "r") as jar:
            if "META-INF/mods.toml" not in jar.namelist():
                return None, "Not a Forge mod"

            with jar.open("META-INF/mods.toml") as f:
                try:
                    data = tomllib.load(f)
                    mods = data.get("mods", [])

                    if not mods:
                        return False, "Forge mod list empty, assuming universal"

                    # specific logic: if ALL mods in the jar are client-only, then the jar is client-only.
                    # In Forge, displayTest="IGNORE_ALL_VERSION" usually implies client-side only.
                    all_client = True
                    for mod in mods:
                        display_test = mod.get("displayTest", "")
                        if display_test != "IGNORE_ALL_VERSION":
                            all_client = False
                            break

                    if all_client:
                        return (
                            True,
                            "Forge displayTest set to 'IGNORE_ALL_VERSION' (Client-only)",
                        )
                    else:
                        return False, "Forge mod has server-side components"

                except (tomllib.TOMLDecodeError, Exception) as e:
                    return None, f"Error parsing mods.toml: {e}"
    except zipfile.BadZipFile:
        return None, "Corrupt JAR file"
    except Exception as e:
        return None, f"Error reading Forge metadata: {e}"


def check_jar_sidedness(file_path: str) -> Tuple[bool, str, str]:
    """
    Determines if a JAR is client-only.
    Returns: (is_client_only, mod_type, reason)
    """
    # Try Fabric first
    is_client, reason = is_fabric_client_only(file_path)

    # If not Fabric, try Forge
    mod_type = "Fabric"
    if is_client is None:
        is_client, reason = is_forge_client_only(file_path)
        mod_type = "Forge"

    # If neither (or error in both), assume Universal/Unknown
    if is_client is None:
        mod_type = "Unknown"
        is_client = False
        reason = "No known metadata found, assuming Universal"

    return is_client, mod_type, reason


from typing import Callable, Dict, Optional, Tuple

# ... existing imports ...

def process_local_modpack(
    source: str, 
    dest: str, 
    dry_run: bool, 
    progress_callback: Optional[Callable[[int, int], None]] = None
):
    """
    Iterates through mods in source, determines sidedness, and copies to dest.
    """
    if not os.path.exists(source):
        logging.error(f"Error: Source directory '{source}' does not exist.")
        raise FileNotFoundError(f"Source directory '{source}' does not exist.")

    if not dry_run:
        os.makedirs(dest, exist_ok=True)

    files = [f for f in os.listdir(source) if f.endswith(".jar")]
    total_mods = len(files)
    processed = 0
    copied = 0
    skipped = 0

    logging.info(f"Scanning {total_mods} JAR files in '{source}'...")
    
    # Initial callback
    if progress_callback:
        progress_callback(0, total_mods)

    def analyze_jar(index: int, filename: str) -> Tuple[int, str, str, bool, str, str]:
        file_path = os.path.join(source, filename)
        is_client, mod_type, reason = check_jar_sidedness(file_path)
        return index, filename, file_path, is_client, mod_type, reason

    max_workers = get_default_worker_count()
    results: Dict[int, Tuple[str, str, bool, str, str]] = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(analyze_jar, index, filename)
            for index, filename in enumerate(files)
        ]

        for future in as_completed(futures):
            index, filename, file_path, is_client, mod_type, reason = future.result()
            results[index] = (filename, file_path, is_client, mod_type, reason)
            
            # Since analyze_jar is done, we could update progress here if we wanted fine-grained
            # async progress, but the original code loops linearly for logging/copying.
            # We'll update in the main loop below for consistent order.

    for index in range(len(files)):
        filename, file_path, is_client, mod_type, reason = results[index]
        processed += 1

        if is_client:
            logging.info(f"[SKIP] {filename} ({mod_type}): {reason}")
            skipped += 1
        else:
            logging.info(f"[COPY] {filename} ({mod_type}): {reason}")
            if not dry_run:
                try:
                    shutil.copy2(file_path, os.path.join(dest, filename))
                except Exception as e:
                    logging.error(f"Failed to copy {filename}: {e}")
            copied += 1
            
        if progress_callback:
            progress_callback(processed, total_mods)

    logging.info("-" * 40)
    logging.info(f"Summary: Processed {processed}/{total_mods} files.")
    logging.info(f"Copied: {copied}")
    logging.info(f"Skipped (Client-only): {skipped}")
    if dry_run:
        logging.info("Dry run complete. No files were moved.")


def get_modrinth_project_version(url_or_slug: str) -> Tuple[str, Optional[str]]:
    """
    Parses Modrinth URL or slug to get project slug and version ID/number.
    Input: "https://modrinth.com/modpack/my-pack/version/1.0.0" or "my-pack"
    Returns: (slug, version)
    """
    if url_or_slug.startswith("http"):
        parsed = urlparse(url_or_slug)
        path_parts = parsed.path.strip("/").split("/")

        # Format: /modpack/SLUG/version/VERSION or /project/SLUG/version/VERSION
        if len(path_parts) >= 2 and (path_parts[0] in ["modpack", "project"]):
            slug = path_parts[1]
            version = (
                path_parts[3]
                if len(path_parts) >= 4 and path_parts[2] == "version"
                else None
            )
            return slug, version

    return url_or_slug, None


def search_modrinth_modpacks(query: str, limit: int = 20) -> list:
    """
    Searches for modpacks on Modrinth.
    Returns a list of project dictionaries (hits).
    """
    session = get_requests_session()
    
    # Facet for modpacks only: [["project_type:modpack"]]
    params = {
        "query": query,
        "facets": '[["project_type:modpack"]]',
        "limit": limit
    }
    
    try:
        response = session.get("https://api.modrinth.com/v2/search", params=params)
        response.raise_for_status()
        data = response.json()
        return data.get("hits", [])
    except Exception as e:
        logging.error(f"Modrinth search failed: {e}")
        return []

def get_modrinth_versions(slug: str) -> list:
    """
    Fetches available versions for a modpack project.
    Returns a list of version dictionaries.
    """
    session = get_requests_session()
    try:
        response = session.get(f"https://api.modrinth.com/v2/project/{slug}/version")
        response.raise_for_status()
        return response.json()
    except Exception as e:
        logging.error(f"Failed to fetch versions for {slug}: {e}")
        return []

def process_modrinth_pack(
    modrinth_url: str, 
    output_file: Optional[str], 
    dry_run: bool,
    pack_version_id: Optional[str] = None,
    progress_callback: Optional[Callable[[int, int, str], None]] = None, # (current, total, status_msg)
    download_callback: Optional[Callable[[int, int], None]] = None       # (current_bytes, total_bytes)
):
    """
    Downloads and filters a Modrinth modpack.
    """
    slug, version_id_from_url = get_modrinth_project_version(modrinth_url)
    
    # Use the explicitly provided version ID if available, otherwise fall back to URL parsed one
    version_id = pack_version_id if pack_version_id else version_id_from_url

    logging.info(f"Fetching: {slug} ({version_id if version_id else 'Latest'})")

    try:
        # Get Project Info
        session = get_requests_session()
        # ... (rest of function logic remains similar until download) ...
        project_resp = session.get(f"https://api.modrinth.com/v2/project/{slug}")
        if project_resp.status_code != 200:
            logging.error(
                f"Failed to fetch project info: {project_resp.status_code} {project_resp.text}"
            )
            raise RuntimeError(f"Failed to fetch project info: {project_resp.status_code}")

        project_data = project_resp.json()
        if project_data.get("project_type") != "modpack":
            logging.error(
                f"Project '{slug}' is not a modpack (type: {project_data.get('project_type')})"
            )
            raise ValueError(f"Project '{slug}' is not a modpack")

        project_title = project_data.get("title", slug)

        # Get Version Info
        if version_id:
            version_resp = session.get(f"https://api.modrinth.com/v2/version/{version_id}")
        else:
            # Get versions list and pick latest
            version_resp = session.get(f"https://api.modrinth.com/v2/project/{slug}/version")

        if version_resp.status_code != 200:
            logging.error(
                f"Failed to fetch version info: {version_resp.status_code} {version_resp.text}"
            )
            raise RuntimeError(f"Failed to fetch version info: {version_resp.status_code}")

        version_data = version_resp.json()
        if isinstance(version_data, list):
            if not version_data:
                logging.error("No versions found for this project.")
                raise ValueError("No versions found for this project.")
            target_version = version_data[0]  # Latest
        else:
            target_version = version_data

        files = target_version.get("files", [])
        mrpack_file = next(
            (f for f in files if f["filename"].endswith(".mrpack")), None
        )

        if not mrpack_file:
            logging.error("No .mrpack file found in the version.")
            raise FileNotFoundError("No .mrpack file found in the version.")

        mrpack_url = mrpack_file["url"]

        # Define output filename
        if not output_file:
            if not dry_run:
                output_file = f"{project_title}-server.mrpack"
            else:
                output_file = "dry-run-output.mrpack"

        # Temporary work area
        with tempfile.TemporaryDirectory() as temp_dir:
            pack_path = os.path.join(temp_dir, "original.mrpack")

            logging.info(f"Downloading: {mrpack_url}")
            
            # Helper to download with progress
            def download_with_progress(url, dest_path):
                with session.get(url, stream=True) as r:
                    r.raise_for_status()
                    total_length = int(r.headers.get('content-length', 0))
                    downloaded = 0
                    
                    if download_callback and total_length > 0:
                        download_callback(0, total_length)
                        
                    with open(dest_path, "wb") as f:
                        for chunk in r.iter_content(chunk_size=65536):
                            f.write(chunk)
                            downloaded += len(chunk)
                            if download_callback and total_length > 0:
                                download_callback(downloaded, total_length)

            if not dry_run:
                download_with_progress(mrpack_url, pack_path)
            else:
                # Mock file for dry run if checking structure logic (but we need real file to extract index)
                # Actually, dry run should probably still download to analyze?
                # The user requirement implied 'process' includes checking mods.
                # So we MUST download to check mods.
                # Dry run usually means "don't produce final output/change system state".
                # We will download to temp, analyze, but not write final output.
                logging.info("(Dry Run) Downloading for analysis...")
                download_with_progress(mrpack_url, pack_path)

            # Extract modrinth.index.json
            extract_dir = os.path.join(temp_dir, "extracted")
            with zipfile.ZipFile(pack_path, "r") as zip_ref:
                zip_ref.extractall(extract_dir)

            index_path = os.path.join(extract_dir, "modrinth.index.json")
            if not os.path.exists(index_path):
                logging.error("Invalid .mrpack: modrinth.index.json missing.")
                raise FileNotFoundError("Invalid .mrpack: modrinth.index.json missing.")

            with open(index_path, "r", encoding="utf-8") as f:
                index_data = json.load(f)

            files_list = index_data.get("files", [])
            new_files_list = []

            # Prepare directory for new pack
            new_pack_dir = os.path.join(temp_dir, "new_pack")
            os.makedirs(new_pack_dir, exist_ok=True)

            # Calculate total size for download progress
            total_mod_bytes = sum(f.get("fileSize", 0) for f in files_list)
            downloaded_mod_bytes = 0
            download_lock = threading.Lock()

            def mod_chunk_callback(chunk_size):
                nonlocal downloaded_mod_bytes
                with download_lock:
                    downloaded_mod_bytes += chunk_size
                    if download_callback and total_mod_bytes > 0:
                        download_callback(downloaded_mod_bytes, total_mod_bytes)

            # Copy overrides if exist
            overrides_src = os.path.join(extract_dir, "overrides")
            if os.path.exists(overrides_src):
                logging.info("Copying overrides directory...")
                if not dry_run:
                    shutil.copytree(
                        overrides_src, os.path.join(new_pack_dir, "overrides")
                    )

            logging.info(f"Processing {len(files_list)} mods...")

            processed = 0
            kept = 0
            skipped = 0
            
            # Reset progress for mod processing
            if progress_callback:
                progress_callback(0, len(files_list), "Processing mods...")

            def check_mod_entry(
                index: int, mod_entry: dict
            ) -> Tuple[int, dict, bool, str, str, str]:
                path = mod_entry.get("path", "")
                filename = os.path.basename(path)
                download_url = mod_entry["downloads"][0]

                logging.debug(f"Checking {filename}...")
                temp_handle, temp_jar_path = tempfile.mkstemp(
                    suffix=".jar", dir=temp_dir
                )
                os.close(temp_handle)

                try:
                    session = get_requests_session()
                    with session.get(download_url, stream=True) as r:
                        r.raise_for_status()
                        with open(temp_jar_path, "wb") as f:
                            for chunk in r.iter_content(chunk_size=65536):
                                f.write(chunk)
                                mod_chunk_callback(len(chunk))
                except Exception as e:
                    try:
                        os.remove(temp_jar_path)
                    except OSError:
                        pass
                    return (
                        index,
                        mod_entry,
                        False,
                        "Unknown",
                        f"Failed to download: {e}",
                        filename,
                    )

                is_client, mod_type, reason = check_jar_sidedness(temp_jar_path)

                try:
                    os.remove(temp_jar_path)
                except OSError:
                    pass

                return index, mod_entry, is_client, mod_type, reason, filename

            max_workers = get_default_worker_count()
            results: Dict[int, Tuple[dict, bool, str, str, str]] = {}
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = [
                    executor.submit(check_mod_entry, index, mod_entry)
                    for index, mod_entry in enumerate(files_list)
                ]

                for future in as_completed(futures):
                    (
                        index,
                        mod_entry,
                        is_client,
                        mod_type,
                        reason,
                        filename,
                    ) = future.result()
                    results[index] = (mod_entry, is_client, mod_type, reason, filename)

            for index in range(len(files_list)):
                processed += 1
                mod_entry, is_client, mod_type, reason, filename = results[index]

                if is_client:
                    logging.info(f"[SKIP] {filename} ({reason})")
                    skipped += 1
                else:
                    logging.info(f"[KEEP] {filename}")
                    new_files_list.append(mod_entry)
                    kept += 1
                    
                if progress_callback:
                    progress_callback(processed, len(files_list), f"Processed {processed}/{len(files_list)}")

            # Write new index
            index_data["files"] = new_files_list
            index_data["name"] = f"{index_data.get('name', project_title)} (Server)"

            if not dry_run:
                with open(
                    os.path.join(new_pack_dir, "modrinth.index.json"),
                    "w",
                    encoding="utf-8",
                ) as f:
                    json.dump(index_data, f, indent=2)

                # Zip up the new pack
                logging.info(f"Creating server pack: {output_file}")
                with zipfile.ZipFile(output_file, "w", zipfile.ZIP_DEFLATED) as zipf:
                    for root, dirs, files in os.walk(new_pack_dir):
                        for file in files:
                            file_path = os.path.join(root, file)
                            arcname = os.path.relpath(file_path, new_pack_dir)
                            zipf.write(file_path, arcname)
            else:
                logging.info("(Dry Run) Would create server pack with:")
                logging.info("  - Overrides included")
                logging.info(f"  - {kept} mods kept")
                logging.info(f"  - {skipped} mods removed")

            logging.info("-" * 40)
            logging.info(f"Summary: Processed {processed} mods.")
            logging.info(f"Kept: {kept}")
            logging.info(f"Skipped (Client-only): {skipped}")

    except Exception as e:
        logging.error(f"An unexpected error occurred: {e}")
        if logging.getLogger().isEnabledFor(logging.DEBUG):
            import traceback

            traceback.print_exc()
        raise

def main():
    parser = argparse.ArgumentParser(
        description="Minecraft Server Pack Builder: filters client-side only mods from local folders or Modrinth packs."
    )

    group = parser.add_mutually_exclusive_group(required=False)
    group.add_argument(
        "--source", "-s", help="Path to the source 'mods' directory (Local mode)"
    )
    group.add_argument(
        "--modrinth-url", "-m", help="Modrinth Modpack URL or Slug (Modrinth mode)"
    )

    parser.add_argument(
        "--destination",
        "-d",
        help="Path to the destination 'mods' directory (Local mode)",
    )
    parser.add_argument(
        "--output-file",
        "-o",
        help="Output path for the generated .mrpack (Modrinth mode)",
    )
    parser.add_argument(
        "--pack-version",
        help="Specific version ID/Number to download (Modrinth mode). Overrides URL version.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Simulate the process without copying/writing files",
    )
    parser.add_argument(
        "--verbose", "-v", action="store_true", help="Enable verbose logging"
    )
    parser.add_argument(
        "--gui", action="store_true", help="Launch the Graphical User Interface"
    )

    args = parser.parse_args()
    setup_logging(args.verbose)

    if args.gui:
        try:
            from gui import run_gui
            run_gui()
        except ImportError as e:
            logging.error(f"Failed to load GUI: {e}")
            logging.error("Ensure PyQt6 is installed: pip install PyQt6")
            sys.exit(1)
        return

    # Validate arguments for CLI mode since group is no longer required=True
    if not args.source and not args.modrinth_url:
        parser.error("one of the arguments --source/-s --modrinth-url/-m is required")

    try:
        if args.source:
            if not args.destination:
                parser.error("--destination is required when using --source")
            process_local_modpack(args.source, args.destination, args.dry_run)
        elif args.modrinth_url:
            process_modrinth_pack(
                args.modrinth_url, 
                args.output_file, 
                args.dry_run,
                pack_version_id=args.pack_version
            )
    except Exception as e:
        sys.exit(1)


if __name__ == "__main__":
    main()