Hartmut
24833ce52e
create_order and add_order_line endpoints were not passing material_override from the request body to the OrderLine constructor. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1573 lines
55 KiB
Python
1573 lines
55 KiB
Python
import io
|
||
import logging
|
||
import os
|
||
import re
|
||
import uuid
|
||
import zipfile
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, func, update
|
||
from sqlalchemy.orm import selectinload
|
||
|
||
from app.database import get_db
|
||
from app.models.order import Order, OrderStatus
|
||
from app.models.order_item import OrderItem
|
||
from app.models.order_line import OrderLine
|
||
from app.models.product import Product
|
||
from app.models.output_type import OutputType
|
||
from app.models.cad_file import CadFile
|
||
from app.models.user import User
|
||
from app.schemas.order import OrderCreate, OrderOut, OrderDetailOut, OrderItemOut, RejectOrderRequest
|
||
from app.schemas.order_line import OrderLineCreate, OrderLineOut
|
||
from app.schemas.product import ProductOut
|
||
from app.schemas.output_type import OutputTypeOut
|
||
from app.services.order_service import generate_order_number
|
||
from app.utils.auth import get_current_user, require_admin_or_pm, require_pm_or_above
|
||
|
||
router = APIRouter(prefix="/orders", tags=["orders"])
|
||
|
||
|
||
def _is_privileged(user: User) -> bool:
|
||
return user.role.value in ("admin", "project_manager")
|
||
|
||
|
||
def _result_path_to_url(result_path: str) -> str | None:
|
||
"""Convert an internal result_path to a servable static URL."""
|
||
if "/renders/" in result_path:
|
||
idx = result_path.index("/renders/")
|
||
return result_path[idx:]
|
||
if "/thumbnails/" in result_path:
|
||
idx = result_path.index("/thumbnails/")
|
||
return result_path[idx:]
|
||
return None
|
||
|
||
|
||
def _build_line_out(line: OrderLine) -> OrderLineOut:
|
||
product_out = ProductOut.model_validate(line.product)
|
||
product_out.thumbnail_url = line.product.thumbnail_url
|
||
product_out.processing_status = line.product.processing_status
|
||
|
||
# Prefer completed render over CAD thumbnail
|
||
thumb = line.product.thumbnail_url
|
||
if line.render_status == "completed" and line.result_path:
|
||
render_url = _result_path_to_url(line.result_path)
|
||
if render_url:
|
||
thumb = render_url
|
||
|
||
# Build OutputTypeOut with pricing convenience fields
|
||
ot_out = None
|
||
if line.output_type:
|
||
ot_out = OutputTypeOut.model_validate(line.output_type)
|
||
if hasattr(line.output_type, 'pricing_tier') and line.output_type.pricing_tier:
|
||
pt = line.output_type.pricing_tier
|
||
ot_out.pricing_tier_name = f"{pt.category_key}/{pt.quality_level}"
|
||
ot_out.price_per_item = float(pt.price_per_item)
|
||
|
||
rp_name: str | None = None
|
||
if hasattr(line, 'render_position') and line.render_position:
|
||
rp_name = line.render_position.name
|
||
|
||
out = OrderLineOut(
|
||
id=line.id,
|
||
order_id=line.order_id,
|
||
product_id=line.product_id,
|
||
product=product_out,
|
||
output_type_id=line.output_type_id,
|
||
output_type=ot_out,
|
||
gewuenschte_bildnummer=line.gewuenschte_bildnummer,
|
||
item_status=line.item_status,
|
||
render_status=line.render_status,
|
||
result_path=line.result_path,
|
||
thumbnail_url=thumb,
|
||
ai_validation_status=line.ai_validation_status,
|
||
ai_validation_result=line.ai_validation_result,
|
||
render_backend_used=line.render_backend_used,
|
||
flamenco_job_id=line.flamenco_job_id,
|
||
unit_price=float(line.unit_price) if line.unit_price is not None else None,
|
||
render_position_id=line.render_position_id,
|
||
render_position_name=rp_name,
|
||
render_log=line.render_log if hasattr(line, 'render_log') else None,
|
||
render_started_at=line.render_started_at if hasattr(line, 'render_started_at') else None,
|
||
render_completed_at=line.render_completed_at if hasattr(line, 'render_completed_at') else None,
|
||
material_override=getattr(line, 'material_override', None),
|
||
notes=line.notes,
|
||
created_at=line.created_at,
|
||
updated_at=line.updated_at,
|
||
)
|
||
return out
|
||
|
||
|
||
async def _load_order_detail(db, order_id: uuid.UUID) -> Order:
|
||
from app.models.output_type import OutputType as OTModel
|
||
from app.models.render_position import ProductRenderPosition
|
||
result = await db.execute(
|
||
select(Order)
|
||
.where(Order.id == order_id)
|
||
.options(
|
||
selectinload(Order.items).selectinload(OrderItem.cad_file),
|
||
selectinload(Order.lines)
|
||
.selectinload(OrderLine.product)
|
||
.selectinload(Product.cad_file),
|
||
selectinload(Order.lines)
|
||
.selectinload(OrderLine.product)
|
||
.selectinload(Product.render_positions),
|
||
selectinload(Order.lines)
|
||
.selectinload(OrderLine.output_type)
|
||
.selectinload(OTModel.pricing_tier),
|
||
selectinload(Order.lines)
|
||
.selectinload(OrderLine.render_position),
|
||
)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
|
||
def _compute_render_progress(lines) -> dict | None:
|
||
"""Compute render progress from order lines that have an output_type."""
|
||
renderable = [l for l in lines if l.output_type_id is not None]
|
||
if not renderable:
|
||
return None
|
||
progress = {"total": len(renderable), "completed": 0, "processing": 0, "failed": 0, "pending": 0, "cancelled": 0}
|
||
for l in renderable:
|
||
status = l.render_status or "pending"
|
||
if status in progress:
|
||
progress[status] += 1
|
||
else:
|
||
progress["pending"] += 1
|
||
return progress
|
||
|
||
|
||
async def _maybe_complete_order(db: AsyncSession, order_id: uuid.UUID):
|
||
"""If all renderable lines are terminal, auto-advance order to completed."""
|
||
lines_result = await db.execute(
|
||
select(OrderLine).where(
|
||
OrderLine.order_id == order_id,
|
||
OrderLine.output_type_id.isnot(None),
|
||
)
|
||
)
|
||
lines = lines_result.scalars().all()
|
||
if not lines:
|
||
return
|
||
all_terminal = all(
|
||
l.render_status in ("completed", "failed", "cancelled")
|
||
for l in lines
|
||
)
|
||
if not all_terminal:
|
||
return
|
||
order_result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = order_result.scalar_one_or_none()
|
||
if order and order.status == OrderStatus.processing:
|
||
order.status = OrderStatus.completed
|
||
order.completed_at = datetime.utcnow()
|
||
order.updated_at = datetime.utcnow()
|
||
await db.commit()
|
||
|
||
|
||
def _order_detail_out(order: Order) -> OrderDetailOut:
|
||
out = OrderDetailOut.model_validate(order)
|
||
out.item_count = len(order.items)
|
||
out.items = [OrderItemOut.model_validate(i) for i in order.items]
|
||
out.line_count = len(order.lines)
|
||
out.lines = [_build_line_out(line) for line in order.lines]
|
||
out.render_progress = _compute_render_progress(order.lines)
|
||
return out
|
||
|
||
|
||
@router.get("/search", response_model=list[OrderDetailOut])
|
||
async def search_orders(
|
||
q: str = Query(""),
|
||
statuses: str = Query(""), # comma-separated: "draft,submitted"
|
||
date_from: str = Query(""),
|
||
date_to: str = Query(""),
|
||
limit: int = Query(50, le=200),
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Full-text search across orders and their items."""
|
||
from sqlalchemy import or_
|
||
|
||
# Parse and validate status list
|
||
valid_statuses = []
|
||
for s in (s.strip() for s in statuses.split(",") if s.strip()):
|
||
try:
|
||
valid_statuses.append(OrderStatus(s))
|
||
except ValueError:
|
||
pass
|
||
|
||
# Eagerly load items + cad_file + lines to avoid lazy-load issues during Pydantic serialisation
|
||
from app.models.output_type import OutputType as OTModel
|
||
order_q = (
|
||
select(Order)
|
||
.options(
|
||
selectinload(Order.items).selectinload(OrderItem.cad_file),
|
||
selectinload(Order.lines).selectinload(OrderLine.product).selectinload(Product.cad_file),
|
||
selectinload(Order.lines).selectinload(OrderLine.product).selectinload(Product.render_positions),
|
||
selectinload(Order.lines).selectinload(OrderLine.output_type).selectinload(OTModel.pricing_tier),
|
||
)
|
||
)
|
||
if not _is_privileged(user):
|
||
order_q = order_q.where(Order.created_by == user.id)
|
||
if valid_statuses:
|
||
order_q = order_q.where(Order.status.in_(valid_statuses))
|
||
if date_from:
|
||
order_q = order_q.where(Order.created_at >= date_from)
|
||
if date_to:
|
||
order_q = order_q.where(Order.created_at <= date_to + "T23:59:59")
|
||
|
||
if q:
|
||
pattern = f"%{q}%"
|
||
item_fields_cols = [
|
||
OrderItem.ebene1, OrderItem.ebene2, OrderItem.baureihe,
|
||
OrderItem.pim_id, OrderItem.produkt_baureihe, OrderItem.gewaehltes_produkt,
|
||
OrderItem.name_cad_modell, OrderItem.lagertyp, OrderItem.notes,
|
||
]
|
||
item_match = or_(*(f.ilike(pattern) for f in item_fields_cols))
|
||
order_match = or_(Order.order_number.ilike(pattern), Order.notes.ilike(pattern))
|
||
|
||
matching_via_items = select(OrderItem.order_id).where(item_match)
|
||
matching_direct = select(Order.id).where(order_match)
|
||
order_q = order_q.where(
|
||
or_(Order.id.in_(matching_via_items), Order.id.in_(matching_direct))
|
||
)
|
||
|
||
order_q = order_q.order_by(Order.updated_at.desc()).limit(limit)
|
||
result = await db.execute(order_q)
|
||
orders = result.scalars().all()
|
||
|
||
# Text fields used for Python-side item filtering
|
||
_item_text_attrs = [
|
||
'ebene1', 'ebene2', 'baureihe', 'pim_id', 'produkt_baureihe',
|
||
'gewaehltes_produkt', 'name_cad_modell', 'lagertyp', 'notes',
|
||
]
|
||
|
||
out = []
|
||
for order in orders:
|
||
if q:
|
||
q_lower = q.lower()
|
||
order_direct = (
|
||
(order.order_number and q_lower in order.order_number.lower())
|
||
or (order.notes and q_lower in order.notes.lower())
|
||
)
|
||
if order_direct:
|
||
items = list(order.items)
|
||
else:
|
||
items = [
|
||
i for i in order.items
|
||
if any(
|
||
getattr(i, attr) and q_lower in getattr(i, attr).lower()
|
||
for attr in _item_text_attrs
|
||
)
|
||
]
|
||
else:
|
||
items = list(order.items)
|
||
|
||
d = OrderDetailOut.model_validate(order)
|
||
d.item_count = len(items)
|
||
d.items = [OrderItemOut.model_validate(i) for i in items]
|
||
d.line_count = len(order.lines)
|
||
d.lines = [_build_line_out(line) for line in order.lines]
|
||
d.render_progress = _compute_render_progress(order.lines)
|
||
out.append(d)
|
||
|
||
return out
|
||
|
||
|
||
@router.get("", response_model=list[OrderOut])
|
||
async def list_orders(
|
||
status: Optional[OrderStatus] = None,
|
||
template_id: Optional[uuid.UUID] = None,
|
||
skip: int = Query(0, ge=0),
|
||
limit: int = Query(50, ge=1, le=200),
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
q = select(Order)
|
||
if not _is_privileged(user):
|
||
q = q.where(Order.created_by == user.id)
|
||
if status:
|
||
q = q.where(Order.status == status)
|
||
if template_id:
|
||
q = q.where(Order.template_id == template_id)
|
||
q = q.order_by(Order.created_at.desc()).offset(skip).limit(limit)
|
||
result = await db.execute(q)
|
||
orders = result.scalars().all()
|
||
|
||
# Attach item_count, line_count, and render_progress
|
||
out = []
|
||
for order in orders:
|
||
cnt_result = await db.execute(
|
||
select(func.count(OrderItem.id)).where(OrderItem.order_id == order.id)
|
||
)
|
||
cnt = cnt_result.scalar() or 0
|
||
line_cnt_result = await db.execute(
|
||
select(func.count(OrderLine.id)).where(OrderLine.order_id == order.id)
|
||
)
|
||
line_cnt = line_cnt_result.scalar() or 0
|
||
|
||
# Compute render progress for renderable lines
|
||
rp_result = await db.execute(
|
||
select(OrderLine.render_status, func.count(OrderLine.id))
|
||
.where(
|
||
OrderLine.order_id == order.id,
|
||
OrderLine.output_type_id.isnot(None),
|
||
)
|
||
.group_by(OrderLine.render_status)
|
||
)
|
||
rp_rows = rp_result.all()
|
||
render_progress = None
|
||
if rp_rows:
|
||
render_progress = {"total": 0, "completed": 0, "processing": 0, "failed": 0, "pending": 0, "cancelled": 0}
|
||
for rs, count in rp_rows:
|
||
s = rs or "pending"
|
||
if s in render_progress:
|
||
render_progress[s] += count
|
||
else:
|
||
render_progress["pending"] += count
|
||
render_progress["total"] += count
|
||
|
||
d = OrderOut.model_validate(order)
|
||
d.item_count = cnt
|
||
d.line_count = line_cnt
|
||
d.render_progress = render_progress
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
@router.post("", response_model=OrderDetailOut, status_code=status.HTTP_201_CREATED)
|
||
async def create_order(
|
||
body: OrderCreate,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
order_number = await generate_order_number(db)
|
||
order = Order(
|
||
order_number=order_number,
|
||
template_id=body.template_id,
|
||
created_by=user.id,
|
||
source_excel=body.source_excel,
|
||
notes=body.notes,
|
||
tenant_id=getattr(user, 'tenant_id', None),
|
||
)
|
||
db.add(order)
|
||
await db.flush()
|
||
|
||
for item_data in body.items:
|
||
item = OrderItem(
|
||
order_id=order.id,
|
||
row_index=item_data.row_index,
|
||
ebene1=item_data.ebene1,
|
||
ebene2=item_data.ebene2,
|
||
baureihe=item_data.baureihe,
|
||
pim_id=item_data.pim_id,
|
||
produkt_baureihe=item_data.produkt_baureihe,
|
||
gewaehltes_produkt=item_data.gewaehltes_produkt,
|
||
name_cad_modell=item_data.name_cad_modell,
|
||
gewuenschte_bildnummer=item_data.gewuenschte_bildnummer,
|
||
lagertyp=item_data.lagertyp,
|
||
medias_rendering=item_data.medias_rendering,
|
||
components=[c.model_dump() for c in item_data.components],
|
||
tenant_id=getattr(user, 'tenant_id', None),
|
||
)
|
||
db.add(item)
|
||
|
||
for line_data in body.lines:
|
||
# Verify product exists
|
||
prod_result = await db.execute(
|
||
select(Product).where(Product.id == line_data.product_id)
|
||
)
|
||
if not prod_result.scalar_one_or_none():
|
||
raise HTTPException(404, detail=f"Product {line_data.product_id} not found")
|
||
line = OrderLine(
|
||
order_id=order.id,
|
||
product_id=line_data.product_id,
|
||
output_type_id=line_data.output_type_id,
|
||
render_position_id=line_data.render_position_id,
|
||
global_render_position_id=line_data.global_render_position_id,
|
||
gewuenschte_bildnummer=line_data.gewuenschte_bildnummer,
|
||
material_override=line_data.material_override,
|
||
notes=line_data.notes,
|
||
tenant_id=getattr(user, 'tenant_id', None),
|
||
)
|
||
db.add(line)
|
||
|
||
await db.commit()
|
||
|
||
order_loaded = await _load_order_detail(db, order.id)
|
||
return _order_detail_out(order_loaded)
|
||
|
||
|
||
@router.get("/{order_id}", response_model=OrderDetailOut)
|
||
async def get_order(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
order = await _load_order_detail(db, order_id)
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
|
||
return _order_detail_out(order)
|
||
|
||
|
||
@router.post("/{order_id}/submit", response_model=OrderOut)
|
||
async def submit_order(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
if order.status != OrderStatus.draft:
|
||
raise HTTPException(400, detail=f"Order is already {order.status.value}")
|
||
|
||
# Require legacy items marked for rendering to have a linked STEP file
|
||
items_result = await db.execute(
|
||
select(OrderItem).where(OrderItem.order_id == order_id)
|
||
)
|
||
items = items_result.scalars().all()
|
||
missing_items = [
|
||
i.name_cad_modell or f"row {i.row_index}"
|
||
for i in items
|
||
if i.medias_rendering and i.cad_file_id is None
|
||
]
|
||
# Require order_lines with output_type_id to have a product with a CAD file
|
||
lines_result = await db.execute(
|
||
select(OrderLine)
|
||
.options(selectinload(OrderLine.product))
|
||
.where(
|
||
OrderLine.order_id == order_id,
|
||
OrderLine.output_type_id.is_not(None),
|
||
)
|
||
)
|
||
lines = lines_result.scalars().all()
|
||
missing_lines = [
|
||
line.product.name or str(line.product.pim_id)
|
||
for line in lines
|
||
if line.product.cad_file_id is None
|
||
]
|
||
missing = missing_items + missing_lines
|
||
if missing:
|
||
raise HTTPException(
|
||
400,
|
||
detail=f"Cannot submit: {len(missing)} rendering item(s) are missing a STEP file: {', '.join(missing[:5])}{'…' if len(missing) > 5 else ''}",
|
||
)
|
||
|
||
order.status = OrderStatus.submitted
|
||
order.submitted_at = datetime.utcnow()
|
||
order.updated_at = datetime.utcnow()
|
||
|
||
# Auto-approve order_lines when submitted (new Product Library workflow
|
||
# has no per-item approval step — submission implies approval)
|
||
await db.execute(
|
||
update(OrderLine)
|
||
.where(OrderLine.order_id == order.id, OrderLine.item_status == "pending")
|
||
.values(item_status="approved")
|
||
)
|
||
|
||
await db.commit()
|
||
await db.refresh(order)
|
||
|
||
# Notify admins/PMs about new submission (broadcast)
|
||
from app.services.notification_service import emit_notification
|
||
await emit_notification(
|
||
db,
|
||
actor_user_id=user.id,
|
||
target_user_id=None,
|
||
action="order.submitted",
|
||
entity_type="order",
|
||
entity_id=str(order.id),
|
||
details={"order_number": order.order_number},
|
||
)
|
||
|
||
# Compute estimated price after commit (pricing_service opens its own transaction)
|
||
from app.services.pricing_service import refresh_order_price
|
||
await refresh_order_price(db, order.id)
|
||
await db.refresh(order)
|
||
|
||
# Broadcast WebSocket event for live UI updates
|
||
try:
|
||
from app.core.websocket import manager as _ws_mgr
|
||
_tid = str(user.tenant_id) if user.tenant_id else None
|
||
if _tid:
|
||
await _ws_mgr.broadcast_to_tenant(_tid, {
|
||
"type": "order_status_change",
|
||
"order_id": str(order.id),
|
||
"status": "submitted",
|
||
})
|
||
except Exception:
|
||
pass
|
||
|
||
return order
|
||
|
||
|
||
@router.delete("/{order_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
async def delete_order(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
if order.status not in (OrderStatus.draft, OrderStatus.submitted, OrderStatus.rejected):
|
||
raise HTTPException(400, detail="Only draft, submitted or rejected orders can be deleted")
|
||
|
||
await db.delete(order)
|
||
await db.commit()
|
||
|
||
|
||
class SplitMissingStepResponse(BaseModel):
|
||
new_order_id: str
|
||
new_order_number: str
|
||
moved_item_count: int
|
||
moved_line_count: int
|
||
|
||
|
||
@router.post("/{order_id}/split-missing-step", response_model=SplitMissingStepResponse)
|
||
async def split_missing_step(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Move all items/lines that block submission (no STEP file) to a new draft order.
|
||
|
||
After this call the original order can be submitted immediately.
|
||
"""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
if order.status != OrderStatus.draft:
|
||
raise HTTPException(400, detail="Only draft orders can be split")
|
||
|
||
# Find legacy OrderItems blocking submission (rendering, no STEP linked)
|
||
items_result = await db.execute(
|
||
select(OrderItem).where(OrderItem.order_id == order_id)
|
||
)
|
||
items_to_move = [
|
||
i for i in items_result.scalars().all()
|
||
if i.medias_rendering and i.cad_file_id is None
|
||
]
|
||
|
||
# Find OrderLines blocking submission (has output type, product has no STEP)
|
||
lines_result = await db.execute(
|
||
select(OrderLine)
|
||
.options(selectinload(OrderLine.product))
|
||
.where(
|
||
OrderLine.order_id == order_id,
|
||
OrderLine.output_type_id.is_not(None),
|
||
)
|
||
)
|
||
lines_to_move = [
|
||
ln for ln in lines_result.scalars().all()
|
||
if ln.product.cad_file_id is None
|
||
]
|
||
|
||
if not items_to_move and not lines_to_move:
|
||
raise HTTPException(400, detail="No items without STEP file found — nothing to split")
|
||
|
||
# Create the new draft order
|
||
new_order_number = await generate_order_number(db)
|
||
new_order = Order(
|
||
order_number=new_order_number,
|
||
template_id=order.template_id,
|
||
created_by=order.created_by,
|
||
source_excel=order.source_excel,
|
||
notes=f"Split from {order.order_number} — awaiting STEP files",
|
||
tenant_id=order.tenant_id,
|
||
)
|
||
db.add(new_order)
|
||
await db.flush()
|
||
|
||
# Move items and lines by reassigning order_id
|
||
for item in items_to_move:
|
||
item.order_id = new_order.id
|
||
for ln in lines_to_move:
|
||
ln.order_id = new_order.id
|
||
|
||
await db.commit()
|
||
|
||
# Refresh estimated_price on both orders
|
||
from app.services.pricing_service import refresh_order_price
|
||
await refresh_order_price(db, order_id)
|
||
await refresh_order_price(db, new_order.id)
|
||
|
||
return SplitMissingStepResponse(
|
||
new_order_id=str(new_order.id),
|
||
new_order_number=new_order_number,
|
||
moved_item_count=len(items_to_move),
|
||
moved_line_count=len(lines_to_move),
|
||
)
|
||
|
||
|
||
class GenerateLinesRequest(BaseModel):
|
||
output_type_ids: list[uuid.UUID]
|
||
|
||
|
||
class GenerateLinesResponse(BaseModel):
|
||
created: int
|
||
skipped: int
|
||
no_product_count: int = 0
|
||
no_step_count: int = 0
|
||
|
||
|
||
@router.post("/{order_id}/generate-lines", response_model=GenerateLinesResponse)
|
||
async def generate_lines_from_items(
|
||
order_id: uuid.UUID,
|
||
body: GenerateLinesRequest,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Bulk-create OrderLines from OrderItems for orders that have no output lines.
|
||
|
||
Looks up each item's product by pim_id / produkt_baureihe, then creates one
|
||
line per product × requested output type (skips duplicates).
|
||
"""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if order.status not in (OrderStatus.draft, OrderStatus.submitted):
|
||
raise HTTPException(400, detail="Only draft or submitted orders support line generation")
|
||
if not body.output_type_ids:
|
||
raise HTTPException(400, detail="At least one output type is required")
|
||
|
||
from app.services.product_service import lookup_product
|
||
|
||
items_result = await db.execute(
|
||
select(OrderItem).where(OrderItem.order_id == order_id)
|
||
)
|
||
items = items_result.scalars().all()
|
||
|
||
# Fetch existing lines to skip duplicates
|
||
existing_result = await db.execute(
|
||
select(OrderLine.product_id, OrderLine.output_type_id)
|
||
.where(OrderLine.order_id == order_id)
|
||
)
|
||
existing_pairs: set[tuple] = {(str(r[0]), str(r[1])) for r in existing_result.all()}
|
||
|
||
created = 0
|
||
skipped = 0
|
||
no_product_count = 0
|
||
no_step_count = 0
|
||
|
||
for item in items:
|
||
# Use the canonical lookup: produkt_baureihe first (unique per product),
|
||
# then pim_id as fallback. pim_id is a category-level code shared by
|
||
# many products so it must NOT be used as the primary key.
|
||
product = await lookup_product(db, item.pim_id, item.produkt_baureihe)
|
||
if not product:
|
||
no_product_count += 1
|
||
continue
|
||
|
||
if product.cad_file_id is None:
|
||
no_step_count += 1
|
||
# Still create the line so it shows in the UI — it will fail at dispatch
|
||
# but the user can upload a STEP file and retry.
|
||
|
||
for type_id in body.output_type_ids:
|
||
pair = (str(product.id), str(type_id))
|
||
if pair in existing_pairs:
|
||
skipped += 1
|
||
continue
|
||
line = OrderLine(
|
||
order_id=order_id,
|
||
product_id=product.id,
|
||
output_type_id=type_id,
|
||
gewuenschte_bildnummer=item.gewuenschte_bildnummer,
|
||
tenant_id=getattr(user, 'tenant_id', None),
|
||
)
|
||
db.add(line)
|
||
existing_pairs.add(pair)
|
||
created += 1
|
||
|
||
await db.commit()
|
||
|
||
# Refresh estimated price
|
||
try:
|
||
from app.services.pricing_service import refresh_order_price
|
||
await refresh_order_price(db, order_id)
|
||
except Exception:
|
||
pass
|
||
|
||
return GenerateLinesResponse(
|
||
created=created,
|
||
skipped=skipped,
|
||
no_product_count=no_product_count,
|
||
no_step_count=no_step_count,
|
||
)
|
||
|
||
|
||
class OrderStatusUpdate(BaseModel):
|
||
status: str
|
||
notes: Optional[str] = None
|
||
|
||
|
||
@router.post("/{order_id}/status", response_model=OrderOut)
|
||
async def update_order_status(
|
||
order_id: uuid.UUID,
|
||
body: OrderStatusUpdate,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Update order status with lifecycle timestamps (admin / PM only)."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
now = datetime.utcnow()
|
||
try:
|
||
new_status = OrderStatus(body.status)
|
||
except ValueError:
|
||
raise HTTPException(400, detail=f"Invalid status: {body.status}")
|
||
|
||
order.status = new_status
|
||
order.updated_at = now
|
||
|
||
if new_status == OrderStatus.processing:
|
||
order.processing_started_at = now
|
||
elif new_status == OrderStatus.completed:
|
||
order.completed_at = now
|
||
elif new_status == OrderStatus.rejected:
|
||
order.rejected_at = now
|
||
|
||
if body.notes is not None:
|
||
order.notes = body.notes
|
||
|
||
# Auto-update order_lines.item_status to match order lifecycle
|
||
if new_status in (OrderStatus.processing, OrderStatus.completed):
|
||
await db.execute(
|
||
update(OrderLine)
|
||
.where(OrderLine.order_id == order.id)
|
||
.values(item_status="approved")
|
||
)
|
||
elif new_status == OrderStatus.rejected:
|
||
await db.execute(
|
||
update(OrderLine)
|
||
.where(OrderLine.order_id == order.id)
|
||
.values(item_status="rejected")
|
||
)
|
||
|
||
await db.commit()
|
||
|
||
# Notify the order creator about status change
|
||
from app.services.notification_service import emit_notification
|
||
await emit_notification(
|
||
db,
|
||
actor_user_id=user.id,
|
||
target_user_id=order.created_by,
|
||
action=f"order.{new_status.value}",
|
||
entity_type="order",
|
||
entity_id=str(order.id),
|
||
details={"order_number": order.order_number},
|
||
)
|
||
|
||
# Dispatch renders when order moves to processing
|
||
if new_status == OrderStatus.processing:
|
||
lines_result = await db.execute(
|
||
select(OrderLine).where(
|
||
OrderLine.order_id == order.id,
|
||
OrderLine.output_type_id.isnot(None),
|
||
OrderLine.render_status == "pending",
|
||
)
|
||
)
|
||
from app.tasks.step_tasks import dispatch_order_line_render
|
||
for line in lines_result.scalars().all():
|
||
dispatch_order_line_render.delay(str(line.id))
|
||
|
||
if new_status == OrderStatus.completed:
|
||
from app.services.pricing_service import refresh_order_price
|
||
await refresh_order_price(db, order.id)
|
||
|
||
await db.refresh(order)
|
||
return order
|
||
|
||
|
||
@router.post("/{order_id}/lines", response_model=OrderLineOut, status_code=status.HTTP_201_CREATED)
|
||
async def add_order_line(
|
||
order_id: uuid.UUID,
|
||
body: OrderLineCreate,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Add a product + output_type line to a draft order."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
if order.status == OrderStatus.draft:
|
||
pass # always allowed for owner/admin
|
||
elif order.status == OrderStatus.submitted and _is_privileged(user):
|
||
pass # admin / PM may add lines to a submitted order (e.g. to fix missing output types)
|
||
else:
|
||
raise HTTPException(400, detail="Can only add lines to draft orders (admins may also add to submitted orders)")
|
||
|
||
prod_result = await db.execute(
|
||
select(Product).options(selectinload(Product.cad_file)).where(Product.id == body.product_id)
|
||
)
|
||
if not prod_result.scalar_one_or_none():
|
||
raise HTTPException(404, detail="Product not found")
|
||
|
||
line = OrderLine(
|
||
order_id=order_id,
|
||
product_id=body.product_id,
|
||
output_type_id=body.output_type_id,
|
||
render_position_id=body.render_position_id,
|
||
global_render_position_id=body.global_render_position_id,
|
||
gewuenschte_bildnummer=body.gewuenschte_bildnummer,
|
||
material_override=body.material_override,
|
||
notes=body.notes,
|
||
tenant_id=getattr(user, 'tenant_id', None),
|
||
)
|
||
db.add(line)
|
||
try:
|
||
await db.commit()
|
||
except Exception:
|
||
await db.rollback()
|
||
raise HTTPException(409, detail="Duplicate line (same product + output_type + position already exists in this order)")
|
||
|
||
await db.refresh(line)
|
||
|
||
# Update estimated_price on the draft order immediately
|
||
from app.services.pricing_service import refresh_order_price
|
||
await refresh_order_price(db, order_id)
|
||
|
||
from app.models.output_type import OutputType as OTModel
|
||
from app.models.render_position import ProductRenderPosition
|
||
result2 = await db.execute(
|
||
select(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.options(
|
||
selectinload(OrderLine.product).selectinload(Product.cad_file),
|
||
selectinload(OrderLine.product).selectinload(Product.render_positions),
|
||
selectinload(OrderLine.output_type).selectinload(OTModel.pricing_tier),
|
||
selectinload(OrderLine.render_position),
|
||
)
|
||
)
|
||
line_loaded = result2.scalar_one()
|
||
return _build_line_out(line_loaded)
|
||
|
||
|
||
@router.get("/{order_id}/check-materials")
|
||
async def check_materials(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Check if all materials in this order's products are mapped to library materials."""
|
||
from app.domains.materials.service import find_unmapped_materials
|
||
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
lines_result = await db.execute(
|
||
select(OrderLine)
|
||
.options(selectinload(OrderLine.product))
|
||
.where(OrderLine.order_id == order_id)
|
||
)
|
||
lines = lines_result.scalars().all()
|
||
|
||
# Collect all unique material names from all products
|
||
all_material_names: list[str] = []
|
||
seen: set[str] = set()
|
||
for line in lines:
|
||
if not line.product or not line.product.cad_part_materials:
|
||
continue
|
||
for entry in line.product.cad_part_materials:
|
||
mat_name = entry.get("material", "")
|
||
if mat_name and mat_name.lower() not in seen:
|
||
seen.add(mat_name.lower())
|
||
all_material_names.append(mat_name)
|
||
|
||
unmapped = await find_unmapped_materials(all_material_names, db)
|
||
total = len(all_material_names)
|
||
mapped = total - len(unmapped)
|
||
|
||
return {
|
||
"unmapped": unmapped,
|
||
"total_materials": total,
|
||
"mapped_count": mapped,
|
||
}
|
||
|
||
|
||
@router.post("/{order_id}/dispatch-renders")
|
||
async def dispatch_renders(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Dispatch (or retry) renders for all pending/failed/cancelled lines (admin/PM only).
|
||
|
||
Auto-advances order to processing if currently submitted or completed.
|
||
"""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
if order.status not in (OrderStatus.submitted, OrderStatus.processing, OrderStatus.completed):
|
||
raise HTTPException(400, detail=f"Cannot dispatch renders for order in {order.status.value} status")
|
||
|
||
lines_result = await db.execute(
|
||
select(OrderLine).where(
|
||
OrderLine.order_id == order.id,
|
||
OrderLine.output_type_id.isnot(None),
|
||
OrderLine.render_status.in_(["pending", "failed", "cancelled"]),
|
||
)
|
||
)
|
||
lines = lines_result.scalars().all()
|
||
|
||
if not lines:
|
||
raise HTTPException(400, detail="No renderable lines with pending, failed, or cancelled status")
|
||
|
||
# Auto-advance to processing if not already there
|
||
if order.status in (OrderStatus.submitted, OrderStatus.completed):
|
||
now = datetime.utcnow()
|
||
order.status = OrderStatus.processing
|
||
order.processing_started_at = now
|
||
order.completed_at = None
|
||
order.updated_at = now
|
||
|
||
# Reset failed/cancelled lines to pending before re-dispatch
|
||
from sqlalchemy import update as sql_update
|
||
for line in lines:
|
||
if line.render_status in ("failed", "cancelled"):
|
||
await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.values(render_status="pending", render_completed_at=None, render_log=None)
|
||
)
|
||
await db.commit()
|
||
|
||
from app.domains.rendering.dispatch_service import dispatch_render_with_workflow
|
||
for line in lines:
|
||
try:
|
||
dispatch_render_with_workflow(str(line.id))
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"dispatch_render_with_workflow failed for %s, falling back: %s",
|
||
line.id, exc,
|
||
)
|
||
from app.tasks.step_tasks import dispatch_order_line_render
|
||
dispatch_order_line_render.delay(str(line.id))
|
||
|
||
return {"dispatched": len(lines), "order_status": order.status.value}
|
||
|
||
|
||
@router.post("/{order_id}/lines/{line_id}/cancel-render")
|
||
async def cancel_line_render(
|
||
order_id: uuid.UUID,
|
||
line_id: uuid.UUID,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Cancel a running render for a single order line (admin/PM only).
|
||
|
||
Cancels the Flamenco job or revokes the Celery task, then marks
|
||
the line as 'cancelled'.
|
||
"""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
line_result = await db.execute(
|
||
select(OrderLine).where(OrderLine.id == line_id, OrderLine.order_id == order_id)
|
||
)
|
||
line = line_result.scalar_one_or_none()
|
||
if not line:
|
||
raise HTTPException(404, detail="Order line not found")
|
||
|
||
if line.render_status not in ("processing", "pending"):
|
||
raise HTTPException(400, detail=f"Line render_status is '{line.render_status}', nothing to cancel")
|
||
|
||
cancelled_backend = line.render_backend_used or "celery"
|
||
errors: list[str] = []
|
||
|
||
# Revoke Celery task (best-effort) using real task ID from job document
|
||
try:
|
||
from app.tasks.celery_app import celery_app
|
||
real_task_id = None
|
||
if line.render_job_doc:
|
||
real_task_id = line.render_job_doc.get("celery_task_id")
|
||
task_id = real_task_id or f"render-{line_id}"
|
||
celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM")
|
||
except Exception as exc:
|
||
errors.append(f"Celery revoke failed: {str(exc)[:200]}")
|
||
|
||
# Mark line as cancelled
|
||
from sqlalchemy import update as sql_update
|
||
now = datetime.utcnow()
|
||
await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.values(
|
||
render_status="cancelled",
|
||
render_completed_at=now,
|
||
render_log={
|
||
"cancelled_by": str(user.id),
|
||
"cancelled_at": now.isoformat(),
|
||
"backend": cancelled_backend,
|
||
"errors": errors or None,
|
||
},
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
# Check if all renderable lines are now terminal → auto-complete order
|
||
await _maybe_complete_order(db, order_id)
|
||
|
||
return {
|
||
"cancelled": True,
|
||
"line_id": str(line.id),
|
||
"backend": cancelled_backend,
|
||
"errors": errors or None,
|
||
}
|
||
|
||
|
||
@router.post("/{order_id}/lines/{line_id}/dispatch-render")
|
||
async def dispatch_single_line_render(
|
||
order_id: uuid.UUID,
|
||
line_id: uuid.UUID,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Dispatch (or retry) a render for a single order line (admin/PM only)."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
line_result = await db.execute(
|
||
select(OrderLine).where(OrderLine.id == line_id, OrderLine.order_id == order.id)
|
||
)
|
||
line = line_result.scalar_one_or_none()
|
||
if not line:
|
||
raise HTTPException(404, detail="Order line not found")
|
||
|
||
if line.render_status not in ("pending", "failed", "cancelled"):
|
||
raise HTTPException(400, detail=f"Cannot dispatch line in {line.render_status} status")
|
||
|
||
# Reset to pending
|
||
from sqlalchemy import update as sql_update
|
||
await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.values(render_status="pending", render_completed_at=None, render_log=None)
|
||
)
|
||
|
||
# Auto-advance order to processing if needed
|
||
if order.status in (OrderStatus.submitted, OrderStatus.completed):
|
||
now = datetime.utcnow()
|
||
order.status = OrderStatus.processing
|
||
order.processing_started_at = now
|
||
order.completed_at = None
|
||
order.updated_at = now
|
||
|
||
await db.commit()
|
||
|
||
from app.domains.rendering.dispatch_service import dispatch_render_with_workflow
|
||
try:
|
||
dispatch_render_with_workflow(str(line.id))
|
||
except Exception as exc:
|
||
logger.warning("dispatch_render_with_workflow failed for %s: %s", line.id, exc)
|
||
from app.tasks.step_tasks import dispatch_order_line_render
|
||
dispatch_order_line_render.delay(str(line.id))
|
||
|
||
return {"dispatched": True, "line_id": str(line.id)}
|
||
|
||
|
||
class BatchMaterialOverrideBody(BaseModel):
|
||
material_override: str | None = None
|
||
|
||
|
||
@router.post("/{order_id}/batch-material-override")
|
||
async def batch_material_override(
|
||
order_id: uuid.UUID,
|
||
body: BatchMaterialOverrideBody,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Set material_override on ALL lines of an order at once."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
if not result.scalar_one_or_none():
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
from sqlalchemy import update as sql_update
|
||
res = await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.order_id == order_id)
|
||
.values(material_override=body.material_override)
|
||
)
|
||
await db.commit()
|
||
return {"updated": res.rowcount, "material_override": body.material_override}
|
||
|
||
|
||
class PatchLineBody(BaseModel):
|
||
material_override: str | None = None
|
||
|
||
|
||
@router.patch("/{order_id}/lines/{line_id}")
|
||
async def patch_order_line(
|
||
order_id: uuid.UUID,
|
||
line_id: uuid.UUID,
|
||
body: PatchLineBody,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Update fields on an order line (admin/PM only)."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
line_result = await db.execute(
|
||
select(OrderLine).where(OrderLine.id == line_id, OrderLine.order_id == order_id)
|
||
)
|
||
line = line_result.scalar_one_or_none()
|
||
if not line:
|
||
raise HTTPException(404, detail="Order line not found")
|
||
|
||
data = body.model_dump(exclude_unset=True)
|
||
from sqlalchemy import update as sql_update
|
||
if data:
|
||
await db.execute(
|
||
sql_update(OrderLine).where(OrderLine.id == line.id).values(**data)
|
||
)
|
||
await db.commit()
|
||
|
||
return {"updated": True, "line_id": str(line.id)}
|
||
|
||
|
||
class RejectLineBody(BaseModel):
|
||
reason: str = ""
|
||
|
||
|
||
@router.post("/{order_id}/lines/{line_id}/reject", status_code=200)
|
||
async def reject_order_line(
|
||
order_id: uuid.UUID,
|
||
line_id: uuid.UUID,
|
||
body: RejectLineBody,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Reject a single order line (admin/PM only).
|
||
|
||
Sets item_status to 'rejected' and stores the reason in the notes field.
|
||
"""
|
||
if not _is_privileged(user):
|
||
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
||
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
line_result = await db.execute(
|
||
select(OrderLine).where(OrderLine.id == line_id, OrderLine.order_id == order_id)
|
||
)
|
||
line = line_result.scalar_one_or_none()
|
||
if not line:
|
||
raise HTTPException(404, detail="Order line not found")
|
||
|
||
from sqlalchemy import update as sql_update
|
||
|
||
notes_value = body.reason.strip() if body.reason and body.reason.strip() else line.notes
|
||
await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.values(
|
||
item_status="rejected",
|
||
notes=notes_value,
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
return {"rejected": True, "line_id": str(line.id), "reason": body.reason}
|
||
|
||
|
||
@router.post("/{order_id}/cancel-renders")
|
||
async def cancel_order_renders(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(require_admin_or_pm),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Cancel all processing/pending renders for an order (admin/PM only)."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
lines_result = await db.execute(
|
||
select(OrderLine).where(
|
||
OrderLine.order_id == order.id,
|
||
OrderLine.output_type_id.isnot(None),
|
||
OrderLine.render_status.in_(["processing", "pending"]),
|
||
)
|
||
)
|
||
lines = lines_result.scalars().all()
|
||
|
||
if not lines:
|
||
raise HTTPException(400, detail="No active renders to cancel")
|
||
|
||
from app.tasks.celery_app import celery_app
|
||
from sqlalchemy import update as sql_update
|
||
|
||
now = datetime.utcnow()
|
||
cancelled_count = 0
|
||
errors: list[str] = []
|
||
|
||
for line in lines:
|
||
# Revoke Celery task using real task ID from job document
|
||
try:
|
||
real_task_id = None
|
||
if line.render_job_doc:
|
||
real_task_id = line.render_job_doc.get("celery_task_id")
|
||
task_id = real_task_id or f"render-{line.id}"
|
||
celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM")
|
||
except Exception:
|
||
pass
|
||
|
||
await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.values(
|
||
render_status="cancelled",
|
||
render_completed_at=now,
|
||
render_log={
|
||
"cancelled_by": str(user.id),
|
||
"cancelled_at": now.isoformat(),
|
||
"backend": line.render_backend_used or "unknown",
|
||
},
|
||
)
|
||
)
|
||
cancelled_count += 1
|
||
|
||
await db.commit()
|
||
|
||
# Check if all renderable lines are now terminal → auto-complete order
|
||
await _maybe_complete_order(db, order_id)
|
||
|
||
# Re-read order status (may have changed)
|
||
await db.refresh(order)
|
||
|
||
return {
|
||
"cancelled": cancelled_count,
|
||
"order_status": order.status.value,
|
||
"errors": errors or None,
|
||
}
|
||
|
||
|
||
@router.post("/{order_id}/reject", response_model=OrderOut)
|
||
async def reject_order(
|
||
order_id: uuid.UUID,
|
||
body: RejectOrderRequest,
|
||
user: User = Depends(require_pm_or_above),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Reject a submitted or processing order (PM / admin only).
|
||
|
||
Cancels all pending/processing render lines and notifies the order creator.
|
||
"""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
if order.status not in (OrderStatus.submitted, OrderStatus.processing):
|
||
raise HTTPException(
|
||
400,
|
||
detail=f"Cannot reject order in '{order.status.value}' status — only submitted or processing orders can be rejected",
|
||
)
|
||
|
||
now = datetime.utcnow()
|
||
order.status = OrderStatus.rejected
|
||
order.rejected_at = now
|
||
order.rejection_reason = body.reason or None
|
||
order.updated_at = now
|
||
|
||
# Cancel all pending/processing render lines
|
||
from sqlalchemy import update as sql_update
|
||
from app.tasks.celery_app import celery_app
|
||
|
||
active_lines_result = await db.execute(
|
||
select(OrderLine).where(
|
||
OrderLine.order_id == order.id,
|
||
OrderLine.render_status.in_(["pending", "processing"]),
|
||
)
|
||
)
|
||
active_lines = active_lines_result.scalars().all()
|
||
for line in active_lines:
|
||
try:
|
||
real_task_id = None
|
||
if line.render_job_doc:
|
||
real_task_id = line.render_job_doc.get("celery_task_id")
|
||
task_id = real_task_id or f"render-{line.id}"
|
||
celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM")
|
||
except Exception:
|
||
pass
|
||
|
||
await db.execute(
|
||
sql_update(OrderLine)
|
||
.where(OrderLine.id == line.id)
|
||
.values(
|
||
render_status="cancelled",
|
||
render_completed_at=now,
|
||
render_log={
|
||
"cancelled_by": str(user.id),
|
||
"cancelled_at": now.isoformat(),
|
||
"reason": "order_rejected",
|
||
},
|
||
)
|
||
)
|
||
|
||
# Mark all order lines as rejected
|
||
await db.execute(
|
||
update(OrderLine)
|
||
.where(OrderLine.order_id == order.id)
|
||
.values(item_status="rejected")
|
||
)
|
||
|
||
await db.commit()
|
||
await db.refresh(order)
|
||
|
||
# Notify the order creator
|
||
if body.notify_client:
|
||
from app.services.notification_service import emit_notification
|
||
reason_text = f" Reason: {body.reason}" if body.reason else ""
|
||
await emit_notification(
|
||
db,
|
||
actor_user_id=user.id,
|
||
target_user_id=order.created_by,
|
||
action="order.rejected",
|
||
entity_type="order",
|
||
entity_id=str(order.id),
|
||
details={
|
||
"order_number": order.order_number,
|
||
"reason": body.reason or "",
|
||
"message": f"Your order {order.order_number} was rejected.{reason_text}",
|
||
},
|
||
)
|
||
|
||
# Broadcast WebSocket event
|
||
try:
|
||
from app.core.websocket import manager as _ws_mgr
|
||
_tid = str(user.tenant_id) if user.tenant_id else None
|
||
if _tid:
|
||
await _ws_mgr.broadcast_to_tenant(_tid, {
|
||
"type": "order_status_change",
|
||
"order_id": str(order.id),
|
||
"status": "rejected",
|
||
})
|
||
except Exception:
|
||
pass
|
||
|
||
return order
|
||
|
||
|
||
@router.post("/{order_id}/resubmit", response_model=OrderOut)
|
||
async def resubmit_order(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Resubmit a rejected order back to draft (creator or PM+).
|
||
|
||
Clears rejection_reason / rejected_at and resets the order to draft
|
||
so the client can correct and resubmit.
|
||
"""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
|
||
if order.status != OrderStatus.rejected:
|
||
raise HTTPException(400, detail=f"Only rejected orders can be resubmitted (current status: {order.status.value})")
|
||
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied — only the order creator or a PM/admin can resubmit")
|
||
|
||
now = datetime.utcnow()
|
||
order.status = OrderStatus.draft
|
||
order.rejected_at = None
|
||
order.rejection_reason = None
|
||
order.updated_at = now
|
||
|
||
await db.commit()
|
||
await db.refresh(order)
|
||
|
||
# Notify PMs/admins about the resubmission (broadcast)
|
||
from app.services.notification_service import emit_notification
|
||
await emit_notification(
|
||
db,
|
||
actor_user_id=user.id,
|
||
target_user_id=None,
|
||
action="order.resubmitted",
|
||
entity_type="order",
|
||
entity_id=str(order.id),
|
||
details={
|
||
"order_number": order.order_number,
|
||
"resubmitted_by": str(user.id),
|
||
},
|
||
)
|
||
|
||
# Broadcast WebSocket event
|
||
try:
|
||
from app.core.websocket import manager as _ws_mgr
|
||
_tid = str(user.tenant_id) if user.tenant_id else None
|
||
if _tid:
|
||
await _ws_mgr.broadcast_to_tenant(_tid, {
|
||
"type": "order_status_change",
|
||
"order_id": str(order.id),
|
||
"status": "draft",
|
||
})
|
||
except Exception:
|
||
pass
|
||
|
||
return order
|
||
|
||
|
||
@router.delete("/{order_id}/lines/{line_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
async def remove_order_line(
|
||
order_id: uuid.UUID,
|
||
line_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Remove a line from a draft order."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
if order.status != OrderStatus.draft:
|
||
raise HTTPException(400, detail="Can only remove lines from draft orders")
|
||
|
||
line_result = await db.execute(
|
||
select(OrderLine).where(OrderLine.id == line_id, OrderLine.order_id == order_id)
|
||
)
|
||
line = line_result.scalar_one_or_none()
|
||
if not line:
|
||
raise HTTPException(404, detail="Order line not found")
|
||
|
||
await db.delete(line)
|
||
await db.commit()
|
||
|
||
# Update estimated_price after removing the line
|
||
from app.services.pricing_service import refresh_order_price
|
||
await refresh_order_price(db, order_id)
|
||
|
||
|
||
@router.get("/{order_id}/download-renders")
|
||
async def download_renders(
|
||
order_id: uuid.UUID,
|
||
user: User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Stream a ZIP of all completed render files for this order."""
|
||
result = await db.execute(select(Order).where(Order.id == order_id))
|
||
order = result.scalar_one_or_none()
|
||
if not order:
|
||
raise HTTPException(404, detail="Order not found")
|
||
if not _is_privileged(user) and order.created_by != user.id:
|
||
raise HTTPException(403, detail="Access denied")
|
||
|
||
lines_result = await db.execute(
|
||
select(OrderLine)
|
||
.where(
|
||
OrderLine.order_id == order_id,
|
||
OrderLine.render_status == "completed",
|
||
OrderLine.result_path.isnot(None),
|
||
)
|
||
.options(
|
||
selectinload(OrderLine.product),
|
||
selectinload(OrderLine.output_type),
|
||
selectinload(OrderLine.render_position),
|
||
)
|
||
)
|
||
lines = lines_result.scalars().all()
|
||
|
||
if not lines:
|
||
raise HTTPException(404, detail="No completed renders found for this order")
|
||
|
||
from app.config import settings as app_settings
|
||
|
||
def _resolve_path(p: str) -> str:
|
||
"""Translate container-relative paths to backend filesystem paths."""
|
||
# Flamenco worker mounts the uploads volume at /shared, backend at /app/uploads
|
||
if p.startswith("/shared/"):
|
||
return app_settings.upload_dir + p[len("/shared"):]
|
||
return p
|
||
|
||
buf = io.BytesIO()
|
||
# Track names used to avoid duplicates
|
||
name_counts: dict[str, int] = {}
|
||
|
||
with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||
for line in lines:
|
||
if not line.result_path:
|
||
continue
|
||
fs_path = _resolve_path(line.result_path)
|
||
if not os.path.isfile(fs_path):
|
||
continue
|
||
# Build a meaningful filename
|
||
product_name = (line.product.name or line.product.pim_id or "product") if line.product else "product"
|
||
ot_name = (line.output_type.name if line.output_type else None) or "render"
|
||
pos_name = (line.render_position.name if line.render_position else None)
|
||
# Sanitize: replace spaces + special chars with underscore
|
||
def _safe(s: str) -> str:
|
||
return re.sub(r"[^\w\-.]", "_", s).strip("_")
|
||
|
||
parts = [_safe(product_name), _safe(ot_name)]
|
||
if pos_name:
|
||
parts.append(_safe(pos_name))
|
||
ext = os.path.splitext(line.result_path)[1] or ".png"
|
||
base_name = "_".join(parts) + ext
|
||
|
||
# Deduplicate
|
||
if base_name in name_counts:
|
||
name_counts[base_name] += 1
|
||
stem, suffix = os.path.splitext(base_name)
|
||
archive_name = f"{stem}_{name_counts[base_name]}{suffix}"
|
||
else:
|
||
name_counts[base_name] = 0
|
||
archive_name = base_name
|
||
|
||
zf.write(fs_path, archive_name)
|
||
|
||
if not zf.infolist():
|
||
raise HTTPException(404, detail="No render files found on disk")
|
||
|
||
buf.seek(0)
|
||
safe_order = re.sub(r"[^\w\-]", "_", order.order_number)
|
||
filename = f"{safe_order}_renders.zip"
|
||
|
||
return StreamingResponse(
|
||
buf,
|
||
media_type="application/zip",
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|