# -*- coding: utf-8 -*-
"""MaemulHOME v1.0 shorts script worker (2026-08-26 r2).

This worker is intentionally limited to the first server-side stage:

    source_ready -> analysis_ready -> script_ready

It does not render or upload videos.  A MySQL advisory lock keeps script
generation sequential even if two scheduler processes start at the same time.
Existing v0.1.9.1/server publishing code is not imported or modified.
"""

from __future__ import annotations

import argparse
import html
import json
import logging
import os
import re
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple


FILE_PATH = Path(__file__).resolve()
for candidate in (FILE_PATH.parent, FILE_PATH.parent.parent):
    value = str(candidate)
    if value not in sys.path:
        sys.path.insert(0, value)

try:
    from db import get_conn as _get_conn  # type: ignore
except ImportError:  # pragma: no cover - checked on the real server
    _get_conn = None


PROJECT_TABLE = "blog_v1_shorts_projects"
ADVISORY_LOCK = "maemulhome_v1_shorts_script_worker"
DEFAULT_API_URL = os.getenv(
    "V1_SHORTS_SCRIPT_API_URL", "http://127.0.0.1:9000/v1/generate-script"
)
DEFAULT_API_KEY = os.getenv(
    "V1_SHORTS_API_KEY", "honghee-shorts-secret-2026"
)
DEFAULT_TIMEOUT = int(os.getenv("V1_SHORTS_SCRIPT_TIMEOUT", "240"))
STALE_MINUTES = int(os.getenv("V1_SHORTS_STALE_MINUTES", "30"))
MAX_SOURCE_CHARS = int(os.getenv("V1_SHORTS_MAX_SOURCE_CHARS", "14000"))
MAX_IMAGES = int(os.getenv("V1_SHORTS_MAX_IMAGES", "30"))

LOGGER = logging.getLogger("v1_shorts_worker")
WORKER_BUILD = "20260826-r2-v1-table"


def open_connection() -> Any:
    if _get_conn is None:
        raise RuntimeError(
            "db.py를 찾을 수 없습니다. 이 파일을 D:/honghee/blog_api/jobs/ 또는 "
            "D:/honghee/blog_api/에 두고 실행해 주세요."
        )
    return _get_conn()


def json_text(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=str)


def clean_text(value: Any, limit: int = MAX_SOURCE_CHARS) -> str:
    text = str(value or "")
    text = re.sub(r"<script\b[^>]*>.*?</script>", " ", text, flags=re.I | re.S)
    text = re.sub(r"<style\b[^>]*>.*?</style>", " ", text, flags=re.I | re.S)
    text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
    text = re.sub(r"</(?:p|div|li|tr|h[1-6])>", "\n", text, flags=re.I)
    text = re.sub(r"<[^>]+>", " ", text)
    text = html.unescape(text)
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    text = re.sub(r"[ \t\f\v]+", " ", text)
    text = re.sub(r" *\n *", "\n", text)
    text = re.sub(r"\n{3,}", "\n\n", text).strip()
    return text[:limit]


def unique_strings(values: Iterable[Any], limit: int = MAX_IMAGES) -> List[str]:
    result: List[str] = []
    seen = set()
    for value in values:
        item = html.unescape(str(value or "")).strip()
        if not item or item in seen:
            continue
        if not re.match(r"^https?://", item, flags=re.I):
            continue
        seen.add(item)
        result.append(item)
        if len(result) >= limit:
            break
    return result


def extract_image_urls(*html_values: Any) -> List[str]:
    candidates: List[str] = []
    for value in html_values:
        text = str(value or "")
        candidates.extend(
            re.findall(
                r"<img\b[^>]*\bsrc\s*=\s*[\"']([^\"']+)[\"']",
                text,
                flags=re.I,
            )
        )
    return unique_strings(candidates)


