update screenshot

MPV/lyric.py

@@ -60,6 +60,7 @@ _LYRIC_VISIBLE_PROP = "user-data/medeia-lyric-visible"
 # to a store via the store DB.
 _ITEM_STORE_PROP = "user-data/medeia-item-store"
 _ITEM_HASH_PROP = "user-data/medeia-item-hash"
+_LEGACY_SUB_TRACK_TITLES = ("medeia-note-sub", "medeia-lyric-sub", "medeia-sub")
 
 # Note: We previously used `osd-overlay`, but some mpv builds return
 # error='invalid parameter' for that command. We now use `show-text`, which is
@@ -540,6 +541,39 @@ def _lyric_duration_ms(idx: int, times: List[float], current_t: float) -> int:
     return 1200
 
 
+def _format_vtt_timestamp(seconds: float) -> str:
+    total_ms = max(0, int(round(float(seconds or 0.0) * 1000.0)))
+    hours = total_ms // 3600000
+    minutes = (total_ms // 60000) % 60
+    secs = (total_ms // 1000) % 60
+    millis = total_ms % 1000
+    return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}"
+
+
+def _lrc_entries_to_vtt_text(entries: List[LrcLine]) -> str:
+    if not entries:
+        return "WEBVTT\n\n"
+
+    lines: List[str] = ["WEBVTT", ""]
+    times = [entry.time_s for entry in entries]
+    for idx, entry in enumerate(entries, start=1):
+        start_s = max(0.0, float(entry.time_s or 0.0))
+        if idx < len(entries):
+            end_s = max(start_s + 0.25, float(times[idx]))
+        else:
+            end_s = start_s + 1.2
+
+        text = str(entry.text or "").replace("\r\n", "\n").replace("\r", "\n")
+        cue_text = text if text.strip() else " "
+
+        lines.append(str(idx))
+        lines.append(f"{_format_vtt_timestamp(start_s)} --> {_format_vtt_timestamp(end_s)}")
+        lines.extend(cue_text.split("\n"))
+        lines.append("")
+
+    return "\n".join(lines)
+
+
 def _unwrap_memory_m3u(text: Optional[str]) -> Optional[str]:
     """Extract the real target URL/path from a memory:// M3U payload."""
     if not isinstance(text, str) or not text.startswith("memory://"):
@@ -596,6 +630,12 @@ def _notes_cache_root() -> Path:
     return root
 
 
+def _generated_sub_root() -> Path:
+    root = Path(tempfile.gettempdir()) / "medeia-mpv-notes"
+    root.mkdir(parents=True, exist_ok=True)
+    return root
+
+
 def _notes_cache_key(store: str, file_hash: str) -> str:
     return hashlib.sha1(
         f"{str(store or '').strip().lower()}:{str(file_hash or '').strip().lower()}".encode(
@@ -790,6 +830,26 @@ def _extract_note_text(notes: Dict[str, str], name: str) -> Optional[str]:
     return text if text.strip() else None
 
 
+def _extract_first_note_text(
+    notes: Dict[str, str],
+    names: List[str],
+    *,
+    predicate: Optional[Any] = None,
+) -> tuple[Optional[str], Optional[str]]:
+    for name in names:
+        candidate = _extract_note_text(notes, name)
+        if not candidate:
+            continue
+        if predicate is not None:
+            try:
+                if not bool(predicate(candidate)):
+                    continue
+            except Exception:
+                continue
+        return name, candidate
+    return None, None
+
+
 def _extract_lrc_from_notes(notes: Dict[str, str]) -> Optional[str]:
     """Return raw LRC text from the note named 'lyric'."""
     return _extract_note_text(notes, "lyric")
@@ -811,18 +871,61 @@ def _looks_like_subtitle_text(text: str) -> bool:
     return False
 
 
-def _extract_sub_from_notes(notes: Dict[str, str]) -> Optional[str]:
-    """Return raw subtitle text from note-backed subtitle/transcript keys."""
+def _extract_sub_from_notes(notes: Dict[str, str]) -> tuple[Optional[str], Optional[str]]:
+    """Return (note_name, subtitle_text) from note-backed subtitle/transcript keys."""
     primary = _extract_note_text(notes, "sub")
     if primary:
-        return primary
-    for note_name in _SUBTITLE_NOTE_ALIASES:
-        candidate = _extract_note_text(notes, note_name)
-        if candidate and _looks_like_subtitle_text(candidate):
-            return candidate
+        return "sub", primary
+    return _extract_first_note_text(
+        notes,
+        list(_SUBTITLE_NOTE_ALIASES),
+        predicate=_looks_like_subtitle_text,
+    )
+
+
+def _display_note_name(note_name: Optional[str]) -> str:
+    text = re.sub(r"\s+", " ", str(note_name or "").replace("_", " ")).strip()
+    if not text:
+        return "subtitle"
+    lowered = text.casefold()
+    if lowered == "lyric":
+        return "lyrics"
+    if lowered == "sub":
+        return "subtitles"
+    return text
+
+
+def _display_media_title(client: MPVIPCClient) -> Optional[str]:
+    for key in ("metadata/by-key/title", "metadata/by-key/Title", "media-title"):
+        try:
+            value = _ipc_get_property(client, key, None)
+        except Exception:
+            value = None
+        if isinstance(value, str):
+            text = re.sub(r"\s+", " ", value).strip()
+            if text:
+                return text
     return None
 
 
+def _generated_subtitle_title(client: MPVIPCClient, *, note_name: Optional[str]) -> str:
+    note_label = _display_note_name(note_name)
+    media_title = _display_media_title(client)
+    if media_title:
+        title = f"{note_label}: {media_title}"
+    else:
+        title = note_label
+    title = re.sub(r"\s+", " ", title).strip()
+    return title[:96] if len(title) > 96 else title
+
+
+def _filename_slug(text: Optional[str], *, default: str) -> str:
+    value = re.sub(r"[^A-Za-z0-9._ -]+", " ", str(text or ""))
+    value = re.sub(r"\s+", "-", value).strip("- ._")
+    value = value[:48]
+    return value or default
+
+
 def _infer_sub_extension(text: str) -> str:
     # Best-effort: mpv generally understands SRT/VTT; choose based on content.
     t = (text or "").lstrip("\ufeff\r\n").lstrip()
@@ -839,39 +942,122 @@ def _infer_sub_extension(text: str) -> str:
     return ".vtt"
 
 
-def _write_temp_sub_file(*, key: str, text: str) -> Path:
+def _write_temp_sub_file(*, key: str, text: str, label: Optional[str] = None) -> Path:
     # Write to a content-addressed temp path so updates force mpv reload.
-    tmp_dir = Path(tempfile.gettempdir()) / "medeia-mpv-notes"
-    tmp_dir.mkdir(parents=True, exist_ok=True)
+    tmp_dir = _generated_sub_root()
 
     ext = _infer_sub_extension(text)
     digest = hashlib.sha1((key + "\n" + (text or "")).encode("utf-8",
                                                              errors="ignore")
                           ).hexdigest()[:16]
-    safe_key = hashlib.sha1((key or "").encode("utf-8",
-                                               errors="ignore")).hexdigest()[:12]
-    path = (tmp_dir / f"sub-{safe_key}-{digest}{ext}").resolve()
+    prefix = _filename_slug(label, default="subtitle")
+    path = (tmp_dir / f"{prefix}-{digest}{ext}").resolve()
     path.write_text(text or "", encoding="utf-8", errors="replace")
     return path
 
 
-def _try_remove_selected_external_sub(client: MPVIPCClient) -> None:
+def _subtitle_track_snapshot(client: MPVIPCClient) -> List[Dict[str, Any]]:
+    raw = _ipc_get_property(client, "track-list", [])
+    return raw if isinstance(raw, list) else []
+
+
+def _track_external_sub_path(track: Dict[str, Any]) -> Optional[Path]:
+    if not isinstance(track, dict):
+        return None
+    for key in ("external-filename", "external_filename", "demux-filename", "demux_filename"):
+        raw = track.get(key)
+        if not isinstance(raw, str):
+            continue
+        text = raw.strip()
+        if not text:
+            continue
+        try:
+            return Path(text).expanduser().resolve()
+        except Exception:
+            return Path(text)
+    return None
+
+
+def _is_medeia_generated_sub_track(track: Dict[str, Any]) -> bool:
+    if not isinstance(track, dict):
+        return False
+    title = str(track.get("title") or "").strip()
+    if title in _LEGACY_SUB_TRACK_TITLES:
+        return True
+    path = _track_external_sub_path(track)
+    if path is None:
+        return False
     try:
-        client.send_command({
-            "command": ["sub-remove"]
-        })
+        path.relative_to(_generated_sub_root().resolve())
+        return True
     except Exception:
+        return False
+
+
+def _find_medeia_sub_track_ids(client: MPVIPCClient) -> List[int]:
+    out: List[int] = []
+    for track in _subtitle_track_snapshot(client):
+        if not isinstance(track, dict):
+            continue
+        if str(track.get("type") or "") != "sub":
+            continue
+        if not _is_medeia_generated_sub_track(track):
+            continue
+        try:
+            track_id = int(track.get("id"))
+        except Exception:
+            continue
+        out.append(track_id)
+    return out
+
+
+def _log_medeia_sub_tracks(client: MPVIPCClient, reason: str) -> None:
+    parts: List[str] = []
+    for track in _subtitle_track_snapshot(client):
+        if not isinstance(track, dict):
+            continue
+        if str(track.get("type") or "") != "sub":
+            continue
+        if not _is_medeia_generated_sub_track(track):
+            continue
+        title = str(track.get("title") or "").strip()
+        source = _track_external_sub_path(track)
+        parts.append(
+            f"id={track.get('id')}"
+            f" title={title!r}"
+            f" selected={bool(track.get('selected'))}"
+            f" external={bool(track.get('external'))}"
+            f" source={source.name if source is not None else '<none>'}"
+        )
+    if parts:
+        _log(f"Medeia subtitle tracks {reason}: " + " | ".join(parts))
+    else:
+        _log(f"Medeia subtitle tracks {reason}: <none>")
+
+
+def _remove_medeia_external_subs(client: MPVIPCClient, *, reason: str = "") -> None:
+    track_ids = _find_medeia_sub_track_ids(client)
+    if not track_ids:
         return
+    _log(f"Removing Medeia subtitle tracks reason={reason or 'unknown'} ids={track_ids}")
+    for track_id in track_ids:
+        try:
+            client.send_command({
+                "command": ["sub-remove", int(track_id)]
+            })
+        except Exception:
+            continue
+    _log_medeia_sub_tracks(client, f"after-remove:{reason or 'unknown'}")
 
 
-def _try_add_external_sub(client: MPVIPCClient, path: Path) -> None:
+def _try_add_external_sub(client: MPVIPCClient, path: Path, *, title: str) -> None:
     try:
         client.send_command(
             {
                 "command": ["sub-add",
                             str(path),
                             "select",
-                            "medeia-sub"]
+                            str(title or _NOTE_SUB_TRACK_TITLE)]
             }
         )
     except Exception:
@@ -1099,7 +1285,7 @@ class _PlaybackState:
     entries: List[LrcLine] = field(default_factory=list)
     times: List[float] = field(default_factory=list)
     loaded_key: Optional[str] = None
-    loaded_mode: Optional[str] = None          # 'lyric' | 'sub' | None
+    loaded_mode: Optional[str] = None          # 'lyric' | 'sub' | 'lyric-sub' | None
     loaded_sub_path: Optional[Path] = None
     last_target: Optional[str] = None
     fetch_attempt_key: Optional[str] = None
@@ -1130,7 +1316,7 @@ class _PlaybackState:
         self.loaded_key = None
         self.loaded_mode = None
         if self.loaded_sub_path is not None:
-            _try_remove_selected_external_sub(client)
+            _remove_medeia_external_subs(client, reason="state-clear")
             self.loaded_sub_path = None
 
 
@@ -1153,6 +1339,7 @@ def run_auto_overlay(
         return 3
 
     _log(f"Auto overlay connected (ipc={getattr(mpv, 'ipc_path', None)})")
+    _remove_medeia_external_subs(client, reason="startup-sweep")
 
     state = _PlaybackState()
     last_idx: Optional[int] = None
@@ -1196,6 +1383,12 @@ def run_auto_overlay(
             if not client.connect():
                 _log("mpv IPC disconnected; exiting MPV.lyric")
                 return 4
+            _remove_medeia_external_subs(client, reason="reconnect-sweep:path")
+            state.clear(client)
+            state.last_target = None
+            last_idx = None
+            last_text = None
+            last_visible = None
             time.sleep(poll_s)
             continue
 
@@ -1207,11 +1400,14 @@ def run_auto_overlay(
             last_visible = visible
         elif last_visible is True and visible is False:
             _osd_clear_and_restore(client)
-            _try_remove_selected_external_sub(client)
+            _remove_medeia_external_subs(client, reason="visibility-off")
+            state.loaded_sub_path = None
             last_idx = None
             last_text = None
             last_visible = visible
         elif last_visible is False and visible is True:
+            if state.loaded_mode in {"sub", "lyric-sub"} and state.loaded_sub_path is None:
+                state.loaded_key = None
             last_idx = None
             last_text = None
             last_visible = visible
@@ -1459,21 +1655,26 @@ def run_auto_overlay(
             except Exception:
                 _log("Loaded notes keys: <error>")
 
-            sub_text = _extract_note_text(notes, "sub")
+            sub_note_name, sub_text = _extract_sub_from_notes(notes)
             if sub_text:
                 # Hand subtitles to mpv's track subsystem; suppress OSD lyric overlay.
                 _osd_clear_and_restore(client)
                 sub_path: Optional[Path] = None
+                sub_title = _generated_subtitle_title(client, note_name=sub_note_name)
                 try:
-                    sub_path = _write_temp_sub_file(key=state.key, text=sub_text)
+                    sub_path = _write_temp_sub_file(key=state.key, text=sub_text, label=sub_title)
                 except Exception as exc:
                     _log(f"Failed to write sub note temp file: {exc}")
 
                 if sub_path is not None:
-                    if state.loaded_sub_path is not None:
-                        _try_remove_selected_external_sub(client)
-                    _try_add_external_sub(client, sub_path)
+                    _remove_medeia_external_subs(client, reason="load-note-sub")
+                    _try_add_external_sub(client, sub_path, title=sub_title)
                     state.loaded_sub_path = sub_path
+                    _log(
+                        f"Loaded note-backed native subtitle track"
+                        f" note={sub_note_name!r} title={sub_title!r} path={sub_path}"
+                    )
+                    _log_medeia_sub_tracks(client, "after-add-note-sub")
 
                 state.entries = []
                 state.times = []
@@ -1481,12 +1682,12 @@ def run_auto_overlay(
                 state.loaded_mode = "sub"
 
             else:
-                # Switching away from sub mode: unload the external subtitle.
-                if state.loaded_mode == "sub" and state.loaded_sub_path is not None:
-                    _try_remove_selected_external_sub(client)
+                # Switching away from native subtitle mode: unload the external subtitle.
+                if state.loaded_sub_path is not None:
+                    _remove_medeia_external_subs(client, reason="switch-away-native-sub")
                     state.loaded_sub_path = None
 
-                lrc_text = _extract_note_text(notes, "lyric")
+                lrc_text = _extract_lrc_from_notes(notes)
                 if not lrc_text:
                     _log("No lyric note found (note name: 'lyric')")
 
@@ -1569,10 +1770,47 @@ def run_auto_overlay(
                 else:
                     _log(f"Loaded lyric note ({len(lrc_text)} chars)")
                     parsed = parse_lrc(lrc_text)
-                    state.entries = parsed
-                    state.times = [e.time_s for e in parsed]
-                    state.loaded_key = state.key
-                    state.loaded_mode = "lyric"
+                    if not parsed:
+                        _log("Lyric note contained no timestamped entries")
+                        _osd_clear_and_restore(client)
+                        state.entries = []
+                        state.times = []
+                        state.loaded_key = state.key
+                        state.loaded_mode = None
+                    else:
+                        lyric_sub_path: Optional[Path] = None
+                        lyric_sub_title = _generated_subtitle_title(client, note_name="lyric")
+                        try:
+                            lyric_sub_text = _lrc_entries_to_vtt_text(parsed)
+                            lyric_sub_path = _write_temp_sub_file(
+                                key=f"{state.key}:lyric",
+                                text=lyric_sub_text,
+                                label=lyric_sub_title,
+                            )
+                        except Exception as exc:
+                            _log(f"Failed to write lyric note temp subtitle: {exc}")
+
+                        if lyric_sub_path is None:
+                            _osd_clear_and_restore(client)
+                            state.entries = []
+                            state.times = []
+                            state.loaded_key = state.key
+                            state.loaded_mode = None
+                        else:
+                            _osd_clear_and_restore(client)
+                            _remove_medeia_external_subs(client, reason="load-lyric-sub")
+                            _try_add_external_sub(client, lyric_sub_path, title=lyric_sub_title)
+                            state.loaded_sub_path = lyric_sub_path
+                            state.entries = []
+                            state.times = []
+                            state.loaded_key = state.key
+                            state.loaded_mode = "lyric-sub"
+                            _log(
+                                f"Loaded lyric note as native subtitle track"
+                                f" title={lyric_sub_title!r} entries={len(parsed)}"
+                                f" path={lyric_sub_path}"
+                            )
+                            _log_medeia_sub_tracks(client, "after-add-lyric-sub")
 
         # ----------------------------------------------------------------
         # 8. Render the current lyric line.
@@ -1590,6 +1828,12 @@ def run_auto_overlay(
             if not client.connect():
                 _log("mpv IPC disconnected; exiting MPV.lyric")
                 return 4
+            _remove_medeia_external_subs(client, reason="reconnect-sweep:time")
+            state.clear(client)
+            state.last_target = None
+            last_idx = None
+            last_text = None
+            last_visible = None
             time.sleep(poll_s)
             continue