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"]), )