# -*- coding: utf-8 -*-
"""썰박스 생성 잡 매니저 — subprocess 감독 + 진행 단계 DB 영속화.

원본(o2o-ssulbox/app/jobs.py)에서 이식하되 4가지를 바꿨다:

1. **SSE 제거** — castad 는 폴링을 쓴다. `_subscribers`/`_emit` 을 통째로 걷어냈다.
2. **step 을 DB 에 영속화** — 원본은 인메모리 `_jobs` 에만 뒀지만, 폴링이 진행률을
   돌려주려면 DB 에 있어야 한다. 나중에 워커를 늘려도 폴링은 그대로 동작한다.
3. **세션 팩토리를 BackgroundSessionLocal 로** — 요청용 풀(20+20)을 장시간 잡이
   잠식하지 않게 한다. castad `video_task.py` 와 같은 관행.
4. **좀비 방지** — 원본은 `p.wait()` 에 타임아웃이 없어 엔진이 걸리면 스레드가
   영구 블록되고 `_running` 이 안 줄어 큐가 멎었다. **감시견 타이머**로 데드라인에
   프로세스를 죽인다. `wait(timeout=)` 만 걸면 안 된다 — stdout 읽기 루프가
   EOF 까지 블록하므로 엔진이 조용히 매달리면 wait() 에 도달조차 못 한다
   (2026-07-29 `zzz/_ssul_timeout_verify.py` 로 확인한 실제 결함).

⚠️ **단일 워커 전제** — `_jobs`/`_running` 이 프로세스 로컬이고, `sweep_orphans` 가
"기동 시 비터미널 잡은 전부 고아"라는 불변식에 의존한다. `--workers` 를 늘리면
워커 B 가 기동하며 워커 A 의 정상 잡을 환불·error 처리한다.
"""

import asyncio
import os
import subprocess
import sys
import threading
import time
from collections import deque
from pathlib import Path
from typing import Any, Optional

from app.database.session import BackgroundSessionLocal
from app.ssulbox.constants import (
    DONE_RE,
    JOB_DIR_RE,
    SCENARIO_ENGINE,
    STEP_NAMES,
    STEP_RE,
    STORE_RE,
    gemini_key,
)
from app.ssulbox.services import place_service, task_service
from app.utils.address_parser import extract_region_from_address
from app.utils.logger import get_logger
from config import ssulbox_settings

logger = get_logger("ssulbox")

# ── 프로세스 로컬 상태 ────────────────────────────────────────
#: 진행 중인 잡의 로그·타이밍 (권위 아님 — 권위는 DB status/step)
_jobs: dict[int, dict[str, Any]] = {}
_queue: deque[int] = deque()
_lock = threading.Lock()
_running = 0
_shutting_down = False
#: 앱 이벤트 루프. 워커 스레드가 DB 작업을 위임할 대상
_loop: Optional[asyncio.AbstractEventLoop] = None


def create_job(
    content_id: int,
    scenario: str,
    input_text: str,
    scenes: int,
    seconds: int,
    *,
    store_name: str | None = None,
    road_address: str | None = None,
    address: str | None = None,
) -> None:
    """잡을 큐에 넣는다.

    **반드시 앱 이벤트 루프(요청 핸들러)에서 호출해야 한다** — 여기서 루프를 캡처해
    워커 스레드가 DB 작업을 위임할 때 쓴다.
    """
    global _loop
    _loop = asyncio.get_running_loop()
    _jobs[content_id] = {
        "id": content_id,
        "scenario": scenario,
        "input": input_text,
        "scenes": scenes,
        "seconds": seconds,
        "status": "queued",
        "step": 0,
        "log": [],
        "output": None,
        "job_dir": None,  # generator 가 stdout 으로 알려준다. 실패 정리 대상
        "store_name": None,  # 검색을 안 거친 경우 생성 로그에서 주워온다
        # 자동완성으로 고른 값. place URL 해석(_resolve_input_url)에만 쓴다 —
        # DB 저장은 create_task 가 이미 했으므로 여기서는 힌트일 뿐이다.
        "store_name_hint": store_name,
        "road_address_hint": road_address,
        "address_hint": address,
        "error": None,
        "timings": {},
    }
    with _lock:
        _queue.append(content_id)
    _pump()


def get_job(content_id: int) -> Optional[dict]:
    """인메모리 진행 정보. 로그 확인용이며 권위는 DB 다."""
    return _jobs.get(content_id)


