feat(backend): add admin dashboard, ledger, handle, and config APIs

Expose authenticated admin endpoints for dashboard stats, feedback/suggestion
handling with status transitions and menu permissions, and hotline/qrcode writes.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/admin.py

@@ -0,0 +1,207 @@
+"""管理端 API：工作台、台账办理、配置、附件上传。"""
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Depends, File, Form, Query, UploadFile
+from sqlalchemy.orm import Session
+
+from app.core import storage
+from app.core.database import get_db
+from app.core.deps import get_current_user, require_menu
+from app.core.response import AppHTTPError, success_response
+from app.core.storage import StorageError
+from app.crud import attachment as attachment_crud
+from app.models.system import SysUser
+from app.schemas.business import HotlineUpdateBody, QrcodeUpdateBody, StatusUpdateBody
+from app.services import admin_ledger_service, config_service, dashboard_service
+
+router = APIRouter(prefix="/web/admin", tags=["admin"])
+
+
+@router.get("/dashboard")
+def dashboard(
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+):
+    return success_response(dashboard_service.get_dashboard(db))
+
+
+@router.get("/feedbacks")
+def list_feedbacks(
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+    status: Optional[str] = Query(None),
+    fbType: Optional[str] = Query(None),
+    keyword: Optional[str] = Query(None),
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+):
+    return success_response(
+        admin_ledger_service.list_feedbacks_page(
+            db,
+            status=status,
+            fb_type=fbType,
+            keyword=keyword,
+            page=page,
+            page_size=pageSize,
+        )
+    )
+
+
+@router.get("/feedbacks/{feedback_id}")
+def get_feedback(
+    feedback_id: str,
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+):
+    return success_response(
+        admin_ledger_service.get_feedback_detail(db, feedback_id)
+    )
+
+
+@router.put("/feedbacks/{feedback_id}/status")
+def update_feedback_status(
+    feedback_id: str,
+    body: StatusUpdateBody,
+    db: Annotated[Session, Depends(get_db)],
+    user: Annotated[SysUser, Depends(require_menu("feedback:handle"))],
+):
+    return success_response(
+        admin_ledger_service.update_feedback_status(
+            db,
+            user,
+            feedback_id,
+            status=body.status,
+            handle_remark=body.handle_remark,
+        )
+    )
+
+
+@router.get("/suggestions")
+def list_suggestions(
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+    status: Optional[str] = Query(None),
+    keyword: Optional[str] = Query(None),
+    page: int = Query(1, ge=1),
+    pageSize: int = Query(10, ge=1, le=100),
+):
+    return success_response(
+        admin_ledger_service.list_suggestions_page(
+            db,
+            status=status,
+            keyword=keyword,
+            page=page,
+            page_size=pageSize,
+        )
+    )
+
+
+@router.get("/suggestions/{suggestion_id}")
+def get_suggestion(
+    suggestion_id: str,
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+):
+    return success_response(
+        admin_ledger_service.get_suggestion_detail(db, suggestion_id)
+    )
+
+
+@router.put("/suggestions/{suggestion_id}/status")
+def update_suggestion_status(
+    suggestion_id: str,
+    body: StatusUpdateBody,
+    db: Annotated[Session, Depends(get_db)],
+    user: Annotated[SysUser, Depends(require_menu("suggestion:handle"))],
+):
+    return success_response(
+        admin_ledger_service.update_suggestion_status(
+            db,
+            user,
+            suggestion_id,
+            status=body.status,
+            handle_remark=body.handle_remark,
+        )
+    )
+
+
+@router.get("/hotline")
+def get_hotline(
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+):
+    return success_response(config_service.get_hotline_payload(db))
+
+
+@router.put("/hotline")
+def put_hotline(
+    body: HotlineUpdateBody,
+    db: Annotated[Session, Depends(get_db)],
+    user: Annotated[SysUser, Depends(require_menu("hotline:write"))],
+):
+    return success_response(
+        config_service.update_hotline(
+            db,
+            user,
+            phone=body.phone,
+            email=body.email,
+            address=body.address,
+            work_time=body.work_time,
+        )
+    )
+
+
+@router.get("/qrcode")
+def get_qrcode(
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+):
+    return success_response(config_service.get_qrcode_payload(db))
+
+
+@router.put("/qrcode")
+def put_qrcode(
+    body: QrcodeUpdateBody,
+    db: Annotated[Session, Depends(get_db)],
+    user: Annotated[SysUser, Depends(require_menu("qrcode:write"))],
+):
+    return success_response(
+        config_service.update_qrcode(
+            db,
+            user,
+            land_url=body.land_url,
+            title=body.title,
+            image_url=body.image_url,
+        )
+    )
+
+
+@router.post("/attachment/upload")
+async def admin_upload(
+    db: Annotated[Session, Depends(get_db)],
+    _user: Annotated[SysUser, Depends(get_current_user)],
+    file: UploadFile = File(...),
+    folder: Optional[str] = Form(""),
+):
+    if not file or not file.filename:
+        raise AppHTTPError(400, "请选择文件")
+    content = await file.read()
+    try:
+        stored = storage.upload_file(content, file.filename, folder or "")
+    except StorageError as exc:
+        raise AppHTTPError(400, exc.message) from exc
+    row = attachment_crud.create_attachment(
+        db,
+        file_name=file.filename,
+        object_key=stored["objectKey"],
+        url=stored["url"],
+        size=len(content),
+    )
+    return success_response(
+        {
+            "id": row.id,
+            "url": row.url,
+            "fileName": row.file_name,
+            "size": row.size,
+        }
+    )

