"""Article search + delete."""
from __future__ import annotations

import logging
from datetime import date, datetime
from pathlib import Path, PurePosixPath
from typing import Any

from fastapi import BackgroundTasks, Body, Depends, HTTPException, Query
from fastapi.responses import FileResponse, Response
from sqlalchemy import desc, func, or_, select
from sqlalchemy.orm import Session

from govcrawler import openstd_download
from govcrawler.models import (
    Article,
    ArticleStandardMeta,
    Attachment,
    CrawlSite,
    CrawlTarget,
    LocalDepartment,
)
from govcrawler.settings import get_settings
from govcrawler.storage.paths import to_os_path
from govcrawler.timeutil import epoch_iso_cn, now_cn_naive

from ._common import iso_cn, _session, router

log = logging.getLogger(__name__)


def _standard_meta_payload(m: ArticleStandardMeta | None) -> dict[str, Any] | None:
    if m is None:
        return None
    return {
        "standard_no": m.standard_no,
        "chinese_title": m.chinese_title,
        "english_title": m.english_title,
        "standard_status": m.standard_status,
        "ccs_codes": m.ccs_codes or [],
        "ics_codes": m.ics_codes or [],
        "publish_date": m.publish_date.isoformat() if m.publish_date else None,
        "implementation_date": (
            m.implementation_date.isoformat() if m.implementation_date else None
        ),
        "competent_department": m.competent_department,
        "technical_committee": m.technical_committee,
        "issuing_body": m.issuing_body,
        "standard_type": m.standard_type,
        "raw_meta_json": m.raw_meta_json,
    }


def _articles_base_stmt():
    return (
        select(
            Article,
            CrawlSite.site_code.label("site_code"),
            CrawlTarget.target_code.label("target_code"),
            LocalDepartment.dept_id.label("dept_id"),
            LocalDepartment.dept_name.label("dept_name"),
            LocalDepartment.short_name.label("dept_short_name"),
        )
        .select_from(Article)
        .join(CrawlSite, CrawlSite.id == Article.site_id, isouter=True)
        .join(CrawlTarget, CrawlTarget.id == Article.target_id, isouter=True)
        .join(LocalDepartment, LocalDepartment.dept_id == Article.dept_id, isouter=True)
    )


def _article_payload(
    a: Article, row: Any | None = None, *, include_heavy: bool = False,
) -> dict[str, Any]:
    payload = {
        "id": a.id,
        "site_id": a.site_id,
        "site_code": getattr(row, "site_code", None),
        "target_id": a.target_id,
        "target_code": getattr(row, "target_code", None),
        "dept_id": getattr(row, "dept_id", a.dept_id),
        "dept_name": getattr(row, "dept_name", None),
        "dept_short_name": getattr(row, "dept_short_name", None),
        "native_post_id": a.native_post_id,
        "url": a.url,
        "url_hash": a.url_hash,
        "title": a.title,
        "publish_time": iso_cn(a.publish_time),
        "source_raw": a.source_raw,
        "channel_name": a.channel_name,
        "channel_path": a.channel_path,
        "content_category": a.content_category,
        "content_subcategory": a.content_subcategory,
        "index_no": a.index_no,
        "doc_no": a.doc_no,
        "publisher": a.publisher,
        "publish_date": a.publish_date.isoformat() if a.publish_date else None,
        "effective_date": a.effective_date.isoformat() if a.effective_date else None,
        "is_effective": a.is_effective,
        "expiry_date": a.expiry_date.isoformat() if a.expiry_date else None,
        "topic_words": a.topic_words,
        "open_category": a.open_category,
        "content_text_len": len(a.content_text or ""),
        "raw_html_path": a.raw_html_path,
        "text_path": a.text_path,
        "has_attachment": a.has_attachment,
        "status": a.status,
        "fetch_strategy": a.fetch_strategy,
        "fetched_at": iso_cn(a.fetched_at),
        "exported_to_rag_at": iso_cn(a.exported_to_rag_at),
        "rag_export_status": a.rag_export_status,
        "rag_export_started_at": iso_cn(a.rag_export_started_at),
        "rag_export_finished_at": iso_cn(a.rag_export_finished_at),
        "rag_export_error": a.rag_export_error,
        "rag_export_task_ids": a.rag_export_task_ids or [],
        "created_at": iso_cn(a.created_at),
        "updated_at": iso_cn(a.updated_at),
    }
    if include_heavy:
        payload["metadata_json"] = a.metadata_json
        payload["content_text"] = a.content_text
        payload["standard_meta"] = _standard_meta_payload(a.standard_meta)
    return payload