def shutdown() -> None:
    """신규 큐잉을 막는다. lifespan shutdown 에서 dispose_engine 전에 호출."""
    global _shutting_down
    _shutting_down = True
    with _lock:
        dropped = len(_queue)
        _queue.clear()
    if dropped:
        logger.info(f"[job_manager] 종료 — 대기 중이던 {dropped}건은 다음 기동 스윕이 처리")


def _run_db(coro, timeout: Optional[int] = None):
    """워커 스레드에서 앱 루프에 DB 코루틴을 위임하고 완료까지 대기한다.

    asyncmy 커넥션 풀은 생성된 이벤트 루프에 바인딩되므로 스레드에서
    `asyncio.run` 을 쓰면 풀이 깨진다. 반드시 앱 루프에 위임해야 한다.
    """
    if _loop is None or _loop.is_closed():
        # 셧다운 중이면 루프가 코루틴을 실행하지 않아 무한정 매달린다.
        coro.close()
        raise RuntimeError("event loop unavailable (shutting down)")
    fut = asyncio.run_coroutine_threadsafe(coro, _loop)
    return fut.result(timeout=timeout or ssulbox_settings.SSULBOX_DB_DELEGATE_TIMEOUT)


async def _mark_running(content_id: int) -> None:
    async with BackgroundSessionLocal() as session:
        await task_service.mark_running(session, content_id)
        await session.commit()


async def _update_step(content_id: int, step: int) -> None:
    async with BackgroundSessionLocal() as session:
        await task_service.update_step(session, content_id, step)
        await session.commit()


async def _get_place_info(
    content_id: int,
) -> tuple[Optional[str], Optional[str], Optional[str]]:
    """현재 저장된 (store_name, region, detail_region_info)."""
    async with BackgroundSessionLocal() as session:
        return await task_service.get_place_info(session, content_id)


async def _save_place(
    content_id: int, store_name: str, region: str, detail: str
) -> None:
    async with BackgroundSessionLocal() as session:
        await task_service.set_place_info(
            session,
            content_id,
            store_name=store_name,
            region=region,
            detail_region_info=detail,
        )
        await session.commit()


async def _finalize(
    content_id: int, mp4: Path, store_name: Optional[str] = None
) -> None:
    cleanup: Optional[Path] = None
    async with BackgroundSessionLocal() as session:
        try:
            # store_name 은 생성 시점에 비어 있을 때만 반영된다(finalize_task 가 판단).
            _, cleanup = await task_service.finalize_task(
                session, content_id, mp4, store_name=store_name
            )
            await session.commit()
        except Exception:
            await session.rollback()
            raise
    # 되돌릴 수 없는 삭제는 커밋이 성공한 뒤에만
    task_service.cleanup_job_dir(cleanup)


async def _try_generate_sns_metadata(content_id: int) -> None:
    """제목/설명/해시태그 생성 실패가 완료 처리에 영향을 주지 않도록 격리합니다."""
    from app.social.services.seo_service import seo_service

    try:
        async with BackgroundSessionLocal() as session:
            await seo_service.generate_and_save_for_ssul(content_id, session)
            await session.commit()
    except Exception as e:
        logger.warning(
            f"[ssul {content_id}] SNS 메타데이터 생성 실패: {e}",
            exc_info=True,
        )


async def _fail(content_id: int, error: str) -> None:
    async with BackgroundSessionLocal() as session:
        try:
            await task_service.fail_task(session, content_id, error)
            await session.commit()
        except Exception:
            await session.rollback()
            raise


def _pump() -> None:
    """동시 실행 한도 안에서 대기 중인 잡을 시작한다."""
    global _running
    if _shutting_down:
        return
    with _lock:
        while _queue and _running < ssulbox_settings.SSULBOX_MAX_CONCURRENT_JOBS:
            content_id = _queue.popleft()
            _running += 1
            threading.Thread(target=_run, args=(content_id,), daemon=True).start()


def _build_command(job: dict) -> tuple[list[str], Path]:
    """generator 실행 커맨드와 작업 디렉터리."""
    engine = SCENARIO_ENGINE[job["scenario"]]
    engine_dir = ssulbox_settings.generator_path / engine
    main_py = engine_dir / "main.py"
    if not main_py.exists():
        raise FileNotFoundError(f"generator not found: {main_py}")

    cmd = [
        sys.executable,
        "-u",
        str(main_py),
        job["input"],
        "--scenes",
        str(job["scenes"]),
        "--seconds",
        str(job["seconds"]),
    ]
    return cmd, engine_dir


