""" 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 """ try: # Controlla sessione Flask (se disponibile) 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 except RuntimeError: # Session non disponibile (fuori da request context) # Fallback: richiedi sempre nuovo token logger.debug("Flask session non disponibile, richiedo nuovo 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 (se disponibile) try: session['cassanova_token'] = new_token session['cassanova_token_expires'] = ( datetime.now() + timedelta(seconds=expires_in) ).isoformat() logger.info(f"Nuovo token salvato in session, scade tra {expires_in}s") except RuntimeError: logger.debug("Session non disponibile, token non salvato") 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 get_stocks(self, salespoint_id: int, batch_size: int = 100) -> Dict: """ Scarica giacenze per un salespoint. Args: salespoint_id: ID del punto vendita batch_size: Numero di risultati per pagina Returns: {'stocks': [...], 'totalCount': N} """ try: data = self._request( 'GET', f'/stocks/{salespoint_id}', params={'start': 0, 'limit': batch_size} ) logger.info(f"Scaricate giacenze per salespoint {salespoint_id}: {len(data.get('stocks', []))} items") return data except Exception as e: logger.error(f"Errore get_stocks: {e}") return {'stocks': [], 'totalCount': 0} 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 app.models.inventory 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() logger.info(f"Scaricati {len(products)} prodotti da Cassanova") cached = 0 errors = 0 products_with_barcodes = 0 for idx, product in enumerate(products): # I prodotti hanno un array 'barcodes', non un singolo 'barcode' barcodes_list = product.get('barcodes', []) if idx < 3: # Log primi 3 prodotti logger.info(f"Prodotto {idx}: {product.get('description')}, barcodes: {barcodes_list}") # Se il prodotto ha barcode, salva in cache per ogni barcode if barcodes_list: products_with_barcodes += 1 for barcode_obj in barcodes_list: # barcode_obj è dict: {value: "...", format: "EAN13", salable: true} barcode_value = barcode_obj.get('value') if isinstance(barcode_obj, dict) else barcode_obj if barcode_value: try: result = self.cache_model.cache_product(tenant_id, str(barcode_value), product) if result: cached += 1 if cached <= 3: # Log primi 3 salvataggi logger.info(f"✓ Salvato barcode {barcode_value}") else: logger.warning(f"cache_product ritornò False per {barcode_value}") errors += 1 except Exception as e: logger.error(f"Errore cache barcode {barcode_value}: {e}") errors += 1 else: # Prodotto senza barcode - non lo mettiamo in cache if idx < 3: logger.debug(f"Prodotto {product.get('id')} senza barcode") # Gestisci anche prodotti multivariant (hanno barcodes nelle variants) if product.get('multivariant') and product.get('variants'): for variant in product.get('variants', []): variant_barcodes = variant.get('barcodes', []) for barcode_obj in variant_barcodes: barcode_value = barcode_obj.get('value') if isinstance(barcode_obj, dict) else barcode_obj if barcode_value: try: # Salva prodotto + info variante product_with_variant = product.copy() product_with_variant['selected_variant'] = variant result = self.cache_model.cache_product(tenant_id, str(barcode_value), product_with_variant) if result: cached += 1 else: errors += 1 except Exception as e: logger.error(f"Errore cache variante barcode {barcode_value}: {e}") errors += 1 result = { 'total': len(products), 'cached': cached, 'errors': errors } logger.info(f"Sincronizzazione completata: {result}") logger.info(f"Prodotti con barcodes: {products_with_barcodes}") return result # ==================================================================== # RICERCA PRODOTTI # ==================================================================== def search_product(self, tenant_id: int, barcode: str = None, description: str = None) -> List[Dict]: """ Cerca prodotti per barcode/ID o nome. IMPORTANTE: Usa SOLO cache locale (nessuna chiamata API). I prodotti devono essere pre-sincronizzati tramite sync_all_products(). Args: tenant_id: ID tenant barcode: Codice a barre o ID prodotto (ricerca esatta) description: Nome prodotto (ricerca parziale) Returns: Lista prodotti trovati (dalla cache locale) """ if barcode: # RICERCA PER BARCODE: SOLO CACHE LOCALE cached = self.cache_model.get_cached_product(tenant_id, barcode) if cached: logger.debug(f"Prodotto {barcode} trovato in cache locale") return [cached] # Non trovato in cache - ritorna vuoto logger.warning(f"Barcode {barcode} non trovato in cache locale. Sincronizza dalla dashboard.") return [] elif description: # RICERCA PER NOME: SOLO CACHE LOCALE products = self.cache_model.search_products_local(tenant_id, description, limit=20) return products else: return [] # ==================================================================== # SALES POINTS (SEDI) # ==================================================================== def get_salespoints(self, tenant_id: int) -> List[Dict]: """ Recupera tutti i Sales Points disponibili dai prodotti sincronizzati. Args: tenant_id: ID tenant Returns: Lista sales points con conteggio prodotti """ import pymysql conn = pymysql.connect(**self.session_model.db_config) cursor = conn.cursor() try: # Estrai sales points unici dai prodotti in cache sql = """ SELECT JSON_EXTRACT(dati_cassanova, '$.idSalesPoint') as sp_id, COUNT(*) as product_count FROM inv_cache_prodotti WHERE id_tenant = %s GROUP BY sp_id ORDER BY product_count DESC """ cursor.execute(sql, (tenant_id,)) results = cursor.fetchall() salespoints = [] for row in results: sp_id = row[0] if sp_id: # Verifica se esiste in inv_sedi cursor.execute(""" SELECT nome, indirizzo, città FROM inv_sedi WHERE id = %s AND id_tenant = %s """, (sp_id, tenant_id)) sede_data = cursor.fetchone() if sede_data: name = sede_data[0] address = f"{sede_data[1]}, {sede_data[2]}" if sede_data[1] else None else: name = f"Sede {sp_id}" address = None salespoints.append({ 'id': int(sp_id), 'name': name, 'address': address, 'product_count': row[1] }) return salespoints finally: cursor.close() conn.close() def update_session_name(self, session_id: int, tenant_id: int, name: str) -> bool: """Aggiorna nome sessione.""" return self.session_model.update_session_name(session_id, tenant_id, name) # ==================================================================== # SESSIONI INVENTARIO # ==================================================================== def start_session(self, tenant_id: int, user_id: int, sede_id: int = 17117) -> 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, id_sede: int = None) -> Optional[int]: """Aggiunge item a sessione.""" return self.item_model.add_item( session_id, tenant_id, barcode, product_data, found_in_api, quantita, id_sede ) def get_session_items(self, session_id: int, tenant_id: int, id_sede: int = None) -> List[Dict]: """Ottiene items di una sessione filtrati per sede.""" return self.item_model.get_session_items(session_id, tenant_id, id_sede) 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) def update_item_quantity(self, item_id: int, session_id: int, new_quantity: int) -> bool: """Modifica quantità item.""" return self.item_model.update_item_quantity(item_id, session_id, new_quantity) def sync_stocks(self, tenant_id: int, salespoint_id: int) -> int: """ Sincronizza giacenze da Cassanova per un salespoint. Args: tenant_id: ID tenant salespoint_id: ID sede da sincronizzare Returns: Numero di giacenze sincronizzate """ try: # Chiama API Cassanova per get stocks import pymysql import json from datetime import datetime # Get stocks da Cassanova response = self.cassanova.get_stocks(salespoint_id) stocks = response.get('stocks', []) conn = pymysql.connect(**self.session_model.db_config) cursor = conn.cursor() synced = 0 for stock in stocks: sql = """ INSERT INTO inv_giacenze ( id_tenant, id_sede, id_prodotto, id_variante, quantita, quantita_in_arrivo, quantita_in_uscita, soglia_minima, gestisci_giacenza, data_sincronizzazione ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW() ) ON DUPLICATE KEY UPDATE quantita = VALUES(quantita), quantita_in_arrivo = VALUES(quantita_in_arrivo), quantita_in_uscita = VALUES(quantita_in_uscita), soglia_minima = VALUES(soglia_minima), gestisci_giacenza = VALUES(gestisci_giacenza), data_sincronizzazione = NOW() """ cursor.execute(sql, ( tenant_id, salespoint_id, stock.get('idProduct'), stock.get('idProductVariant'), stock.get('quantity', 0), stock.get('incomingQuantity', 0), stock.get('outgoingQuantity', 0), stock.get('warningLevel', 0), stock.get('manageStock', True) )) synced += 1 conn.commit() cursor.close() conn.close() logger.info(f"Sincronizzate {synced} giacenze per sede {salespoint_id}") return synced except Exception as e: logger.error(f"Errore sync_stocks: {e}") return 0 return InventoryService()