main.py

# # uvicorn main:app --host 0.0.0.0 --port 8000
# """
# FastAPI service for the 5-step subtitle pipeline.

# Project storage contract:

# src/
# ├── input/
# │   └── <project_id>/
# │       └── file.<original-extension>
# └── output/
#     └── <project_id>/
#         ├── audios/
#         ├── diarization/
#         ├── segments/
#         ├── transcribe/
#         ├── subtitles/
#         ├── service/
#         │   ├── status.json
#         │   └── pipeline.log
#         └── ...

# API:

# POST /projects
#     Start a new project from a remote HTTP/HTTPS media URL.

# GET /projects/{project_id}
#     Return current persistent project status.

# GET /projects/{project_id}/logs
#     Return recent project log lines.

# POST /projects/{project_id}/resume
#     Resume the failed/interrupted project from the first incomplete step.

# GET /health
#     Health check.

# The processing job is intentionally single-worker inside this process.
# That is important for your CPU-heavy ML pipeline because Step 2/4 should not
# be duplicated into multiple independent OS processes.

# For production across multiple machines/processes, move the job execution to
# a durable queue such as Celery/RQ/Arq with Redis. FastAPI's BackgroundTasks
# are suitable for smaller/simple background work but FastAPI itself documents
# a queue/worker system as preferable for heavy background computation.
# """

# from __future__ import annotations

# import asyncio
# import json
# import logging
# import os
# import re
# import shutil
# import tempfile
# import threading
# import time
# from concurrent.futures import ThreadPoolExecutor
# from datetime import datetime, timezone
# from email.message import Message
# from pathlib import Path
# from typing import Any
# from urllib.parse import urlparse, unquote

# import httpx
# from fastapi import FastAPI, HTTPException, status
# from fastapi.responses import FileResponse
# from pydantic import BaseModel, Field, HttpUrl

# from src.pipeline.step_001_separate import run_audio_separation
# from src.pipeline.step_002_diarize import run_diarization
# from src.pipeline.step_003_segment import run_segment_preparation
# from src.pipeline.step_004_transcribe import run_transcription
# from src.pipeline.step_005_original_subtitle import run_step_5

# # ============================================================================
# # CONSOLE & FILE LOGGING SETUP
# # ============================================================================

# logging.basicConfig(
#     level=logging.INFO,
#     format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
# )
# console_logger = logging.getLogger("subtitle-pipeline-console")

# # ============================================================================
# # CONFIGURATION
# # ============================================================================

# BASE_DIR = Path(__file__).resolve().parent
# SRC_DIR = BASE_DIR / "src"

# INPUT_ROOT = SRC_DIR / "input"
# OUTPUT_ROOT = SRC_DIR / "output"

# MAX_DOWNLOAD_BYTES = int(
#     os.getenv(
#         "MAX_DOWNLOAD_BYTES",
#         str(4 * 1024 * 1024 * 1024),
#     )
# )

# DOWNLOAD_CHUNK_BYTES = 1024 * 1024  # 1 MiB
# HTTP_TIMEOUT_SECONDS = float(
#     os.getenv(
#         "HTTP_TIMEOUT_SECONDS",
#         "120",
#     )
# )

# MAX_LOG_LINES = int(
#     os.getenv(
#         "MAX_LOG_LINES",
#         "5000",
#     )
# )

# MAX_STATUS_LOG_LINES = int(
#     os.getenv(
#         "MAX_STATUS_LOG_LINES",
#         "100",
#     )
# )

# MAX_CONCURRENT_PIPELINES = 1

# SUPPORTED_EXTENSIONS = {
#     ".mp3", ".wav", ".flac", ".m4a", ".aac", ".ogg", ".opus", ".wma",
#     ".mp4", ".mkv", ".mov", ".avi", ".webm", ".m4v", ".mpeg", ".mpg",
#     ".ts", ".mts", ".m2ts", ".3gp",
# }

# EXTENSION_BY_CONTENT_TYPE = {
#     "audio/mpeg": ".mp3",
#     "audio/wav": ".wav",
#     "audio/x-wav": ".wav",
#     "audio/flac": ".flac",
#     "audio/x-flac": ".flac",
#     "audio/mp4": ".m4a",
#     "audio/aac": ".aac",
#     "audio/ogg": ".ogg",
#     "audio/opus": ".opus",
#     "video/mp4": ".mp4",
#     "video/quicktime": ".mov",
#     "video/x-matroska": ".mkv",
#     "video/webm": ".webm",
#     "video/mpeg": ".mpg",
# }

# MEDIA_MAGIC_FALLBACK_EXTENSION = ".bin"

# STEPS = [
#     {"number": 1, "name": "audio_separation"},
#     {"number": 2, "name": "speaker_diarization"},
#     {"number": 3, "name": "speaker_segment_preparation"},
#     {"number": 4, "name": "speaker_aware_transcription"},
#     {"number": 5, "name": "original_subtitle_generation"},
# ]

# # ============================================================================
# # FASTAPI APP
# # ============================================================================

# app = FastAPI(
#     title="Subtitle Pipeline Service",
#     version="1.0.0",
#     description=(
#         "Project-based FastAPI orchestration for audio separation, "
#         "speaker diarization, segmentation, transcription, and subtitles."
#     ),
# )

# # ============================================================================
# # GLOBAL JOB EXECUTOR
# # ============================================================================

# PIPELINE_EXECUTOR = ThreadPoolExecutor(
#     max_workers=MAX_CONCURRENT_PIPELINES,
#     thread_name_prefix="subtitle-pipeline",
# )

# ACTIVE_PROJECTS: set[str] = set()
# ACTIVE_PROJECTS_LOCK = threading.Lock()
# PROJECT_EVENTS: dict[str, asyncio.Event] = {}
# PROJECT_EVENTS_LOCK = threading.Lock()

# # ============================================================================
# # REQUEST MODELS
# # ============================================================================

# class CreateProjectRequest(BaseModel):
#     project_id: str = Field(
#         ...,
#         min_length=1,
#         max_length=128,
#         description="Directory-safe project identifier.",
#     )
#     file_url: HttpUrl = Field(
#         ...,
#         description="Public/signed HTTP(S) URL of the source media file.",
#     )

# class ResumeProjectResponse(BaseModel):
#     project_id: str
#     status: str
#     resumed_from_step: int | None
#     message: str

# # ============================================================================
# # LOGGING / PERSISTENT STATE
# # ============================================================================

# class ProjectLogger:
#     def __init__(self, project_id: str) -> None:
#         self.project_id = project_id
#         self.project_dir = OUTPUT_ROOT / project_id
#         self.service_dir = self.project_dir / "service"
#         self.log_path = self.service_dir / "pipeline.log"
#         self.status_path = self.service_dir / "status.json"

#         self.service_dir.mkdir(parents=True, exist_ok=True)
#         self._lock = threading.Lock()

#         self.logger = logging.getLogger(f"subtitle-project.{project_id}")
#         self.logger.setLevel(logging.INFO)
#         self.logger.propagate = False

#         if not self.logger.handlers:
#             handler = logging.FileHandler(self.log_path, encoding="utf-8")
#             handler.setFormatter(
#                 logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
#             )
#             self.logger.addHandler(handler)

#     def write(self, message: str, level: int = logging.INFO) -> None:
#         with self._lock:
#             self.logger.log(level, message)
#             # Detailed Console Output with Timestamp
#             timestamp = datetime.now(timezone.utc).isoformat()
#             console_msg = f"[{timestamp}] [Project: {self.project_id}] {message}"
#             if level == logging.ERROR:
#                 console_logger.error(console_msg)
#             elif level == logging.WARNING:
#                 console_logger.warning(console_msg)
#             else:
#                 console_logger.info(console_msg)

#     def recent_lines(self, limit: int = MAX_STATUS_LOG_LINES) -> list[str]:
#         if not self.log_path.exists():
#             return []
#         with self.log_path.open("r", encoding="utf-8", errors="replace") as handle:
#             lines = handle.readlines()
#         return lines[-limit:]

# # ============================================================================
# # PROJECT STATE UTILITIES
# # ============================================================================

# def utc_now() -> str:
#     return datetime.now(timezone.utc).isoformat()

# def _project_state_path(project_id: str) -> Path:
#     return OUTPUT_ROOT / project_id / "service" / "status.json"

# def _default_step_state() -> dict[str, Any]:
#     return {
#         "status": "pending",
#         "started_at": None,
#         "completed_at": None,
#         "error": None,
#         "result": None,
#     }

# def _initial_state(project_id: str, source_url: str) -> dict[str, Any]:
#     return {
#         "schema_version": "1.0",
#         "project_id": project_id,
#         "source_url": source_url,
#         "source_file": None,
#         "status": "queued",
#         "current_step": None,
#         "current_step_name": None,
#         "last_completed_step": 0,
#         "created_at": utc_now(),
#         "updated_at": utc_now(),
#         "completed_at": None,
#         "error": None,
#         "steps": {
#             str(step["number"]): {**step, **_default_step_state()}
#             for step in STEPS
#         },
#         "logs": [],
#     }

# def _read_state(project_id: str) -> dict[str, Any] | None:
#     path = _project_state_path(project_id)
#     if not path.exists():
#         return None
#     try:
#         with path.open("r", encoding="utf-8") as handle:
#             return json.load(handle)
#     except json.JSONDecodeError:
#         return None

# def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
#     path.parent.mkdir(parents=True, exist_ok=True)
#     temp = path.with_name(f".{path.name}.tmp-{os.getpid()}")
#     try:
#         with temp.open("w", encoding="utf-8") as handle:
#             json.dump(payload, handle, indent=2, ensure_ascii=False)
#             handle.write("\n")
#         os.replace(temp, path)
#     finally:
#         temp.unlink(missing_ok=True)

