ajyy/backend/app/services/config_service.py

"""热线 / 二维码配置读写"""
import re
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from app.core.response import AppHTTPError
from app.crud import config as config_crud
from app.crud import oper_log as oper_log_crud

_MULTI_SPLIT_RE = re.compile(r"[,，、;\n\r]+")


def parse_multi(raw: Optional[str]) -> List[str]:
    """将存储串拆成去重后的列表（兼容逗号/顿号/分号/换行）。"""
    if not raw or not str(raw).strip():
        return []
    seen = set()
    out: List[str] = []
    for part in _MULTI_SPLIT_RE.split(str(raw)):
        item = part.strip()
        if not item or item in seen:
            continue
        seen.add(item)
        out.append(item)
    return out


def join_multi(items: Optional[List[str]]) -> str:
    """规范化列表为逗号分隔存储串。"""
    if not items:
        return ""
    return ",".join(parse_multi(",".join(str(p) for p in items if p is not None)))


# 兼容旧名称
parse_phones = parse_multi
join_phones = join_multi


def get_hotline_payload(db: Session) -> Dict[str, Any]:
    row = config_crud.get_hotline(db)
    if not row:
        raise AppHTTPError(404, "热线配置不存在")
    phones = parse_multi(row.phone)
    emails = parse_multi(row.email)
    addresses = parse_multi(row.address)
    return {
        "id": row.id,
        "phone": phones[0] if phones else "",
        "phones": phones,
        "email": emails[0] if emails else "",
        "emails": emails,
        "address": addresses[0] if addresses else "",
        "addresses": addresses,
        "workTime": row.work_time or "",
    }


def get_qrcode_payload(db: Session) -> Dict[str, Any]:
    row = config_crud.get_qrcode(db)
    if not row:
        raise AppHTTPError(404, "二维码配置不存在")
    return {
        "id": row.id,
        "landUrl": row.land_url or "",
        "title": row.title or "",
        "imageUrl": row.image_url or "",
    }


def update_hotline(
    db: Session,
    user,
    *,
    phone: Optional[str] = None,
    phones: Optional[List[str]] = None,
    email: Optional[str] = None,
    emails: Optional[List[str]] = None,
    address: Optional[str] = None,
    addresses: Optional[List[str]] = None,
    work_time: Optional[str] = None,
) -> Dict[str, Any]:
    row = config_crud.get_hotline(db)
    if not row:
        raise AppHTTPError(404, "热线配置不存在")

    stored_phone = phone
    if phones is not None:
        stored_phone = join_multi(phones)
    elif phone is not None:
        stored_phone = join_multi(parse_multi(phone))

    stored_email = email
    if emails is not None:
        stored_email = join_multi(emails)
    elif email is not None:
        stored_email = join_multi(parse_multi(email))

    stored_address = address
    if addresses is not None:
        stored_address = join_multi(addresses)
    elif address is not None:
        stored_address = join_multi(parse_multi(address))

    config_crud.update_hotline(
        db,
        row,
        phone=stored_phone,
        email=stored_email,
        address=stored_address,
        work_time=work_time,
        update_by=user.username,
    )
    oper_log_crud.create_oper_log(
        db,
        module="hotline",
        action="update",
        user_id=user.id,
        username=user.username,
        detail="更新热线配置",
    )
    return get_hotline_payload(db)


def update_qrcode(
    db: Session,
    user,
    *,
    land_url: Optional[str] = None,
    title: Optional[str] = None,
    image_url: Optional[str] = None,
) -> Dict[str, Any]:
    row = config_crud.get_qrcode(db)
    if not row:
        raise AppHTTPError(404, "二维码配置不存在")
    config_crud.update_qrcode(
        db,
        row,
        land_url=land_url,
        title=title,
        image_url=image_url,
        update_by=user.username,
    )
    oper_log_crud.create_oper_log(
        db,
        module="qrcode",
        action="update",
        user_id=user.id,
        username=user.username,
        detail="更新二维码配置",
    )
    return get_qrcode_payload(db)