atc-agents/api/presentation_upload.py

"""Upload PPT/PPTX decks and convert to presentation JSON + HTML."""

from __future__ import annotations

import json
import os
import re
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import httpx

PRESENTATIONS_DIR = Path(os.getenv("PRESENTATIONS_DIR", "/data/presentations"))
LIVE_OVERRIDE_DIR = PRESENTATIONS_DIR / "live-override"
DOCLING_URL = os.getenv("DOCLING_URL", "http://docling-serve:5001").rstrip("/")


def _ensure_dir() -> Path:
    PRESENTATIONS_DIR.mkdir(parents=True, exist_ok=True)
    return PRESENTATIONS_DIR


def list_decks() -> list[dict[str, Any]]:
    _ensure_dir()
    decks = []
    for meta_path in sorted(PRESENTATIONS_DIR.glob("*/meta.json"), key=lambda p: p.stat().st_mtime, reverse=True):
        try:
            meta = json.loads(meta_path.read_text())
            decks.append(meta)
        except Exception:
            continue
    return decks


def _safe_name(name: str) -> str:
    return re.sub(r"[^a-zA-Z0-9._-]+", "_", name)[:80]


def pptx_to_slides(path: Path) -> list[dict[str, Any]]:
    from pptx import Presentation
    from pptx.enum.shapes import MSO_SHAPE_TYPE

    prs = Presentation(str(path))
    slides: list[dict[str, Any]] = []
    for idx, slide in enumerate(prs.slides, start=1):
        bullets: list[str] = []
        title = ""
        for shape in slide.shapes:
            if not hasattr(shape, "text"):
                continue
            text = (shape.text or "").strip()
            if not text:
                continue
            if shape.shape_type == MSO_SHAPE_TYPE.PLACEHOLDER and not title:
                title = text.split("\n")[0][:120]
            else:
                for line in text.split("\n"):
                    line = line.strip()
                    if line and line != title:
                        bullets.append(line[:240])
        if not title:
            title = f"Slide {idx}"
        slides.append({
            "id": f"upload-{idx}",
            "title": title,
            "subtitle": "",
            "bullets": bullets[:12] or ["(empty slide)"],
            "kind": "upload",
        })
    return slides


async def docling_enrich(path: Path) -> list[dict[str, Any]] | None:
    """Optional: parse via Docling for richer structure."""
    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            with path.open("rb") as f:
                r = await client.post(
                    f"{DOCLING_URL}/v1/convert/file",
                    files={"files": (path.name, f, "application/octet-stream")},
                    data={"to_formats": "md"},
                )
            if r.status_code >= 400:
                return None
            data = r.json()
            md = ""
            if isinstance(data, dict):
                doc = data.get("document") or data.get("result") or data
                if isinstance(doc, dict):
                    md = doc.get("md_content") or doc.get("markdown") or ""
                elif isinstance(doc, str):
                    md = doc
            if not md:
                return None
            slides = []
            chunks = [c.strip() for c in re.split(r"\n#{1,2}\s+", md) if c.strip()]
            for i, chunk in enumerate(chunks[:40], start=1):
                lines = [ln.strip() for ln in chunk.split("\n") if ln.strip()]
                title = lines[0][:120] if lines else f"Slide {i}"
                bullets = [ln.lstrip("-•* ").strip() for ln in lines[1:13] if ln.strip()]
                slides.append({
                    "id": f"docling-{i}",
                    "title": title,
                    "subtitle": "Docling parsed",
                    "bullets": bullets or ["—"],
                    "kind": "upload",
                })
            return slides if slides else None
    except Exception:
        return None


async def save_upload(filename: str, content: bytes) -> dict[str, Any]:
    _ensure_dir()
    deck_id = str(uuid.uuid4())[:8]
    deck_dir = PRESENTATIONS_DIR / deck_id
    deck_dir.mkdir(parents=True, exist_ok=True)
    safe = _safe_name(filename)
    dest = deck_dir / safe
    dest.write_bytes(content)

    slides: list[dict[str, Any]] = []
    source = "pptx"
    if safe.lower().endswith((".pptx", ".ppt")):
        slides = pptx_to_slides(dest)
        docling_slides = await docling_enrich(dest)
        if docling_slides and len(docling_slides) >= len(slides):
            slides = docling_slides
            source = "docling+pptx"
    else:
        docling_slides = await docling_enrich(dest)
        if docling_slides:
            slides = docling_slides
            source = "docling"

    if not slides:
        slides = [{
            "id": "upload-1",
            "title": safe,
            "subtitle": "Uploaded file",
            "bullets": [f"File stored at {dest.name}", "Could not auto-parse slides — open in editor or re-upload PPTX"],
            "kind": "upload",
        }]

    payload = {
        "id": deck_id,
        "filename": safe,
        "source": source,
        "ts": datetime.now(timezone.utc).isoformat(),
        "title": safe.rsplit(".", 1)[0],
        "subtitle": "Uploaded presentation",
        "slide_count": len(slides),
        "slides": slides,
    }
    (deck_dir / "meta.json").write_text(json.dumps(payload, indent=2, default=str))
    (deck_dir / "deck.json").write_text(json.dumps(payload, default=str))
    return payload


def get_deck(deck_id: str) -> dict[str, Any] | None:
    path = PRESENTATIONS_DIR / deck_id / "meta.json"
    if not path.exists():
        return None
    return json.loads(path.read_text())


def get_live_override() -> dict[str, Any] | None:
    path = LIVE_OVERRIDE_DIR / "meta.json"
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text())
    except Exception:
        return None


