"""ImageService/ReferenceImageService/SceneImageService 공용 helper.

Phase 3b.2에서 ImageService의 네 private 메서드(`_image_to_dict`,
`_get_latest_world_guide`, `_auto_set_primary`, `_build_lineage_fields`)를
module-level 함수로 승격. ReferenceImageService가 독립 클래스로 이관되어도
동일 helper를 공유할 수 있게 한다. SceneImageService(Phase 3b.3)도 재사용 예정.

기존 ImageService의 instance 메서드(`self._xxx`)는 내부적으로 본 모듈의 함수로
위임하는 shim으로 유지된다 — api/v1/images.py 및 테스트 호환.
"""
from __future__ import annotations

import json
import logging
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)

from sqlalchemy.orm import Session as OrmSession

from app.core.file_paths import to_relative_image_path
from app.core.version_registry import get_module_info
from app.models.project import ImageAsset, ProjectSettings, WorldGuide


__all__ = [
    "image_to_dict",
    "get_latest_world_guide",
    "auto_set_primary",
    "build_lineage_fields",
    "load_project_llm_config",
    "fill_missing_t2i_prompts",
    "populate_t2i_prompts",
]


def populate_t2i_prompts(
    db: OrmSession,
    stills: List[Dict[str, Any]],
    stills_orm: List[Any],
    entities: List[Dict[str, Any]],
    openai_client: Any,
) -> None:
    """Populate T2I prompt fields on stills from ORM + runtime fallback.

    W5 F22 Phase B.20 (2026-04-22): scene_image_service.generate_images의
    T2I loader 블록을 이관. 처리:
      1) stills_orm에서 t2i_prompt_cinematic/closeup + t2i_variations_json(JSON decode) 로드
      2) cinematic 없고 variations도 없는 still만 T2IVisualConverter로 런타임 생성
      3) convert_scenes 결과를 still_data에 쓰고 db.commit

    stills와 stills_orm은 인덱스로 1:1 매칭. 이후 caller는
    ``fill_missing_t2i_prompts(stills)``로 최종 fallback을 적용한다.
    Mutates stills in-place. 반환값 없음.
    """
    has_missing_t2i = False
    for si, still_data in enumerate(stills):
        still_orm = stills_orm[si]
        still_data["t2i_prompt_cinematic"] = still_orm.t2i_prompt_cinematic or ""
        still_data["t2i_prompt_closeup"] = still_orm.t2i_prompt_closeup or ""
        try:
            still_data["t2i_variations"] = (
                json.loads(still_orm.t2i_variations_json)
                if still_orm.t2i_variations_json
                else []
            )
        except (json.JSONDecodeError, TypeError):
            still_data["t2i_variations"] = []
        if not still_data["t2i_prompt_cinematic"]:
            has_missing_t2i = True

    if not has_missing_t2i:
        return

    # t2i_variations_json에 T2I가 있으면 레거시 변환기 불필요
    actually_missing = [
        s for s in stills
        if not s.get("t2i_prompt_cinematic") and not s.get("t2i_variations")
    ]
    if not actually_missing:
        return

    logger.warning(
        "T2I prompts missing for %d stills — generating at runtime",
        len(actually_missing),
    )
    from app.modules.t2i_visual_converter import T2IVisualConverter
    converter = T2IVisualConverter(llm_client=openai_client)
    entity_t2i_map = {e["id"]: e.get("t2i_prompt", "") for e in entities}
    scene_dicts = [
        {
            "id": s["id"],
            "still_frame_prompt": s.get("still_frame_prompt", ""),
            "screenplay_scene_heading": s.get("screenplay_scene_heading", ""),
            "visible_entities_json": s.get("visible_entities_json", "[]"),
        }
        for s in actually_missing
    ]
    scene_t2i_map = converter.convert_scenes(scene_dicts, entity_t2i_map)
    for still_data in stills:
        if not still_data["t2i_prompt_cinematic"] and still_data["id"] in scene_t2i_map:
            t2i = scene_t2i_map[still_data["id"]]
            if isinstance(t2i, dict):
                still_data["t2i_prompt_cinematic"] = t2i.get("a", "")
                still_data["t2i_prompt_closeup"] = t2i.get("b", "")
            elif isinstance(t2i, str):
                still_data["t2i_prompt_cinematic"] = t2i
    db.commit()