def _is_place_url(text: str) -> bool:
    """네이버 지도 링크로 보이는가 — 프론트 `isNaverUrl` 과 같은 판정."""
    t = (text or "").strip().lower()
    return t.startswith("http") and (
        "naver.me" in t or "map.naver" in t or "place.naver" in t
    )


def _collect_place_info(content_id: int, job: dict) -> None:
    """비어 있는 업장명·지역을 크롤링으로 채운다(있는 값은 유지).

    **빠진 값이 있을 때만 크롤링한다.** 검색으로 고른 경우 create 시점에 둘 다
    채워져 있으므로 브라우저를 띄우지 않는다. 다만 검색 경로에서도 지역만 비는
    경우(주소에서 시/군을 못 뽑는 등)가 있어, 존재 여부를 실제로 확인하고 판단한다.

    수집 실패는 삼킨다. 이 정보가 없어도 생성은 place_url 만으로 진행된다.
    """
    if not _is_place_url(job.get("input", "")):
        return
    try:
        store_name, region, detail = _run_db(_get_place_info(content_id))
        if store_name and region and detail:
            return  # 채울 것이 없다

        detail = _run_db(
            place_service.fetch_place_detail(job["input"]),
            # Playwright 기동 + 상세 파싱까지 DB 위임 기본 타임아웃(60s)보다 길 수 있다
            timeout=120,
        )
        if not detail:
            return

        title = detail.get("title") or ""
        road = detail.get("roadAddress") or ""
        jibun = detail.get("address") or ""
        # castad `/home/crawl` 과 동일하게 **도로명·지번을 모두** 넘긴다.
        # 도로명에서 시/군 추출이 실패하면 지번으로 재시도한다(한쪽만 넘기면 놓친다).
        new_region = extract_region_from_address(road or None, jibun or None)
        new_detail = road or jibun  # 도로명 우선, 없으면 지번
        if title:
            job["store_name"] = title
        if title or new_region or new_detail:
            _run_db(_save_place(content_id, title, new_region, new_detail))
    except Exception as e:
        logger.warning(
            f"[ssul {content_id}] 업장 정보 수집 실패(생성은 계속): "
            f"{type(e).__name__}: {e}"
        )


def _resolve_input_url(job: dict) -> None:
    """업장명 입력을 네이버 지도 place URL 로 바꾼다(자동완성 선택 경로).

    generator 는 place 페이지를 크롤링하므로 URL 이 있어야 정확한 가게를 잡는다.
    해석은 castad `NvMapPwScraper` 경로를 재사용한다(ADO2 자동완성과 동일).

    실패하면 원래 입력(업장명)을 그대로 둔다 — generator 가 업장명만으로도
    네이버 브리핑을 시도한다. 정확도는 떨어지지만 생성을 막지는 않는다.
    """
    raw = (job.get("input") or "").strip()
    store = (job.get("store_name_hint") or "").strip()
    if not store or _is_place_url(raw):
        return
    try:
        url = _run_db(
            place_service.resolve_place_url(
                store,
                address=job.get("address_hint") or "",
                road_address=job.get("road_address_hint") or "",
            ),
            # Playwright 기동 + 최대 3단 재시도라 DB 위임 기본 타임아웃보다 길다
            timeout=120,
        )
        if url:
            job["input"] = url
            logger.info(f"[_resolve_input_url] {store!r} → {url}")
        else:
            logger.warning(f"[_resolve_input_url] 해석 실패, 업장명으로 진행 - {store!r}")
    except Exception as e:
        logger.warning(
            f"[_resolve_input_url] 실패(업장명으로 진행): {type(e).__name__}: {e}"
        )