# def _write_state(state: dict[str, Any]) -> None:
#     state["updated_at"] = utc_now()
#     state["logs"] = state.get("logs", [])[-MAX_STATUS_LOG_LINES:]
#     _atomic_write_json(_project_state_path(state["project_id"]), state)

#     with PROJECT_EVENTS_LOCK:
#         event = PROJECT_EVENTS.get(state["project_id"])
#     if event:
#         event.set()

# def _record_event(
#     state: dict[str, Any],
#     logger: ProjectLogger,
#     message: str,
#     *,
#     level: int = logging.INFO,
# ) -> None:
#     logger.write(message, level=level)
#     logs = state.setdefault("logs", [])
#     logs.append({"timestamp": utc_now(), "message": message})
#     state["updated_at"] = utc_now()
#     _write_state(state)

# # ============================================================================
# # PROJECT ID / PATH SAFETY
# # ============================================================================

# PROJECT_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")

# def validate_project_id(project_id: str) -> str:
#     value = str(project_id).strip()
#     if not PROJECT_ID_PATTERN.fullmatch(value):
#         raise HTTPException(
#             status_code=status.HTTP_400_BAD_REQUEST,
#             detail=(
#                 "project_id must contain only letters, numbers, "
#                 "dot, underscore, and hyphen, and must not exceed 128 characters."
#             ),
#         )
#     return value

# def project_paths(project_id: str) -> dict[str, Path]:
#     project_id = validate_project_id(project_id)
#     root = OUTPUT_ROOT / project_id
#     return {
#         "project_dir": root,
#         "input_dir": INPUT_ROOT / project_id,
#         "output_dir": root,
#         "audios": root / "audios",
#         "diarization": root / "diarization",
#         "segments": root / "segments",
#         "transcribe": root / "transcribe",
#         "subtitles": root / "subtitles",
#         "service": root / "service",
#     }

# # ============================================================================
# # URL DOWNLOAD
# # ============================================================================

# def _extension_from_url(url: str) -> str | None:
#     parsed = urlparse(url)
#     name = unquote(Path(parsed.path).name)
#     suffix = Path(name).suffix.lower()
#     if suffix in SUPPORTED_EXTENSIONS:
#         return suffix
#     return None

# def _extension_from_content_type(content_type: str | None) -> str | None:
#     if not content_type:
#         return None
#     value = content_type.split(";", 1)[0].strip().lower()
#     return EXTENSION_BY_CONTENT_TYPE.get(value)

# def _extension_from_content_disposition(header: str | None) -> str | None:
#     if not header:
#         return None
#     message = Message()
#     message["content-disposition"] = header
#     filename = message.get_filename()
#     if not filename:
#         return None
#     suffix = Path(filename).suffix.lower()
#     if suffix in SUPPORTED_EXTENSIONS:
#         return suffix
#     return None

# def _sniff_extension(first_bytes: bytes) -> str | None:
#     if first_bytes.startswith(b"RIFF") and first_bytes[8:12] == b"WAVE":
#         return ".wav"
#     if first_bytes.startswith(b"ID3"):
#         return ".mp3"
#     if first_bytes.startswith(b"OggS"):
#         return ".ogg"
#     if first_bytes.startswith(b"fLaC"):
#         return ".flac"
#     if len(first_bytes) >= 12:
#         if first_bytes[4:8] == b"ftyp":
#             return ".mp4"
#     if first_bytes.startswith(b"\x1a\x45\xdf\xa3"):
#         return ".mkv"
#     return None

# def _safe_download_destination(project_id: str, extension: str) -> Path:
#     paths = project_paths(project_id)
#     input_dir = paths["input_dir"]
#     input_dir.mkdir(parents=True, exist_ok=True)
#     return input_dir / f"file{extension}"

# def download_source_file(
#     project_id: str,
#     source_url: str,
#     logger: ProjectLogger,
# ) -> Path:
#     logger.write(f"Downloading source media: {source_url}")
#     with httpx.Client(
#         follow_redirects=True,
#         timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),
#         headers={"User-Agent": "subtitle-pipeline/1.0 (media downloader)"},
#     ) as client:
#         try:
#             with client.stream("GET", source_url) as response:
#                 response.raise_for_status()
#                 content_type = response.headers.get("content-type")
#                 content_length_header = response.headers.get("content-length")

#                 if content_length_header:
#                     try:
#                         content_length = int(content_length_header)
#                     except ValueError:
#                         content_length = None

#                     if content_length is not None and content_length > MAX_DOWNLOAD_BYTES:
#                         raise ValueError("Remote media exceeds MAX_DOWNLOAD_BYTES.")

#                 extension = (
#                     _extension_from_content_disposition(response.headers.get("content-disposition"))
#                     or _extension_from_url(source_url)
#                     or _extension_from_content_type(content_type)
#                 )

#                 if not extension:
#                     extension = MEDIA_MAGIC_FALLBACK_EXTENSION

#                 destination = _safe_download_destination(project_id, extension)
#                 partial = destination.with_suffix(destination.suffix + ".part")
#                 partial.unlink(missing_ok=True)

#                 total = 0
#                 sniff = bytearray()

#                 with partial.open("wb") as output:
#                     for chunk in response.iter_bytes(chunk_size=DOWNLOAD_CHUNK_BYTES):
#                         if not chunk:
#                             continue
#                         if len(sniff) < 64:
#                             needed = 64 - len(sniff)
#                             sniff.extend(chunk[:needed])
#                         total += len(chunk)
#                         if total > MAX_DOWNLOAD_BYTES:
#                             raise ValueError("Remote media exceeded MAX_DOWNLOAD_BYTES.")
#                         output.write(chunk)

#                 if destination.suffix == MEDIA_MAGIC_FALLBACK_EXTENSION:
#                     sniffed = _sniff_extension(bytes(sniff))
#                     if sniffed:
#                         sniffed_destination = destination.with_suffix(sniffed)
#                         sniffed_destination.unlink(missing_ok=True)
#                         os.replace(partial, sniffed_destination)
#                         destination = sniffed_destination
#                     else:
#                         os.replace(partial, destination)
#                 else:
#                     os.replace(partial, destination)

#                 if destination.stat().st_size <= 0:
#                     raise ValueError("Downloaded media file is empty.")

#                 logger.write(
#                     f"Download complete: {destination} ({destination.stat().st_size:,} bytes)"
#                 )
#                 return destination

#         except Exception:
#             try:
#                 if "partial" in locals():
#                     partial.unlink(missing_ok=True)
#             except Exception:
#                 pass
#             raise

# # ============================================================================
# # PIPELINE EXECUTION
# # ============================================================================

# def _result_to_jsonable(value: Any) -> Any:
#     if value is None:
#         return None
#     if isinstance(value, (str, int, float, bool)):
#         return value
#     if isinstance(value, Path):
#         return str(value)
#     if isinstance(value, dict):
#         return {str(key): _result_to_jsonable(item) for key, item in value.items()}
#     if isinstance(value, (list, tuple)):
#         return [_result_to_jsonable(item) for item in value]
#     return str(value)

# def _run_step_1(project_id: str, source_file: Path) -> dict[str, Any]:
#     paths = project_paths(project_id)
#     vocal_path, instrument_path = run_audio_separation(
#         input_file_path=source_file,
#         output_dir=paths["audios"],
#     )
#     return {"vocal_path": str(vocal_path), "instrument_path": str(instrument_path)}

# def _run_step_2(project_id: str) -> dict[str, Any]:
#     return run_diarization(project_id)

# def _run_step_3(project_id: str) -> dict[str, Any]:
#     return run_segment_preparation(project_id)

# def _run_step_4(project_id: str) -> dict[str, Any]:
#     return run_transcription(project_id)

# def _run_step_5(project_id: str) -> dict[str, Any]:
#     return run_step_5(project_id)

# def _step_function(number: int):
#     return {
#         1: _run_step_1,
#         2: _run_step_2,
#         3: _run_step_3,
#         4: _run_step_4,
#         5: _run_step_5,
#     }[number]

# def _first_incomplete_step(state: dict[str, Any]) -> int | None:
#     for step in STEPS:
#         step_state = state["steps"][str(step["number"])]
#         if step_state.get("status") != "success":
#             return step["number"]
#     return None

# def _mark_step_started(state: dict[str, Any], step_number: int) -> None:
#     step_state = state["steps"][str(step_number)]
#     step_state["status"] = "running"
#     step_state["started_at"] = utc_now()
#     step_state["completed_at"] = None
#     step_state["error"] = None

#     state["status"] = "running"
#     state["current_step"] = step_number
#     state["current_step_name"] = STEPS[step_number - 1]["name"]
#     _write_state(state)

# def _mark_step_success(state: dict[str, Any], step_number: int, result: Any) -> None:
#     step_state = state["steps"][str(step_number)]
#     step_state["status"] = "success"
#     step_state["completed_at"] = utc_now()
#     step_state["error"] = None
#     step_state["result"] = _result_to_jsonable(result)

#     state["last_completed_step"] = max(state.get("last_completed_step", 0), step_number)
#     state["error"] = None
#     _write_state(state)

# def _mark_step_failed(state: dict[str, Any], step_number: int, error: Exception) -> None:
#     step_state = state["steps"][str(step_number)]
#     step_state["status"] = "failed"
#     step_state["completed_at"] = utc_now()
#     step_state["error"] = str(error)

#     state["status"] = "failed"
#     state["error"] = str(error)
#     state["current_step"] = step_number
#     state["current_step_name"] = STEPS[step_number - 1]["name"]
#     _write_state(state)

# def _run_pipeline_job(project_id: str) -> None:
#     logger = ProjectLogger(project_id)
#     state = _read_state(project_id)

#     if not state:
#         logger.write("Cannot start pipeline: project state missing.", logging.ERROR)
#         return

#     source_file_value = state.get("source_file")
#     if not source_file_value:
#         state["status"] = "failed"
#         state["error"] = "source_file missing from project state."
#         _write_state(state)
#         logger.write("Pipeline aborted: source_file missing.", logging.ERROR)
#         return

