ajyy/backend/app/services/admin_ledger_service.py

"""管理端反馈 / 建言办理"""
from typing import Any, Dict, List, Optional, Set

from sqlalchemy.orm import Session

from app.core.response import AppHTTPError
from app.crud import feedback as feedback_crud
from app.crud import oper_log as oper_log_crud
from app.crud import suggestion as suggestion_crud
from app.models.business import Feedback, Suggestion
from app.models.system import SysUser

FEEDBACK_TRANSITIONS: Dict[str, Set[str]] = {
    "待处理": {"处理中", "已办结"},
    "处理中": {"已办结"},
    "已办结": set(),
}

SUGGESTION_TRANSITIONS: Dict[str, Set[str]] = {
    "待查阅": {"已采纳", "已回复"},
    "已采纳": {"已回复"},
    "已回复": {"已采纳"},
}


def _iso(dt) -> Optional[str]:
    if not dt:
        return None
    return dt.isoformat(sep=" ", timespec="seconds")


def feedback_to_dict(row: Feedback) -> Dict[str, Any]:
    return {
        "id": row.id,
        "bizNo": row.biz_no,
        "fbType": row.fb_type,
        "targetName": row.target_name,
        "targetDept": row.target_dept,
        "targetPost": row.target_post,
        "content": row.content,
        "reporterName": row.reporter_name,
        "reporterUnit": row.reporter_unit,
        "reporterPost": row.reporter_post,
        "reporterPhone": row.reporter_phone,
        "attachmentUrls": list(row.attachments or []),
        "status": row.status,
        "handleRemark": row.handle_remark,
        "createTime": _iso(row.create_time),
        "updateTime": _iso(row.update_time),
    }


def suggestion_to_dict(row: Suggestion) -> Dict[str, Any]:
    return {
        "id": row.id,
        "bizNo": row.biz_no,
        "content": row.content,
        "contactName": row.contact_name,
        "contactUnit": row.contact_unit,
        "contactPost": row.contact_post,
        "contactPhone": row.contact_phone,
        "attachmentUrls": list(row.attachments or []),
        "status": row.status,
        "handleRemark": row.handle_remark,
        "createTime": _iso(row.create_time),
        "updateTime": _iso(row.update_time),
    }


def _assert_transition(
    current: str, new_status: str, table: Dict[str, Set[str]]
) -> None:
    if current == new_status:
        return
    allowed = table.get(current, set())
    if new_status not in allowed:
        raise AppHTTPError(400, f"不允许从「{current}」变更为「{new_status}」")


def list_feedbacks_page(
    db: Session,
    *,
    status: Optional[str],
    fb_type: Optional[str],
    keyword: Optional[str],
    page: int,
    page_size: int,
) -> Dict[str, Any]:
    items, total = feedback_crud.list_feedbacks(
        db,
        status=status,
        fb_type=fb_type,
        keyword=keyword,
        page=page,
        page_size=page_size,
    )
    return {
        "list": [feedback_to_dict(r) for r in items],
        "total": total,
        "page": page,
        "pageSize": page_size,
    }


def get_feedback_detail(db: Session, feedback_id: str) -> Dict[str, Any]:
    row = feedback_crud.get_feedback(db, feedback_id)
    if not row:
        raise AppHTTPError(404, "反馈不存在")
    return feedback_to_dict(row)


def update_feedback_status(
    db: Session,
    user: SysUser,
    feedback_id: str,
    *,
    status: str,
    handle_remark: Optional[str] = None,
) -> Dict[str, Any]:
    row = feedback_crud.get_feedback(db, feedback_id)
    if not row:
        raise AppHTTPError(404, "反馈不存在")
    old = row.status
    _assert_transition(old, status, FEEDBACK_TRANSITIONS)
    feedback_crud.update_feedback_status(
        db,
        row,
        status=status,
        handle_remark=handle_remark,
        update_by=user.username,
    )
    oper_log_crud.create_oper_log(
        db,
        module="feedback",
        action="status",
        user_id=user.id,
        username=user.username,
        detail=f"{row.biz_no}: {old}→{status}; {(handle_remark or '')[:120]}",
    )
    return feedback_to_dict(row)


def list_suggestions_page(
    db: Session,
    *,
    status: Optional[str],
    keyword: Optional[str],
    page: int,
    page_size: int,
) -> Dict[str, Any]:
    items, total = suggestion_crud.list_suggestions(
        db, status=status, keyword=keyword, page=page, page_size=page_size
    )
    return {
        "list": [suggestion_to_dict(r) for r in items],
        "total": total,
        "page": page,
        "pageSize": page_size,
    }


def get_suggestion_detail(db: Session, suggestion_id: str) -> Dict[str, Any]:
    row = suggestion_crud.get_suggestion(db, suggestion_id)
    if not row:
        raise AppHTTPError(404, "建言不存在")
    return suggestion_to_dict(row)


def update_suggestion_status(
    db: Session,
    user: SysUser,
    suggestion_id: str,
    *,
    status: str,
    handle_remark: Optional[str] = None,
) -> Dict[str, Any]:
    row = suggestion_crud.get_suggestion(db, suggestion_id)
    if not row:
        raise AppHTTPError(404, "建言不存在")
    old = row.status
    _assert_transition(old, status, SUGGESTION_TRANSITIONS)
    suggestion_crud.update_suggestion_status(
        db,
        row,
        status=status,
        handle_remark=handle_remark,
        update_by=user.username,
    )
    oper_log_crud.create_oper_log(
        db,
        module="suggestion",
        action="status",
        user_id=user.id,
        username=user.username,
        detail=f"{row.biz_no}: {old}→{status}; {(handle_remark or '')[:120]}",
    )
    return suggestion_to_dict(row)