SupplyAgent/supply_infra/video_discovery_gates.py

"""find_agent P0 候选硬门槛与基础时效规则。"""
from __future__ import annotations

import json
import re
from dataclasses import asdict, dataclass
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from typing import Any, Mapping
from zoneinfo import ZoneInfo

from supply_infra.config import get_infra_settings


@dataclass(frozen=True)
class FindAgentGateRules:
    rule_version: str
    timezone: str
    min_duration_seconds: int
    min_share_count: int
    min_content_50_plus_ratio: float
    min_account_50_plus_ratio: float
    festival_lead_days: int
    event_max_age_days: int
    seasonal_max_age_days: int


_DAYPART_RULES: tuple[tuple[str, tuple[str, ...], time, time], ...] = (
    ("morning", ("早上好", "早安", "晨安", "早晨好"), time(5, 0), time(10, 30)),
    ("noon", ("午安", "中午好", "午间好"), time(11, 0), time(14, 0)),
    ("evening", ("晚上好", "晚安", "夜安"), time(18, 0), time(1, 0)),
)

_FIXED_FESTIVALS: dict[str, tuple[int, int]] = {
    "元旦": (1, 1),
    "情人节": (2, 14),
    "妇女节": (3, 8),
    "劳动节": (5, 1),
    "儿童节": (6, 1),
    "建党节": (7, 1),
    "建军节": (8, 1),
    "教师节": (9, 10),
    "国庆": (10, 1),
    "国庆节": (10, 1),
    "圣诞": (12, 25),
    "圣诞节": (12, 25),
}

# 业务运行期内使用的农历节日阳历日期。超出覆盖年份时返回 unknown，不猜测日期。
_LUNAR_FESTIVALS: dict[int, dict[str, tuple[int, int]]] = {
    2025: {
        "春节": (1, 29),
        "元宵": (2, 12),
        "元宵节": (2, 12),
        "端午": (5, 31),
        "端午节": (5, 31),
        "七夕": (8, 29),
        "中秋": (10, 6),
        "中秋节": (10, 6),
        "重阳": (10, 29),
        "重阳节": (10, 29),
    },
    2026: {
        "春节": (2, 17),
        "元宵": (3, 3),
        "元宵节": (3, 3),
        "端午": (6, 19),
        "端午节": (6, 19),
        "七夕": (8, 19),
        "中秋": (9, 25),
        "中秋节": (9, 25),
        "重阳": (10, 18),
        "重阳节": (10, 18),
    },
    2027: {
        "春节": (2, 6),
        "元宵": (2, 20),
        "元宵节": (2, 20),
        "端午": (6, 9),
        "端午节": (6, 9),
        "七夕": (8, 8),
        "中秋": (9, 15),
        "中秋节": (9, 15),
        "重阳": (10, 8),
        "重阳节": (10, 8),
    },
}

_EVENT_MARKERS = (
    "最新消息",
    "突发",
    "刚刚",
    "今日新闻",
    "赛事结果",
    "比赛结果",
)
_RELATIVE_DATE_MARKERS = ("今天", "今日", "明天", "昨日", "昨天")


def get_gate_rules() -> FindAgentGateRules:
    settings = get_infra_settings()
    return FindAgentGateRules(
        rule_version=settings.find_agent_rule_version,
        timezone=settings.scheduler_timezone,
        min_duration_seconds=settings.find_agent_min_duration_seconds,
        min_share_count=settings.find_agent_min_share_count,
        min_content_50_plus_ratio=settings.find_agent_min_content_50_plus_ratio,
        min_account_50_plus_ratio=settings.find_agent_min_account_50_plus_ratio,
        festival_lead_days=settings.find_agent_festival_lead_days,
        event_max_age_days=settings.find_agent_event_max_age_days,
        seasonal_max_age_days=settings.find_agent_seasonal_max_age_days,
    )