#     source_file = Path(source_file_value)
#     if not source_file.exists():
#         state["status"] = "failed"
#         state["error"] = f"Source file not found: {source_file}"
#         _write_state(state)
#         logger.write(state["error"], logging.ERROR)
#         return

#     start_step = _first_incomplete_step(state)
#     if start_step is None:
#         state["status"] = "completed"
#         state["current_step"] = None
#         state["current_step_name"] = None
#         state["completed_at"] = state.get("completed_at") or utc_now()
#         _write_state(state)
#         logger.write("Project already completed; nothing to do.")
#         return

#     _record_event(state, logger, f"Pipeline starting from Step {start_step}.")

#     try:
#         for number in range(start_step, 6):
#             step = STEPS[number - 1]
#             step_name = step["name"]

#             _mark_step_started(state, number)
#             _record_event(state, logger, f"STEP {number:03d} STARTED | {step_name}")

#             try:
#                 if number == 1:
#                     result = _step_function(number)(project_id, source_file)
#                 else:
#                     result = _step_function(number)(project_id)
#             except Exception as exc:
#                 _mark_step_failed(state, number, exc)
#                 _record_event(
#                     state,
#                     logger,
#                     f"STEP {number:03d} FAILED | {step_name} | {exc}",
#                     level=logging.ERROR,
#                 )
#                 return

#             _mark_step_success(state, number, result)
#             _record_event(state, logger, f"STEP {number:03d} COMPLETED | {step_name}")

#         state["status"] = "completed"
#         state["current_step"] = None
#         state["current_step_name"] = None
#         state["completed_at"] = utc_now()
#         state["error"] = None
#         _write_state(state)
#         _record_event(state, logger, "PIPELINE COMPLETED SUCCESSFULLY.")

#     finally:
#         with ACTIVE_PROJECTS_LOCK:
#             ACTIVE_PROJECTS.discard(project_id)

# def _submit_pipeline(project_id: str) -> None:
#     with ACTIVE_PROJECTS_LOCK:
#         if project_id in ACTIVE_PROJECTS:
#             raise RuntimeError("Project is already running.")
#         ACTIVE_PROJECTS.add(project_id)

#     try:
#         PIPELINE_EXECUTOR.submit(_run_pipeline_job, project_id)
#     except Exception:
#         with ACTIVE_PROJECTS_LOCK:
#             ACTIVE_PROJECTS.discard(project_id)
#         raise

# # ============================================================================
# # API ROUTES
# # ============================================================================

# @app.get("/")
# async def root() -> dict[str, Any]:
#     console_logger.info(f"[{utc_now()}] Root endpoint '/' accessed.")
#     with ACTIVE_PROJECTS_LOCK:
#         active_count = len(ACTIVE_PROJECTS)
#     return {
#         "message": "Subtitle Pipeline Service is active and running.",
#         "version": "1.0.0",
#         "active_projects_count": active_count,
#         "docs_url": "/docs",
#         "health_check": "/health"
#     }

# @app.get("/health")
# async def health() -> dict[str, Any]:
#     with ACTIVE_PROJECTS_LOCK:
#         active = sorted(ACTIVE_PROJECTS)
#     console_logger.info(f"[{utc_now()}] Health check endpoint accessed.")
#     return {
#         "status": "ok",
#         "service": "subtitle-pipeline",
#         "active_projects": active,
#         "max_concurrent_pipelines": MAX_CONCURRENT_PIPELINES,
#         "output_root": str(OUTPUT_ROOT.resolve()),
#     }

# @app.post("/projects", status_code=status.HTTP_202_ACCEPTED)
# async def create_project(request: CreateProjectRequest) -> dict[str, Any]:
#     project_id = validate_project_id(request.project_id)
#     source_url = str(request.file_url)
#     paths = project_paths(project_id)

#     paths["service"].mkdir(parents=True, exist_ok=True)
#     existing = _read_state(project_id)

#     if existing:
#         raise HTTPException(
#             status_code=status.HTTP_409_CONFLICT,
#             detail=f"Project '{project_id}' already exists. Use GET for status or POST /resume.",
#         )

#     logger = ProjectLogger(project_id)
#     state = _initial_state(project_id, source_url)
#     _write_state(state)
#     logger.write(f"Project created: {project_id}")

#     try:
#         source_file = await asyncio.to_thread(
#             download_source_file,
#             project_id,
#             source_url,
#             logger,
#         )

#         state = _read_state(project_id)
#         if not state:
#             raise RuntimeError("Project state disappeared after download.")

#         state["source_file"] = str(source_file.resolve())
#         state["status"] = "queued"
#         state["error"] = None
#         _write_state(state)

#         _record_event(state, logger, "Source media downloaded successfully. Submitting pipeline.")
#         _submit_pipeline(project_id)

#     except Exception as exc:
#         state = _read_state(project_id) or state
#         state["status"] = "failed"
#         state["error"] = str(exc)
#         _write_state(state)
#         logger.write(f"PROJECT CREATION FAILED | {exc}", level=logging.ERROR)
#         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc

#     return {
#         "project_id": project_id,
#         "status": "queued",
#         "source_file": state["source_file"],
#         "status_url": f"/projects/{project_id}",
#         "logs_url": f"/projects/{project_id}/logs",
#         "resume_url": f"/projects/{project_id}/resume",
#     }

# @app.get("/projects/{project_id}")
# async def get_project(project_id: str) -> dict[str, Any]:
#     project_id = validate_project_id(project_id)
#     state = _read_state(project_id)

#     if not state:
#         raise HTTPException(
#             status_code=status.HTTP_404_NOT_FOUND,
#             detail=f"Project '{project_id}' does not exist.",
#         )

#     with ACTIVE_PROJECTS_LOCK:
#         running = project_id in ACTIVE_PROJECTS

#     state["runtime"] = {"running_in_this_process": running}
#     return state

# @app.get("/projects/{project_id}/logs")
# async def get_project_logs(project_id: str, limit: int = 100) -> dict[str, Any]:
#     project_id = validate_project_id(project_id)
#     state = _read_state(project_id)

#     if not state:
#         raise HTTPException(
#             status_code=status.HTTP_404_NOT_FOUND,
#             detail=f"Project '{project_id}' does not exist.",
#         )

#     limit = max(1, min(limit, MAX_LOG_LINES))
#     logger = ProjectLogger(project_id)

#     return {
#         "project_id": project_id,
#         "log_file": str(logger.log_path.resolve()),
#         "lines": logger.recent_lines(limit),
#     }

# @app.post("/projects/{project_id}/resume", response_model=ResumeProjectResponse)
# async def resume_project(project_id: str) -> ResumeProjectResponse:
#     project_id = validate_project_id(project_id)
#     state = _read_state(project_id)

#     if not state:
#         raise HTTPException(
#             status_code=status.HTTP_404_NOT_FOUND,
#             detail=f"Project '{project_id}' does not exist.",
#         )

#     if not state.get("source_file"):
#         raise HTTPException(
#             status_code=status.HTTP_409_CONFLICT,
#             detail="Cannot resume because the source media file is missing from project state.",
#         )

#     if not Path(state["source_file"]).exists():
#         raise HTTPException(
#             status_code=status.HTTP_409_CONFLICT,
#             detail="Cannot resume because the downloaded source file no longer exists.",
#         )

#     with ACTIVE_PROJECTS_LOCK:
#         if project_id in ACTIVE_PROJECTS:
#             raise HTTPException(
#                 status_code=status.HTTP_409_CONFLICT,
#                 detail="Project is already running.",
#             )

#     start_step = _first_incomplete_step(state)
#     if start_step is None:
#         return ResumeProjectResponse(
#             project_id=project_id,
#             status="completed",
#             resumed_from_step=None,
#             message="Project is already complete.",
#         )

#     step_state = state["steps"][str(start_step)]
#     step_state["status"] = "pending"
#     step_state["error"] = None
#     step_state["started_at"] = None
#     step_state["completed_at"] = None

#     state["status"] = "queued"
#     state["error"] = None
#     state["current_step"] = start_step
#     state["current_step_name"] = STEPS[start_step - 1]["name"]
#     _write_state(state)

#     logger = ProjectLogger(project_id)
#     _record_event(
#         state,
#         logger,
#         f"RESUME REQUESTED | restarting from Step {start_step:03d} | {STEPS[start_step - 1]['name']}",
#     )

#     try:
#         _submit_pipeline(project_id)
#     except RuntimeError as exc:
#         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc

#     return ResumeProjectResponse(
#         project_id=project_id,
#         status="queued",
#         resumed_from_step=start_step,
#         message=f"Project queued for resume from Step {start_step:03d}.",
#     )

# @app.get("/projects/{project_id}/subtitle")
# async def get_project_subtitle(project_id: str):
#     project_id = validate_project_id(project_id)
#     state = _read_state(project_id)

#     if not state:
#         raise HTTPException(
#             status_code=status.HTTP_404_NOT_FOUND,
#             detail={
#                 "status": "not_found",
#                 "message": f"Project '{project_id}' does not exist.",
#             },
#         )

#     step_5 = state.get("steps", {}).get("5", {})
#     step_5_result = step_5.get("result") or {}
#     srt_path_value = step_5_result.get("srt_path")

#     if srt_path_value:
#         srt_path = Path(srt_path_value)
#         if srt_path.exists() and srt_path.is_file():
#             return FileResponse(
#                 path=str(srt_path),
#                 media_type="application/x-subrip",
#                 filename=srt_path.name,
#                 headers={"Content-Disposition": f'attachment; filename="{srt_path.name}"'},
#             )

