Add podtx rename --from-title to fix missing episode numbers

README.md

@@ -59,6 +59,7 @@ podtx transcribe https://example.com/audio/ep01.mp3 --engine whisper
 | `podtx sync [feed]` | Transcribe new episodes |
 | `podtx transcribe <target>` | One-shot RSS / URL / file |
 | `podtx format <json\|--feed\|--all>` | Re-format existing transcript JSON (no ASR) |
+| `podtx rename --from-title --feed\|--all` | Fix `_000_` filenames from title episode numbers |
 
 ### Useful flags
 
@@ -86,6 +87,15 @@ podtx format --feed corecursive-coding-stories --readable --cleanup
 podtx format --all --readable --cleanup
 ```
 
+Rename already-transcribed files whose episode number was missing (`_000_`)
+when the title embeds a clear number (same rules as filename inference above):
+
+```bash
+podtx rename --from-title --feed syntax-tasty-web-development-treats --dry-run
+podtx rename --from-title --feed syntax-tasty-web-development-treats
+podtx rename --from-title --all
+```
+
 ## Data & config
 
 - Data (SQLite state, transcripts, temp audio): `~/.local/share/podcast-transcriber/`

src/podtx/cli.py

@@ -22,6 +22,7 @@
     reformat_many,
     reformat_transcript,
 )
+from podtx.rename_cmd import rename_many_from_title
 from podtx.rss import FeedParseError, parse_feed, suggest_unique_slug
 
 app = typer.Typer(
@@ -554,4 +555,88 @@ def format_cmd(
         )
 
     if result.failed:
-        raise typer.Exit(1)
+        raise typer.Exit(1)
+
+
+@app.command("rename")
+def rename_cmd(
+    from_title: bool = typer.Option(
+        False,
+        "--from-title",
+        help="Infer episode numbers from JSON titles and rename sibling outputs",
+    ),
+    feed: Optional[str] = typer.Option(
+        None,
+        "--feed",
+        help="Rename transcripts for a feed slug",
+    ),
+    all_feeds: bool = typer.Option(
+        False,
+        "--all",
+        help="Rename transcripts across the whole library",
+    ),
+    dry_run: bool = typer.Option(
+        False,
+        "--dry-run",
+        help="Print planned renames without writing",
+    ),
+    data_dir: Optional[Path] = typer.Option(
+        None, "--data-dir", help="Override data directory"
+    ),
+    quiet: bool = typer.Option(False, "--quiet", "-q"),
+) -> None:
+    """Rename transcript outputs to fix missing episode numbers (no ASR).
+
+    Currently supports ``--from-title`` with ``--feed`` or ``--all``.
+    """
+    if not from_title:
+        err_console.print("[red]Specify --from-title[/red] (only rename mode supported)")
+        raise typer.Exit(1)
+
+    if sum([feed is not None, all_feeds]) != 1:
+        err_console.print("[red]Specify exactly one of:[/red] `--feed <slug>` or `--all`")
+        raise typer.Exit(1)
+
+    settings = load_settings(data_dir=data_dir)
+    ensure_data_dirs(settings)
+    transcripts_root = settings.transcripts_dir()
+    try:
+        targets = discover_transcript_jsons(
+            transcripts_root,
+            feed=None if all_feeds else feed,
+        )
+    except TranscriptJsonError as exc:
+        err_console.print(f"[red]{exc}[/red]")
+        raise typer.Exit(1) from exc
+
+    if not targets:
+        err_console.print("[dim]No transcript JSON files found.[/dim]")
+        raise typer.Exit(1)
+
+    if not quiet:
+        scope = "all feeds" if all_feeds else f"feed {feed}"
+        mode = "Dry-run rename" if dry_run else "Renaming"
+        console.print(f"[bold]{mode} {len(targets)} transcript(s)[/bold] ({scope})")
+
+    db = Database(settings.state_db_path())
+    try:
+        result = rename_many_from_title(targets, dry_run=dry_run, db=None if dry_run else db)
+    finally:
+        db.close()
+
+    if not quiet:
+        for old, new in result.renames:
+            verb = "Would rename" if dry_run else "Renamed"
+            console.print(f"[green]{verb}[/green] {old.name} → {new.name}")
+        for path, message in result.skips:
+            console.print(f"[dim]Skipped[/dim] {path.name}: {message}")
+        for path, message in result.errors:
+            err_console.print(f"[red]Failed[/red] {path.name}: {message}")
+        prefix = "Dry-run done" if dry_run else "Done"
+        console.print(
+            f"[bold]{prefix}[/bold]: {result.ok} ok, "
+            f"{result.skipped} skipped, {result.failed} failed"
+        )
+
+    if result.failed:
+        raise typer.Exit(1)

src/podtx/db.py

@@ -186,6 +186,32 @@ def mark_done(
         )
         self._conn.commit()
 
+    def update_episode_paths(
+        self,
+        *,
+        feed_id: int,
+        guid: str,
+        episode_num: int,
+        output_paths: list[Path],
+    ) -> bool:
+        """Update episode number and output paths after a rename (no status change)."""
+        cur = self._conn.execute(
+            """
+            UPDATE episodes
+            SET episode_num = ?,
+                output_paths_json = ?
+            WHERE feed_id = ? AND guid = ?
+            """,
+            (
+                episode_num,
+                json.dumps([str(p) for p in output_paths]),
+                feed_id,
+                guid,
+            ),
+        )
+        self._conn.commit()
+        return cur.rowcount > 0
+
     def mark_error(self, *, feed_id: int, guid: str, message: str) -> None:
         self._conn.execute(
             """

src/podtx/rename_cmd.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field, replace
+from pathlib import Path
+
+from podtx.db import Database
+from podtx.format_cmd import TranscriptJsonError, load_transcript_json
+from podtx.models import Episode
+from podtx.naming import parse_episode_number_from_title, transcript_basename
+
+# Sibling outputs that share a transcript basename.
+_SIBLING_EXTS = (".json", ".txt", ".srt", ".vtt")
+
+
+class RenameError(ValueError):
+    pass
+
+
+@dataclass
+class RenameAction:
+    old_json: Path
+    new_basename: str
+    episode_num: int
+    moves: list[tuple[Path, Path]]
+
+
+@dataclass
+class BatchRenameResult:
+    ok: int = 0
+    skipped: int = 0
+    failed: int = 0
+    renames: list[tuple[Path, Path]] = field(default_factory=list)
+    skips: list[tuple[Path, str]] = field(default_factory=list)
+    errors: list[tuple[Path, str]] = field(default_factory=list)
+
+
+def resolve_episode_number_for_rename(episode: Episode) -> int | None:
+    """Prefer a positive JSON/RSS episode number; else parse a clear title number."""
+    if episode.episode_num is not None and episode.episode_num > 0:
+        return episode.episode_num
+    parsed = parse_episode_number_from_title(episode.title)
+    if parsed is not None and parsed > 0:
+        return parsed
+    return None
+
+
+def plan_rename_from_title(json_path: Path) -> RenameAction | None:
+    """Build a rename plan from JSON title/episode metadata.
+
+    Returns ``None`` when no rename is needed (already correct, or no clear number).
+    Raises ``RenameError`` if a target basename already exists.
+    """
+    path = json_path.expanduser()
+    episode, _ = load_transcript_json(path)
+    num = resolve_episode_number_for_rename(episode)
+    if num is None:
+        return None
+
+    updated = replace(episode, episode_num=num)
+    new_basename = transcript_basename(updated)
+    old_basename = path.stem
+    if new_basename == old_basename:
+        return None
+
+    parent = path.parent
+    moves: list[tuple[Path, Path]] = []
+    for ext in _SIBLING_EXTS:
+        src = parent / f"{old_basename}{ext}"
+        if not src.is_file():
+            continue
+        dest = parent / f"{new_basename}{ext}"
+        if dest.exists():
+            raise RenameError(f"Target already exists: {dest}")
+        moves.append((src, dest))
+
+    if not moves:
+        return None
+
+    return RenameAction(
+        old_json=path,
+        new_basename=new_basename,
+        episode_num=num,
+        moves=moves,
+    )
+
+
+def apply_rename(
+    action: RenameAction,
+    *,
+    dry_run: bool = False,
+    db: Database | None = None,
+) -> Path:
+    """Apply a rename plan. Returns the new JSON path (even on dry-run)."""
+    new_json = action.old_json.parent / f"{action.new_basename}.json"
+    if dry_run:
+        return new_json
+
+    for src, dest in action.moves:
+        src.rename(dest)
+
+    payload = json.loads(new_json.read_text(encoding="utf-8"))
+    payload["episode"] = action.episode_num
+    new_json.write_text(
+        json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
+        encoding="utf-8",
+    )
+
+    if db is not None:
+        guid = str(payload.get("guid") or "")
+        feed = db.get_feed(new_json.parent.name)
+        if feed is not None and guid:
+            db.update_episode_paths(
+                feed_id=feed.id,
+                guid=guid,
+                episode_num=action.episode_num,
+                output_paths=[dest for _, dest in action.moves],
+            )
+
+    return new_json
+
+
+def rename_many_from_title(
+    json_paths: list[Path],
+    *,
+    dry_run: bool = False,
+    db: Database | None = None,
+) -> BatchRenameResult:
+    """Rename many transcript JSON trees; continue on per-file skip/error."""
+    result = BatchRenameResult()
+    for path in json_paths:
+        try:
+            plan = plan_rename_from_title(path)
+        except (TranscriptJsonError, RenameError, OSError, ValueError) as exc:
+            result.failed += 1
+            result.errors.append((path, str(exc)))
+            continue
+
+        if plan is None:
+            result.skipped += 1
+            result.skips.append((path, "no rename needed"))
+            continue
+
+        try:
+            new_json = apply_rename(plan, dry_run=dry_run, db=db)
+        except (OSError, ValueError) as exc:
+            result.failed += 1
+            result.errors.append((path, str(exc)))
+            continue
+
+        result.ok += 1
+        result.renames.append((path, new_json))
+    return result

tests/test_format_cmd.py

@@ -216,6 +216,20 @@ def test_cli_format_all(tmp_path: Path) -> None:
     assert "2" in result.stdout or "Wrote" in result.stdout
 
 
+def test_cli_format_feed_reports_failures(tmp_path: Path) -> None:
+    root = _library(tmp_path)
+    bad = root / "feed-a" / "broken.json"
+    bad.write_text("{not-json", encoding="utf-8")
+    result = runner.invoke(
+        app,
+        ["format", "--feed", "feed-a", "--data-dir", str(tmp_path), "--cleanup"],
+    )
+    assert result.exit_code != 0
+    out = result.stdout + result.stderr
+    assert "Failed" in out
+    assert "failed" in out
+
+
 def test_cli_format_requires_target(tmp_path: Path) -> None:
     result = runner.invoke(app, ["format", "--cleanup"])
     assert result.exit_code != 0