def build_rule_snapshot(now: datetime | None = None) -> dict[str, Any]:
    rules = get_gate_rules()
    timezone = ZoneInfo(rules.timezone)
    evaluated_at = now or datetime.now(timezone)
    if evaluated_at.tzinfo is None:
        evaluated_at = evaluated_at.replace(tzinfo=timezone)
    else:
        evaluated_at = evaluated_at.astimezone(timezone)
    return {
        **asdict(rules),
        "current_datetime": evaluated_at.isoformat(timespec="seconds"),
        "current_date": evaluated_at.strftime("%Y-%m-%d"),
    }


def load_rule_snapshot(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 {})
    fallback = build_rule_snapshot()
    return {**fallback, **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)):
        timestamp = float(value)
        if timestamp > 10_000_000_000:
            timestamp /= 1000
        try:
            parsed = datetime.fromtimestamp(timestamp, timezone)
        except (OSError, OverflowError, ValueError):
            return None
    else:
        text = str(value).strip()
        if not text:
            return None
        if re.fullmatch(r"\d{10,13}", text):
            return parse_datetime_value(int(text), timezone_name=timezone_name)
        normalized = text.replace("Z", "+00:00")
        parsed = None
        for candidate in (
            normalized,
            normalized.replace("/", "-"),
        ):
            try:
                parsed = datetime.fromisoformat(candidate)
                break
            except ValueError:
                continue
        if parsed is None:
            for pattern in ("%Y%m%d", "%Y-%m-%d", "%Y/%m/%d"):
                try:
                    parsed = datetime.strptime(text, pattern)
                    break
                except ValueError:
                    continue
        if parsed is None:
            return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone)
    return parsed.astimezone(timezone)


def normalize_duration_seconds(value: Any, *, unit: str = "seconds") -> Decimal | None:
    if value in (None, "") or isinstance(value, bool):
        return None
    try:
        number = Decimal(str(value))
    except (ArithmeticError, ValueError):
        return None
    if unit == "milliseconds":
        number /= Decimal("1000")
    if number < 0:
        return None
    return number.quantize(Decimal("0.001"))


def _candidate_text(candidate: Mapping[str, Any]) -> str:
    fragments = [str(candidate.get("title") or "")]
    tags = candidate.get("tags")
    if tags is None:
        tags = candidate.get("tags_json")
    if isinstance(tags, str):
        try:
            loaded = json.loads(tags)
        except (TypeError, ValueError):
            loaded = tags
        tags = loaded
    if isinstance(tags, list):
        fragments.extend(str(item) for item in tags)
    elif tags:
        fragments.append(str(tags))
    evidence = candidate.get("temporal_evidence")
    if evidence is None:
        evidence = candidate.get("temporal_evidence_json")
    if evidence:
        fragments.append(str(evidence))
    return " ".join(fragments)


def _time_in_window(current: time, start: time, end: time) -> bool:
    if start <= end:
        return start <= current <= end
    return current >= start or current <= end


def _festival_dates(year: int) -> dict[str, date]:
    result = {
        name: date(year, month, day)
        for name, (month, day) in _FIXED_FESTIVALS.items()
    }
    for name, (month, day) in _LUNAR_FESTIVALS.get(year, {}).items():
        result[name] = date(year, month, day)
    return result


def _nearest_festival_date(name: str, current: date) -> date | None:
    candidates = [
        festival_date
        for year in (current.year - 1, current.year, current.year + 1)
        if (festival_date := _festival_dates(year).get(name)) is not None
    ]
    if not candidates:
        return None
    return min(candidates, key=lambda value: abs((value - current).days))