#     subtitles_dir = OUTPUT_ROOT / project_id / "subtitles"
#     if subtitles_dir.exists():
#         srt_files = sorted(
#             subtitles_dir.glob("*.srt"),
#             key=lambda path: path.stat().st_mtime,
#             reverse=True,
#         )
#         if srt_files:
#             srt_path = srt_files[0]
#             return FileResponse(
#                 path=str(srt_path),
#                 media_type="application/x-subrip",
#                 filename=srt_path.name,
#                 headers={"Content-Disposition": f'attachment; filename="{srt_path.name}"'},
#             )

#     logger = ProjectLogger(project_id)
#     return {
#         "status": "not_ready",
#         "message": "Subtitle file has not been created yet.",
#         "project_id": project_id,
#         "project_status": state.get("status"),
#         "current_step": state.get("current_step"),
#         "current_step_name": state.get("current_step_name"),
#         "last_completed_step": state.get("last_completed_step", 0),
#         "subtitle_exists": False,
#         "error": state.get("error"),
#         "logs": logger.recent_lines(20),
#     }

#     '''


# curl -X POST \
#   "https://friendly-space-train-xrvvggqpg95jc57j-8000.app.github.dev/projects" \
#   -H "Content-Type: application/json" \
#   -d '{
#     "project_id": "test-conversation-002",
#     "file_url": "https://huggingface.co/datasets/zhangjinyang/TalkingHead-1KH-audio/resolve/main/val_cropped_clips/1lSejjfNHpw_0075_S1_E728_L671_T47_R1471_B847.mp4"
#   }'

#   '''


# uvicorn main:app --host 0.0.0.0 --port 8000
"""
FastAPI service for the 5-step subtitle pipeline.

Project storage contract:

src/
├── input/
│   └── <project_id>/
│       └── file.<original-extension>
└── output/
    └── <project_id>/
        ├── audios/
        ├── diarization/
        ├── segments/
        ├── transcribe/
        ├── subtitles/
        ├── service/
        │   ├── status.json
        │   └── pipeline.log
        └── ...

API:

POST /projects
    Start a new project from a remote HTTP/HTTPS media URL.

GET /projects/{project_id}
    Return current persistent project status.

GET /projects/{project_id}/logs
    Return recent project log lines.

POST /projects/{project_id}/resume
    Resume the failed/interrupted project from the first incomplete step.

GET /health
    Health check.

The processing job is intentionally single-worker inside this process.
That is important for your CPU-heavy ML pipeline because Step 2/4 should not
be duplicated into multiple independent OS processes.

For production across multiple machines/processes, move the job execution to
a durable queue such as Celery/RQ/Arq with Redis. FastAPI's BackgroundTasks
are suitable for smaller/simple background work but FastAPI itself documents
a queue/worker system as preferable for heavy background computation.
"""

from __future__ import annotations

import asyncio
import copy
import json
import logging
import os
import re
import shutil
import sys
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from email.message import Message
from pathlib import Path
from typing import Any
from urllib.parse import urlparse, unquote

import httpx
from fastapi import FastAPI, HTTPException, status
from fastapi.responses import FileResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, HttpUrl

from src.pipeline.step_001_separate import run_audio_separation
from src.pipeline.step_002_diarize import run_diarization
from src.pipeline.step_003_segment import run_segment_preparation
from src.pipeline.step_004_transcribe import run_transcription
from src.pipeline.step_005_original_subtitle import run_step_5


# ============================================================================
# CONFIGURATION
# ============================================================================

BASE_DIR = Path(__file__).resolve().parent
SRC_DIR = BASE_DIR / "src"

INPUT_ROOT = SRC_DIR / "input"
OUTPUT_ROOT = SRC_DIR / "output"

MAX_DOWNLOAD_BYTES = int(
    os.getenv(
        "MAX_DOWNLOAD_BYTES",
        str(4 * 1024 * 1024 * 1024),
    )
)

DOWNLOAD_CHUNK_BYTES = 1024 * 1024  # 1 MiB
HTTP_TIMEOUT_SECONDS = float(
    os.getenv(
        "HTTP_TIMEOUT_SECONDS",
        "120",
    )
)

MAX_LOG_LINES = int(
    os.getenv(
        "MAX_LOG_LINES",
        "5000",
    )
)

MAX_STATUS_LOG_LINES = int(
    os.getenv(
        "MAX_STATUS_LOG_LINES",
        "100",
    )
)

# Webhook (optional, sent per-project) delivery timeout.
WEBHOOK_TIMEOUT_SECONDS = float(
    os.getenv(
        "WEBHOOK_TIMEOUT_SECONDS",
        "15",
    )
)

# Whether logs are also mirrored to stdout (colorized) in addition to the
# per-project log file on disk.
LOG_TO_CONSOLE = os.getenv(
    "LOG_TO_CONSOLE",
    "true",
).strip().lower() in {
    "1",
    "true",
    "yes",
}

# ============================================================================
# TIMEZONE (INDIAN STANDARD TIME) — all human-facing log timestamps use IST.
# ============================================================================

IST_TZ = timezone(
    timedelta(
        hours=5,
        minutes=30,
    ),
    name="IST",
)


def ist_now() -> datetime:
    return datetime.now(
        IST_TZ
    )


def ist_now_iso() -> str:
    return ist_now().isoformat()


def ist_now_display() -> str:
    return ist_now().strftime(
        "%Y-%m-%d %H:%M:%S.%f"
    )[:-3] + " IST"


# ============================================================================
# ANSI COLORS — used to make per-step / per-event log lines colorful.
# ============================================================================

class Ansi:
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"

    BLACK = "\033[30m"
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    MAGENTA = "\033[35m"
    CYAN = "\033[36m"
    WHITE = "\033[37m"

    BRIGHT_RED = "\033[91m"
    BRIGHT_GREEN = "\033[92m"
    BRIGHT_YELLOW = "\033[93m"
    BRIGHT_BLUE = "\033[94m"
    BRIGHT_MAGENTA = "\033[95m"
    BRIGHT_CYAN = "\033[96m"
    BRIGHT_WHITE = "\033[97m"


# Each pipeline step gets its own distinct color so its log lines are
# instantly recognizable when scanning pipeline.log or the console.
STEP_COLORS: dict[int, str] = {
    1: Ansi.BRIGHT_CYAN,
    2: Ansi.BRIGHT_MAGENTA,
    3: Ansi.BRIGHT_YELLOW,
    4: Ansi.BRIGHT_BLUE,
    5: Ansi.BRIGHT_GREEN,
}

# Event-type colors (STARTED / COMPLETED / FAILED / etc.) layered on top of
# the step color.
EVENT_COLORS: dict[str, str] = {
    "STARTED": Ansi.BLUE,
    "RUNNING": Ansi.CYAN,
    "COMPLETED": Ansi.GREEN,
    "SUCCESS": Ansi.GREEN,
    "FAILED": Ansi.RED,
    "ERROR": Ansi.RED,
    "INFO": Ansi.WHITE,
    "WEBHOOK": Ansi.MAGENTA,
    "RESUME": Ansi.YELLOW,
    "PROJECT": Ansi.BRIGHT_WHITE,
}

LEVEL_COLORS: dict[int, str] = {
    logging.DEBUG: Ansi.WHITE,
    logging.INFO: Ansi.CYAN,
    logging.WARNING: Ansi.YELLOW,
    logging.ERROR: Ansi.RED,
    logging.CRITICAL: Ansi.BOLD + Ansi.BRIGHT_RED,
}


def _colorize(
    text: str,
    color: str,
) -> str:
    return f"{color}{text}{Ansi.RESET}"


def step_log_message(
    step_number: int | None,
    event_type: str,
    message: str,
) -> str:
    """
    Build a colorful, per-step log message such as:

        [STEP 002] STARTED | speaker_diarization

    Colors are ANSI escape codes. They render nicely in any terminal that
    tails pipeline.log (e.g. `tail -f` / `cat`) and in the console handler.
    They are harmless (ignored/visible as-is) in non-ANSI viewers.
    """
    step_color = (
        STEP_COLORS.get(
            step_number,
            Ansi.WHITE,
        )
        if step_number is not None
        else Ansi.BRIGHT_WHITE
    )

    event_color = EVENT_COLORS.get(
        event_type,
        Ansi.WHITE,
    )

    step_tag = (
        _colorize(
            f"[STEP {step_number:03d}]",
            Ansi.BOLD + step_color,
        )
        if step_number is not None
        else _colorize(
            "[PROJECT]",
            Ansi.BOLD + Ansi.BRIGHT_WHITE,
        )
    )

    event_tag = _colorize(
        event_type,
        Ansi.BOLD + event_color,
    )

    return (
        f"{step_tag} {event_tag} | {message}"
    )

# Only one whole pipeline job executes at a time in this process.
#
# This is deliberate. Step 2/4 are heavy CPU model workloads, and concurrent
# project jobs would fight over the same CPU/RAM resources.
MAX_CONCURRENT_PIPELINES = 1

SUPPORTED_EXTENSIONS = {
    ".mp3",
    ".wav",
    ".flac",
    ".m4a",
    ".aac",
    ".ogg",
    ".opus",
    ".wma",
    ".mp4",
    ".mkv",
    ".mov",
    ".avi",
    ".webm",
    ".m4v",
    ".mpeg",
    ".mpg",
    ".ts",
    ".mts",
    ".m2ts",
    ".3gp",
}

EXTENSION_BY_CONTENT_TYPE = {
    "audio/mpeg": ".mp3",
    "audio/wav": ".wav",
    "audio/x-wav": ".wav",
    "audio/flac": ".flac",
    "audio/x-flac": ".flac",
    "audio/mp4": ".m4a",
    "audio/aac": ".aac",
    "audio/ogg": ".ogg",
    "audio/opus": ".opus",
    "video/mp4": ".mp4",
    "video/quicktime": ".mov",
    "video/x-matroska": ".mkv",
    "video/webm": ".webm",
    "video/mpeg": ".mpg",
}

MEDIA_MAGIC_FALLBACK_EXTENSION = ".bin"

