#!/usr/bin/env python3 """ Diagnostic: compare mod folder names on disk against the server's canonical names, and optionally rename mismatched folders + fix broken junctions. Matching strategy (most-reliable first): 1. Read meta.cpp from the local folder -> steam_id -> server by_steam_id 2. Normalize folder name (@CBA_A3 -> cbaa3) -> server by_name Status codes: OK disk name matches server name exactly MISMATCH disk name differs from server canonical name NOT_ON_SERVER server has no folder for this mod (new mod? typo?) ID_COLLISION local meta.cpp has wrong steam_id (belongs to a different mod) Usage: python check_names.py # report only python check_names.py --fix # rename mismatched folders + fix junctions python check_names.py --fix-ids # fix wrong steam_ids in local meta.cpp files python check_names.py --fix --fix-ids # both python check_names.py --group shared --fix-ids """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path from arma_modlist_tools.compat import fix_console_encoding from arma_modlist_tools.config import load_config from arma_modlist_tools.fetcher import ( _normalize_name, _parse_meta_cpp, build_server_index, ) from arma_modlist_tools.linker import _is_junction, create_junction, remove_junction fix_console_encoding() # Column widths W_DISK = 44 W_SERVER = 50 W_GROUP = 24 _PUBLISHEDID_RE = re.compile(r"(publishedid\s*=\s*)\d+(\s*;)", re.IGNORECASE) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _server_name_from_url(url: str) -> str: return url.rstrip("/").split("/")[-1] def _read_local_steam_id(mod_dir: Path) -> str | None: for candidate in (mod_dir / "meta.cpp", mod_dir / "Meta.cpp"): if candidate.exists(): try: return _parse_meta_cpp(candidate.read_text(encoding="utf-8", errors="ignore")) except OSError: pass return None def _lookup_detailed( local_name: str, mod_dir: Path, index: dict, ) -> tuple[str | None, str | None]: """ Return (server_name, local_steam_id_used). - server_name: canonical server folder name, or None if not found - local_steam_id_used: the steam_id that was read from local meta.cpp (present regardless of whether the lookup succeeded, so callers can report what the wrong ID is) """ local_sid = _read_local_steam_id(mod_dir) if local_sid: url = index["by_steam_id"].get(local_sid) if url: return _server_name_from_url(url), local_sid # Name-based fallback url = index["by_name"].get(_normalize_name(local_name)) if url: return _server_name_from_url(url), local_sid return None, local_sid def _collect_mod_folders(cfg, group_filter: str | None) -> list[tuple[str, Path]]: result = [] if not cfg.downloads.is_dir(): return result for group_dir in sorted(cfg.downloads.iterdir()): if not group_dir.is_dir(): continue if group_filter and group_dir.name != group_filter: continue for mod_dir in sorted(group_dir.iterdir()): if mod_dir.is_dir() and mod_dir.name.startswith("@"): result.append((group_dir.name, mod_dir)) return result # --------------------------------------------------------------------------- # Classification # --------------------------------------------------------------------------- def _resolve_status( disk_name: str, server_name: str | None, local_sid: str | None, ok_disk_names: set[str], ) -> tuple[str, str]: """ Return (status, display_col) after false-positive filtering. ID_COLLISION: steam_id lookup returned a name that's already correctly held by a different folder — the local meta.cpp has the wrong publishedid. """ if server_name is None: return "NOT_ON_SERVER", "---" if server_name == disk_name: return "OK", server_name if server_name in ok_disk_names: # This folder's local steam_id belongs to a mod that already has a # correct folder on disk — the meta.cpp is wrong. sid_info = f" (local id: {local_sid})" if local_sid else "" return "ID_COLLISION", f"{server_name}{sid_info}" return "MISMATCH", server_name # --------------------------------------------------------------------------- # Fix: rename folder + update junction # --------------------------------------------------------------------------- def _fix_mismatch( group: str, mod_dir: Path, server_name: str, cfg, ) -> tuple[bool, str]: new_dir = mod_dir.parent / server_name if new_dir.exists(): return False, f"target already exists: {new_dir}" old_link = cfg.arma_dir / mod_dir.name removed_old_link = False if _is_junction(old_link): ok, err = remove_junction(old_link) if not ok: return False, f"could not remove old junction {old_link}: {err}" removed_old_link = True try: mod_dir.rename(new_dir) except OSError as exc: if removed_old_link: create_junction(old_link, mod_dir) return False, f"rename failed: {exc}" new_link = cfg.arma_dir / server_name if cfg.arma_dir.exists(): if not (_is_junction(new_link) or new_link.exists()): create_junction(new_link, new_dir) return True, f"{mod_dir.name} -> {server_name}" # --------------------------------------------------------------------------- # Fix: rewrite wrong steam_id in local meta.cpp # --------------------------------------------------------------------------- def _build_comparison_id_map(cfg) -> dict[str, tuple[str, str]]: """ Load comparison.json and return {normalized_name: (steam_id, display_name)} for every mod that has a steam_id. """ if not cfg.comparison.exists(): return {} try: data = json.loads(cfg.comparison.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} result: dict[str, tuple[str, str]] = {} all_mods = list(data.get("shared", {}).get("mods", [])) for preset_data in data.get("unique", {}).values(): all_mods.extend(preset_data.get("mods", [])) for mod in all_mods: sid = mod.get("steam_id") name = mod.get("name", "") if sid and name: result[_normalize_name(name)] = (sid, name) return result def _write_steam_id(mod_dir: Path, new_sid: str) -> tuple[bool, str]: """ Overwrite / insert publishedid in mod_dir/meta.cpp. Returns (success, message). """ meta_path = mod_dir / "meta.cpp" if meta_path.exists(): try: text = meta_path.read_text(encoding="utf-8", errors="ignore") except OSError as exc: return False, str(exc) old_sid_match = _parse_meta_cpp(text) if _PUBLISHEDID_RE.search(text): new_text = _PUBLISHEDID_RE.sub(rf"\g<1>{new_sid}\g<2>", text) else: new_text = text.rstrip() + f"\npublishedid = {new_sid};\n" try: meta_path.write_text(new_text, encoding="utf-8") except OSError as exc: return False, str(exc) old_str = old_sid_match or "missing" return True, f"meta.cpp: {old_str} -> {new_sid}" else: try: meta_path.write_text(f"publishedid = {new_sid};\n", encoding="utf-8") except OSError as exc: return False, str(exc) return True, f"meta.cpp created with id {new_sid}" # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: cfg = load_config() parser = argparse.ArgumentParser( description="Check and fix mod folder name mismatches and wrong steam IDs." ) parser.add_argument("--group", "-g", metavar="GROUP", help="Only check this group (default: all)") parser.add_argument("--fix", action="store_true", help="Rename MISMATCH folders and fix junctions") parser.add_argument("--fix-ids", action="store_true", help="Rewrite wrong publishedid in meta.cpp using comparison.json") args = parser.parse_args() # ---- Build server index ---- print(f"\nBuilding server index...") index = build_server_index(cfg.server_url, cfg.server_auth) print(f" {len(index['by_steam_id'])} mods indexed by steam_id, " f"{len(index['by_name'])} by name\n") # ---- Collect local mod folders ---- mods = _collect_mod_folders(cfg, args.group) if not mods: print("No @Mod folders found in downloads/.\n") sys.exit(0) # ---- Pass 1: raw lookup ---- # (group, mod_dir, server_name | None, local_steam_id | None) raw: list[tuple[str, Path, str | None, str | None]] = [] for group, mod_dir in mods: server_name, local_sid = _lookup_detailed(mod_dir.name, mod_dir, index) raw.append((group, mod_dir, server_name, local_sid)) # ---- Pass 2: build ok_disk_names for collision detection ---- ok_disk_names: set[str] = { mod_dir.name for _, mod_dir, server_name, _ in raw if server_name == mod_dir.name } # ---- Print table ---- ok_count = mismatch_count = collision_count = unknown_count = 0 mismatches: list[tuple[str, Path, str]] = [] collisions: list[tuple[str, Path, str | None]] = [] # (group, mod_dir, local_sid) print(f" {'Disk name':<{W_DISK}} {'Group':<{W_GROUP}} {'Status / Server name':<{W_SERVER}}") print(f" {'-'*W_DISK} {'-'*W_GROUP} {'-'*W_SERVER}") for group, mod_dir, server_name, local_sid in raw: status, server_col = _resolve_status(mod_dir.name, server_name, local_sid, ok_disk_names) if status == "OK": ok_count += 1 detail = f"OK ({server_col})" elif status == "MISMATCH": mismatch_count += 1 mismatches.append((group, mod_dir, server_name)) detail = f"MISMATCH -> {server_col}" elif status == "ID_COLLISION": collision_count += 1 collisions.append((group, mod_dir, local_sid)) detail = f"ID_COLLISION {server_col}" else: unknown_count += 1 detail = "NOT_ON_SERVER" print(f" {mod_dir.name:<{W_DISK}} {group:<{W_GROUP}} {detail}") print(f"\n {ok_count} OK, {mismatch_count} mismatch, " f"{collision_count} id_collision, {unknown_count} not on server\n") any_action = args.fix or args.fix_ids nothing_to_do = (not mismatches) and (not collisions) if nothing_to_do: if mismatch_count == 0 and collision_count == 0: print(" All folder names match. Safe to run full pipeline.\n") sys.exit(0) if not any_action: hints = [] if mismatches: hints.append("--fix to rename mismatched folders") if collisions: hints.append("--fix-ids to correct wrong steam IDs in meta.cpp") print(" Run with " + " and ".join(hints) + ".\n") sys.exit(0) # ---- --fix: rename mismatched folders ---- if args.fix and mismatches: print(f"Fixing {len(mismatches)} name mismatch(es)...\n") fixed = failed = 0 for group, mod_dir, server_name in mismatches: ok, msg = _fix_mismatch(group, mod_dir, server_name, cfg) if ok: print(f" [+] {msg}") fixed += 1 else: print(f" [X] {mod_dir.name} FAILED: {msg}") failed += 1 print(f"\n Done: {fixed} renamed" + (f", {failed} failed" if failed else "") + "\n") # ---- --fix-ids: rewrite wrong steam_ids ---- if args.fix_ids and collisions: id_map = _build_comparison_id_map(cfg) if not id_map: print(" WARNING: comparison.json not found or empty. " "Run 'python run.py --skip-fetch --skip-link' first.\n") sys.exit(1) print(f"Fixing {len(collisions)} steam_id collision(s)...\n") fixed = failed = unknown = 0 for group, mod_dir, bad_sid in collisions: norm = _normalize_name(mod_dir.name) match = id_map.get(norm) if match is None: print(f" [?] {mod_dir.name:<{W_DISK}} not in comparison.json — skipped") unknown += 1 continue correct_sid, display_name = match if bad_sid == correct_sid: print(f" [=] {mod_dir.name:<{W_DISK}} id already correct ({correct_sid})") fixed += 1 continue ok, msg = _write_steam_id(mod_dir, correct_sid) if ok: print(f" [+] {mod_dir.name:<{W_DISK}} {msg}") fixed += 1 else: print(f" [X] {mod_dir.name:<{W_DISK}} FAILED: {msg}") failed += 1 print(f"\n Done: {fixed} fixed" + (f", {unknown} skipped (not in comparison.json)" if unknown else "") + (f", {failed} failed" if failed else "") + "\n") print(" Re-run check_names.py to verify.\n") if __name__ == "__main__": main()