backend/app/crud/config.py

@@ -22,3 +22,50 @@ def get_qrcode(db: Session) -> Optional[QrcodeConfig]:
         .order_by(QrcodeConfig.create_time.asc())
         .first()
     )
+
+
+def update_hotline(
+    db: Session,
+    row: HotlineConfig,
+    *,
+    phone: Optional[str] = None,
+    email: Optional[str] = None,
+    address: Optional[str] = None,
+    work_time: Optional[str] = None,
+    update_by: Optional[str] = None,
+) -> HotlineConfig:
+    if phone is not None:
+        row.phone = phone
+    if email is not None:
+        row.email = email
+    if address is not None:
+        row.address = address
+    if work_time is not None:
+        row.work_time = work_time
+    if update_by:
+        row.update_by = update_by
+    db.commit()
+    db.refresh(row)
+    return row
+
+
+def update_qrcode(
+    db: Session,
+    row: QrcodeConfig,
+    *,
+    land_url: Optional[str] = None,
+    title: Optional[str] = None,
+    image_url: Optional[str] = None,
+    update_by: Optional[str] = None,
+) -> QrcodeConfig:
+    if land_url is not None:
+        row.land_url = land_url
+    if title is not None:
+        row.title = title
+    if image_url is not None:
+        row.image_url = image_url
+    if update_by:
+        row.update_by = update_by
+    db.commit()
+    db.refresh(row)
+    return row

backend/app/crud/feedback.py

