597 lines
24 KiB
Python
597 lines
24 KiB
Python
from fastapi import APIRouter, Depends, Query, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select
|
|
from datetime import date
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
from app.core.database.db_session import get_db
|
|
from app.core.permissions.RoleChecker import RoleChecker
|
|
from app.models.UserModel import User
|
|
# For customer oauth verification
|
|
from app.api.v1.routers.CheckoutRouter import get_current_customer
|
|
from app.models.EcomCustomerModel import EcomCustomer
|
|
from app.models.ServiceModel import ServiceJob
|
|
|
|
from app.schemas.ServiceSchema import (
|
|
ServiceJobCreate, ServiceJobResponse, ServiceJobIntakeCreate, ServiceJobIntakeResponse,
|
|
ServiceJobInspectionCreate, ServiceJobInspectionResponse, ServiceJobRescheduleRequest,
|
|
ServiceCatalogResponse, ServiceJobMediaBatchCreate, ServiceJobMediaResponse
|
|
)
|
|
from app.schemas.ServiceQuoteSchema import ServiceJobQuoteCreate, ServiceJobQuoteResponse
|
|
from app.schemas.ServicePaymentSchema import ServicePaymentCreate, RazorpayVerificationRequest
|
|
|
|
from app.services.ServiceJobService import ServiceJobService
|
|
from app.services.SlotAllocationService import SlotAllocationService
|
|
from app.services.InspectionService import InspectionService
|
|
from app.services.QuoteService import QuoteService
|
|
from app.services.ServicePaymentService import ServicePaymentService
|
|
from app.repositories.ServiceRepository import ServiceRepository
|
|
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
def get_current_actor(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
if not credentials:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
token = credentials.credentials
|
|
try:
|
|
from app.core.Token import verify_access_token
|
|
payload = verify_access_token(token)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=f"Token validation failed: {str(e)}",
|
|
headers={"WWW-Authenticate": "Bearer"}
|
|
)
|
|
role = payload.get("role", "")
|
|
sub = payload.get("sub")
|
|
if role == "customer":
|
|
customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == sub).first()
|
|
if not customer:
|
|
raise HTTPException(status_code=404, detail="Customer not found")
|
|
return {"type": "customer", "obj": customer}
|
|
elif role in ["Super Admin", "Admin", "Manager", "Technician"]:
|
|
user = db.query(User).filter(User.user_id == sub).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Admin user not found")
|
|
return {"type": "admin", "obj": user}
|
|
else:
|
|
raise HTTPException(status_code=403, detail="Not authorized")
|
|
|
|
def get_optional_actor(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
|
|
db: Session = Depends(get_db)
|
|
) -> Optional[dict]:
|
|
if not credentials:
|
|
return None
|
|
token = credentials.credentials
|
|
try:
|
|
from app.core.Token import verify_access_token
|
|
payload = verify_access_token(token)
|
|
role = payload.get("role", "")
|
|
sub = payload.get("sub")
|
|
if role == "customer":
|
|
customer = db.query(EcomCustomer).filter(EcomCustomer.customer_id == sub).first()
|
|
if customer:
|
|
return {"type": "customer", "obj": customer}
|
|
elif role in ["Super Admin", "Admin", "Manager", "Technician"]:
|
|
user = db.query(User).filter(User.user_id == sub).first()
|
|
if user:
|
|
return {"type": "admin", "obj": user}
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
router = APIRouter(prefix="/api/v1/service", tags=["Repairs & Scheduling Services"])
|
|
|
|
job_service = ServiceJobService()
|
|
allocation_service = SlotAllocationService()
|
|
inspection_service = InspectionService()
|
|
quote_service = QuoteService()
|
|
payment_service = ServicePaymentService()
|
|
service_repo = ServiceRepository()
|
|
|
|
@router.get("/catalog", response_model=List[ServiceCatalogResponse])
|
|
def get_catalog_services(db: Session = Depends(get_db)):
|
|
"""Get active repair & diagnostics catalog options."""
|
|
return service_repo.get_all_catalog_services(db)
|
|
|
|
@router.get("/slots/available")
|
|
def get_available_appointment_slots(
|
|
service_id: str,
|
|
target_date: date = Query(...),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Dynamically fetch schedule slots based on technician availability and duration."""
|
|
return allocation_service.get_available_slots(db, target_date, service_id)
|
|
|
|
@router.post("/booking/create")
|
|
def create_online_service_booking(
|
|
payload: ServiceJobCreate,
|
|
actor: dict = Depends(get_current_actor),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Book a repair job slot and lock capacity (starts 10-minute slot hold)."""
|
|
if actor["type"] == "customer":
|
|
customer_id = actor["obj"].customer_id
|
|
else:
|
|
# Admin flow
|
|
if payload.customer_id:
|
|
customer_id = payload.customer_id
|
|
elif payload.customer_name:
|
|
import ulid
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
parts = payload.customer_name.strip().split(maxsplit=1)
|
|
first_name = parts[0]
|
|
last_name = parts[1] if len(parts) > 1 else ""
|
|
|
|
u_str = str(ulid.ULID())
|
|
raw_phone = (payload.customer_phone or "").strip()
|
|
digits_only = "".join(c for c in raw_phone if c.isdigit())
|
|
phone_suffix = digits_only[-10:] if len(digits_only) >= 10 else digits_only
|
|
|
|
email = (payload.customer_email or f"walkin_{u_str}@ifixkart.com").strip()
|
|
phone = raw_phone if raw_phone else f"W{u_str[-15:]}"
|
|
|
|
existing = None
|
|
if phone_suffix:
|
|
existing = db.query(EcomCustomer).filter(
|
|
(EcomCustomer.phone.like(f"%{phone_suffix}%")) | (EcomCustomer.email == email)
|
|
).first()
|
|
elif payload.customer_email:
|
|
existing = db.query(EcomCustomer).filter(EcomCustomer.email == email).first()
|
|
|
|
if existing:
|
|
customer_id = existing.customer_id
|
|
else:
|
|
try:
|
|
new_cust = EcomCustomer(
|
|
customer_id=str(ulid.ULID()),
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
email=email,
|
|
phone=phone,
|
|
is_active=True
|
|
)
|
|
db.add(new_cust)
|
|
db.commit()
|
|
db.refresh(new_cust)
|
|
customer_id = new_cust.customer_id
|
|
except IntegrityError:
|
|
db.rollback()
|
|
found = db.query(EcomCustomer).filter(
|
|
(EcomCustomer.phone.like(f"%{phone_suffix}%")) if phone_suffix else (EcomCustomer.email == email)
|
|
).first()
|
|
if found:
|
|
customer_id = found.customer_id
|
|
else:
|
|
customer_id = str(ulid.ULID())
|
|
else:
|
|
raise HTTPException(status_code=400, detail="customer_id or customer_name required for admin booking")
|
|
|
|
return job_service.create_online_booking(db, customer_id, payload)
|
|
|
|
@router.post("/jobs/{job_id}/intake", response_model=ServiceJobIntakeResponse)
|
|
def record_walk_in_device_intake(
|
|
job_id: str,
|
|
payload: ServiceJobIntakeCreate,
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Record physical checklists (SIM tray, scratches, power status) during device receipt."""
|
|
return inspection_service.create_device_intake(db, job_id, current_user.user_id, payload)
|
|
|
|
@router.post("/jobs/{job_id}/inspect", response_model=ServiceJobInspectionResponse)
|
|
def submit_technician_diagnostic_findings(
|
|
job_id: str,
|
|
payload: ServiceJobInspectionCreate,
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Submit technician diagnostic findings and inspection status."""
|
|
return inspection_service.submit_inspection(db, job_id, current_user.user_id, payload)
|
|
|
|
@router.post("/jobs/{job_id}/quotes")
|
|
@router.post("/jobs/{job_id}/quotes/create")
|
|
def create_or_revise_repair_quote(
|
|
job_id: str,
|
|
payload: ServiceJobQuoteCreate,
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Generate or revise estimate quote for customer approval."""
|
|
return quote_service.create_or_revise_quote(db, job_id, current_user.user_id, payload)
|
|
|
|
@router.post("/jobs/{job_id}/quotes/{quote_id}/respond")
|
|
def respond_to_quote_estimate(
|
|
job_id: str,
|
|
quote_id: str,
|
|
action: str = Query(..., regex="^(ACCEPT|REJECT)$"),
|
|
actor: dict = Depends(get_current_actor),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Customer approves or declines line items in revised estimate quote."""
|
|
customer_id = actor["obj"].customer_id if actor["type"] == "customer" else "SYSTEM"
|
|
return quote_service.respond_to_quote(db, quote_id, customer_id, action)
|
|
|
|
@router.post("/jobs/{job_id}/payments/initiate")
|
|
def initiate_milestone_payment(
|
|
job_id: str,
|
|
payload: ServicePaymentCreate,
|
|
actor: dict = Depends(get_current_actor),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Initiate Razorpay order for advance deposit or final balance payment."""
|
|
customer_id = actor["obj"].customer_id if actor["type"] == "customer" else None
|
|
return payment_service.initiate_payment(db, job_id, payload, customer_id)
|
|
|
|
@router.post("/payments/verify")
|
|
@router.post("/jobs/{job_id}/payments/verify")
|
|
def verify_milestone_payment(
|
|
payload: RazorpayVerificationRequest,
|
|
job_id: Optional[str] = None,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Verify Razorpay payment signature & update job/payment ledger."""
|
|
return payment_service.verify_payment(db, job_id or "", payload)
|
|
|
|
@router.post("/jobs/{job_id}/reschedule")
|
|
def reschedule_active_appointment(
|
|
job_id: str,
|
|
payload: ServiceJobRescheduleRequest,
|
|
actor: dict = Depends(get_current_actor),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Customer or staff reschedules repair appointment time slot."""
|
|
customer_id = actor["obj"].customer_id if actor["type"] == "customer" else None
|
|
return job_service.reschedule_appointment(db, job_id, customer_id, payload)
|
|
|
|
@router.get("/jobs/{job_id}")
|
|
def get_service_job_details(
|
|
job_id: str,
|
|
actor: Optional[dict] = Depends(get_optional_actor),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Retrieve service job, current appointment, timeline events, active quotes, and media proof."""
|
|
job = db.get(ServiceJob, job_id) or db.query(ServiceJob).filter(ServiceJob.job_no == job_id).first()
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Service job not found")
|
|
|
|
if actor and actor["type"] == "customer" and job.customer_id and job.customer_id != actor["obj"].customer_id:
|
|
raise HTTPException(status_code=403, detail="Not authorized to view this job")
|
|
|
|
from app.models.ServiceModel import ServiceAppointment, ServiceJobEvent
|
|
from app.models.ServiceQuoteModel import ServiceJobQuote
|
|
from app.models.ServicePaymentModel import ServicePayment
|
|
|
|
# Active appointment
|
|
app_stmt = select(ServiceAppointment).where(
|
|
ServiceAppointment.service_job_id == job.job_id,
|
|
ServiceAppointment.status.in_(["HELD", "CONFIRMED"])
|
|
)
|
|
appt = db.execute(app_stmt).scalar_one_or_none()
|
|
|
|
# Timeline events
|
|
event_stmt = select(ServiceJobEvent).where(ServiceJobEvent.job_id == job.job_id).order_by(ServiceJobEvent.timestamp.asc())
|
|
events = list(db.execute(event_stmt).scalars().all())
|
|
|
|
# Latest quote
|
|
quote_stmt = select(ServiceJobQuote).where(
|
|
ServiceJobQuote.service_job_id == job.job_id,
|
|
ServiceJobQuote.status.in_(["PENDING_CUSTOMER", "ACCEPTED", "REJECTED"])
|
|
).order_by(ServiceJobQuote.version.desc())
|
|
quote = db.execute(quote_stmt).scalars().first()
|
|
|
|
# Payments
|
|
pay_stmt = select(ServicePayment).where(ServicePayment.service_job_id == job.job_id)
|
|
payments = list(db.execute(pay_stmt).scalars().all())
|
|
|
|
# Latest inspection (for damage description)
|
|
from app.models.ServiceModel import ServiceJobInspection
|
|
insp_stmt = select(ServiceJobInspection).where(
|
|
ServiceJobInspection.service_job_id == job.job_id
|
|
).order_by(ServiceJobInspection.created_at.desc())
|
|
inspection = db.execute(insp_stmt).scalars().first()
|
|
|
|
from app.models.EcomCustomerModel import EcomCustomer
|
|
cust = db.get(EcomCustomer, job.customer_id) if job.customer_id else None
|
|
|
|
# Fetch job media items
|
|
media_list = inspection_service.get_job_media(db, job.job_id)
|
|
|
|
# Build response payload
|
|
return {
|
|
"job_id": job.job_id,
|
|
"job_no": job.job_no,
|
|
"status": job.status,
|
|
"customer": {
|
|
"customer_id": job.customer_id,
|
|
"name": f"{cust.first_name} {cust.last_name}".strip() if cust else "Guest Customer",
|
|
"email": cust.email if cust else "N/A",
|
|
"phone": cust.phone if cust else "N/A"
|
|
},
|
|
"service_name": job.service_name_snapshot or (job.service.name if job.service else "Repair Service"),
|
|
"base_price": float(job.base_price_snapshot if job.base_price_snapshot is not None else (job.service.base_price if job.service else 0.0)),
|
|
"device_brand": job.device.brand if job.device else (job.brand_id or "Generic"),
|
|
"device_model": job.device.model if job.device else (job.model_id or "Device"),
|
|
"fulfillment_type": job.fulfillment_type or "WALK_IN",
|
|
"fulfillment_fee": float(job.fulfillment_fee or (250.0 if job.fulfillment_type == "DOORSTEP_PICKUP" else 0.0)),
|
|
"courier_name": getattr(job, "courier_name", None),
|
|
"awb_number": getattr(job, "awb_number", None),
|
|
"pickup_status": getattr(job, "pickup_status", None),
|
|
"delivery_address": job.delivery_address,
|
|
"media": media_list,
|
|
"appointment": {
|
|
"appointment_id": appt.appointment_id,
|
|
"scheduled_start": appt.scheduled_start.isoformat(),
|
|
"scheduled_end": appt.scheduled_end.isoformat(),
|
|
"status": appt.status
|
|
} if appt else None,
|
|
"quote": {
|
|
"quote_id": quote.quote_id,
|
|
"version": quote.version,
|
|
"subtotal": float(quote.subtotal),
|
|
"tax": float(quote.tax),
|
|
"additional_damage_amount": float(quote.additional_damage_amount),
|
|
"total": float(quote.total),
|
|
"status": quote.status,
|
|
"reason": quote.reason,
|
|
"expires_at": quote.expires_at.isoformat() if quote.expires_at else None,
|
|
"additional_damage_description": inspection.additional_damage if inspection else None,
|
|
} if quote else None,
|
|
"events": [
|
|
{
|
|
"event_type": ev.event_type,
|
|
"timestamp": ev.timestamp.isoformat(),
|
|
"notes": ev.notes
|
|
} for ev in events
|
|
],
|
|
"payments": [
|
|
{
|
|
"payment_type": p.payment_type,
|
|
"amount": float(p.amount),
|
|
"status": p.status,
|
|
"paid_at": p.paid_at.isoformat() if p.paid_at else None
|
|
} for p in payments
|
|
]
|
|
}
|
|
|
|
@router.get("/jobs")
|
|
def list_service_jobs_admin(
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""List all repair & diagnostics jobs for dashboard queues."""
|
|
from app.models.ServiceModel import ServiceJob
|
|
from app.models.EcomCustomerModel import EcomCustomer
|
|
stmt = select(ServiceJob).order_by(ServiceJob.created_at.desc())
|
|
jobs = db.execute(stmt).scalars().all()
|
|
|
|
cust_ids = {j.customer_id for j in jobs if j.customer_id}
|
|
cust_map = {}
|
|
if cust_ids:
|
|
custs = db.execute(select(EcomCustomer).where(EcomCustomer.customer_id.in_(cust_ids))).scalars().all()
|
|
for c in custs:
|
|
cust_map[c.customer_id] = {
|
|
"name": f"{c.first_name} {c.last_name}".strip(),
|
|
"email": c.email,
|
|
"phone": c.phone or "N/A"
|
|
}
|
|
|
|
return [
|
|
{
|
|
"job_id": j.job_id,
|
|
"job_no": j.job_no,
|
|
"customer_id": j.customer_id,
|
|
"customer_name": cust_map.get(j.customer_id, {}).get("name") or (f"Customer ({j.customer_id[:8]})" if j.customer_id else "Guest Customer"),
|
|
"customer_email": cust_map.get(j.customer_id, {}).get("email") or "N/A",
|
|
"customer_phone": cust_map.get(j.customer_id, {}).get("phone") or "N/A",
|
|
"status": j.status,
|
|
"service_name": j.service_name_snapshot or (j.custom_service_name if j.custom_service_name else (j.service.name if j.service else "Custom Repair")),
|
|
"base_price": float(j.base_price_snapshot if j.base_price_snapshot is not None else (j.service.base_price if j.service else 0.0)),
|
|
"device_brand": j.device.brand if j.device else (j.brand_id or "Generic"),
|
|
"device_model": j.device.model if j.device else (j.model_id or "Device"),
|
|
"fulfillment_type": j.fulfillment_type or "WALK_IN",
|
|
"fulfillment_fee": float(j.fulfillment_fee or (250.0 if j.fulfillment_type == "DOORSTEP_PICKUP" else 0.0)),
|
|
"courier_name": getattr(j, "courier_name", None),
|
|
"awb_number": getattr(j, "awb_number", None),
|
|
"pickup_status": getattr(j, "pickup_status", None),
|
|
"delivery_address": j.delivery_address,
|
|
"created_at": j.created_at.isoformat()
|
|
} for j in jobs
|
|
]
|
|
|
|
@router.post("/jobs/{job_id}/logistics")
|
|
def update_service_job_logistics(
|
|
job_id: str,
|
|
payload: dict,
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Update courier name, AWB number, and pickup status for doorstep pickup / courier jobs."""
|
|
from app.models.ServiceModel import ServiceJob, ServiceJobEvent
|
|
import ulid
|
|
|
|
job = db.get(ServiceJob, job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
|
|
courier_name = payload.get("courier_name")
|
|
awb_number = payload.get("awb_number")
|
|
pickup_status = payload.get("pickup_status")
|
|
|
|
if courier_name is not None:
|
|
job.courier_name = courier_name
|
|
if awb_number is not None:
|
|
job.awb_number = awb_number
|
|
if pickup_status is not None:
|
|
job.pickup_status = pickup_status
|
|
|
|
note_parts = []
|
|
if courier_name: note_parts.append(f"Courier: {courier_name}")
|
|
if awb_number: note_parts.append(f"AWB: {awb_number}")
|
|
if pickup_status: note_parts.append(f"Pickup Status: {pickup_status}")
|
|
|
|
event = ServiceJobEvent(
|
|
event_id=str(ulid.ULID()),
|
|
job_id=job_id,
|
|
event_type="LOGISTICS_UPDATED",
|
|
performed_by=current_user.user_id,
|
|
notes="Logistics updated: " + ", ".join(note_parts) if note_parts else "Logistics details updated."
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
db.refresh(job)
|
|
return {
|
|
"status": "success",
|
|
"job_id": job.job_id,
|
|
"courier_name": job.courier_name,
|
|
"awb_number": job.awb_number,
|
|
"pickup_status": job.pickup_status
|
|
}
|
|
|
|
@router.post("/jobs/{job_id}/media")
|
|
def attach_service_job_media(
|
|
job_id: str,
|
|
payload: ServiceJobMediaBatchCreate,
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Attach media files (video/photos) under category (INSPECTION_DONE, READY_FOR_DELIVERY, etc.)."""
|
|
return inspection_service.batch_upload_job_media(db, job_id, payload.category, payload.file_ids)
|
|
|
|
@router.get("/jobs/{job_id}/media")
|
|
def get_service_job_media(
|
|
job_id: str,
|
|
category: Optional[str] = Query(None),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Get media proof uploaded for a service job."""
|
|
return inspection_service.get_job_media(db, job_id, category)
|
|
|
|
@router.post("/jobs/{job_id}/status")
|
|
def update_service_job_status_admin(
|
|
job_id: str,
|
|
status: str = Query(..., description="New status value"),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Admin/Technician manual status override for a service job."""
|
|
from app.models.ServiceModel import ServiceJob, ServiceJobEvent
|
|
import ulid
|
|
from datetime import datetime
|
|
|
|
job = db.get(ServiceJob, job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
|
|
job.status = status
|
|
|
|
# Credential Lifecycle: Purge lock credentials upon repair completion / delivery
|
|
if status in ["DELIVERED", "COMPLETED", "CLOSED", "READY_FOR_DELIVERY"]:
|
|
if job.lock_credential_encrypted:
|
|
job.lock_credential_encrypted = None
|
|
job.lock_credential_deleted_at = datetime.utcnow()
|
|
|
|
event = ServiceJobEvent(
|
|
event_id=str(ulid.ULID()),
|
|
job_id=job_id,
|
|
event_type=status,
|
|
performed_by=current_user.user_id,
|
|
notes=f"Status updated to {status} by technician."
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
db.refresh(job)
|
|
return {"status": "success", "new_status": job.status}
|
|
|
|
@router.get("/jobs/{job_id}/credentials")
|
|
def get_service_job_credentials_admin(
|
|
job_id: str,
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin", "Manager", "Technician"])),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Retrieve and decrypt customer device lock credentials with access logging audit."""
|
|
from app.models.ServiceModel import ServiceJob
|
|
from app.core.security.LockValidator import decrypt_credential
|
|
from datetime import datetime
|
|
|
|
job = db.get(ServiceJob, job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
|
|
if not job.lock_credential_encrypted:
|
|
return {
|
|
"lock_type": job.lock_type,
|
|
"passcode": None,
|
|
"is_deleted": job.lock_credential_deleted_at is not None,
|
|
"deleted_at": job.lock_credential_deleted_at
|
|
}
|
|
|
|
# Record access audit metadata
|
|
now = datetime.utcnow()
|
|
job.lock_credential_accessed_at = now
|
|
job.lock_credential_accessed_by = current_user.user_id
|
|
db.commit()
|
|
|
|
decrypted = decrypt_credential(job.lock_credential_encrypted)
|
|
|
|
return {
|
|
"lock_type": job.lock_type,
|
|
"passcode": decrypted,
|
|
"accessed_at": now,
|
|
"accessed_by": current_user.user_id
|
|
}
|
|
|
|
@router.get("/jobs/{job_id}/private-video")
|
|
def stream_private_condition_video(
|
|
job_id: str,
|
|
actor: Optional[dict] = Depends(get_optional_actor),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Access control endpoint for viewing private pre-courier device condition video."""
|
|
from app.models.ServiceModel import ServiceJob
|
|
from app.models.FileModel import FileUpload
|
|
from fastapi.responses import FileResponse
|
|
from pathlib import Path
|
|
|
|
job = db.get(ServiceJob, job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
|
|
if not job.pre_dispatch_video_id:
|
|
raise HTTPException(status_code=404, detail="No pre-dispatch video attached to this job")
|
|
|
|
# Authorization Check: Actor must be admin/technician or the customer owning the job
|
|
is_authorized = False
|
|
if actor:
|
|
if actor["type"] == "admin":
|
|
is_authorized = True
|
|
elif actor["type"] == "customer" and actor["obj"].customer_id == job.customer_id:
|
|
is_authorized = True
|
|
|
|
if not is_authorized:
|
|
raise HTTPException(status_code=403, detail="Not authorized to access this private video")
|
|
|
|
# Fetch file record
|
|
file_record = db.query(FileUpload).filter(FileUpload.file_id == job.pre_dispatch_video_id).first()
|
|
if not file_record:
|
|
raise HTTPException(status_code=404, detail="Video file record not found")
|
|
|
|
file_path = Path(file_record.storage_path.lstrip("/"))
|
|
if not file_path.is_absolute():
|
|
from app.core.config import BACKEND_ROOT
|
|
file_path = BACKEND_ROOT / file_path
|
|
|
|
if not file_path.exists():
|
|
raise HTTPException(status_code=404, detail="Video media file missing on server disk")
|
|
|
|
return FileResponse(file_path, media_type=file_record.mime_type or "video/mp4")
|