"""C-3 공학 정적 코퍼스 1회 일괄 ingest (plan crawl-24x7-1).

National Board 기술 아티클(~86, ASP.NET 구식 — 기사 앵커가 싱글쿼트 href) +
TWI Job Knowledge(~153, sitemap 기반). 지속 크롤링이 아니라 아카이브 일괄 +
저빈도 증분 유형 — 스케줄러 미등록, 수동 CLI:

  docker exec hyungi_document_server-fastapi-1 \
      python -m workers.static_corpus_ingest --corpus all --limit 3   # 검증용
  docker exec -d hyungi_document_server-fastapi-1 \
      python -m workers.static_corpus_ingest --corpus all             # 전체 (~45분)

  ※ -d 백그라운드 실행 시 중단은 host pkill 이 아니라 컨테이너 내부 PID kill
    ([[feedback_docker_exec_orphan_kill]]).

멱등: edit_url(정규화)+file_hash dedup — 재실행 = 신규분만 (그대로 monthly 증분 절차).
politeness: fetch_page 재사용 (per-domain 1 + 5~15s jitter + robots).
원본 보존·승격 필드: fulltext_worker 와 동일 규약 (재추출 가능 상태 유지).
실패는 degrade 없이 skip + 말미 목록 출력 (정적 코퍼스 — RSS 요약 같은 격하 대상 부재).
"""

import argparse
import asyncio
import hashlib
import re
from datetime import datetime, timezone
from html import unescape

from sqlalchemy import select

from core.crawl_politeness import CrawlBlocked, CrawlFetchError, CrawlSkip, fetch_page
from core.database import async_session
from core.utils import setup_logger
from models.document import Document
from models.news_source import NewsSource
from models.queue import enqueue_stage
from workers.fulltext_worker import (
    _WEB_MIN_BODY_LEN,
    _extract_body,
    _raw_html_path,
    _save_raw_html,
    _strip_article_footer,
)
from workers.news_collector import _article_hash, _normalize_url

logger = setup_logger("static_corpus")

_NB_LISTING = "https://www.nationalboard.org/Index.aspx?pageID=164"
_TWI_SITEMAP = "https://www.twi-global.com/sitemap.xml"


async def _discover_national_board() -> list[str]:
    """목록 페이지의 기사 앵커 — 싱글쿼트 href 가 기본형이라 양쪽 인용부호 매칭."""
    html_text, _ = await fetch_page(_NB_LISTING)
    ids = sorted(
        {int(i) for i in re.findall(
            r"href=['\"]/?Index\.aspx\?pageID=164&(?:amp;)?ID=(\d+)['\"]", html_text)}
    )
    return [f"https://www.nationalboard.org/Index.aspx?pageID=164&ID={i}" for i in ids]


async def _discover_twi() -> list[str]:
    """sitemap 에서 job-knowledge 시리즈만 (faqs/published-papers 는 향후 증분 후보)."""
    xml_text, _ = await fetch_page(
        _TWI_SITEMAP,
        content_types=("text/xml", "application/xml", "text/html"),
    )
    urls = re.findall(
        r"<loc>(https://www\.twi-global\.com/technical-knowledge/job-knowledge/[^<]+)</loc>",
        xml_text,
    )
    return sorted({u for u in urls if not u.rstrip("/").endswith("job-knowledge")})


CORPORA = {
    "national-board": {
        "source_name": "National Board 기술 아티클",
        "listing_url": _NB_LISTING,
        "discover": _discover_national_board,
        "fetch_method": "page",
    },
    "twi": {
        "source_name": "TWI Job Knowledge",
        "listing_url": _TWI_SITEMAP,
        "discover": _discover_twi,
        "fetch_method": "sitemap+page",
    },
}


async def _get_or_create_source(session, spec: dict) -> NewsSource:
    """레지스트리 행 — 출처 추적 + crawl_raw src_{id} 경로 + A-8 패널 가시성.

    enabled=False: 6h 뉴스 사이클 비대상 (피드가 없는 정적 코퍼스 — 증분은 본 CLI 재실행).
    """
    result = await session.execute(
        select(NewsSource).where(NewsSource.name == spec["source_name"])
    )
    source = result.scalars().first()
    if source is None:
        source = NewsSource(
            name=spec["source_name"],
            feed_url=spec["listing_url"],
            feed_type="rss",
            fetch_method=spec["fetch_method"],
            fulltext_policy="none",
            source_channel="crawl",
            category="Engineering",
            language="en",
            country="US" if "national" in spec["source_name"].lower() else "GB",
            enabled=False,
        )
        session.add(source)
        await session.flush()
    return source


def _page_title(html_text: str, fallback: str) -> str:
    m = re.search(r'<meta\s+property="og:title"\s+content="([^"]+)"', html_text)
    if not m:
        m = re.search(r"<title[^>]*>([^<]+)</title>", html_text, re.I)
    title = unescape(m.group(1)).strip() if m else ""
    # 사이트 접미 잡음 제거 (TWI 는 ' - TWI', NB 는 'National Board ...' 꼬리표)
    title = re.sub(r"\s*[-|·]\s*(TWI|National Board[^-|]*)\s*$", "", title).strip()
    return title or fallback


