shaguabijia-app-server/app/admin/routers/config.py

"""admin 运营配置:列出所有可配项 + 改某项(带审计 + 改值校验)。

配置项定义见 app.core.config_schema.CONFIG_DEFS;业务读配置 fallback 默认(见 app_config repo)。
权限:operator / finance 都可改(运营调奖励、财务调额度),super 恒可。
"""
from __future__ import annotations

from typing import Annotated, Any

from fastapi import APIRouter, Depends, HTTPException, Request

from app.admin.audit import write_audit
from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
from app.admin.schemas.config import ConfigItemOut, ConfigUpdateRequest
from app.core import limit_policy
from app.core.config_schema import CONFIG_DEFS
from app.core.rewards import SIGNIN_CYCLE_LEN
from app.models.admin import AdminUser
from app.models.app_config import AppConfig
from app.repositories import app_config

router = APIRouter(
    prefix="/admin/api/config",
    tags=["admin-config"],
    dependencies=[Depends(get_current_admin)],
)


def _validate(key: str, value: Any) -> None:
    """按配置项 type 校验新值,不合法抛 ValueError(router 转 400)。"""
    t = CONFIG_DEFS[key]["type"]
    if t == "int":
        if not isinstance(value, int) or isinstance(value, bool) or value < 0:
            raise ValueError("需为非负整数")
        minimum = CONFIG_DEFS[key].get("min")
        maximum = CONFIG_DEFS[key].get("max")
        if minimum is not None and value < minimum:
            raise ValueError(f"不能小于 {minimum}")
        if maximum is not None and value > maximum:
            raise ValueError(f"不能大于 {maximum}")
    elif t == "int_list":
        if not isinstance(value, list) or not value:
            raise ValueError("需为非空整数列表")
        if not all(isinstance(x, int) and not isinstance(x, bool) and x >= 0 for x in value):
            raise ValueError("列表元素需为非负整数")
        if key == "signin_rewards" and len(value) != SIGNIN_CYCLE_LEN:
            raise ValueError(f"签到档位必须正好 {SIGNIN_CYCLE_LEN} 个(对应 {SIGNIN_CYCLE_LEN} 天循环)")
    elif t == "dict_str_int":
        if not isinstance(value, dict) or not all(
            isinstance(k, str) and isinstance(v, int) and not isinstance(v, bool)
            for k, v in value.items()
        ):
            raise ValueError("需为 {字符串: 整数} 映射")
    elif t == "bool":
        if not isinstance(value, bool):
            raise ValueError("需为布尔值")


def _limit_item(
    key: str,
    values: dict[str, int],
    bundle: AppConfig | None,
) -> ConfigItemOut:
    definition = CONFIG_DEFS[key]
    rule_code = limit_policy.RULE_CODE_BY_CONFIG_KEY[key]
    return ConfigItemOut(
        key=key,
        value=values[rule_code],
        label=definition["label"],
        group=definition["group"],
        type=definition["type"],
        help=definition.get("help"),
        default=definition["default"],
        overridden=bundle is not None,
        updated_at=(
            bundle.updated_at.isoformat() if bundle is not None else None
        ),
    )


def _item(db, key: str) -> ConfigItemOut:
    if key in limit_policy.RULE_CODE_BY_CONFIG_KEY:
        return _limit_item(
            key,
            limit_policy.get_global_limits(db),
            db.get(AppConfig, limit_policy.LIMIT_POLICY_GLOBAL_KEY),
        )
    for item in app_config.list_all(db):
        if item["key"] == key:
            return ConfigItemOut(**item)
    raise HTTPException(status_code=404, detail="未知配置项")


@router.get("", response_model=list[ConfigItemOut], summary="所有可配项 + 当前值(不含 hidden)")
def list_config(db: AdminDb) -> list[ConfigItemOut]:
    # hidden 项(已下线/由专用页管理,如福利页任务·里程碑·看广告调参、首页轮播数据源)不在本页渲染。
    legacy_items = {
        item["key"]: item for item in app_config.list_all(db)
    }
    values = limit_policy.get_global_limits(db)
    bundle = db.get(AppConfig, limit_policy.LIMIT_POLICY_GLOBAL_KEY)
    out: list[ConfigItemOut] = []
    for key, definition in CONFIG_DEFS.items():
        if definition.get("hidden"):
            continue
        if key in limit_policy.RULE_CODE_BY_CONFIG_KEY:
            out.append(_limit_item(key, values, bundle))
        else:
            out.append(ConfigItemOut(**legacy_items[key]))
    return out


@router.patch("/{key}", response_model=ConfigItemOut, summary="改某项配置(带审计)")
def update_config(
    key: str,
    body: ConfigUpdateRequest,
    request: Request,
    admin: Annotated[AdminUser, Depends(require_role("operator", "finance"))],
    db: AdminDb,
) -> ConfigItemOut:
    if key not in CONFIG_DEFS:
        raise HTTPException(status_code=404, detail="未知配置项")
    try:
        _validate(key, body.value)
        rule_code = limit_policy.RULE_CODE_BY_CONFIG_KEY.get(key)
        if rule_code is not None:
            before = limit_policy.get_global_limits(db)[rule_code]
            limit_policy.set_global_limits(
                db,
                {rule_code: body.value},
                admin_id=admin.id,
                commit=False,
            )
        else:
            before = app_config.get_value(db, key)
            app_config.set_value(
                db,
                key,
                body.value,
                admin_id=admin.id,
                commit=False,
            )
        if key == "ad_daily_limit":
            # 旧系统配置接口过去只有一个广告日上限。仍有人直接调用时，同时同步
            # 新的激励视频/Draw 两项，避免旧入口写入后业务实际值不变。
            limit_policy.set_global_limits(
                db,
                {
                    "ad.reward_video.daily": body.value,
                    "ad.feed.daily": body.value,
                },
                admin_id=admin.id,
                commit=False,
            )
    except ValueError as e:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(e)) from e
    write_audit(
        db, admin, action="config.set", target_type="config", target_id=key,
        detail={"before": before, "after": body.value}, ip=get_client_ip(request), commit=False,
    )
    db.commit()
    return _item(db, key)