app/services/photo_vision.py

"""Photo retrieval and optional vision analysis for section generation."""

from __future__ import annotations

import asyncio
import base64
import json
import logging
from datetime import UTC, datetime
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.db.models import ReportSectionPhoto
from app.storage.photo_paths import resolve_report_photo_path

logger = logging.getLogger(__name__)

_VISION_CACHE_FILENAME = "_ai_vision_observations.json"


def _section_photo_vision_model() -> str:
    explicit = (getattr(settings, "section_photo_vision_model", None) or "").strip()
    if explicit:
        return explicit
    return (settings.chat_model or "").strip() or "gpt-4o"


def _vision_cache_path(tenant_id: str, report_id: str, section_code: str) -> Path:
    return (
        settings.upload_dir
        / tenant_id
        / "report_photos"
        / str(report_id)
        / str(section_code)
        / _VISION_CACHE_FILENAME
    )


def invalidate_section_photo_vision_cache(
    tenant_id: str,
    report_id: str,
    section_code: str,
) -> None:
    """Drop cached vision output after new uploads."""
    path = _vision_cache_path(tenant_id, report_id, section_code)
    try:
        if path.is_file():
            path.unlink()
    except OSError as exc:
        logger.debug("Could not remove photo vision cache %s: %s", path, exc)


def _photo_paths_from_rows(
    rows: list[ReportSectionPhoto],
    *,
    tenant_id: str,
    report_id: str,
    section_code: str,
) -> list[tuple[Path, str]]:
    photo_paths: list[tuple[Path, str]] = []
    for r in rows:
        resolved = resolve_report_photo_path(
            r.file_path,
            tenant_id=tenant_id,
            report_id=report_id,
            section_code=section_code,
            photo_id=r.id,
        )
        if resolved:
            photo_paths.append((Path(resolved), r.content_type))
    return photo_paths


def _data_url_for_image(path: Path, content_type: str) -> str:
    raw = path.read_bytes()
    b64 = base64.b64encode(raw).decode("ascii")
    ct = (content_type or "image/jpeg").strip().lower()
    if not ct.startswith("image/"):
        ct = f"image/{ct}"
    return f"data:{ct};base64,{b64}"


def _openai_vision_image_part(path: Path, content_type: str) -> dict:
    """Chat Completions vision part (not Responses API input_image format)."""
    return {
        "type": "image_url",
        "image_url": {"url": _data_url_for_image(path, content_type), "detail": "high"},
    }


def _openai_vision_user_message(prompt: str, image_parts: list[dict]) -> dict:
    return {
        "role": "user",
        "content": [{"type": "text", "text": prompt}, *image_parts],
    }


def _dedupe_observation_lines(lines: list[str], *, max_items: int) -> list[str]:
    seen: set[str] = set()
    out: list[str] = []
    for ln in lines:
        key = ln.lower().strip()[:160]
        if not key or key in seen:
            continue
        seen.add(key)
        out.append(ln)
        if len(out) >= max_items:
            break
    return out


def _read_vision_cache(
    path: Path,
    *,
    photo_ids: list[str],
) -> tuple[list[str], str | None] | None:
    if not path.is_file():
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(payload, dict):
        return None
    cached_ids = payload.get("photo_ids")
    if not isinstance(cached_ids, list) or sorted(str(x) for x in cached_ids) != sorted(photo_ids):
        return None
    obs = payload.get("observations")
    note = payload.get("note")
    observations = [str(x) for x in obs] if isinstance(obs, list) else []
    note_out = str(note) if isinstance(note, str) and note.strip() else None
    return observations, note_out


def _write_vision_cache(
    path: Path,
    *,
    photo_ids: list[str],
    observations: list[str],
    note: str | None,
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "photo_ids": photo_ids,
        "observations": observations,
        "note": note,
        "analyzed_at": datetime.now(UTC).isoformat(),
    }
    path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")


async def get_section_photo_rows(
    db: AsyncSession,
    *,
    tenant_id: str,
    report_id: str,
    section_code: str,
) -> list[ReportSectionPhoto]:
    res = await db.execute(
        select(ReportSectionPhoto)
        .where(
            ReportSectionPhoto.tenant_id == tenant_id,
            ReportSectionPhoto.report_id == report_id,
            ReportSectionPhoto.section_code == section_code,
        )
        .order_by(ReportSectionPhoto.created_at.asc())
    )
    return list(res.scalars().all())


