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 evaluate_candidate_gate(candidate: Mapping[str, Any], rule_snapshot: Mapping[str, Any] | str | None) -> dict[str, Any]:
    rules = _rules(rule_snapshot)
    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]] = []

    def threshold(name: str, actual: float | None, minimum: float, missing: str, low: str) -> None:
        if actual is None:
            status, reason = "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": False})

    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")
    threshold("share_count", shares, min_shares, "SHARE_COUNT_UNKNOWN", "SHARE_COUNT_TOO_LOW")

    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 = False
    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,
        "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,
    }