def table_exists(conn: Any, table_name: str) -> bool:
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT COUNT(*) AS cnt
              FROM information_schema.TABLES
             WHERE TABLE_SCHEMA = DATABASE()
               AND TABLE_NAME = %s
            """,
            (table_name,),
        )
        row = cur.fetchone() or {}
    return int(row.get("cnt") or 0) > 0


def acquire_lock(conn: Any) -> bool:
    with conn.cursor() as cur:
        cur.execute("SELECT GET_LOCK(%s, 0) AS acquired", (ADVISORY_LOCK,))
        row = cur.fetchone() or {}
    return int(row.get("acquired") or 0) == 1


def release_lock(conn: Any) -> None:
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT RELEASE_LOCK(%s)", (ADVISORY_LOCK,))
    except Exception:
        LOGGER.exception("advisory lock release failed")


def claim_project(conn: Any, project_id: int = 0) -> Optional[Dict[str, Any]]:
    """Claim one project while holding the process-wide advisory lock."""
    conn.begin()
    try:
        where = """
            (
                (p.workflow_status = 'source_ready')
                OR
                (p.workflow_status = 'analysis_ready'
                 AND p.updated_at < DATE_SUB(NOW(), INTERVAL %s MINUTE))
            )
        """
        params: List[Any] = [STALE_MINUTES]
        if project_id > 0:
            where += " AND p.id = %s"
            params.append(project_id)

        with conn.cursor() as cur:
            cur.execute(
                f"""
                SELECT p.*
                  FROM {PROJECT_TABLE} p
                 WHERE {where}
                 ORDER BY p.created_at ASC, p.id ASC
                 LIMIT 1
                 FOR UPDATE
                """,
                tuple(params),
            )
            project = cur.fetchone()
            if not project:
                conn.rollback()
                return None

            cur.execute(
                f"""
                UPDATE {PROJECT_TABLE}
                   SET workflow_status = 'analysis_ready',
                       project_status = 'processing',
                       script_status = 'processing',
                       error_message = NULL,
                       updated_at = NOW()
                 WHERE id = %s
                """,
                (project["id"],),
            )
        conn.commit()
        project["workflow_status"] = "analysis_ready"
        return project
    except Exception:
        conn.rollback()
        raise


def fetch_one_optional(
    conn: Any, table_name: str, query: str, params: Sequence[Any]
) -> Dict[str, Any]:
    if not table_exists(conn, table_name):
        return {}
    with conn.cursor() as cur:
        cur.execute(query, tuple(params))
        return cur.fetchone() or {}


def fetch_source_bundle(conn: Any, project: Dict[str, Any]) -> Dict[str, Any]:
    project_id = int(project.get("id") or 0)
    article_no = str(project.get("article_no") or "").strip()
    realtor_id = int(project.get("realtor_id") or 0)

    draft = fetch_one_optional(
        conn,
        "blog_article_drafts",
        """
        SELECT id, realtor_id, article_no, draft_title, draft_html,
               clipboard_html, plain_text, ai_summary,
               header_image_url, representative_image_url, blog_url,
               blog_id, blog_post_no, updated_at
          FROM blog_article_drafts
         WHERE article_no = %s
         ORDER BY id DESC
         LIMIT 1
        """,
        (article_no,),
    )

    publish_queue = fetch_one_optional(
        conn,
        "blog_publish_queue",
        """
        SELECT id, realtor_id, draft_id, article_no, publish_title,
               publish_result_url, blog_id, blog_post_no,
               queue_status, finished_at
          FROM blog_publish_queue
         WHERE article_no = %s
           AND realtor_id = %s
         ORDER BY id DESC
         LIMIT 1
        """,
        (article_no, realtor_id),
    )

    realtor = fetch_one_optional(
        conn,
        "blog_realtors",
        """
        SELECT id, office_name, representative_name, office_phone,
               mobile_phone, region, office_address, office_detail_address
          FROM blog_realtors
         WHERE id = %s
         LIMIT 1
        """,
        (realtor_id,),
    )

    title = str(
        project.get("source_post_title")
        or project.get("post_title")
        or publish_queue.get("publish_title")
        or draft.get("draft_title")
        or project.get("shorts_title")
        or f"네이버 부동산 매물 {article_no}"
    ).strip()

    blog_url = str(
        project.get("blog_post_url")
        or publish_queue.get("publish_result_url")
        or draft.get("blog_url")
        or ""
    ).strip()

    source_text = clean_text(
        draft.get("plain_text")
        or draft.get("draft_html")
        or draft.get("clipboard_html")
        or project.get("shorts_script")
        or ""
    )

    images = extract_image_urls(draft.get("draft_html"), draft.get("clipboard_html"))
    images = unique_strings(
        [
            draft.get("representative_image_url"),
            draft.get("header_image_url"),
            *images,
        ]
    )

    if not source_text:
        raise RuntimeError(
            f"숏츠 원본 본문을 찾지 못했습니다(project_id={project_id}, article_no={article_no})."
        )
    if not blog_url:
        raise RuntimeError(
            f"발행된 블로그 URL이 없습니다(project_id={project_id}, article_no={article_no})."
        )

    return {
        "project_id": project_id,
        "article_no": article_no,
        "realtor_id": realtor_id,
        "title": title,
        "blog_url": blog_url,
        "content": source_text,
        "image_urls": images,
        "image_count": len(images),
        "draft_id": draft.get("id"),
        "publish_queue_id": publish_queue.get("id"),
        "office_name": realtor.get("office_name") or "",
        "representative_name": realtor.get("representative_name") or "",
        "office_phone": realtor.get("office_phone") or "",
        "mobile_phone": realtor.get("mobile_phone") or "",
        "region": realtor.get("region") or "",
    }


def build_payload(project: Dict[str, Any], source: Dict[str, Any]) -> Dict[str, Any]:
    platforms = [
        value.strip()
        for value in str(project.get("publish_platforms") or "youtube").split(",")
        if value.strip()
    ]
    if not platforms:
        platforms = ["youtube"]

    content_lines = [
        "부동산 숏츠 제작용 원본 정보",
        f"매물번호: {source['article_no']}",
        f"원본 제목: {source['title']}",
        f"블로그 URL: {source['blog_url']}",
        f"중개사무소: {source['office_name']}",
        f"대표자: {source['representative_name']}",
        f"지역: {source['region']}",
        "",
        "원본 본문:",
        source["content"],
        "",
        "작성 조건:",
        "- 45초 세로형 부동산 숏츠",
        "- 첫 3초에 핵심 매력과 가격을 자연스럽게 제시",
        "- 허위·과장·확정 수익 표현 금지",
        "- 짧고 읽기 쉬운 한국어 자막 문장",
        "- 마지막은 중개사무소 상담 문의 안내",
        "- JSON 또는 API가 요구하는 정형 응답 사용",
    ]

    return {
        "project_id": int(project.get("id") or 0),
        "realtor_id": int(project.get("realtor_id") or 0),
        "membership_id": int(project.get("membership_id") or 0),
        "source_type": project.get("source_type") or "blog_auto",
        "title": source["title"],
        "source_url": source["blog_url"],
        "article_no": source["article_no"],
        "content": "\n".join(content_lines),
        "script_type": project.get("script_type") or "basic",
        "generation_mode": project.get("generation_mode") or "render_video",
        "platforms": platforms,
        "duration_seconds": 45,
        "video_width": 1080,
        "video_height": 1920,
        "image_urls": source["image_urls"],
    }


def call_script_api(payload: Dict[str, Any]) -> Tuple[Dict[str, Any], str]:
    body = json_text(payload).encode("utf-8")
    request = urllib.request.Request(
        DEFAULT_API_URL,
        data=body,
        headers={
            "Content-Type": "application/json; charset=utf-8",
            "X-API-KEY": DEFAULT_API_KEY,
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=DEFAULT_TIMEOUT) as response:
            raw = response.read().decode("utf-8", errors="replace")
            status = int(getattr(response, "status", 200))
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"숏츠 대본 API HTTP {exc.code}: {raw[:1000]}") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(f"숏츠 대본 API 연결 실패: {exc}") from exc

    if status >= 400:
        raise RuntimeError(f"숏츠 대본 API HTTP {status}: {raw[:1000]}")
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"숏츠 대본 API 응답이 JSON이 아닙니다: {raw[:1000]}") from exc
    if not isinstance(data, dict):
        raise RuntimeError("숏츠 대본 API 응답 형식이 올바르지 않습니다.")
    if data.get("ok") is False:
        raise RuntimeError(str(data.get("error") or data.get("message") or "숏츠 대본 생성 실패"))
    return data, raw


def nested_value(data: Dict[str, Any], paths: Sequence[Sequence[str]]) -> Any:
    for path in paths:
        value: Any = data
        for key in path:
            if not isinstance(value, dict) or key not in value:
                value = None
                break
            value = value[key]
        if value not in (None, "", [], {}):
            return value
    return None


def parse_api_result(data: Dict[str, Any], fallback_title: str) -> Dict[str, Any]:
    script = nested_value(
        data,
        [
            ("shorts_script",),
            ("script",),
            ("text",),
            ("result", "shorts_script"),
            ("result", "script"),
            ("data", "shorts_script"),
            ("data", "script"),
        ],
    )
    if not isinstance(script, str) or not script.strip():
        raise RuntimeError("숏츠 대본 API 응답에 script가 없습니다.")

    title = nested_value(
        data,
        [
            ("shorts_title",),
            ("title",),
            ("result", "shorts_title"),
            ("result", "title"),
            ("data", "shorts_title"),
            ("data", "title"),
        ],
    )
    hashtags = nested_value(
        data,
        [
            ("hashtags",),
            ("result", "hashtags"),
            ("data", "hashtags"),
        ],
    )
    video_prompt = nested_value(
        data,
        [
            ("video_prompt",),
            ("result", "video_prompt"),
            ("data", "video_prompt"),
        ],
    )
    candidates = nested_value(
        data,
        [
            ("script_candidates",),
            ("result", "script_candidates"),
            ("data", "script_candidates"),
        ],
    )

    if isinstance(hashtags, list):
        hashtags = " ".join(str(item).strip() for item in hashtags if str(item).strip())

    return {
        "shorts_title": str(title or fallback_title).strip()[:255],
        "shorts_script": script.strip(),
        "hashtags": str(hashtags or "").strip(),
        "video_prompt": str(video_prompt or "").strip(),
        "script_candidates": json_text(candidates) if candidates not in (None, "") else None,
    }


def save_success(
    conn: Any,
    project_id: int,
    source: Dict[str, Any],
    payload: Dict[str, Any],
    raw_response: str,
    result: Dict[str, Any],
) -> None:
    with conn.cursor() as cur:
        cur.execute(
            f"""
            UPDATE {PROJECT_TABLE}
               SET source_post_title = %s,
                   shorts_title = %s,
                   shorts_script = %s,
                   video_prompt = %s,
                   hashtags = %s,
                   script_request_json = %s,
                   script_response_json = %s,
                   script_generated_at = NOW(),
                   image_count = %s,
                   project_status = 'processing',
                   script_status = 'done',
                   workflow_status = 'script_ready',
                   render_status = 'pending',
                   error_message = NULL,
                   updated_at = NOW()
             WHERE id = %s
            """,
            (
                source["title"],
                result["shorts_title"],
                result["shorts_script"],
                result["video_prompt"],
                result["hashtags"],
                json_text({"payload": payload, "source": source}),
                raw_response,
                int(source["image_count"]),
                project_id,
            ),
        )
    conn.commit()


def save_failure(conn: Any, project_id: int, error: Exception) -> None:
    message = str(error).strip()[:60000]
    try:
        with conn.cursor() as cur:
            cur.execute(
                f"""
                UPDATE {PROJECT_TABLE}
                   SET project_status = 'failed',
                       script_status = 'failed',
                       workflow_status = 'failed',
                       error_message = %s,
                       updated_at = NOW()
                 WHERE id = %s
                """,
                (message, project_id),
            )
        conn.commit()
    except Exception:
        conn.rollback()
        LOGGER.exception("failed to record project error: project_id=%s", project_id)


def process_once(project_id: int = 0, dry_run: bool = False) -> int:
    conn = open_connection()
    project: Optional[Dict[str, Any]] = None
    locked = False
    try:
        if not table_exists(conn, PROJECT_TABLE):
            raise RuntimeError(f"{PROJECT_TABLE} 테이블이 없습니다.")
        locked = acquire_lock(conn)
        if not locked:
            LOGGER.info("다른 v1 숏츠 Worker가 실행 중이므로 건너뜁니다.")
            return 0

        project = claim_project(conn, project_id=project_id)
        if not project:
            if project_id > 0:
                with conn.cursor() as cur:
                    cur.execute(
                        f"""
                        SELECT id, realtor_id, article_no, project_status,
                               workflow_status, script_status, render_status,
                               error_message, updated_at
                          FROM {PROJECT_TABLE}
                         WHERE id = %s
                         LIMIT 1
                        """,
                        (project_id,),
                    )
                    state = cur.fetchone()
                if state:
                    LOGGER.warning(
                        "project_id=%s 상태가 처리 조건과 다릅니다: %s",
                        project_id,
                        json_text(state),
                    )
                else:
                    LOGGER.warning(
                        "현재 연결된 DB의 %s 테이블에 project_id=%s가 없습니다.",
                        PROJECT_TABLE,
                        project_id,
                    )
            else:
                LOGGER.info("처리할 source_ready 숏츠 작업이 없습니다.")
            return 0

        pid = int(project["id"])
        LOGGER.info(
            "숏츠 대본 작업 시작: project_id=%s, realtor_id=%s, article_no=%s",
            pid,
            project.get("realtor_id"),
            project.get("article_no"),
        )

        source = fetch_source_bundle(conn, project)
        payload = build_payload(project, source)

        if dry_run:
            with conn.cursor() as cur:
                cur.execute(
                    f"""
                    UPDATE {PROJECT_TABLE}
                       SET workflow_status = 'source_ready',
                           project_status = 'queued',
                           script_status = 'pending',
                           updated_at = NOW()
                     WHERE id = %s
                    """,
                    (pid,),
                )
            conn.commit()
            print(json.dumps({"ok": True, "dry_run": True, "source": source}, ensure_ascii=False, indent=2, default=str))
            return 0

        response, raw = call_script_api(payload)
        parsed = parse_api_result(response, fallback_title=source["title"])
        save_success(conn, pid, source, payload, raw, parsed)
        LOGGER.info(
            "숏츠 대본 생성 완료: project_id=%s, images=%s, title=%s",
            pid,
            source["image_count"],
            parsed["shorts_title"],
        )
        return 1
    except Exception as exc:
        conn.rollback()
        if project:
            save_failure(conn, int(project["id"]), exc)
        LOGGER.exception("숏츠 대본 작업 실패: %s", exc)
        return 2
    finally:
        if locked:
            release_lock(conn)
        conn.close()


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="MaemulHOME v1.0 sequential shorts script worker"
    )
    parser.add_argument("--project-id", type=int, default=0, help="특정 프로젝트만 처리")
    parser.add_argument("--dry-run", action="store_true", help="원본 수집만 검증하고 상태 복원")
    parser.add_argument("--loop", action="store_true", help="계속 실행하며 대기 작업 확인")
    parser.add_argument("--poll-seconds", type=int, default=30, help="loop 확인 간격")
    parser.add_argument("--log-level", default="INFO", choices=("DEBUG", "INFO", "WARNING", "ERROR"))
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    logging.basicConfig(
        level=getattr(logging, args.log_level),
        format="%(asctime)s [%(levelname)s] %(message)s",
    )
    LOGGER.info("v1 숏츠 Worker 빌드: %s / table=%s", WORKER_BUILD, PROJECT_TABLE)

    if not args.loop:
        return process_once(project_id=args.project_id, dry_run=args.dry_run)

    LOGGER.info("v1 숏츠 순차 Worker 시작: poll=%s초", max(5, args.poll_seconds))
    while True:
        process_once(project_id=args.project_id, dry_run=args.dry_run)
        if args.project_id > 0:
            return 0
        time.sleep(max(5, args.poll_seconds))


if __name__ == "__main__":
    raise SystemExit(main())