STEPS = [
    {
        "number": 1,
        "name": "audio_separation",
    },
    {
        "number": 2,
        "name": "speaker_diarization",
    },
    {
        "number": 3,
        "name": "speaker_segment_preparation",
    },
    {
        "number": 4,
        "name": "speaker_aware_transcription",
    },
    {
        "number": 5,
        "name": "original_subtitle_generation",
    },
]


# ============================================================================
# FASTAPI APP
# ============================================================================

app = FastAPI(
    title="Subtitle Pipeline Service",
    version="1.0.0",
    description=(
        "Project-based FastAPI orchestration for audio separation, "
        "speaker diarization, segmentation, transcription, and subtitles."
    ),
)

# Configure CORS to allow requests from anywhere
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],       # Allows requests from any origin
    allow_credentials=True,    # Allows cookies/auth headers
    allow_methods=["*"],       # Allows all HTTP methods (GET, POST, etc.)
    allow_headers=["*"],       # Allows all headers
)

# ============================================================================
# GLOBAL JOB EXECUTOR
# ============================================================================

PIPELINE_EXECUTOR = ThreadPoolExecutor(
    max_workers=MAX_CONCURRENT_PIPELINES,
    thread_name_prefix="subtitle-pipeline",
)

ACTIVE_PROJECTS: set[str] = set()
ACTIVE_PROJECTS_LOCK = threading.Lock()

# Per-project asyncio event used only to wake log/status waiters if later
# extended to WebSocket/SSE. The persistent JSON files remain the source of
# truth.
PROJECT_EVENTS: dict[str, asyncio.Event] = {}

PROJECT_EVENTS_LOCK = threading.Lock()


# ============================================================================
# REQUEST MODELS
# ============================================================================

class CreateProjectRequest(BaseModel):
    project_id: str = Field(
        ...,
        min_length=1,
        max_length=128,
        description="Directory-safe project identifier.",
    )

    file_url: HttpUrl = Field(
        ...,
        description="Public/signed HTTP(S) URL of the source media file.",
    )

    webhook_url: HttpUrl | None = Field(
        default=None,
        description=(
            "Optional HTTP(S) webhook URL. If present, the service will "
            "POST a JSON payload to it whenever a pipeline step completes "
            "(with that step's detailed logs plus all previous logs for "
            "this project), and once more when the whole pipeline "
            "completes (with all logs and the subtitle file path/"
            "download link)."
        ),
    )


class ResumeProjectResponse(BaseModel):
    project_id: str
    status: str
    resumed_from_step: int | None
    message: str


# ============================================================================
# LOGGING / PERSISTENT STATE
# ============================================================================

class ISTColorFormatter(logging.Formatter):
    """
    Logging formatter that:
      1. Renders timestamps in Indian Standard Time (IST, UTC+05:30).
      2. Colorizes the level name using ANSI escape codes so pipeline.log
         (and the console mirror) is colorful and easy to scan.

    The per-step / per-event coloring (STARTED / COMPLETED / FAILED, etc.)
    is applied separately, in the message text itself, via
    `step_log_message()`, so a single log line ends up carrying both a
    level color and a step/event color.
    """

    def __init__(
        self,
        fmt: str,
        colorize: bool = True,
    ) -> None:
        super().__init__(fmt)
        self.colorize = colorize

    def formatTime(
        self,
        record: logging.LogRecord,
        datefmt: str | None = None,
    ) -> str:
        dt = datetime.fromtimestamp(
            record.created,
            tz=IST_TZ,
        )

        if datefmt:
            return dt.strftime(
                datefmt
            )

        return (
            dt.strftime(
                "%Y-%m-%d %H:%M:%S.%f"
            )[:-3]
            + " IST"
        )

    def format(
        self,
        record: logging.LogRecord,
    ) -> str:
        if not self.colorize:
            return super().format(
                record
            )

        # Work on a shallow copy so we never mutate the shared LogRecord
        # (multiple handlers format the same record).
        colored_record = copy.copy(
            record
        )

        level_color = LEVEL_COLORS.get(
            colored_record.levelno,
            Ansi.WHITE,
        )

        colored_record.levelname = _colorize(
            f"{colored_record.levelname:<8}",
            level_color,
        )

        return super().format(
            colored_record
        )


class ProjectLogger:
    """
    File-backed project logger.

    Every project gets:
        src/output/<project_id>/service/pipeline.log

    A project-specific status.json is also updated after every important
    transition so status survives API process restarts.
    """

    def __init__(
        self,
        project_id: str,
    ) -> None:
        self.project_id = project_id

        self.project_dir = (
            OUTPUT_ROOT / project_id
        )

        self.service_dir = (
            self.project_dir / "service"
        )

        self.log_path = (
            self.service_dir / "pipeline.log"
        )

        self.status_path = (
            self.service_dir / "status.json"
        )

        self.service_dir.mkdir(
            parents=True,
            exist_ok=True,
        )

        self._lock = threading.Lock()

        # Avoid stacking handlers when the same service object is reconstructed
        # during development/reload.
        self.logger = logging.getLogger(
            f"subtitle-project.{project_id}"
        )
        self.logger.setLevel(logging.INFO)
        self.logger.propagate = False

        if not self.logger.handlers:
            file_handler = logging.FileHandler(
                self.log_path,
                encoding="utf-8",
            )
            file_handler.setFormatter(
                ISTColorFormatter(
                    "%(asctime)s | %(levelname)s | %(message)s",
                    colorize=True,
                )
            )
            self.logger.addHandler(file_handler)

            if LOG_TO_CONSOLE:
                console_handler = logging.StreamHandler(
                    sys.stdout
                )
                console_handler.setFormatter(
                    ISTColorFormatter(
                        "%(asctime)s | %(levelname)s | "
                        f"[{project_id}] "
                        "%(message)s",
                        colorize=True,
                    )
                )
                self.logger.addHandler(console_handler)

    def write(
        self,
        message: str,
        level: int = logging.INFO,
    ) -> None:
        with self._lock:
            self.logger.log(
                level,
                message,
            )

    def write_step(
        self,
        step_number: int | None,
        event_type: str,
        message: str,
        level: int = logging.INFO,
    ) -> str:
        """
        Write a colorful, per-step log line (both to file and, if enabled,
        console) and return the fully-composed message text so callers can
        reuse the exact same text (e.g. for state["logs"] / webhooks).
        """
        composed = step_log_message(
            step_number,
            event_type,
            message,
        )

        self.write(
            composed,
            level=level,
        )

        return composed

    def recent_lines(
        self,
        limit: int = MAX_STATUS_LOG_LINES,
    ) -> list[str]:
        if not self.log_path.exists():
            return []

        # Tail efficiently enough for a project log file while keeping the
        # implementation dependency-free.
        with self.log_path.open(
            "r",
            encoding="utf-8",
            errors="replace",
        ) as handle:
            lines = handle.readlines()

        return lines[-limit:]

    def all_lines(
        self,
    ) -> list[str]:
        """
        Full project log, unbounded. Used for the "all previous logs
        related to this project" payload sent to the webhook, and for the
        final completed-pipeline webhook event.
        """
        if not self.log_path.exists():
            return []

        with self.log_path.open(
            "r",
            encoding="utf-8",
            errors="replace",
        ) as handle:
            return handle.readlines()

    def lines_for_step(
        self,
        step_number: int,
    ) -> list[str]:
        """
        Just the lines related to a single step (STARTED/COMPLETED/FAILED
        and anything else logged with that step's [STEP NNN] tag).
        Used for the "this step's detailed logs" webhook payload.
        """
        marker = f"STEP {step_number:03d}]"

        return [
            line
            for line in self.all_lines()
            if marker in line
        ]


# ============================================================================
# PROJECT STATE
# ============================================================================

def utc_now() -> str:
    return datetime.now(
        timezone.utc
    ).isoformat()


def _project_state_path(
    project_id: str,
) -> Path:
    return (
        OUTPUT_ROOT
        / project_id
        / "service"
        / "status.json"
    )


def _default_step_state() -> dict[str, Any]:
    return {
        "status": "pending",
        "started_at": None,
        "completed_at": None,
        "error": None,
        "result": None,
    }


def _initial_state(
    project_id: str,
    source_url: str,
    webhook_url: str | None = None,
) -> dict[str, Any]:
    return {
        "schema_version": "1.0",
        "project_id": project_id,
        "source_url": source_url,
        "source_file": None,
        "webhook_url": webhook_url,
        "status": "queued",
        "current_step": None,
        "current_step_name": None,
        "last_completed_step": 0,
        "created_at": utc_now(),
        "updated_at": utc_now(),
        "completed_at": None,
        "error": None,
        "steps": {
            str(step["number"]): {
                **step,
                **_default_step_state(),
            }
            for step in STEPS
        },
        "logs": [],
    }


def _read_state(
    project_id: str,
) -> dict[str, Any] | None:
    path = _project_state_path(
        project_id
    )

    if not path.exists():
        return None

    try:
        with path.open(
            "r",
            encoding="utf-8",
        ) as handle:
            return json.load(handle)
    except json.JSONDecodeError:
        return None


def _atomic_write_json(
    path: Path,
    payload: dict[str, Any],
) -> None:
    path.parent.mkdir(
        parents=True,
        exist_ok=True,
    )

    temp = path.with_name(
        f".{path.name}.tmp-{os.getpid()}"
    )

    try:
        with temp.open(
            "w",
            encoding="utf-8",
        ) as handle:
            json.dump(
                payload,
                handle,
                indent=2,
                ensure_ascii=False,
            )
            handle.write("\n")

        os.replace(
            temp,
            path,
        )
    finally:
        temp.unlink(
            missing_ok=True,
        )


def _write_state(
    state: dict[str, Any],
) -> None:
    state["updated_at"] = utc_now()

    # Keep the state object compact; logs themselves live in pipeline.log.
    state["logs"] = state.get(
        "logs",
        [],
    )[-MAX_STATUS_LOG_LINES:]

    _atomic_write_json(
        _project_state_path(
            state["project_id"]
        ),
        state,
    )

    with PROJECT_EVENTS_LOCK:
        event = PROJECT_EVENTS.get(
            state["project_id"]
        )

    if event:
        event.set()


