webapp/app/services/vas_task_service.py

# app/services/task_service.py
from sqlalchemy.orm import Session
from typing import List
from app.utils.search import apply_keyword_search
from app.utils.pagination import paginate
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.vas_task import VasTask
from app.schemas.vas_task import VasTaskCreate, VasTaskUpdate
from datetime import datetime

class VasTaskService:

    def create(db: Session, data: VasTaskCreate):
        rec = VasTask(**data.dict(), status='pending', created_at=datetime.utcnow())
        db.add(rec)
        db.commit()
        db.refresh(rec)
        return rec

    def list_task(
        db: Session,
        status: str = None,
        routing_key: str = None,
        script_version: str = None,
        keyword: str = None,
        page: int = 0,
        size: int = 10,
    ):
        query = db.query(VasTask)
        
        if status:
            query = query.filter(VasTask.status == status)
        
        if routing_key:
            query = query.filter(VasTask.routing_key == routing_key)

        if script_version:
            query = query.filter(VasTask.script_version == script_version)

        query = apply_keyword_search(
            query=query,
            model=VasTask,
            keyword=keyword,
            fields=["order_id", "routing_key", "user_inputs"]
        )
        
        query = query.order_by(
            VasTask.priority.desc(),
            VasTask.id.asc()
        )
        return paginate(query, page, size)
    
    def update(db: Session, id: int, payload: VasTaskUpdate):
        obj = db.query(VasTask).filter(VasTask.id == id).first()
        if not obj:
            raise NotFoundError("Task not exist")
        data = payload.dict(exclude_unset=True)  # ⭐ 关键

        for key, value in data.items():
            setattr(obj, key, value)
        db.commit()
        db.refresh(obj)
        return obj
        
    def get_active_task_by_order_id(db: Session, order_id:str):
        recs = db.query(VasTask).filter(
            VasTask.status == "pending",
            VasTask.order_id == order_id
        ).all()
        return recs
    
    def return_to_queue(db: Session, id:int):
        rec = db.query(VasTask).filter_by(id=id).first()
        if not rec:
            raise NotFoundError("Task not exist")
        rec.status = 'pending'
        db.commit()
        db.refresh(rec)
        return rec

    def manual_confirm(db: Session, id:int):
        rec = db.query(VasTask).filter_by(id=id).first()
        if not rec:
            raise NotFoundError("Task not exist")
        rec.status = 'completed'
        db.commit()
        db.refresh(rec)
        return rec