Switched api endpoints

Switched from api.php to episode.php

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("<pre>", "").replace("</pre>", "").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,
         }
     )
 

apps/media_library/naming.py

@@ -18,14 +18,30 @@ EPISODE_PATTERN = re.compile(
         (?:[\.\-_\s]*e?(?P<episode_end>\d{1,4}))?
     )
     |
-    (?:
-        (?P<abs_episode>\d{1,4})
-    )
+    (?P<abs_episode>\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,
     )
-