Daniel Bedeleanu
9fd20fff7c
- Create QuantityDisplay component with tap-to-edit UI + +/- buttons
- Implement useQuantityAdjustment hook with optimistic updates, debouncing
- Add PATCH /items/{itemId} backend endpoint with audit logging
- Components support inline quantity adjustment without modal friction
866 lines
33 KiB
Python
866 lines
33 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, strip_exif_orientation
|
|
from ..services.image_storage import save_image, get_unique_filename
|
|
from ..logger import log
|
|
|
|
# [H-02] Rate limiter for extract-label endpoint
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
|
|
router = APIRouter(
|
|
prefix="/items",
|
|
tags=["Items"]
|
|
)
|
|
|
|
@router.get("/search")
|
|
def search_items(
|
|
q: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[PHASE-5-T1] Search items across all text fields (name, part_number, barcode, description, category, notes).
|
|
|
|
Real-time search with relevance scoring. Returns max 50 results.
|
|
- q: query string (min 1 char, max 100 chars)
|
|
- Returns: List of matching items ordered by relevance
|
|
"""
|
|
# Validate query
|
|
if not q or len(q) < 1 or len(q) > 100:
|
|
return []
|
|
|
|
query_lower = q.lower()
|
|
|
|
# Get all items (we'll do client-side scoring for flexible matching)
|
|
all_items = db.query(models.Item).all()
|
|
|
|
# Score each item based on matches across all text fields
|
|
scored_items = []
|
|
for item in all_items:
|
|
score = 0
|
|
|
|
# Name match (highest priority)
|
|
if item.name:
|
|
item_name_lower = item.name.lower()
|
|
if query_lower == item_name_lower:
|
|
score += 500 # Exact match
|
|
elif item_name_lower.startswith(query_lower):
|
|
score += 250 # Prefix match
|
|
elif query_lower in item_name_lower:
|
|
score += 100 # Substring match
|
|
|
|
# Part number match
|
|
if item.part_number:
|
|
pn_lower = item.part_number.lower()
|
|
if query_lower == pn_lower:
|
|
score += 200
|
|
elif pn_lower.startswith(query_lower):
|
|
score += 150
|
|
elif query_lower in pn_lower:
|
|
score += 50
|
|
|
|
# Barcode match
|
|
if item.barcode:
|
|
barcode_lower = item.barcode.lower()
|
|
if query_lower == barcode_lower:
|
|
score += 180
|
|
elif query_lower in barcode_lower:
|
|
score += 40
|
|
|
|
# Description match
|
|
if item.description:
|
|
desc_lower = item.description.lower()
|
|
if query_lower in desc_lower:
|
|
score += 30
|
|
|
|
# Category match
|
|
if item.category:
|
|
cat_lower = item.category.lower()
|
|
if query_lower in cat_lower:
|
|
score += 20
|
|
|
|
# Type match
|
|
if item.type:
|
|
type_lower = item.type.lower()
|
|
if query_lower in type_lower:
|
|
score += 15
|
|
|
|
# OCR text / specs
|
|
if item.ocr_text:
|
|
ocr_lower = item.ocr_text.lower()
|
|
if query_lower in ocr_lower:
|
|
score += 10
|
|
|
|
if score > 0:
|
|
scored_items.append((item, score))
|
|
|
|
# Sort by score (descending), then by name for consistency
|
|
scored_items.sort(key=lambda x: (-x[1], x[0].name or ""))
|
|
|
|
# Return top 50 results
|
|
results = [item for item, score in scored_items[:50]]
|
|
return results
|
|
|
|
@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."
|
|
)
|
|
|
|
# Strip EXIF orientation so Gemini analyzes raw image (not rotated)
|
|
# Backend will process the same raw image
|
|
contents_no_exif = strip_exif_orientation(contents)
|
|
log.info(f"[EXTRACT] Sending {len(contents_no_exif)} bytes to Gemini (EXIF orientation stripped)")
|
|
|
|
result = extract_label_info(contents_no_exif, mode=mode)
|
|
log.info(f"[EXTRACT] Gemini returned: {type(result).__name__}")
|
|
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()
|
|
|
|
# Exclude image_processing fields from database item creation (backward compatible)
|
|
item_data = item.model_dump(exclude={"extracted_image_bytes", "image_processing"})
|
|
db_item = models.Item(**item_data)
|
|
db.add(db_item)
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
|
|
# NEW: Auto-save photo if extracted_image_bytes and image_processing provided
|
|
if item.extracted_image_bytes and item.image_processing:
|
|
try:
|
|
import base64
|
|
image_bytes = base64.b64decode(item.extracted_image_bytes)
|
|
log.info(f"[CREATE_ITEM] Received {len(image_bytes)} bytes for photo processing (base64 decoded)")
|
|
|
|
# Strip EXIF orientation to match what Gemini analyzed
|
|
image_bytes = strip_exif_orientation(image_bytes)
|
|
log.info(f"[CREATE_ITEM] After EXIF strip: {len(image_bytes)} bytes")
|
|
|
|
photo_result = _auto_save_photo_from_extraction(
|
|
item_id=db_item.id,
|
|
image_bytes=image_bytes,
|
|
crop_bounds=item.image_processing.get("crop_bounds"),
|
|
rotation_degrees=item.image_processing.get("rotation_degrees", 0),
|
|
db=db
|
|
)
|
|
|
|
if photo_result["status"] == "ok":
|
|
db.refresh(db_item) # Reload to get updated photo fields
|
|
else:
|
|
log.warning(f"Photo auto-save skipped for item {db_item.id}: {photo_result.get('reason')}")
|
|
except Exception as e:
|
|
log.error(f"Exception during auto-save for item {db_item.id}: {e}")
|
|
# Don't fail item creation
|
|
|
|
# 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.patch("/{item_id}", response_model=schemas.Item)
|
|
def update_item_quantity(
|
|
item_id: int,
|
|
body: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: auth.TokenData = Depends(auth.get_current_user)
|
|
):
|
|
"""[PHASE-5-T4] Update item quantity via PATCH. Supports direct quantity adjustment without modal."""
|
|
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")
|
|
|
|
# Extract and validate quantity from body
|
|
if "quantity" not in body:
|
|
raise HTTPException(status_code=400, detail="Missing 'quantity' field")
|
|
|
|
try:
|
|
new_quantity = int(body["quantity"])
|
|
except (ValueError, TypeError):
|
|
raise HTTPException(status_code=400, detail="Quantity must be an integer")
|
|
|
|
if new_quantity < 0:
|
|
raise HTTPException(status_code=400, detail="Quantity must be non-negative")
|
|
|
|
# Record old quantity for audit log
|
|
old_quantity = db_item.quantity
|
|
quantity_delta = new_quantity - old_quantity
|
|
|
|
# Update quantity
|
|
db_item.quantity = new_quantity
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
|
|
# Create audit log entry
|
|
audit = models.AuditLog(
|
|
user_id=current_user.sub,
|
|
action="UPDATE_QUANTITY",
|
|
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,
|
|
quantity_change=quantity_delta,
|
|
details=f"Quantity: {old_quantity} → {new_quantity}"
|
|
)
|
|
db.add(audit)
|
|
db.commit()
|
|
|
|
log.info(f"[PATCH /items/{item_id}] USER[{current_user.sub}] Updated quantity: {old_quantity} → {new_quantity}")
|
|
|
|
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()
|
|
|
|
# [CLEANUP] Delete associated image files
|
|
if db_item.photo_path:
|
|
try:
|
|
photo_file = Path(db_item.photo_path.lstrip("/"))
|
|
if photo_file.exists():
|
|
photo_file.unlink()
|
|
log.info(f"Deleted photo file: {db_item.photo_path}")
|
|
except Exception as e:
|
|
log.warning(f"Failed to delete photo file {db_item.photo_path}: {e}")
|
|
|
|
if db_item.photo_thumbnail_path:
|
|
try:
|
|
thumb_file = Path(db_item.photo_thumbnail_path.lstrip("/"))
|
|
if thumb_file.exists():
|
|
thumb_file.unlink()
|
|
log.info(f"Deleted thumbnail file: {db_item.photo_thumbnail_path}")
|
|
except Exception as e:
|
|
log.warning(f"Failed to delete thumbnail file {db_item.photo_thumbnail_path}: {e}")
|
|
|
|
# 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)}"
|
|
)
|
|
|
|
|
|
def _auto_save_photo_from_extraction(
|
|
item_id: int,
|
|
image_bytes: bytes,
|
|
crop_bounds: Optional[Dict[str, int]],
|
|
rotation_degrees: Optional[float],
|
|
db: Session
|
|
) -> Dict[str, str]:
|
|
"""
|
|
Helper function to save extracted photos with AI-guided crop/rotation.
|
|
|
|
This function is called after item creation if image_processing metadata exists.
|
|
It gracefully handles missing/invalid data without throwing exceptions.
|
|
|
|
Args:
|
|
item_id: ID of the item to attach the photo to
|
|
image_bytes: Raw photo bytes
|
|
crop_bounds: Optional crop bounds dict {x, y, width, height} (in pixels)
|
|
rotation_degrees: Optional rotation in degrees (-360 to +360, clockwise)
|
|
db: SQLAlchemy session
|
|
|
|
Returns:
|
|
{status: "ok"} if photo saved successfully
|
|
{status: "skipped", reason: "..."} if data invalid or missing
|
|
|
|
Behavior:
|
|
- Validates crop_bounds (all keys present, all ints >= 0)
|
|
- Validates rotation_degrees (numeric, -360 to +360)
|
|
- Skips gracefully if crop_bounds is None (no exceptions)
|
|
- Skips gracefully on invalid data (logs warning, returns skipped)
|
|
- Updates item.photo_path, photo_thumbnail_path, photo_upload_date
|
|
- Never throws exceptions
|
|
"""
|
|
from ..logger import log
|
|
|
|
try:
|
|
# Validate item exists
|
|
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
|
|
if not db_item:
|
|
log.warning(f"Auto-save photo: Item {item_id} not found, skipping")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Item {item_id} not found"
|
|
}
|
|
|
|
# Validate image_bytes
|
|
if not image_bytes or len(image_bytes) == 0:
|
|
log.warning(f"Auto-save photo for item {item_id}: No image bytes provided")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": "Empty image bytes"
|
|
}
|
|
|
|
# Validate crop_bounds (if provided)
|
|
crop_bounds_validated = None
|
|
if crop_bounds is not None:
|
|
if not isinstance(crop_bounds, dict):
|
|
log.warning(f"Auto-save photo for item {item_id}: crop_bounds is not a dict, skipping")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": "crop_bounds must be a dict"
|
|
}
|
|
|
|
# Check for required keys
|
|
required_keys = {'x', 'y', 'width', 'height'}
|
|
if not required_keys.issubset(crop_bounds.keys()):
|
|
missing = required_keys - set(crop_bounds.keys())
|
|
log.warning(f"Auto-save photo for item {item_id}: Missing crop_bounds keys: {missing}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Missing crop_bounds keys: {missing}"
|
|
}
|
|
|
|
# Validate all values are integers >= 0
|
|
try:
|
|
crop_bounds_validated = {}
|
|
for key in required_keys:
|
|
val = crop_bounds[key]
|
|
# Convert to int if it's numeric
|
|
if isinstance(val, (int, float)):
|
|
int_val = int(val)
|
|
else:
|
|
raise ValueError(f"Non-numeric value for {key}: {val}")
|
|
|
|
if int_val < 0:
|
|
raise ValueError(f"Negative value for {key}: {int_val}")
|
|
|
|
crop_bounds_validated[key] = int_val
|
|
|
|
except (ValueError, TypeError) as e:
|
|
log.warning(f"Auto-save photo for item {item_id}: Invalid crop_bounds: {str(e)}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Invalid crop_bounds: {str(e)}"
|
|
}
|
|
|
|
# Validate rotation_degrees (optional but if provided, must be valid)
|
|
if rotation_degrees is not None:
|
|
try:
|
|
rot = float(rotation_degrees)
|
|
if rot < -360 or rot > 360:
|
|
log.warning(f"Auto-save photo for item {item_id}: rotation_degrees {rot} out of range [-360, 360]")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"rotation_degrees {rot} out of range [-360, 360]"
|
|
}
|
|
except (ValueError, TypeError) as e:
|
|
log.warning(f"Auto-save photo for item {item_id}: Invalid rotation_degrees: {str(e)}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Invalid rotation_degrees: {str(e)}"
|
|
}
|
|
|
|
# All validation passed, proceed with processing
|
|
try:
|
|
# Save original image (EXIF-stripped, unprocessed) for debugging
|
|
category = db_item.category or "items"
|
|
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}"
|
|
debug_filename = get_unique_filename(filename_base, category, existing_files, variant="debug_original")
|
|
|
|
try:
|
|
original_debug_path = save_image(image_bytes, category, debug_filename.replace("_debug_original.jpg", ""), variant="debug_original")
|
|
log.info(f"[AUTO_SAVE] Saved original image for debugging: {original_debug_path}")
|
|
except Exception as e:
|
|
log.warning(f"[AUTO_SAVE] Failed to save debug original image: {e}")
|
|
original_debug_path = None
|
|
|
|
# Process image (crop + rotation + compression + thumbnail)
|
|
processor = ImageProcessor()
|
|
log.info(f"[AUTO_SAVE] Processing photo: crop_bounds={crop_bounds_validated}, rotation_degrees={rotation_degrees or 0}")
|
|
process_result = processor.process_photo(image_bytes, crop_bounds_validated, rotation_degrees=rotation_degrees or 0)
|
|
log.info(f"[AUTO_SAVE] Processing result: status={process_result.get('status')}")
|
|
|
|
if process_result.get('status') != 'success':
|
|
error_msg = process_result.get('error', 'Unknown error')
|
|
log.warning(f"Auto-save photo for item {item_id}: Image processing failed: {error_msg}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Image processing failed: {error_msg}"
|
|
}
|
|
|
|
# Get processed bytes
|
|
cropped_bytes = process_result.get('cropped_image_bytes')
|
|
thumbnail_bytes = process_result.get('thumbnail_bytes')
|
|
|
|
if not cropped_bytes or not thumbnail_bytes:
|
|
log.warning(f"Auto-save photo for item {item_id}: No image data from processing")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": "No image data from processing"
|
|
}
|
|
|
|
# Get category for file storage
|
|
category = db_item.category or "items"
|
|
|
|
# Get unique filenames
|
|
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:
|
|
log.warning(f"Auto-save photo for item {item_id}: Failed to save image: {str(e)}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": 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:
|
|
log.warning(f"Auto-save photo for item {item_id}: Failed to save thumbnail: {str(e)}")
|
|
# Clean up original if thumbnail save fails
|
|
try:
|
|
old_photo = Path(original_path.lstrip("/"))
|
|
if old_photo.exists():
|
|
old_photo.unlink()
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Failed to save thumbnail: {str(e)}"
|
|
}
|
|
|
|
# Update database
|
|
db_item.photo_path = original_path
|
|
db_item.photo_thumbnail_path = thumbnail_path
|
|
db_item.photo_upload_date = datetime.now(timezone.utc)
|
|
|
|
# Store image processing metadata in labels_data (including original debug image path)
|
|
if not db_item.labels_data:
|
|
db_item.labels_data = "{}"
|
|
|
|
try:
|
|
labels = json.loads(db_item.labels_data)
|
|
except (json.JSONDecodeError, TypeError):
|
|
labels = {}
|
|
|
|
if "image_processing" not in labels:
|
|
labels["image_processing"] = {}
|
|
|
|
# Add the original image path for debugging
|
|
labels["image_processing"]["original_photo_path"] = original_debug_path
|
|
labels["image_processing"]["crop_bounds"] = crop_bounds_validated
|
|
labels["image_processing"]["rotation_degrees"] = rotation_degrees or 0
|
|
|
|
db_item.labels_data = json.dumps(labels)
|
|
log.info(f"[AUTO_SAVE] Updated labels_data with original_photo_path: {original_debug_path}")
|
|
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
|
|
log.info(f"Auto-save photo for item {item_id}: Success")
|
|
return {
|
|
"status": "ok"
|
|
}
|
|
|
|
except Exception as e:
|
|
db.rollback()
|
|
log.warning(f"Auto-save photo for item {item_id}: Unexpected error: {str(e)}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Unexpected error: {str(e)}"
|
|
}
|
|
|
|
except Exception as e:
|
|
# Catch-all for any unexpected errors (never throw)
|
|
log.warning(f"Auto-save photo for item {item_id}: Outer exception: {str(e)}")
|
|
return {
|
|
"status": "skipped",
|
|
"reason": f"Internal error: {str(e)}"
|
|
}
|