def fill_missing_t2i_prompts(stills: List[Dict[str, Any]]) -> None:
    """Fill missing t2i_prompt_cinematic / t2i_prompt_closeup on stills in-place.

    W5 F22 Phase B.18 (2026-04-22): scene_image_service.generate_images의
    fallback 블록을 이관. 각 still에 대해 3단 fallback:
      1) t2i_variations[0/1].t2i_prompt (이미 파싱된 리스트)
      2) still_frame_prompt 원본
      3) 그대로 둠 (빈 문자열)

    Mutates stills in-place. 반환값 없음.
    """
    for still_data in stills:
        if not still_data.get("t2i_prompt_cinematic"):
            # t2i_variations에서 첫 번째 변형 사용 (이미 파싱됨)
            t2i_vars = still_data.get("t2i_variations", [])
            if t2i_vars and isinstance(t2i_vars, list):
                still_data["t2i_prompt_cinematic"] = t2i_vars[0].get("t2i_prompt", "")
                if len(t2i_vars) > 1:
                    still_data["t2i_prompt_closeup"] = t2i_vars[1].get("t2i_prompt", "")
        if not still_data.get("t2i_prompt_cinematic"):
            still_data["t2i_prompt_cinematic"] = still_data.get("still_frame_prompt", "")
        if not still_data.get("t2i_prompt_closeup"):
            still_data["t2i_prompt_closeup"] = still_data.get("still_frame_prompt", "")


def load_project_llm_config(db: OrmSession, project_id: str) -> Dict[str, Any]:
    """프로젝트 LLM 설정 로드. 없으면 빈 dict.

    W5 F22 Phase B.3에서 scene_image_service._load_project_llm_config을
    module-level helper로 승격. scene_variation_service의 recommend_variations
    등에서 공유.
    """
    try:
        ps = db.query(ProjectSettings).filter(ProjectSettings.project_id == project_id).first()
        if ps and ps.llm_config_json:
            return json.loads(ps.llm_config_json)
    except Exception as exc:
        logger.warning("llm_config_json load failed for project %s: %s", project_id, exc)
    return {}


def image_to_dict(
    img: ImageAsset,
    db: Optional[OrmSession] = None,
    *,
    actual_refs_cache: Optional[Dict[str, List[str]]] = None,
) -> Dict[str, Any]:
    """Convert an ImageAsset ORM object to a dict.

    Task 5 (single-vs-batch reference contract §4.4):
    - reference_image_ids = lineage (visible_entities character/prop UUID list)
    - actual_attached_refs = actual labels attached to LLM call (llm_call_log join)

    Resolution order for actual_attached_refs:
      1. actual_refs_cache 전달 시 (list endpoint, N+1 회피) → cache.get(still_id)
      2. db 전달 시 (detail endpoint) → _lookup_actual_refs single query
      3. 둘 다 None → actual_attached_refs = None (backward compat, 기존 caller)

    list endpoint 사용 패턴 (N+1 회피 의무):
        still_ids = [img.still_id for img in images if img.still_id]
        cache = _lookup_actual_refs_batch(db, project_id, episode_id, still_ids)
        return [image_to_dict(img, actual_refs_cache=cache) for img in images]
    """
    # ImageAsset.file_path는 ImagePathType이 ORM read 시 절대 경로로 환원한다.
    # 직렬화 시점에는 다시 상대로 변환하여 절대 호스트 경로 leak을 차단한다 (Codex B2).
    actual_refs: Optional[List[str]] = None
    if actual_refs_cache is not None:
        actual_refs = actual_refs_cache.get(img.still_id or "")
    elif db is not None and img.still_id:
        actual_refs = _lookup_actual_refs(
            db, img.project_id, img.episode_id, img.still_id,
        )

    return {
        "id": img.id,
        "asset_type": img.asset_type,
        # 2a (2026-07-01): image role taxonomy(fp/항공/배경/실내외마네킹) 노출.
        "pipeline_role": getattr(img, "pipeline_role", None),
        # #8 (2026-07-02): 중간과정/변형 구분 — Entities/캔버스가 중간물·거부본을
        # 최종 참조와 동레벨로 섞지 않도록 컬럼 노출 (기존 필드 불변).
        "is_intermediate": bool(getattr(img, "is_intermediate", False)),
        "disposition": getattr(img, "disposition", None),
        "entity_id": img.entity_id,
        "still_id": img.still_id,
        "episode_id": img.episode_id,
        "file_path": to_relative_image_path(img.file_path) if img.file_path else "",
        "prompt_used": img.prompt_used,
        "generation_model": img.generation_model,
        "width": img.width,
        "height": img.height,
        "status": img.status,
        "review_notes": img.review_notes or "",
        "validation_score": img.validation_score,
        "validation_result": img.validation_result,
        "sanitization_strategy": img.sanitization_strategy,
        "original_prompt": img.original_prompt,
        "sanitization_note": img.sanitization_note,
        "variant_type": img.variant_type,
        "angle_applied": img.angle_applied,
        "color_applied": img.color_applied,
        "source_image_id": img.source_image_id,
        "is_primary": bool(img.is_primary),
        "prompt_type": img.prompt_type,
        "code_version": img.code_version,
        "prompt_file_version": img.prompt_file_version,
        "reference_image_ids": img.reference_image_ids or "[]",
        "actual_attached_refs": actual_refs,
        "theme_label": img.theme_label,
        "created_at": img.created_at,
    }


