Update src/graph/graph_loader.py

src/graph/graph_loader.py

@@ -10,14 +10,18 @@ class KnowledgeGraphPersister:
         """
         Inizializza il driver Neo4j usando le variabili d'ambiente per sicurezza.
         """
-        uri = os.getenv("NEO4J_URI", "bolt://localhost:7687")
+        uri = os.getenv("NEO4J_URI", "")
         user = os.getenv("NEO4J_USER", "neo4j")
-        password = os.getenv("NEO4J_PASSWORD", "activa_semantic_lab")
+        password = os.getenv("NEO4J_PASSWORD", "")
         
-        try:
+       try:
             self.driver = GraphDatabase.driver(uri, auth=(user, password))
             self.driver.verify_connectivity()
-            print(f"✅ Connesso a Neo4j ({uri}) successfully.")
+            print(f"✅ Connesso a Neo4j ({uri}).")
+            
+            # Creazione indici all'avvio (Fondamentale per la velocità dei MERGE)
+            self._create_constraints()
+            
         except Exception as e:
             print(f"❌ Errore critico connessione Neo4j: {e}")
             self.driver = None
@@ -26,90 +30,93 @@ class KnowledgeGraphPersister:
         if self.driver:
             self.driver.close()
 
+    def _create_constraints(self):
+        """
+        Crea un vincolo di unicità sulla proprietà URI.
+        Senza questo, MERGE diventa lentissimo (Full Table Scan).
+        """
+        if not self.driver: return
+        query = "CREATE CONSTRAINT resource_uri_unique IF NOT EXISTS FOR (n:Resource) REQUIRE n.uri IS UNIQUE"
+        with self.driver.session() as session:
+            try:
+                session.run(query)
+                print("⚡ Vincoli/Indici Neo4j verificati.")
+            except Exception as e:
+                print(f"⚠️ Warning creazione indici: {e}")
+
     def sanitize_name(self, name):
         """
-        Normalizza i nomi per creare URI coerenti (Canonicalization base).
+        Canonicalization base.
         """
         if not name: return "Unknown"
-        # Rimuove caratteri speciali e spazi extra, mantiene coerenza maiuscole/minuscole
+        # Rimuove spazi extra e normalizza.
+        # Nota: In produzione usare slugify o urllib.parse.quote per URI robusti
         return name.strip().replace(" ", "_").replace("'", "").replace('"', "")
 
+    def sanitize_predicate(self, pred):
+        """
+        Pulisce il predicato per evitare Cypher Injection, dato che non può essere parametrizzato.
+        """
+        # Accetta solo caratteri alfanumerici e underscore. Upper case per convenzione Neo4j.
+        clean = "".join(x for x in pred if x.isalnum() or x == "_")
+        return clean.upper() if clean else "RELATED_TO"
+
     def save_triples(self, triples):
         """
-        Salva le triple in BATCH (ottimizzazione performance).
-        Usa UNWIND per processare liste di dati in un'unica transazione.
+        Salva le triple usando VERO Batching (UNWIND).
+        Raggruppa le triple per predicato per aggirare il limite di parametrizzazione delle relazioni.
         """
-        if not self.driver:
-            print("⚠️ Driver non connesso. Impossibile salvare.")
+        if not self.driver or not triples:
             return
 
-        if not triples:
-            return
+        print(f"💾 Preparazione Batch di {len(triples)} triple...")
 
-        print(f"💾 Salvataggio BATCH di {len(triples)} triple su Neo4j...")
+        # 1. Raggruppamento per Predicato
+        # Struttura: { "LOCATED_IN": [ {subj:..., obj:..., ...}, ... ], "HAS_TYPE": [...] }
+        batched_by_pred = defaultdict(list)
         
-        # 1. Prepariamo i dati come lista di dizionari (Payload leggero)
-        batch_data = []
         for t in triples:
-            batch_data.append({
+            safe_pred = self.sanitize_predicate(t.predicate)
+            
+            item = {
                 "subj_uri": self.sanitize_name(t.subject),
                 "subj_label": t.subject,
-                "pred": t.predicate, # Nota: Il predicato dinamico richiede attenzione in Cypher
                 "obj_uri": self.sanitize_name(t.object),
                 "obj_label": t.object,
-                "conf": t.confidence,
-                "src": t.source
-            })
+                "conf": float(t.confidence), # Assicura float nativo
+                "src": t.source or "unknown"
+            }
+            batched_by_pred[safe_pred].append(item)
 
