hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/retrieval/plan_helpers.py

"""Derived helpers for ResolvedSearchPlan."""

from __future__ import annotations

from typing import TYPE_CHECKING

from .method_acronym import resolve_method_acronym


def _venues(plan: "ResolvedSearchPlan") -> list[str]:
    return [str(v).strip() for v in (plan.venues or []) if str(v).strip()]


def _keywords(plan: "ResolvedSearchPlan") -> list[str]:
    return [str(k).strip() for k in (plan.keywords or []) if str(k).strip()]

if TYPE_CHECKING:
    from .recall_context import RecallContext
    from .search_plan import ResolvedSearchPlan


def primary_venue(plan: "ResolvedSearchPlan") -> str | None:
    v = _venues(plan)
    return v[0] if v else None


def method_acronym_for(plan: "ResolvedSearchPlan", ctx: "RecallContext | None" = None) -> str:
    if (plan.method_acronym or "").strip():
        return str(plan.method_acronym).strip()
    if ctx is not None:
        return str(ctx.search_kwargs.get("method_acronym") or "").strip()
    return ""


def is_pinned_single_year(plan: "ResolvedSearchPlan") -> bool:
    yf, yt = plan.year_from, plan.year_to
    return bool(_venues(plan) and isinstance(yf, int) and isinstance(yt, int) and yf == yt)


def is_strict_venue_match(plan: "ResolvedSearchPlan") -> bool:
    return is_pinned_single_year(plan) and bool(plan.main_conference_proceedings_only)


def use_venue_proceedings_journal(plan: "ResolvedSearchPlan") -> bool:
    return bool(_venues(plan))


def pinned_research_topic(plan: "ResolvedSearchPlan") -> str:
    """Search topic after removing pinned venue/year terms."""
    from ...core.search.normalize import extract_pinned_topic_terms

    venue = primary_venue(plan) or ""
    year = plan.year_from if isinstance(plan.year_from, int) else None
    return extract_pinned_topic_terms(
        query=plan.query or "",
        merged_kw=list(plan.keywords or []),
        venue=venue,
        year=year,
    ).strip()


def is_venue_browse_plan(plan: "ResolvedSearchPlan") -> bool:
    """True for pure venue+year browsing."""
    if not is_pinned_single_year(plan) or not plan.main_conference_proceedings_only or not _venues(plan):
        return False
    if method_acronym_for(plan) or resolve_method_acronym(plan.query or "", _keywords(plan)):
        return False
    if plan.target_titles or plan.authors:
        return False
    return not bool(pinned_research_topic(plan))


def effective_max_results(plan: "ResolvedSearchPlan", requested: int) -> int:
    return max(int(requested), 30) if is_venue_browse_plan(plan) else int(requested)


def effective_recall_max_candidates(plan: "ResolvedSearchPlan", current: int) -> int:
    cur = int(current or 24)
    return max(cur, 24) if is_venue_browse_plan(plan) else cur


def tavily_configured() -> bool:
    try:
        from ...settings import get_settings

        return bool(str(getattr(get_settings(), "tavily_api_key", "") or "").strip())
    except Exception:
        return False


def should_supplement_from_proceedings_site(plan: "ResolvedSearchPlan") -> bool:
    """Venue searches can use proceedings supplement when Tavily is configured."""
    return (
        bool(_venues(plan))
        and tavily_configured()
    )


def should_supplement_from_intent_dict(intent: dict[str, Any]) -> bool:
    venues = [str(v).strip() for v in (intent.get("venues") or []) if str(v).strip()]
    if not venues:
        return False
    yf, yt = intent.get("year_from"), intent.get("year_to")
    try:
        pinned = yf is not None and yt is not None and int(yf) == int(yt)
    except (TypeError, ValueError):
        pinned = False
    return pinned and bool(intent.get("main_conference_proceedings_only")) and tavily_configured()