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, }