def _append_contains_filter(filters: list[Any], column: Any, value: str | None) -> None:
    if value:
        filters.append(column.ilike(f"%{value}%"))


def _article_filters(
    *,
    q: str | None = None,
    article_id: int | None = None,
    site: str | None = None,
    target: str | None = None,
    dept: int | None = None,
    status: str | None = None,
    has_attachment: bool | None = None,
    is_effective: bool | None = None,
    native_post_id: str | None = None,
    url: str | None = None,
    url_hash: str | None = None,
    title: str | None = None,
    source_raw: str | None = None,
    channel_name: str | None = None,
    channel_path: str | None = None,
    content_category: str | None = None,
    content_subcategory: str | None = None,
    index_no: str | None = None,
    doc_no: str | None = None,
    publisher: str | None = None,
    topic_words: str | None = None,
    open_category: str | None = None,
    fetch_strategy: str | None = None,
    rag_export_status: str | None = None,
    publish_date_from: date | None = None,
    publish_date_to: date | None = None,
    effective_date_from: date | None = None,
    effective_date_to: date | None = None,
    expiry_date_from: date | None = None,
    expiry_date_to: date | None = None,
    fetched_from: datetime | None = None,
    fetched_to: datetime | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    since: datetime | None = None,
) -> list[Any]:
    filters: list[Any] = []
    if q:
        like = f"%{q}%"
        filters.append(or_(Article.title.ilike(like), Article.content_text.ilike(like)))
    if article_id is not None:
        filters.append(Article.id == article_id)
    if site:
        filters.append(CrawlSite.site_code == site)
    if target:
        filters.append(CrawlTarget.target_code == target)
    if dept is not None:
        filters.append(Article.dept_id == dept)
    if status:
        filters.append(Article.status == status)
    if has_attachment is not None:
        filters.append(Article.has_attachment.is_(has_attachment))
    if is_effective is not None:
        filters.append(Article.is_effective.is_(is_effective))
    if url_hash:
        filters.append(Article.url_hash == url_hash)
    if fetch_strategy:
        filters.append(Article.fetch_strategy == fetch_strategy)
    if rag_export_status:
        filters.append(Article.rag_export_status == rag_export_status)

    _append_contains_filter(filters, Article.native_post_id, native_post_id)
    _append_contains_filter(filters, Article.url, url)
    _append_contains_filter(filters, Article.title, title)
    _append_contains_filter(filters, Article.source_raw, source_raw)
    _append_contains_filter(filters, Article.channel_name, channel_name)
    _append_contains_filter(filters, Article.channel_path, channel_path)
    _append_contains_filter(filters, Article.content_category, content_category)
    _append_contains_filter(filters, Article.content_subcategory, content_subcategory)
    _append_contains_filter(filters, Article.index_no, index_no)
    _append_contains_filter(filters, Article.doc_no, doc_no)
    _append_contains_filter(filters, Article.publisher, publisher)
    _append_contains_filter(filters, Article.topic_words, topic_words)
    _append_contains_filter(filters, Article.open_category, open_category)

    if publish_date_from:
        filters.append(Article.publish_date >= publish_date_from)
    if publish_date_to:
        filters.append(Article.publish_date <= publish_date_to)
    if effective_date_from:
        filters.append(Article.effective_date >= effective_date_from)
    if effective_date_to:
        filters.append(Article.effective_date <= effective_date_to)
    if expiry_date_from:
        filters.append(Article.expiry_date >= expiry_date_from)
    if expiry_date_to:
        filters.append(Article.expiry_date <= expiry_date_to)
    if since:
        filters.append(Article.fetched_at >= since)
    if fetched_from:
        filters.append(Article.fetched_at >= fetched_from)
    if fetched_to:
        filters.append(Article.fetched_at <= fetched_to)
    if created_from:
        filters.append(Article.created_at >= created_from)
    if created_to:
        filters.append(Article.created_at <= created_to)
    return filters


