SupplyAgent/supply_infra/business_harness/evaluate.py

from __future__ import annotations

import hashlib
import json
import uuid
from datetime import datetime
from decimal import Decimal
from typing import Any

from sqlalchemy import func, select

from supply_infra.db.models.business_harness import (
    DailyDemandPackage,
    DailyDemandTask,
    DemandAttributionSnapshot,
    DemandEvaluationSnapshot,
    DemandScoreContribution,
    EvidencePackage,
    PlatformDemandCategoryRel,
    PlatformDemandVersion,
    RawDemandExpression,
    StandardDemandAlias,
    StrategyVersion,
)
from supply_infra.db.models.demand_grade import DemandGrade
from supply_infra.db.session import get_session
from supply_infra.pipeline.contracts import StepContext
from supply_infra.pipeline.dates import china_now

_NAMESPACE = uuid.UUID("53a14b2c-8ec3-5b88-a684-aa09737b6570")
_STRATEGY_KEY = "harness-v1"
_STRATEGY_DEFINITION: dict[str, Any] = {
    "validity_weights": {
        "legacy_grade_score": 0.35,
        "prior_heat": 0.20,
        "source_coverage": 0.15,
        "posterior_rov": 0.15,
        "attributed_feedback": 0.15,
    },
    "local_priority_weights": {
        "validity": 0.30,
        "prior_heat": 0.20,
        "grade_band": 0.15,
        "supply_gap": 0.20,
        "attributed_feedback": 0.15,
    },
    "action_thresholds": {
        "assure_supply": {"validity": 0.75, "priority": 0.55},
        "priority": {"validity": 0.60, "priority": 0.70},
        "targeted_validation": {"validity": 0.45, "confidence_below": 0.75},
        "exploration": {"validity": 0.25},
    },
    "posterior_min_samples": 1,
    "exploration_allocation": 1,
    "schema_version": 1,
}


def _uuid(key: str) -> str:
    return str(uuid.uuid5(_NAMESPACE, key))


def _hash(payload: Any) -> str:
    canonical = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


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


def _json_list(raw: str | None) -> list[Any]:
    if not raw:
        return []
    try:
        value = json.loads(raw)
    except (TypeError, json.JSONDecodeError):
        return []
    return value if isinstance(value, list) else []


def _weighted_score(
    dimensions: list[dict[str, Any]],
) -> tuple[Decimal, list[dict[str, Any]]]:
    available = [item for item in dimensions if item["normalized"] is not None]
    weight_sum = sum(float(item["weight"]) for item in available)
    if weight_sum <= 0:
        return Decimal("0.00000"), []
    contributions: list[dict[str, Any]] = []
    total = 0.0
    for item in available:
        effective_weight = float(item["weight"]) / weight_sum
        contribution = _clamp(float(item["normalized"])) * effective_weight
        total += contribution
        contributions.append(
            {
                **item,
                "effective_weight": effective_weight,
                "contribution": contribution,
            }
        )
    return Decimal(str(round(_clamp(total), 5))), contributions


def _grade_band(grade: str) -> float:
    return {
        "S": 1.0,
        "A": 0.85,
        "B": 0.65,
        "C": 0.40,
        "D": 0.20,
    }.get(str(grade).upper(), 0.30)


def _action_tier(
    definition: dict[str, Any],
    validity: float,
    priority: float,
    confidence: float,
    *,
    has_posterior: bool,
) -> str:
    thresholds = definition["action_thresholds"]
    assure = thresholds["assure_supply"]
    if (
        has_posterior
        and validity >= assure["validity"]
        and priority >= assure["priority"]
    ):
        return "assure_supply"
    priority_rule = thresholds["priority"]
    if validity >= priority_rule["validity"] and priority >= priority_rule["priority"]:
        return "priority"
    validate = thresholds["targeted_validation"]
    if (
        validity >= validate["validity"]
        and confidence < validate["confidence_below"]
    ):
        return "targeted_validation"
    if validity >= thresholds["exploration"]["validity"]:
        return "exploration"
    return "suppress"