def _record_event(
    state: dict[str, Any],
    logger: ProjectLogger,
    message: str,
    *,
    level: int = logging.INFO,
) -> None:
    logger.write(
        message,
        level=level,
    )

    logs = state.setdefault(
        "logs",
        [],
    )

    logs.append(
        {
            "timestamp_ist": ist_now_iso(),
            "message": message,
        }
    )

    state["updated_at"] = utc_now()

    _write_state(
        state
    )


# ============================================================================
# PROJECT ID / PATH SAFETY
# ============================================================================

PROJECT_ID_PATTERN = re.compile(
    r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)


def validate_project_id(
    project_id: str,
) -> str:
    value = str(
        project_id
    ).strip()

    if not PROJECT_ID_PATTERN.fullmatch(
        value
    ):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=(
                "project_id must contain only letters, numbers, "
                "dot, underscore, and hyphen, and must not exceed 128 characters."
            ),
        )

    return value


def project_paths(
    project_id: str,
) -> dict[str, Path]:
    project_id = validate_project_id(
        project_id
    )

    root = (
        OUTPUT_ROOT
        / project_id
    )

    return {
        "project_dir": root,
        "input_dir": INPUT_ROOT / project_id,
        "output_dir": root,
        "audios": root / "audios",
        "diarization": root / "diarization",
        "segments": root / "segments",
        "transcribe": root / "transcribe",
        "subtitles": root / "subtitles",
        "service": root / "service",
    }


# ============================================================================
# URL DOWNLOAD
# ============================================================================

def _extension_from_url(
    url: str,
) -> str | None:
    parsed = urlparse(
        url
    )

    name = unquote(
        Path(
            parsed.path
        ).name
    )

    suffix = Path(
        name
    ).suffix.lower()

    if suffix in SUPPORTED_EXTENSIONS:
        return suffix

    return None


def _extension_from_content_type(
    content_type: str | None,
) -> str | None:
    if not content_type:
        return None

    value = content_type.split(
        ";",
        1,
    )[0].strip().lower()

    return EXTENSION_BY_CONTENT_TYPE.get(
        value
    )


def _extension_from_content_disposition(
    header: str | None,
) -> str | None:
    if not header:
        return None

    message = Message()

    message["content-disposition"] = header

    filename = message.get_filename()

    if not filename:
        return None

    suffix = Path(
        filename
    ).suffix.lower()

    if suffix in SUPPORTED_EXTENSIONS:
        return suffix

    return None


def _sniff_extension(
    first_bytes: bytes,
) -> str | None:
    """
    Best-effort media signature detection.

    FFmpeg remains the final authority for what the file can actually decode.
    This function only chooses a useful extension when the URL has no media
    suffix and the content type is generic.
    """
    if first_bytes.startswith(
        b"RIFF"
    ) and first_bytes[8:12] == b"WAVE":
        return ".wav"

    if first_bytes.startswith(
        b"ID3"
    ):
        return ".mp3"

    if first_bytes.startswith(
        b"OggS"
    ):
        return ".ogg"

    if first_bytes.startswith(
        b"fLaC"
    ):
        return ".flac"

    if len(first_bytes) >= 12:
        if first_bytes[4:8] == b"ftyp":
            return ".mp4"

    if first_bytes.startswith(
        b"\x1a\x45\xdf\xa3"
    ):
        return ".mkv"

    return None


def _safe_download_destination(
    project_id: str,
    extension: str,
) -> Path:
    paths = project_paths(
        project_id
    )

    input_dir = paths[
        "input_dir"
    ]

    input_dir.mkdir(
        parents=True,
        exist_ok=True,
    )

    return (
        input_dir
        / f"file{extension}"
    )


def download_source_file(
    project_id: str,
    source_url: str,
    logger: ProjectLogger,
) -> Path:
    """
    Stream the remote object directly to disk.

    The complete file is never loaded into RAM.

    A .part file is used until the download completes. This makes resume/retry
    safe: an interrupted download is never mistaken for a usable source.
    """
    logger.write_step(
        None,
        "PROJECT",
        f"Downloading source media: {source_url} "
        f"| started_at={ist_now_display()}",
    )

    with httpx.Client(
        follow_redirects=True,
        timeout=httpx.Timeout(
            HTTP_TIMEOUT_SECONDS
        ),
        headers={
            "User-Agent": (
                "subtitle-pipeline/1.0 "
                "(media downloader)"
            ),
        },
    ) as client:

        try:
            with client.stream(
                "GET",
                source_url,
            ) as response:

                response.raise_for_status()

                content_type = response.headers.get(
                    "content-type"
                )

                content_length_header = response.headers.get(
                    "content-length"
                )

                if content_length_header:
                    try:
                        content_length = int(
                            content_length_header
                        )
                    except ValueError:
                        content_length = None

                    if (
                        content_length is not None
                        and content_length > MAX_DOWNLOAD_BYTES
                    ):
                        raise ValueError(
                            "Remote media exceeds MAX_DOWNLOAD_BYTES."
                        )

                extension = (
                    _extension_from_content_disposition(
                        response.headers.get(
                            "content-disposition"
                        )
                    )
                    or _extension_from_url(
                        source_url
                    )
                    or _extension_from_content_type(
                        content_type
                    )
                )

                if not extension:
                    extension = MEDIA_MAGIC_FALLBACK_EXTENSION

                destination = _safe_download_destination(
                    project_id,
                    extension,
                )

                partial = destination.with_suffix(
                    destination.suffix
                    + ".part"
                )

                # Remove old partial state for a clean new download.
                partial.unlink(
                    missing_ok=True
                )

                total = 0
                sniff = bytearray()

                with partial.open(
                    "wb"
                ) as output:

                    for chunk in response.iter_bytes(
                        chunk_size=DOWNLOAD_CHUNK_BYTES
                    ):
                        if not chunk:
                            continue

                        if len(sniff) < 64:
                            needed = 64 - len(sniff)
                            sniff.extend(
                                chunk[:needed]
                            )

                        total += len(
                            chunk
                        )

                        if total > MAX_DOWNLOAD_BYTES:
                            raise ValueError(
                                "Remote media exceeded MAX_DOWNLOAD_BYTES."
                            )

                        output.write(
                            chunk
                        )

                if (
                    destination.suffix == MEDIA_MAGIC_FALLBACK_EXTENSION
                ):
                    sniffed = _sniff_extension(
                        bytes(sniff)
                    )

                    if sniffed:
                        sniffed_destination = (
                            destination.with_suffix(
                                sniffed
                            )
                        )

                        sniffed_destination.unlink(
                            missing_ok=True
                        )

                        os.replace(
                            partial,
                            sniffed_destination,
                        )

                        destination = sniffed_destination

                    else:
                        os.replace(
                            partial,
                            destination,
                        )
                else:
                    os.replace(
                        partial,
                        destination,
                    )

                if destination.stat().st_size <= 0:
                    raise ValueError(
                        "Downloaded media file is empty."
                    )

                logger.write_step(
                    None,
                    "PROJECT",
                    "Download complete: "
                    f"{destination} "
                    f"({destination.stat().st_size:,} bytes) "
                    f"| completed_at={ist_now_display()}",
                )

                return destination

        except Exception:
            # If an exception occurs before the final rename, clean the partial
            # file. A future resume will download a complete source again.
            try:
                if "partial" in locals():
                    partial.unlink(
                        missing_ok=True
                    )
            except Exception:
                pass

            raise


# ============================================================================
# PIPELINE EXECUTION
# ============================================================================

def _result_to_jsonable(
    value: Any,
) -> Any:
    if value is None:
        return None

    if isinstance(
        value,
        (
            str,
            int,
            float,
            bool,
        ),
    ):
        return value

    if isinstance(
        value,
        Path,
    ):
        return str(
            value
        )

    if isinstance(
        value,
        dict,
    ):
        return {
            str(key): _result_to_jsonable(
                item
            )
            for key, item in value.items()
        }

    if isinstance(
        value,
        (list, tuple),
    ):
        return [
            _result_to_jsonable(
                item
            )
            for item in value
        ]

    return str(
        value
    )


# ============================================================================
# OPTIONAL PROJECT WEBHOOK
# ============================================================================

def _subtitle_download_info(
    project_id: str,
    state: dict[str, Any],
) -> dict[str, Any]:
    """
    Resolve the generated subtitle file path (if any) plus the existing
    download route for it, for inclusion in webhook payloads.
    """
    step_5 = state.get(
        "steps",
        {},
    ).get(
        "5",
        {},
    )

    step_5_result = (
        step_5.get("result")
        or {}
    )

    srt_path_value = step_5_result.get(
        "srt_path"
    )

    subtitle_path: str | None = None

    if srt_path_value and Path(
        srt_path_value
    ).exists():
        subtitle_path = str(
            Path(
                srt_path_value
            ).resolve()
        )
    else:
        subtitles_dir = (
            OUTPUT_ROOT
            / project_id
            / "subtitles"
        )

        if subtitles_dir.exists():
            srt_files = sorted(
                subtitles_dir.glob(
                    "*.srt"
                ),
                key=lambda path: path.stat().st_mtime,
                reverse=True,
            )

            if srt_files:
                subtitle_path = str(
                    srt_files[0].resolve()
                )

    return {
        "subtitle_file_path": subtitle_path,
        "subtitle_ready": subtitle_path is not None,
        # This route already exists in this API and streams the .srt file.
        "subtitle_download_url": (
            f"/projects/{project_id}/subtitle"
        ),
    }


