revernomad17/arma-modlist-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add migrate step to move mod folders between groups on preset change

Browse Source 
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)
This commit is contained in:
Tran G. (Revernomad) Khoa
2026-04-14 15:08:40 +07:00
parent 45cb023513
commit 48637ffe90
6 changed files with 341 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
arma_modlist_tools/__init__.py
View File

@@ -12,6 +12,7 @@ from .config import load_config, Config
from .compat import is_windows, is_linux, get_os_label, fix_console_encoding
from .reporter import build_missing_report, save_missing_report
from .cleaner import find_orphan_folders, folder_size
from .migrator import migrate_mod_groups
__all__ = [
# parser
@@ -32,4 +33,6 @@ __all__ = [
"build_missing_report", "save_missing_report",
# cleaner
"find_orphan_folders", "folder_size",
# migrator
"migrate_mod_groups",
]

186
arma_modlist_tools/migrator.py Normal file
View File

@@ -0,0 +1,186 @@
"""
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

18
gui/app.py
View File

@@ -151,7 +151,7 @@ class ArmaModManagerApp(ctk.CTk):
# run.py calls fix_console_encoding() at import time, which needs
# the real sys.stdout.buffer. Import it before we redirect stdout.
try:
from run import step_fetch, step_link
from run import step_migrate, step_fetch, step_link
except Exception as _import_err:
self.after(0, lambda: self.post_log(
f"\n✗ Failed to load pipeline: {_import_err}\n"
@@ -165,7 +165,7 @@ class ArmaModManagerApp(ctk.CTk):
from arma_modlist_tools.compare import compare_presets
# Step 1 — Parse selected presets
_hdr("Step 1 / 4", t("pipeline.step1_name"))
_hdr("Step 1 / 5", t("pipeline.step1_name"))
cfg.modlist_json.mkdir(exist_ok=True)
presets = []
for fp in sorted(cfg.modlist_html.glob("*.html")):
@@ -182,7 +182,7 @@ class ArmaModManagerApp(ctk.CTk):
presets.append(preset)
# Step 2 — Compare
_hdr("Step 2 / 4", t("pipeline.step2_name"))
_hdr("Step 2 / 5", t("pipeline.step2_name"))
result = compare_presets(*presets)
cfg.comparison.write_text(
json.dumps(result, indent=2, ensure_ascii=False),
@@ -193,12 +193,16 @@ class ArmaModManagerApp(ctk.CTk):
print(f" Shared: {result['shared']['mod_count']} | "
f"Unique: {total_unique}")
# Step 3 — Fetch
_hdr("Step 3 / 4", t("pipeline.step3_name"))
# Step 3 — Migrate
_hdr("Step 3 / 5", t("pipeline.step3_name"))
step_migrate(cfg)
# Step 4 — Fetch
_hdr("Step 4 / 5", t("pipeline.step4_name"))
step_fetch(cfg)
# Step 4 — Link
_hdr("Step 4 / 4", t("pipeline.step4_name"))
# Step 5 — Link
_hdr("Step 5 / 5", t("pipeline.step5_name"))
groups = (
sorted(p.name for p in cfg.downloads.iterdir() if p.is_dir())
if cfg.downloads.is_dir() else []

10
gui/locales.py
View File

@@ -29,8 +29,9 @@ _EN: dict[str, str] = {
"pipeline.starting": "Pipeline started",
"pipeline.step1_name": "Parse presets",
"pipeline.step2_name": "Compare presets",
"pipeline.step3_name": "Download mods",
"pipeline.step4_name": "Link mods",
"pipeline.step3_name": "Migrate mod groups",
"pipeline.step4_name": "Download mods",
"pipeline.step5_name": "Link mods",
# ── app.py dialogs ────────────────────────────────────────────────────────
"app.dlg_presets_title": "Not enough presets selected",
@@ -270,8 +271,9 @@ _VI: dict[str, str] = {
"pipeline.starting": "Pipeline đã bắt đầu",
"pipeline.step1_name": "Phân tích preset",
"pipeline.step2_name": "So sánh preset",
"pipeline.step3_name": "Tải mod",
"pipeline.step4_name": "Liên kết mod",
"pipeline.step3_name": "Di chuyển nhóm mod",
"pipeline.step4_name": "Tải mod",
"pipeline.step5_name": "Liên kết mod",
# ── app.py dialogs ────────────────────────────────────────────────────────
"app.dlg_presets_title": "Chưa chọn đủ preset",

24
run.py
View File

@@ -78,6 +78,22 @@ def step_compare(cfg) -> None:
print(f" -> {cfg.comparison}")
def step_migrate(cfg) -> None:
if not cfg.comparison.exists():
print(f" NOTE: {cfg.comparison} not found — skipping migration.")
return
if not cfg.downloads.is_dir():
print(f" NOTE: downloads dir missing ({cfg.downloads}) — nothing to migrate.")
return
from arma_modlist_tools.migrator import migrate_mod_groups
comparison = json.loads(cfg.comparison.read_text(encoding="utf-8"))
arma_dir = cfg.arma_dir if cfg.arma_dir.is_dir() else None
result = migrate_mod_groups(cfg.downloads, arma_dir, comparison)
if result["errors"]:
for name, err in result["errors"].items():
print(f" ERROR {name}: {err}")
def step_fetch(cfg) -> None:
if not cfg.comparison.exists():
print(f" ERROR: {cfg.comparison} not found. Run parse + compare first.")
@@ -178,6 +194,7 @@ def main() -> None:
parser = argparse.ArgumentParser(description="Run the full mod management pipeline.")
parser.add_argument("--skip-parse", action="store_true")
parser.add_argument("--skip-compare", action="store_true")
parser.add_argument("--skip-migrate", action="store_true")
parser.add_argument("--skip-fetch", action="store_true")
parser.add_argument("--skip-link", action="store_true")
parser.add_argument("--group", "-g", metavar="GROUP",
@@ -187,13 +204,15 @@ def main() -> None:
steps = [
(not args.skip_parse, "Parse presets"),
(not args.skip_compare, "Compare presets"),
(not args.skip_migrate, "Migrate mods"),
(not args.skip_fetch, "Fetch mods"),
(not args.skip_link, "Link mods"),
]
active_count = sum(1 for run, _ in steps if run)
step_num = 0
if args.skip_parse and args.skip_compare and args.skip_fetch and args.skip_link:
if (args.skip_parse and args.skip_compare and args.skip_migrate
and args.skip_fetch and args.skip_link):
print("All steps skipped — nothing to do.")
sys.exit(0)
@@ -201,6 +220,7 @@ def main() -> None:
(cfg,),
(cfg,),
(cfg,),
(cfg,),
None, # handled separately
]):
if not run:
@@ -221,6 +241,8 @@ def main() -> None:
step_parse(cfg)
elif name == "Compare presets":
step_compare(cfg)
elif name == "Migrate mods":
step_migrate(cfg)
elif name == "Fetch mods":
step_fetch(cfg)

112
test_suite.py
View File

@@ -2351,6 +2351,118 @@ test("_find_folder: missing meta.cpp silently skipped",
_with_tmp(_test_ff_missing_meta_cpp_skipped))
# ---------------------------------------------------------------------------
# migrator — migrate_mod_groups
# ---------------------------------------------------------------------------
group("migrator — migrate_mod_groups")
from arma_modlist_tools.migrator import migrate_mod_groups as _migrate
def _make_mod(dl: Path, grp: str, folder: str, steam_id: str | None = None) -> Path:
"""Create a minimal mod folder under downloads/group/folder."""
d = dl / grp / folder
d.mkdir(parents=True)
if steam_id:
(d / "meta.cpp").write_text(f"publishedid = {steam_id};\n", encoding="utf-8")
(d / "dummy.pbo").write_bytes(b"\x00" * 8)
return d
def _test_migrate_already_correct():
comp = {"shared": {"mods": [{"name": "CBA_A3", "steam_id": "450814997"}]}, "unique": {}}
with tempfile.TemporaryDirectory() as d:
dl = Path(d) / "downloads"
_make_mod(dl, "shared", "@CBA_A3", "450814997")
result = _migrate(dl, None, comp)
assert_eq(result["moved"], 0)
assert_eq(result["skipped_correct"], 1)
def _test_migrate_moves_by_steam_id():
comp = {"shared": {"mods": []}, "unique": {
"A_v1": {"mods": [{"name": "ACE3", "steam_id": "463939057"}]}
}}
with tempfile.TemporaryDirectory() as d:
dl = Path(d) / "downloads"
old = _make_mod(dl, "A", "@ACE3", "463939057")
result = _migrate(dl, None, comp)
assert_eq(result["moved"], 1)
assert not old.exists(), "old folder must be gone"
assert (dl / "A_v1" / "@ACE3").exists(), "new folder must exist"
assert (dl / "A_v1" / "@ACE3" / "dummy.pbo").exists(), "files preserved"
def _test_migrate_moves_by_normalized_name():
"""No meta.cpp — matching falls back to normalised folder name."""
comp = {"shared": {"mods": [{"name": "CBA_A3", "steam_id": None}]}, "unique": {
"A": {"mods": []}
}}
with tempfile.TemporaryDirectory() as d:
dl = Path(d) / "downloads"
_make_mod(dl, "A", "@CBA_A3", steam_id=None)
result = _migrate(dl, None, comp)
assert_eq(result["moved"], 1)
assert (dl / "shared" / "@CBA_A3").exists()
def _test_migrate_skips_dest_exists():
comp = {"shared": {"mods": [{"name": "CBA_A3", "steam_id": "450814997"}]}, "unique": {}}
with tempfile.TemporaryDirectory() as d:
dl = Path(d) / "downloads"
_make_mod(dl, "A", "@CBA_A3", "450814997")
_make_mod(dl, "shared", "@CBA_A3", "450814997")
result = _migrate(dl, None, comp)
assert_eq(result["moved"], 0)
assert_eq(result["skipped_dest_exists"], 1)
def _test_migrate_skips_not_on_disk():
comp = {"shared": {"mods": [{"name": "CBA_A3", "steam_id": "450814997"}]}, "unique": {}}
with tempfile.TemporaryDirectory() as d:
dl = Path(d) / "downloads"
dl.mkdir()
result = _migrate(dl, None, comp)
assert_eq(result["skipped_not_found"], 1)
assert_eq(result["moved"], 0)
def _test_migrate_removes_stale_junction():
from arma_modlist_tools.linker import create_junction, _is_junction
comp = {"shared": {"mods": [{"name": "ACE3", "steam_id": "463939057"}]}, "unique": {}}
with tempfile.TemporaryDirectory() as dl_d, \
tempfile.TemporaryDirectory() as arma_d:
dl = Path(dl_d) / "downloads"
arma = Path(arma_d)
old = _make_mod(dl, "A", "@ACE3", "463939057")
link = arma / "@ACE3"
create_junction(link, old)
assert _is_junction(link), "precondition: junction must exist"
result = _migrate(dl, arma, comp)
assert_eq(result["moved"], 1)
assert_eq(result["junction_removed"], 1)
assert not _is_junction(link), "stale junction must be removed"
assert (dl / "shared" / "@ACE3").exists()
def _test_migrate_missing_downloads_dir():
comp = {"shared": {"mods": [{"name": "X", "steam_id": "1"}]}, "unique": {}}
with tempfile.TemporaryDirectory() as d:
result = _migrate(Path(d) / "nonexistent", None, comp)
assert_eq(result["moved"], 0)
assert_eq(result["skipped_not_found"], 1)
test("migrator: already in correct group → no move", _test_migrate_already_correct)
test("migrator: moves mod by steam_id to new group", _test_migrate_moves_by_steam_id)
test("migrator: moves mod by normalized name (no meta.cpp)", _test_migrate_moves_by_normalized_name)
test("migrator: skips when destination already exists", _test_migrate_skips_dest_exists)
test("migrator: skips mod not on disk", _test_migrate_skips_not_on_disk)
test("migrator: removes stale junction before moving", _test_migrate_removes_stale_junction)
test("migrator: no-op when downloads dir missing", _test_migrate_missing_downloads_dir)
# ---------------------------------------------------------------------------
# Summary
# ---------------------------------------------------------------------------