def _lookup_actual_refs(
    db: "OrmSession",
    project_id: str,
    episode_id: str,
    still_id: str,
) -> Optional[List[str]]:
    """detail endpoint — 1 still 의 가장 최근 llm_call_log entry ref labels.

    Task 5 (spec §4.4): operation_type IN ('single_scene_image_gen', 'scene_image_gen')
    + metadata_json.still_id 매칭 + 가장 최근 created_at row.
    log entry 없거나 retention 지나면 None.
    """
    from sqlalchemy import text as _text
    sql = _text("""
        SELECT reference_image_ids FROM llm_call_log
        WHERE project_id = :pid
          AND episode_id = :eid
          AND operation_type IN ('single_scene_image_gen', 'scene_image_gen')
          AND metadata_json::text LIKE :sid_pat
        ORDER BY created_at DESC LIMIT 1
    """)
    try:
        row = db.execute(
            sql, {"pid": project_id, "eid": episode_id, "sid_pat": f'%"still_id": "{still_id}"%'},
        ).first()
    except Exception as exc:
        logger.warning("actual_refs lookup failed for still %s: %s", still_id, exc)
        return None
    if not row:
        return None
    try:
        return json.loads(row[0] or "[]")
    except Exception:
        return None


def _lookup_actual_refs_batch(
    db: "OrmSession",
    project_id: str,
    episode_id: str,
    still_ids: List[str],
) -> Dict[str, List[str]]:
    """list endpoint — episode 전체 still_id 의 latest log entry 1 query 로 lookup.

    Task 5 (spec §4.4) — N+1 회피.
    PostgreSQL DISTINCT ON 으로 still_id 별 가장 최근 created_at row 1개씩.

    Returns: {still_id: ref_labels_list}. log entry 없는 still_id 는 dict 미포함
    (caller 가 cache.get(sid) → None handling).
    Parse 실패 row 는 빈 list 로 isolated (다른 row 영향 없음).
    """
    if not still_ids:
        return {}
    from sqlalchemy import text as _text

    sql = _text("""
        SELECT DISTINCT ON ((metadata_json::jsonb)->>'still_id')
            (metadata_json::jsonb)->>'still_id' AS still_id,
            reference_image_ids
        FROM llm_call_log
        WHERE project_id = :pid
          AND episode_id = :eid
          AND operation_type IN ('single_scene_image_gen', 'scene_image_gen')
          AND (metadata_json::jsonb)->>'still_id' = ANY(:sids)
        ORDER BY (metadata_json::jsonb)->>'still_id', created_at DESC
    """)
    try:
        rows = db.execute(sql, {"pid": project_id, "eid": episode_id, "sids": still_ids}).all()
    except Exception as exc:
        logger.warning("actual_refs batch lookup failed: %s", exc)
        return {}

    result: Dict[str, List[str]] = {}
    for sid, ref_ids_json in rows:
        if sid is None:
            continue
        try:
            result[sid] = json.loads(ref_ids_json or "[]")
        except Exception:
            result[sid] = []
    return result


def get_latest_world_guide(db: OrmSession, project_id: str, episode_id: str) -> Dict[str, Any]:
    """Get the latest world guide for an episode (JSON-parsed)."""
    wg = (
        db.query(WorldGuide)
        .filter(WorldGuide.project_id == project_id, WorldGuide.episode_id == episode_id)
        .order_by(WorldGuide.created_at.desc())
        .first()
    )
    if wg:
        try:
            return json.loads(wg.guide_json)
        except json.JSONDecodeError as exc:
            logger.warning("guide_json parse failed for project %s: %s", project_id, exc)
    return {}


#: 사람이 손으로 올린 자산의 표식 — `image_upload_service` 가 쓰고
#: 여기서 읽는다. **한 자리에서** 정한다(두 곳에 적으면 한쪽만 고쳐진다).
MANUAL_UPLOAD_PROMPT = "uploaded"


