shaguabijia-app-server/app/repositories/app_config.py

"""运营配置读写。

get_value:业务读配置,DB 没有则 fallback 到 CONFIG_DEFS 默认(= 原 rewards 常量,空表行为不变)。
不缓存:配置读频率低(每次福利操作读一次,key 主键查极快),admin 改了立即生效、跨进程一致。
"""
from __future__ import annotations

from typing import Any

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.core.config_schema import CONFIG_DEFS
from app.models.app_config import AppConfig


def get_value(db: Session, key: str) -> Any:
    """读配置值。DB 有用 DB,没有 fallback 到 CONFIG_DEFS 默认。未知 key 抛 KeyError。"""
    row = db.get(AppConfig, key)
    if row is not None:
        return row.value
    return CONFIG_DEFS[key]["default"]


def set_value(
    db: Session, key: str, value: Any, *, admin_id: int, commit: bool = True
) -> AppConfig:
    """写配置(admin 用)。未知 key 抛 KeyError。commit=False 让调用方与审计同事务。"""
    if key not in CONFIG_DEFS:
        raise KeyError(f"unknown config key: {key}")
    row = db.get(AppConfig, key)
    if row is None:
        row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
        db.add(row)
    else:
        row.value = value
        row.updated_by_admin_id = admin_id
    if commit:
        db.commit()
        db.refresh(row)
    else:
        db.flush()
    return row


def list_all(db: Session) -> list[dict]:
    """所有可配项:default + 当前值(DB 有则用 DB) + 元信息。给 admin 配置页渲染。"""
    overrides = {r.key: r for r in db.execute(select(AppConfig)).scalars().all()}
    out: list[dict] = []
    for key, defn in CONFIG_DEFS.items():
        row = overrides.get(key)
        out.append(
            {
                "key": key,
                "label": defn["label"],
                "group": defn["group"],
                "type": defn["type"],
                "help": defn.get("help"),
                "default": defn["default"],
                "value": row.value if row is not None else defn["default"],
                "overridden": row is not None,
                "updated_at": row.updated_at.isoformat() if row is not None else None,
            }
        )
    return out


# ── 最新 App 版本(OTA 检查更新)──────────────────────────────────────────────
# 版本信息不是"运营可配置项",不进 CONFIG_DEFS:它由发布流程(发车)/应急整组写入、客户端读取做
# OTA,语义是"发布系统状态"而非"运营调参"。故走下面专用读写,直接操作 AppConfig 表(复用表不复用 repo)。
APP_VERSION_KEY = "latest_app_version"


def get_app_version(db: Session) -> dict | None:
    """读最新 App 版本信息(OTA)。无记录返回 None(客户端视为无更新)。"""
    row = db.get(AppConfig, APP_VERSION_KEY)
    return row.value if row is not None else None


def set_app_version(db: Session, data: dict, *, commit: bool = True) -> AppConfig:
    """整组原子覆盖最新版本信息(发布流程/应急写,非 admin,故 updated_by_admin_id 留空)。"""
    row = db.get(AppConfig, APP_VERSION_KEY)
    if row is None:
        row = AppConfig(key=APP_VERSION_KEY, value=data, updated_by_admin_id=None)
        db.add(row)
    else:
        row.value = data
    if commit:
        db.commit()
        db.refresh(row)
    else:
        db.flush()
    return row


# ── 穿山甲广告配置(admin 可配,客户端动态拉)──────────────────────────────────
# 同 app_version:结构化 dict,复用 AppConfig 表不进 CONFIG_DEFS(它是一组关联配置,不是散项)。
# 默认值 = 客户端当前内置的硬编码 ID(AdConfig.kt),空库时下发默认 = 维持现状。
# reward_mkey 是 GroMore S2S 验签密钥(机密),只后端验签用、绝不下发客户端(见 platform/ad-config)。
AD_CONFIG_KEY = "ad_config"
_AD_CONFIG_DEFAULTS: dict[str, Any] = {
    "app_id": "5830519",                  # 穿山甲应用ID(正式)
    "reward_code_id": "104099389",        # 福利页激励视频位
    # 比价与领券共用同一穿山甲 Draw 代码位(默认 104098712 = 后台「Draw信息流」位),靠 feed_scene
    # (comparison/coupon)区分收益;运营可在后台把两者拆成不同代码位。字段从旧 *_feed_code_id 改名为 *_draw_code_id。
    "compare_draw_code_id": "104098712",  # 比价 Draw 代码位(104098712 = Draw 信息流位)
    "coupon_draw_code_id": "104098712",   # 领券 Draw 代码位(初始同比价,运营可拆)
    "reward_mkey": "",                    # 激励位 GroMore 验签密钥(空则回退 .env PANGLE_REWARD_SECRET*)
    "reward_enabled": True,               # 福利激励视频开关
    "compare_ad_enabled": True,           # 比价广告开关
    "coupon_ad_enabled": True,            # 领券广告开关
    "withdrawal_ad_enabled": True,        # 提现激励视频开关(关=直接放行提现,不挡用户的钱)
}


def get_ad_config(db: Session) -> dict:
    """读广告配置。DB 无则返回默认(=客户端内置值,维持现状);DB 有则与默认 merge
    (新增字段向后兼容,老记录缺的字段用默认补)。"""
    row = db.get(AppConfig, AD_CONFIG_KEY)
    merged = dict(_AD_CONFIG_DEFAULTS)
    if row is not None and isinstance(row.value, dict):
        merged.update(row.value)
    return merged


def set_ad_config(db: Session, data: dict, *, admin_id: int, commit: bool = True) -> AppConfig:
    """admin 写广告配置。与现有值 merge(部分更新),只接受已知字段(防脏键)。"""
    row = db.get(AppConfig, AD_CONFIG_KEY)
    merged = dict(_AD_CONFIG_DEFAULTS)
    if row is not None and isinstance(row.value, dict):
        merged.update(row.value)
    merged.update({k: v for k, v in data.items() if k in _AD_CONFIG_DEFAULTS})
    if row is None:
        row = AppConfig(key=AD_CONFIG_KEY, value=merged, updated_by_admin_id=admin_id)
        db.add(row)
    else:
        row.value = merged
        row.updated_by_admin_id = admin_id
    if commit:
        db.commit()
        db.refresh(row)
    else:
        db.flush()
    return row