SupplyAgent/find_agent_v2/gates.py

"""Deterministic candidate gates owned by find_agent_v2."""

from __future__ import annotations

import json
import os
import re
from datetime import date, datetime, time
from decimal import Decimal
from typing import Any, Mapping
from zoneinfo import ZoneInfo


def _env_number(name: str, default: float) -> float:
    try:
        return float(os.getenv(name, str(default)))
    except ValueError:
        return default


def build_rule_snapshot(now: datetime | None = None) -> dict[str, Any]:
    timezone_name = os.getenv("FIND_AGENT_V2_TIMEZONE", "Asia/Shanghai")
    timezone = ZoneInfo(timezone_name)
    current = now or datetime.now(timezone)
    current = current.replace(tzinfo=timezone) if current.tzinfo is None else current.astimezone(timezone)
    return {
        "rule_version": os.getenv("FIND_AGENT_V2_RULE_VERSION", "find-agent-v2-gate-v1"),
        "timezone": timezone_name,
        "min_duration_seconds": int(_env_number("FIND_AGENT_V2_MIN_DURATION_SECONDS", 30)),
        "min_share_count": int(_env_number("FIND_AGENT_V2_MIN_SHARE_COUNT", 1000)),
        "min_content_50_plus_ratio": _env_number("FIND_AGENT_V2_MIN_CONTENT_50_PLUS_RATIO", 0.20),
        "min_account_50_plus_ratio": _env_number("FIND_AGENT_V2_MIN_ACCOUNT_50_PLUS_RATIO", 0.20),
        "event_max_age_days": int(_env_number("FIND_AGENT_V2_EVENT_MAX_AGE_DAYS", 7)),
        "seasonal_max_age_days": int(_env_number("FIND_AGENT_V2_SEASONAL_MAX_AGE_DAYS", 180)),
        "current_datetime": current.isoformat(timespec="seconds"),
        "current_date": current.strftime("%Y-%m-%d"),
    }


def _rules(value: Mapping[str, Any] | str | None) -> dict[str, Any]:
    if isinstance(value, str):
        try:
            loaded = json.loads(value)
        except (TypeError, ValueError):
            loaded = {}
    else:
        loaded = dict(value or {})
    return {**build_rule_snapshot(), **loaded}


def parse_datetime_value(value: Any, *, timezone_name: str = "Asia/Shanghai") -> datetime | None:
    if value in (None, "") or isinstance(value, bool):
        return None
    timezone = ZoneInfo(timezone_name)
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, date):
        parsed = datetime.combine(value, time.min)
    elif isinstance(value, (int, float, Decimal)):
        number = float(value)
        if number > 10_000_000_000:
            number /= 1000
        try:
            parsed = datetime.fromtimestamp(number, timezone)
        except (OSError, OverflowError, ValueError):
            return None
    else:
        text = str(value).strip()
        if re.fullmatch(r"\d{10,13}", text):
            return parse_datetime_value(int(text), timezone_name=timezone_name)
        parsed = None
        for candidate in (text.replace("Z", "+00:00"), text.replace("/", "-")):
            try:
                parsed = datetime.fromisoformat(candidate)
                break
            except ValueError:
                continue
        if parsed is None:
            return None
    return parsed.replace(tzinfo=timezone) if parsed.tzinfo is None else parsed.astimezone(timezone)


def _number(value: Any) -> float | None:
    if value in (None, "") or isinstance(value, bool):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _strong(candidate: Mapping[str, Any]) -> bool:
    scores = [_number(candidate.get(key)) for key in ("relevance_score", "elder_score", "share_score")]
    value = _number(candidate.get("value_score"))
    return value is not None and value >= 0.65 or all(score is not None and score >= floor for score, floor in zip(scores, (0.70, 0.70, 0.65), strict=True))