@@ -1,7 +1,8 @@
 """反馈 CRUD"""
 from datetime import datetime
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
+from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from app.models.business import Feedback
@@ -57,3 +58,59 @@ def create_feedback(
     db.commit()
     db.refresh(row)
     return row
+
+
+def get_feedback(db: Session, feedback_id: str) -> Optional[Feedback]:
+    return (
+        db.query(Feedback)
+        .filter(Feedback.id == feedback_id, Feedback.del_flag == "0")
+        .first()
+    )
+
+
+def list_feedbacks(
+    db: Session,
+    *,
+    status: Optional[str] = None,
+    fb_type: Optional[str] = None,
+    keyword: Optional[str] = None,
+    page: int = 1,
+    page_size: int = 10,
+) -> Tuple[List[Feedback], int]:
+    q = db.query(Feedback).filter(Feedback.del_flag == "0")
+    if status:
+        q = q.filter(Feedback.status == status)
+    if fb_type:
+        q = q.filter(Feedback.fb_type == fb_type)
+    if keyword:
+        like = f"%{keyword}%"
+        q = q.filter(
+            or_(
+                Feedback.content.like(like),
+                Feedback.biz_no.like(like),
+                Feedback.target_name.like(like),
+                Feedback.reporter_name.like(like),
+            )
+        )
+    q = q.order_by(Feedback.create_time.desc())
+    total = q.count()
+    items = q.offset((page - 1) * page_size).limit(page_size).all()
+    return items, total
+
+
+def update_feedback_status(
+    db: Session,
+    row: Feedback,
+    *,
+    status: str,
+    handle_remark: Optional[str] = None,
+    update_by: Optional[str] = None,
+) -> Feedback:
+    row.status = status
+    if handle_remark is not None:
+        row.handle_remark = handle_remark
+    if update_by:
+        row.update_by = update_by
+    db.commit()
+    db.refresh(row)
+    return row

backend/app/crud/oper_log.py

@@ -0,0 +1,41 @@
+"""操作日志 CRUD"""
+import uuid
+from typing import List, Optional, Tuple
+
+from sqlalchemy.orm import Session
+
+from app.models.system import OperLog
+
+
+def create_oper_log(
+    db: Session,
+    *,
+    module: str,
+    action: str,
+    user_id: Optional[str] = None,
+    username: Optional[str] = None,
+    detail: Optional[str] = None,
+    ip: Optional[str] = None,
+) -> OperLog:
+    row = OperLog(
+        id=str(uuid.uuid4()),
+        user_id=user_id,
+        username=username,
+        module=module,
+        action=action,
+        detail=detail,
+        ip=ip,
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return row
+
+
+def list_oper_logs(
+    db: Session, *, page: int = 1, page_size: int = 10
+) -> Tuple[List[OperLog], int]:
+    q = db.query(OperLog).order_by(OperLog.create_time.desc())
+    total = q.count()
+    items = q.offset((page - 1) * page_size).limit(page_size).all()
+    return items, total

backend/app/crud/suggestion.py

@@ -1,7 +1,8 @@
 """建言 CRUD"""
 from datetime import datetime
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
+from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from app.models.business import Suggestion
@@ -53,3 +54,55 @@ def create_suggestion(
     db.commit()
     db.refresh(row)
     return row
+
+
+def get_suggestion(db: Session, suggestion_id: str) -> Optional[Suggestion]:
+    return (
+        db.query(Suggestion)
+        .filter(Suggestion.id == suggestion_id, Suggestion.del_flag == "0")
+        .first()
+    )
+
+
+def list_suggestions(
+    db: Session,
+    *,
+    status: Optional[str] = None,
+    keyword: Optional[str] = None,
+    page: int = 1,
+    page_size: int = 10,
+) -> Tuple[List[Suggestion], int]:
+    q = db.query(Suggestion).filter(Suggestion.del_flag == "0")
+    if status:
+        q = q.filter(Suggestion.status == status)
+    if keyword:
+        like = f"%{keyword}%"
+        q = q.filter(
+            or_(
+                Suggestion.content.like(like),
+                Suggestion.biz_no.like(like),
+                Suggestion.contact_name.like(like),
+            )
+        )
+    q = q.order_by(Suggestion.create_time.desc())
+    total = q.count()
+    items = q.offset((page - 1) * page_size).limit(page_size).all()
+    return items, total
+
+
+def update_suggestion_status(
+    db: Session,
+    row: Suggestion,
+    *,
+    status: str,
+    handle_remark: Optional[str] = None,
+    update_by: Optional[str] = None,
+) -> Suggestion:
+    row.status = status
+    if handle_remark is not None:
+        row.handle_remark = handle_remark
+    if update_by:
+        row.update_by = update_by
+    db.commit()
+    db.refresh(row)
+    return row

backend/app/crud/user.py

@@ -30,6 +30,37 @@ def get_role_by_id(db: Session, role_id: str) -> Optional[SysRole]:
     )
 
 
+def get_role_by_code(db: Session, role_code: str) -> Optional[SysRole]:
+    return (
+        db.query(SysRole)
+        .filter(SysRole.role_code == role_code, SysRole.del_flag == "0")
+        .first()
+    )
+
+
+def create_user(
+    db: Session,
+    *,
+    username: str,
+    password_hash: str,
+    display_name: str,
+    role_id: str,
+    status: str = "0",
+) -> SysUser:
+    row = SysUser(
+        username=username,
+        password_hash=password_hash,
+        display_name=display_name,
+        role_id=role_id,
+        status=status,
+        del_flag="0",
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return row
+
+
 def get_user_with_role(
     db: Session, user_id: str
 ) -> Tuple[Optional[SysUser], Optional[SysRole]]:

backend/app/main.py

@@ -8,6 +8,7 @@ from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 
 from app.api.auth import router as auth_router
+from app.api.admin import router as admin_router
 from app.api.public import router as public_router
 from app.core.config import get_settings
 from app.core.database import SessionLocal, init_db
@@ -45,6 +46,7 @@ app.add_middleware(
 
 app.include_router(auth_router, prefix=settings.API_V1_PREFIX)
 app.include_router(public_router, prefix=settings.API_V1_PREFIX)
+app.include_router(admin_router, prefix=settings.API_V1_PREFIX)
 
 _upload_dir = Path(settings.LOCAL_STORAGE_DIR)
 _upload_dir.mkdir(parents=True, exist_ok=True)

backend/app/schemas/business.py

@@ -34,3 +34,21 @@ class SuggestionCreateBody(CamelModel):
     contact_post: Optional[str] = None
     contact_phone: Optional[str] = None
     attachment_urls: Optional[List[str]] = None
+
+
+class StatusUpdateBody(CamelModel):
+    status: str
+    handle_remark: Optional[str] = None
+
+
+class HotlineUpdateBody(CamelModel):
+    phone: Optional[str] = None
+    email: Optional[str] = None
+    address: Optional[str] = None
+    work_time: Optional[str] = None
+
+
+class QrcodeUpdateBody(CamelModel):
+    land_url: Optional[str] = None
+    title: Optional[str] = None
+    image_url: Optional[str] = None

backend/app/services/admin_ledger_service.py

@@ -0,0 +1,195 @@
+"""管理端反馈 / 建言办理"""
+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,
+        "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)

backend/app/services/config_service.py

@@ -1,10 +1,11 @@
-"""热线 / 二维码公开读取"""
-from typing import Any, Dict
+"""热线 / 二维码配置读写"""
+from typing import Any, Dict, 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
 
 
 def get_hotline_payload(db: Session) -> Dict[str, Any]:
@@ -30,3 +31,65 @@ def get_qrcode_payload(db: Session) -> Dict[str, Any]:
         "title": row.title or "",
         "imageUrl": row.image_url or "",
     }
+
+
+def update_hotline(
+    db: Session,
+    user,
+    *,
+    phone: Optional[str] = None,
+    email: Optional[str] = None,
+    address: Optional[str] = None,
+    work_time: Optional[str] = None,
+) -> Dict[str, Any]:
+    row = config_crud.get_hotline(db)
+    if not row:
+        raise AppHTTPError(404, "热线配置不存在")
+    config_crud.update_hotline(
+        db,
+        row,
+        phone=phone,
+        email=email,
+        address=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)

backend/app/services/dashboard_service.py

@@ -0,0 +1,59 @@
+"""工作台统计"""
+from datetime import datetime, timedelta
+from typing import Any, Dict, List
+
+from sqlalchemy.orm import Session
+
+from app.crud import feedback as feedback_crud
+from app.models.business import Feedback, Suggestion
+
+
+def _feedback_summary(row: Feedback) -> Dict[str, Any]:
+    content = row.content or ""
+    return {
+        "id": row.id,
+        "bizNo": row.biz_no,
+        "fbType": row.fb_type,
+        "content": content[:80],
+        "status": row.status,
+        "createTime": row.create_time.isoformat(sep=" ", timespec="seconds")
+        if row.create_time
+        else None,
+    }
+
+
+def get_dashboard(db: Session) -> Dict[str, Any]:
+    feedback_total = (
+        db.query(Feedback).filter(Feedback.del_flag == "0").count()
+    )
+    complaint_pending = (
+        db.query(Feedback)
+        .filter(
+            Feedback.del_flag == "0",
+            Feedback.fb_type == "投诉",
+            Feedback.status.in_(["待处理", "处理中"]),
+        )
+        .count()
+    )
+    suggestion_total = (
+        db.query(Suggestion).filter(Suggestion.del_flag == "0").count()
+    )
+    week_ago = datetime.utcnow() - timedelta(days=7)
+    week_fb = (
+        db.query(Feedback)
+        .filter(Feedback.del_flag == "0", Feedback.create_time >= week_ago)
+        .count()
+    )
+    week_sg = (
+        db.query(Suggestion)
+        .filter(Suggestion.del_flag == "0", Suggestion.create_time >= week_ago)
+        .count()
+    )
+    recent, _ = feedback_crud.list_feedbacks(db, page=1, page_size=5)
+    return {
+        "feedbackTotal": feedback_total,
+        "complaintPending": complaint_pending,
+        "suggestionTotal": suggestion_total,
+        "weekNew": week_fb + week_sg,
+        "recentFeedback": [_feedback_summary(r) for r in recent],
+    }

backend/tests/test_admin_feedback.py

@@ -0,0 +1,157 @@
+"""管理端台账 / 办理 / 配置 API 测试"""
+import pytest
+
+from app.core.security import hash_password
+from app.crud import user as user_crud
+from app.main import app as fastapi_app
+from app.core.database import get_db
+
+
+def _login(client, username: str, password: str) -> dict:
+    r = client.post(
+        "/api/web/auth/login", json={"username": username, "password": password}
+    )
+    assert r.status_code == 200, r.text
+    token = r.json()["data"]["token"]
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _ensure_role_user(client, username: str, role_code: str, password: str = "Pass@123"):
+    """在 client fixture 共享的 session 上插入角色用户。"""
+    # 取出 override 的 session
+    gen = fastapi_app.dependency_overrides[get_db]()
+    db = next(gen)
+    role = user_crud.get_role_by_code(db, role_code)
+    assert role is not None
+    existing = user_crud.get_user_by_username(db, username)
+    if not existing:
+        user_crud.create_user(
+            db,
+            username=username,
+            password_hash=hash_password(password),
+            display_name=username,
+            role_id=role.id,
+        )
+    return _login(client, username, password)
+
+
+@pytest.fixture()
+def admin_headers(client):
+    return _login(client, "admin", "Admin@123")
+
+
+@pytest.fixture()
+def handler_headers(client):
+    return _ensure_role_user(client, "handler1", "handler")
+
+
+@pytest.fixture()
+def viewer_headers(client):
+    return _ensure_role_user(client, "viewer1", "viewer")
+
+
+def test_viewer_cannot_handle(client, viewer_headers):
+    fb = client.post(
+        "/api/public/feedback", json={"fbType": "投诉", "content": "xxxx"}
+    ).json()["data"]
+    r = client.put(
+        f"/api/web/admin/feedbacks/{fb['id']}/status",
+        headers=viewer_headers,
+        json={"status": "处理中"},
+    )
+    assert r.status_code == 403
+    assert r.json()["code"] == 403
+
+
+def test_handler_can_handle_feedback(client, handler_headers):
+    fb = client.post(
+        "/api/public/feedback", json={"fbType": "投诉", "content": "需要办理的内容"}
+    ).json()["data"]
+    r = client.put(
+        f"/api/web/admin/feedbacks/{fb['id']}/status",
+        headers=handler_headers,
+        json={"status": "处理中", "handleRemark": "已接单"},
+    )
+    assert r.status_code == 200
+    assert r.json()["data"]["status"] == "处理中"
+    assert r.json()["data"]["handleRemark"] == "已接单"
+
+
+def test_invalid_feedback_transition(client, handler_headers):
+    fb = client.post(
+        "/api/public/feedback", json={"fbType": "表扬", "content": "好评内容测试"}
+    ).json()["data"]
+    # 待处理不能直接回到非法状态（例如空跳转到不存在状态）
+    r = client.put(
+        f"/api/web/admin/feedbacks/{fb['id']}/status",
+        headers=handler_headers,
+        json={"status": "已办结"},
+    )
+    assert r.status_code == 200  # 待处理→已办结 允许
+    r2 = client.put(
+        f"/api/web/admin/feedbacks/{fb['id']}/status",
+        headers=handler_headers,
+        json={"status": "待处理"},
+    )
+    assert r2.status_code == 400
+
+
+def test_handler_cannot_update_hotline(client, handler_headers):
+    r = client.put(
+        "/api/web/admin/hotline",
+        headers=handler_headers,
+        json={"phone": "10086"},
+    )
+    assert r.status_code == 403
+
+
+def test_admin_update_hotline(client, admin_headers):
+    r = client.put(
+        "/api/web/admin/hotline",
+        headers=admin_headers,
+        json={"phone": "0551-88888888", "workTime": "工作日 9-17"},
+    )
+    assert r.status_code == 200
+    assert r.json()["data"]["phone"] == "0551-88888888"
+    pub = client.get("/api/public/hotline")
+    assert pub.json()["data"]["phone"] == "0551-88888888"
+
+
+def test_dashboard(client, admin_headers):
+    client.post(
+        "/api/public/feedback", json={"fbType": "投诉", "content": "仪表盘反馈一条"}
+    )
+    r = client.get("/api/web/admin/dashboard", headers=admin_headers)
+    assert r.status_code == 200
+    data = r.json()["data"]
+    assert data["feedbackTotal"] >= 1
+    assert "recentFeedback" in data
+    assert "complaintPending" in data
+
+
+def test_list_feedbacks(client, admin_headers):
+    client.post(
+        "/api/public/feedback", json={"fbType": "投诉", "content": "列表筛选关键字甲"}
+    )
+    r = client.get(
+        "/api/web/admin/feedbacks",
+        headers=admin_headers,
+        params={"keyword": "关键字甲", "page": 1, "pageSize": 10},
+    )
+    assert r.status_code == 200
+    data = r.json()["data"]
+    assert data["total"] >= 1
+    assert data["list"][0]["content"].find("关键字甲") >= 0
+
+
+def test_suggestion_handle(client, handler_headers):
+    sg = client.post(
+        "/api/public/suggestion", json={"content": "一条建言内容足够长"}
+    ).json()["data"]
+    r = client.put(
+        f"/api/web/admin/suggestions/{sg['id']}/status",
+        headers=handler_headers,
+        json={"status": "已采纳"},
+    )
+    assert r.status_code == 200
+    assert r.json()["data"]["status"] == "已采纳"