services/render_engine/renderer/jobs/manager.py

from __future__ import annotations

import concurrent.futures
import http.client
import json
import shutil
import threading
import urllib.request
from urllib.parse import urlparse
from dataclasses import asdict
from pathlib import Path
from typing import Callable

from renderer.core.config import Settings
from renderer.core.models import AIReelsRequest, JobRecord, RenderRequest
from renderer.core.render_engine import RenderEngine
from renderer.core.security import create_download_token
from renderer.core.utils import new_id, now, read_json, write_json


class JobManager:
    def __init__(self, settings: Settings | None = None) -> None:
        self.settings = settings or Settings()
        self.settings.ensure_dirs()
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=self.settings.max_workers)
        self.lock = threading.Lock()

    def submit_render(self, request: RenderRequest) -> str:
        return self._submit(
            lambda job_id, log: RenderEngine(self.settings, log=log).render(request, job_id),
            callback_url=request.callback_url,
            export_target=request.export_target,
        )

    def submit_ai_reels(self, request: AIReelsRequest) -> str:
        return self._submit(lambda job_id, log: RenderEngine(self.settings, log=log).ai_reels(request, job_id))

    def submit_task(
        self,
        handler: Callable[[str, Callable[[str], None]], object],
        *,
        callback_url: str | None = None,
        export_target: str | None = None,
    ) -> str:
        return self._submit(handler, callback_url=callback_url, export_target=export_target)

    def submit_batch(self, requests: list[RenderRequest]) -> list[str]:
        job_ids: list[str] = []
        for request in requests:
            job_ids.append(self.submit_render(request))
        return job_ids

    def get(self, job_id: str) -> JobRecord:
        data = read_json(self._record_path(job_id), None)
        if data is None:
            raise KeyError(job_id)
        return JobRecord(**data)

    def cancel(self, job_id: str) -> JobRecord:
        record = self.get(job_id)
        if record.state == "PENDING":
            self._update(job_id, state="CANCELLED", failure_reason="Cancelled before execution")
        elif record.state == "RUNNING":
            self._update(job_id, state="CANCEL_REQUESTED", failure_reason="Cancellation requested")
        return self.get(job_id)

    def cleanup(self, older_than_seconds: int | None = None) -> dict[str, int]:
        cutoff = now() - (older_than_seconds or self.settings.job_retention_seconds)
        removed_jobs = 0
        removed_exports = 0
        for path in self.settings.jobs_dir.glob("*.json"):
            record = JobRecord(**read_json(path, {}))
            if record.updated_at >= cutoff or record.state in {"PENDING", "RUNNING", "CANCEL_REQUESTED"}:
                continue
            if record.output_path:
                output = Path(record.output_path)
                if output.exists():
                    output.unlink()
                    removed_exports += 1
            path.unlink(missing_ok=True)
            removed_jobs += 1
        uploads = self.settings.temp_dir / "uploads"
        if uploads.exists():
            shutil.rmtree(uploads, ignore_errors=True)
        return {"removed_jobs": removed_jobs, "removed_exports": removed_exports}

    def summary(self) -> dict:
        records: list[JobRecord] = []
        for path in self.settings.jobs_dir.glob("*.json"):
            try:
                records.append(JobRecord(**read_json(path, {})))
            except Exception:
                continue
        state_counts: dict[str, int] = {}
        for record in records:
            state_counts[record.state] = state_counts.get(record.state, 0) + 1
        active = [
            {
                "job_id": record.job_id,
                "state": record.state,
                "created_at": record.created_at,
                "updated_at": record.updated_at,
                "metrics": record.metrics,
            }
            for record in sorted(records, key=lambda item: item.updated_at, reverse=True)
            if record.state in {"PENDING", "RUNNING", "CANCEL_REQUESTED"}
        ]
        return {
            "total_jobs": len(records),
            "state_counts": state_counts,
            "active_jobs": active,
            "max_workers": self.settings.max_workers,
        }

    def _submit(
        self,
        handler: Callable[[str, Callable[[str], None]], object],
        callback_url: str | None = None,
        export_target: str | None = None,
    ) -> str:
        job_id = new_id()
        record = JobRecord(
            job_id=job_id,
            state="PENDING",
            created_at=now(),
            updated_at=now(),
            download_token=create_download_token(self.settings.signing_secret, job_id),
            callback_url=callback_url,
            export_target=export_target,
        )
        self._save(record)
        self.executor.submit(self._run_with_retries, job_id, handler)
        return job_id

    def _run_with_retries(self, job_id: str, handler: Callable[[str, Callable[[str], None]], object]) -> None:
        attempts = 0
        while attempts < self.settings.max_retries:
            attempts += 1
            if self.get(job_id).state == "CANCELLED":
                self._send_callback(job_id)
                return
            self._update(job_id, state="RUNNING", metrics={"attempt": attempts})
            try:
                if self.get(job_id).state == "CANCEL_REQUESTED":
                    self._update(job_id, state="CANCELLED", failure_reason="Cancelled before render started")
                    self._send_callback(job_id)
                    return
                result = handler(job_id, lambda message: self.append_log(job_id, message))
                record = self.get(job_id)
                output_path = getattr(result, "output_path", None)
                export_path = self._export_copy(output_path, record.export_target, job_id) if output_path else None
                self._update(
                    job_id,
                    state="COMPLETED",
                    output_path=str(output_path) if output_path else None,
                    export_path=export_path,
                    commands=getattr(result, "commands", []),
                    logs=getattr(result, "logs", []),
                    metrics=getattr(result, "metrics", {}) | {"attempt": attempts},
                )
                self._send_callback(job_id)
                return
            except Exception as exc:
                self.append_log(job_id, f"Attempt {attempts} failed: {exc}")
                if attempts >= self.settings.max_retries:
                    self._update(job_id, state="FAILED", failure_reason=str(exc), metrics={"attempt": attempts})
                    self._send_callback(job_id)

    def append_log(self, job_id: str, message: str) -> None:
        with self.lock:
            record = self.get(job_id)
            record.logs.append(message)
            record.updated_at = now()
            self._save(record)

    def _update(self, job_id: str, **changes) -> None:
        with self.lock:
            record = self.get(job_id)
            for key, value in changes.items():
                if key == "metrics" and record.metrics and isinstance(value, dict):
                    record.metrics.update(value)
                else:
                    setattr(record, key, value)
            record.updated_at = now()
            self._save(record)

    def _record_path(self, job_id: str) -> Path:
        return self.settings.jobs_dir / f"{job_id}.json"

    def _save(self, record: JobRecord) -> None:
        write_json(self._record_path(record.job_id), asdict(record))

    def _export_copy(self, output_path: Path, export_target: str | None, job_id: str) -> str | None:
        if not export_target:
            return None
        if export_target != "local":
            if export_target.startswith(("http://", "https://")):
                _put_file(export_target, output_path)
                return export_target
            return None
        target_dir = self.settings.storage_dir / job_id
        target_dir.mkdir(parents=True, exist_ok=True)
        target = target_dir / output_path.name
        shutil.copy2(output_path, target)
        return str(target)

    def _send_callback(self, job_id: str) -> None:
        try:
            record = self.get(job_id)
        except KeyError:
            return
        if not record.callback_url:
            return
        payload = json.dumps(asdict(record), default=str).encode("utf-8")
        request = urllib.request.Request(
            record.callback_url,
            data=payload,
            headers={"Content-Type": "application/json", "User-Agent": "ava2lon-studio-callback/2.0"},
            method="POST",
        )
        try:
            urllib.request.urlopen(request, timeout=10).read()
        except Exception as exc:
            self.append_log(job_id, f"Callback delivery failed: {exc}")


def _put_file(url: str, path: Path) -> None:
    parsed = urlparse(url)
    connection_cls = http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection
    connection = connection_cls(parsed.netloc, timeout=60)
    target = parsed.path or "/"
    if parsed.query:
        target += f"?{parsed.query}"
    headers = {
        "Content-Type": "video/mp4",
        "Content-Length": str(path.stat().st_size),
        "User-Agent": "ava2lon-studio-export/2.0",
    }
    connection.putrequest("PUT", target)
    for key, value in headers.items():
        connection.putheader(key, value)
    connection.endheaders()
    with path.open("rb") as source:
        while True:
            chunk = source.read(1024 * 1024)
            if not chunk:
                break
            connection.send(chunk)
    response = connection.getresponse()
    body = response.read()
    connection.close()
    if response.status >= 400:
        raise RuntimeError(f"Export upload failed with HTTP {response.status}: {body[:500]!r}")