def is_manual_upload(img: ImageAsset) -> bool:
    """사람이 손으로 올린 자산인가 — 코드가 남긴 표식을 코드가 읽는다."""
    return (img.prompt_used or "") == MANUAL_UPLOAD_PROMPT


def auto_set_primary(db: OrmSession, project_id: str, img: ImageAsset) -> None:
    """Auto-set the newly created image as primary, unsetting others.

    Composite images (prompt_used starts with [composite:] or [outlook_id:]) are
    excluded — composite 이미지들은 generate_composite_image에서 자체적으로
    is_primary를 관리한다. Composite끼리 / 비-composite끼리만 primary 경쟁.

    ## ★사람이 올린 대표는 파이프라인이 안 내린다 (2026-09-20)

    사용자가 손으로 올린 최종본이 **걷기 한 번에 사라졌다** — 새 이미지를
    한 장도 안 사도 캐시된 산출을 다시 영속하면서 이 함수가 같은 still 의
    형제를 **전부** 내렸기 때문이다(실측: 오전에 올린 8장이 저녁 걷기에
    덮였다. 현재 이 화의 사람 대표는 16장).

    그래서 같은 still 에 **사람이 올린 대표**가 있으면 이 자산을 대표로
    승격하지 않고 그 자리를 지킨다. 새 자산은 **저장은 된다** — 지우지
    않으므로 사용자가 보고 고를 수 있다.

    ★사람이 올린 자산 자신이 들어올 때는 그대로 대표가 된다(사람이
     바꾸려는 것이다).
    ★레버: `settings.image_protect_manual_primary` 를 끄면 종전 동작.
    """
    prompt = img.prompt_used or ""
    is_composite = prompt.startswith("[composite:") or prompt.startswith("[outlook_id:")
    if img.entity_id:
        siblings = (
            db.query(ImageAsset)
            .filter(
                ImageAsset.project_id == project_id,
                ImageAsset.entity_id == img.entity_id,
                ImageAsset.id != img.id,
            )
            .all()
        )
        for sib in siblings:
            sib_prompt = sib.prompt_used or ""
            sib_is_composite = sib_prompt.startswith("[composite:") or sib_prompt.startswith("[outlook_id:")
            if is_composite == sib_is_composite:
                sib.is_primary = 0

    if img.still_id:
        siblings = (
            db.query(ImageAsset)
            .filter(
                ImageAsset.project_id == project_id,
                ImageAsset.still_id == img.still_id,
                ImageAsset.id != img.id,
            )
            .all()
        )
        # ★사람이 올린 대표가 있으면 이 자산은 대표가 되지 않는다.
        _protect = True
        try:
            from app.core.config import settings as _settings

            _protect = bool(getattr(
                _settings, "image_protect_manual_primary", True))
        except Exception:                             # noqa: BLE001
            pass
        if (_protect and not is_manual_upload(img)
                and any(is_manual_upload(s) and s.is_primary
                        for s in siblings)):
            logger.warning(
                "auto_set_primary: still %s 에 **사람이 올린 대표**가 있다 "
                "— 새 자산 %s 는 저장하되 대표로 올리지 않는다",
                img.still_id, img.id)
            img.is_primary = 0
            return
        for sib in siblings:
            sib.is_primary = 0

    img.is_primary = 1


def build_lineage_fields(
    db: OrmSession,
    project_id: str,
    module_name: str,
    ref_entity_ids: Optional[List[str]] = None,
    prompt_type: Optional[str] = None,
) -> Dict[str, Any]:
    """Build lineage fields for an image asset (code_version / reference_image_ids)."""
    info = get_module_info(module_name)
    ref_image_ids: List[str] = []
    if ref_entity_ids:
        for eid in ref_entity_ids:
            ref_img = (
                db.query(ImageAsset)
                .filter(
                    ImageAsset.project_id == project_id,
                    ImageAsset.entity_id == eid,
                    ImageAsset.asset_type == "reference",
                    ImageAsset.is_primary == 1,
                )
                .first()
            )
            if not ref_img:
                ref_img = (
                    db.query(ImageAsset)
                    .filter(
                        ImageAsset.project_id == project_id,
                        ImageAsset.entity_id == eid,
                        ImageAsset.asset_type == "reference",
                    )
                    .order_by(ImageAsset.created_at.desc())
                    .first()
                )
            if ref_img:
                ref_image_ids.append(ref_img.id)

    return {
        "prompt_type": prompt_type,
        "code_version": info["version"],
        "prompt_file_version": info.get("prompt_dependency"),
        "reference_image_ids": json.dumps(ref_image_ids),
    }