@router.get("/api/articles/search")
def search_articles(
    q: str | None = Query(None, description="substring of title or content_text"),
    article_id: int | None = Query(None, description="article.id"),
    site: str | None = Query(None, description="crawl_site.site_code"),
    target: str | None = Query(None, description="crawl_target.target_code"),
    dept: int | None = Query(None, description="local_department.dept_id (article.dept_id)"),
    status: str | None = Query(None, description="ready | failed | raw"),
    has_attachment: bool | None = Query(None),
    is_effective: bool | None = Query(None),
    native_post_id: str | None = Query(None),
    url: str | None = Query(None),
    url_hash: str | None = Query(None),
    title: str | None = Query(None),
    source_raw: str | None = Query(None),
    channel_name: str | None = Query(None),
    channel_path: str | None = Query(None),
    content_category: str | None = Query(None),
    content_subcategory: str | None = Query(None),
    index_no: str | None = Query(None),
    doc_no: str | None = Query(None),
    publisher: str | None = Query(None),
    topic_words: str | None = Query(None),
    open_category: str | None = Query(None),
    fetch_strategy: str | None = Query(None),
    rag_export_status: str | None = Query(None),
    publish_date_from: date | None = None,
    publish_date_to: date | None = None,
    effective_date_from: date | None = None,
    effective_date_to: date | None = None,
    expiry_date_from: date | None = None,
    expiry_date_to: date | None = None,
    fetched_from: datetime | None = None,
    fetched_to: datetime | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    since: datetime | None = None,
    limit: int = Query(20, ge=1, le=200),
    offset: int = Query(0, ge=0),
    s: Session = Depends(_session),
) -> dict[str, Any]:
    """Server-side pagination. UI keeps `limit` small (default 20 = one page)
    and pages by varying `offset`. Returns `total` so the front-end can
    render correct pagination controls.

    Historical context: the UI previously asked for limit=5000 and did
    client-side slicing. Once total articles crossed 5000 every record
    past the cap silently disappeared from the UI. limit is now capped
    at 200 to enforce real pagination."""
    # Build the WHERE clause once, apply to both COUNT and SELECT.
    filters = _article_filters(
        q=q, article_id=article_id, site=site, target=target, dept=dept,
        status=status, has_attachment=has_attachment, is_effective=is_effective,
        native_post_id=native_post_id, url=url, url_hash=url_hash, title=title,
        source_raw=source_raw, channel_name=channel_name, channel_path=channel_path,
        content_category=content_category, content_subcategory=content_subcategory,
        index_no=index_no, doc_no=doc_no, publisher=publisher,
        topic_words=topic_words, open_category=open_category,
        fetch_strategy=fetch_strategy, rag_export_status=rag_export_status,
        publish_date_from=publish_date_from, publish_date_to=publish_date_to,
        effective_date_from=effective_date_from, effective_date_to=effective_date_to,
        expiry_date_from=expiry_date_from, expiry_date_to=expiry_date_to,
        fetched_from=fetched_from, fetched_to=fetched_to,
        created_from=created_from, created_to=created_to, since=since,
    )

    # Total count over the filtered set. Joins are needed when filters
    # touch CrawlSite/CrawlTarget; the LEFT OUTER joins are safe (no
    # duplication) and Article.id is unique.
    count_stmt = (
        select(func.count(Article.id))
        .select_from(Article)
        .join(CrawlSite, CrawlSite.id == Article.site_id, isouter=True)
        .join(CrawlTarget, CrawlTarget.id == Article.target_id, isouter=True)
    )
    for f in filters:
        count_stmt = count_stmt.where(f)
    total = s.execute(count_stmt).scalar() or 0

    stmt = _articles_base_stmt().order_by(desc(Article.fetched_at)).offset(offset).limit(limit)
    for f in filters:
        stmt = stmt.where(f)
    rows = s.execute(stmt).all()
    return {
        "count": len(rows),
        "total": total,
        "limit": limit,
        "offset": offset,
        "items": [
            {
                **_article_payload(r.Article, r),
            }
            for r in rows
        ],
    }


