demand_server/app/strategies/config_store.py

import json
import re
from dataclasses import dataclass
from typing import Any

from sqlalchemy import text

from app.core.config import settings
from app.db.mysql import SessionLocal

IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


def _safe_identifier(name: str) -> str:
    if not IDENTIFIER_RE.match(name):
        raise ValueError(f"invalid sql identifier: {name}")
    return name


@dataclass(frozen=True)
class StrategyConfigRecord:
    strategy_id: str
    name: str
    version: str
    params: dict[str, Any]
    active: bool


def _parse_params(raw: object) -> dict[str, Any]:
    if raw is None:
        return {}
    if isinstance(raw, dict):
        return raw
    if isinstance(raw, (bytes, bytearray)):
        raw = raw.decode("utf-8")
    if isinstance(raw, str):
        text_value = raw.strip()
        if not text_value:
            return {}
        return json.loads(text_value)
    raise TypeError(f"unsupported params type: {type(raw)!r}")


def ensure_strategy_tables() -> None:
    config_table = _safe_identifier(settings.strategy_config_table)
    staging_table = _safe_identifier(settings.strategy_staging_table)

    config_sql = f"""
    CREATE TABLE IF NOT EXISTS {config_table}
    (
        id           BIGINT       NOT NULL AUTO_INCREMENT COMMENT '自增id',
        strategy_id  VARCHAR(64)  NOT NULL COMMENT '全局唯一id，英文名称',
        name         VARCHAR(128) NOT NULL COMMENT '策略名称',
        version      VARCHAR(32)  NOT NULL COMMENT '策略版本，用于追踪代码变更',
        params       JSON         NULL COMMENT '策略运行参数，如阈值、窗口期等',
        active       TINYINT(1)   NOT NULL DEFAULT 1 COMMENT '是否启用，0=不参与产出和合并',
        create_time  TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        updated_time TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
        PRIMARY KEY (id),
        UNIQUE KEY uk_strategy_id (strategy_id)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='策略配置表'
    """

    staging_sql = f"""
    CREATE TABLE IF NOT EXISTS {staging_table}
    (
        id                 BIGINT        NOT NULL AUTO_INCREMENT COMMENT '自增id',
        strategy_config_id VARCHAR(64)   NOT NULL COMMENT '来源策略ID，关联 strategy_config.strategy_id',
        strategy           VARCHAR(32)   NULL COMMENT '策略名称',
        demand_id          VARCHAR(64)   NULL COMMENT '需求id',
        demand_name        VARCHAR(512)  NULL COMMENT '需求',
        weight             DOUBLE        NULL COMMENT '需求权重',
        `type`             VARCHAR(64)   NULL COMMENT '需求类型',
        video_count        INT           NULL COMMENT '视频数量',
        video_list         JSON          NULL COMMENT '视频列表',
        extend             JSON          NULL COMMENT '拓展字段',
        batch_date         VARCHAR(32)   NULL COMMENT '归属批次日期，e.g. 20250526',
        create_time        TIMESTAMP     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
        PRIMARY KEY (id),
        KEY idx_strategy_config_id (strategy_config_id),
        KEY idx_batch_date (batch_date),
        KEY idx_demand_id (demand_id)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='策略暂存表'
    """

    with SessionLocal() as session:
        session.execute(text(config_sql))
        session.execute(text(staging_sql))
        session.commit()


def fetch_all_configs() -> list[StrategyConfigRecord]:
    config_table = _safe_identifier(settings.strategy_config_table)
    query = text(
        f"""
        SELECT strategy_id, name, version, params, active
        FROM {config_table}
        ORDER BY strategy_id ASC
        """
    )
    with SessionLocal() as session:
        rows = session.execute(query).mappings().all()

    return [
        StrategyConfigRecord(
            strategy_id=str(row["strategy_id"]),
            name=str(row["name"]),
            version=str(row["version"]),
            params=_parse_params(row["params"]),
            active=bool(row["active"]),
        )
        for row in rows
    ]


def fetch_config_by_id(strategy_id: str) -> StrategyConfigRecord | None:
    config_table = _safe_identifier(settings.strategy_config_table)
    query = text(
        f"""
        SELECT strategy_id, name, version, params, active
        FROM {config_table}
        WHERE strategy_id = :strategy_id
        LIMIT 1
        """
    )
    with SessionLocal() as session:
        row = session.execute(query, {"strategy_id": strategy_id}).mappings().first()

    if row is None:
        return None

    return StrategyConfigRecord(
        strategy_id=str(row["strategy_id"]),
        name=str(row["name"]),
        version=str(row["version"]),
        params=_parse_params(row["params"]),
        active=bool(row["active"]),
    )