from __future__ import annotations from typing import Any from sqlalchemy import select from supply_infra.db.models.business_harness import ( ContentDemandCoverage, ContentPerformanceFact, DailyDemandPackage, DailyDemandTask, DemandAttributionSnapshot, DemandEvaluationSnapshot, DemandScoreContribution, EvidencePackage, PlatformDemand, PlatformDemandCategoryRel, PlatformDemandVersion, RawDemandExpression, StandardDemandAlias, StrategyVersion, ) from supply_infra.db.session import get_session def get_published_demand_package( *, biz_dt: str | None = None, run_id: str | None = None, ) -> dict[str, Any] | None: with get_session() as session: stmt = select(DailyDemandPackage).where( DailyDemandPackage.status == "published" ) if biz_dt: stmt = stmt.where(DailyDemandPackage.biz_dt == biz_dt) if run_id: stmt = stmt.where(DailyDemandPackage.run_id == run_id) package = session.scalar( stmt.order_by( DailyDemandPackage.biz_dt.desc(), DailyDemandPackage.package_version.desc(), DailyDemandPackage.created_at.desc(), ).limit(1) ) if package is None: return None evidence = session.get(EvidencePackage, package.evidence_package_id) strategy = session.get(StrategyVersion, package.strategy_version_id) tasks = list( session.scalars( select(DailyDemandTask) .where( DailyDemandTask.demand_package_id == package.demand_package_id ) .order_by(DailyDemandTask.action_tier, DailyDemandTask.task_id) ).all() ) items: list[dict[str, Any]] = [] for task in tasks: evaluation = session.get( DemandEvaluationSnapshot, task.evaluation_id, ) contributions = list( session.scalars( select(DemandScoreContribution) .where( DemandScoreContribution.evaluation_id == task.evaluation_id ) .order_by( DemandScoreContribution.score_type, DemandScoreContribution.dimension, ) ).all() ) payload = dict(task.task_payload_json) payload["task_id"] = task.task_id payload["evaluation_id"] = task.evaluation_id payload["score_contributions"] = [ { "score_type": row.score_type, "dimension": row.dimension, "raw_value": row.raw_value_json, "normalized_value": float(row.normalized_value), "configured_weight": float(row.configured_weight), "effective_weight": float(row.effective_weight), "contribution": float(row.contribution), "data_status": row.data_status, "reason": row.reason, } for row in contributions ] if evaluation is not None: payload["decision_reason"] = evaluation.decision_reason payload["missing_dimensions"] = ( evaluation.missing_dimensions_json ) items.append(payload) return { "demand_package_id": package.demand_package_id, "run_id": package.run_id, "biz_dt": package.biz_dt, "package_version": package.package_version, "status": package.status, "content_hash": package.content_hash, "item_count": package.item_count, "published_at": ( package.published_at.isoformat() if package.published_at is not None else None ), "evidence_package": ( { "package_id": evidence.package_id, "status": evidence.status, "source_snapshot_hash": evidence.source_snapshot_hash, "source_versions": evidence.source_versions_json, "evidence_count": evidence.evidence_count, } if evidence is not None else None ), "strategy": ( { "strategy_version_id": strategy.strategy_version_id, "version_key": strategy.version_key, "definition": strategy.definition_json, "change_reason": strategy.change_reason, } if strategy is not None else None ), "items": items, } def get_platform_demand_history( platform_demand_id: str, ) -> dict[str, Any] | None: with get_session() as session: demand = session.get(PlatformDemand, platform_demand_id) if demand is None: return None versions = list( session.scalars( select(PlatformDemandVersion) .where( PlatformDemandVersion.platform_demand_id == platform_demand_id ) .order_by(PlatformDemandVersion.version_no.desc()) ).all() ) version_payloads: list[dict[str, Any]] = [] for row in versions: category_relations = list( session.scalars( select(PlatformDemandCategoryRel) .where( PlatformDemandCategoryRel.platform_demand_version_id == row.platform_demand_version_id ) .order_by( PlatformDemandCategoryRel.relation_type, PlatformDemandCategoryRel.category_id, ) ).all() ) evidence_rows = list( session.scalars( select(RawDemandExpression) .join( StandardDemandAlias, StandardDemandAlias.expression_id == RawDemandExpression.expression_id, ) .where( StandardDemandAlias.term_id == row.term_id, RawDemandExpression.package_id == row.package_id, ) .order_by( RawDemandExpression.source_type, RawDemandExpression.source_record_id, ) ).all() ) evaluation = session.scalar( select(DemandEvaluationSnapshot).where( DemandEvaluationSnapshot.platform_demand_version_id == row.platform_demand_version_id ) ) contributions = ( list( session.scalars( select(DemandScoreContribution) .where( DemandScoreContribution.evaluation_id == evaluation.evaluation_id ) .order_by( DemandScoreContribution.score_type, DemandScoreContribution.dimension, ) ).all() ) if evaluation is not None else [] ) coverages = list( session.scalars( select(ContentDemandCoverage) .where( ContentDemandCoverage.platform_demand_version_id == row.platform_demand_version_id ) .order_by( ContentDemandCoverage.content_id, ContentDemandCoverage.coverage_role, ) ).all() ) coverage_payloads: list[dict[str, Any]] = [] for coverage in coverages: attributions = list( session.scalars( select(DemandAttributionSnapshot) .where( DemandAttributionSnapshot.coverage_id == coverage.coverage_id ) .order_by(DemandAttributionSnapshot.created_at) ).all() ) attribution_payloads: list[dict[str, Any]] = [] for attribution in attributions: fact = session.get( ContentPerformanceFact, attribution.performance_fact_id, ) attribution_payloads.append( { "attribution_id": attribution.attribution_id, "support_state": attribution.support_state, "attributable_rov": ( float(attribution.attributable_rov) if attribution.attributable_rov is not None else None ), "attributable_vov": ( float(attribution.attributable_vov) if attribution.attributable_vov is not None else None ), "attribution_confidence": float( attribution.attribution_confidence ), "validity_influence": float( attribution.validity_influence ), "priority_influence": float( attribution.priority_influence ), "reason": attribution.reason, "performance_fact": ( { "performance_fact_id": ( fact.performance_fact_id ), "biz_dt": fact.biz_dt, "source": fact.source, "exposure_count": fact.exposure_count, "sample_size": fact.sample_size, "content_quality": ( float(fact.content_quality) if fact.content_quality is not None else None ), "rov": ( float(fact.rov) if fact.rov is not None else None ), "vov": ( float(fact.vov) if fact.vov is not None else None ), "observed_at": fact.observed_at.isoformat(), } if fact is not None else None ), } ) coverage_payloads.append( { "coverage_id": coverage.coverage_id, "task_id": coverage.task_id, "content_id": coverage.content_id, "content_url": coverage.content_url, "source_type": coverage.source_type, "coverage_role": coverage.coverage_role, "coverage_score": float(coverage.coverage_score), "evidence": coverage.evidence_json, "status": coverage.status, "content_published_at": ( coverage.content_published_at.isoformat() if coverage.content_published_at is not None else None ), "attributions": attribution_payloads, } ) version_payloads.append( { "platform_demand_version_id": ( row.platform_demand_version_id ), "version_no": row.version_no, "biz_dt": row.biz_dt, "run_id": row.run_id, "evidence_package_id": row.package_id, "name": row.name, "description": row.description, "cognition_confidence": float(row.cognition_confidence), "reason": row.reason, "change_type": row.change_type, "evidence_hash": row.evidence_hash, "raw_evidence": [ { "expression_id": evidence.expression_id, "source_type": evidence.source_type, "source_record_id": evidence.source_record_id, "raw_text": evidence.raw_text, "original_payload": ( evidence.original_payload_json ), "content_hash": evidence.content_hash, "data_quality": evidence.data_quality, } for evidence in evidence_rows ], "category_relations": [ { "category_id": int(relation.category_id), "relation_type": relation.relation_type, "relation_source": relation.relation_source, "reason": relation.reason, "confidence": float(relation.confidence), "is_inferred": bool(relation.is_inferred), "status": relation.status, } for relation in category_relations ], "evaluation": ( { "evaluation_id": evaluation.evaluation_id, "strategy_version_id": ( evaluation.strategy_version_id ), "validity_score": float( evaluation.validity_score ), "local_supply_priority": float( evaluation.local_supply_priority ), "data_confidence": float( evaluation.data_confidence ), "posterior_state": evaluation.posterior_state, "action_tier": evaluation.action_tier, "metrics": evaluation.metrics_json, "missing_dimensions": ( evaluation.missing_dimensions_json ), "decision_reason": evaluation.decision_reason, "score_contributions": [ { "score_type": contribution.score_type, "dimension": contribution.dimension, "raw_value": ( contribution.raw_value_json ), "normalized_value": float( contribution.normalized_value ), "configured_weight": float( contribution.configured_weight ), "effective_weight": float( contribution.effective_weight ), "contribution": float( contribution.contribution ), "data_status": contribution.data_status, "reason": contribution.reason, } for contribution in contributions ], } if evaluation is not None else None ), "content_coverage": coverage_payloads, } ) return { "platform_demand_id": demand.platform_demand_id, "name": demand.name, "description": demand.description, "status": demand.status, "lifecycle_state": demand.lifecycle_state, "versions": version_payloads, }