francwa
ba6f016d49
- Extract MetadataStore from SubtitleMetadataStore (alfred/infrastructure/metadata/).
Generic load/save + typed update helpers (update_parse, update_probe, update_tmdb)
for the per-release .alfred/metadata.yaml.
- SubtitleMetadataStore becomes a thin facade — owns subtitle_history shape,
delegates I/O to MetadataStore.
- Agent._execute_tool_call auto-persists successful analyze_release / probe_media /
find_media_imdb_id results to the release's .alfred file. find_media_imdb_id
follows release_focus when it has no path argument.
- New tools:
· read_release_metadata(release_path) — cacheable, key=release_path.
Returns the .alfred content or has_metadata=false.
· query_library(name) — substring scan across configured library roots.
- Both new tools added to CORE_TOOLS (always visible).
366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Filesystem tools for folder management."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
import alfred as _alfred_pkg
|
|
from alfred.application.filesystem import (
|
|
CreateSeedLinksUseCase,
|
|
ListFolderUseCase,
|
|
ManageSubtitlesUseCase,
|
|
MoveMediaUseCase,
|
|
SetFolderPathUseCase,
|
|
)
|
|
from alfred.application.filesystem.detect_media_type import detect_media_type
|
|
from alfred.application.filesystem.enrich_from_probe import enrich_from_probe
|
|
from alfred.application.filesystem.resolve_destination import (
|
|
resolve_episode_destination as _resolve_episode_destination,
|
|
)
|
|
from alfred.application.filesystem.resolve_destination import (
|
|
resolve_movie_destination as _resolve_movie_destination,
|
|
)
|
|
from alfred.application.filesystem.resolve_destination import (
|
|
resolve_season_destination as _resolve_season_destination,
|
|
)
|
|
from alfred.application.filesystem.resolve_destination import (
|
|
resolve_series_destination as _resolve_series_destination,
|
|
)
|
|
from alfred.infrastructure.filesystem import FileManager, create_folder, move
|
|
from alfred.infrastructure.filesystem.ffprobe import probe
|
|
from alfred.infrastructure.filesystem.find_video import find_video_file
|
|
from alfred.infrastructure.metadata import MetadataStore
|
|
from alfred.infrastructure.persistence import get_memory
|
|
|
|
_LEARNED_ROOT = Path(_alfred_pkg.__file__).parent.parent / "data" / "knowledge"
|
|
|
|
|
|
def move_media(source: str, destination: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/move_media.yaml."""
|
|
file_manager = FileManager()
|
|
use_case = MoveMediaUseCase(file_manager)
|
|
return use_case.execute(source, destination).to_dict()
|
|
|
|
|
|
def move_to_destination(source: str, destination: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/move_to_destination.yaml."""
|
|
parent = str(Path(destination).parent)
|
|
result = create_folder(parent)
|
|
if result["status"] != "ok":
|
|
return result
|
|
return move(source, destination)
|
|
|
|
|
|
def resolve_season_destination(
|
|
release_name: str,
|
|
tmdb_title: str,
|
|
tmdb_year: int,
|
|
confirmed_folder: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_season_destination.yaml."""
|
|
return _resolve_season_destination(
|
|
release_name, tmdb_title, tmdb_year, confirmed_folder
|
|
).to_dict()
|
|
|
|
|
|
def resolve_episode_destination(
|
|
release_name: str,
|
|
source_file: str,
|
|
tmdb_title: str,
|
|
tmdb_year: int,
|
|
tmdb_episode_title: str | None = None,
|
|
confirmed_folder: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_episode_destination.yaml."""
|
|
return _resolve_episode_destination(
|
|
release_name,
|
|
source_file,
|
|
tmdb_title,
|
|
tmdb_year,
|
|
tmdb_episode_title,
|
|
confirmed_folder,
|
|
).to_dict()
|
|
|
|
|
|
def resolve_movie_destination(
|
|
release_name: str,
|
|
source_file: str,
|
|
tmdb_title: str,
|
|
tmdb_year: int,
|
|
) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_movie_destination.yaml."""
|
|
return _resolve_movie_destination(
|
|
release_name, source_file, tmdb_title, tmdb_year
|
|
).to_dict()
|
|
|
|
|
|
def resolve_series_destination(
|
|
release_name: str,
|
|
tmdb_title: str,
|
|
tmdb_year: int,
|
|
confirmed_folder: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/resolve_series_destination.yaml."""
|
|
return _resolve_series_destination(
|
|
release_name, tmdb_title, tmdb_year, confirmed_folder
|
|
).to_dict()
|
|
|
|
|
|
def create_seed_links(
|
|
library_file: str, original_download_folder: str
|
|
) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/create_seed_links.yaml."""
|
|
file_manager = FileManager()
|
|
use_case = CreateSeedLinksUseCase(file_manager)
|
|
return use_case.execute(library_file, original_download_folder).to_dict()
|
|
|
|
|
|
def manage_subtitles(source_video: str, destination_video: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/manage_subtitles.yaml."""
|
|
file_manager = FileManager()
|
|
use_case = ManageSubtitlesUseCase(file_manager)
|
|
return use_case.execute(source_video, destination_video).to_dict()
|
|
|
|
|
|
def learn(pack: str, category: str, key: str, values: list[str]) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/learn.yaml."""
|
|
_VALID_PACKS = {"subtitles"}
|
|
_VALID_CATEGORIES = {"languages", "types", "formats"}
|
|
|
|
if pack not in _VALID_PACKS:
|
|
return {
|
|
"status": "error",
|
|
"error": "unknown_pack",
|
|
"message": f"Unknown pack '{pack}'. Valid: {sorted(_VALID_PACKS)}",
|
|
}
|
|
|
|
if category not in _VALID_CATEGORIES:
|
|
return {
|
|
"status": "error",
|
|
"error": "unknown_category",
|
|
"message": f"Unknown category '{category}'. Valid: {sorted(_VALID_CATEGORIES)}",
|
|
}
|
|
|
|
learned_path = _LEARNED_ROOT / "subtitles_learned.yaml"
|
|
_LEARNED_ROOT.mkdir(parents=True, exist_ok=True)
|
|
|
|
data: dict = {}
|
|
if learned_path.exists():
|
|
try:
|
|
with open(learned_path, encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
except Exception as e:
|
|
return {"status": "error", "error": "read_failed", "message": str(e)}
|
|
|
|
cat_data = data.setdefault(category, {})
|
|
entry = cat_data.setdefault(key, {"tokens": []})
|
|
existing = entry.get("tokens", [])
|
|
new_tokens = [v for v in values if v not in existing]
|
|
entry["tokens"] = existing + new_tokens
|
|
|
|
tmp = learned_path.with_suffix(".yaml.tmp")
|
|
try:
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
yaml.safe_dump(
|
|
data, f, allow_unicode=True, default_flow_style=False, sort_keys=False
|
|
)
|
|
tmp.rename(learned_path)
|
|
except Exception as e:
|
|
tmp.unlink(missing_ok=True)
|
|
return {"status": "error", "error": "write_failed", "message": str(e)}
|
|
|
|
return {
|
|
"status": "ok",
|
|
"pack": pack,
|
|
"category": category,
|
|
"key": key,
|
|
"added_count": len(new_tokens),
|
|
"tokens": entry["tokens"],
|
|
}
|
|
|
|
|
|
def set_path_for_folder(folder_name: str, path_value: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/set_path_for_folder.yaml."""
|
|
file_manager = FileManager()
|
|
use_case = SetFolderPathUseCase(file_manager)
|
|
response = use_case.execute(folder_name, path_value)
|
|
return response.to_dict()
|
|
|
|
|
|
def analyze_release(release_name: str, source_path: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/analyze_release.yaml."""
|
|
from alfred.domain.release.services import parse_release # noqa: PLC0415
|
|
|
|
path = Path(source_path)
|
|
parsed = parse_release(release_name)
|
|
parsed.media_type = detect_media_type(parsed, path)
|
|
|
|
probe_used = False
|
|
if parsed.media_type not in ("unknown", "other"):
|
|
video_file = find_video_file(path)
|
|
if video_file:
|
|
media_info = probe(video_file)
|
|
if media_info:
|
|
enrich_from_probe(parsed, media_info)
|
|
probe_used = True
|
|
|
|
return {
|
|
"status": "ok",
|
|
"media_type": parsed.media_type,
|
|
"parse_path": parsed.parse_path,
|
|
"title": parsed.title,
|
|
"year": parsed.year,
|
|
"season": parsed.season,
|
|
"episode": parsed.episode,
|
|
"episode_end": parsed.episode_end,
|
|
"quality": parsed.quality,
|
|
"source": parsed.source,
|
|
"codec": parsed.codec,
|
|
"group": parsed.group,
|
|
"languages": parsed.languages,
|
|
"audio_codec": parsed.audio_codec,
|
|
"audio_channels": parsed.audio_channels,
|
|
"bit_depth": parsed.bit_depth,
|
|
"hdr_format": parsed.hdr_format,
|
|
"edition": parsed.edition,
|
|
"site_tag": parsed.site_tag,
|
|
"is_season_pack": parsed.is_season_pack,
|
|
"probe_used": probe_used,
|
|
}
|
|
|
|
|
|
def probe_media(source_path: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/probe_media.yaml."""
|
|
path = Path(source_path)
|
|
if not path.exists():
|
|
return {
|
|
"status": "error",
|
|
"error": "not_found",
|
|
"message": f"{source_path} does not exist",
|
|
}
|
|
|
|
media_info = probe(path)
|
|
if media_info is None:
|
|
return {
|
|
"status": "error",
|
|
"error": "probe_failed",
|
|
"message": "ffprobe failed to read the file",
|
|
}
|
|
|
|
return {
|
|
"status": "ok",
|
|
"video": {
|
|
"codec": media_info.video_codec,
|
|
"resolution": media_info.resolution,
|
|
"width": media_info.width,
|
|
"height": media_info.height,
|
|
"duration_seconds": media_info.duration_seconds,
|
|
"bitrate_kbps": media_info.bitrate_kbps,
|
|
},
|
|
"audio_tracks": [
|
|
{
|
|
"index": t.index,
|
|
"codec": t.codec,
|
|
"channels": t.channels,
|
|
"channel_layout": t.channel_layout,
|
|
"language": t.language,
|
|
"is_default": t.is_default,
|
|
}
|
|
for t in media_info.audio_tracks
|
|
],
|
|
"subtitle_tracks": [
|
|
{
|
|
"index": t.index,
|
|
"codec": t.codec,
|
|
"language": t.language,
|
|
"is_default": t.is_default,
|
|
"is_forced": t.is_forced,
|
|
}
|
|
for t in media_info.subtitle_tracks
|
|
],
|
|
"audio_languages": media_info.audio_languages,
|
|
"is_multi_audio": media_info.is_multi_audio,
|
|
}
|
|
|
|
|
|
def list_folder(folder_type: str, path: str = ".") -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/list_folder.yaml."""
|
|
file_manager = FileManager()
|
|
use_case = ListFolderUseCase(file_manager)
|
|
response = use_case.execute(folder_type, path)
|
|
return response.to_dict()
|
|
|
|
|
|
def read_release_metadata(release_path: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/read_release_metadata.yaml."""
|
|
path = Path(release_path)
|
|
if not path.exists():
|
|
return {
|
|
"status": "error",
|
|
"error": "not_found",
|
|
"message": f"{release_path} does not exist",
|
|
}
|
|
root = path if path.is_dir() else path.parent
|
|
store = MetadataStore(root)
|
|
if not store.exists():
|
|
return {
|
|
"status": "ok",
|
|
"release_path": str(root),
|
|
"has_metadata": False,
|
|
"metadata": {},
|
|
}
|
|
return {
|
|
"status": "ok",
|
|
"release_path": str(root),
|
|
"has_metadata": True,
|
|
"metadata": store.load(),
|
|
}
|
|
|
|
|
|
def query_library(name: str) -> dict[str, Any]:
|
|
"""Thin tool wrapper — semantics live in alfred/agent/tools/specs/query_library.yaml."""
|
|
needle = name.strip().lower()
|
|
if not needle:
|
|
return {
|
|
"status": "error",
|
|
"error": "empty_name",
|
|
"message": "name must be a non-empty string",
|
|
}
|
|
|
|
memory = get_memory()
|
|
roots = memory.ltm.library_paths.to_dict() or {}
|
|
if not roots:
|
|
return {
|
|
"status": "error",
|
|
"error": "no_libraries",
|
|
"message": "No library paths configured — call set_path_for_folder first.",
|
|
}
|
|
|
|
matches: list[dict[str, Any]] = []
|
|
for collection, root in roots.items():
|
|
root_path = Path(root)
|
|
if not root_path.is_dir():
|
|
continue
|
|
for entry in root_path.iterdir():
|
|
if not entry.is_dir():
|
|
continue
|
|
if needle not in entry.name.lower():
|
|
continue
|
|
store = MetadataStore(entry)
|
|
matches.append(
|
|
{
|
|
"collection": collection,
|
|
"name": entry.name,
|
|
"path": str(entry),
|
|
"has_metadata": store.exists(),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"status": "ok",
|
|
"query": name,
|
|
"match_count": len(matches),
|
|
"matches": matches,
|
|
}
|