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 )