demand_server/app/strategies/sources/hot_content_sync_log.py

from typing import Any

from sqlalchemy import text

from app.core.config import settings
from app.db.hot_content_mysql import HotContentSessionLocal


def _hot_strategy_name(strategy: str | None = None) -> str:
    value = (strategy or settings.hot_demand_pool_strategy or "新热事件").strip()
    return value or "新热事件"


def fetch_hot_content_sync_log_by_partition(
    *,
    partition_dt: str,
    strategy: str | None = None,
) -> list[dict[str, Any]]:
    strategy_value = _hot_strategy_name(strategy)
    query = text(
        """
        SELECT
            partition_dt,
            strategy,
            demand_id,
            demand_name,
            demand_type,
            weight
        FROM hot_content_odps_sync_log
        WHERE partition_dt = :partition_dt
          AND strategy = :strategy
        ORDER BY id ASC
        """
    )
    with HotContentSessionLocal() as session:
        rows = session.execute(
            query,
            {
                "partition_dt": partition_dt,
                "strategy": strategy_value,
            },
        ).mappings().all()

    results: list[dict[str, Any]] = []
    for row in rows:
        demand_id = str(row.get("demand_id") or "").strip()
        demand_name = str(row.get("demand_name") or "").strip()
        if not demand_id or not demand_name:
            continue
        results.append(
            {
                "partition_dt": str(row.get("partition_dt") or partition_dt).strip(),
                "strategy": str(row.get("strategy") or strategy_value).strip(),
                "demand_id": demand_id,
                "demand_name": demand_name,
                "demand_type": str(row.get("demand_type") or "").strip() or None,
                "weight": row.get("weight"),
            }
        )
    return results