hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/search_intent/parsing.py

"""搜索意图解析 —— LLM 意图提取、JSON 修复、重试纠错与意图规范化."""

from __future__ import annotations

import json
import logging
import re
from datetime import datetime
from typing import TYPE_CHECKING, Any

logger = logging.getLogger(__name__)

from app.core.search.paper_searcher import _sanitize_author_list_for_query


from .arxiv_normalization import sanitize_arxiv_categories, sanitize_arxiv_id_list
from ...utils.common import dedupe_strings_preserve_order
from .venue_phrases import sanitize_venue_tokens

if TYPE_CHECKING:
    from ...agents.support.search_models import SearchIntent

def _search_intent_cls():
    from ...agents.support.search_models import SearchIntent

    return SearchIntent

def strip_retrieval_meta_from_query(q: str) -> str:
    s = (q or "").strip()
    return " ".join(s.split()) if s else ""

def _fence_inner_after_first_fence(t: str) -> str | None:
    first = t.find("```")
    if first < 0:
        return None
    rest = t[first + 3 :]
    rl = rest.lstrip()
    if rl[:4].lower() == "json":
        rl = rl[4:].lstrip(" \t\r\n")
    end = rl.find("```")
    body = rl[:end].strip() if end >= 0 else rl.strip()
    return body or None

def extract_json_object(text: str) -> dict[str, Any | None]:
    t = (text or "").strip()
    if not t:
        return None
    if "```" in t and (boxed := _fence_inner_after_first_fence(t)):
        t = boxed
    start = t.find("{")
    end = t.rfind("}")
    if start < 0 or end <= start:
        return None
    raw = t[start : end + 1]
    raw = re.sub(r",\s*([\]}])", r"\1", raw)
    try:
        out = json.loads(raw)
        return out if isinstance(out, dict) else None
    except json.JSONDecodeError:
        return None

_SORT_ALLOWED = frozenset({"relevance", "date"})
_CONFIDENCE_ALLOWED = frozenset({"high", "medium", "low"})
_SEARCH_STRATEGY_ALLOWED = frozenset(
    {"keyword_matching", "semantic_search", "hybrid", "targeted_lookup"}
)

def _intent_str_list_field(source: Any, cap: int) -> list[str]:
    if not isinstance(source, list):
        return []
    return [str(x).strip() for x in source if str(x).strip()][:cap]

def _intent_str_or_list_field(source: Any, cap: int) -> list[str] | None:
    if isinstance(source, str) and source.strip():
        return [source.strip()][:cap]
    if isinstance(source, list):
        return [str(x).strip() for x in source if str(x).strip()][:cap]
    return None

def _intent_year_field(v: Any) -> int | None:
    if isinstance(v, (int, float)) and 1900 <= int(v) <= 2100:
        return int(v)
    return None

def _intent_bounded_int(v: Any, lo: int, hi: int) -> int | None:
    if isinstance(v, (int, float)):
        return max(lo, min(int(v), hi))
    return None

def _intent_flag_merge(flags: dict[str, Any], d: dict[str, Any], key: str) -> Any:
    return flags[key] if key in flags else d.get(key)

def _intent_use_tavily_raw(nested: bool, flags: dict[str, Any], d: dict[str, Any]) -> Any:
    if nested:
        return flags["use_tavily"] if "use_tavily" in flags else d.get("use_tavily")
    return d.get("use_tavily")

