SupplyAgent/supply_infra/business_harness/content_feedback.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 select

from supply_infra.db.models.business_harness import (
    ContentDemandCoverage,
    ContentPerformanceFact,
    DailyDemandPackage,
    DailyDemandTask,
    DemandAttributionSnapshot,
)
from supply_infra.db.models.video_discovery import (
    VideoDiscoveryCandidate,
    VideoDiscoveryRun,
)
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")


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 _attribution_values(
    coverage: ContentDemandCoverage,
    fact: ContentPerformanceFact,
) -> dict[str, Any]:
    coverage_score = float(coverage.coverage_score)
    quality = float(fact.content_quality) if fact.content_quality is not None else 0.5
    sample_factor = _clamp(float(fact.sample_size or 0) / 1000.0)
    confidence = _clamp(coverage_score * quality * sample_factor)
    if fact.rov is None and fact.vov is None:
        return {
            "support_state": "inconclusive",
            "attributable_rov": None,
            "attributable_vov": None,
            "confidence": confidence,
            "validity_influence": 0.0,
            "priority_influence": 0.0,
            "reason": "内容承接关系已建立，但缺少 ROV/VOV，暂不影响次日评分",
        }

    attributable_rov = (
        float(fact.rov) * coverage_score if fact.rov is not None else None
    )
    attributable_vov = (
        float(fact.vov) * coverage_score if fact.vov is not None else None
    )
    signal = attributable_rov if attributable_rov is not None else 0.0
    support_state = (
        "support"
        if signal > 0.05
        else "oppose"
        if signal < -0.05
        else "inconclusive"
    )
    validity_influence = max(-0.1, min(0.1, signal * confidence * 0.1))
    priority_influence = max(
        -0.1,
        min(0.1, (attributable_vov or signal) * confidence * 0.1),
    )
    return {
        "support_state": support_state,
        "attributable_rov": attributable_rov,
        "attributable_vov": attributable_vov,
        "confidence": confidence,
        "validity_influence": validity_influence,
        "priority_influence": priority_influence,
        "reason": (
            f"覆盖度={coverage_score:.3f}、内容质量={quality:.3f}、"
            f"样本因子={sample_factor:.3f}，归因状态={support_state}"
        ),
    }


def _ensure_attribution(
    session,
    coverage: ContentDemandCoverage,
    fact: ContentPerformanceFact,
    strategy_version_id: str,
) -> DemandAttributionSnapshot:
    attribution_id = _uuid(
        f"demand-attribution:{coverage.coverage_id}:"
        f"{fact.performance_fact_id}:{strategy_version_id}"
    )
    existing = session.get(DemandAttributionSnapshot, attribution_id)
    if existing is not None:
        return existing
    values = _attribution_values(coverage, fact)
    row = DemandAttributionSnapshot(
        attribution_id=attribution_id,
        platform_demand_version_id=coverage.platform_demand_version_id,
        coverage_id=coverage.coverage_id,
        performance_fact_id=fact.performance_fact_id,
        strategy_version_id=strategy_version_id,
        support_state=values["support_state"],
        attributable_rov=(
            Decimal(str(values["attributable_rov"]))
            if values["attributable_rov"] is not None
            else None
        ),
        attributable_vov=(
            Decimal(str(values["attributable_vov"]))
            if values["attributable_vov"] is not None
            else None
        ),
        attribution_confidence=Decimal(str(round(values["confidence"], 5))),
        validity_influence=Decimal(
            str(round(values["validity_influence"], 7))
        ),
        priority_influence=Decimal(
            str(round(values["priority_influence"], 7))
        ),
        reason=values["reason"],
    )
    session.add(row)
    return row