async def _vision_batch_call_async(
    *,
    image_parts: list[dict],
    prompt: str,
    api_key: str,
    model: str,
    section_code: str,
) -> str:
    from app.llm.openai_chat import chat_completions_create_raw

    return await chat_completions_create_raw(
        messages=[
            {"role": "system", "content": "You are a careful building inspection assistant."},
            _openai_vision_user_message(prompt, image_parts),
        ],
        model=model,
        max_tokens=800,
        temperature=0.1,
        phase="photo_vision",
        section_id=section_code,
        api_key=api_key,
    )


async def _vision_batches_async(
    *,
    photo_paths: list[tuple[Path, str]],
    section_code: str,
    api_key: str,
    model: str,
    batch_size: int,
    max_observations: int,
) -> tuple[list[str], str | None]:
    """Run vision batches with throttled async OpenAI calls."""
    if not photo_paths:
        return [], "Photos were uploaded for this section, but they could not be read; content is text-led."

    n_files = len(photo_paths)
    bs = max(1, int(batch_size))
    num_batches = (n_files + bs - 1) // bs
    merged: list[str] = []

    for b in range(num_batches):
        start = b * bs
        chunk = photo_paths[start : start + bs]
        image_parts: list[dict] = []
        for p, ct in chunk:
            try:
                image_parts.append(_openai_vision_image_part(p, ct))
            except Exception as exc:  # noqa: BLE001
                logger.warning("Could not read photo for vision: %s", exc)
        if not image_parts:
            continue

        subset_lo = start + 1
        subset_hi = start + len(image_parts)
        if num_batches > 1:
            scope = (
                f"This is subset {b + 1} of {num_batches} ({subset_lo}-{subset_hi} of {n_files} images for this section). "
                "Describe only what is visible in these images; avoid duplicating generic section headers.\n"
            )
        else:
            scope = ""

        prompt = (
            "You are assisting a UK residential surveyor writing a RICS Home Survey report section.\n"
            f"RICS section code: {section_code}.\n"
            f"{scope}"
            "Analyze the photos and produce 3–10 concise bullet observations for this set.\n"
            "Rules:\n"
            "- Only state what is visible.\n"
            "- If uncertain, say 'unclear' and suggest a verification.\n"
            "- Use surveyor tone and building terminology.\n"
            "- Do not invent measurements.\n"
            "- Focus on condition/defects/materials/obvious hazards.\n"
        )

        text = await _vision_batch_call_async(
            image_parts=image_parts,
            prompt=prompt,
            api_key=api_key,
            model=model,
            section_code=section_code,
        )
        for line in text.splitlines():
            ln = line.strip().lstrip("-•").strip()
            if ln:
                merged.append(ln)

    if not merged:
        return [], "Photos were uploaded, but no usable observations were produced."

    return _dedupe_observation_lines(merged, max_items=max_observations), None


async def run_section_photo_vision_analysis(
    db: AsyncSession,
    *,
    tenant_id: str,
    report_id: str,
    section_code: str,
) -> tuple[list[str], str | None]:
    """Run vision for all photos in a section and persist observations to the section cache."""
    rows = await get_section_photo_rows(
        db,
        tenant_id=tenant_id,
        report_id=str(report_id),
        section_code=str(section_code),
    )
    if not rows:
        return [], None

    if not settings.section_photo_vision_enabled or not (settings.openai_api_key or "").strip():
        return [], "Photos were uploaded for this section, but photo analysis is unavailable; content is text-led."

    photo_ids = [str(r.id) for r in rows]
    cache_path = _vision_cache_path(tenant_id, str(report_id), str(section_code))
    cached = _read_vision_cache(cache_path, photo_ids=photo_ids)
    if cached is not None:
        return cached

    photo_paths = _photo_paths_from_rows(
        rows,
        tenant_id=tenant_id,
        report_id=str(report_id),
        section_code=str(section_code),
    )
    if not photo_paths:
        note = "Photos were uploaded for this section, but they could not be read; content is text-led."
        _write_vision_cache(cache_path, photo_ids=photo_ids, observations=[], note=note)
        return [], note

    model = _section_photo_vision_model()
    try:
        observations, note = await _vision_batches_async(
            photo_paths=photo_paths,
            section_code=str(section_code),
            api_key=settings.openai_api_key or "",
            model=model,
            batch_size=int(settings.section_photo_vision_batch_size),
            max_observations=int(settings.section_photo_vision_max_observations),
        )
        _write_vision_cache(
            cache_path,
            photo_ids=photo_ids,
            observations=observations,
            note=note,
        )
        if observations:
            logger.info(
                "Photo vision complete section=%s report=%s images=%d observations=%d",
                section_code,
                report_id,
                len(photo_paths),
                len(observations),
            )
        elif note:
            logger.warning(
                "Photo vision note section=%s report=%s: %s",
                section_code,
                report_id,
                note,
            )
        return observations, note
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "Photo vision failed section=%s report=%s: %s",
            section_code,
            report_id,
            exc,
        )
        note = "Photos were uploaded for this section, but automated photo analysis failed; content is text-led."
        _write_vision_cache(cache_path, photo_ids=photo_ids, observations=[], note=note)
        return [], note


