PayrollManagementSystem/backend/app/services/system_setting_service.py

"""系统设置（公司信息）业务编排。

单例行语义：整个系统恒一行（id=1，迁移种子保底）。logo 只存**相对路径**
（uploads/logo/ 下固定名覆盖），文件不入库的理由：图片进数据库会让每次
读取与备份放大传输，且无法独立替换；DB 只持有一行指针。

logo 安全：后缀白名单 + magic bytes 双重校验（上传内容不可信，仅看文件名
后缀可被伪造）；固定文件名落盘，**不使用上传方文件名**——杜绝路径拼接与
覆盖任意文件。
"""

import logging
from pathlib import Path

from sqlalchemy.orm import Session

from app.core.config import get_settings
from app.core.constants import ACTION_UPDATE, MODULE_SYSTEM_SETTINGS
from app.core.exceptions import BadRequestError, NotFoundError
from app.models import SystemSetting, User
from app.schemas.setting import SettingOut, SettingPublicOut, SettingUpdateIn
from app.services import audit_service
from app.services.audit_service import AuditRecord, RequestMeta

logger = logging.getLogger(__name__)

# magic bytes：PNG 固定 8 字节签名；JPEG 以 SOI + APP0/JFIF 起始
_MAGIC = {
    "png": b"\x89PNG\r\n\x1a\n",
    "jpg": b"\xff\xd8\xff",
}
LOGO_MAX_BYTES = 2 * 1024 * 1024  # 2MB：页眉/登录页展示尺寸上限，防上传滥用

SETTINGS_ID = 1  # 全系统单行主键


def get_setting(session: Session) -> SystemSetting:
    """取公司信息单行（种子保底 id=1；防御式兜底：极端环境下缺行则补建）。"""
    setting = session.get(SystemSetting, SETTINGS_ID)
    if setting is None:
        setting = SystemSetting(id=SETTINGS_ID, company_name="衍水集团")
        session.add(setting)
        session.flush()
    return setting


def to_out(setting: SystemSetting) -> SettingOut:
    """管理端出参（logo_url 带更新时间版本参数，替换图片后浏览器不再用旧缓存）。"""
    return SettingOut(
        company_name=setting.company_name,
        logo_url=_logo_url(setting),
    )


def to_public_out(setting: SystemSetting) -> SettingPublicOut:
    """匿名端点出参（白名单字段，独立 schema 见模块说明）。"""
    return SettingPublicOut(
        company_name=setting.company_name,
        logo_url=_logo_url(setting),
    )


def _logo_url(setting: SystemSetting) -> str | None:
    if not setting.logo_path:
        return None
    version = int(setting.update_time.timestamp()) if setting.update_time else 0
    return f"/api/v1/system/settings/logo/public?v={version}"


def update_company(
    session: Session,
    payload: SettingUpdateIn,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> SettingOut:
    """更新公司名称（审计留痕）。"""
    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_SYSTEM_SETTINGS,
            action=ACTION_UPDATE,
            summary=f"修改公司名称为「{payload.company_name}」",
            request_data=payload.model_dump(),
        ),
        meta=meta,
    ):
        setting = get_setting(session)
        setting.company_name = payload.company_name.strip()
        session.commit()
        return to_out(setting)


def save_logo(
    session: Session,
    content: bytes,
    ext: str,
    *,
    operator: User,
    meta: RequestMeta | None = None,
) -> SettingOut:
    """上传 logo：校验 → 固定名覆盖落盘 → 更新相对路径（审计留痕）。

    旧扩展名文件一并清理（logo.png 换 jpg 后残留文件会继续占盘）。
    """
    ext = ext.lower()
    if ext not in _MAGIC:
        raise BadRequestError("logo 仅支持 png/jpg 格式")
    if len(content) > LOGO_MAX_BYTES:
        raise BadRequestError("logo 文件不能超过 2MB")
    if not content.startswith(_MAGIC[ext]):
        raise BadRequestError("文件内容与扩展名不符（仅接受真实图片）")

    with audit_service.audit(
        session,
        operator=operator,
        record=AuditRecord(
            module=MODULE_SYSTEM_SETTINGS,
            action=ACTION_UPDATE,
            summary="更新公司 logo",
            request_data={"ext": ext, "size": len(content)},
        ),
        meta=meta,
    ):
        setting = get_setting(session)
        old_path = setting.logo_path

        # 先写新文件、commit 成功后再清旧扩展名（若把清理放在 commit 之前，
        # commit 失败会让库指旧路径而旧文件已删——断链无补偿）；同扩展名
        # 由写入覆盖，无需也不会误删新文件
        target = _logo_dir() / f"logo.{ext}"
        target.write_bytes(content)

        setting.logo_path = f"logo/logo.{ext}"
        session.commit()

        old_ext = Path(old_path).suffix.lstrip(".") if old_path else None
        if old_ext and old_ext != ext:
            _cleanup_old_logo(old_ext)
        logger.info("更新公司 logo | size=%s ext=%s", len(content), ext)
        return to_out(setting)


def read_logo(session: Session) -> tuple[Path, str]:
    """按库中记录定位 logo 文件，返回 (路径, 扩展名)。未上传抛 404。"""
    setting = get_setting(session)
    if not setting.logo_path:
        raise NotFoundError("尚未上传 logo")
    path = get_settings().upload_dir / setting.logo_path
    if not path.is_file():
        raise NotFoundError("logo 文件不存在，请重新上传")
    return path, path.suffix.lstrip(".")


def _logo_dir() -> Path:
    directory = get_settings().upload_dir / "logo"
    directory.mkdir(parents=True, exist_ok=True)
    return directory


def _cleanup_old_logo(old_ext: str) -> None:
    if old_ext in _MAGIC:
        old_file = _logo_dir() / f"logo.{old_ext}"
        if old_file.exists():
            old_file.unlink(missing_ok=True)