zhuzihao
8a2f72d366
Co-authored-by: zzhyyyyy <2685922758@qq.com> Reviewed-on: #63 Co-authored-by: zhuzihao <zhuzihao@wonderable.ai> Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""反馈页二维码卡配置读写(运营后台可配 → App 反馈页同步)。
|
|
|
|
整张卡(开关 + 二维码图 + 三行文案)作为一个 JSON 对象,复用通用 app_config 表
|
|
(key=feedback_qr_card,value 存这个 dict)。表里没这行 → 返回内置默认(= App 反馈页原
|
|
硬编码文案,空配置 = 改造前现状,图片走 App 本地兜底)。admin 改 → 写一行 → App 下次拉
|
|
GET /api/v1/feedback/config 即生效(跨进程一致)。
|
|
|
|
不经 app.repositories.app_config(那层按 CONFIG_DEFS 白名单限定 key,且会出现在系统配置页);
|
|
本 key 由本模块独占维护,不进 CONFIG_DEFS,所以不会污染系统配置页。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.app_config import AppConfig
|
|
|
|
_KEY = "feedback_qr_card"
|
|
|
|
# 默认值镜像 App 反馈页原硬编码:空配置时整体行为与改造前一致。
|
|
_DEFAULTS: dict[str, Any] = {
|
|
"enabled": True,
|
|
"image_url": None, # None = App 用本地 asset 兜底(group_qr.png),再不行画占位伪码
|
|
"title": "长按图片保存二维码",
|
|
"group_name": "傻瓜比价官方群",
|
|
"subtitle": "一起唠嗑共创、解锁新玩法",
|
|
}
|
|
|
|
_FIELDS = tuple(_DEFAULTS.keys())
|
|
|
|
|
|
def _merge(raw: Any) -> dict[str, Any]:
|
|
"""把 DB 存的(可能不全的)dict 叠加到默认上,得到完整配置(5 个字段,无 updated_at)。"""
|
|
out = dict(_DEFAULTS)
|
|
if isinstance(raw, dict):
|
|
for k in _FIELDS:
|
|
v = raw.get(k)
|
|
if v is not None: # image_url 的 None 与默认 None 等价,无需特判
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def get_config(db: Session) -> dict[str, Any]:
|
|
"""完整配置 + updated_at(admin 读 / App 读共用;App schema 会忽略多余的 updated_at)。"""
|
|
row = db.get(AppConfig, _KEY)
|
|
cfg = _merge(row.value if row is not None else None)
|
|
cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
|
|
return cfg
|
|
|
|
|
|
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
|
|
"""整体覆写该行(value 须为完整 5 字段 dict),返回合并后的完整配置(含 updated_at)。"""
|
|
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 # 重新赋整个 dict,SQLAlchemy 能侦测到变更(in-place 改才侦测不到)
|
|
row.updated_by_admin_id = admin_id
|
|
if commit:
|
|
db.commit()
|
|
db.refresh(row)
|
|
else:
|
|
db.flush()
|
|
out = _merge(row.value)
|
|
out["updated_at"] = row.updated_at.isoformat() if row.updated_at else None
|
|
return out
|
|
|
|
|
|
def update_config(
|
|
db: Session,
|
|
*,
|
|
enabled: bool | None = None,
|
|
title: str | None = None,
|
|
group_name: str | None = None,
|
|
subtitle: str | None = None,
|
|
admin_id: int,
|
|
commit: bool = True,
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""改文案/开关(只改传了的字段;图片走 set_image)。返回 (before, after) 供审计。"""
|
|
row = db.get(AppConfig, _KEY)
|
|
before = _merge(row.value if row is not None else None)
|
|
new_value = {k: before[k] for k in _FIELDS}
|
|
if enabled is not None:
|
|
new_value["enabled"] = enabled
|
|
if title is not None:
|
|
new_value["title"] = title.strip()
|
|
if group_name is not None:
|
|
new_value["group_name"] = group_name.strip()
|
|
if subtitle is not None:
|
|
new_value["subtitle"] = subtitle.strip()
|
|
after = _write(db, new_value, admin_id=admin_id, commit=commit)
|
|
return before, after
|
|
|
|
|
|
def set_image(
|
|
db: Session, image_url: str | None, *, admin_id: int, commit: bool = True
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
"""设置/清空二维码图 URL。返回 (before, after);before['image_url'] 供调用方删旧文件。"""
|
|
row = db.get(AppConfig, _KEY)
|
|
before = _merge(row.value if row is not None else None)
|
|
new_value = {k: before[k] for k in _FIELDS}
|
|
new_value["image_url"] = image_url
|
|
after = _write(db, new_value, admin_id=admin_id, commit=commit)
|
|
return before, after
|