def schedule_section_photo_vision_analysis(
    *,
    tenant_id: str,
    report_id: str,
    section_code: str,
) -> None:
    """Fire-and-forget vision after upload (uses its own DB session)."""
    if not settings.section_photo_analyze_on_upload:
        return
    if not settings.section_photo_vision_enabled or not (settings.openai_api_key or "").strip():
        return

    async def _job() -> None:
        from app.db.database import get_session_factory

        factory = get_session_factory()
        try:
            async with factory() as db:
                await run_section_photo_vision_analysis(
                    db,
                    tenant_id=tenant_id,
                    report_id=str(report_id),
                    section_code=str(section_code),
                )
        except Exception:  # noqa: BLE001
            logger.exception(
                "Background photo vision failed report=%s section=%s",
                report_id,
                section_code,
            )

    try:
        asyncio.get_running_loop().create_task(_job())
    except RuntimeError:
        asyncio.run(_job())


async def get_photo_observations_for_section(
    db: AsyncSession | None,
    *,
    tenant_id: str,
    report_id: str | None,
    section_code: str,
) -> tuple[list[str], str | None]:
    """Return (observations, note) from cache or a live vision run."""
    if db is None or not report_id:
        return [], None

    rows = await get_section_photo_rows(
        db,
        tenant_id=tenant_id,
        report_id=str(report_id),
        section_code=str(section_code),
    )
    if not rows:
        return [], None

    if not settings.section_photo_vision_enabled or not (settings.openai_api_key or "").strip():
        return [], "Photos were uploaded for this section, but photo analysis is unavailable; content is text-led."

    return await run_section_photo_vision_analysis(
        db,
        tenant_id=tenant_id,
        report_id=str(report_id),
        section_code=str(section_code),
    )


async def enrich_bullets_with_section_photos(
    db: AsyncSession,
    *,
    tenant_id: str,
    report_id: str,
    section_code: str,
    bullets: list[str],
    survey_level: int | None = None,
) -> list[str]:
    """Append photo-policy notes and vision observations to section bullets (generation + agentic)."""
    from app.services.photo_policy import PhotoPolicy, classify_section_photo_policy_async

    out = list(bullets)
    try:
        photo_policy, _ = await classify_section_photo_policy_async(
            db, tenant_id, section_code, survey_level=survey_level
        )
    except Exception:  # noqa: BLE001
        photo_policy = None

    try:
        photo_obs, photo_note = await get_photo_observations_for_section(
            db,
            tenant_id=tenant_id,
            report_id=report_id,
            section_code=section_code,
        )
    except Exception:  # noqa: BLE001
        photo_obs, photo_note = [], None

    if photo_policy == PhotoPolicy.requires_image and not photo_obs and not photo_note:
        out.extend(
            [
                "We were unable to verify this during inspection. "
                "No photos were provided for this section; confirm condition via site inspection "
                "or photo evidence before sign-off."
            ]
        )
    elif photo_note:
        out.append(f"Photo policy note: {photo_note}")
    elif photo_obs:
        out.extend(["Photo observations:"] + [f"- {x}" for x in photo_obs])
    return out