launcher/server/pack_manager.py

import hashlib
import os
import zipfile
from datetime import datetime
from pathlib import Path
import json
from typing import Optional, Dict, List
import structlog
import asyncio
import aiofiles

from models import PackMeta, FileEntry

logger = structlog.get_logger(__name__)

PACKS_DIR = Path("packs")
DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)

# Cache for loaded manifests
_manifest_cache: Dict[str, PackMeta] = {}

# Pack config cache
_packs_config: Optional[Dict] = None

def load_packs_config() -> dict:
    global _packs_config
    if _packs_config is not None:
        return _packs_config
    config_path = DATA_DIR / "packs_config.json"
    if config_path.exists():
        try:
            _packs_config = json.loads(config_path.read_text())
        except Exception:
            _packs_config = {}
    else:
        _packs_config = {}
    return _packs_config

def save_packs_config(config: dict):
    global _packs_config
    _packs_config = config
    config_path = DATA_DIR / "packs_config.json"
    config_path.write_text(json.dumps(config, indent=2))

def is_pack_disabled(pack_name: str) -> bool:
    config = load_packs_config()
    return config.get(pack_name, {}).get("disabled", False)

def toggle_pack_disabled(pack_name: str) -> bool:
    config = load_packs_config()
    entry = config.setdefault(pack_name, {})
    entry["disabled"] = not entry.get("disabled", False)
    save_packs_config(config)
    return entry["disabled"]

def get_pack_presets(pack_name: str) -> List[str]:
    presets_dir = PACKS_DIR / pack_name / "presets"
    if not presets_dir.exists() or not presets_dir.is_dir():
        return []
    return sorted([d.name for d in presets_dir.iterdir() if d.is_dir()])

def get_cached_manifest(pack_name: str) -> Optional[PackMeta]:
    """Get manifest from memory cache if available"""
    return _manifest_cache.get(pack_name)

def get_meta_path(pack_name: str) -> Path:
    return DATA_DIR / f"{pack_name}.meta"

def calculate_sha256_sync(file_path: Path) -> str:
    """Calculate SHA256 hash of a file (synchronous version)"""
    hash_sha = hashlib.sha256()
    with open(file_path, 'rb') as f:
        while chunk := f.read(8192):
            hash_sha.update(chunk)
    return hash_sha.hexdigest()

async def calculate_sha256(file_path: Path) -> str:
    """Calculate SHA256 hash of a file (async)"""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, calculate_sha256_sync, file_path)