def _send_webhook(
    state: dict[str, Any],
    logger: ProjectLogger,
    event: str,
    *,
    step_number: int | None = None,
    step_state: dict[str, Any] | None = None,
) -> None:
    """
    Best-effort webhook delivery.

    Fired:
      - after every step completes (success or failure): includes that
        step's own detailed logs, plus every log line for the project so
        far ("previous logs too, related to that project").
      - once more when the whole pipeline finishes: includes ALL logs for
        the project plus the subtitle file path / download link.

    Failures here are logged but NEVER raised — a webhook outage must not
    break the pipeline.
    """
    webhook_url = state.get(
        "webhook_url"
    )

    if not webhook_url:
        return

    project_id = state[
        "project_id"
    ]

    payload: dict[str, Any] = {
        "event": event,
        "project_id": project_id,
        "sent_at_ist": ist_now_iso(),
        "status": state.get("status"),
        "current_step": state.get("current_step"),
        "current_step_name": state.get("current_step_name"),
        "last_completed_step": state.get(
            "last_completed_step",
            0,
        ),
        "error": state.get("error"),
        "logs_so_far": logger.all_lines(),
    }

    if step_number is not None:
        payload["step"] = {
            "number": step_number,
            "name": STEPS[step_number - 1]["name"],
            "status": (step_state or {}).get("status"),
            "started_at": (step_state or {}).get("started_at"),
            "completed_at": (step_state or {}).get("completed_at"),
            "error": (step_state or {}).get("error"),
            "result": (step_state or {}).get("result"),
        }
        payload["step_logs"] = logger.lines_for_step(
            step_number
        )

    if event == "pipeline_completed":
        payload.update(
            _subtitle_download_info(
                project_id,
                state,
            )
        )
        payload["completed_at"] = state.get(
            "completed_at"
        )

    try:
        with httpx.Client(
            timeout=httpx.Timeout(
                WEBHOOK_TIMEOUT_SECONDS
            ),
        ) as client:
            response = client.post(
                webhook_url,
                json=payload,
            )

        logger.write_step(
            step_number,
            "WEBHOOK",
            (
                f"Webhook delivered | event={event} | "
                f"url={webhook_url} | "
                f"response_status={response.status_code}"
            ),
        )

    except Exception as exc:
        logger.write_step(
            step_number,
            "WEBHOOK",
            (
                f"Webhook delivery FAILED | event={event} | "
                f"url={webhook_url} | error={exc}"
            ),
            level=logging.WARNING,
        )


def _run_step_1(
    project_id: str,
    source_file: Path,
) -> dict[str, Any]:
    paths = project_paths(
        project_id
    )

    vocal_path, instrument_path = (
        run_audio_separation(
            input_file_path=source_file,
            output_dir=paths["audios"],
        )
    )

    return {
        "vocal_path": str(
            vocal_path
        ),
        "instrument_path": str(
            instrument_path
        ),
    }


def _run_step_2(
    project_id: str,
) -> dict[str, Any]:
    return run_diarization(
        project_id
    )


def _run_step_3(
    project_id: str,
) -> dict[str, Any]:
    return run_segment_preparation(
        project_id
    )


def _run_step_4(
    project_id: str,
) -> dict[str, Any]:
    return run_transcription(
        project_id
    )


def _run_step_5(
    project_id: str,
) -> dict[str, Any]:
    return run_step_5(
        project_id
    )


def _step_function(
    number: int,
):
    return {
        1: _run_step_1,
        2: _run_step_2,
        3: _run_step_3,
        4: _run_step_4,
        5: _run_step_5,
    }[number]


def _first_incomplete_step(
    state: dict[str, Any],
) -> int | None:
    for step in STEPS:
        step_state = state[
            "steps"
        ][
            str(step["number"])
        ]

        if step_state.get(
            "status"
        ) != "success":
            return step["number"]

    return None


def _mark_step_started(
    state: dict[str, Any],
    step_number: int,
) -> None:
    step_state = state[
        "steps"
    ][
        str(step_number)
    ]

    step_state["status"] = "running"
    step_state["started_at"] = utc_now()
    step_state["completed_at"] = None
    step_state["error"] = None

    state["status"] = "running"
    state["current_step"] = step_number
    state["current_step_name"] = STEPS[
        step_number - 1
    ][
        "name"
    ]

    _write_state(
        state
    )


def _mark_step_success(
    state: dict[str, Any],
    step_number: int,
    result: Any,
) -> None:
    step_state = state[
        "steps"
    ][
        str(step_number)
    ]

    step_state["status"] = "success"
    step_state["completed_at"] = utc_now()
    step_state["error"] = None
    step_state["result"] = (
        _result_to_jsonable(
            result
        )
    )

    state["last_completed_step"] = max(
        state.get(
            "last_completed_step",
            0,
        ),
        step_number,
    )

    state["error"] = None

    _write_state(
        state
    )


def _mark_step_failed(
    state: dict[str, Any],
    step_number: int,
    error: Exception,
) -> None:
    step_state = state[
        "steps"
    ][
        str(step_number)
    ]

    step_state["status"] = "failed"
    step_state["completed_at"] = utc_now()
    step_state["error"] = str(
        error
    )

    state["status"] = "failed"
    state["error"] = str(
        error
    )
    state["current_step"] = step_number
    state["current_step_name"] = STEPS[
        step_number - 1
    ][
        "name"
    ]

    _write_state(
        state
    )


def _run_pipeline_job(
    project_id: str,
) -> None:
    """
    Worker function.

    It intentionally executes the five heavy stages sequentially.

    A process restart does not destroy progress information because state.json
    is checkpointed after each step. The resume route uses this same state.
    """
    logger = ProjectLogger(
        project_id
    )

    state = _read_state(
        project_id
    )

    if not state:
        logger.write(
            "Cannot start pipeline: project state missing.",
            logging.ERROR,
        )
        return

    source_file_value = state.get(
        "source_file"
    )

    if not source_file_value:
        state["status"] = "failed"
        state["error"] = (
            "source_file missing from project state."
        )

        _write_state(
            state
        )

        logger.write(
            "Pipeline aborted: source_file missing.",
            logging.ERROR,
        )

        return

    source_file = Path(
        source_file_value
    )

    if not source_file.exists():
        state["status"] = "failed"
        state["error"] = (
            f"Source file not found: {source_file}"
        )

        _write_state(
            state
        )

        logger.write(
            state["error"],
            logging.ERROR,
        )

        return

    start_step = (
        _first_incomplete_step(
            state
        )
    )

    if start_step is None:
        state["status"] = "completed"
        state["current_step"] = None
        state["current_step_name"] = None
        state["completed_at"] = (
            state.get(
                "completed_at"
            )
            or utc_now()
        )

        _write_state(
            state
        )

        logger.write(
            "Project already completed; nothing to do."
        )

        return

    _record_event(
        state,
        logger,
        step_log_message(
            None,
            "PROJECT",
            f"Pipeline starting from Step {start_step:03d} "
            f"at {ist_now_display()}.",
        ),
    )

    try:
        for number in range(
            start_step,
            6,
        ):
            step = STEPS[
                number - 1
            ]

            step_name = step[
                "name"
            ]

            _mark_step_started(
                state,
                number,
            )

            _record_event(
                state,
                logger,
                step_log_message(
                    number,
                    "STARTED",
                    f"{step_name} | started_at={ist_now_display()}",
                ),
            )

            try:
                if number == 1:
                    result = _step_function(
                        number
                    )(
                        project_id,
                        source_file,
                    )
                else:
                    result = _step_function(
                        number
                    )(
                        project_id,
                    )

            except Exception as exc:
                _mark_step_failed(
                    state,
                    number,
                    exc,
                )

                _record_event(
                    state,
                    logger,
                    step_log_message(
                        number,
                        "FAILED",
                        f"{step_name} | error={exc} | "
                        f"failed_at={ist_now_display()}",
                    ),
                    level=logging.ERROR,
                )

                # Notify the optional webhook: this step's detailed logs +
                # every previous log line for this project.
                _send_webhook(
                    state,
                    logger,
                    event="step_failed",
                    step_number=number,
                    step_state=state["steps"][str(number)],
                )

                return

            _mark_step_success(
                state,
                number,
                result,
            )

            _record_event(
                state,
                logger,
                step_log_message(
                    number,
                    "COMPLETED",
                    f"{step_name} | completed_at={ist_now_display()}",
                ),
            )

            # Notify the optional webhook: this step's detailed logs +
            # every previous log line for this project so far.
            _send_webhook(
                state,
                logger,
                event="step_completed",
                step_number=number,
                step_state=state["steps"][str(number)],
            )

        state["status"] = "completed"
        state["current_step"] = None
        state["current_step_name"] = None
        state["completed_at"] = utc_now()
        state["error"] = None

        _write_state(
            state
        )

        _record_event(
            state,
            logger,
            step_log_message(
                None,
                "SUCCESS",
                "PIPELINE COMPLETED SUCCESSFULLY at "
                f"{ist_now_display()}.",
            ),
        )

        # Final webhook: all logs for the project + subtitle file path /
        # download link.
        _send_webhook(
            state,
            logger,
            event="pipeline_completed",
        )

    finally:
        with ACTIVE_PROJECTS_LOCK:
            ACTIVE_PROJECTS.discard(
                project_id
            )


# ============================================================================
# JOB SUBMISSION
# ============================================================================

def _submit_pipeline(
    project_id: str,
) -> None:
    with ACTIVE_PROJECTS_LOCK:
        if (
            project_id
            in ACTIVE_PROJECTS
        ):
            raise RuntimeError(
                "Project is already running."
            )

        ACTIVE_PROJECTS.add(
            project_id
        )

    try:
        PIPELINE_EXECUTOR.submit(
            _run_pipeline_job,
            project_id,
        )
    except Exception:
        with ACTIVE_PROJECTS_LOCK:
            ACTIVE_PROJECTS.discard(
                project_id
            )
        raise


# ============================================================================
# API ROUTES
# ============================================================================

