gsfdc/server/app/models/content.py

import json
from datetime import datetime, timezone

from sqlalchemy import DateTime, Integer, String, Text, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base

def _mysql_args() -> dict:
    return {"mysql_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}

DEFAULTS: dict[str, str] = {
    "site_name": "甘肃省房地产价格评估行业协会",
    "logo_file_id": "",
    "banner_file_ids": "[]",
    "icp_text": "陇ICP备19001227号",
    "police_text": "甘公网安备62010202001984号",
    "friend_links": "[]",
    "link_cirea": "",
    "link_housing": "",
    "complaint_workdays": "",
    "expert_reply_workdays": "",
    "upload_max_bytes": "10485760",
    "login_max_failures": "5",
    "login_lock_minutes": "15",
}

SETTING_KEYS = tuple(DEFAULTS)
_JSON_KEYS = {"banner_file_ids", "friend_links"}
_NUMBER_KEYS = {"upload_max_bytes", "login_max_failures", "login_lock_minutes"}
_OPTIONAL_NUMBER_KEYS = {"complaint_workdays", "expert_reply_workdays"}


def _now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


class SiteSetting(Base):
    __tablename__ = "site_setting"
    __table_args__ = _mysql_args()

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    key: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    value: Mapped[str] = mapped_column(Text, nullable=False, default="")


class AdminFile(Base):
    __tablename__ = "admin_file"
    __table_args__ = _mysql_args()

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    bucket: Mapped[str] = mapped_column(String(16), nullable=False, index=True)
    object_key: Mapped[str] = mapped_column(String(255), nullable=False)
    filename: Mapped[str] = mapped_column(String(255), nullable=False)
    content_type: Mapped[str] = mapped_column(String(128), nullable=False, default="")
    size: Mapped[int] = mapped_column(Integer, nullable=False)
    created_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=_now)


class SensitiveWord(Base):
    __tablename__ = "sensitive_word"
    __table_args__ = _mysql_args()

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    word: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=_now)


def ensure_site_defaults(db) -> None:
    existing = set(db.scalars(select(SiteSetting.key)))
    missing = [
        SiteSetting(key=key, value=value)
        for key, value in DEFAULTS.items()
        if key not in existing
    ]
    if not missing:
        return
    db.add_all(missing)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()


def read_upload_limit(db) -> int | None:
    row = db.scalar(select(SiteSetting).where(SiteSetting.key == "upload_max_bytes"))
    if row is None:
        return None
    text = str(row.value).strip()
    if not text.isdigit():
        return None
    value = int(text)
    if value <= 0:
        return None
    return value


def save_settings(db, values: dict[str, str]) -> None:
    ensure_site_defaults(db)
    for key, value in values.items():
        row = db.scalar(select(SiteSetting).where(SiteSetting.key == key))
        if row is None:
            db.add(SiteSetting(key=key, value=value))
        else:
            row.value = value
    db.commit()


def _present(key: str, raw: str):
    if key in _JSON_KEYS:
        if not raw:
            return []
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            return []
        return parsed if isinstance(parsed, list) else []
    if key in _NUMBER_KEYS:
        text = str(raw).strip()
        if text.isdigit():
            return int(text)
        return raw
    if key in _OPTIONAL_NUMBER_KEYS:
        text = "" if raw is None else str(raw).strip()
        if not text:
            return ""
        if text.isdigit():
            return int(text)
        return text
    return "" if raw is None else str(raw)


def _public_urls(db, file_ids: list) -> list[str]:
    from app.services.storage import public_object_url

    urls: list[str] = []
    for item in file_ids:
        if isinstance(item, bool):
            continue
        try:
            file_id = int(str(item).strip())
        except (TypeError, ValueError):
            continue
        if file_id <= 0:
            continue
        row = db.get(AdminFile, file_id)
        if row is None or row.bucket != "public" or not row.object_key:
            continue
        urls.append(public_object_url(row.object_key))
    return urls


def settings_view(db) -> dict:
    ensure_site_defaults(db)
    rows = db.scalars(select(SiteSetting)).all()
    raw = {row.key: row.value for row in rows}
    view = {key: _present(key, raw.get(key, DEFAULTS.get(key, ""))) for key in SETTING_KEYS}
    logo_id = raw.get("logo_file_id", "")
    logo_urls = _public_urls(db, [logo_id]) if str(logo_id).strip() else []
    view["logoUrl"] = logo_urls[0] if logo_urls else ""
    banners = view["banner_file_ids"] if isinstance(view["banner_file_ids"], list) else []
    view["bannerUrls"] = _public_urls(db, banners)
    return view