def save_live_override(body: dict[str, Any]) -> dict[str, Any]:
    """Persist user edits for the Live Cluster deck."""
    LIVE_OVERRIDE_DIR.mkdir(parents=True, exist_ok=True)
    (LIVE_OVERRIDE_DIR / "assets").mkdir(parents=True, exist_ok=True)
    existing = get_live_override() or {"id": "live", "source": "live-override", "editable": True}
    slides = _clean_slides(body.get("slides"))
    payload = {
        **existing,
        "id": "live",
        "source": "live-override",
        "editable": True,
        "title": str(body.get("title") or existing.get("title") or "Live Cluster")[:120],
        "subtitle": str(body.get("subtitle") or existing.get("subtitle") or "")[:300],
        "ts": datetime.now(timezone.utc).isoformat(),
        "slides": slides,
        "slide_count": len(slides),
    }
    (LIVE_OVERRIDE_DIR / "meta.json").write_text(json.dumps(payload, indent=2, default=str))
    return payload


def clear_live_override() -> bool:
    import shutil
    if not LIVE_OVERRIDE_DIR.exists():
        return True
    shutil.rmtree(LIVE_OVERRIDE_DIR, ignore_errors=True)
    return True


def _clean_slides(incoming_slides: Any) -> list[dict[str, Any]]:
    clean_slides: list[dict[str, Any]] = []
    for i, s in enumerate(incoming_slides or [], start=1):
        if not isinstance(s, dict):
            continue
        bullets = [str(b).strip()[:400] for b in (s.get("bullets") or []) if str(b).strip()]
        slide: dict[str, Any] = {
            "id": str(s.get("id") or f"slide-{i}"),
            "title": str(s.get("title") or f"Slide {i}")[:200],
            "subtitle": str(s.get("subtitle") or "")[:300],
            "bullets": bullets,
            "image": str(s.get("image") or "")[:300],
            "kind": str(s.get("kind") or "narrative")[:40],
        }
        if s.get("animation"):
            slide["animation"] = str(s.get("animation"))[:40]
        clean_slides.append(slide)
    return clean_slides or [_blank_slide(1)]


def _blank_slide(idx: int = 1) -> dict[str, Any]:
    return {
        "id": f"slide-{idx}",
        "title": f"Slide {idx}",
        "subtitle": "",
        "bullets": ["Double-click to edit this point"],
        "image": "",
        "kind": "narrative",
    }


def create_deck(title: str = "Untitled deck", slides: list[dict[str, Any]] | None = None) -> dict[str, Any]:
    """Create a new editable deck (optionally seeded from existing slides)."""
    _ensure_dir()
    deck_id = str(uuid.uuid4())[:8]
    deck_dir = PRESENTATIONS_DIR / deck_id
    (deck_dir / "assets").mkdir(parents=True, exist_ok=True)

    seeded = [dict(s) for s in (slides or [])] or [_blank_slide(1)]
    payload = {
        "id": deck_id,
        "filename": "",
        "source": "editor",
        "editable": True,
        "ts": datetime.now(timezone.utc).isoformat(),
        "title": (title or "Untitled deck").strip()[:120] or "Untitled deck",
        "subtitle": "Editable deck",
        "slide_count": len(seeded),
        "slides": seeded,
    }
    (deck_dir / "meta.json").write_text(json.dumps(payload, indent=2, default=str))
    (deck_dir / "deck.json").write_text(json.dumps(payload, default=str))
    return payload


def save_deck(deck_id: str, body: dict[str, Any]) -> dict[str, Any] | None:
    """Overwrite an existing editable deck's content. Returns None if missing."""
    deck_dir = PRESENTATIONS_DIR / deck_id
    meta_path = deck_dir / "meta.json"
    if not meta_path.exists():
        return None
    existing = json.loads(meta_path.read_text())

    clean_slides = _clean_slides(body.get("slides"))

    existing.update({
        "title": str(body.get("title") or existing.get("title") or "Untitled deck")[:120],
        "subtitle": str(body.get("subtitle") or existing.get("subtitle") or "")[:300],
        "editable": True,
        "source": existing.get("source") or "editor",
        "ts": datetime.now(timezone.utc).isoformat(),
        "slides": clean_slides,
        "slide_count": len(clean_slides),
    })
    meta_path.write_text(json.dumps(existing, indent=2, default=str))
    (deck_dir / "deck.json").write_text(json.dumps(existing, default=str))
    return existing


def delete_deck(deck_id: str) -> bool:
    import shutil
    deck_dir = PRESENTATIONS_DIR / deck_id
    if not deck_dir.exists():
        return False
    shutil.rmtree(deck_dir, ignore_errors=True)
    return True


def save_image(deck_id: str, filename: str, content: bytes) -> dict[str, Any] | None:
    """Store an image in the deck's assets folder; return its served URL."""
    if deck_id == "live":
        deck_dir = LIVE_OVERRIDE_DIR
    else:
        deck_dir = PRESENTATIONS_DIR / deck_id
        if not deck_dir.exists():
            return None
    assets = deck_dir / "assets"
    assets.mkdir(parents=True, exist_ok=True)
    ext = ""
    if "." in filename:
        ext = "." + _safe_name(filename.rsplit(".", 1)[-1].lower())[:8]
    name = f"{uuid.uuid4().hex[:12]}{ext}"
    (assets / name).write_bytes(content)
    return {"name": name, "url": f"/api/presentation/decks/{deck_id}/assets/{name}"}


def get_asset_path(deck_id: str, name: str) -> Path | None:
    safe = _safe_name(name)
    base = LIVE_OVERRIDE_DIR if deck_id == "live" else PRESENTATIONS_DIR / deck_id
    path = base / "assets" / safe
    if not path.exists() or not path.is_file():
        return None
    return path