def evaluate_temporal_status(
    candidate: Mapping[str, Any],
    rule_snapshot: Mapping[str, Any] | str | None,
) -> dict[str, Any]:
    rules = load_rule_snapshot(rule_snapshot)
    timezone_name = str(rules["timezone"])
    timezone = ZoneInfo(timezone_name)
    current = parse_datetime_value(
        rules.get("current_datetime"),
        timezone_name=timezone_name,
    ) or datetime.now(timezone)
    publish_at = parse_datetime_value(
        candidate.get("publish_at"),
        timezone_name=timezone_name,
    )
    text = _candidate_text(candidate)
    inferred_type = str(candidate.get("temporal_type") or "").strip() or "evergreen"
    evidence: dict[str, Any] = {
        "evaluated_at": current.isoformat(timespec="seconds"),
        "publish_at": publish_at.isoformat(timespec="seconds") if publish_at else None,
        "matched_terms": [],
    }

    if publish_at is None:
        if _can_compensate_temporal_unknown(candidate, text=text):
            return {
                "temporal_type": inferred_type or "evergreen",
                "status": "pass",
                "reason_code": None,
                "reason": "缺少发布时间，但无时效敏感表述且其他指标支持通过",
                "evidence": {**evidence, "compensated": True},
            }
        return {
            "temporal_type": inferred_type,
            "status": "unknown",
            "reason_code": "TEMPORAL_UNKNOWN",
            "reason": "缺少真实发布时间，无法完成时效校验",
            "evidence": evidence,
        }

    for daypart, terms, start, end in _DAYPART_RULES:
        matched = [term for term in terms if term in text]
        if not matched:
            continue
        inferred_type = "daypart"
        evidence["matched_terms"].extend(matched)
        evidence["valid_time"] = (
            f"{start.strftime('%H:%M')}-{end.strftime('%H:%M')}"
        )
        if not _time_in_window(current.timetz().replace(tzinfo=None), start, end):
            return {
                "temporal_type": inferred_type,
                "status": "fail",
                "reason_code": "DAYPART_EXPIRED",
                "reason": f"当前时间不在{daypart}内容有效时段",
                "evidence": evidence,
            }

    matched_festivals: list[tuple[str, date]] = []
    known_names = set(_FIXED_FESTIVALS)
    for yearly in _LUNAR_FESTIVALS.values():
        known_names.update(yearly)
    for name in sorted(known_names, key=len, reverse=True):
        if name not in text:
            continue
        festival_date = _nearest_festival_date(name, current.date())
        if festival_date is None:
            return {
                "temporal_type": "festival",
                "status": "unknown",
                "reason_code": "TEMPORAL_UNKNOWN",
                "reason": f"规则版本未覆盖当前相邻年份的{name}日期",
                "evidence": {**evidence, "matched_terms": [name]},
            }
        matched_festivals.append((name, festival_date))
        break
    if matched_festivals:
        inferred_type = "festival"
        name, festival_date = matched_festivals[0]
        lead_days = int(rules["festival_lead_days"])
        valid_from = festival_date - timedelta(days=lead_days)
        valid_to = festival_date + timedelta(days=1)
        evidence.update(
            {
                "matched_terms": [name],
                "valid_from": valid_from.isoformat(),
                "valid_to": valid_to.isoformat(),
            }
        )
        if not valid_from <= current.date() <= valid_to:
            return {
                "temporal_type": inferred_type,
                "status": "fail",
                "reason_code": "FESTIVAL_OUT_OF_WINDOW",
                "reason": f"{name}内容不在当前有效窗口",
                "evidence": evidence,
            }

    matched_relative = [term for term in _RELATIVE_DATE_MARKERS if term in text]
    if matched_relative and publish_at.date() != current.date():
        evidence["matched_terms"].extend(matched_relative)
        return {
            "temporal_type": "event",
            "status": "fail",
            "reason_code": "RELATIVE_DATE_EXPIRED",
            "reason": "内容包含相对日期表述，但并非当天发布",
            "evidence": evidence,
        }

    if any(marker in text for marker in _EVENT_MARKERS):
        inferred_type = "event"
    age_days = max(0.0, (current - publish_at).total_seconds() / 86400)
    evidence["content_age_days"] = round(age_days, 3)
    if inferred_type == "event" and age_days > int(rules["event_max_age_days"]):
        return {
            "temporal_type": inferred_type,
            "status": "fail",
            "reason_code": "EVENT_EXPIRED",
            "reason": "事件型内容超过允许的新鲜度窗口",
            "evidence": evidence,
        }
    if (
        inferred_type == "seasonal"
        and age_days > int(rules["seasonal_max_age_days"])
    ):
        return {
            "temporal_type": inferred_type,
            "status": "fail",
            "reason_code": "SEASONAL_EXPIRED",
            "reason": "季节型内容超过允许的新鲜度窗口",
            "evidence": evidence,
        }

    explicit_status = str(candidate.get("temporal_status") or "").strip()
    if explicit_status in {"fail", "unknown"}:
        reason_code = (
            "TEMPORAL_EXPIRED" if explicit_status == "fail" else "TEMPORAL_UNKNOWN"
        )
        return {
            "temporal_type": inferred_type,
            "status": explicit_status,
            "reason_code": reason_code,
            "reason": "候选补证结果未通过时间有效性判断",
            "evidence": evidence,
        }
    return {
        "temporal_type": inferred_type,
        "status": "pass",
        "reason_code": None,
        "reason": "发布时间与基础时间语义有效",
        "evidence": evidence,
    }


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 _semantic_score(candidate: Mapping[str, Any], field: str) -> float | None:
    return _number(candidate.get(field))


