423 lines
17 KiB
Python
423 lines
17 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
from .tag import Tag
|
|
|
|
|
|
class File:
|
|
def __init__(self, file_path: Path, tagmanager=None, index=None) -> None:
|
|
self.file_path = file_path
|
|
self.filename = file_path.name
|
|
# Unified pool index; when set, metadata is persisted there (see
|
|
# PoolIndex). Without an index a File is an in-memory object only — no
|
|
# metadata is written to disk (there are no per-file sidecars anymore).
|
|
self.index = index
|
|
self.new = True
|
|
self.ignored = False
|
|
self.tags: list[Tag] = []
|
|
self.tagmanager = tagmanager
|
|
# new: optional date string "YYYY-MM-DD" (assigned manually)
|
|
self.date: str | None = None
|
|
# movie-library fields
|
|
self.title: str | None = None
|
|
# Optional ČSFD film URL — a second database pointer kept alongside the
|
|
# TMDb id (which lives in the filename). Stored only, not fetched.
|
|
self.csfd_url: str | None = None
|
|
# ISO timestamp of when the movie was added to the pool (set on import);
|
|
# None for older/unknown → falls back to the file's ctime.
|
|
self.added: str | None = None
|
|
# Cached metadata used for folder grouping (year, rating) — filled from
|
|
# the movie's NFO (which Jellyfin writes).
|
|
self.meta: dict | None = None
|
|
# full_paths of tags derived automatically from the NFO (vs. user-added);
|
|
# only these are regenerated on refresh — user tags are preserved.
|
|
self.auto_tag_paths: set[str] = set()
|
|
# free-form per-movie attributes (e.g. "collection_sort"); usable in
|
|
# miniDLNA filename templates via name_context.
|
|
self.attributes: dict[str, str] = {}
|
|
# Identity key of the movie record this file belongs to (set on load /
|
|
# first save); ``None`` only for an index-less, in-memory File.
|
|
self._movie_key: str | None = None
|
|
self.get_metadata()
|
|
|
|
def _identity_from_path(self) -> tuple[str, str, str | None, int | None, str]:
|
|
"""``(key, title, year, tmdb_id, label)`` implied by this file's location.
|
|
|
|
A file inside its own ``Title (rok) [tmdbid]`` folder is one version of
|
|
that movie — identity comes from the **folder**, and the rest of its own
|
|
stem is the version label. A flat file carries both in its own name.
|
|
"""
|
|
from .movie import identity_key
|
|
from .naming import (
|
|
is_canonical_pool_stem,
|
|
parse_3d_format,
|
|
parse_pool_stem,
|
|
parse_tmdb_id,
|
|
)
|
|
|
|
parent = self.file_path.parent.name
|
|
grouped = is_canonical_pool_stem(parent)
|
|
identity_stem = parent if grouped else self.file_path.stem
|
|
parsed = parse_pool_stem(identity_stem)
|
|
year = str(parsed[1]) if parsed else None
|
|
tmdb_id = parse_tmdb_id(identity_stem)
|
|
title = self.title or (parsed[0] if parsed else self.file_path.stem)
|
|
|
|
label = ""
|
|
part = (
|
|
re.fullmatch(
|
|
rf"{re.escape(parent)}[\s._-]*(?:part|pt|cd|dvd|disc|disk)[\s._-]*(\d{{1,2}})",
|
|
self.file_path.stem,
|
|
re.IGNORECASE,
|
|
)
|
|
if grouped
|
|
else None
|
|
)
|
|
if part:
|
|
# one part of a film split across files: "<folder>-part2"
|
|
label = f"part{int(part.group(1))}"
|
|
elif grouped and self.file_path.stem.startswith(f"{parent} - "):
|
|
label = self.file_path.stem[len(parent) + 3 :]
|
|
elif parse_3d_format(self.file_path.stem):
|
|
label = parse_3d_format(self.file_path.stem) or ""
|
|
return identity_key(tmdb_id, title, year), title, year, tmdb_id, label
|
|
|
|
def get_metadata(self) -> None:
|
|
"""Load this file's metadata from the movie record that owns it.
|
|
|
|
Since the pool index became identity-keyed (see :mod:`src.core.movie`),
|
|
a File is a *view* on one media entry of a movie record — every version
|
|
of the same film therefore reads the same title/tags/ČSFD/… A file the
|
|
index doesn't know yet joins the movie of the same identity when there
|
|
is one (that's how a newly added version inherits the shared metadata
|
|
instead of blanking it), otherwise it starts a new record.
|
|
"""
|
|
if self.index is None:
|
|
self._init_new_metadata()
|
|
return
|
|
|
|
found = self.index.find_movie_by_media(self.file_path)
|
|
if found is not None:
|
|
self._movie_key, record = found
|
|
self._apply_record(record)
|
|
return
|
|
|
|
self._init_new_metadata()
|
|
key, *_ = self._identity_from_path()
|
|
existing = self.index.get_movie(key)
|
|
if existing is not None:
|
|
self._movie_key = key
|
|
self._apply_record(existing)
|
|
self.save_metadata()
|
|
|
|
def _init_new_metadata(self) -> None:
|
|
self.new = True
|
|
self.ignored = False
|
|
self.tags = []
|
|
self.date = None
|
|
self.title = None
|
|
self.csfd_url = None
|
|
self.meta = None
|
|
self.auto_tag_paths = set()
|
|
self.attributes = {}
|
|
self.added = None
|
|
|
|
def _build_record(self) -> dict:
|
|
data = {
|
|
"new": self.new,
|
|
"ignored": self.ignored,
|
|
# ukládáme full_path tagů
|
|
"tags": [
|
|
tag.full_path if isinstance(tag, Tag) else tag for tag in self.tags
|
|
],
|
|
# which of the tags were derived automatically (from the NFO)
|
|
"auto_tags": sorted(self.auto_tag_paths),
|
|
# date může být None
|
|
"date": self.date,
|
|
"title": self.title,
|
|
"csfd_url": self.csfd_url,
|
|
"attributes": self.attributes,
|
|
"added": self.added,
|
|
}
|
|
if self.meta is not None:
|
|
data["meta"] = dict(self.meta)
|
|
return data
|
|
|
|
def _apply_record(self, data: dict) -> None:
|
|
self.new = data.get("new", True)
|
|
self.ignored = data.get("ignored", False)
|
|
self.tags = []
|
|
self.date = data.get("date", None)
|
|
self.title = data.get("title", None)
|
|
self.csfd_url = data.get("csfd_url", None)
|
|
# provenance of auto (NFO-derived) tags; legacy key "csfd_tags" honored
|
|
self.auto_tag_paths = set(data.get("auto_tags", data.get("csfd_tags", [])))
|
|
self.attributes = dict(data.get("attributes", {}))
|
|
self.added = data.get("added", None)
|
|
raw_meta = data.get("meta")
|
|
if raw_meta is None: # legacy: pull year/rating out of the old csfd cache
|
|
old = data.get("csfd_cache") or {}
|
|
raw_meta = (
|
|
{"year": old.get("year"), "rating": old.get("rating")} if old else None
|
|
)
|
|
self.meta = dict(raw_meta) if raw_meta else None
|
|
|
|
if not self.tagmanager:
|
|
return
|
|
|
|
for tag_str in data.get("tags", []):
|
|
if "/" in tag_str:
|
|
category, name = tag_str.split("/", 1)
|
|
tag = self.tagmanager.add_tag(category, name)
|
|
self.tags.append(tag)
|
|
|
|
def save_metadata(self):
|
|
"""Persist metadata into this file's movie record (no-op without index).
|
|
|
|
Writes the shared fields onto the movie record and makes sure the
|
|
record lists this file among its media. A file the index has never seen
|
|
starts a new record under its own identity key.
|
|
"""
|
|
if self.index is None:
|
|
return
|
|
from .movie import ROLE_VIDEO
|
|
|
|
derived_key, title, year, tmdb_id, label = self._identity_from_path()
|
|
key = self._movie_key or derived_key
|
|
record = self.index.get_movie(key) or {}
|
|
record.update(self._build_record())
|
|
# The file's actual location is the truth — a move must update this.
|
|
record["library"] = self._library_name()
|
|
record["title"] = title
|
|
record["year"] = year
|
|
record["tmdb_id"] = tmdb_id
|
|
|
|
# Keep this file listed exactly once, with the label its name implies.
|
|
rel = self.index._key(self.file_path)
|
|
media = [m for m in record.get("media", []) if m.get("file") != rel]
|
|
media.append({"role": ROLE_VIDEO, "label": label, "file": rel})
|
|
record["media"] = media
|
|
|
|
self.index.set_movie(key, record)
|
|
self._movie_key = key
|
|
|
|
def _library_name(self) -> str:
|
|
"""Top-level pool folder this file sits in (its library), or ``""``."""
|
|
if self.index is None:
|
|
return ""
|
|
rel = self.index._key(self.file_path)
|
|
parts = Path(rel).parts
|
|
return parts[0] if len(parts) > 1 else ""
|
|
|
|
def added_timestamp(self) -> float:
|
|
"""Epoch seconds for 'date added' — the stored ``added`` or the file mtime.
|
|
|
|
Used to order the "recently added" folder (newer = larger). The fallback
|
|
is **mtime**, not ctime: creating the Filmotéka hardlinks bumps the source
|
|
inode's ctime, which would make ctime-based ordering meaningless, whereas
|
|
mtime (preserved from the source on import) stays stable.
|
|
"""
|
|
if self.added:
|
|
try:
|
|
from datetime import datetime
|
|
|
|
return datetime.fromisoformat(self.added).timestamp()
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
return self.file_path.stat().st_mtime
|
|
except OSError:
|
|
return 0.0
|
|
|
|
def name_context(self) -> dict:
|
|
"""Fields for a miniDLNA filename template (see config tag schema).
|
|
|
|
Provides title / year / rating / ext / stem / filename from the movie's
|
|
metadata (year & rating come from the NFO cache, year falling back to a
|
|
``Rok`` tag, then the canonical filename). Missing values render empty.
|
|
"""
|
|
cache = self.meta or {}
|
|
year = cache.get("year")
|
|
if year is None:
|
|
for t in self.tags:
|
|
if isinstance(t, Tag) and t.category == "Rok" and t.name.isdigit():
|
|
year = t.name
|
|
break
|
|
if year is None:
|
|
# Last resort: recover it from a canonical "Title (YYYY)" filename.
|
|
from .naming import parse_pool_stem
|
|
|
|
parsed = parse_pool_stem(self.file_path.stem)
|
|
if parsed is not None:
|
|
year = parsed[1]
|
|
rating = cache.get("rating")
|
|
# user attributes first; core fields take precedence over same-named keys
|
|
context = dict(self.attributes)
|
|
context.update(
|
|
{
|
|
"title": self.title or self.file_path.stem,
|
|
"year": "" if year is None else str(year),
|
|
"rating": "" if rating is None else rating,
|
|
"ext": self.file_path.suffix,
|
|
"stem": self.file_path.stem,
|
|
"filename": self.filename,
|
|
}
|
|
)
|
|
return context
|
|
|
|
def set_attribute(self, key: str, value: str | None) -> None:
|
|
"""Set a free-form attribute (e.g. 'collection_sort'); None/'' removes it."""
|
|
key = key.strip()
|
|
if not key:
|
|
return
|
|
if value:
|
|
self.attributes[key] = value
|
|
else:
|
|
self.attributes.pop(key, None)
|
|
self.save_metadata()
|
|
|
|
def save_location(self) -> None:
|
|
"""Persist a rename/move of this file without touching shared metadata.
|
|
|
|
Every version of a film holds its own in-memory copy of the movie
|
|
record's shared fields (title, tags, ČSFD, …). A sibling loaded before
|
|
someone changed those fields would write its stale copy back on a plain
|
|
:meth:`save_metadata`. Re-reading the record first means only this
|
|
file's own media entry — path and popisek — actually changes.
|
|
"""
|
|
if self.index is not None and self.index.find_movie_by_media(self.file_path):
|
|
self.get_metadata()
|
|
self.save_metadata()
|
|
|
|
def delete_metadata(self) -> None:
|
|
"""Drop this file from its movie record; the last one takes the record.
|
|
|
|
Removing one version of a multi-version movie leaves the movie (and its
|
|
metadata) in place — only when its last media entry goes does the whole
|
|
record go with it.
|
|
"""
|
|
if self.index is None:
|
|
return
|
|
found = self.index.find_movie_by_media(self.file_path)
|
|
if found is None:
|
|
return
|
|
key, record = found
|
|
rel = self.index._key(self.file_path)
|
|
record["media"] = [m for m in record.get("media", []) if m.get("file") != rel]
|
|
if record["media"]:
|
|
self.index.set_movie(key, record)
|
|
else:
|
|
self.index.delete_movie(key)
|
|
|
|
def relocate(self, new_path: Path) -> None:
|
|
"""Point this File at a new path, moving its media entry along.
|
|
|
|
The physical file must already have been moved/renamed by the caller;
|
|
this only repoints the movie record's media entry (and the File), so a
|
|
rename never detaches a file from its movie.
|
|
"""
|
|
if self.index is not None:
|
|
found = self.index.find_movie_by_media(self.file_path)
|
|
if found is not None:
|
|
key, record = found
|
|
old_rel = self.index._key(self.file_path)
|
|
new_rel = self.index._key(Path(new_path))
|
|
for media in record.get("media", []):
|
|
if media.get("file") == old_rel:
|
|
media["file"] = new_rel
|
|
self.index.set_movie(key, record)
|
|
self._movie_key = key
|
|
self.file_path = Path(new_path)
|
|
self.filename = self.file_path.name
|
|
|
|
def set_date(self, date_str: str | None):
|
|
"""Nastaví datum (např. '2025-09-25') nebo None pro smazání."""
|
|
if date_str is None or date_str == "":
|
|
self.date = None
|
|
else:
|
|
# neprvádíme složitou validaci zde; očekáváme 'YYYY-MM-DD'
|
|
self.date = date_str
|
|
self.save_metadata()
|
|
|
|
def apply_nfo(self, nfo, schema: list[dict] | None = None) -> dict:
|
|
"""Regenerate the auto (NFO-derived) tags from the movie's NFO.
|
|
|
|
Reads genres/year/countries/rating from ``nfo`` (``nfo.NfoData``) per the
|
|
tag schema's source fields and (re)creates those tags; user-added tags are
|
|
kept. Also caches year/rating (for folder grouping) and fills the title
|
|
when missing. ``nfo`` of ``None`` (no NFO yet) is a no-op.
|
|
|
|
Returns:
|
|
dict with 'success' and 'tags_added'.
|
|
"""
|
|
if nfo is None:
|
|
return {"success": False, "error": "NFO nenalezeno", "tags_added": []}
|
|
|
|
if schema is None:
|
|
from .config import DEFAULT_TAG_SCHEMA
|
|
|
|
schema = DEFAULT_TAG_SCHEMA
|
|
|
|
self.meta = {"year": nfo.year, "rating": nfo.rating}
|
|
|
|
# Drop previous auto (NFO) tags; keep user-added ones.
|
|
self.tags = [t for t in self.tags if t.full_path not in self.auto_tag_paths]
|
|
self.auto_tag_paths = set()
|
|
|
|
tags_added: list[str] = []
|
|
for entry in schema:
|
|
field = entry.get("csfd_field") # schema key = source field name
|
|
if not field:
|
|
continue # user-only category
|
|
category = entry["category"]
|
|
for value in nfo.field_values(field):
|
|
tag_obj = (
|
|
self.tagmanager.add_tag(category, value)
|
|
if self.tagmanager
|
|
else Tag(category, value)
|
|
)
|
|
if tag_obj not in self.tags:
|
|
self.tags.append(tag_obj)
|
|
tags_added.append(tag_obj.full_path)
|
|
self.auto_tag_paths.add(tag_obj.full_path)
|
|
|
|
if nfo.title and not self.title:
|
|
self.title = nfo.title
|
|
|
|
self.save_metadata()
|
|
return {"success": True, "tags_added": tags_added}
|
|
|
|
def add_tag(self, tag):
|
|
# tag může být Tag nebo string
|
|
from .tag import Tag as TagClass
|
|
|
|
if isinstance(tag, str):
|
|
if "/" in tag and self.tagmanager:
|
|
category, name = tag.split("/", 1)
|
|
tag_obj = self.tagmanager.add_tag(category, name)
|
|
else:
|
|
tag_obj = TagClass("default", tag)
|
|
elif isinstance(tag, TagClass):
|
|
tag_obj = tag
|
|
else:
|
|
return
|
|
if tag_obj not in self.tags:
|
|
self.tags.append(tag_obj)
|
|
self.save_metadata()
|
|
|
|
def remove_tag(self, tag):
|
|
# tag může být Tag nebo string (full_path)
|
|
if isinstance(tag, str):
|
|
if "/" in tag:
|
|
category, name = tag.split("/", 1)
|
|
tag_obj = Tag(category, name)
|
|
else:
|
|
tag_obj = Tag("default", tag)
|
|
elif isinstance(tag, Tag):
|
|
tag_obj = tag
|
|
else:
|
|
return
|
|
if tag_obj in self.tags:
|
|
self.tags.remove(tag_obj)
|
|
self.save_metadata()
|