def _ensure_strategy(session) -> StrategyVersion:
    strategy = session.scalar(
        select(StrategyVersion)
        .where(StrategyVersion.status == "active")
        .order_by(StrategyVersion.activated_at.desc())
        .limit(1)
    )
    if strategy is not None:
        return strategy
    legacy = session.scalar(
        select(StrategyVersion).where(StrategyVersion.version_key == _STRATEGY_KEY)
    )
    if legacy is not None:
        legacy.status = "active"
        legacy.activated_at = china_now()
        session.flush()
        return legacy
    strategy = StrategyVersion(
        strategy_version_id=_uuid(f"strategy-version:{_STRATEGY_KEY}"),
        version_key=_STRATEGY_KEY,
        status="active",
        definition_json=_STRATEGY_DEFINITION,
        change_reason="平台需求双评分与五类行动层初始显式策略",
        created_by="system",
        activated_at=china_now(),
    )
    session.add(strategy)
    session.flush()
    return strategy


def _raw_dimension(
    name: str,
    raw: Any,
    normalized: float | None,
    weight: float,
    reason: str,
) -> dict[str, Any]:
    return {
        "dimension": name,
        "raw": {"value": raw},
        "normalized": normalized,
        "weight": weight,
        "data_status": "available" if normalized is not None else "missing",
        "reason": reason,
    }