@router.get("/api/articles/search/ids")
def search_articles_ids(
    q: str | None = Query(None),
    article_id: int | None = Query(None),
    site: str | None = Query(None),
    target: str | None = Query(None),
    dept: int | None = Query(None),
    status: str | None = Query(None),
    has_attachment: bool | None = Query(None),
    is_effective: bool | None = Query(None),
    native_post_id: str | None = Query(None),
    url: str | None = Query(None),
    url_hash: str | None = Query(None),
    title: str | None = Query(None),
    source_raw: str | None = Query(None),
    channel_name: str | None = Query(None),
    channel_path: str | None = Query(None),
    content_category: str | None = Query(None),
    content_subcategory: str | None = Query(None),
    index_no: str | None = Query(None),
    doc_no: str | None = Query(None),
    publisher: str | None = Query(None),
    topic_words: str | None = Query(None),
    open_category: str | None = Query(None),
    fetch_strategy: str | None = Query(None),
    rag_export_status: str | None = Query(None),
    publish_date_from: date | None = None,
    publish_date_to: date | None = None,
    effective_date_from: date | None = None,
    effective_date_to: date | None = None,
    expiry_date_from: date | None = None,
    expiry_date_to: date | None = None,
    fetched_from: datetime | None = None,
    fetched_to: datetime | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    since: datetime | None = None,
    s: Session = Depends(_session),
) -> dict[str, Any]:
    """Lightweight IDs-only counterpart to /api/articles/search. Used by the
    UI's "select all matching filter" button — paginated rendering means
    the row set only has the current page, but bulk-delete should be able
    to act on every record matching the filter regardless of page."""
    stmt = (
        select(Article.id)
        .select_from(Article)
        .join(CrawlSite, CrawlSite.id == Article.site_id, isouter=True)
        .join(CrawlTarget, CrawlTarget.id == Article.target_id, isouter=True)
        .order_by(desc(Article.fetched_at))
    )
    for f in _article_filters(
        q=q, article_id=article_id, site=site, target=target, dept=dept,
        status=status, has_attachment=has_attachment, is_effective=is_effective,
        native_post_id=native_post_id, url=url, url_hash=url_hash, title=title,
        source_raw=source_raw, channel_name=channel_name, channel_path=channel_path,
        content_category=content_category, content_subcategory=content_subcategory,
        index_no=index_no, doc_no=doc_no, publisher=publisher,
        topic_words=topic_words, open_category=open_category,
        fetch_strategy=fetch_strategy, rag_export_status=rag_export_status,
        publish_date_from=publish_date_from, publish_date_to=publish_date_to,
        effective_date_from=effective_date_from, effective_date_to=effective_date_to,
        expiry_date_from=expiry_date_from, expiry_date_to=expiry_date_to,
        fetched_from=fetched_from, fetched_to=fetched_to,
        created_from=created_from, created_to=created_to, since=since,
    ):
        stmt = stmt.where(f)
    ids = [row[0] for row in s.execute(stmt).all()]
    return {"ids": ids, "total": len(ids)}


def retry_failed_articles_to_rag(article_ids: list[int]) -> None:
    """Run selected RAG import jobs sequentially in a FastAPI background task."""
    from govcrawler.rag.exporter import RagExporter

    exporter = RagExporter()
    try:
        for article_id in article_ids:
            exporter.export_pending(article_id=article_id, source="manual")
    finally:
        exporter.close()


def import_all_pending_articles_to_rag(article_ids: list[int]) -> None:
    """Run one full-table RAG import snapshot in the background."""
    from govcrawler.rag.exporter import RagExporter

    exporter = RagExporter()
    try:
        for article_id in article_ids:
            exporter.export_pending(article_id=article_id, source="manual")
    finally:
        exporter.close()


@router.post("/api/articles/rag-import-all")
def import_all_rag_articles(
    background: BackgroundTasks,
    s: Session = Depends(_session),
) -> dict[str, Any]:
    """Queue all RAG-pending or RAG-failed ready articles for ingestion.

    This is intentionally not selection-based: the Articles page's "全部入库"
    button means the whole article table, not the checked rows or current page.
    """
    eligible_ids = [
        article_id
        for (article_id,) in (
            s.query(Article.id)
            .filter(
                Article.status == "ready",
                Article.exported_to_rag_at.is_(None),
                or_(
                    Article.rag_export_status.is_(None),
                    Article.rag_export_status.in_(("pending", "failed")),
                ),
            )
            .order_by(Article.fetched_at.asc(), Article.id.asc())
            .all()
        )
    ]
    eligible_count = len(eligible_ids)
    if eligible_ids:
        background.add_task(import_all_pending_articles_to_rag, eligible_ids)
    return {
        "queued": bool(eligible_ids),
        "eligible_count": eligible_count,
    }


