ranking.py

from __future__ import annotations

from datetime import datetime


def clamp(value: float) -> float:
    return max(0.0, min(1.0, value))


def token_overlap_score(query: str, text: str) -> float:
    query_tokens = {token for token in query.lower().split() if len(token) > 2}
    text_tokens = {token for token in text.lower().split() if len(token) > 2}
    if not query_tokens:
        return 0.0
    return len(query_tokens & text_tokens) / len(query_tokens)


def compute_feature_set(item: dict, query: str, domain_profiles: list[dict]) -> dict[str, float]:
    text = " ".join([item.get("title", ""), item.get("abstract", ""), " ".join(item.get("keywords", []))])
    relevance_score = token_overlap_score(query, text)
    venue_score = venue_authority_score(item.get("venue", ""), domain_profiles)
    citation_score = citation_to_score(item.get("citation_count"))
    recency_score = recency_to_score(item.get("year"))
    survey_foundation_score = survey_foundation_to_score(item)
    cross_domain_score = cross_domain_to_score(item, domain_profiles)
    domain_profile_boost = profile_boost(item, domain_profiles)

    return {
        "relevance_score": round(clamp(relevance_score), 4),
        "venue_score": round(clamp(venue_score), 4),
        "citation_score": round(clamp(citation_score), 4),
        "recency_score": round(clamp(recency_score), 4),
        "survey_foundation_score": round(clamp(survey_foundation_score), 4),
        "cross_domain_score": round(clamp(cross_domain_score), 4),
        "domain_profile_boost": round(clamp(domain_profile_boost), 4),
    }


def venue_authority_score(venue_name: str, domain_profiles: list[dict]) -> float:
    for profile in domain_profiles:
        for venue in profile.get("top_venues", []):
            if venue["name"].lower() == venue_name.lower():
                return float(venue["weight"])
    return 0.42


def citation_to_score(citations: int | None) -> float:
    if citations is None:
        return 0.3
    if citations >= 5000:
        return 1.0
    if citations >= 1000:
        return 0.9
    if citations >= 300:
        return 0.75
    if citations >= 100:
        return 0.6
    if citations >= 30:
        return 0.45
    return 0.2


def recency_to_score(year: int | None) -> float:
    if year is None:
        return 0.35
    delta = datetime.utcnow().year - year
    if delta <= 1:
        return 1.0
    if delta <= 2:
        return 0.88
    if delta <= 4:
        return 0.72
    if delta <= 7:
        return 0.52
    return 0.28


def survey_foundation_to_score(item: dict) -> float:
    text = " ".join([item.get("title", ""), item.get("abstract", ""), " ".join(item.get("keywords", []))]).lower()
    if any(keyword in text for keyword in ["survey", "review", "benchmark", "foundation", "foundational"]):
        return 0.92
    if item.get("citation_count", 0) and item.get("citation_count", 0) > 400:
        return 0.78
    return 0.45


def cross_domain_to_score(item: dict, domain_profiles: list[dict]) -> float:
    if len(domain_profiles) < 2:
        return 0.2
    text = " ".join([item.get("title", ""), item.get("abstract", ""), " ".join(item.get("keywords", []))]).lower()
    matched = 0
    for profile in domain_profiles:
        domain_key = profile["key"]
        if domain_key == "medicine" and any(keyword in text for keyword in ["medical", "clinical", "patient", "health"]):
            matched += 1
        elif domain_key == "computer_science" and any(keyword in text for keyword in ["model", "learning", "neural", "graph"]):
            matched += 1
        elif domain_key == "biology" and any(keyword in text for keyword in ["genome", "protein", "cell", "biolog"]):
            matched += 1
        else:
            matched += 0
    return matched / len(domain_profiles)


def profile_boost(item: dict, domain_profiles: list[dict]) -> float:
    venue_boost = venue_authority_score(item.get("venue", ""), domain_profiles)
    keyword_text = " ".join(item.get("keywords", [])).lower()
    keyword_boost = 0.2 if any(keyword in keyword_text for keyword in ["survey", "benchmark", "review"]) else 0.0
    return min(1.0, venue_boost + keyword_boost)


def calculate_importance(features: dict[str, float]) -> dict:
    relevance = clamp(features["relevance_score"])
    venue = clamp(features["venue_score"])
    citation = clamp(features["citation_score"])
    recency = clamp(features["recency_score"])
    survey_foundation = clamp(features["survey_foundation_score"])
    cross_domain = clamp(features["cross_domain_score"])
    domain_profile = clamp(features["domain_profile_boost"])

    final_score = round(
        100
        * (
            0.30 * relevance
            + 0.22 * venue
            + 0.16 * citation
            + 0.10 * recency
            + 0.10 * survey_foundation
            + 0.07 * cross_domain
            + 0.05 * domain_profile
        ),
        1,
    )

    reasons = []
    if relevance > 0.60:
        reasons.append("Strong relevance to the query intent")
    if venue > 0.90:
        reasons.append("Published in a top-priority venue profile")
    if citation > 0.75:
        reasons.append("Strong citation signal")
    if cross_domain > 0.60:
        reasons.append("Useful for cross-domain surveying")
    if survey_foundation > 0.80:
        reasons.append("Useful as survey, benchmark, or foundational reading")
    if not reasons:
        reasons.append("Retrieved as a relevant supporting paper")

    reading_level = (
        "starter" if survey_foundation > 0.8 or citation > 0.85 else "frontier" if recency > 0.85 else "advanced"
    )

    return {
        "final_score": final_score,
        "reasons": reasons,
        "reading_level": reading_level,
    }