SupplyAgent/supply_infra/db/models/demand_grade.py

from __future__ import annotations

from datetime import datetime
from decimal import Decimal

from sqlalchemy import BigInteger, Integer, Numeric, String, Text, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column

from supply_infra.db.base import Base


class DemandGrade(Base):
    """需求分级结果 — 对 multi_demand_pool_di 中的需求按先验/后验热度评级。"""

    __tablename__ = "demand_grade"
    __table_args__ = (UniqueConstraint("biz_dt", "demand_name", name="uk_demand_grade"),)

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    biz_dt: Mapped[str] = mapped_column(String(32), nullable=False, comment="业务日期YYYYMMDD")
    demand_name: Mapped[str] = mapped_column(
        String(256), nullable=False, comment="需求名称（去重后的代表词）"
    )
    category_ids: Mapped[str | None] = mapped_column(
        Text, nullable=True, comment="归属的全局树节点id列表JSON数组（展示快照，权威关系见demand_grade_category_rel）"
    )
    grade: Mapped[str] = mapped_column(String(4), nullable=False, comment="等级 S/A/B/C/D")
    score: Mapped[Decimal | None] = mapped_column(
        Numeric(6, 2),
        nullable=True,
        comment="需求自身来源归一分0-100（来源内排名后对已有来源取均值）",
    )
    prior_total_score: Mapped[Decimal | None] = mapped_column(
        Numeric(16, 8), nullable=True, comment="落库时的先验 total_score 快照"
    )
    posterior_rov_avg: Mapped[Decimal | None] = mapped_column(
        Numeric(16, 8), nullable=True, comment="落库时的后验 rov_diff 快照（real_rov_7d_avg）"
    )
    posterior_rov_count: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, comment="落库时的后验样本数快照"
    )
    has_posterior: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, comment="是否有后验验证数据 0-无 1-有"
    )
    related_pool_ids: Mapped[str] = mapped_column(
        Text,
        nullable=False,
        comment="关联的原始 multi_demand_pool_di.id 列表JSON数组（必填，用于回溯原始需求）",
    )
    video_list: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        comment="关联视频列表JSON（最多10个，由related_pool_ids对应原始行的video_list合并去重得到）",
    )
    strategies: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
        comment="来源策略列表JSON数组（由related_pool_ids对应原始行的strategy去重得到）",
    )
    reason: Mapped[str] = mapped_column(Text, nullable=False, comment="判断依据")
    create_time: Mapped[datetime] = mapped_column(
        nullable=False,
        server_default=func.now(),
        comment="创建时间",
    )
    update_time: Mapped[datetime] = mapped_column(
        nullable=False,
        server_default=func.now(),
        onupdate=func.now(),
        comment="更新时间",
    )