[#1] Fix ingest_id name error and isolate FULLTEXT index creation

ingest_csv.py

@@ -54,6 +54,20 @@ def ingest_file(filename, engine):
         
     print(f"\n🚀 Found {filename}! Starting grouped vertical partition ingestion...")
     
+    # Initialize ingestion status record
+    init_ingestion_status_table(engine)
+    ingest_id = None
+    try:
+        with engine.begin() as conn:
+            conn.execute(text("""
+                INSERT INTO ingestion_status (filename, start_time, rows_loaded, status)
+                VALUES (:filename, NOW(), 0, 'RUNNING')
+            """), {"filename": filename})
+            res_id = conn.execute(text("SELECT LAST_INSERT_ID()"))
+            ingest_id = res_id.scalar()
+    except Exception as e:
+        print(f"Warning: Could not create ingestion status record: {e}")
+
     chunk_size = 10000 
     total_processed = 0
     max_chunks = int(os.environ.get('MAX_CHUNKS', '0'))
@@ -180,12 +194,16 @@ def ingest_file(filename, engine):
             print(f"   Successfully appended {total_processed} rows into grouped tables...", end="\r")
             
             # Update rows loaded in database
-            with engine.begin() as conn:
-                conn.execute(text("""
-                    UPDATE ingestion_status 
-                    SET rows_loaded = :rows
-                    WHERE id = :id
-                """), {"rows": total_processed, "id": ingest_id})
+            if ingest_id is not None:
+                try:
+                    with engine.begin() as conn:
+                        conn.execute(text("""
+                            UPDATE ingestion_status 
+                            SET rows_loaded = :rows
+                            WHERE id = :id
+                        """), {"rows": total_processed, "id": ingest_id})
+                except Exception as e:
+                    print(f"Warning: Could not update ingestion progress: {e}")
                 
             if total_processed % 50000 == 0:
                 notifier.send_alert(f"Ingestion Milestone: {total_processed} rows processed")
@@ -196,32 +214,38 @@ def ingest_file(filename, engine):
     notifier.send_alert(f"Ingestion Finished: {filename}")
     print(f"\n✅ Finished importing {filename}.")
     
-    # Auto-create required FULLTEXT search indexes if not present
-    print("🚀 Creating FULLTEXT indexes on products_core for Streamlit search RAG...")
-    with engine.begin() as conn:
+    if ingest_id is not None:
         try:
-            res = conn.execute(text("SHOW INDEX FROM products_core WHERE Key_name = 'idx_prod_name'"))
-            if not res.fetchall():
-                conn.execute(text("ALTER TABLE products_core ADD FULLTEXT INDEX idx_prod_name (product_name)"))
-                print("Created FULLTEXT index: idx_prod_name")
+            with engine.begin() as conn:
+                conn.execute(text("""
+                    UPDATE ingestion_status 
+                    SET end_time = NOW(), status = 'COMPLETED'
+                    WHERE id = :id
+                """), {"id": ingest_id})
         except Exception as e:
-            print(f"Warning: Could not create index idx_prod_name: {e}")
+            print(f"Warning: Could not update ingestion status: {e}")
             
+    # Auto-create required FULLTEXT search indexes if not present
+    print("🚀 Creating FULLTEXT indexes on products_core for Streamlit search RAG...")
+    for idx_name, cols in [
+        ('idx_prod_name', 'product_name'),
+        ('idx_ing_text', 'ingredients_text'),
+        ('idx_prod_ing', 'product_name, ingredients_text')
+    ]:
         try:
-            res = conn.execute(text("SHOW INDEX FROM products_core WHERE Key_name = 'idx_ing_text'"))
-            if not res.fetchall():
-                conn.execute(text("ALTER TABLE products_core ADD FULLTEXT INDEX idx_ing_text (ingredients_text)"))
-                print("Created FULLTEXT index: idx_ing_text")
-        except Exception as e:
-            print(f"Warning: Could not create index idx_ing_text: {e}")
-
-        try:
-            res = conn.execute(text("SHOW INDEX FROM products_core WHERE Key_name = 'idx_prod_ing'"))
-            if not res.fetchall():
-                conn.execute(text("ALTER TABLE products_core ADD FULLTEXT INDEX idx_prod_ing (product_name, ingredients_text)"))
-                print("Created FULLTEXT index: idx_prod_ing")
+            with engine.connect() as conn:
+                res = conn.execute(text(f"SHOW INDEX FROM products_core WHERE Key_name = '{idx_name}'"))
+                exists = len(res.fetchall()) > 0
+            
+            if not exists:
+                print(f"Index {idx_name} not found. Creating index...")
+                with engine.begin() as conn:
+                    conn.execute(text(f"ALTER TABLE products_core ADD FULLTEXT INDEX {idx_name} ({cols})"))
+                print(f"Created FULLTEXT index: {idx_name}")
+            else:
+                print(f"Index {idx_name} already exists.")
         except Exception as e:
-            print(f"Warning: Could not create index idx_prod_ing: {e}")
+            print(f"Warning: Could not create index {idx_name}: {e}")
             
     return True