Implement duplicate resolution strategy by quality and enhance related documentation

- Updated `duplicate_resolve.py` to introduce a new strategy for keeping files based on quality, considering resolution, source, codec, and size.
- Enhanced `planner.py` to utilize the new quality-based strategy during plan generation, updating quarantine reasons accordingly.
- Modified `README.md` to document the new `plan.duplicate_keep` options, including `by_quality`, and provided detailed descriptions of each strategy.
- Added unit tests in `test_duplicate_resolve.py` to validate the new quality-based resolution logic.
- Updated `analysis.json` and `plan.json` with new timestamps and IDs to reflect recent changes.

These updates improve the Video Library Manager's ability to handle duplicate files more effectively, ensuring users retain the highest quality versions.
This commit is contained in:
windyboy
2026-02-11 08:44:48 +08:00
parent 79f5ddf1f5
commit 1f55eab304
9 changed files with 297 additions and 20 deletions
+7 -2
View File
@@ -324,9 +324,14 @@ def validate_config(config: Config) -> list[str]:
errors.append("tmdb_region must be a string when set")
if not isinstance(config.tmdb_include_adult, bool):
errors.append("tmdb_include_adult must be a boolean")
if config.duplicate_keep not in ("by_reputation", "first_seen", "manual"):
if config.duplicate_keep not in (
"by_reputation",
"first_seen",
"manual",
"by_quality",
):
errors.append(
f"duplicate_keep must be one of 'by_reputation', 'first_seen', 'manual', got: {config.duplicate_keep!r}"
f"duplicate_keep must be one of 'by_reputation', 'first_seen', 'manual', 'by_quality', got: {config.duplicate_keep!r}"
)
return errors
+85 -2
View File
@@ -1,7 +1,8 @@
"""Duplicate group resolution: choose which file to keep when consuming analysis."""
import re
from pathlib import Path
from typing import Union
from typing import Optional, Union
from vlm.models import MovieIdentity, SeriesIdentity
@@ -9,10 +10,12 @@ from vlm.models import MovieIdentity, SeriesIdentity
def choose_keep_index(
items: list[tuple[Path, Union[MovieIdentity, SeriesIdentity]]],
strategy: str,
quality_comparison: Optional[list[dict]] = None,
) -> int | None:
"""Choose the index of the item to keep in a duplicate group.
Strategies:
- by_quality: Prefer higher quality (resolution > source > codec > size). Requires quality_comparison.
- by_reputation: Prefer items with reputation_score; then sort by score desc,
then reputation_votes desc, then first_seen (input order). Keep index 0 after sort.
- first_seen: Keep the first item (index 0).
@@ -20,7 +23,9 @@ def choose_keep_index(
Args:
items: List of (path, identity) for the duplicate group.
strategy: One of "by_reputation", "first_seen", "manual".
strategy: One of "by_quality", "by_reputation", "first_seen", "manual".
quality_comparison: List of quality dicts (filename, path, size_bytes, resolution?, codec?)
aligned with items. Required when strategy is "by_quality".
Returns:
Index in [0, len(items)) to keep, or None for manual.
@@ -29,11 +34,89 @@ def choose_keep_index(
return None
if strategy == "first_seen":
return 0
if strategy == "by_quality":
if quality_comparison is None or len(quality_comparison) != len(items):
return 0 # Fallback to first if quality data missing/mismatched
return _by_quality_index(items, quality_comparison)
if strategy == "by_reputation":
return _by_reputation_index(items)
return 0
def _parse_resolution_tier(resolution: Optional[str], path: Path) -> int:
"""Parse resolution to tier. Higher = better. 2160=4, 1080=3, 720=2, 0=unknown."""
if resolution:
m = re.search(r"(\d{3,4})[x×](\d{3,4})", resolution, re.I)
if m:
h = int(m.group(2))
if h >= 2160:
return 4
if h >= 1080:
return 3
if h >= 720:
return 2
return 1
s = path.name.upper()
if "2160" in s or "4K" in s or "UHD" in s:
return 4
if "1080" in s:
return 3
if "720" in s:
return 2
if "480" in s or "576" in s:
return 1
return 0
def _parse_source_tier(path: Path) -> int:
"""Parse source type from path. BluRay=3, WEB-DL=2, HDTV=1, 0=other."""
s = path.name.upper()
if "BLURAY" in s or "BLU-RAY" in s or "BD" in s:
return 3
if "WEB-DL" in s or "WEBRIP" in s or "WEB" in s:
return 2
if "HDTV" in s or "HDTVRIP" in s:
return 1
return 0
def _parse_codec_tier(codec: Optional[str], path: Path) -> int:
"""Parse codec to tier. hevc/x265=2, h264/x264=1, 0=other."""
if codec:
c = codec.lower()
if "hevc" in c or "h265" in c or "x265" in c:
return 2
if "h264" in c or "x264" in c or "avc" in c:
return 1
s = path.name.upper()
if "X265" in s or "H265" in s or "HEVC" in s:
return 2
if "X264" in s or "H264" in s or "AVC" in s:
return 1
return 0
def _by_quality_index(
items: list[tuple[Path, Union[MovieIdentity, SeriesIdentity]]],
quality_comparison: list[dict],
) -> int:
"""Sort by: resolution desc, source desc, codec desc, size desc, index asc. Return best index."""
def key(idx_item: tuple[int, tuple[Path, Union[MovieIdentity, SeriesIdentity]]]) -> tuple:
idx, (path, _) = idx_item
qc = quality_comparison[idx] if idx < len(quality_comparison) else {}
resolution = qc.get("resolution")
codec = qc.get("codec")
size = qc.get("size_bytes", 0) or 0
res_tier = _parse_resolution_tier(resolution, path)
src_tier = _parse_source_tier(path)
codec_tier = _parse_codec_tier(codec, path)
return (-res_tier, -src_tier, -codec_tier, -size, idx)
indexed = list(enumerate(items))
indexed.sort(key=key)
return indexed[0][0]
def _by_reputation_index(
items: list[tuple[Path, Union[MovieIdentity, SeriesIdentity]]],
) -> int:
+25 -10
View File
@@ -22,6 +22,7 @@ from vlm.models import (
)
QUARANTINE_REASON_DUPLICATE = "重复项(已按外部评分保留一条)"
QUARANTINE_REASON_DUPLICATE_BY_QUALITY = "重复项(已按画质保留最优)"
def generate_plan(
@@ -50,26 +51,40 @@ def generate_plan(
for dup in analysis_data.get("duplicates", []):
paths = dup.get("files", [])
indices = [path_to_index[p] for p in paths if p in path_to_index]
items = [
(identities[i][0].path, identities[i][1])
for i in indices
if identities[i][1] is not None
and isinstance(identities[i][1], (MovieIdentity, SeriesIdentity))
]
path_to_qc = {qc.get("path"): qc for qc in dup.get("quality_comparison", [])}
items = []
valid_indices = []
for i in indices:
identity = identities[i][1]
if identity is not None and isinstance(
identity, (MovieIdentity, SeriesIdentity)
):
items.append((identities[i][0].path, identity))
valid_indices.append(i)
if not items:
continue
keep_idx = choose_keep_index(items, config.duplicate_keep)
quality_list = [
path_to_qc.get(str(p), {}) for p, _ in items
]
keep_idx = choose_keep_index(
items, config.duplicate_keep, quality_comparison=quality_list
)
if keep_idx is None:
continue
keep_identity_index = indices[keep_idx]
quarantine_indices = set(indices) - {keep_identity_index}
keep_identity_index = valid_indices[keep_idx]
quarantine_indices = set(valid_indices) - {keep_identity_index}
reason = (
QUARANTINE_REASON_DUPLICATE_BY_QUALITY
if config.duplicate_keep == "by_quality"
else QUARANTINE_REASON_DUPLICATE
)
for i in quarantine_indices:
vf = identities[i][0]
operations[i] = FileOperation(
operation_type="quarantine",
source_path=vf.path,
destination_path=None,
reason=QUARANTINE_REASON_DUPLICATE,
reason=reason,
has_conflict=False,
conflict_reason=None,
)