219 lines
7.7 KiB
Python
219 lines
7.7 KiB
Python
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import Select, func, or_, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.models.repair import Repair, RepairIntakeEvent, RepairStatusHistory
|
|
from app.schemas.repair import RepairCreate, RepairStatusUpdate, RepairUpdate
|
|
|
|
|
|
class RepairRepository:
|
|
@staticmethod
|
|
def query(
|
|
*,
|
|
status: str | None = None,
|
|
q: str | None = None,
|
|
customer_id: int | None = None,
|
|
source: str | None = None,
|
|
priority: str | None = None,
|
|
) -> Select[tuple[Repair]]:
|
|
query = select(Repair).options(selectinload(Repair.history))
|
|
if status:
|
|
query = query.where(Repair.status == status)
|
|
if customer_id is not None:
|
|
query = query.where(Repair.customer_id == customer_id)
|
|
if source:
|
|
query = query.where(Repair.source == source)
|
|
if priority:
|
|
query = query.where(Repair.priority == priority)
|
|
if q:
|
|
term = f"%{q.strip()}%"
|
|
query = query.where(
|
|
or_(
|
|
Repair.repair_number.ilike(term),
|
|
Repair.customer_name.ilike(term),
|
|
Repair.device_manufacturer.ilike(term),
|
|
Repair.device_model.ilike(term),
|
|
Repair.fault_description.ilike(term),
|
|
Repair.source_reference.ilike(term),
|
|
)
|
|
)
|
|
return query
|
|
|
|
@staticmethod
|
|
def list(
|
|
db: Session,
|
|
*,
|
|
status: str | None = None,
|
|
q: str | None = None,
|
|
customer_id: int | None = None,
|
|
source: str | None = None,
|
|
priority: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> tuple[list[Repair], int]:
|
|
base_query = RepairRepository.query(
|
|
status=status,
|
|
q=q,
|
|
customer_id=customer_id,
|
|
source=source,
|
|
priority=priority,
|
|
)
|
|
total = db.scalar(select(func.count()).select_from(base_query.subquery())) or 0
|
|
items = list(
|
|
db.scalars(
|
|
base_query
|
|
.order_by(Repair.created_at.desc(), Repair.id.desc())
|
|
.offset(offset)
|
|
.limit(limit)
|
|
)
|
|
)
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def get_by_id(db: Session, repair_id: int) -> Repair | None:
|
|
return db.scalar(select(Repair).options(selectinload(Repair.history)).where(Repair.id == repair_id))
|
|
|
|
@staticmethod
|
|
def get_next_number(db: Session, year: int) -> str:
|
|
prefix = f"R{year}-"
|
|
latest = db.scalar(
|
|
select(Repair.repair_number)
|
|
.where(Repair.repair_number.like(f"{prefix}%"))
|
|
.order_by(Repair.repair_number.desc())
|
|
.limit(1)
|
|
)
|
|
next_number = 1
|
|
if latest:
|
|
next_number = int(latest.split("-")[-1]) + 1
|
|
return f"{prefix}{next_number:06d}"
|
|
|
|
@staticmethod
|
|
def create(db: Session, payload: RepairCreate, *, repair_number: str) -> Repair:
|
|
repair = Repair(repair_number=repair_number, **RepairRepository._payload_data(payload))
|
|
db.add(repair)
|
|
db.flush()
|
|
RepairRepository.add_history(db, repair.id, old_status=None, new_status=repair.status, note="Reparatur angelegt", actor_user_id=None)
|
|
db.commit()
|
|
db.refresh(repair)
|
|
return RepairRepository.get_by_id(db, repair.id) or repair
|
|
|
|
@staticmethod
|
|
def update(db: Session, repair: Repair, payload: RepairUpdate) -> Repair:
|
|
for key, value in RepairRepository._payload_data(payload).items():
|
|
setattr(repair, key, value)
|
|
RepairRepository._apply_status_timestamps(repair)
|
|
db.commit()
|
|
db.refresh(repair)
|
|
return RepairRepository.get_by_id(db, repair.id) or repair
|
|
|
|
@staticmethod
|
|
def update_status(db: Session, repair: Repair, payload: RepairStatusUpdate, *, actor_user_id: int | None) -> Repair:
|
|
old_status = repair.status
|
|
repair.status = payload.status
|
|
RepairRepository._apply_status_timestamps(repair)
|
|
RepairRepository.add_history(
|
|
db,
|
|
repair.id,
|
|
old_status=old_status,
|
|
new_status=payload.status,
|
|
note=payload.note,
|
|
actor_user_id=actor_user_id,
|
|
)
|
|
db.commit()
|
|
db.refresh(repair)
|
|
return RepairRepository.get_by_id(db, repair.id) or repair
|
|
|
|
@staticmethod
|
|
def cancel(db: Session, repair: Repair, *, actor_user_id: int | None) -> Repair:
|
|
old_status = repair.status
|
|
repair.status = "cancelled"
|
|
repair.completed_at = datetime.now(UTC)
|
|
RepairRepository.add_history(
|
|
db,
|
|
repair.id,
|
|
old_status=old_status,
|
|
new_status="cancelled",
|
|
note="Reparatur storniert",
|
|
actor_user_id=actor_user_id,
|
|
)
|
|
db.commit()
|
|
db.refresh(repair)
|
|
return repair
|
|
|
|
@staticmethod
|
|
def add_history(db: Session, repair_id: int, *, old_status: str | None, new_status: str, note: str | None, actor_user_id: int | None) -> RepairStatusHistory:
|
|
history = RepairStatusHistory(
|
|
repair_id=repair_id,
|
|
old_status=old_status,
|
|
new_status=new_status,
|
|
note=note,
|
|
actor_user_id=actor_user_id,
|
|
)
|
|
db.add(history)
|
|
return history
|
|
|
|
@staticmethod
|
|
def get_history(db: Session, repair_id: int) -> list[RepairStatusHistory]:
|
|
return list(
|
|
db.scalars(
|
|
select(RepairStatusHistory)
|
|
.where(RepairStatusHistory.repair_id == repair_id)
|
|
.order_by(RepairStatusHistory.created_at.desc(), RepairStatusHistory.id.desc())
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def count_by_status(db: Session, status: str) -> int:
|
|
return db.scalar(select(func.count(Repair.id)).where(Repair.status == status)) or 0
|
|
|
|
@staticmethod
|
|
def create_intake_event(
|
|
db: Session,
|
|
*,
|
|
external_source: str,
|
|
external_reference: str | None,
|
|
payload: dict,
|
|
status: str,
|
|
error_message: str | None = None,
|
|
repair_id: int | None = None,
|
|
processed_at: datetime | None = None,
|
|
) -> RepairIntakeEvent:
|
|
event = RepairIntakeEvent(
|
|
external_source=external_source,
|
|
external_reference=external_reference,
|
|
payload=payload,
|
|
status=status,
|
|
error_message=error_message,
|
|
repair_id=repair_id,
|
|
processed_at=processed_at,
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
db.refresh(event)
|
|
return event
|
|
|
|
@staticmethod
|
|
def update_intake_event(db: Session, event: RepairIntakeEvent, *, status: str, repair_id: int | None = None, error_message: str | None = None) -> RepairIntakeEvent:
|
|
event.status = status
|
|
event.repair_id = repair_id
|
|
event.error_message = error_message
|
|
event.processed_at = datetime.now(UTC)
|
|
db.commit()
|
|
db.refresh(event)
|
|
return event
|
|
|
|
@staticmethod
|
|
def _payload_data(payload: RepairCreate | RepairUpdate) -> dict:
|
|
data = payload.model_dump()
|
|
data["customer_email"] = str(payload.customer_email or "")
|
|
data["device_serial_number"] = payload.device_serial_number or None
|
|
data["source_reference"] = payload.source_reference or None
|
|
return data
|
|
|
|
@staticmethod
|
|
def _apply_status_timestamps(repair: Repair) -> None:
|
|
if repair.status == "approved" and repair.approved_at is None:
|
|
repair.approved_at = datetime.now(UTC)
|
|
if repair.status in {"completed", "cancelled"} and repair.completed_at is None:
|
|
repair.completed_at = datetime.now(UTC)
|