# db_manager.py # --- CRITICAL FIX FOR CHROMADB SQLITE VERSION --- import sys try: __import__('pysqlite3') sys.modules['sqlite3'] = sys.modules.pop('pysqlite3') except ImportError: pass # ----------------------------------------------- import os import json import hashlib import chromadb from pathlib import Path from supabase_client import get_supabase_client from config import KNOWLEDGE_CACHE_DIR, CRAWL_CACHE_DIR, VECTOR_DB_PATH def list_all_events() -> list[str]: """Fetches a list of all event_keys from the knowledge_cache.""" supabase = get_supabase_client() if not supabase: print("[DB_MANAGER_ERROR] Supabase client not initialized. Cannot list events.") return[] try: response = supabase.table('knowledge_cache').select('event_key').order('updated_at', desc=True).execute() if response.data: return [item['event_key'] for item in response.data] return[] except Exception as e: print(f"[DB_MANAGER_ERROR] Failed to list events: {e}") return[] def get_event_details(event_key: str) -> dict: """Fetches the full JSON data for a specific event_key.""" supabase = get_supabase_client() if not supabase: print(f"[DB_MANAGER_ERROR] Supabase client not initialized. Cannot get details for '{event_key}'.") return {} try: response = supabase.table('knowledge_cache').select('data').eq('event_key', event_key).single().execute() return response.data.get('data', {}) if response.data else {} except Exception as e: print(f"[DB_MANAGER_ERROR] Failed to get details for '{event_key}': {e}") return {} def delete_event_from_supabase(event_key: str) -> tuple[bool, str]: """ Performs a comprehensive "hard delete" of an event across all related Supabase services. This deletes from knowledge_cache, vector_store, and Storage. """ supabase = get_supabase_client() if not supabase: error_message = "Supabase client not initialized. Cannot delete event." print(f"[DB_MANAGER_ERROR] {error_message}") return False, error_message try: print(f"[DB_MANAGER] Starting hard delete for event_key: {event_key}") vector_response = supabase.table('vector_store').select('metadata').eq('metadata->>event_id', event_key).execute() urls_to_delete = set() if vector_response.data: for item in vector_response.data: source_url = item.get('metadata', {}).get('source_url') if source_url: urls_to_delete.add(source_url) print(f" - Deleting vectors for '{event_key}'...") supabase.table('vector_store').delete().eq('metadata->>event_id', event_key).execute() print(" - Vector deletion complete.") print(f" - Deleting knowledge cache entry for '{event_key}'...") supabase.table('knowledge_cache').delete().eq('event_key', event_key).execute() print(" - Knowledge cache deletion complete.") if urls_to_delete: print(f" - Found {len(urls_to_delete)} unique source files to delete from Storage...") files_to_delete =[f"crawl_cache/{hashlib.md5(url.encode()).hexdigest()}.md" for url in urls_to_delete] if files_to_delete: supabase.storage.from_('crawl_cache').remove(files_to_delete) print(" - Storage file deletion complete.") success_message = f"Successfully deleted all data for event '{event_key}' from Supabase." print(f"[DB_MANAGER] {success_message}") return True, success_message except Exception as e: error_message = f"An error occurred while deleting '{event_key}': {e}" print(f"[DB_MANAGER_ERROR] {error_message}") return False, error_message def clear_local_event_cache(event_key: str) -> tuple[bool, str]: """ Clears all LOCAL cache files associated with a specific event_key. This includes knowledge cache, crawl cache, and the local vector DB collection. """ print(f"[DB_MANAGER] Starting local cache clear for event_key: {event_key}") deleted_items =[] try: # Step 1: Find all associated crawl URLs from Supabase first for robustness supabase = get_supabase_client() urls_to_delete = set() if supabase: vector_response = supabase.table('vector_store').select('metadata').eq('metadata->>event_id', event_key).execute() if vector_response.data: for item in vector_response.data: source_url = item.get('metadata', {}).get('source_url') if source_url: urls_to_delete.add(source_url) # Step 2: Delete local knowledge cache file kc_file = Path(KNOWLEDGE_CACHE_DIR) / f"{event_key}.json" if kc_file.exists(): kc_file.unlink() deleted_items.append(f"Local knowledge cache: {kc_file.name}") print(f" - Deleted local knowledge cache: {kc_file.name}") # Step 3: Delete local crawl cache files based on URLs found if urls_to_delete: print(f" - Found {len(urls_to_delete)} source URLs to clear from local crawl cache.") for url in urls_to_delete: url_hash = hashlib.md5(url.encode()).hexdigest() cc_file = Path(CRAWL_CACHE_DIR) / f"{url_hash}.md" if cc_file.exists(): cc_file.unlink() deleted_items.append(f"Local crawl cache for: {url.split('/')[-1]}") print(f" - Deleted local crawl cache: {cc_file.name}") else: print(" - No associated crawl cache files found to clear.") # Step 4: Delete the local vector DB collection for this event try: client = chromadb.PersistentClient(path=VECTOR_DB_PATH) collection_name = f"coll_{event_key}" collections = client.list_collections() if any(c.name == collection_name for c in collections): client.delete_collection(name=collection_name) deleted_items.append(f"Local vector DB collection: '{collection_name}'") print(f" - Deleted local vector collection: '{collection_name}'") else: print(f" - No local vector collection named '{collection_name}' found to delete.") except Exception as e: print(f" - [WARNING] Could not delete vector collection (it may not exist): {e}") if not deleted_items: message = f"No local cache found for event '{event_key}'. Nothing to clear." print(f"[DB_MANAGER] {message}") return True, message message = f"Successfully cleared local cache for '{event_key}'. Removed {len(deleted_items)} item(s)." print(f"[DB_MANAGER] {message}") return True, message except Exception as e: error_message = f"An error occurred while clearing local cache for '{event_key}': {e}" print(f"[DB_MANAGER_ERROR] {error_message}") return False, error_message def list_run_outputs() -> list[dict]: """Fetches a list of all run outputs from the run_outputs table.""" supabase = get_supabase_client() if not supabase: print("[DB_MANAGER_ERROR] Supabase client not initialized. Cannot list run outputs.") return[] try: response = supabase.table('run_outputs').select('id, created_at, agent_mode, filename, event_count, run_data').order('created_at', desc=True).execute() return response.data if response.data else [] except Exception as e: print(f"[DB_MANAGER_ERROR] Failed to list run outputs: {e}") return[] def delete_run_output(run_id: str) -> tuple[bool, str]: """Deletes a specific run output record from Supabase.""" supabase = get_supabase_client() if not supabase: error_message = "Supabase client not initialized. Cannot delete run output." print(f"[DB_MANAGER_ERROR] {error_message}") return False, error_message try: supabase.table('run_outputs').delete().eq('id', run_id).execute() success_message = f"Successfully deleted run output record {run_id}." print(f"[DB_MANAGER] {success_message}") return True, success_message except Exception as e: error_message = f"An error occurred while deleting run output {run_id}: {e}" print(f"[DB_MANAGER_ERROR] {error_message}") return False, error_message