feat: videos upload command

.editorconfig

@@ -12,4 +12,4 @@ insert_final_newline = true
 max_line_length = 100
 
 [*.{yml,yaml,toml}]
-indent_size = 2
+indent_size = 2

README.md

@@ -117,6 +117,66 @@ Notes:
 - `liveBroadcasts.insert`/`update` are write operations (~50 quota units each);
   see [API quota](#api-quota) below.
 
+## Uploading videos
+
+Bulk-upload videos from a directory. Each video file needs a YAML sidecar with
+the same basename describing its metadata.
+
+```bash
+yts videos upload ./outbox                    # dry-run preview
+yts videos upload ./outbox --execute          # actually upload
+yts videos upload ./outbox --execute --max 3  # cap to N uploads (quota budget)
+```
+
+Directory layout:
+
+```
+outbox/
+  holiday.mp4
+  holiday.yaml    # required
+  holiday.jpg     # optional thumbnail (or .png), max 2 MB
+```
+
+Sidecar (`holiday.yaml`):
+
+```yaml
+title: Holiday recap 2026
+description: |
+  Multi-line description.
+privacy: private              # private | unlisted | public
+publish_at: 2026-06-01T10:00:00+02:00   # optional; forces privacy=private
+tags: [travel, vlog]
+category_id: "22"             # YouTube category id; default 22 (People & Blogs)
+default_language: nl
+default_audio_language: nl
+made_for_kids: false
+```
+
+After a successful upload the tool patches the sidecar with the resulting
+`video_id` and `uploaded_at`, so re-running on the same directory is safe; only
+sidecars without a `video_id` get uploaded.
+
+Quota: `videos.insert` costs ~1600 quota units, so on the default 10k/day
+quota you can upload ~6 videos per day. Use `--max` to cap a run explicitly.
+
+Show what's scheduled for the future:
+
+```bash
+yts videos list --scheduled
+```
+
+## Development
+
+Clone the repo, sync dev dependencies, and install the pre-commit hook so
+`ruff check` and `ruff format` run on every commit (same checks CI runs):
+
+```bash
+uv sync --group dev
+uv run pre-commit install
+```
+
+Run the suite manually with `uv run pytest` and `uv run pre-commit run --all-files`.
+
 ## API quota
 
 The YouTube Data API enforces a default quota of 10_000 units per project per day. Most read

pyproject.toml

@@ -26,6 +26,8 @@ dependencies = [
     "google-auth-oauthlib>=1.0.0",
     "google-api-python-client>=2.0.0",
     "packaging>=21.0",
+    "pydantic>=2.5",
+    "ruamel.yaml>=0.18",
 ]
 
 [dependency-groups]

src/ytstudio/commands/upload.py

@@ -0,0 +1,155 @@
+from datetime import UTC, datetime
+from pathlib import Path
+
+import typer
+from googleapiclient.errors import HttpError
+from rich.progress import BarColumn, Progress, TextColumn, TimeRemainingColumn
+
+from ytstudio.services import get_data_service
+from ytstudio.ui import console, create_table, dim
+from ytstudio.upload_pipeline import (
+    DiscoveryError,
+    UploadJob,
+    ValidationError,
+    discover,
+    set_thumbnail,
+    upload_video,
+    validate_jobs,
+    write_back,
+)
+
+
+class _QuotaExceeded(Exception):
+    def __init__(self, cause: HttpError) -> None:
+        self.cause = cause
+
+
+def _is_quota_exceeded(e: HttpError) -> bool:
+    return any(detail.get("reason") == "quotaExceeded" for detail in e.error_details or [])
+
+
+def _upload_one(service, job: UploadJob) -> str:
+    file_size = job.video_path.stat().st_size
+    with Progress(
+        TextColumn("[bold blue]{task.fields[name]}"),
+        BarColumn(),
+        TextColumn("{task.percentage:>3.0f}%"),
+        TimeRemainingColumn(),
+        console=console,
+        transient=True,
+    ) as progress:
+        task_id = progress.add_task("upload", total=file_size, name=job.video_path.name)
+
+        def _on_progress(done: int, total: int) -> None:
+            progress.update(task_id, completed=done)
+
+        try:
+            video_id = upload_video(service, job, on_progress=_on_progress)
+        except HttpError as e:
+            if _is_quota_exceeded(e):
+                raise _QuotaExceeded(e) from e
+            raise
+
+    if job.thumbnail_path is not None:
+        try:
+            set_thumbnail(service, video_id=video_id, thumbnail_path=job.thumbnail_path)
+        except HttpError as e:
+            console.print(f"[yellow]thumbnail for {job.video_path.name} failed: {e}[/yellow]")
+
+    # Print success first so the video_id is visible even if write-back fails.
+    console.print(f"[green]ok[/green] {job.video_path.name} -> https://youtu.be/{video_id}")
+
+    try:
+        write_back(
+            job.sidecar_path,
+            video_id=video_id,
+            uploaded_at_iso=datetime.now(UTC).isoformat(),
+        )
+    except Exception as e:
+        console.print(
+            f"[red]write-back failed[/red] for {job.sidecar_path.name}: {e}\n"
+            f"[yellow]Sidecar not patched. Add manually to avoid duplicate upload:[/yellow]\n"
+            f"  video_id: {video_id}"
+        )
+
+    return video_id
+
+
+def upload(
+    path: Path = typer.Argument(..., help="Video file or directory of videos with yaml sidecars"),
+    execute: bool = typer.Option(False, "--execute", help="Actually upload (default is dry-run)"),
+    max_uploads: int = typer.Option(
+        0, "--max", "-m", help="Maximum number of uploads in this run (0 = no limit)"
+    ),
+) -> None:
+    """Upload one or more videos described by yaml sidecars."""
+    try:
+        jobs = discover(path)
+    except DiscoveryError as e:
+        console.print(f"[red]{e}[/red]")
+        raise typer.Exit(1) from e
+
+    try:
+        validate_jobs(jobs)
+    except ValidationError as e:
+        console.print(f"[red]Validation failed:[/red]\n{e}")
+        raise typer.Exit(1) from e
+
+    pending = [j for j in jobs if not j.already_uploaded]
+    skipped = [j for j in jobs if j.already_uploaded]
+
+    if not pending:
+        console.print(dim("Nothing to upload (all sidecars already have video_id)."))
+        return
+
+    table = create_table()
+    table.add_column("File", style="yellow")
+    table.add_column("Title", style="cyan")
+    table.add_column("Privacy")
+    table.add_column("Publish at")
+    table.add_column("Thumb")
+
+    for job in pending:
+        spec = job.spec
+        publish_at = spec.publish_at.isoformat() if spec.publish_at else "-"
+        thumb = "yes" if job.thumbnail_path else "-"
+        table.add_row(job.video_path.name, spec.title, spec.privacy.value, publish_at, thumb)
+
+    console.print(table)
+
+    for job in skipped:
+        console.print(dim(f"skipped (already uploaded): {job.video_path.name}"))
+
+    if not execute:
+        console.print("\n[dim]Dry-run. Re-run with --execute to upload.[/dim]")
+        return
+
+    if max_uploads > 0:
+        pending = pending[:max_uploads]
+
+    service = get_data_service()
+
+    estimated = len(pending) * 1600
+    console.print(dim(f"\nGoing to upload {len(pending)} videos (~{estimated} quota units)."))
+
+    succeeded = 0
+    quota_hit = False
+    for job in pending:
+        try:
+            _upload_one(service, job)
+        except _QuotaExceeded:
+            console.print(
+                f"\n[red]Quota exceeded after {succeeded} upload(s).[/red] "
+                "Daily YouTube API quota exhausted; resets at midnight Pacific Time (PT)."
+            )
+            quota_hit = True
+            break
+        except HttpError as e:
+            console.print(f"[red]X {job.video_path.name}: {e}[/red]")
+            continue
+
+        succeeded += 1
+
+    console.print(f"\n[bold]Done:[/bold] {succeeded}/{len(pending)} uploaded.")
+    if quota_hit:
+        raise typer.Exit(1)

src/ytstudio/commands/videos.py

@@ -1,11 +1,13 @@
 import json
 import re
 from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
 
 import typer
 from googleapiclient.errors import HttpError
 
 from ytstudio.api import api, handle_api_error
+from ytstudio.commands.upload import upload as _upload_cmd
 from ytstudio.services import get_data_service
 from ytstudio.ui import (
     console,
@@ -37,6 +39,7 @@ class Video:
     default_language: str | None = None
     default_audio_language: str | None = None
     localizations: dict = field(default_factory=dict)
+    scheduled_publish_at: str | None = None
 
 
 def format_duration(iso_duration: str) -> str:
@@ -95,6 +98,7 @@ def fetch_video(data_service, video_id: str) -> Video | None:
         default_language=snippet.get("defaultLanguage"),
         default_audio_language=snippet.get("defaultAudioLanguage"),
         localizations=item.get("localizations", {}),
+        scheduled_publish_at=item.get("status", {}).get("publishAt"),
     )
 
 
@@ -163,6 +167,7 @@ def fetch_videos(
                 localizations=data.get("localizations", {}),
                 default_language=snippet.get("defaultLanguage"),
                 default_audio_language=snippet.get("defaultAudioLanguage"),
+                scheduled_publish_at=data.get("status", {}).get("publishAt"),
             )
 
             all_videos.append(video)
@@ -194,12 +199,26 @@ def list_videos(
     has_localization: str = typer.Option(
         None, "--has-localization", help="Filter by available translation (e.g., en, nl)"
     ),
+    scheduled: bool = typer.Option(
+        False, "--scheduled", help="Only show videos scheduled for future publish"
+    ),
 ):
     """List your YouTube videos"""
     service = get_data_service()
     result = fetch_videos(service, limit, page_token)
     videos: list[Video] = result["videos"]
 
+    if scheduled:
+        now = datetime.now(UTC)
+
+        def _is_future(v: Video) -> bool:
+            ts = v.scheduled_publish_at
+            if not ts:
+                return False
+            return datetime.fromisoformat(ts.replace("Z", "+00:00")) > now
+
+        videos = [v for v in videos if v.privacy == "private" and _is_future(v)]
+
     if audio_lang:
         videos = [v for v in videos if v.default_audio_language == audio_lang]
     if meta_lang:
@@ -471,3 +490,6 @@ def search_replace(
             failed += 1
 
     console.print(f"\n[bold]Done:[/bold] {success} updated, {failed} failed")
+
+
+app.command("upload")(_upload_cmd)