""" ENIGMA Inventario - Inventory Service FIXED ============================================ Service completo con: - Token Cassanova in Flask session (PERSISTENTE) - Integrazione con models che usano tabelle reali - Cache prodotti locale - Workflow completo inventario Author: HetGi & Claude Date: 2026-01-14 (FIXED VERSION) """ import requests import json from datetime import datetime, timedelta from typing import Optional, Dict, List, Any from flask import session import logging logger = logging.getLogger(__name__) class CassanovaService: """Service per API Cassanova con token in Flask session.""" def __init__(self, api_key: str, hostname: str = "https://api.cassanova.com"): self.api_key = api_key self.hostname = hostname self.api_version = "1.0.0" def _get_token(self) -> str: """ Ottiene token da Flask session o ne richiede uno nuovo. Returns: Token valido """ # Controlla sessione Flask token = session.get('cassanova_token') expires_at = session.get('cassanova_token_expires') if token and expires_at: expires_dt = datetime.fromisoformat(expires_at) if datetime.now() < (expires_dt - timedelta(seconds=60)): logger.debug("Token valido in sessione Flask") return token # Richiedi nuovo token logger.info("Richiesta nuovo token Cassanova...") url = f"{self.hostname}/apikey/token" headers = { "Content-Type": "application/json", "X-Requested-With": "*" } payload = {"apiKey": self.api_key} response = requests.post(url, headers=headers, json=payload, timeout=10) response.raise_for_status() data = response.json() new_token = data['access_token'] expires_in = data.get('expires_in', 3600) # Salva in Flask session session['cassanova_token'] = new_token session['cassanova_token_expires'] = ( datetime.now() + timedelta(seconds=expires_in) ).isoformat() logger.info(f"Nuovo token ottenuto, scade tra {expires_in}s") return new_token def _request(self, method: str, endpoint: str, **kwargs) -> Dict: """ Esegue richiesta HTTP con autenticazione. Args: method: GET, POST, etc. endpoint: /products, etc. **kwargs: params, json, etc. Returns: Response JSON """ token = self._get_token() url = f"{self.hostname}{endpoint}" headers = { "Content-Type": "application/json", "X-Version": self.api_version, "Authorization": f"Bearer {token}" } response = requests.request(method, url, headers=headers, timeout=30, **kwargs) response.raise_for_status() return response.json() def search_product_by_barcode(self, barcode: str) -> Optional[Dict]: """Cerca prodotto per barcode.""" try: data = self._request('GET', '/products', params={'barcode': barcode, 'start': 0, 'limit': 1}) products = data.get('products', []) return products[0] if products else None except Exception as e: logger.error(f"Errore ricerca barcode: {e}") return None def search_products(self, description: str = None, start: int = 0, limit: int = 100) -> Dict: """Cerca prodotti.""" try: params = {'start': start, 'limit': limit} if description: params['description'] = description return self._request('GET', '/products', params=params) except Exception as e: logger.error(f"Errore ricerca prodotti: {e}") return {'totalCount': 0, 'products': []} def get_all_products(self, batch_size: int = 100) -> List[Dict]: """Scarica TUTTI i prodotti (paginato).""" all_products = [] start = 0 try: while True: data = self.search_products(start=start, limit=batch_size) products = data.get('products', []) total = data.get('totalCount', 0) if not products: break all_products.extend(products) logger.info(f"Scaricati {len(all_products)}/{total} prodotti...") if len(all_products) >= total: break start += batch_size logger.info(f"Download completato: {len(all_products)} prodotti") return all_products except Exception as e: logger.error(f"Errore download prodotti: {e}") return all_products def create_inventory_service(db_config: Dict, cassanova_api_key: str): """ Factory per creare InventoryService. Args: db_config: {host, user, password, database} cassanova_api_key: API Key Cassanova Returns: InventoryService instance """ from models_inventory_fixed import InventorySession, InventoryItem, ProductCache class InventoryService: """Service completo inventario.""" def __init__(self): self.session_model = InventorySession(db_config) self.item_model = InventoryItem(db_config) self.cache_model = ProductCache(db_config) self.cassanova = CassanovaService(api_key=cassanova_api_key) # ==================================================================== # SINCRONIZZAZIONE PRODOTTI # ==================================================================== def sync_all_products(self, tenant_id: int) -> Dict[str, Any]: """ Sincronizza TUTTI i prodotti da Cassanova alla cache locale. Args: tenant_id: ID tenant Returns: { 'total': int, 'cached': int, 'errors': int } """ logger.info(f"Inizio sincronizzazione prodotti per tenant {tenant_id}") products = self.cassanova.get_all_products() cached = 0 errors = 0 for product in products: barcode = product.get('barcode') if barcode: try: self.cache_model.cache_product(tenant_id, barcode, product) cached += 1 except Exception as e: logger.error(f"Errore cache prodotto {barcode}: {e}") errors += 1 result = { 'total': len(products), 'cached': cached, 'errors': errors } logger.info(f"Sincronizzazione completata: {result}") return result # ==================================================================== # RICERCA PRODOTTI # ==================================================================== def search_product(self, tenant_id: int, barcode: str = None, description: str = None) -> List[Dict]: """ Cerca prodotti per barcode o nome. Prima cerca in cache locale, poi su Cassanova API. Args: tenant_id: ID tenant barcode: Codice a barre (ricerca esatta) description: Nome prodotto (ricerca parziale) Returns: Lista prodotti trovati """ if barcode: # Ricerca per barcode: cache → API cached = self.cache_model.get_cached_product(tenant_id, barcode) if cached: logger.debug(f"Prodotto {barcode} trovato in cache") return [cached] # Non in cache, cerca API product = self.cassanova.search_product_by_barcode(barcode) if product: # Salva in cache self.cache_model.cache_product(tenant_id, barcode, product) return [product] return [] elif description: # Ricerca per nome: solo API data = self.cassanova.search_products(description=description, limit=20) return data.get('products', []) else: return [] # ==================================================================== # SESSIONI INVENTARIO # ==================================================================== def start_session(self, tenant_id: int, user_id: int, sede_id: int = 1) -> Optional[int]: """Crea nuova sessione.""" return self.session_model.create_session(tenant_id, user_id, sede_id) def get_current_session(self, tenant_id: int, user_id: int) -> Optional[Dict]: """Ottiene sessione aperta corrente.""" return self.session_model.get_active_session(tenant_id, user_id) def get_session_details(self, session_id: int, tenant_id: int) -> Optional[Dict]: """Ottiene dettagli sessione.""" return self.session_model.get_session(session_id, tenant_id) def close_session(self, session_id: int, tenant_id: int, note: str = None) -> bool: """Chiude sessione.""" return self.session_model.close_session(session_id, tenant_id, note) def get_recent_sessions(self, tenant_id: int, limit: int = 10) -> List[Dict]: """Lista sessioni recenti.""" return self.session_model.get_recent_sessions(tenant_id, limit) # ==================================================================== # ITEM INVENTARIO # ==================================================================== def add_item(self, session_id: int, tenant_id: int, barcode: str, product_data: Optional[Dict] = None, found_in_api: bool = False, quantita: int = 1) -> Optional[int]: """Aggiunge item a sessione.""" return self.item_model.add_item( session_id, tenant_id, barcode, product_data, found_in_api, quantita ) def get_session_items(self, session_id: int, tenant_id: int) -> List[Dict]: """Ottiene items di una sessione.""" return self.item_model.get_session_items(session_id, tenant_id) def delete_item(self, item_id: int, session_id: int) -> bool: """Elimina item da sessione.""" return self.item_model.delete_item(item_id, session_id) return InventoryService()