fix: 5 findings from codebase audit round 5
cli.py:
- _smaller_duplicate_ids: walrus operator eliminates double p.get("id")
per element; truthiness check replaces dead "is not None" guard (all
persons in by_name are guaranteed to have a truthy id after the
line-75 gate)
- Extract _excl() helper inside _handle_duplicate_people — replaces 4
identical [p for p in lst if p.get("id") not in skip_ids] expressions
across all return paths
jobs.py:
- Extract _valid_people() — shared filter for interactive_configure and
auto_configure; uses (p.get("name") or "").strip() to match cli.py's
whitespace-strip gate, preventing whitespace-only Immich names from
reaching _build_job and creating blank Frigate person labels
- Hoist queued_ids set before the display loop in interactive_configure:
O(N) set lookup per render instead of O(N×|jobs|) linear scan
This commit is contained in:
+9
-6
@@ -82,14 +82,17 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
|
||||
def _smaller_duplicate_ids(groups: dict) -> set[str]:
|
||||
"""IDs of all but the largest person in each duplicate group."""
|
||||
return {
|
||||
p.get("id")
|
||||
pid
|
||||
for ps in groups.values()
|
||||
for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
|
||||
if p.get("id") is not None
|
||||
if (pid := p.get("id"))
|
||||
}
|
||||
|
||||
skip_ids = _smaller_duplicate_ids(duplicates)
|
||||
|
||||
def _excl(lst: list[dict]) -> list[dict]:
|
||||
return [p for p in lst if p.get("id") not in skip_ids]
|
||||
|
||||
if not Config.MERGE_DUPLICATE_PEOPLE:
|
||||
rprint("\n[bold yellow]⚠ Duplicate person names detected in Immich:[/bold yellow]")
|
||||
for name, ps in sorted(duplicates.items()):
|
||||
@@ -111,7 +114,7 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
|
||||
)
|
||||
# Return deduplicated list — keep only the largest per name so that
|
||||
# downstream job creation never runs two jobs for the same Frigate folder.
|
||||
return [p for p in people if p.get("id") not in skip_ids]
|
||||
return _excl(people)
|
||||
|
||||
# Auto-merge: survivor = largest asset count, rest merge into it inside Immich
|
||||
merged_any = False
|
||||
@@ -144,12 +147,12 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
|
||||
" — possible transient error or expired API key;"
|
||||
" proceeding with pre-merge list. Check IMMICH_API_KEY if this recurs."
|
||||
)
|
||||
return [p for p in people if p.get("id") not in skip_ids]
|
||||
return _excl(people)
|
||||
# Filter out the smaller duplicate from any group whose merge failed — those
|
||||
# IDs still exist in Immich and would produce two jobs for the same folder.
|
||||
# IDs from groups that merged successfully are already gone from Immich, so
|
||||
# this filter is a no-op for them.
|
||||
return [p for p in fresh if p.get("id") not in skip_ids]
|
||||
return _excl(fresh)
|
||||
|
||||
# All merges failed — fall back to local deduplication (keep largest per name) so
|
||||
# downstream job creation never runs two jobs for the same Frigate folder.
|
||||
@@ -157,7 +160,7 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
|
||||
" [yellow]All merges failed — applying local deduplication"
|
||||
" to avoid overwriting output.[/yellow]"
|
||||
)
|
||||
return [p for p in people if p.get("id") not in skip_ids]
|
||||
return _excl(people)
|
||||
|
||||
|
||||
_UNSUPPORTED_VARS = [
|
||||
|
||||
+11
-4
@@ -192,13 +192,20 @@ def _configure_person(person: dict, people: list[dict]) -> dict | None:
|
||||
return job
|
||||
|
||||
|
||||
def _valid_people(people: list[dict]) -> list[dict]:
|
||||
return sorted(
|
||||
[p for p in people if (p.get("name") or "").strip() and p.get("id")],
|
||||
key=lambda x: x["name"],
|
||||
)
|
||||
|
||||
|
||||
def interactive_configure(people: list[dict]) -> list[dict]:
|
||||
"""Interactive phase: select person(s), mode, and configure training strategy.
|
||||
|
||||
Supports multi-person batch mode — after configuring one person,
|
||||
prompts to add another.
|
||||
"""
|
||||
valid_people = sorted([p for p in people if p.get("name") and p.get("id")], key=lambda x: x["name"])
|
||||
valid_people = _valid_people(people)
|
||||
|
||||
if not valid_people:
|
||||
rprint("[red]No people found with names in Immich.[/red]")
|
||||
@@ -209,9 +216,9 @@ def interactive_configure(people: list[dict]) -> list[dict]:
|
||||
while True:
|
||||
# Select person
|
||||
console.print("\n[bold cyan]Select Person to Train:[/bold cyan]")
|
||||
queued_ids = {j["person"]["id"] for j in jobs}
|
||||
for idx, p in enumerate(valid_people, 1):
|
||||
# Mark already-queued people
|
||||
marker = " [dim](queued)[/dim]" if any(j["person"]["id"] == p.get("id") for j in jobs) else ""
|
||||
marker = " [dim](queued)[/dim]" if p.get("id") in queued_ids else ""
|
||||
console.print(f" [bold]{idx}.[/bold] {p['name']}{marker}")
|
||||
|
||||
p_choice = IntPrompt.ask("Enter Number", choices=[str(i) for i in range(1, len(valid_people) + 1)])
|
||||
@@ -230,7 +237,7 @@ def interactive_configure(people: list[dict]) -> list[dict]:
|
||||
|
||||
def auto_configure(people: list[dict]) -> list[dict]:
|
||||
"""Non-interactive: configure jobs for all named people automatically."""
|
||||
valid_people = sorted([p for p in people if p.get("name") and p.get("id")], key=lambda x: x["name"])
|
||||
valid_people = _valid_people(people)
|
||||
|
||||
if not valid_people:
|
||||
rprint("[red]No people found with names in Immich.[/red]")
|
||||
|
||||
Reference in New Issue
Block a user