@router.post("/api/articles/rag-retry")
def retry_failed_rag_articles(
    background: BackgroundTasks,
    payload: dict[str, Any] = Body(...),
    s: Session = Depends(_session),
) -> dict[str, Any]:
    """Queue selected unexported articles for RAG ingestion.

    Only ready articles that are not already exported and whose RAG export
    state is empty/pending/failed are eligible. Other selected ids are reported
    as skipped so the UI can send the full current selection safely.
    """
    article_ids = payload.get("article_ids") or []
    if not isinstance(article_ids, list) or not article_ids:
        raise HTTPException(400, "article_ids must be a non-empty array")
    if len(article_ids) > 200:
        raise HTTPException(400, "article_ids must be <= 200")

    normalized_ids: list[int] = []
    for raw_id in article_ids:
        try:
            normalized_ids.append(int(raw_id))
        except (TypeError, ValueError):
            raise HTTPException(400, "article_ids must contain integers")
    normalized_ids = list(dict.fromkeys(normalized_ids))

    rows = (
        s.query(Article)
        .filter(Article.id.in_(normalized_ids))
        .all()
    )
    by_id = {row.id: row for row in rows}
    eligible: list[int] = []
    skipped: list[dict[str, Any]] = []
    for article_id in normalized_ids:
        row = by_id.get(article_id)
        if row is None:
            skipped.append({"article_id": article_id, "reason": "not_found"})
            continue
        if row.status != "ready":
            skipped.append({"article_id": article_id, "reason": "article_not_ready"})
            continue
        if row.exported_to_rag_at is not None:
            skipped.append({"article_id": article_id, "reason": "already_exported"})
            continue
        if row.rag_export_status not in {None, "failed", "pending"}:
            skipped.append({"article_id": article_id, "reason": "rag_status_not_importable"})
            continue
        eligible.append(article_id)
        row.rag_export_status = "pending"
        row.rag_export_started_at = None
        row.rag_export_finished_at = None
        row.rag_export_error = None
        row.rag_export_task_ids = []

    if eligible:
        s.commit()
        background.add_task(retry_failed_articles_to_rag, eligible)
    else:
        s.rollback()

    return {
        "queued_count": len(eligible),
        "skipped_count": len(skipped),
        "article_ids": eligible,
        "skipped": skipped,
    }


def _safe_delete_file(rel_path: str | None, *, data_dir: Path) -> bool:
    """Delete a file under data_dir if it sits inside it. Path-traversal safe."""
    if not rel_path:
        return False
    abs_path = to_os_path(data_dir, PurePosixPath(rel_path))
    try:
        abs_path.resolve().relative_to(data_dir.resolve())
    except Exception:
        return False
    if not abs_path.exists():
        return False
    try:
        abs_path.unlink()
        return True
    except OSError as e:
        log.warning("delete file failed path=%s err=%s", abs_path, e)
        return False


def _article_file_paths(a: Article) -> list[str]:
    paths: list[str] = []
    for att in a.attachments:
        if att.file_path:
            paths.append(att.file_path)
    if a.raw_html_path:
        paths.append(a.raw_html_path)
    if a.text_path:
        paths.append(a.text_path)
    return paths


def _delete_article_paths(paths: list[str], *, data_dir: Path) -> list[str]:
    files_removed: list[str] = []
    for rel_path in paths:
        if _safe_delete_file(rel_path, data_dir=data_dir):
            files_removed.append(rel_path)
    return files_removed


def _delete_article_files(a: Article, *, data_dir: Path) -> list[str]:
    return _delete_article_paths(_article_file_paths(a), data_dir=data_dir)


@router.delete("/api/articles/{article_id}")
def delete_article(article_id: int, s: Session = Depends(_session)) -> dict[str, Any]:
    """Hard-delete an article + all its attachments (DB rows + files on disk).

    The DB cascades: attachment.article_id has ondelete=CASCADE, so the
    attachment rows go automatically. We additionally remove the on-disk
    files (raw_html, articles_text, attachment payloads) so a re-crawl of
    the same URL doesn't trip the url_hash UNIQUE check while leaving
    orphan files lying around.
    """
    a = s.get(Article, article_id)
    if a is None:
        raise HTTPException(404, "article not found")
    data_dir = Path(get_settings().data_dir)

    files_removed = _delete_article_files(a, data_dir=data_dir)

    s.delete(a)  # cascades to attachment rows via ondelete=CASCADE
    s.commit()
    return {
        "deleted": True,
        "article_id": article_id,
        "attachments_removed": len(files_removed),
        "files_removed": files_removed,
    }


