fix: complete rate limiter integration and report command structure

- Add RequestRateLimiter class to providers/base.py with wait() method
- Integrate rate limiter through enrichment pipeline (enrich_identities_data → _enrich_record_with_fresh_providers → _build_providers → TMDBProvider)
- Add _cmd wrapper functions to commands/report.py for CLI imports
- Fix SeriesIdentity import in reports.py
- Add sidecar file tracking to planner review_context for move operations
- Update tests to match new rate limiter signature

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
windyboy
2026-09-25 14:03:01 +08:00
co-authored by Claude Sonnet 4.5
parent dfa18ed405
commit 0f0636bf19
6 changed files with 85 additions and 11 deletions
+30
View File
@@ -131,3 +131,33 @@ def report_duplicates(ctx: CLIContext, format: str, input: Path, output: Optiona
def report_summary(ctx: CLIContext, input: Path, output: Optional[Path]): def report_summary(ctx: CLIContext, input: Path, output: Optional[Path]):
"""Show library statistics.""" """Show library statistics."""
run_command(ctx, lambda: _run_summary(ctx, input, output), stage="summary report") run_command(ctx, lambda: _run_summary(ctx, input, output), stage="summary report")
def report_inventory_cmd(ctx: CLIContext, format: str, input: Path, output: Optional[Path]) -> None:
"""Implementation for inventory report command."""
run_command(ctx, lambda: _run_inventory(ctx, format, input, output), stage="inventory report")
def report_completeness_cmd(
ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]
) -> None:
"""Implementation for completeness report command."""
run_command(
ctx, lambda: _run_completeness(ctx, format, input, output, plan),
stage="completeness report", json_errors=True,
)
def report_duplicates_cmd(
ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]
) -> None:
"""Implementation for duplicates report command."""
run_command(
ctx, lambda: _run_duplicates(ctx, format, input, output, plan),
stage="duplicate report", json_errors=True,
)
def report_summary_cmd(ctx: CLIContext, input: Path, output: Optional[Path]) -> None:
"""Implementation for summary report command."""
run_command(ctx, lambda: _run_summary(ctx, input, output), stage="summary report")
+8 -1
View File
@@ -15,6 +15,7 @@ from vlm.cache import EnrichmentCache
from vlm.config import Config from vlm.config import Config
from vlm.parser import normalize_title from vlm.parser import normalize_title
from vlm.providers import ProviderResult, TMDBAuthError, TMDBProvider, TMDBProviderError from vlm.providers import ProviderResult, TMDBAuthError, TMDBProvider, TMDBProviderError
from vlm.providers.base import RequestRateLimiter
from vlm.utils import sanitize_path_component from vlm.utils import sanitize_path_component
RefreshMode = str RefreshMode = str
@@ -62,6 +63,7 @@ def enrich_identities_data(
refresh_all = refresh_mode == "refresh_all" refresh_all = refresh_mode == "refresh_all"
max_workers = max(1, int(config.enrichment_max_concurrency)) max_workers = max(1, int(config.enrichment_max_concurrency))
rate_limiter = RequestRateLimiter(min_interval=0.25)
for section, media_type in (("movies", "movie"), ("series", "series"), ("anime", "anime")): for section, media_type in (("movies", "movie"), ("series", "series"), ("anime", "anime")):
records = identities_data.get(section, []) records = identities_data.get(section, [])
@@ -108,6 +110,7 @@ def enrich_identities_data(
config, config,
request_timeout=request_timeout, request_timeout=request_timeout,
retries=retries, retries=retries,
rate_limiter=rate_limiter,
) )
_apply_payload(record, payload) _apply_payload(record, payload)
cache.put_identity(identity_key, fingerprint, payload) cache.put_identity(identity_key, fingerprint, payload)
@@ -125,6 +128,7 @@ def enrich_identities_data(
config, config,
request_timeout, request_timeout,
retries, retries,
rate_limiter,
) )
future_map[future] = (record, identity_key, fingerprint) future_map[future] = (record, identity_key, fingerprint)
@@ -197,7 +201,7 @@ def _update_stats_after_enrich(
_emit_progress(stats, progress_callback) _emit_progress(stats, progress_callback)
def _build_providers(config: Config, *, request_timeout: int, retries: int) -> list: def _build_providers(config: Config, *, request_timeout: int, retries: int, rate_limiter: RequestRateLimiter) -> list:
providers = [] providers = []
unsupported: list[str] = [] unsupported: list[str] = []
for name in config.enrichment_providers: for name in config.enrichment_providers:
@@ -213,6 +217,7 @@ def _build_providers(config: Config, *, request_timeout: int, retries: int) -> l
timeout_seconds=request_timeout, timeout_seconds=request_timeout,
retries=retries, retries=retries,
min_interval_seconds=0.25, min_interval_seconds=0.25,
rate_limiter=rate_limiter,
) )
) )
else: else:
@@ -232,11 +237,13 @@ def _enrich_record_with_fresh_providers(
config: Config, config: Config,
request_timeout: int, request_timeout: int,
retries: int, retries: int,
rate_limiter: RequestRateLimiter,
) -> tuple[dict, int, list[dict[str, str]], str]: ) -> tuple[dict, int, list[dict[str, str]], str]:
providers = _build_providers( providers = _build_providers(
config, config,
request_timeout=request_timeout, request_timeout=request_timeout,
retries=retries, retries=retries,
rate_limiter=rate_limiter,
) )
return _enrich_record( return _enrich_record(
record, record,
+13
View File
@@ -26,6 +26,7 @@ from vlm.plan_review import (
build_review_context, build_review_context,
normalized_path_key, normalized_path_key,
) )
from vlm.scanner import find_sidecar_companions
from vlm.utils import ( from vlm.utils import (
is_sample_path, is_sample_path,
is_within_root, is_within_root,
@@ -584,6 +585,18 @@ def _stamp_review_context_on_operations(
duplicate_group_id=gid, duplicate_group_id=gid,
keep_candidate=keep_candidate, keep_candidate=keep_candidate,
) )
if op.operation_type != "no-op" and op.destination_path:
sidecars = find_sidecar_companions(vf.path)
if sidecars:
dest_dir = op.destination_path.parent
ctx["sidecars"] = [
{
"name": s.name,
"path": str(s),
"proposed_destination_path": str(dest_dir / s.name),
}
for s in sidecars
]
stamped.append(replace(op, review_context=ctx)) stamped.append(replace(op, review_context=ctx))
continue continue
stamped.append(op) stamped.append(op)
+25
View File
@@ -2,10 +2,35 @@
from __future__ import annotations from __future__ import annotations
import threading
import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Optional, Protocol from typing import Optional, Protocol
class RequestRateLimiter:
"""Thread-safe rate limiter for API requests.
Coordinates request timing across concurrent workers to respect
provider rate limits.
"""
def __init__(self, min_interval: float):
"""Initialize with minimum seconds between requests."""
self._min_interval = min_interval
self._last_request_time = 0.0
self._lock = threading.Lock()
def wait(self) -> None:
"""Block until enough time has passed since the last request."""
with self._lock:
now = time.monotonic()
elapsed = now - self._last_request_time
if elapsed < self._min_interval:
time.sleep(self._min_interval - elapsed)
self._last_request_time = time.monotonic()
@dataclass @dataclass
class ProviderResult: class ProviderResult:
"""Normalized provider output used by enrichment pipeline.""" """Normalized provider output used by enrichment pipeline."""
+1 -1
View File
@@ -14,7 +14,7 @@ from datetime import datetime, timezone
from io import StringIO from io import StringIO
from pathlib import Path from pathlib import Path
from vlm.models import DuplicateGroup, MovieIdentity, SeasonCompleteness, VideoFile from vlm.models import DuplicateGroup, MovieIdentity, SeasonCompleteness, SeriesIdentity, VideoFile
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
+8 -9
View File
@@ -7,7 +7,7 @@ import pytest
from vlm.config import Config from vlm.config import Config
from vlm.enrichment import _build_display_title, _build_providers, enrich_identities_data from vlm.enrichment import _build_display_title, _build_providers, enrich_identities_data
from vlm.providers.base import ProviderResult from vlm.providers.base import ProviderResult, RequestRateLimiter
from vlm.providers.tmdb import TMDBAuthError from vlm.providers.tmdb import TMDBAuthError
@@ -188,7 +188,7 @@ def test_build_providers_rejects_unknown_provider(tmp_path):
) )
with pytest.raises(ValueError, match="Unsupported enrichment providers"): with pytest.raises(ValueError, match="Unsupported enrichment providers"):
_build_providers(config, request_timeout=3, retries=1) _build_providers(config, request_timeout=3, retries=1, rate_limiter=RequestRateLimiter(0.25))
def test_refresh_all_clears_stale_enrichment_fields(tmp_path, monkeypatch): def test_refresh_all_clears_stale_enrichment_fields(tmp_path, monkeypatch):
@@ -387,15 +387,14 @@ def test_enrich_respects_max_concurrency_for_uncached_records(tmp_path, monkeypa
thread_ids: set[int] = set() thread_ids: set[int] = set()
lock = threading.Lock() lock = threading.Lock()
seen_limiters: dict[str, object] = {} seen_limiters: list[object] = []
def _fake_enrich(record, media_type, config_obj, request_timeout, retries, rate_limiters=None): def _fake_enrich(record, media_type, config_obj, request_timeout, retries, rate_limiter=None):
with lock: with lock:
if rate_limiters is not None and "tmdb" in rate_limiters: if rate_limiter is not None:
limiter = rate_limiters["tmdb"] if not seen_limiters:
if "limiter" not in seen_limiters: seen_limiters.append(rate_limiter)
seen_limiters["limiter"] = limiter
else: else:
assert limiter is seen_limiters["limiter"] assert rate_limiter is seen_limiters[0]
time.sleep(0.01) time.sleep(0.01)
with lock: with lock:
thread_ids.add(threading.get_ident()) thread_ids.add(threading.get_ident())