def _temporal(candidate: Mapping[str, Any], rules: Mapping[str, Any]) -> dict[str, Any]:
    timezone_name = str(rules["timezone"])
    current = parse_datetime_value(rules.get("current_datetime"), timezone_name=timezone_name) or datetime.now(ZoneInfo(timezone_name))
    published = parse_datetime_value(candidate.get("publish_at"), timezone_name=timezone_name)
    title = str(candidate.get("title") or "")
    temporal_type = str(candidate.get("temporal_type") or "evergreen")
    if published is None:
        compensated = _strong(candidate) and not any(word in title for word in ("今天", "今日", "明天", "刚刚", "突发"))
        return {"status": "pass" if compensated else "unknown", "reason_code": None if compensated else "TEMPORAL_UNKNOWN", "temporal_type": temporal_type, "evidence": {"publish_at": None, "compensated": compensated}}
    age_days = max(0.0, (current - published).total_seconds() / 86400)
    reason_code = None
    if any(word in title for word in ("今天", "今日", "明天", "昨日", "昨天")) and published.date() != current.date():
        reason_code = "RELATIVE_DATE_EXPIRED"
    elif temporal_type == "event" and age_days > int(rules["event_max_age_days"]):
        reason_code = "EVENT_EXPIRED"
    elif temporal_type == "seasonal" and age_days > int(rules["seasonal_max_age_days"]):
        reason_code = "SEASONAL_EXPIRED"
    return {"status": "fail" if reason_code else "pass", "reason_code": reason_code, "temporal_type": temporal_type, "evidence": {"publish_at": published.isoformat(timespec="seconds"), "content_age_days": round(age_days, 3)}}


def evaluate_candidate_gate(candidate: Mapping[str, Any], rule_snapshot: Mapping[str, Any] | str | None) -> dict[str, Any]:
    rules = _rules(rule_snapshot)
    temporal = _temporal(candidate, rules)
    duration = _number(candidate.get("duration_seconds"))
    shares = _number(candidate.get("share_count"))
    content_ratio = _number(candidate.get("content_50_plus_ratio"))
    account_ratio = _number(candidate.get("account_50_plus_ratio"))
    checks: list[dict[str, Any]] = [{"name": "temporal", **temporal}]

    def threshold(name: str, actual: float | None, minimum: float, missing: str, low: str, compensate: bool = False) -> None:
        if actual is None:
            status, reason = ("pass", None) if compensate else ("fail", missing)
        else:
            status, reason = ("pass", None) if actual >= minimum else ("fail", low)
        checks.append({"name": name, "status": status, "reason_code": reason, "actual": actual, "threshold": minimum, "compensated": actual is None and compensate})

    min_duration = float(rules["min_duration_seconds"])
    min_shares = float(rules["min_share_count"])
    threshold("duration_seconds", duration, min_duration, "DURATION_UNKNOWN", "DURATION_TOO_SHORT", compensate=_strong(candidate) or shares is not None and shares >= min_shares * 1.5)
    likes, plays = _number(candidate.get("like_count")), _number(candidate.get("play_count"))
    threshold("share_count", shares, min_shares, "SHARE_COUNT_UNKNOWN", "SHARE_COUNT_TOO_LOW", compensate=_strong(candidate) or bool(likes and likes >= 5000) or bool(plays and plays >= 50000))

    min_content = float(rules["min_content_50_plus_ratio"])
    min_account = float(rules["min_account_50_plus_ratio"])
    portrait_pass = bool(content_ratio is not None and content_ratio >= min_content or account_ratio is not None and account_ratio >= min_account)
    portrait_compensated = content_ratio is None and account_ratio is None and _strong(candidate)
    checks.append({
        "name": "elder_portrait", "status": "pass" if portrait_pass or portrait_compensated else "fail",
        "reason_code": None if portrait_pass or portrait_compensated else "CONTENT_PORTRAIT_MISSING" if content_ratio is None and account_ratio is None else "PORTRAIT_50_PLUS_TOO_LOW",
        "actual": {"content_50_plus_ratio": content_ratio, "account_50_plus_ratio": account_ratio},
        "threshold": {"min_content_50_plus_ratio": min_content, "min_account_50_plus_ratio": min_account},
        "compensated": portrait_compensated,
    })
    failed = [str(check["reason_code"]) for check in checks if check.get("status") != "pass" and check.get("reason_code")]
    content_status = "missing" if content_ratio is None else "pass" if content_ratio >= min_content else "fail"
    account_status = "missing" if account_ratio is None else "pass" if account_ratio >= min_account else "fail"
    return {
        "rule_version": str(rules["rule_version"]), "status": "pass" if not failed else "fail",
        "primary_eligible": not failed, "failed_reason_codes": failed, "checks": checks,
        "temporal": temporal, "content_portrait_status": content_status,
        "account_portrait_status": account_status,
        "portrait_conflict": content_status in {"pass", "fail"} and account_status in {"pass", "fail"} and content_status != account_status,
    }