def _run(content_id: int) -> None:
    """워커 스레드 본체. subprocess 를 감독하며 stdout 마커로 진행을 파싱한다."""
    global _running
    job = _jobs[content_id]
    proc: Optional[subprocess.Popen] = None

    try:
        _run_db(_mark_running(content_id))
        job["status"] = "running"

        # 업장명만 들어온 경우(자동완성 선택) place URL 로 바꾼다.
        # **요청 핸들러가 아니라 여기서 하는 이유**: 해석에 10초 이상 걸려
        # `/ssul/create` 응답이 그만큼 막히고 사용자가 폼에 묶인다.
        # 워커로 옮기면 요청은 즉시 끝나고 진행 화면이 바로 뜬다.
        _resolve_input_url(job)

        # 업장명·주소를 **항상** 확보한다. 검색으로 고르지 않고 place URL 을
        # 붙여넣은 경우 create 시점에 아무것도 없으므로 여기서 크롤링한다.
        # 생성 **전에** 하는 이유: 목록에 이름 없는 카드가 뜨는 구간을 없앤다.
        # 실패해도 생성은 계속한다(목록 표시·필터용 부가 정보다).
        _collect_place_info(content_id, job)

        # ⚠️ 커맨드는 위 두 단계가 끝난 **뒤에** 만든다 — job["input"] 이 갱신되므로
        # 먼저 만들면 해석 전 값(업장명)이 그대로 엔진에 넘어간다.
        cmd, engine_dir = _build_command(job)

        env = dict(os.environ)
        env["PYTHONIOENCODING"] = "utf-8"
        key = gemini_key()
        if key:
            env["GEMINI_API_KEY"] = key

        proc = subprocess.Popen(
            cmd,
            cwd=str(engine_dir),
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            encoding="utf-8",
            errors="replace",
            bufsize=1,
        )

        started = step_started = time.monotonic()
        timeout_s = ssulbox_settings.SSULBOX_JOB_TIMEOUT_SECONDS

        # ⚠️ 벽시계 감시견이 필요한 이유 — `proc.wait(timeout=)` 만으로는 못 막는다.
        # 아래 `for raw in proc.stdout` 은 EOF 까지 블록한다. 엔진이 출력 없이
        # 매달리면(응답 없는 API 호출 등) EOF 가 오지 않아 워커 스레드가 영구
        # 정지하고 `_running` 이 안 줄어 **큐 전체가 멎는다**. wait() 의 타임아웃은
        # stdout 이 닫힌 뒤에야 평가되므로 정작 그 상황에는 도달하지 못한다.
        # 데드라인에 프로세스를 죽여야 stdout 이 닫히고 루프가 풀린다.
        timed_out = threading.Event()

        def _on_deadline() -> None:
            timed_out.set()
            logger.error(f"[ssul {content_id}] TIMEOUT {timeout_s}s — 프로세스 종료")
            _kill(proc)

        watchdog = threading.Timer(timeout_s, _on_deadline)
        watchdog.daemon = True
        watchdog.start()

        try:
            for raw in proc.stdout:  # type: ignore[union-attr]
                line = raw.rstrip()
                if not line:
                    continue
                logger.info(f"[ssul {content_id}] {line}")
                job["log"].append(line)
                if len(job["log"]) > 300:
                    del job["log"][:-300]

                m = STEP_RE.search(line)
                if m:
                    n = int(m.group(1))
                    if n != job["step"]:
                        now = time.monotonic()
                        prev = job["step"]
                        label = "준비·크롤링" if prev == 0 else STEP_NAMES[prev - 1]
                        job["timings"][label] = round(now - step_started, 1)
                        step_started = now
                        job["step"] = n
                        # 폴링이 읽을 수 있도록 DB 에 반영한다
                        try:
                            _run_db(_update_step(content_id, n))
                        except Exception as e:
                            logger.warning(f"[ssul {content_id}] step 저장 실패: {e}")

                d = DONE_RE.search(line)
                if d:
                    job["output"] = d.group(1).strip()

                # 실패 시 지울 대상. 완료 마커가 없어도 알 수 있는 유일한 경로다.
                jd = JOB_DIR_RE.search(line)
                if jd:
                    job["job_dir"] = jd.group(1).strip()

                # place URL 을 붙여넣어 검색을 안 거친 경우의 업장명 확보 경로.
                # '?' 는 엔진이 이름을 못 얻었을 때 찍는 placeholder 라 버린다.
                sm = STORE_RE.search(line)
                if sm:
                    name = sm.group(1).strip()
                    if name and name != "?":
                        job["store_name"] = name

            # stdout 이 닫혔으니 종료는 임박했다. 짧은 여유만 준다.
            code = proc.wait(timeout=30)
        finally:
            watchdog.cancel()

        if timed_out.is_set():
            # 감시견이 죽인 것이지 정상 종료가 아니다. 아래 except 로 넘긴다.
            raise subprocess.TimeoutExpired(cmd, timeout_s)

        summary = " · ".join(f"{k} {v}s" for k, v in job["timings"].items()) or "(마커 없음)"
        logger.info(
            f"[ssul {content_id}] 단계별 {summary} · 총 {time.monotonic() - started:.1f}s"
        )

        if code != 0:
            raise RuntimeError(f"생성 프로세스 종료코드 {code}")
        if not job["output"]:
            raise RuntimeError("완성 마커를 찾지 못했습니다")

        mp4 = Path(job["output"])
        if not mp4.is_absolute():
            mp4 = (engine_dir / mp4).resolve()

        # finalize 는 Blob 업로드·포스터를 포함해 오래 걸리므로 위임 타임아웃을 넉넉히
        _run_db(
            _finalize(content_id, mp4, job.get("store_name")), timeout=600
        )
        # ADO2 video_task 와 같이 완료 커밋 뒤에 SEO를 돌린다. 실패해도 영상은 유지.
        try:
            _run_db(_try_generate_sns_metadata(content_id), timeout=240)
        except Exception as e:
            logger.warning(
                f"[ssul {content_id}] SNS 메타데이터 위임 실패: {type(e).__name__}: {e}"
            )
        job["status"] = "done"
        job["step"] = 4

    except subprocess.TimeoutExpired as e:
        # 원본은 타임아웃이 없어 엔진이 걸리면 큐가 영구 정지했다.
        # 두 경로로 들어온다: 감시견 데드라인, 또는 stdout 이 닫혔는데도
        # 프로세스가 안 끝나는 경우(위 `wait(timeout=30)`).
        job["status"] = "error"
        job["error"] = f"생성 시간 초과 ({e.timeout}s)"
        logger.error(f"[ssul {content_id}] TIMEOUT — 프로세스 종료")
        _kill(proc)
        _safe_fail(content_id, job["error"])

    except Exception as e:
        job["status"] = "error"
        job["error"] = f"{type(e).__name__}: {e}"
        logger.error(f"[ssul {content_id}] FAILED - {job['error']}", exc_info=True)
        _kill(proc)
        _safe_fail(content_id, job["error"])

    finally:
        with _lock:
            _running -= 1
        _pump()


