demand_server/app/strategies/sources/supply_demand_content.py

import json
from typing import Any

from sqlalchemy import text

from app.core.config import settings
from app.db.supply_mysql import SupplySessionLocal
from app.strategies.config_store import _safe_identifier


def _parse_ext_data(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)
    return {}


def fetch_demand_content_by_dt(dt: str) -> list[dict[str, Any]]:
    table = _safe_identifier(settings.supply_demand_content_table)
    query = text(
        f"""
        SELECT
            merge_leve2,
            name,
            reason,
            suggestion,
            ext_data,
            score,
            dt
        FROM {table}
        WHERE dt = :dt
          AND name IS NOT NULL
          AND TRIM(name) <> ''
        ORDER BY score DESC, name ASC
        """
    )
    with SupplySessionLocal() as session:
        rows = session.execute(query, {"dt": dt}).mappings().all()

    results: list[dict[str, Any]] = []
    for row in rows:
        ext_data = _parse_ext_data(row["ext_data"])
        results.append(
            {
                "merge_leve2": row.get("merge_leve2"),
                "name": str(row["name"]).strip(),
                "reason": row.get("reason"),
                "suggestion": row.get("suggestion"),
                "ext_data": ext_data,
                "demand_type": ext_data.get("type"),
                "score": row.get("score"),
                "dt": str(row.get("dt") or dt).strip(),
            }
        )
    return results