"""Equipment sync service: import from Rentman, upsert into DB, invalidate cache.""" import logging import os import httpx from datetime import datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.equipment import EquipmentCache from app.models.sync_log import SyncLog from app.services.rentman_service import RentmanService from app.cache import cache logger = logging.getLogger(__name__) IMAGES_DIR = "/data/images/equipment" class SyncService: """Orchestrates equipment import from Rentman into the local database.""" def __init__(self, db: AsyncSession, rentman: RentmanService | None = None) -> None: self.db = db self.rentman = rentman or RentmanService() async def run_sync(self) -> dict[str, Any]: """Execute an incremental equipment sync. 1. Create sync_log entry (status=running) 2. Paginate GET /equipment from Rentman 3. Compare updateHash with DB, only process changed items 4. Download images only for changed items 5. Mark missing items as unavailable 6. Invalidate Redis cache (equipment:*) 7. Update sync_log (status=completed or failed) Returns dict with sync_id, items_processed, status. """ log_entry = SyncLog( sync_type="equipment", status="running", started_at=datetime.utcnow(), ) self.db.add(log_entry) await self.db.commit() await self.db.refresh(log_entry) sync_id = log_entry.id items_processed = 0 items_failed = 0 error_message: str | None = None try: # Ensure images directory exists os.makedirs(IMAGES_DIR, exist_ok=True) # Fetch all equipment from Rentman all_equipment = await self.rentman.get_all_equipment(limit=100) # Build set of rentman_ids from API for availability check api_rentman_ids = set() # Load all existing equipment from DB for hash comparison result = await self.db.execute(select(EquipmentCache)) existing_items = {item.rentman_id: item for item in result.scalars().all()} for raw_item in all_equipment: rentman_id = str(raw_item.get("id", "")) api_rentman_ids.add(rentman_id) new_hash = raw_item.get("updateHash", "") existing = existing_items.get(rentman_id) # Skip if hash unchanged and item exists if existing and existing.update_hash == new_hash and new_hash: continue try: transformed = await self.rentman.transform_equipment(raw_item) await self._upsert_equipment(transformed) # Download image if S3 URL available if transformed.get("images"): await self._download_image( transformed["rentman_id"], transformed["images"][0], existing.id if existing else None, ) items_processed += 1 except Exception as exc: logger.warning("Failed to upsert equipment item %s: %s", rentman_id, exc) items_failed += 1 # Mark missing items as unavailable for rentman_id, existing in existing_items.items(): if rentman_id not in api_rentman_ids and existing.available: existing.available = False logger.info("Marked equipment %s as unavailable (not in API)", rentman_id) await self.db.commit() # Invalidate Redis cache await cache.delete_pattern("equipment:*") status_val = "completed" except Exception as exc: logger.error("Equipment sync failed: %s", exc) error_message = str(exc) status_val = "failed" # Update log entry log_entry.status = status_val log_entry.items_processed = items_processed log_entry.items_failed = items_failed log_entry.error_message = error_message log_entry.completed_at = datetime.utcnow() await self.db.commit() return { "sync_id": sync_id, "items_processed": items_processed, "items_failed": items_failed, "status": status_val, } async def _upsert_equipment(self, data: dict[str, Any]) -> None: """Insert or update a single equipment row by rentman_id.""" result = await self.db.execute( select(EquipmentCache).where(EquipmentCache.rentman_id == data["rentman_id"]) ) existing = result.scalar_one_or_none() if existing: existing.name = data["name"] existing.number = data.get("number", "") existing.category = data.get("category", "") existing.subcategory = data.get("subcategory", "") existing.description = data.get("description", "") existing.specifications = data.get("specifications") existing.images = data.get("images") existing.update_hash = data.get("update_hash") existing.rental_price = data.get("rental_price") existing.brand = data.get("brand", "") existing.available = data.get("available", True) else: new_item = EquipmentCache( rentman_id=data["rentman_id"], name=data["name"], number=data.get("number", ""), category=data.get("category", ""), subcategory=data.get("subcategory", ""), description=data.get("description", ""), specifications=data.get("specifications"), images=data.get("images"), update_hash=data.get("update_hash"), rental_price=data.get("rental_price"), brand=data.get("brand", ""), available=data.get("available", True), ) self.db.add(new_item) async def _download_image(self, rentman_id: str, s3_url: str, db_id: int | None) -> None: """Download image from S3 URL and save locally. File path: /data/images/equipment/{db_id or rentman_id}.jpg """ # Determine the file identifier - prefer DB id if available # We need to flush to get the ID for new items if db_id is None: await self.db.flush() result = await self.db.execute( select(EquipmentCache).where(EquipmentCache.rentman_id == rentman_id) ) item = result.scalar_one_or_none() if item: db_id = item.id if db_id is None: logger.warning("Could not determine DB id for image download: %s", rentman_id) return image_path = os.path.join(IMAGES_DIR, f"{db_id}.jpg") # Skip if already exists if os.path.exists(image_path): return try: async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.get(s3_url) resp.raise_for_status() with open(image_path, "wb") as f: f.write(resp.content) logger.info("Downloaded image for equipment %s -> %s", db_id, image_path) except Exception as exc: logger.warning("Failed to download image for %s: %s", rentman_id, exc) async def get_last_sync(self) -> dict[str, Any]: """Return the most recent sync_log entry summary.""" result = await self.db.execute( select(SyncLog).order_by(SyncLog.started_at.desc()).limit(1) ) log = result.scalar_one_or_none() if not log: return {"last_sync": None, "items_processed": 0, "status": "never"} return { "last_sync": log.started_at, "items_processed": log.items_processed, "status": log.status, } async def get_sync_log_paginated(self, page: int = 1, page_size: int = 20) -> dict[str, Any]: """Return paginated sync log entries.""" offset = (page - 1) * page_size result = await self.db.execute( select(SyncLog) .order_by(SyncLog.started_at.desc()) .offset(offset) .limit(page_size) ) logs = result.scalars().all() count_result = await self.db.execute(select(SyncLog)) total = len(count_result.scalars().all()) return { "items": logs, "total": total, "page": page, "page_size": page_size, "total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0, }