def _kill(proc: Optional[subprocess.Popen]) -> None:
    if proc is None or proc.poll() is not None:
        return
    try:
        proc.kill()
    except Exception as e:
        logger.warning(f"[job_manager] 프로세스 종료 실패: {e}")


def _safe_fail(content_id: int, error: str) -> None:
    """환불·상태 갱신. 이것마저 실패하면 다음 기동의 고아 스윕이 재처리한다."""
    try:
        _run_db(_fail(content_id, error))
    except Exception as e:
        logger.error(
            f"[ssul {content_id}] 실패 처리 실패(스윕이 재처리): {type(e).__name__}: {e}"
        )
    finally:
        # DB 처리 성패와 무관하게 디스크는 정리한다. 남겨봐야 쓸 곳이 없다.
        _cleanup_failed_dir(content_id)


def _cleanup_failed_dir(content_id: int) -> None:
    """실패한 잡의 중간 산출물 삭제.

    castad `video_task.py` 는 `finally` 로 성공·실패 양쪽을 정리한다. 썰박스도
    맞춘다. 실패 시엔 generator 자신의 자산 정리(`main.py` 의 `out_mp4.exists()`
    조건)마저 건너뛰므로 **오히려 실패가 가장 많이 남긴다** — 이미지·음성 전체.
    """
    raw = (_jobs.get(content_id) or {}).get("job_dir")
    if not raw:
        # 폴더 마커 전에 죽었다면 만들어진 것도 없다
        return
    try:
        job_dir = Path(raw).resolve()
        root = ssulbox_settings.output_path.resolve()
        # stdout 에서 읽어온 경로다. output 밖은 무슨 일이 있어도 지우지 않는다.
        # `job_dir.parent != root` 는 `output/<엔진>` 통째 삭제를 막는다
        # (정상 경로는 항상 `output/<엔진>/<작업>` 이라 2단계 아래다).
        if root not in job_dir.parents or job_dir.parent == root:
            logger.warning(f"[ssul {content_id}] output 밖 경로라 정리 생략: {job_dir}")
            return
        task_service.cleanup_job_dir(job_dir)
    except Exception as e:
        logger.warning(f"[ssul {content_id}] 실패 정리 실패: {type(e).__name__}: {e}")