@app.get(
    "/",
)
async def root() -> dict[str, Any]:
    """
    Root endpoint — simple service banner + a directory of available
    routes, so hitting the base URL is never a 404.
    """
    with ACTIVE_PROJECTS_LOCK:
        active = sorted(
            ACTIVE_PROJECTS
        )

    return {
        "service": "Subtitle Pipeline Service",
        "version": app.version,
        "status": "ok",
        "message": (
            "Subtitle Pipeline API is running. "
            "See /docs for interactive API documentation."
        ),
        "server_time_ist": ist_now_display(),
        "active_projects": active,
        "endpoints": {
            "root": "GET /",
            "health": "GET /health",
            "docs": "GET /docs",
            "create_project": "POST /projects",
            "project_status": "GET /projects/{project_id}",
            "project_logs": "GET /projects/{project_id}/logs",
            "resume_project": "POST /projects/{project_id}/resume",
            "download_subtitle": "GET /projects/{project_id}/subtitle",
        },
    }


@app.get(
    "/health",
)
async def health() -> dict[str, Any]:
    with ACTIVE_PROJECTS_LOCK:
        active = sorted(
            ACTIVE_PROJECTS
        )

    return {
        "status": "ok",
        "service": "subtitle-pipeline",
        "active_projects": active,
        "max_concurrent_pipelines": (
            MAX_CONCURRENT_PIPELINES
        ),
        "output_root": str(
            OUTPUT_ROOT.resolve()
        ),
    }


@app.post(
    "/projects",
    status_code=status.HTTP_202_ACCEPTED,
)
async def create_project(
    request: CreateProjectRequest,
) -> dict[str, Any]:
    project_id = validate_project_id(
        request.project_id
    )

    source_url = str(
        request.file_url
    )

    webhook_url = (
        str(
            request.webhook_url
        )
        if request.webhook_url
        else None
    )

    paths = project_paths(
        project_id
    )

    service_dir = paths[
        "service"
    ]

    service_dir.mkdir(
        parents=True,
        exist_ok=True,
    )

    existing = _read_state(
        project_id
    )

    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=(
                f"Project '{project_id}' already exists. "
                "Use GET for status or POST /resume to continue it."
            ),
        )

    logger = ProjectLogger(
        project_id
    )

    state = _initial_state(
        project_id,
        source_url,
        webhook_url,
    )

    _write_state(
        state
    )

    logger.write_step(
        None,
        "PROJECT",
        (
            f"Project created: {project_id} | source_url={source_url} | "
            "webhook_url="
            + (
                webhook_url
                if webhook_url
                else "(none)"
            )
        ),
    )

    try:
        source_file = await asyncio.to_thread(
            download_source_file,
            project_id,
            source_url,
            logger,
        )

        state = _read_state(
            project_id
        )

        if not state:
            raise RuntimeError(
                "Project state disappeared after download."
            )

        state["source_file"] = str(
            source_file.resolve()
        )
        state["status"] = "queued"
        state["error"] = None

        _write_state(
            state
        )

        _record_event(
            state,
            logger,
            step_log_message(
                None,
                "PROJECT",
                "Source media downloaded successfully. "
                f"Submitting pipeline at {ist_now_display()}.",
            ),
        )

        _submit_pipeline(
            project_id
        )

    except Exception as exc:
        state = _read_state(
            project_id
        ) or state

        state["status"] = "failed"
        state["error"] = str(
            exc
        )

        _write_state(
            state
        )

        logger.write_step(
            None,
            "FAILED",
            f"PROJECT CREATION FAILED | {exc} | "
            f"failed_at={ist_now_display()}",
            level=logging.ERROR,
        )

        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(
                exc
            ),
        ) from exc

    return {
        "project_id": project_id,
        "status": "queued",
        "source_file": (
            state["source_file"]
        ),
        "status_url": (
            f"/projects/{project_id}"
        ),
        "logs_url": (
            f"/projects/{project_id}/logs"
        ),
        "resume_url": (
            f"/projects/{project_id}/resume"
        ),
    }


@app.get(
    "/projects/{project_id}",
)
async def get_project(
    project_id: str,
) -> dict[str, Any]:
    project_id = validate_project_id(
        project_id
    )

    state = _read_state(
        project_id
    )

    if not state:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=(
                f"Project '{project_id}' does not exist."
            ),
        )

    with ACTIVE_PROJECTS_LOCK:
        running = (
            project_id
            in ACTIVE_PROJECTS
        )

    state["runtime"] = {
        "running_in_this_process": running,
    }

    return state


@app.get(
    "/projects/{project_id}/logs",
)
async def get_project_logs(
    project_id: str,
    limit: int = 100,
) -> dict[str, Any]:
    project_id = validate_project_id(
        project_id
    )

    state = _read_state(
        project_id
    )

    if not state:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=(
                f"Project '{project_id}' does not exist."
            ),
        )

    limit = max(
        1,
        min(
            limit,
            MAX_LOG_LINES,
        ),
    )

    logger = ProjectLogger(
        project_id
    )

    return {
        "project_id": project_id,
        "log_file": str(
            logger.log_path.resolve()
        ),
        "lines": logger.recent_lines(
            limit
        ),
    }


@app.post(
    "/projects/{project_id}/resume",
    response_model=ResumeProjectResponse,
)
async def resume_project(
    project_id: str,
) -> ResumeProjectResponse:
    project_id = validate_project_id(
        project_id
    )

    state = _read_state(
        project_id
    )

    if not state:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=(
                f"Project '{project_id}' does not exist."
            ),
        )

    if not state.get(
        "source_file"
    ):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=(
                "Cannot resume because the source media file "
                "is missing from project state."
            ),
        )

    if not Path(
        state["source_file"]
    ).exists():
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=(
                "Cannot resume because the downloaded source file "
                "no longer exists."
            ),
        )

    with ACTIVE_PROJECTS_LOCK:
        if project_id in ACTIVE_PROJECTS:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=(
                    "Project is already running."
                ),
            )

    start_step = _first_incomplete_step(
        state
    )

    if start_step is None:
        return ResumeProjectResponse(
            project_id=project_id,
            status="completed",
            resumed_from_step=None,
            message=(
                "Project is already complete."
            ),
        )

    # Reset the failed step to pending. Successful previous steps stay success.
    step_state = state[
        "steps"
    ][
        str(start_step)
    ]

    step_state["status"] = "pending"
    step_state["error"] = None
    step_state["started_at"] = None
    step_state["completed_at"] = None

    state["status"] = "queued"
    state["error"] = None
    state["current_step"] = start_step
    state["current_step_name"] = STEPS[
        start_step - 1
    ][
        "name"
    ]

    _write_state(
        state
    )

    logger = ProjectLogger(
        project_id
    )

    _record_event(
        state,
        logger,
        step_log_message(
            start_step,
            "RESUME",
            f"Restarting from Step {start_step:03d} | "
            f"{STEPS[start_step - 1]['name']} | "
            f"requested_at={ist_now_display()}",
        ),
    )

    try:
        _submit_pipeline(
            project_id
        )
    except RuntimeError as exc:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(
                exc
            ),
        ) from exc

    return ResumeProjectResponse(
        project_id=project_id,
        status="queued",
        resumed_from_step=start_step,
        message=(
            f"Project queued for resume from Step {start_step:03d}."
        ),
    )



@app.get("/projects/{project_id}/subtitle")
async def get_project_subtitle(project_id: str):
    project_id = validate_project_id(project_id)

    state = _read_state(project_id)

    if not state:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail={
                "status": "not_found",
                "message": f"Project '{project_id}' does not exist.",
            },
        )

    # ---------------------------------------------------------
    # Get the exact SRT path generated by Step 5
    # ---------------------------------------------------------
    step_5 = state.get("steps", {}).get("5", {})
    step_5_result = step_5.get("result") or {}

    srt_path_value = step_5_result.get("srt_path")

    # ---------------------------------------------------------
    # If Step 5 already generated the SRT, use its exact path
    # ---------------------------------------------------------
    if srt_path_value:
        srt_path = Path(srt_path_value)

        if srt_path.exists() and srt_path.is_file():
            return FileResponse(
                path=str(srt_path),
                media_type="application/x-subrip",
                filename=srt_path.name,
                headers={
                    "Content-Disposition": (
                        f'attachment; filename="{srt_path.name}"'
                    )
                },
            )

    # ---------------------------------------------------------
    # Fallback: search the subtitles directory for any .srt file
    # This handles cases where Step 5 did not save srt_path.
    # ---------------------------------------------------------
    subtitles_dir = (
        OUTPUT_ROOT
        / project_id
        / "subtitles"
    )

    if subtitles_dir.exists():
        srt_files = sorted(
            subtitles_dir.glob("*.srt"),
            key=lambda path: path.stat().st_mtime,
            reverse=True,
        )

        if srt_files:
            srt_path = srt_files[0]

            return FileResponse(
                path=str(srt_path),
                media_type="application/x-subrip",
                filename=srt_path.name,
                headers={
                    "Content-Disposition": (
                        f'attachment; filename="{srt_path.name}"'
                    )
                },
            )

    # ---------------------------------------------------------
    # SRT not available yet
    # ---------------------------------------------------------
    logger = ProjectLogger(project_id)

    return {
        "status": "not_ready",
        "message": "Subtitle file has not been created yet.",
        "project_id": project_id,
        "project_status": state.get("status"),
        "current_step": state.get("current_step"),
        "current_step_name": state.get("current_step_name"),
        "last_completed_step": state.get("last_completed_step", 0),
        "subtitle_exists": False,
        "error": state.get("error"),
        "logs": logger.recent_lines(20),
    }


# ============================================================================
# OPTIONAL SHUTDOWN
# ============================================================================

# @app.on_event(
#     "shutdown"
# )
# def shutdown_event() -> None:
#     PIPELINE_EXECUTOR.shutdown(
#         wait=False,
#         cancel_futures=False,
#     )