def search_intent_from_dict(d: dict[str, Any]) -> SearchIntent:
    nested = isinstance(d.get("search"), dict)
    if nested:
        s = dict(d["search"])
        flags = dict(d["flags"]) if isinstance(d.get("flags"), dict) else {}
        rank = dict(d["ranking"]) if isinstance(d.get("ranking"), dict) else {}
    else:
        s, flags, rank = dict(d), {}, {}

    Si = _search_intent_cls()
    intent = Si()

    intent.query = str(s.get("query") or "").strip()[:500]
    intent.keywords = _intent_str_list_field(s.get("keywords") or [], 16)
    intent.authors = _intent_str_list_field(s.get("authors") or d.get("authors") or [], 8)
    intent.venues = _intent_str_list_field(s.get("venues") or [], 8)

    tt_raw = s.get("target_titles") or s.get("paper_titles") or s.get("paper_title") or []
    if (tt_parsed := _intent_str_or_list_field(tt_raw, 6)) is not None:
        intent.target_titles = tt_parsed
    ta_raw = s.get("target_authors") or s.get("target_author") or []
    if (ta_parsed := _intent_str_or_list_field(ta_raw, 6)) is not None:
        intent.target_authors = ta_parsed

    if (yf := _intent_year_field(s.get("year_from"))) is not None:
        intent.year_from = yf
    if (yt := _intent_year_field(s.get("year_to"))) is not None:
        intent.year_to = yt

    sort = str(s.get("sort") or "relevance").lower()
    intent.sort = sort if sort in _SORT_ALLOWED else "relevance"

    if (mr := _intent_bounded_int(s.get("max_results", 10), 5, 30)) is not None:
        intent.max_results = mr

    intent.arxiv_categories = _intent_str_list_field(s.get("arxiv_categories") or [], 12)

    axl_raw = s.get("arxiv_id_list") or s.get("pinned_arxiv_ids") or d.get("arxiv_id_list")
    if (ax_parsed := _intent_str_or_list_field(axl_raw, 8)) is not None:
        intent.arxiv_id_list = ax_parsed

    intent.is_short_acronym = bool(_intent_flag_merge(flags, d, "is_short_acronym"))
    intent.wants_classic = bool(_intent_flag_merge(flags, d, "wants_classic"))
    intent.wants_recent = bool(_intent_flag_merge(flags, d, "wants_recent"))
    intent.main_conference_proceedings_only = bool(
        _intent_flag_merge(flags, d, "main_conference_proceedings_only")
    )
    ut = _intent_use_tavily_raw(nested, flags, d)
    intent.use_tavily = None if ut is None else bool(ut)

    conf = str(_intent_flag_merge(flags, d, "confidence_level") or "").strip().lower()
    if conf in _CONFIDENCE_ALLOWED:
        intent.confidence_level = conf

    strat = str(_intent_flag_merge(flags, d, "search_strategy") or "").strip().lower()
    if strat in _SEARCH_STRATEGY_ALLOWED:
        intent.search_strategy = strat

    llm_sources = flags.get("sources") or d.get("sources") or []
    if isinstance(llm_sources, list) and llm_sources:
        allowed = {"arxiv", "dblp", "openalex", "tavily"}
        intent.sources = [str(src).strip().lower() for src in llm_sources if str(src).strip().lower() in allowed]

    rk_strat = str(flags.get("ranking_strategy") or d.get("ranking_strategy") or "").strip().lower()
    if rk_strat in ("date", "relevance", "hybrid"):
        intent.ranking_strategy = rk_strat

    use_llm = rank.get("use_llm_rank", rank.get("use_two_stage_rerank", d.get("use_llm_rank", d.get("use_two_stage_rerank", True))))
    intent.use_llm_rank = bool(use_llm)
    rc_raw = rank.get("rerank_recall_max", rank.get("rerank_coarse_top_n", d.get("rerank_recall_max", d.get("rerank_coarse_top_n", 24))))
    if (rc := _intent_bounded_int(rc_raw, 8, 60)) is not None:
        intent.rerank_recall_max = rc

    rat = rank.get("rationale", d.get("ranking_rationale"))
    if rat is not None:
        intent.ranking_rationale = str(rat).strip()[:800]

    return intent

def finalize_llm_intent(intent: SearchIntent, profile: str) -> SearchIntent:
    intent.query = strip_retrieval_meta_from_query((intent.query or "")[:500])
    intent.keywords = dedupe_strings_preserve_order(list(intent.keywords or []), max_n=16)

    if not (intent.query or "").strip() and intent.keywords:
        intent.query = intent.keywords[0][:500]

    if not (intent.query or "").strip() and (intent.authors or []):
        intent.query = str((intent.authors or [])[0]).strip()[:500]

    from ...utils.author_query_match import normalize_author_names

    intent.authors = normalize_author_names(
        [str(x).strip() for x in (intent.authors or []) if str(x).strip()]
    )[:8]
    raw_venues = [str(x).strip() for x in (intent.venues or []) if str(x).strip()]
    intent.venues = sanitize_venue_tokens(raw_venues)[:8]
    intent.target_titles = [str(x).strip() for x in (intent.target_titles or []) if str(x).strip()][:6]
    from ..retrieval.method_acronym import is_method_acronym_token

    q_strip = (intent.query or "").strip()
    if intent.venues and is_method_acronym_token(q_strip):
        intent.keywords = [q_strip]
        intent.is_short_acronym = True
        intent.use_tavily = True
    intent.target_authors = [str(x).strip() for x in (intent.target_authors or []) if str(x).strip()][:6]
    intent.authors = _sanitize_author_list_for_query(intent.query or "", intent.authors)
    intent.max_results = max(5, min(int(intent.max_results or 10), 30))
    intent.rerank_recall_max = max(8, min(int(intent.rerank_recall_max or 24), 60))
    conf = str(intent.confidence_level or "").strip().lower()
    intent.confidence_level = conf if conf in ("high", "medium", "low") else "medium"
    strat = str(intent.search_strategy or "").strip().lower()
    intent.search_strategy = (
        strat
        if strat in ("keyword_matching", "semantic_search", "hybrid", "targeted_lookup")
        else "hybrid"
    )
    if profile == "novelty":
        intent.sort = "date"
    elif intent.sort not in ("relevance", "date"):
        intent.sort = "relevance"

    if bool(intent.wants_recent) and intent.sort != "date":
        intent.sort = "date"

    axl = [str(x).strip() for x in (intent.arxiv_id_list or []) if str(x).strip()][:8]
    intent.arxiv_id_list = axl

    if intent.year_from is not None and (intent.year_from < 1900 or intent.year_from > 2100):
        intent.year_from = None
    if intent.year_to is not None and (intent.year_to < 1900 or intent.year_to > 2100):
        intent.year_to = None
    _ensure_intent_year_window_ordered(intent)
    return intent