@router.post("/api/articles/{article_id}/refetch")
def refetch_article_now(article_id: int, s: Session = Depends(_session)) -> dict[str, Any]:
    """Immediately re-fetch a single article URL.

    This intentionally bypasses the per-site task queue: the article detail
    action is for one known URL, so operators expect it to finish in this
    request instead of waiting behind target-level crawls.
    """
    row = s.execute(
        select(Article, CrawlTarget.target_code.label("target_code"))
        .select_from(Article)
        .join(CrawlTarget, CrawlTarget.id == Article.target_id, isouter=True)
        .where(Article.id == article_id)
    ).first()
    if row is None:
        raise HTTPException(404, "article not found")
    a = row.Article
    target_code = row.target_code
    if not target_code:
        raise HTTPException(400, "article is missing crawl target")
    url = a.url
    if not url:
        raise HTTPException(400, "article is missing url")

    data_dir = Path(get_settings().data_dir)
    old_file_paths = _article_file_paths(a)
    s.commit()

    try:
        from govcrawler.pipeline import fetch_and_store
    except ImportError as e:
        raise HTTPException(501, f"pipeline.fetch_and_store not wired for v2: {e}")

    try:
        result = fetch_and_store(
            target_code=target_code,
            url=url,
            replace_article_id=article_id,
        )
    except Exception as e:
        log.exception("article refetch failed article_id=%s target=%s url=%s", article_id, target_code, url)
        return {
            "executed": True,
            "article_id": article_id,
            "target_code": target_code,
            "url": url,
            "files_removed": [],
            "result": {"status": "failed", "reason": f"{type(e).__name__}: {e}"},
        }
    files_removed: list[str] = []
    if result.get("status") == "ready":
        s.expire_all()
        fresh = s.get(Article, article_id)
        current_file_paths = set(_article_file_paths(fresh)) if fresh else set()
        stale_file_paths = [p for p in old_file_paths if p not in current_file_paths]
        files_removed = _delete_article_paths(stale_file_paths, data_dir=data_dir)

    return {
        "executed": True,
        "article_id": article_id,
        "target_code": target_code,
        "url": url,
        "files_removed": files_removed,
        "result": result,
    }


@router.post("/api/articles/bulk-delete")
def bulk_delete_articles(
    payload: dict[str, Any] = Body(...),
    s: Session = Depends(_session),
) -> dict[str, Any]:
    """Hard-delete a batch of articles (and their attachments + on-disk
    files). Used by the 文章列表 multi-select action. Cap 1000 / call.

    Body shape (one of):
      {"article_ids": [1,2,3,...]}     # explicit ids
      {"target_code": "..."}           # delete every article tagged with
                                        # this crawl_target (handy for
                                        # 'wipe and re-run with --force')
      {"site_code": "..."}             # delete every article from a site
    """
    article_ids = payload.get("article_ids") or []
    target_code = (payload.get("target_code") or "").strip() or None
    site_code = (payload.get("site_code") or "").strip() or None

    if not article_ids and not target_code and not site_code:
        raise HTTPException(400, "provide article_ids, target_code, or site_code")
    if article_ids and not isinstance(article_ids, list):
        raise HTTPException(400, "article_ids must be an array")

    q = s.query(Article)
    if article_ids:
        if len(article_ids) > 1000:
            raise HTTPException(400, "article_ids must be <= 1000")
        q = q.filter(Article.id.in_(article_ids))
    if target_code:
        from govcrawler.models import CrawlTarget
        t = s.query(CrawlTarget).filter_by(target_code=target_code).first()
        if t is None:
            raise HTTPException(404, f"target not found: {target_code}")
        q = q.filter(Article.target_id == t.id)
    if site_code:
        from govcrawler.models import CrawlSite
        site = s.query(CrawlSite).filter_by(site_code=site_code).first()
        if site is None:
            raise HTTPException(404, f"site not found: {site_code}")
        q = q.filter(Article.site_id == site.id)

    rows = q.all()
    if len(rows) > 1000:
        raise HTTPException(400, f"refusing to delete {len(rows)} articles in one call (>1000) — narrow the filter")

    data_dir = Path(get_settings().data_dir)
    files_removed = 0
    for a in rows:
        for att in a.attachments:
            if _safe_delete_file(att.file_path, data_dir=data_dir):
                files_removed += 1
        if _safe_delete_file(a.raw_html_path, data_dir=data_dir):
            files_removed += 1
        if _safe_delete_file(a.text_path, data_dir=data_dir):
            files_removed += 1
        s.delete(a)
    s.commit()
    return {
        "deleted_count": len(rows),
        "files_removed": files_removed,
    }