def _has_strong_semantic_scores(candidate: Mapping[str, Any]) -> bool:
    """R/E/S 综合优秀，或 V 足够高，可用于缺失字段补偿。"""
    relevance = _semantic_score(candidate, "relevance_score")
    elder = _semantic_score(candidate, "elder_score")
    share = _semantic_score(candidate, "share_score")
    value = _semantic_score(candidate, "value_score")
    if (
        relevance is not None
        and elder is not None
        and share is not None
        and relevance >= 0.70
        and elder >= 0.70
        and share >= 0.65
    ):
        return True
    return value is not None and value >= 0.65


def _text_has_temporal_markers(text: str) -> bool:
    for _, terms, _, _ in _DAYPART_RULES:
        if any(term in text for term in terms):
            return True
    known_names = set(_FIXED_FESTIVALS)
    for yearly in _LUNAR_FESTIVALS.values():
        known_names.update(yearly)
    if any(name in text for name in known_names):
        return True
    if any(marker in text for marker in _EVENT_MARKERS + _RELATIVE_DATE_MARKERS):
        return True
    return False


def _can_compensate_missing_portrait(
    candidate: Mapping[str, Any],
    account_ratio: float | None,
    *,
    min_account_ratio: float,
) -> bool:
    if account_ratio is not None and account_ratio >= max(min_account_ratio, 0.35):
        return True
    if account_ratio is not None and account_ratio >= min_account_ratio:
        return _has_strong_semantic_scores(candidate)
    elder = _semantic_score(candidate, "elder_score")
    return elder is not None and elder >= 0.75 and _has_strong_semantic_scores(candidate)


def _can_compensate_missing_duration(
    candidate: Mapping[str, Any],
    shares: float | None,
    *,
    min_share_count: float,
) -> bool:
    if shares is not None and shares >= min_share_count * 1.5:
        return True
    relevance = _semantic_score(candidate, "relevance_score")
    return (
        relevance is not None
        and relevance >= 0.80
        and _has_strong_semantic_scores(candidate)
    )


def _can_compensate_missing_shares(candidate: Mapping[str, Any]) -> bool:
    likes = _number(candidate.get("like_count"))
    plays = _number(candidate.get("play_count"))
    if likes is not None and likes >= 5000:
        return True
    if plays is not None and plays >= 50000:
        return True
    share = _semantic_score(candidate, "share_score")
    return share is not None and share >= 0.75 and _has_strong_semantic_scores(candidate)


def _can_compensate_temporal_unknown(
    candidate: Mapping[str, Any],
    *,
    text: str,
) -> bool:
    if _text_has_temporal_markers(text):
        return False
    if _has_strong_semantic_scores(candidate):
        return True
    shares = _number(candidate.get("share_count"))
    return shares is not None and shares >= 2000