def _evaluation_inputs(
    session,
    version: PlatformDemandVersion,
    grade: DemandGrade,
    definition: dict[str, Any],
) -> tuple[
    dict[str, Any],
    list[dict[str, Any]],
    list[dict[str, Any]],
    list[str],
]:
    strategies = [str(value) for value in _json_list(grade.strategies)]
    videos = [str(value) for value in _json_list(grade.video_list)]
    posterior_available = bool(grade.has_posterior and grade.posterior_rov_count > 0)
    posterior_raw = (
        float(grade.posterior_rov_avg)
        if posterior_available and grade.posterior_rov_avg is not None
        else None
    )
    attributed = session.execute(
        select(
            func.avg(DemandAttributionSnapshot.validity_influence),
            func.avg(DemandAttributionSnapshot.priority_influence),
            func.count(),
        )
        .join(
            PlatformDemandVersion,
            PlatformDemandVersion.platform_demand_version_id
            == DemandAttributionSnapshot.platform_demand_version_id,
        )
        .where(
            PlatformDemandVersion.platform_demand_id
            == version.platform_demand_id,
            DemandAttributionSnapshot.support_state.in_(("support", "oppose")),
        )
    ).one()
    attributed_validity = (
        float(attributed[0]) if attributed[0] is not None else None
    )
    attributed_priority = (
        float(attributed[1]) if attributed[1] is not None else None
    )
    validity_weights = definition["validity_weights"]
    validity_dimensions = [
        _raw_dimension(
            "legacy_grade_score",
            float(grade.score) if grade.score is not None else None,
            _clamp(float(grade.score) / 100.0) if grade.score is not None else None,
            float(validity_weights["legacy_grade_score"]),
            "旧分级来源内排名归一到 0~1，仅作为过渡信号",
        ),
        _raw_dimension(
            "prior_heat",
            (
                float(grade.prior_total_score)
                if grade.prior_total_score is not None
                else None
            ),
            (
                _clamp(float(grade.prior_total_score) / 4.0)
                if grade.prior_total_score is not None
                else None
            ),
            float(validity_weights["prior_heat"]),
            "四类先验总分按理论范围 0~4 归一",
        ),
        _raw_dimension(
            "source_coverage",
            {"strategies": strategies, "count": len(strategies)},
            _clamp(len(strategies) / 4.0),
            float(validity_weights["source_coverage"]),
            "来源覆盖越多，需求成立的交叉支持越充分",
        ),
        _raw_dimension(
            "posterior_rov",
            {
                "rov": posterior_raw,
                "sample_count": int(grade.posterior_rov_count or 0),
            },
            _clamp(0.5 + posterior_raw / 2.0) if posterior_raw is not None else None,
            float(validity_weights["posterior_rov"]),
            "仅在存在有效样本时使用后验；缺失不按零处理",
        ),
        _raw_dimension(
            "attributed_feedback",
            {
                "validity_influence": attributed_validity,
                "sample_count": int(attributed[2] or 0),
            },
            (
                _clamp(0.5 + attributed_validity * 5.0)
                if attributed_validity is not None
                else None
            ),
            float(validity_weights["attributed_feedback"]),
            "只消费已关联内容且完成表现归因的支持/反对反馈",
        ),
    ]
    validity, _ = _weighted_score(validity_dimensions)
    local_weights = definition["local_priority_weights"]
    local_dimensions = [
        _raw_dimension(
            "validity",
            float(validity),
            float(validity),
            float(local_weights["validity"]),
            "局部供给首先受需求成立度约束",
        ),
        _raw_dimension(
            "prior_heat",
            (
                float(grade.prior_total_score)
                if grade.prior_total_score is not None
                else None
            ),
            (
                _clamp(float(grade.prior_total_score) / 4.0)
                if grade.prior_total_score is not None
                else None
            ),
            float(local_weights["prior_heat"]),
            "当前主题内先验热度",
        ),
        _raw_dimension(
            "grade_band",
            grade.grade,
            _grade_band(grade.grade),
            float(local_weights["grade_band"]),
            "旧等级仅作为过渡期的相对位置快照",
        ),
        _raw_dimension(
            "supply_gap",
            {"known_content_count": len(videos)},
            1.0 - _clamp(len(videos) / 10.0) if grade.video_list is not None else None,
            float(local_weights["supply_gap"]),
            "已有内容越少，局部补供价值越高；未知供给不按零处理",
        ),
        _raw_dimension(
            "attributed_feedback",
            {
                "priority_influence": attributed_priority,
                "sample_count": int(attributed[2] or 0),
            },
            (
                _clamp(0.5 + attributed_priority * 5.0)
                if attributed_priority is not None
                else None
            ),
            float(local_weights["attributed_feedback"]),
            "内容表现归因对次日局部供给优先级的影响",
        ),
    ]
    missing = [
        item["dimension"]
        for item in validity_dimensions + local_dimensions
        if item["normalized"] is None
    ]
    metrics = {
        "grade": grade.grade,
        "strategies": strategies,
        "videos": videos,
        "posterior_available": posterior_available,
        "posterior_rov_count": int(grade.posterior_rov_count or 0),
        "attributed_feedback_count": int(attributed[2] or 0),
    }
    return metrics, validity_dimensions, local_dimensions, list(dict.fromkeys(missing))


def _store_contributions(
    session,
    evaluation_id: str,
    score_type: str,
    contributions: list[dict[str, Any]],
) -> None:
    for item in contributions:
        session.add(
            DemandScoreContribution(
                contribution_id=_uuid(
                    f"score-contribution:{evaluation_id}:"
                    f"{score_type}:{item['dimension']}"
                ),
                evaluation_id=evaluation_id,
                score_type=score_type,
                dimension=item["dimension"],
                raw_value_json=item["raw"],
                normalized_value=Decimal(str(round(item["normalized"], 7))),
                configured_weight=Decimal(str(round(item["weight"], 7))),
                effective_weight=Decimal(
                    str(round(item["effective_weight"], 7))
                ),
                contribution=Decimal(str(round(item["contribution"], 7))),
                data_status=item["data_status"],
                reason=item["reason"],
            )
        )


