ajyy/backend/app/crud/feedback.py

"""反馈 CRUD"""
from datetime import datetime
from typing import List, Optional, Tuple

from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.models.business import Feedback


def next_biz_no(db: Session, prefix: str = "FK") -> str:
    """生成业务编号：前缀 + YYYYMMDD + 3 位日序号。"""
    date_part = datetime.now().strftime("%Y%m%d")
    head = f"{prefix}{date_part}"
    last = (
        db.query(Feedback)
        .filter(Feedback.biz_no.like(f"{head}%"))
        .order_by(Feedback.biz_no.desc())
        .first()
    )
    seq = 1
    if last and last.biz_no and len(last.biz_no) >= 3:
        try:
            seq = int(last.biz_no[-3:]) + 1
        except ValueError:
            seq = 1
    return f"{head}{seq:03d}"


def create_feedback(
    db: Session,
    *,
    biz_no: str,
    fb_type: str,
    content: str,
    target_name: Optional[str] = None,
    target_dept: Optional[str] = None,
    target_post: Optional[str] = None,
    reporter_name: Optional[str] = None,
    reporter_unit: Optional[str] = None,
    reporter_post: Optional[str] = None,
    reporter_phone: Optional[str] = None,
    attachments: Optional[List[str]] = None,
    status: str = "待处理",
) -> Feedback:
    row = Feedback(
        biz_no=biz_no,
        fb_type=fb_type,
        content=content,
        target_name=target_name,
        target_dept=target_dept,
        target_post=target_post,
        reporter_name=reporter_name,
        reporter_unit=reporter_unit,
        reporter_post=reporter_post,
        reporter_phone=reporter_phone,
        attachments=list(attachments or []),
        status=status,
        del_flag="0",
    )
    db.add(row)
    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