-        # 2. Query Batch Ottimizzata
-        # Nota: In Cypher non si può parametrizzare il TIPO di relazione (es. :RELAZIONE).
-        # Per performance pura con relazioni dinamiche, usiamo APOC o un approccio ibrido.
-        # Qui usiamo un approccio sicuro iterando nel driver ma con transazione unica,
-        # oppure raggruppiamo per tipo di relazione. 
-        
-        # Approccio Migliore per MVP: Transazione singola
+        # 2. Esecuzione Transazioni (Una per tipo di relazione)
         with self.driver.session() as session:
-            try:
-                session.execute_write(self._batch_write_tx, batch_data)
-                print("✅ Batch completato.")
-            except Exception as e:
-                print(f"⚠️ Errore durante il salvataggio batch: {e}")
+            for pred, data_list in batched_by_pred.items():
+                try:
+                    session.execute_write(self._unwind_write_tx, pred, data_list)
+                    print(f"   -> Inserite {len(data_list)} relazioni :{pred}")
+                except Exception as e:
+                    print(f"⚠️ Errore batch per relazione :{pred} -> {e}")
 
-    @staticmethod
-    def _batch_write_tx(tx, batch_data):
-        """Funzione transazionale interna."""
-        for item in batch_data:
-            # Usiamo MERGE per evitare duplicati
-            # Usiamo apoc.create.relationship se disponibile per predicati dinamici,
-            # altrimenti usiamo string formatting controllata (safe perché interna).
-            
-            # Sanitizzazione predicato per evitare injection (solo caratteri sicuri)
-            safe_pred = "".join(x for x in item['pred'] if x.isalnum() or x in "_:")
-            if not safe_pred: safe_pred = "RELATED_TO"
-
-            query = (
-                f"MERGE (s:Resource {{uri: $subj_uri}}) "
-                f"ON CREATE SET s.label = $subj_label "
-                f"MERGE (o:Resource {{uri: $obj_uri}}) "
-                f"ON CREATE SET o.label = $obj_label "
-                f"MERGE (s)-[r:`{safe_pred}`]->(o) "
-                f"SET r.confidence = $conf, r.source = $src"
-            )
-            
-            tx.run(query, item)
+        print("✅ Salvataggio completato.")
 
-# --- TEST ISOLATO ---
-if __name__ == "__main__":
-    # Creiamo un mock per testare senza dipendenze esterne
-    from collections import namedtuple
-    MockTriple = namedtuple("MockTriple", ["subject", "predicate", "object", "confidence", "source"])
-
-    triples = [
-        MockTriple("Batch Node 1", "TEST_BATCH", "Batch Node 2", 0.99, "test_doc_1"),
-        MockTriple("Batch Node 2", "IS_RELATED_TO", "Batch Node 3", 0.85, "test_doc_1")
-    ]
-
-    # Assicurati di avere le variabili d'ambiente o fallback attivi
-    persister = KnowledgeGraphPersister()
-    persister.save_triples(triples)
-    persister.close()
+    @staticmethod
+    def _unwind_write_tx(tx, predicate, batch_data):
+        """
+        Usa UNWIND per inserire migliaia di righe in un colpo solo.
+        Molto più performante su rete cloud.
+        """
+        # La query è dinamica SOLO sul tipo di relazione (sanitizzato prima),
+        # tutto il resto passa come parametro lista ($batch).
+        query = (
+            f"UNWIND $batch AS row "
+            f"MERGE (s:Resource {{uri: row.subj_uri}}) "
+            f"ON CREATE SET s.label = row.subj_label "
+            f"MERGE (o:Resource {{uri: row.obj_uri}}) "
+            f"ON CREATE SET o.label = row.obj_label "
+            f"MERGE (s)-[r:`{predicate}`]->(o) "
+            f"SET r.confidence = row.conf, "
+            f"    r.source = row.src, "
+            f"    r.last_updated = datetime()"
+        )
+        
+        tx.run(query, batch=batch_data)