Daniel Bedeleanu
e46777b933
- Database: Added photo_path, photo_thumbnail_path, photo_upload_date fields
- Services: ImageStorage (file ops) + ImageProcessor (image processing pipeline)
- API: POST/PUT /api/items/{id}/photo upload endpoints with validation
- API: GET /api/items/{id} returns photo URLs
- Static: FastAPI StaticFiles mount for /images/ directory
- Tests: 127+ comprehensive tests across all components
- Security: Fixed race conditions, path traversal, crop validation
- Ready for Phase 2 (frontend UI)
Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
426 lines
15 KiB
Python
426 lines
15 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File, Request
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func
|
|
from typing import List, Optional, Dict
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from slowapi import Limiter
|
|
from slowapi.util import get_remote_address
|
|
from pathlib import Path
|
|
from .. import models, schemas, auth
|
|
from ..database import get_db
|
|
from ..services.image_processing import ImageProcessor
|
|
from ..services.image_storage import save_image, get_unique_filename
|
|
|
|
# [H-02] Rate limiter for extract-label endpoint
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
|
|
router = APIRouter(
|
|
prefix="/items",
|
|
tags=["Items"]
|
|
)
|
|
|
|
@router.get("/stats")
|
|
def read_item_stats(
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Item statistics — only for authenticated users."""
|
|
total_categories = db.query(models.Category).count()
|
|
total_items = db.query(models.Item).count()
|
|
|
|
# Count items per category string
|
|
items_per_category = db.query(models.Item.category, func.count(models.Item.id))\
|
|
.group_by(models.Item.category).all()
|
|
|
|
return {
|
|
"total_categories": total_categories,
|
|
"total_items": total_items,
|
|
"items_distribution": {cat: count for cat, count in items_per_category if cat}
|
|
}
|
|
|
|
@router.get("/", response_model=List[schemas.Item])
|
|
def read_items(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] List of items — only for authenticated users."""
|
|
items = db.query(models.Item).offset(skip).limit(limit).all()
|
|
return items
|
|
|
|
@router.get("/{item_id}", response_model=schemas.Item)
|
|
def read_item(
|
|
item_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Get item — only for authenticated users. [PHASE1-T5] Include photo URLs if set."""
|
|
item = db.query(models.Item).filter(models.Item.id == item_id).first()
|
|
if item is None:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
# Build photo object if photo_path is set
|
|
if item.photo_path and item.photo_thumbnail_path and item.photo_upload_date:
|
|
item.photo = schemas.PhotoResponse(
|
|
thumbnail_url=item.photo_thumbnail_path,
|
|
full_url=item.photo_path,
|
|
uploaded_at=item.photo_upload_date
|
|
)
|
|
else:
|
|
item.photo = None
|
|
|
|
return item
|
|
|
|
_ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
|
|
_MAX_IMAGE_SIZE = 10 * 1024 * 1024 # 10 MB
|
|
|
|
@limiter.limit("10/minute")
|
|
@router.post("/extract-label")
|
|
async def extract_label(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
mode: str = "item", # 'item' or 'box'
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Extract label from image — only for authenticated users. [H-02] Rate limit: 10 req/min per IP."""
|
|
from ..ai_vision import extract_label_info
|
|
|
|
# [SECURITY FIX H-03] Validate MIME type and maximum size
|
|
if file.content_type not in _ALLOWED_IMAGE_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
|
detail=f"File type not allowed: {file.content_type}. Accepted: {', '.join(_ALLOWED_IMAGE_TYPES)}"
|
|
)
|
|
|
|
contents = await file.read()
|
|
|
|
if len(contents) > _MAX_IMAGE_SIZE:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail="File exceeds 10MB limit."
|
|
)
|
|
|
|
result = extract_label_info(contents, mode=mode)
|
|
return result
|
|
|
|
@router.post("/", response_model=schemas.Item, status_code=status.HTTP_201_CREATED)
|
|
def create_item(
|
|
item: schemas.ItemCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Create item — only for authenticated users. [M-02] user_id from token."""
|
|
# [DUPLICATE CHECK] Prevent duplicate part numbers
|
|
if item.barcode:
|
|
existing = db.query(models.Item).filter(models.Item.barcode == item.barcode).first()
|
|
if existing:
|
|
# Return existing item data for user comparison
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={
|
|
"message": f"Item with Part Number '{item.barcode}' already exists in inventory.",
|
|
"existing_id": existing.id,
|
|
"existing_item": schemas.Item.model_validate(existing).model_dump(mode='json')
|
|
}
|
|
)
|
|
|
|
# [AUTO-PERSIST] Create Category/Color if not exists
|
|
if item.category:
|
|
cat = db.query(models.Category).filter(models.Category.name == item.category).first()
|
|
if not cat:
|
|
db.add(models.Category(name=item.category))
|
|
db.commit()
|
|
|
|
if item.color:
|
|
col = db.query(models.Color).filter(models.Color.name == item.color).first()
|
|
if not col:
|
|
db.add(models.Color(name=item.color))
|
|
db.commit()
|
|
|
|
db_item = models.Item(**item.model_dump())
|
|
db.add(db_item)
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
|
|
# Audit log the creation — [M-02] user_id from token, not from body
|
|
# Capture full snapshot
|
|
item_snapshot = {
|
|
"barcode": db_item.barcode,
|
|
"name": db_item.name,
|
|
"category": db_item.category,
|
|
"type": db_item.type,
|
|
"part_number": db_item.part_number,
|
|
"color": db_item.color,
|
|
"specs": db_item.specs,
|
|
"box_label": db_item.box_label,
|
|
"image_url": db_item.image_url
|
|
}
|
|
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="CREATE_ITEM",
|
|
target_item_id=db_item.id,
|
|
target_item_name=db_item.name,
|
|
target_item_pn=db_item.part_number,
|
|
target_item_barcode=db_item.barcode,
|
|
target_snapshot=json.dumps(item_snapshot),
|
|
quantity_change=item.quantity
|
|
)
|
|
db.add(audit)
|
|
db.commit()
|
|
|
|
return db_item
|
|
|
|
@router.put("/{item_id}", response_model=schemas.Item)
|
|
def update_item(
|
|
item_id: int,
|
|
item: schemas.ItemCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[C-01] Update item — only for authenticated users."""
|
|
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
|
|
if not db_item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
# [AUTO-PERSIST] Create Category/Color if not exists
|
|
if item.category:
|
|
cat = db.query(models.Category).filter(models.Category.name == item.category).first()
|
|
if not cat:
|
|
db.add(models.Category(name=item.category))
|
|
db.commit()
|
|
|
|
if item.color:
|
|
col = db.query(models.Color).filter(models.Color.name == item.color).first()
|
|
if not col:
|
|
db.add(models.Color(name=item.color))
|
|
db.commit()
|
|
|
|
update_data = item.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_item, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
return db_item
|
|
|
|
@router.delete("/{item_id}")
|
|
def delete_item(
|
|
item_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_admin)
|
|
):
|
|
"""[C-01] Delete item — only for authenticated users. InterventionItems are cleared; AuditLogs are KEPT for history."""
|
|
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
|
|
if not db_item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
# [AUDIT] Log the deletion to database and disk
|
|
from ..logger import log
|
|
log.warning(f"USER[{current_user.sub}] DELETING ITEM: ID={item_id}, Name={db_item.name}, PN={db_item.part_number}")
|
|
|
|
item_snapshot = {
|
|
"barcode": db_item.barcode,
|
|
"name": db_item.name,
|
|
"category": db_item.category,
|
|
"type": db_item.type,
|
|
"part_number": db_item.part_number,
|
|
"color": db_item.color,
|
|
"specs": db_item.specs,
|
|
"box_label": db_item.box_label,
|
|
"image_url": db_item.image_url,
|
|
"final_quantity": db_item.quantity
|
|
}
|
|
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="DELETE_ITEM",
|
|
target_item_id=item_id,
|
|
target_item_name=db_item.name,
|
|
target_item_pn=db_item.part_number,
|
|
target_item_barcode=db_item.barcode,
|
|
target_snapshot=json.dumps(item_snapshot),
|
|
details=f"Final Quantity: {db_item.quantity}"
|
|
)
|
|
db.add(audit)
|
|
|
|
# [CLEANUP] Delete related InterventionItems to prevent foreign key issues
|
|
db.query(models.InterventionItem).filter(models.InterventionItem.item_id == item_id).delete()
|
|
|
|
# Audit Logs in database are NOT deleted here to preserve history of actions
|
|
db.delete(db_item)
|
|
db.commit()
|
|
return {"message": "Item deleted successfully. History logs preserved."}
|
|
|
|
|
|
@router.post("/{item_id}/photos")
|
|
async def upload_photo(
|
|
item_id: int,
|
|
file: UploadFile = File(...),
|
|
crop_bounds: Optional[str] = "",
|
|
replace_existing: Optional[str] = "",
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""
|
|
[PHASE1-T4] Upload/replace photo for an item.
|
|
|
|
- Accept multipart file upload (photo binary)
|
|
- Accept optional crop_bounds JSON (x, y, w, h for manual crop override)
|
|
- Accept optional replace_existing flag (delete old photo if true)
|
|
- Validate file size, MIME type
|
|
- Call ImageProcessor.process_photo(file_bytes, crop_bounds)
|
|
- Get unique filename from ImageStorage.get_unique_filename()
|
|
- Save original and thumbnail using ImageStorage.save_image()
|
|
- Delete old photo file if replacing
|
|
- Update Item.photo_path, photo_thumbnail_path, photo_upload_date
|
|
- Return: {status: "ok", photo: {thumbnail_url, full_url, uploaded_at}}
|
|
"""
|
|
# Verify item exists
|
|
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
|
|
if not db_item:
|
|
raise HTTPException(status_code=404, detail="Item not found")
|
|
|
|
# Validate file type
|
|
if file.content_type not in _ALLOWED_IMAGE_TYPES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
|
detail=f"File type not allowed: {file.content_type}. Accepted: {', '.join(_ALLOWED_IMAGE_TYPES)}"
|
|
)
|
|
|
|
# Read file and validate size
|
|
file_bytes = await file.read()
|
|
if len(file_bytes) > _MAX_IMAGE_SIZE:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail="File exceeds 10MB limit."
|
|
)
|
|
|
|
try:
|
|
# Parse crop_bounds if provided
|
|
crop_bounds_dict = None
|
|
if crop_bounds and crop_bounds.strip():
|
|
try:
|
|
crop_bounds_dict = json.loads(crop_bounds)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid crop_bounds JSON"
|
|
)
|
|
|
|
# Parse replace_existing flag (comes as form string)
|
|
should_replace = replace_existing and replace_existing.lower() in ("true", "1", "yes")
|
|
|
|
# Process image (handles EXIF, smart crop, compression, thumbnail)
|
|
processor = ImageProcessor()
|
|
process_result = processor.process_photo(file_bytes, crop_bounds_dict)
|
|
|
|
if process_result['status'] != 'success':
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Image processing failed: {process_result.get('error', 'Unknown error')}"
|
|
)
|
|
|
|
# Get processed bytes
|
|
cropped_bytes = process_result['cropped_image_bytes']
|
|
thumbnail_bytes = process_result['thumbnail_bytes']
|
|
|
|
if not cropped_bytes or not thumbnail_bytes:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Failed to process image data"
|
|
)
|
|
|
|
# Get category for file storage (use "items" as default category if no category set)
|
|
category = db_item.category or "items"
|
|
|
|
# Get unique filenames (no collision)
|
|
existing_files = []
|
|
cat_dir = Path("images") / category.lower()
|
|
if cat_dir.exists():
|
|
existing_files = [f.name for f in cat_dir.iterdir() if f.is_file()]
|
|
|
|
filename_base = db_item.name or f"item_{item_id}"
|
|
original_filename = get_unique_filename(filename_base, category, existing_files, variant="original")
|
|
thumbnail_filename = get_unique_filename(filename_base, category, existing_files, variant="thumb")
|
|
|
|
# Save original image
|
|
try:
|
|
original_path = save_image(cropped_bytes, category, original_filename.replace("_original.jpg", ""), variant="original")
|
|
except (OSError, IOError) as e:
|
|
if "No space left" in str(e):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
|
|
detail="Disk space full"
|
|
)
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Failed to save image: {str(e)}"
|
|
)
|
|
|
|
# Save thumbnail
|
|
try:
|
|
thumbnail_path = save_image(thumbnail_bytes, category, thumbnail_filename.replace("_thumb.jpg", ""), variant="thumb")
|
|
except (OSError, IOError) as e:
|
|
if "No space left" in str(e):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
|
|
detail="Disk space full"
|
|
)
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Failed to save thumbnail: {str(e)}"
|
|
)
|
|
|
|
# Delete old photo files if replacing
|
|
if should_replace and db_item.photo_path:
|
|
try:
|
|
old_photo = Path(db_item.photo_path.lstrip("/"))
|
|
if old_photo.exists():
|
|
old_photo.unlink()
|
|
except Exception as e:
|
|
# Log but don't fail if cleanup fails
|
|
from ..logger import log
|
|
log.warning(f"Failed to delete old photo: {str(e)}")
|
|
|
|
try:
|
|
if db_item.photo_thumbnail_path:
|
|
old_thumb = Path(db_item.photo_thumbnail_path.lstrip("/"))
|
|
if old_thumb.exists():
|
|
old_thumb.unlink()
|
|
except Exception as e:
|
|
from ..logger import log
|
|
log.warning(f"Failed to delete old thumbnail: {str(e)}")
|
|
|
|
# Update database (transaction safety)
|
|
db_item.photo_path = original_path
|
|
db_item.photo_thumbnail_path = thumbnail_path
|
|
db_item.photo_upload_date = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
|
|
# Return response with URLs
|
|
return {
|
|
"status": "ok",
|
|
"photo": {
|
|
"thumbnail_url": db_item.photo_thumbnail_path,
|
|
"full_url": db_item.photo_path,
|
|
"uploaded_at": db_item.photo_upload_date
|
|
}
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
db.rollback()
|
|
from ..logger import log
|
|
log.error(f"Unexpected error in photo upload: {str(e)}", exc_info=True)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Internal server error: {str(e)}"
|
|
)
|