# ---------------------------------------------------------------------------
# Article detail proxies for the admin UI.
# ---------------------------------------------------------------------------
# These routes mirror the public RAG handlers in govcrawler/api/app.py but
# live under /admin/* so the dashboard's modal / "查看原始 HTML" / 附件下载
# links inherit the admin Basic Auth gate. Without these mirrors the modal
# called the public path /api/articles/<id> directly — that's protected by
# _RagTokenGate (Bearer scheme), and browser-cached Basic Auth from /admin
# never auto-flows to a different scheme realm, so the UI got 401 even
# while the operator was logged in. Same logic, just registered under the
# admin router.

def _serve_admin_article_file(rel_path: str | None, media_type: str,
                              download_name: str) -> FileResponse:
    if not rel_path:
        raise HTTPException(404, "path not set on article")
    data_dir = Path(get_settings().data_dir)
    abs_path = to_os_path(data_dir, PurePosixPath(rel_path))
    try:
        abs_path.resolve().relative_to(data_dir.resolve())
    except Exception:
        raise HTTPException(400, "invalid path")
    if not abs_path.exists():
        raise HTTPException(404, "file missing on disk")
    return FileResponse(
        path=str(abs_path), filename=download_name, media_type=media_type,
    )


@router.get("/api/articles/{article_id}")
def admin_get_article(article_id: int, s: Session = Depends(_session)) -> dict[str, Any]:
    a = s.get(Article, article_id)
    if a is None:
        raise HTTPException(404, "article not found")
    _ = a.site, a.target  # eager-load codes
    site_code = a.site.site_code if a.site_id and a.site else None
    target_code = a.target.target_code if a.target_id and a.target else None
    payload = _article_payload(a, include_heavy=True)
    payload.update({"site_code": site_code, "target_code": target_code})
    payload["attachments"] = [
            {
                "id": att.id,
                "file_name": att.file_name,
                "file_ext": att.file_ext,
                "size_bytes": att.size_bytes,
                "file_hash": att.file_hash,
                "source_url": att.source_url,
            }
            for att in a.attachments
        ]
    return payload


def _openstd_hcno(a: Article) -> str | None:
    metadata = a.metadata_json if isinstance(a.metadata_json, dict) else {}
    public_meta = metadata.get("public_meta") if isinstance(metadata, dict) else {}
    candidates = [
        public_meta.get("openstd_hcno") if isinstance(public_meta, dict) else None,
        metadata.get("openstd_hcno") if isinstance(metadata, dict) else None,
        a.native_post_id,
    ]
    for value in candidates:
        if isinstance(value, str) and value.strip():
            return value.strip().upper()
    return None


def _openstd_fallback_filename(a: Article) -> str:
    stem = " ".join(x for x in [a.doc_no, a.title] if x).strip()
    return f"{stem or _openstd_hcno(a) or 'openstd-standard'}.pdf"


@router.post("/api/articles/{article_id}/openstd-download/start")
def admin_openstd_download_start(
    article_id: int, s: Session = Depends(_session),
) -> dict[str, Any]:
    a = s.get(Article, article_id)
    if a is None:
        raise HTTPException(404, "article not found")
    site_code = a.site.site_code if a.site_id and a.site else None
    if site_code != "openstd_samr":
        raise HTTPException(400, "openstd download is only supported for openstd_samr")
    hcno = _openstd_hcno(a)
    if not hcno:
        raise HTTPException(400, "openstd hcno missing")
    try:
        sess = openstd_download.start_session(hcno, article_id=article_id)
    except ValueError as e:
        raise HTTPException(400, str(e))
    except Exception as e:
        raise HTTPException(502, f"failed to start openstd download session: {e}")
    return {
        "session_id": sess.session_id,
        "captcha_url": f"/admin/api/openstd-download-sessions/{sess.session_id}/captcha",
        "expires_at": epoch_iso_cn(sess.expires_at),
    }


@router.get("/api/openstd-download-sessions/{session_id}/captcha")
def admin_openstd_download_captcha(session_id: str) -> Response:
    try:
        sess = openstd_download.refresh_captcha(session_id)
    except KeyError:
        raise HTTPException(404, "openstd download session expired")
    except Exception as e:
        raise HTTPException(502, f"failed to refresh openstd captcha: {e}")
    return Response(content=sess.captcha_bytes, media_type=sess.captcha_content_type)