def evaluate_candidate_gate(
    candidate: Mapping[str, Any],
    rule_snapshot: Mapping[str, Any] | str | None,
) -> dict[str, Any]:
    rules = load_rule_snapshot(rule_snapshot)
    temporal = evaluate_temporal_status(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",
            "status": temporal["status"],
            "reason_code": temporal["reason_code"],
            "actual": temporal["evidence"],
        }
    ]

    min_duration = float(rules["min_duration_seconds"])
    min_shares = float(rules["min_share_count"])
    min_content_ratio = float(rules["min_content_50_plus_ratio"])
    min_account_ratio = float(rules["min_account_50_plus_ratio"])

    def threshold_check(
        name: str,
        actual: float | None,
        threshold: float,
        *,
        missing_code: str,
        low_code: str,
        compensate_missing: bool = False,
    ) -> None:
        compensated = False
        if actual is None:
            if compensate_missing:
                status = "pass"
                reason_code = None
                compensated = True
            else:
                status = "fail"
                reason_code = missing_code
        elif actual < threshold:
            status = "fail"
            reason_code = low_code
        else:
            status = "pass"
            reason_code = None
        checks.append(
            {
                "name": name,
                "status": status,
                "reason_code": reason_code,
                "actual": actual,
                "threshold": threshold,
                "compensated": compensated,
            }
        )

    threshold_check(
        "duration_seconds",
        duration,
        min_duration,
        missing_code="DURATION_UNKNOWN",
        low_code="DURATION_TOO_SHORT",
        compensate_missing=_can_compensate_missing_duration(
            candidate,
            shares,
            min_share_count=min_shares,
        ),
    )
    threshold_check(
        "share_count",
        shares,
        min_shares,
        missing_code="SHARE_COUNT_UNKNOWN",
        low_code="SHARE_COUNT_TOO_LOW",
        compensate_missing=_can_compensate_missing_shares(candidate),
    )

    content_pass = (
        content_ratio is not None and content_ratio >= min_content_ratio
    )
    account_pass = (
        account_ratio is not None and account_ratio >= min_account_ratio
    )
    portrait_compensated = False
    if content_pass or account_pass:
        portrait_status = "pass"
        portrait_reason_code = None
    elif content_ratio is None and account_ratio is None:
        portrait_compensated = _can_compensate_missing_portrait(
            candidate,
            account_ratio,
            min_account_ratio=min_account_ratio,
        )
        if portrait_compensated:
            portrait_status = "pass"
            portrait_reason_code = None
        else:
            portrait_status = "fail"
            portrait_reason_code = "CONTENT_PORTRAIT_MISSING"
    else:
        portrait_status = "fail"
        if content_ratio is not None and account_ratio is not None:
            portrait_reason_code = "PORTRAIT_50_PLUS_TOO_LOW"
        elif content_ratio is None:
            portrait_reason_code = "CONTENT_PORTRAIT_MISSING"
        else:
            portrait_reason_code = "ACCOUNT_50_PLUS_TOO_LOW"

    checks.append(
        {
            "name": "elder_portrait",
            "status": portrait_status,
            "reason_code": portrait_reason_code,
            "actual": {
                "content_50_plus_ratio": content_ratio,
                "account_50_plus_ratio": account_ratio,
            },
            "threshold": {
                "min_content_50_plus_ratio": min_content_ratio,
                "min_account_50_plus_ratio": min_account_ratio,
            },
            "compensated": portrait_compensated,
        }
    )

    failed_codes = [
        str(check["reason_code"])
        for check in checks
        if check["status"] != "pass" and check.get("reason_code")
    ]
    content_status = (
        "missing"
        if content_ratio is None
        else (
            "pass"
            if content_ratio >= float(rules["min_content_50_plus_ratio"])
            else "fail"
        )
    )
    account_status = (
        "missing"
        if account_ratio is None
        else (
            "pass"
            if account_ratio >= float(rules["min_account_50_plus_ratio"])
            else "fail"
        )
    )
    portrait_conflict = (
        content_status in {"pass", "fail"}
        and account_status in {"pass", "fail"}
        and content_status != account_status
    )
    return {
        "rule_version": str(rules["rule_version"]),
        "status": "pass" if not failed_codes else "fail",
        "primary_eligible": not failed_codes,
        "failed_reason_codes": failed_codes,
        "checks": checks,
        "temporal": temporal,
        "content_portrait_status": content_status,
        "account_portrait_status": account_status,
        "portrait_conflict": portrait_conflict,
    }