ajyy/backend/app/services/dashboard_service.py

"""工作台统计"""
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],
    }