@router.post("/api/openstd-download-sessions/{session_id}/submit")
def admin_openstd_download_submit(
    session_id: str,
    payload: dict[str, Any] = Body(default_factory=dict),
    s: Session = Depends(_session),
) -> dict[str, Any]:
    captcha = str(payload.get("captcha") or "").strip()
    try:
        sess = openstd_download.get_session(session_id)
    except KeyError:
        raise HTTPException(404, "openstd download session expired")
    if sess.article_id is None:
        raise HTTPException(400, "openstd download session is missing article binding")
    a = s.get(Article, sess.article_id)
    if a is None:
        openstd_download.close_session(session_id)
        raise HTTPException(404, "article not found")
    site_code = a.site.site_code if a.site_id and a.site else None
    target_code = a.target.target_code if a.target_id and a.target else None
    if site_code != "openstd_samr":
        openstd_download.close_session(session_id)
        raise HTTPException(400, "openstd download is only supported for openstd_samr")
    when = a.publish_time or a.fetched_at or now_cn_naive()
    try:
        d = openstd_download.submit_captcha_and_download(
            session_id,
            captcha,
            site=site_code,
            column=target_code or "openstd_samr",
            when=when,
            article_key=f"a_{a.id}",
            fallback_name=_openstd_fallback_filename(a),
        )
    except openstd_download.OpenStdCaptchaInvalid:
        raise HTTPException(
            409,
            {
                "reason": "captcha_invalid",
                "captcha_url": f"/admin/api/openstd-download-sessions/{session_id}/captcha",
            },
        )
    except KeyError:
        raise HTTPException(404, "openstd download session expired")
    except Exception as e:
        raise HTTPException(502, f"openstd download failed: {e}")

    existing = s.scalar(
        select(Attachment).where(
            Attachment.article_id == a.id,
            Attachment.file_hash == d.file_hash,
        )
    )
    if existing is not None:
        att = existing
        if d.source_url and not att.source_url:
            att.source_url = d.source_url
        try:
            new_path = to_os_path(Path(get_settings().data_dir), PurePosixPath(d.file_path))
            if str(new_path) != str(to_os_path(Path(get_settings().data_dir), PurePosixPath(att.file_path))):
                new_path.unlink(missing_ok=True)
        except Exception:
            pass
    else:
        att = Attachment(
            article_id=a.id,
            file_name=d.file_name,
            file_ext=d.file_ext,
            size_bytes=d.size_bytes,
            file_path=str(d.file_path),
            file_hash=d.file_hash,
            source_url=d.source_url,
        )
        s.add(att)
    a.has_attachment = True
    s.commit()
    return {
        "status": "ok",
        "article_id": a.id,
        "attachment": {
            "id": att.id,
            "file_name": att.file_name,
            "file_ext": att.file_ext,
            "size_bytes": att.size_bytes,
            "file_hash": att.file_hash,
            "source_url": att.source_url,
        },
    }


@router.get("/api/articles/{article_id}/raw-html")
def admin_get_article_raw_html(
    article_id: int, s: Session = Depends(_session),
) -> FileResponse:
    a = s.get(Article, article_id)
    if a is None:
        raise HTTPException(404, "article not found")
    return _serve_admin_article_file(
        a.raw_html_path, "text/html; charset=utf-8", f"article-{article_id}.html",
    )


@router.get("/api/articles/{article_id}/text")
def admin_get_article_text(
    article_id: int, s: Session = Depends(_session),
) -> FileResponse:
    a = s.get(Article, article_id)
    if a is None:
        raise HTTPException(404, "article not found")
    return _serve_admin_article_file(
        a.text_path, "text/plain; charset=utf-8", f"article-{article_id}.txt",
    )


@router.get("/api/articles/{article_id}/attachments/{attachment_id}")
def admin_download_attachment(
    article_id: int, attachment_id: int, s: Session = Depends(_session),
) -> FileResponse:
    att = s.get(Attachment, attachment_id)
    if att is None or att.article_id != article_id:
        raise HTTPException(404, "attachment not found")
    if not att.file_path:
        raise HTTPException(404, "attachment file_path empty")
    data_dir = Path(get_settings().data_dir)
    abs_path = to_os_path(data_dir, PurePosixPath(att.file_path))
    try:
        abs_path.resolve().relative_to(data_dir.resolve())
    except Exception:
        raise HTTPException(400, "invalid path")
    if not abs_path.exists():
        raise HTTPException(404, "file missing on disk")
    return FileResponse(
        path=str(abs_path),
        filename=att.file_name or abs_path.name,
        media_type="application/octet-stream",
    )