def _ensure_intent_year_window_ordered(intent: SearchIntent) -> None:
    yf = getattr(intent, "year_from", None)
    yt = getattr(intent, "year_to", None)
    if isinstance(yf, int) and isinstance(yt, int) and yf > yt:
        intent.year_from, intent.year_to = yt, yf

def infer_target_edition_year_for_recent(*, is_latest: bool = True, settings: Any | None = None) -> int:
    """Prompt hint for the latest likely conference edition year."""
    _ = is_latest, settings
    y_now = int(datetime.now().year)
    return max(1990, y_now - 1)


def format_intent_llm_prompt(
    template: str,
    user_text: str,
    profile: str,
    *,
    correction_hint: str | None = None,
) -> str:
    now = datetime.now()
    edition_year = infer_target_edition_year_for_recent(is_latest=True)
    base = template.format(
        user_text=(user_text or "").strip()[:3500],
        profile=profile,
        current_date_iso=now.strftime("%Y-%m-%d"),
        current_year=now.year,
        suggested_edition_year=edition_year,
    )
    hint = (correction_hint or "").strip()
    if not hint:
        return base
    return (
        f"{base}\n\n"
        "## 修正要求（上次输出无效，请重新生成完整 JSON）\n"
        f"{hint}\n\n"
        "只输出一个 JSON 对象，不要 markdown 代码块或解释文字。"
    )


def build_intent_retry_correction_hint(
    exc: BaseException,
    *,
    user_message: str,
    last_output: str | None = None,
) -> str:
    em = str(exc or "").strip()
    em_l = em.lower()
    msg = (user_message or "").strip()[:400]
    parts: list[str] = [f"用户查询：「{msg}」。"]
    if "empty_query" in em_l or "intent_parse_empty" in em_l:
        parts.append(
            "上次 JSON 缺少有效检索锚点：query、venues、authors、target_titles、arxiv_id_list 至少应有一项非空；"
            "若用户只提会议/最新，请把会议写入 venues，并按上文日期规则填写 year_from/year_to。"
        )
    elif "json" in em_l or "未返回有效" in em:
        parts.append("上次未返回可解析的 JSON 对象。")
    else:
        parts.append(f"上次解析失败：{em[:400]}。")
    out_snip = (last_output or "").strip()
    if out_snip:
        parts.append(f"上次模型输出片段（供对照，勿照抄错误）：{out_snip[:600]}")
    return " ".join(parts)


def apply_llm_intent_hygiene(intent: SearchIntent, _raw_user_text: str | None = None) -> None:
    """Clean parsed fields without changing venue/year semantics."""
    _ = _raw_user_text
    if intent is None:
        return
    intent.venues = sanitize_venue_tokens(list(intent.venues or []))[:8]
    intent.keywords = dedupe_strings_preserve_order(list(intent.keywords or []), max_n=16)
    intent.arxiv_categories = sanitize_arxiv_categories(list(intent.arxiv_categories or []))
    intent.arxiv_id_list = sanitize_arxiv_id_list(list(intent.arxiv_id_list or []))


__all__ = [
    "extract_json_object",
    "search_intent_from_dict",
    "finalize_llm_intent",
    "apply_llm_intent_hygiene",
    "build_intent_retry_correction_hint",
    "format_intent_llm_prompt",
    "infer_target_edition_year_for_recent",
    "_ensure_intent_year_window_ordered",
]