async def scan_pack(pack_name: str, force_rescan: bool = False) -> PackMeta:
    """Scan pack directory and update manifest if needed"""
    pack_path = PACKS_DIR / pack_name
    
    if not pack_path.exists() or not pack_path.is_dir():
        raise FileNotFoundError(f"Pack {pack_name} not found")

    meta_path = get_meta_path(pack_name)
    current_meta: Optional[PackMeta] = None

    # Check cache first
    if not force_rescan and pack_name in _manifest_cache:
        return _manifest_cache[pack_name]

    # Load existing meta if available
    if meta_path.exists():
        try:
            async with aiofiles.open(meta_path, 'r', encoding='utf-8') as f:
                data = json.loads(await f.read())
                current_meta = PackMeta.model_validate(data)
        except Exception as e:
            logger.warning(f"Failed to load existing meta for pack {pack_name}: {e}")
            current_meta = None

    new_files: Dict[str, FileEntry] = {}
    changed = False

    # Get ignored directories (always use current list, not cached from meta)
    ignored_dirs = [
        "resourcepacks", "shaderpacks", "saves", "logs",
        "crash-reports", "screenshots", "journeymap",
        "presets"
    ]

    # Walk through pack directory
    for root, dirs, files in os.walk(pack_path):
        # Filter ignored directories
        dirs[:] = [d for d in dirs if d not in ignored_dirs]

        for file in files:
            file_path = Path(root) / file
            rel_path = file_path.relative_to(pack_path).as_posix()

            # Skip metadata files that should not be part of the pack
            if file in ("description.txt", "instance.json"):
                continue

            # Skip if in ignored directory
            if any(ignored_dir in rel_path.split('/') for ignored_dir in ignored_dirs):
                continue

            # Validate .jar files — skip corrupted ones
            if file.endswith(".jar") and not _is_valid_zip(file_path):
                logger.warning(f"Skipping corrupt jar in pack {pack_name}: {rel_path}")
                continue

            stat = file_path.stat()
            file_hash = await calculate_sha256(file_path)

            entry = FileEntry(
                path=rel_path,
                hash=file_hash,
                size=stat.st_size,
                added_at=datetime.utcfromtimestamp(stat.st_ctime),
                modified_at=datetime.utcfromtimestamp(stat.st_mtime)
            )

            new_files[rel_path] = entry

            # Check if file changed
            if current_meta and (rel_path not in current_meta.files or 
                                 current_meta.files[rel_path].hash != file_hash):
                changed = True

    # Check if we need to update
    if not current_meta or changed or len(new_files) != len(current_meta.files if current_meta else 0):
        version = (current_meta.version + 1) if current_meta else 1
        
        # Load instance.json for pack metadata
        minecraft_version = "1.20.4"
        loader_type = "vanilla"
        loader_version = None
        asset_index = None
        
        pack_config_path = pack_path / "instance.json"
        if pack_config_path.exists():
            try:
                async with aiofiles.open(pack_config_path, 'r', encoding='utf-8') as f:
                    config = json.loads(await f.read())
                    minecraft_version = config.get("minecraftVersion", minecraft_version)
                    loader_type = config.get("loaderType", loader_type)
                    loader_version = config.get("loaderVersion")
                    asset_index = config.get("assetIndex")
            except Exception as e:
                logger.warning(f"Failed to load instance.json for {pack_name}: {e}")
        
        # Create new manifest
        new_meta = PackMeta(
            pack_name=pack_name,
            version=version,
            files=new_files,
            updated_at=datetime.utcnow(),
            ignored_dirs=ignored_dirs,
            minecraft_version=minecraft_version,
            loader_type=loader_type,
            loader_version=loader_version,
            asset_index=asset_index
        )
        
        async with aiofiles.open(meta_path, 'w', encoding='utf-8') as f:
            await f.write(new_meta.model_dump_json(indent=2))
        
        # Update cache
        _manifest_cache[pack_name] = new_meta
        
        logger.info(f"Pack updated: {pack_name} v{version}, {len(new_files)} files")
        return new_meta
    
    # No changes, use existing
    if current_meta:
        _manifest_cache[pack_name] = current_meta
        return current_meta
    
    # Should not happen
    raise Exception(f"Failed to scan pack {pack_name}")

def _is_valid_zip(path: Path) -> bool:
    """Check if a file is a valid ZIP archive"""
    try:
        with zipfile.ZipFile(path, 'r') as zf:
            return zf.testzip() is None
    except zipfile.BadZipFile:
        return False
    except Exception:
        return False


def get_merged_manifest(pack_name: str, preset: str) -> PackMeta:
    """Get manifest merged with preset files. Preset files override base files."""
    base_meta = get_cached_manifest(pack_name)
    if not base_meta:
        raise FileNotFoundError(f"Pack {pack_name} not found")

    preset_dir = PACKS_DIR / pack_name / "presets" / preset
    if not preset_dir.exists() or not preset_dir.is_dir():
        raise FileNotFoundError(f"Preset {preset} not found for pack {pack_name}")

    merged_files = dict(base_meta.files)

    for root, dirs, files in os.walk(preset_dir):
        for file in files:
            file_path = Path(root) / file
            rel_path = file_path.relative_to(preset_dir).as_posix()
            stat = file_path.stat()
            file_hash = calculate_sha256_sync(file_path)
            merged_files[rel_path] = FileEntry(
                path=rel_path,
                hash=file_hash,
                size=stat.st_size,
                added_at=datetime.utcfromtimestamp(stat.st_ctime),
                modified_at=datetime.utcfromtimestamp(stat.st_mtime)
            )

    return PackMeta(
        pack_name=base_meta.pack_name,
        version=base_meta.version,
        files=merged_files,
        updated_at=base_meta.updated_at,
        ignored_dirs=base_meta.ignored_dirs,
        minecraft_version=base_meta.minecraft_version,
        loader_type=base_meta.loader_type,
        loader_version=base_meta.loader_version,
        asset_index=base_meta.asset_index
    )