48637ffe90
Before step_fetch, scan all downloads/ subdirs and move any mod that comparison.json now assigns to a different group. Matching uses steam_id (via meta.cpp publishedid) first, normalized name as fallback. Stale junctions in arma_dir are removed before the folder move so step_link can re-create them pointing to the new location. - New arma_modlist_tools/migrator.py: migrate_mod_groups() - run.py: step_migrate(), --skip-migrate flag, wired into dispatch loop - gui/app.py: step_migrate inserted as Step 3/5 between compare and fetch - gui/locales.py: add step3/4/5 names (en + vi), renumber old 3->4, 4->5 - test_suite.py: 7 new migrator tests (158 total, 0 failed)
187 lines
6.8 KiB
Python
187 lines
6.8 KiB
Python
"""
|
|
arma_modlist_tools.migrator
|
|
~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
Move locally-downloaded mod folders to match the group assignments in
|
|
comparison.json, avoiding redundant re-downloads when presets are renamed
|
|
or reorganised.
|
|
|
|
A mod "needs migration" when it exists on disk under
|
|
``downloads/{old_group}/@FolderName`` but comparison.json now assigns it
|
|
to ``downloads/{new_group}/@FolderName``.
|
|
|
|
Algorithm
|
|
---------
|
|
1. Build a *local index*: scan every ``downloads/{group}/@ModDir`` and read
|
|
its ``meta.cpp`` to find its steam_id. Also record its normalised folder
|
|
name. Result: ``{steam_id: (group, path)}`` and
|
|
``{norm_name: (group, path)}``.
|
|
|
|
2. Build a *target list* from comparison.json: for every mod in every group,
|
|
record which group comparison now assigns it to.
|
|
|
|
3. For each target entry:
|
|
- Find the mod in the local index (steam_id first, normalised name fallback).
|
|
- If not found on disk: skip (step_fetch will download it).
|
|
- If already in the correct group: skip.
|
|
- If the destination already exists: skip (step_fetch handles file-level sync).
|
|
- Otherwise: remove any stale junction first, then move old → new.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from .fetcher import _normalize_name, _parse_meta_cpp
|
|
from .linker import _is_junction, remove_junction
|
|
|
|
|
|
def _build_local_index(downloads: Path) -> dict:
|
|
"""Scan ``downloads/`` and return a two-key index of existing mod folders.
|
|
|
|
:returns: ``{"by_steam_id": {sid: (group, path)},
|
|
"by_norm_name": {norm: (group, path)}}``
|
|
|
|
First match wins for both keys (sorted directory order is deterministic).
|
|
"""
|
|
by_steam_id: dict[str, tuple[str, Path]] = {}
|
|
by_norm_name: dict[str, tuple[str, Path]] = {}
|
|
|
|
if not downloads.is_dir():
|
|
return {"by_steam_id": by_steam_id, "by_norm_name": by_norm_name}
|
|
|
|
for group_dir in sorted(downloads.iterdir()):
|
|
if not group_dir.is_dir():
|
|
continue
|
|
group_name = group_dir.name
|
|
for mod_dir in sorted(group_dir.iterdir()):
|
|
if not mod_dir.is_dir() or not mod_dir.name.startswith("@"):
|
|
continue
|
|
|
|
norm = _normalize_name(mod_dir.name)
|
|
if norm not in by_norm_name:
|
|
by_norm_name[norm] = (group_name, mod_dir)
|
|
|
|
meta = mod_dir / "meta.cpp"
|
|
if meta.exists():
|
|
try:
|
|
sid = _parse_meta_cpp(
|
|
meta.read_text(encoding="utf-8", errors="replace")
|
|
)
|
|
if sid and sid not in by_steam_id:
|
|
by_steam_id[sid] = (group_name, mod_dir)
|
|
except OSError:
|
|
pass
|
|
|
|
return {"by_steam_id": by_steam_id, "by_norm_name": by_norm_name}
|
|
|
|
|
|
def _build_target_list(comparison: dict) -> list[tuple[str, str | None, str]]:
|
|
"""Flatten comparison.json into ``[(new_group, steam_id_or_None, mod_name)]``.
|
|
|
|
Shared mods → group ``"shared"``; unique mods → group = preset name.
|
|
"""
|
|
entries: list[tuple[str, str | None, str]] = []
|
|
for mod in comparison.get("shared", {}).get("mods", []):
|
|
entries.append(("shared", mod.get("steam_id") or None, mod.get("name", "")))
|
|
for preset, pdata in comparison.get("unique", {}).items():
|
|
for mod in pdata.get("mods", []):
|
|
entries.append((preset, mod.get("steam_id") or None, mod.get("name", "")))
|
|
return entries
|
|
|
|
|
|
def migrate_mod_groups(
|
|
downloads: Path,
|
|
arma_dir: Path | None,
|
|
comparison: dict,
|
|
) -> dict:
|
|
"""Move locally-downloaded mod folders to match *comparison* group assignments.
|
|
|
|
:param downloads: Path to the ``downloads/`` directory.
|
|
:param arma_dir: Path to the Arma 3 server directory used for junction
|
|
cleanup. Pass ``None`` to skip junction removal.
|
|
:param comparison: Parsed ``comparison.json`` dict.
|
|
:returns: Result dict::
|
|
|
|
{
|
|
"moved": int,
|
|
"junction_removed": int,
|
|
"skipped_correct": int,
|
|
"skipped_dest_exists": int,
|
|
"skipped_not_found": int,
|
|
"errors": {mod_name: error_message},
|
|
}
|
|
"""
|
|
result: dict = {
|
|
"moved": 0,
|
|
"junction_removed": 0,
|
|
"skipped_correct": 0,
|
|
"skipped_dest_exists": 0,
|
|
"skipped_not_found": 0,
|
|
"errors": {},
|
|
}
|
|
|
|
local = _build_local_index(downloads)
|
|
|
|
for new_group, steam_id, mod_name in _build_target_list(comparison):
|
|
# 1. Locate on disk — steam_id first, normalised name fallback
|
|
entry: tuple[str, Path] | None = None
|
|
if steam_id:
|
|
entry = local["by_steam_id"].get(steam_id)
|
|
if entry is None:
|
|
entry = local["by_norm_name"].get(_normalize_name(mod_name))
|
|
|
|
if entry is None:
|
|
result["skipped_not_found"] += 1
|
|
continue
|
|
|
|
old_group, old_path = entry
|
|
|
|
# 2. Already in the correct group?
|
|
if old_group == new_group:
|
|
result["skipped_correct"] += 1
|
|
continue
|
|
|
|
# 3. Destination already present — let step_fetch handle file-level sync
|
|
new_path = downloads / new_group / old_path.name
|
|
if new_path.exists():
|
|
result["skipped_dest_exists"] += 1
|
|
continue
|
|
|
|
# 4. Remove stale junction so step_link can re-create it at the new path
|
|
if arma_dir is not None and arma_dir.is_dir():
|
|
link = arma_dir / old_path.name
|
|
if _is_junction(link):
|
|
ok, err = remove_junction(link)
|
|
if ok:
|
|
result["junction_removed"] += 1
|
|
else:
|
|
print(f" MIGRATE WARNING: cannot remove junction "
|
|
f"{link.name}: {err}")
|
|
|
|
# 5. Move folder (shutil.move handles cross-device gracefully)
|
|
try:
|
|
new_path.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(old_path), str(new_path))
|
|
print(f" MIGRATE moved: {old_path.name} {old_group} -> {new_group}")
|
|
result["moved"] += 1
|
|
|
|
# Update index so later targets don't re-match the now-moved path
|
|
if steam_id:
|
|
local["by_steam_id"][steam_id] = (new_group, new_path)
|
|
local["by_norm_name"][_normalize_name(old_path.name)] = (
|
|
new_group, new_path
|
|
)
|
|
|
|
except OSError as exc:
|
|
print(f" MIGRATE ERROR: {old_path.name}: {exc}")
|
|
result["errors"][mod_name] = str(exc)
|
|
|
|
print(
|
|
f" Moved: {result['moved']} "
|
|
f"Junctions removed: {result['junction_removed']} "
|
|
f"Already correct: {result['skipped_correct']} "
|
|
f"Dest exists: {result['skipped_dest_exists']} "
|
|
f"Not on disk: {result['skipped_not_found']}"
|
|
)
|
|
return result
|