async def _ingest_one(session, source: NewsSource, url: str) -> str:
    """기사 1건. 반환: 'ok' / 'dup' / 'skip'(추출부족·차단)."""
    normalized_url = _normalize_url(url)
    existing = await session.execute(
        select(Document).where(Document.edit_url.in_([normalized_url, url])).limit(1)
    )
    if existing.scalars().first():
        return "dup"

    try:
        html_text, final_url = await fetch_page(url)
    except (CrawlBlocked, CrawlSkip, CrawlFetchError) as e:
        logger.warning(f"[{source.name}] fetch 실패 skip: {url} — {type(e).__name__}: {e}")
        return "skip"

    body, engine, engine_ver = _extract_body(html_text)
    if not engine:
        logger.warning(f"[{source.name}] 추출 실패 skip (< {_WEB_MIN_BODY_LEN}자): {url}")
        return "skip"
    clean_body = _strip_article_footer(body.replace("\x00", ""))
    if len(clean_body) < _WEB_MIN_BODY_LEN:
        logger.warning(f"[{source.name}] 푸터 제거 후 본문 부족 skip: {url}")
        return "skip"

    title = _page_title(html_text, fallback=url.rsplit("/", 1)[-1][:90])
    article_id = _article_hash(title, "static", source.name)
    dup2 = await session.execute(
        select(Document).where(Document.file_hash == article_id).limit(1)
    )
    if dup2.scalars().first():
        return "dup"

    now = datetime.now(timezone.utc)
    raw_path = _raw_html_path(source.id, article_id, now)
    raw_saved = True
    try:
        _save_raw_html(raw_path, html_text)
    except OSError as e:
        raw_saved = False
        logger.error(f"[{source.name}] 원본 보존 실패 (ingest 는 진행): {e}")

    doc = Document(
        file_path=f"crawl/{source.name}/{article_id}",
        file_hash=article_id,
        file_format="article",
        file_size=0,  # 아래 extracted_text 확정 후 재계산
        file_type="note",
        title=title,
        extracted_text=f"{title}\n\n{clean_body}",
        extracted_at=now,
        extractor_version=f"static+page@{engine}",
        md_content=clean_body,
        md_status="success",
        md_extraction_engine=engine,
        md_extraction_engine_version=engine_ver,
        md_format_version="1.0",
        md_generated_at=now,
        md_source_hash=hashlib.sha256(html_text.encode("utf-8", errors="replace")).hexdigest(),
        md_content_hash=hashlib.sha256(clean_body.encode("utf-8")).hexdigest(),
        content_origin="extracted",
        source_channel="crawl",
        data_origin="external",
        edit_url=normalized_url,
        review_status="approved",
        ai_domain="Engineering",
        ai_sub_group=source.name,
        ai_tags=[f"Engineering/{source.name}"],
        extract_meta={
            "source_id": source.id,
            "source_name": source.name,
            "published_at": None,  # 정적 코퍼스 — 페이지 발행일 비신뢰, 색인은 채널 게이트로 무조건
            "fulltext": {
                "status": "static_corpus",
                "engine": engine,
                "final_url": final_url,
                "raw_html_path": str(raw_path) if raw_saved else None,
                "body_chars": len(clean_body),
                "resolved_at": now.isoformat(),
            },
        },
    )
    doc.file_size = len(doc.extracted_text.encode())
    session.add(doc)
    await session.flush()

    # crawl 채널 = 발행일 무관 전량 색인 (summarize 는 맥미니 큐 — D-4 lag 관찰 대상)
    await enqueue_stage(session, doc.id, "summarize")
    await enqueue_stage(session, doc.id, "embed")
    await enqueue_stage(session, doc.id, "chunk")
    logger.info(f"[{source.name}] ingest {len(clean_body)}자 ({engine}): {title[:60]}")
    return "ok"


async def run(corpus: str = "all", limit: int = 0) -> None:
    targets = list(CORPORA) if corpus == "all" else [corpus]
    for key in targets:
        spec = CORPORA[key]
        async with async_session() as session:
            source = await _get_or_create_source(session, spec)
            await session.commit()
            source_id = source.id

        try:
            urls = await spec["discover"]()
        except (CrawlBlocked, CrawlSkip, CrawlFetchError) as e:
            logger.error(f"[{spec['source_name']}] 목록 수집 실패 — corpus 건너뜀: {e}")
            continue
        if limit:
            urls = urls[:limit]
        logger.info(f"[{spec['source_name']}] 대상 {len(urls)}건 (limit={limit or '없음'})")

        counts = {"ok": 0, "dup": 0, "skip": 0}
        failed: list[str] = []
        for i, url in enumerate(urls, 1):
            # 커밋 10건 단위 — 장시간 배치 중단 시 진행분 보존
            async with async_session() as session:
                src = await session.get(NewsSource, source_id)
                status = await _ingest_one(session, src, url)
                await session.commit()
            counts[status] += 1
            if status == "skip":
                failed.append(url)
            if i % 10 == 0:
                logger.info(f"[{spec['source_name']}] 진행 {i}/{len(urls)} {counts}")

        logger.info(f"[{spec['source_name']}] 완료: {counts}")
        if failed:
            logger.warning(
                f"[{spec['source_name']}] skip {len(failed)}건 — 재시도는 CLI 재실행(멱등):\n  "
                + "\n  ".join(failed)
            )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="C-3 정적 코퍼스 일괄 ingest")
    parser.add_argument("--corpus", choices=[*CORPORA, "all"], default="all")
    parser.add_argument("--limit", type=int, default=0, help="corpus 당 상한 (0=전체)")
    args = parser.parse_args()
    asyncio.run(run(args.corpus, args.limit))
