diff --git a/apps/media_library/metadata.py b/apps/media_library/metadata.py
index 30949f1c..1642f8b5 100644
--- a/apps/media_library/metadata.py
+++ b/apps/media_library/metadata.py
@@ -19,6 +19,7 @@ METADATA_CACHE_TIMEOUT = 60 * 60 * 6 # 6 hours
TMDB_VALIDATION_CACHE_SUCCESS_TTL = 60 * 30 # 30 minutes
TMDB_VALIDATION_CACHE_FAILURE_TTL = 60 * 5 # 5 minutes
MOVIEDB_SEARCH_URL = "https://movie-db.org/api.php"
+MOVIEDB_EPISODE_URL = "https://movie-db.org/episode.php"
MOVIEDB_HEALTH_CACHE_KEY = "movie-db:health"
MOVIEDB_HEALTH_SUCCESS_TTL = 60 * 15 # 15 minutes
MOVIEDB_HEALTH_FAILURE_TTL = 60 * 2 # 2 minutes
@@ -30,6 +31,9 @@ _REQUESTS_SESSION.mount(
tmdb.REQUESTS_SESSION = _REQUESTS_SESSION
+_MOVIEDB_SEARCH_CACHE: dict[tuple[str, str], tuple[dict | None, str | None]] = {}
+
+
def validate_tmdb_api_key(api_key: str, *, use_cache: bool = True) -> tuple[bool, str | None]:
"""
Validate a TMDB API key by calling the configuration endpoint.
@@ -329,29 +333,106 @@ def _parse_release_year(value: Optional[str]) -> Optional[int]:
return None
-def fetch_movie_db_metadata(media_item: MediaItem) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
- if media_item.item_type not in {
- MediaItem.TYPE_MOVIE,
- MediaItem.TYPE_SHOW,
- MediaItem.TYPE_EPISODE,
- }:
- return None, "Movie-DB fallback does not support this media type."
+def _php_value_from_string(value: str):
+ lowered = value.lower()
+ if lowered == "true":
+ return True
+ if lowered == "false":
+ return False
+ if lowered in {"null", ""}:
+ return None
+ try:
+ if "." in value:
+ return float(value)
+ return int(value)
+ except ValueError:
+ return value
+
+
+def _convert_numeric_container(value):
+ if isinstance(value, dict):
+ converted = {k: _convert_numeric_container(v) for k, v in value.items()}
+ numeric = True
+ indices: list[int] = []
+ for key in converted.keys():
+ try:
+ indices.append(int(key))
+ except (TypeError, ValueError):
+ numeric = False
+ break
+ if numeric:
+ return [converted[str(idx)] for idx in sorted(indices)]
+ return converted
+ if isinstance(value, list):
+ return [_convert_numeric_container(item) for item in value]
+ return value
+
+
+def _parse_php_array_lines(lines: list[str], idx: int = 0):
+ def parse_body(current_idx: int):
+ result: dict[str, Any] = {}
+ while current_idx < len(lines):
+ token = lines[current_idx]
+ if token == ")":
+ return result, current_idx + 1
+ if token.startswith("[") and "] =>" in token:
+ key_part, value_part = token.split("] =>", 1)
+ key = key_part[1:].strip()
+ value = value_part.strip()
+ if value == "Array":
+ sub_value, next_idx = _parse_php_array_lines(lines, current_idx + 1)
+ result[key] = sub_value
+ current_idx = next_idx
+ continue
+ result[key] = _php_value_from_string(value)
+ current_idx += 1
+ return result, current_idx
+
+ if idx >= len(lines):
+ return {}, idx
+
+ token = lines[idx]
+ if token.lower() == "array":
+ idx += 1
+ if idx < len(lines) and lines[idx] == "(":
+ idx += 1
+ parsed, idx = parse_body(idx)
+ return _convert_numeric_container(parsed), idx
+ return {}, idx
+
+
+def _parse_movie_db_php_response(payload: str) -> Optional[dict | list]:
+ if not payload:
+ return None
+ stripped = payload.replace("", "").replace("", "").strip()
+ if not stripped.lower().startswith("array"):
+ return None
+ lines = [line.strip() for line in stripped.splitlines() if line.strip()]
+ parsed, _ = _parse_php_array_lines(lines, 0)
+ return parsed
+
+
+def _movie_db_search(media_type: str, title: str) -> tuple[Optional[dict], Optional[str]]:
+ normalized_key = (media_type, (title or "").strip().lower())
+ if normalized_key in _MOVIEDB_SEARCH_CACHE:
+ return _MOVIEDB_SEARCH_CACHE[normalized_key]
- media_type = "movie" if media_item.is_movie else "tv"
try:
response = _REQUESTS_SESSION.get(
MOVIEDB_SEARCH_URL,
- params={"type": media_type, "query": media_item.title},
+ params={"type": media_type, "query": title},
timeout=6,
)
except requests.RequestException as exc:
message = f"Movie-DB search failed: {exc}"
logger.warning(message)
+ _MOVIEDB_SEARCH_CACHE[normalized_key] = (None, message)
return None, message
if response.status_code != 200:
message = f"Movie-DB search returned status {response.status_code}."
logger.debug(message)
+ _MOVIEDB_SEARCH_CACHE[normalized_key] = (None, message)
return None, message
try:
@@ -359,6 +440,7 @@ def fetch_movie_db_metadata(media_item: MediaItem) -> Tuple[Optional[Dict[str, A
except ValueError:
message = "Movie-DB search returned an unexpected response format."
logger.debug(message)
+ _MOVIEDB_SEARCH_CACHE[normalized_key] = (None, message)
return None, message
results = []
@@ -370,28 +452,189 @@ def fetch_movie_db_metadata(media_item: MediaItem) -> Tuple[Optional[Dict[str, A
if not results:
message = "Movie-DB search returned no matches."
logger.debug(message)
+ _MOVIEDB_SEARCH_CACHE[normalized_key] = (None, message)
return None, message
candidate = results[0]
- release_date = candidate.get("release_date") or candidate.get("first_air_date")
- release_year = _parse_release_year(release_date) or media_item.release_year
+ _MOVIEDB_SEARCH_CACHE[normalized_key] = (candidate, None)
+ return candidate, None
+
+
+def _movie_db_fetch_episode_details(tv_id: Any, season: int, episode: int) -> tuple[Optional[dict], Optional[str]]:
+ try:
+ response = _REQUESTS_SESSION.get(
+ MOVIEDB_EPISODE_URL,
+ params={"tvId": tv_id, "season": season, "episode": episode},
+ timeout=6,
+ )
+ except requests.RequestException as exc:
+ message = f"Movie-DB episode lookup failed: {exc}"
+ logger.warning(message)
+ return None, message
+
+ if response.status_code != 200:
+ message = f"Movie-DB episode endpoint returned status {response.status_code}."
+ logger.debug(message)
+ return None, message
+
+ parsed = _parse_movie_db_php_response(response.text)
+ if not parsed or not isinstance(parsed, dict):
+ message = "Movie-DB episode endpoint returned an unexpected payload."
+ logger.debug(message)
+ return None, message
+
+ return _convert_numeric_container(parsed), None
+
+
+def fetch_movie_db_metadata(media_item: MediaItem) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
+ if media_item.item_type not in {
+ MediaItem.TYPE_MOVIE,
+ MediaItem.TYPE_SHOW,
+ MediaItem.TYPE_EPISODE,
+ }:
+ return None, "Movie-DB fallback does not support this media type."
+
+ if media_item.item_type == MediaItem.TYPE_MOVIE:
+ candidate, message = _movie_db_search("movie", media_item.title)
+ if not candidate:
+ return None, message
+
+ release_date = candidate.get("release_date") or candidate.get("first_air_date")
+ release_year = _parse_release_year(release_date) or media_item.release_year
+ metadata = _to_serializable(
+ {
+ "tmdb_id": str(candidate.get("id") or media_item.tmdb_id or ""),
+ "title": candidate.get("title")
+ or candidate.get("name")
+ or media_item.title,
+ "synopsis": candidate.get("overview") or media_item.synopsis,
+ "release_year": release_year,
+ "poster": build_image_url(candidate.get("poster_path")),
+ "backdrop": build_image_url(candidate.get("backdrop_path")),
+ "runtime_minutes": candidate.get("runtime"),
+ "genres": [],
+ "cast": [],
+ "crew": [],
+ "source": "movie-db",
+ "raw": candidate,
+ }
+ )
+ return metadata, None
+
+ if media_item.item_type == MediaItem.TYPE_SHOW:
+ candidate, message = _movie_db_search("tv", media_item.title)
+ if not candidate:
+ return None, message
+ tv_id = candidate.get("id")
+ release_date = candidate.get("first_air_date") or candidate.get("release_date")
+ release_year = _parse_release_year(release_date) or media_item.release_year
+ metadata = _to_serializable(
+ {
+ "tmdb_id": str(candidate.get("id") or media_item.tmdb_id or ""),
+ "title": candidate.get("name") or candidate.get("title") or media_item.title,
+ "synopsis": candidate.get("overview") or media_item.synopsis,
+ "release_year": release_year,
+ "poster": build_image_url(candidate.get("poster_path")),
+ "backdrop": build_image_url(candidate.get("backdrop_path")),
+ "runtime_minutes": candidate.get("episode_run_time"),
+ "genres": [],
+ "cast": [],
+ "crew": [],
+ "movie_db_tv_id": str(tv_id) if tv_id is not None else None,
+ "source": "movie-db",
+ "raw": candidate,
+ }
+ )
+ return metadata, None
+
+ # Episode fallback
+ parent_series = media_item.parent or MediaItem.objects.filter(pk=media_item.parent_id).first()
+ if not parent_series:
+ return None, "Unable to determine parent series for episode."
+
+ tv_id = None
+ if isinstance(parent_series.metadata, dict):
+ tv_id = parent_series.metadata.get("movie_db_tv_id")
+
+ if not tv_id:
+ series_candidate, message = _movie_db_search("tv", parent_series.title)
+ if not series_candidate:
+ return None, message
+ tv_id = series_candidate.get("id")
+ if isinstance(parent_series.metadata, dict):
+ updated_metadata = dict(parent_series.metadata)
+ else:
+ updated_metadata = {}
+ updated_metadata["movie_db_tv_id"] = str(tv_id) if tv_id is not None else None
+ parent_series.metadata = updated_metadata
+ parent_series.save(update_fields=["metadata", "updated_at"])
+
+ if tv_id is None:
+ return None, "Movie-DB did not return a tvId for this series."
+
+ if media_item.season_number is None or media_item.episode_number is None:
+ return None, "Episode numbers are required for Movie-DB fallback."
+
+ try:
+ season_number = int(media_item.season_number)
+ episode_number = int(media_item.episode_number)
+ except (TypeError, ValueError):
+ return None, "Episode numbers are required for Movie-DB fallback."
+
+ episode_details, message = _movie_db_fetch_episode_details(tv_id, season_number, episode_number)
+ if not episode_details:
+ return None, message
+
+ release_year = _parse_release_year(episode_details.get("air_date")) or media_item.release_year
+ guest_stars = episode_details.get("guest_stars") or []
+ crew_entries = episode_details.get("crew") or []
+
+ cast: list[dict[str, Any]] = []
+ for star in guest_stars:
+ if not isinstance(star, dict):
+ continue
+ name = star.get("name") or star.get("original_name")
+ if not name:
+ continue
+ cast.append(
+ {
+ "name": name,
+ "character": star.get("character"),
+ "profile_url": build_image_url(star.get("profile_path")),
+ }
+ )
+
+ crew: list[dict[str, Any]] = []
+ for member in crew_entries:
+ if not isinstance(member, dict):
+ continue
+ name = member.get("name") or member.get("original_name")
+ if not name:
+ continue
+ crew.append(
+ {
+ "name": name,
+ "job": member.get("job"),
+ "department": member.get("department"),
+ "profile_url": build_image_url(member.get("profile_path")),
+ }
+ )
metadata = _to_serializable(
{
- "tmdb_id": str(candidate.get("id") or media_item.tmdb_id or ""),
- "title": candidate.get("title")
- or candidate.get("name")
- or media_item.title,
- "synopsis": candidate.get("overview") or media_item.synopsis,
+ "tmdb_id": str(episode_details.get("id") or media_item.tmdb_id or ""),
+ "title": episode_details.get("name") or media_item.title,
+ "synopsis": episode_details.get("overview") or media_item.synopsis,
"release_year": release_year,
- "poster": build_image_url(candidate.get("poster_path")),
- "backdrop": build_image_url(candidate.get("backdrop_path")),
- "runtime_minutes": candidate.get("runtime"),
+ "poster": build_image_url(episode_details.get("still_path")),
+ "backdrop": build_image_url(episode_details.get("still_path")),
+ "runtime_minutes": episode_details.get("runtime"),
"genres": [],
- "cast": [],
- "crew": [],
+ "cast": cast,
+ "crew": crew,
+ "movie_db_tv_id": str(tv_id) if tv_id is not None else None,
"source": "movie-db",
- "raw": candidate,
+ "raw": episode_details,
}
)
diff --git a/apps/media_library/naming.py b/apps/media_library/naming.py
index 21dcf096..7bb7402f 100644
--- a/apps/media_library/naming.py
+++ b/apps/media_library/naming.py
@@ -18,14 +18,30 @@ EPISODE_PATTERN = re.compile(
(?:[\.\-_\s]*e?(?P\d{1,4}))?
)
|
- (?:
- (?P\d{1,4})
- )
+ (?P\d{1,4})
""",
re.IGNORECASE | re.VERBOSE,
)
+def _safe_number(value):
+ if isinstance(value, (list, tuple, set)):
+ for entry in value:
+ normalized = _safe_number(entry)
+ if normalized is not None:
+ return normalized
+ return None
+ if value in (None, "", []):
+ return None
+ try:
+ return int(value)
+ except (TypeError, ValueError):
+ try:
+ return int(float(value))
+ except (TypeError, ValueError):
+ return None
+
+
def _strip_extension(file_name: str) -> str:
base, _ext = os.path.splitext(file_name)
return base
@@ -83,29 +99,40 @@ def classify_media_entry(
series_name = _ensure_series_name_from_path(relative_path, default=guess_data.get("title") or base_name)
if library.library_type == Library.LIBRARY_TYPE_SHOWS:
- season_number = guess_data.get("season")
- episode_number = guess_data.get("episode")
+ season_number = _safe_number(guess_data.get("season"))
+ episode_number = _safe_number(guess_data.get("episode"))
if season_number is None:
season_number = _season_from_segments(segments)
if episode_number is None:
- episode_number = guess_data.get("episode_list", [None])[0] if guess_data.get("episode_list") else None
+ episode_number = _safe_number(guess_data.get("episode_list"))
+
+ episode_list_raw = guess_data.get("episode_list") or []
+ if not episode_list_raw and isinstance(guess_data.get("episode"), (list, tuple)):
+ episode_list_raw = list(guess_data.get("episode") or [])
+ episode_list = []
+ for item in episode_list_raw:
+ number = _safe_number(item)
+ if number is not None and number not in episode_list:
+ episode_list.append(number)
if episode_number is None:
match = EPISODE_PATTERN.search(file_name)
if match:
- episode_number = match.group("episode")
- if episode_number:
- try:
- episode_number = int(episode_number)
- except (TypeError, ValueError):
- episode_number = None
+ episode_number = _safe_number(match.group("episode") or match.group("abs_episode"))
if season_number is None and match.group("season"):
try:
season_number = int(match.group("season"))
except (TypeError, ValueError):
season_number = None
+ if match.group("episode_end"):
+ end_number = _safe_number(match.group("episode_end"))
+ if end_number:
+ if end_number not in episode_list:
+ episode_list.append(end_number)
+ if episode_number is not None and episode_number not in episode_list:
+ episode_list.append(episode_number)
detected_type = MediaItem.TYPE_EPISODE if season_number is not None and episode_number is not None else MediaItem.TYPE_SHOW
normalized_title = normalize_title(series_name or base_name)
@@ -114,6 +141,10 @@ def classify_media_entry(
data["season"] = season_number
if episode_number is not None:
data["episode"] = episode_number
+ if episode_list:
+ episode_list_sorted = sorted({n for n in episode_list if isinstance(n, int)})
+ if episode_list_sorted:
+ data["episode_list"] = episode_list_sorted
if series_name:
data["series"] = series_name
@@ -137,4 +168,3 @@ def classify_media_entry(
year=guess_data.get("year"),
data=data,
)
-