def ingest_content_performance_fact(
    *,
    content_id: str,
    biz_dt: str,
    source: str,
    raw_payload: dict[str, Any],
    observed_at: datetime,
    exposure_count: int | None = None,
    sample_size: int | None = None,
    content_quality: float | None = None,
    rov: float | None = None,
    vov: float | None = None,
) -> dict[str, Any]:
    """Append one immutable observation and attribute it to known demand coverage."""
    if not content_id.strip() or not source.strip():
        raise ValueError("content_id and source are required")
    if len(biz_dt) != 8 or not biz_dt.isdigit():
        raise ValueError("biz_dt must be YYYYMMDD")
    if sample_size is not None and sample_size < 0:
        raise ValueError("sample_size must be non-negative")
    if exposure_count is not None and exposure_count < 0:
        raise ValueError("exposure_count must be non-negative")
    if content_quality is not None and not 0 <= content_quality <= 1:
        raise ValueError("content_quality must be between 0 and 1")

    canonical_payload = {
        "raw_payload": raw_payload,
        "exposure_count": exposure_count,
        "sample_size": sample_size,
        "content_quality": content_quality,
        "rov": rov,
        "vov": vov,
        "observed_at": observed_at.isoformat(),
    }
    payload_hash = _hash(canonical_payload)
    fact_id = _uuid(
        f"content-performance:{content_id}:{biz_dt}:{source}:{payload_hash}"
    )
    with get_session() as session:
        fact = session.get(ContentPerformanceFact, fact_id)
        idempotent_replay = fact is not None
        if fact is None:
            fact = ContentPerformanceFact(
                performance_fact_id=fact_id,
                content_id=content_id.strip(),
                biz_dt=biz_dt,
                source=source.strip()[:64],
                payload_hash=payload_hash,
                exposure_count=exposure_count,
                sample_size=sample_size,
                content_quality=(
                    Decimal(str(content_quality))
                    if content_quality is not None
                    else None
                ),
                rov=Decimal(str(rov)) if rov is not None else None,
                vov=Decimal(str(vov)) if vov is not None else None,
                raw_payload_json=raw_payload,
                observed_at=observed_at,
            )
            session.add(fact)
            session.flush()

        rows = session.execute(
            select(ContentDemandCoverage, DailyDemandPackage.strategy_version_id)
            .join(
                DailyDemandTask,
                DailyDemandTask.task_id == ContentDemandCoverage.task_id,
            )
            .join(
                DailyDemandPackage,
                DailyDemandPackage.demand_package_id
                == DailyDemandTask.demand_package_id,
            )
            .where(ContentDemandCoverage.content_id == content_id.strip())
        ).all()
        attribution_ids: list[str] = []
        for coverage, strategy_version_id in rows:
            attribution = _ensure_attribution(
                session,
                coverage,
                fact,
                str(strategy_version_id),
            )
            attribution_ids.append(attribution.attribution_id)
        session.flush()
        return {
            "success": True,
            "performance_fact_id": fact.performance_fact_id,
            "content_id": fact.content_id,
            "attribution_ids": attribution_ids,
            "attribution_count": len(attribution_ids),
            "idempotent_replay": idempotent_replay,
        }


