SupplyAgent/supply_infra/business_harness/materialize.py

from __future__ import annotations

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

from sqlalchemy import func, select

from supply_infra.db.models.business_harness import (
    EvidencePackage,
    PlatformDemand,
    PlatformDemandCategoryRel,
    PlatformDemandVersion,
    RawDemandExpression,
    StandardDemandAlias,
    StandardDemandTerm,
)
from supply_infra.db.models.demand_grade import DemandGrade
from supply_infra.db.models.demand_grade_category_rel import DemandGradeCategoryRel
from supply_infra.db.models.demand_feedback import DemandFeedback
from supply_infra.db.repositories.multi_demand_pool_di_repo import (
    MultiDemandPoolDiRepository,
)
from supply_infra.db.session import get_session
from supply_infra.pipeline.contracts import StepContext

_NAMESPACE = uuid.UUID("53a14b2c-8ec3-5b88-a684-aa09737b6570")


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


def _normalized_text(value: str) -> str:
    return " ".join(unicodedata.normalize("NFKC", str(value)).strip().casefold().split())


def _hash_payload(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 _parse_int_list(raw: str | None) -> list[int]:
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
    except (TypeError, json.JSONDecodeError):
        return []
    if not isinstance(parsed, list):
        return []
    result: list[int] = []
    for value in parsed:
        try:
            result.append(int(value))
        except (TypeError, ValueError):
            continue
    return list(dict.fromkeys(result))


def _score_confidence(score: Decimal | float | None) -> Decimal:
    if score is None:
        return Decimal("0.50000")
    normalized = max(0.0, min(1.0, float(score) / 100.0))
    return Decimal(str(round(normalized, 5)))


def _pool_payload(row: Any) -> dict[str, Any]:
    return {
        "id": int(row.id),
        "strategy": row.strategy,
        "demand_id": row.demand_id,
        "demand_name": row.demand_name,
        "weight": float(row.weight) if row.weight is not None else None,
        "type": row.type,
        "video_count": int(row.video_count or 0),
        "video_list": row.video_list,
        "extend": row.extend,
        "real_rov_7d": (
            float(row.real_rov_7d) if row.real_rov_7d is not None else None
        ),
        "real_vov_7d": (
            float(row.real_vov_7d) if row.real_vov_7d is not None else None
        ),
        "biz_dt": row.biz_dt,
    }


def _existing_result(session, package: EvidencePackage) -> dict[str, Any]:
    version_count = int(
        session.scalar(
            select(func.count())
            .select_from(PlatformDemandVersion)
            .where(PlatformDemandVersion.run_id == package.run_id)
        )
        or 0
    )
    return {
        "success": True,
        "idempotent_replay": True,
        "package_id": package.package_id,
        "biz_dt": package.biz_dt,
        "evidence_count": package.evidence_count,
        "platform_demand_versions": version_count,
        "source_snapshot_hash": package.source_snapshot_hash,
    }


def materialize_platform_demands(context: StepContext) -> dict[str, Any]:
    """
    Freeze raw expressions and map today's DemandGrade rows into stable,
    append-only platform-demand versions.

    This is an intentionally conservative MVP: one current representative
    demand name maps to one stable canonical object. Future merge/split review
    can re-point terms without changing raw evidence.
    """
    package_id = _uuid(f"evidence-package:{context.run_id}")
    with get_session() as session:
        existing_package = session.scalar(
            select(EvidencePackage).where(EvidencePackage.run_id == context.run_id)
        )
        if existing_package is not None:
            return _existing_result(session, existing_package)

        grades = list(
            session.scalars(
                select(DemandGrade)
                .where(DemandGrade.biz_dt == context.biz_dt)
                .order_by(DemandGrade.id)
            ).all()
        )
        if not grades:
            return {
                "success": False,
                "error_code": "platform_demand_no_grades",
                "error": f"biz_dt={context.biz_dt} 没有可物化的 DemandGrade",
            }

        grade_ids = [int(grade.id) for grade in grades]
        category_map: dict[int, list[int]] = {}
        category_stmt = select(
            DemandGradeCategoryRel.demand_grade_id,
            DemandGradeCategoryRel.category_id,
        ).where(DemandGradeCategoryRel.demand_grade_id.in_(grade_ids))
        for grade_id, category_id in session.execute(category_stmt).all():
            category_map.setdefault(int(grade_id), []).append(int(category_id))

        package = EvidencePackage(
            package_id=package_id,
            run_id=context.run_id,
            biz_dt=context.biz_dt,
            package_version=1,
            status="building",
            source_snapshot_hash="0" * 64,
            source_versions_json={
                "run_id": context.run_id,
                "biz_dt": context.biz_dt,
                "date_snapshot": context.date_snapshot,
                "config_snapshot": context.config_snapshot,
                "input_snapshot": context.input_snapshot,
            },
            evidence_count=0,
        )
        session.add(package)
        session.flush()

        all_content_hashes: set[str] = set()
        created_expressions: set[str] = set()
        version_count = 0
        relation_count = 0

        for grade in grades:
            normalized = _normalized_text(grade.demand_name)
            canonical_key = _hash_payload({"normalized_text": normalized})
            term_id = _uuid(f"standard-term:{canonical_key}")
            platform_demand_id = _uuid(f"platform-demand:{canonical_key}")

            term = session.scalar(
                select(StandardDemandTerm).where(
                    StandardDemandTerm.canonical_key == canonical_key
                )
            )
            if term is None:
                term = StandardDemandTerm(
                    term_id=term_id,
                    canonical_key=canonical_key,
                    canonical_text=grade.demand_name,
                    normalized_text=normalized,
                    status="active",
                )
                session.add(term)

            platform = session.scalar(
                select(PlatformDemand).where(
                    PlatformDemand.canonical_key == canonical_key
                )
            )
            if platform is None:
                platform = PlatformDemand(
                    platform_demand_id=platform_demand_id,
                    canonical_key=canonical_key,
                    name=grade.demand_name,
                    description=f"可搜索、可供给的平台需求：{grade.demand_name}",
                    status="active",
                    lifecycle_state="new",
                )
                session.add(platform)
            session.flush()

            pool_ids = _parse_int_list(grade.related_pool_ids)
            pool_rows = MultiDemandPoolDiRepository(session).get_by_ids(pool_ids)
            source_items: list[tuple[str, str, dict[str, Any]]] = []
            for pool_row in pool_rows:
                source_items.append(
                    (
                        "multi_demand_pool_di",
                        str(pool_row.id),
                        _pool_payload(pool_row),
                    )
                )
            feedback_rows = list(
                session.scalars(
                    select(DemandFeedback).where(
                        DemandFeedback.platform_demand_id == platform_demand_id,
                        DemandFeedback.processing_status == "accepted",
                        DemandFeedback.consumed_run_id.is_(None),
                    )
                ).all()
            )
            for feedback in feedback_rows:
                source_items.append(
                    (
                        "demand_feedback",
                        str(feedback.id),
                        {
                            "id": int(feedback.id),
                            "demand_name": (
                                feedback.content
                                or feedback.reason_code
                                or feedback.feedback_action
                            ),
                            "feedback_action": feedback.feedback_action,
                            "reason_code": feedback.reason_code,
                            "content": feedback.content,
                            "resolution_reason": feedback.resolution_reason,
                            "impact_json": feedback.impact_json,
                            "feedback_user_id": feedback.feedback_user_id,
                            "created_at": (
                                feedback.created_at.isoformat()
                                if feedback.created_at is not None
                                else None
                            ),
                        },
                    )
                )
                feedback.consumed_run_id = context.run_id
            if not source_items:
                source_items.append(
                    (
                        "demand_grade",
                        str(grade.id),
                        {
                            "id": int(grade.id),
                            "demand_name": grade.demand_name,
                            "grade": grade.grade,
                            "score": (
                                float(grade.score) if grade.score is not None else None
                            ),
                            "reason": grade.reason,
                            "biz_dt": grade.biz_dt,
                        },
                    )
                )

            expression_ids: list[str] = []
            expression_hashes: list[str] = []
            for source_type, source_record_id, source_payload in source_items:
                expression_id = _uuid(
                    f"raw-expression:{package_id}:{source_type}:{source_record_id}"
                )
                content_hash = _hash_payload(source_payload)
                all_content_hashes.add(content_hash)
                expression_ids.append(expression_id)
                expression_hashes.append(content_hash)
                expression = session.get(RawDemandExpression, expression_id)
                if expression is None:
                    expression = RawDemandExpression(
                        expression_id=expression_id,
                        package_id=package_id,
                        source_type=source_type,
                        source_record_id=source_record_id,
                        raw_text=str(source_payload["demand_name"]),
                        original_payload_json=source_payload,
                        content_hash=content_hash,
                        observed_biz_dt=context.biz_dt,
                        data_quality="available",
                    )
                    session.add(expression)
                    created_expressions.add(expression_id)
                alias_id = _uuid(f"standard-alias:{term_id}:{expression_id}")
                if session.get(StandardDemandAlias, alias_id) is None:
                    session.add(
                        StandardDemandAlias(
                            alias_id=alias_id,
                            term_id=term_id,
                            expression_id=expression_id,
                            alias_text=str(source_payload["demand_name"]),
                            relation_type="source_expression",
                            confidence=Decimal("1.00000"),
                            reason="由当日原始需求表达直接归一到代表需求词",
                        )
                    )

            category_ids = sorted(set(category_map.get(int(grade.id), [])))
            if not category_ids:
                category_ids = sorted(set(_parse_int_list(grade.category_ids)))
            evidence_hash = _hash_payload(
                {
                    "grade_id": int(grade.id),
                    "grade": grade.grade,
                    "score": grade.score,
                    "reason": grade.reason,
                    "expression_hashes": sorted(expression_hashes),
                    "category_ids": category_ids,
                }
            )
            existing_version = session.scalar(
                select(PlatformDemandVersion).where(
                    PlatformDemandVersion.platform_demand_id == platform_demand_id,
                    PlatformDemandVersion.run_id == context.run_id,
                )
            )
            if existing_version is not None:
                continue
            previous = session.scalar(
                select(PlatformDemandVersion)
                .where(
                    PlatformDemandVersion.platform_demand_id == platform_demand_id
                )
                .order_by(PlatformDemandVersion.version_no.desc())
                .limit(1)
            )
            version_no = int(previous.version_no) + 1 if previous is not None else 1
            change_type = (
                "created"
                if previous is None
                else (
                    "unchanged"
                    if previous.evidence_hash == evidence_hash
                    else "updated"
                )
            )
            version_id = _uuid(
                f"platform-demand-version:{platform_demand_id}:{context.run_id}"
            )
            confidence = _score_confidence(grade.score)
            session.add(
                PlatformDemandVersion(
                    platform_demand_version_id=version_id,
                    platform_demand_id=platform_demand_id,
                    term_id=term_id,
                    package_id=package_id,
                    run_id=context.run_id,
                    source_demand_grade_id=int(grade.id),
                    version_no=version_no,
                    biz_dt=context.biz_dt,
                    name=grade.demand_name,
                    description=f"可搜索、可验证的平台需求：{grade.demand_name}",
                    cognition_confidence=confidence,
                    reason=grade.reason,
                    change_type=change_type,
                    evidence_hash=evidence_hash,
                )
            )
            for index, category_id in enumerate(category_ids):
                relation_type = (
                    "primary_attachment" if index == 0 else "auxiliary_attachment"
                )
                session.add(
                    PlatformDemandCategoryRel(
                        rel_id=_uuid(
                            f"platform-category-rel:{version_id}:"
                            f"{category_id}:{relation_type}"
                        ),
                        platform_demand_version_id=version_id,
                        category_id=category_id,
                        relation_type=relation_type,
                        relation_source="demand_grade_snapshot",
                        reason=grade.reason,
                        confidence=confidence,
                        is_inferred=True,
                        status="active",
                        valid_from_biz_dt=context.biz_dt,
                    )
                )
                relation_count += 1
            version_count += 1

        package.source_snapshot_hash = _hash_payload(sorted(all_content_hashes))
        package.evidence_count = len(created_expressions)
        package.status = "frozen"
        session.flush()

        return {
            "success": True,
            "idempotent_replay": False,
            "package_id": package.package_id,
            "biz_dt": context.biz_dt,
            "evidence_count": package.evidence_count,
            "platform_demand_versions": version_count,
            "category_relations": relation_count,
            "source_snapshot_hash": package.source_snapshot_hash,
        }