def publish_daily_demand_package(context: StepContext) -> dict[str, Any]:
    with get_session() as session:
        existing = session.scalar(
            select(DailyDemandPackage).where(
                DailyDemandPackage.run_id == context.run_id
            )
        )
        if existing is not None:
            return {
                "success": existing.status == "published",
                "idempotent_replay": True,
                "demand_package_id": existing.demand_package_id,
                "biz_dt": existing.biz_dt,
                "item_count": existing.item_count,
                "status": existing.status,
                "content_hash": existing.content_hash,
            }

        evidence_package = session.scalar(
            select(EvidencePackage).where(EvidencePackage.run_id == context.run_id)
        )
        if evidence_package is None or evidence_package.status != "frozen":
            return {
                "success": False,
                "error_code": "evidence_package_not_frozen",
                "error": "平台需求证据包不存在或尚未冻结",
            }
        versions = list(
            session.scalars(
                select(PlatformDemandVersion)
                .where(PlatformDemandVersion.run_id == context.run_id)
                .order_by(PlatformDemandVersion.platform_demand_version_id)
            ).all()
        )
        if not versions:
            return {
                "success": False,
                "error_code": "platform_demand_versions_missing",
                "error": "没有可发布的平台需求版本",
            }

        strategy = _ensure_strategy(session)
        previous_package_version = int(
            session.scalar(
                select(func.max(DailyDemandPackage.package_version)).where(
                    DailyDemandPackage.biz_dt == context.biz_dt
                )
            )
            or 0
        )
        package_id = _uuid(f"daily-demand-package:{context.run_id}")
        package = DailyDemandPackage(
            demand_package_id=package_id,
            run_id=context.run_id,
            evidence_package_id=evidence_package.package_id,
            strategy_version_id=strategy.strategy_version_id,
            biz_dt=context.biz_dt,
            package_version=previous_package_version + 1,
            status="building",
            content_hash="0" * 64,
            item_count=0,
            published_at=None,
        )
        session.add(package)
        session.flush()

        task_payloads: list[dict[str, Any]] = []
        tier_counts: dict[str, int] = {}
        for version in versions:
            grade = session.get(DemandGrade, int(version.source_demand_grade_id))
            if grade is None:
                raise ValueError(
                    "平台需求版本引用的 DemandGrade 不存在: "
                    f"{version.source_demand_grade_id}"
                )
            definition = dict(strategy.definition_json)
            metrics, validity_dims, local_dims, missing = _evaluation_inputs(
                session,
                version,
                grade,
                definition,
            )
            validity, validity_contributions = _weighted_score(validity_dims)
            local_priority, local_contributions = _weighted_score(local_dims)
            configured_validity_weight = sum(float(item["weight"]) for item in validity_dims)
            available_validity_weight = sum(
                float(item["weight"])
                for item in validity_dims
                if item["normalized"] is not None
            )
            data_confidence = Decimal(
                str(
                    round(
                        _clamp(
                            available_validity_weight
                            / configured_validity_weight
                        ),
                        5,
                    )
                )
            )
            has_posterior = bool(metrics["posterior_available"])
            action_tier = _action_tier(
                definition,
                float(validity),
                float(local_priority),
                float(data_confidence),
                has_posterior=has_posterior,
            )
            posterior_state = "available" if has_posterior else "missing"
            evaluation_id = _uuid(
                f"demand-evaluation:{version.platform_demand_version_id}"
            )
            reason = (
                f"成立度={validity}，局部供给优先级={local_priority}，"
                f"数据置信度={data_confidence}，行动层={action_tier}；"
                f"缺失维度={missing or ['无']}。"
            )
            evaluation = DemandEvaluationSnapshot(
                evaluation_id=evaluation_id,
                platform_demand_version_id=version.platform_demand_version_id,
                run_id=context.run_id,
                strategy_version_id=strategy.strategy_version_id,
                biz_dt=context.biz_dt,
                validity_score=validity,
                local_supply_priority=local_priority,
                data_confidence=data_confidence,
                posterior_state=posterior_state,
                action_tier=action_tier,
                metrics_json=metrics,
                missing_dimensions_json=missing,
                decision_reason=reason,
            )
            session.add(evaluation)
            _store_contributions(
                session,
                evaluation_id,
                "validity",
                validity_contributions,
            )
            _store_contributions(
                session,
                evaluation_id,
                "local_supply_priority",
                local_contributions,
            )

            relation_rows = list(
                session.scalars(
                    select(PlatformDemandCategoryRel).where(
                        PlatformDemandCategoryRel.platform_demand_version_id
                        == version.platform_demand_version_id
                    )
                ).all()
            )
            aliases = [
                str(value)
                for value in session.scalars(
                    select(RawDemandExpression.raw_text)
                    .join(
                        StandardDemandAlias,
                        StandardDemandAlias.expression_id
                        == RawDemandExpression.expression_id,
                    )
                    .where(StandardDemandAlias.term_id == version.term_id)
                ).all()
            ]
            search_terms = list(dict.fromkeys([version.name, *aliases]))
            existing_content = [str(value) for value in metrics["videos"]]
            hypotheses = (
                ["需要通过新内容验证该需求的真实 ROV/VOV"]
                if not has_posterior
                else ["验证后验表现是否能在新增内容上持续"]
            )
            task_payload = {
                "platform_demand_id": version.platform_demand_id,
                "platform_demand_version_id": version.platform_demand_version_id,
                "version_no": version.version_no,
                "biz_dt": context.biz_dt,
                "run_id": context.run_id,
                "strategy_version": strategy.version_key,
                "name": version.name,
                "description": version.description,
                "reason": version.reason,
                "category_relations": [
                    {
                        "category_id": int(row.category_id),
                        "relation_type": row.relation_type,
                        "relation_source": row.relation_source,
                        "reason": row.reason,
                        "confidence": float(row.confidence),
                        "is_inferred": bool(row.is_inferred),
                    }
                    for row in relation_rows
                ],
                "validity_score": float(validity),
                "local_supply_priority": float(local_priority),
                "data_confidence": float(data_confidence),
                "posterior_state": posterior_state,
                "action_tier": action_tier,
                "search_terms": search_terms,
                "exclude_terms": [],
                "hit_rules": {
                    "semantic_match_required": True,
                    "must_reference_platform_demand_version": True,
                },
                "hypotheses": hypotheses,
                "evidence_gaps": missing,
                "existing_content": existing_content,
                "allocation": {
                    "mode": action_tier,
                    "exploration_quota": (
                        definition["exploration_allocation"]
                        if action_tier == "exploration"
                        else 0
                    ),
                },
                "decision_reason": reason,
            }
            session.add(
                DailyDemandTask(
                    task_id=_uuid(
                        f"daily-demand-task:{package_id}:"
                        f"{version.platform_demand_version_id}"
                    ),
                    demand_package_id=package_id,
                    platform_demand_version_id=version.platform_demand_version_id,
                    evaluation_id=evaluation_id,
                    action_tier=action_tier,
                    search_terms_json=search_terms,
                    exclude_terms_json=[],
                    hit_rules_json=task_payload["hit_rules"],
                    hypotheses_json=hypotheses,
                    evidence_gaps_json=missing,
                    existing_content_json=existing_content,
                    allocation_json=task_payload["allocation"],
                    task_payload_json=task_payload,
                )
            )
            task_payloads.append(task_payload)
            tier_counts[action_tier] = tier_counts.get(action_tier, 0) + 1

        package.item_count = len(task_payloads)
        package.content_hash = _hash(task_payloads)
        package.status = "published"
        package.published_at = china_now()
        session.flush()
        return {
            "success": True,
            "idempotent_replay": False,
            "demand_package_id": package_id,
            "biz_dt": context.biz_dt,
            "package_version": package.package_version,
            "strategy_version": strategy.version_key,
            "status": package.status,
            "item_count": package.item_count,
            "tier_counts": tier_counts,
            "content_hash": package.content_hash,
            "published_at": (
                package.published_at.isoformat()
                if isinstance(package.published_at, datetime)
                else None
            ),
        }