def materialize_content_feedback(context: StepContext) -> dict[str, Any]:
    """Persist Find Agent coverage, raw performance facts and attribution."""
    with get_session() as session:
        package = session.scalar(
            select(DailyDemandPackage).where(
                DailyDemandPackage.run_id == context.run_id,
                DailyDemandPackage.status == "published",
            )
        )
        if package is None:
            return {
                "success": False,
                "error_code": "daily_demand_package_missing",
                "error": "找片结果无法关联到已发布任务包",
            }
        discovery_runs = list(
            session.scalars(
                select(VideoDiscoveryRun).where(
                    VideoDiscoveryRun.demand_package_id
                    == package.demand_package_id
                )
            ).all()
        )
        run_by_id = {str(row.run_id): row for row in discovery_runs}
        candidates = (
            list(
                session.scalars(
                    select(VideoDiscoveryCandidate).where(
                        VideoDiscoveryCandidate.run_id.in_(run_by_id),
                        VideoDiscoveryCandidate.decision_bucket == "primary",
                    )
                ).all()
            )
            if run_by_id
            else []
        )
        coverage_created = 0
        facts_created = 0
        attributions_created = 0
        for candidate in candidates:
            discovery_run = run_by_id[str(candidate.run_id)]
            if (
                not discovery_run.daily_demand_task_id
                or not discovery_run.platform_demand_version_id
            ):
                raise ValueError(
                    f"video discovery run {discovery_run.run_id} 缺少统一任务引用"
                )
            task = session.get(
                DailyDemandTask,
                discovery_run.daily_demand_task_id,
            )
            if task is None:
                raise ValueError(
                    f"daily demand task not found: {discovery_run.daily_demand_task_id}"
                )
            coverage_id = _uuid(
                "content-coverage:"
                f"{discovery_run.platform_demand_version_id}:"
                f"{candidate.aweme_id}:primary"
            )
            coverage = session.get(ContentDemandCoverage, coverage_id)
            if coverage is None:
                relevance = (
                    float(candidate.relevance_score)
                    if candidate.relevance_score is not None
                    else 0.5
                )
                coverage = ContentDemandCoverage(
                    coverage_id=coverage_id,
                    platform_demand_version_id=(
                        discovery_run.platform_demand_version_id
                    ),
                    task_id=task.task_id,
                    content_id=str(candidate.aweme_id),
                    content_url=candidate.content_link,
                    source_type="find_agent",
                    coverage_role="primary",
                    coverage_score=Decimal(str(round(_clamp(relevance), 5))),
                    evidence_json={
                        "decision_reason": candidate.decision_reason,
                        "source_keywords_json": candidate.source_keywords_json,
                        "gate_results_json": candidate.gate_results_json,
                        "video_discovery_run_id": discovery_run.run_id,
                    },
                    status="discovered",
                    discovered_at=candidate.create_time,
                    content_published_at=candidate.publish_at,
                )
                session.add(coverage)
                session.flush()
                coverage_created += 1

            raw_fact = {
                "play_count": candidate.play_count,
                "like_count": candidate.like_count,
                "comment_count": candidate.comment_count,
                "collect_count": candidate.collect_count,
                "share_count": candidate.share_count,
                "gate_status": candidate.gate_status,
                "value_score": (
                    float(candidate.value_score)
                    if candidate.value_score is not None
                    else None
                ),
            }
            payload_hash = _hash(raw_fact)
            fact_id = _uuid(
                f"content-performance:{candidate.aweme_id}:"
                f"{context.biz_dt}:find_agent:{payload_hash}"
            )
            fact = session.get(ContentPerformanceFact, fact_id)
            if fact is None:
                quality = 1.0 if candidate.gate_status == "pass" else 0.5
                fact = ContentPerformanceFact(
                    performance_fact_id=fact_id,
                    content_id=str(candidate.aweme_id),
                    biz_dt=context.biz_dt,
                    source="find_agent_snapshot",
                    payload_hash=payload_hash,
                    exposure_count=candidate.play_count,
                    sample_size=candidate.play_count,
                    content_quality=Decimal(str(quality)),
                    rov=None,
                    vov=None,
                    raw_payload_json=raw_fact,
                    observed_at=candidate.update_time or china_now(),
                )
                session.add(fact)
                session.flush()
                facts_created += 1
            before = session.get(
                DemandAttributionSnapshot,
                _uuid(
                    f"demand-attribution:{coverage.coverage_id}:"
                    f"{fact.performance_fact_id}:{package.strategy_version_id}"
                ),
            )
            _ensure_attribution(
                session,
                coverage,
                fact,
                package.strategy_version_id,
            )
            if before is None:
                attributions_created += 1

        session.flush()
        return {
            "success": True,
            "biz_dt": context.biz_dt,
            "demand_package_id": package.demand_package_id,
            "candidate_count": len(candidates),
            "coverage_created": coverage_created,
            "performance_facts_created": facts_created,
            "attributions_created": attributions_created,
        }