# FlightMonitor/data/aircraft_database_manager.py import sqlite3 import csv import os import threading # For identifying the thread from typing import Optional, Dict, Any, Tuple, Callable try: from ..utils.logger import get_logger logger = get_logger(__name__) except ImportError: import logging logger = logging.getLogger(__name__) if not logger.hasHandlers(): logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s [%(levelname)s] %(funcName)s (%(threadName)s): %(message)s' ) logger.warning("AircraftDatabaseManager using fallback standard Python logger.") DEFAULT_AIRCRAFT_DB_FILENAME = "aircraft_database.db" try: from . import config as app_config AIRCRAFT_DB_ROOT_DIRECTORY = getattr(app_config, "DATABASE_DIRECTORY", "flight_data_history") logger.info(f"AircraftDatabaseManager will use directory from app_config: {AIRCRAFT_DB_ROOT_DIRECTORY}") except ImportError: AIRCRAFT_DB_ROOT_DIRECTORY = "flight_data_history" logger.warning(f"AircraftDatabaseManager: app_config not found. Using fallback directory: {AIRCRAFT_DB_ROOT_DIRECTORY}") class AircraftDatabaseManager: """ Manages the local SQLite database containing static aircraft details. Handles thread-safe connections for operations that might be called from different threads. """ def __init__(self, db_name: str = DEFAULT_AIRCRAFT_DB_FILENAME, db_root_dir: Optional[str] = None): actual_db_root_dir = db_root_dir if db_root_dir is not None else AIRCRAFT_DB_ROOT_DIRECTORY self.db_path: str = os.path.join(os.path.abspath(actual_db_root_dir), db_name) self.conn: Optional[sqlite3.Connection] = None self._main_thread_id: int = threading.get_ident() logger.info(f"Aircraft database path set to: {self.db_path}") db_parent_dir = os.path.dirname(self.db_path) if not os.path.exists(db_parent_dir): try: os.makedirs(db_parent_dir, exist_ok=True) logger.info(f"Aircraft database directory created: {db_parent_dir}") except OSError as e: logger.error(f"Critical error creating directory {db_parent_dir}: {e}. DB might be inaccessible.", exc_info=True) raise RuntimeError(f"Unable to create database directory: {db_parent_dir}") from e if threading.get_ident() == self._main_thread_id: self._connect_safely() if self.conn: self._create_table() else: logger.error("Main database connection failed on init. Cannot ensure aircraft_details table exists.") logger.info(f"AircraftDatabaseManager initialized for DB: {self.db_path}. Main thread ID: {self._main_thread_id}") def _get_thread_safe_connection(self) -> Optional[sqlite3.Connection]: current_thread_id = threading.get_ident() if current_thread_id == self._main_thread_id: if self.conn is None or not self._is_connection_alive(self.conn): logger.debug(f"Main thread ({current_thread_id}): Re-establishing or initial main connection.") self._connect_safely() return self.conn else: logger.debug(f"Worker thread ({current_thread_id}): Creating new local, temporary DB connection to {self.db_path}") try: local_conn = sqlite3.connect(self.db_path, timeout=10.0) local_conn.row_factory = sqlite3.Row return local_conn except sqlite3.Error as e: logger.error(f"Worker thread ({current_thread_id}): Error creating local DB connection: {e}", exc_info=True) return None def _connect_safely(self): if threading.get_ident() != self._main_thread_id: logger.error("_connect_safely called from non-main thread. This is for self.conn only.") return try: if self.conn and not self._is_connection_alive(self.conn): try: self.conn.close(); logger.debug("Closed invalid main DB connection.") except sqlite3.Error: pass # Ignore errors on close if already problematic self.conn = None if self.conn is None: self.conn = sqlite3.connect(self.db_path, timeout=10.0) self.conn.row_factory = sqlite3.Row logger.info(f"Main thread: Connected/Reconnected to aircraft database: {self.db_path}") except sqlite3.Error as e: logger.error(f"Main thread: Error connecting/reconnecting to aircraft database {self.db_path}: {e}", exc_info=True) self.conn = None def _is_connection_alive(self, conn_to_check: Optional[sqlite3.Connection]) -> bool: if conn_to_check is None: return False try: conn_to_check.execute("SELECT 1").fetchone() return True except sqlite3.Error: return False def _create_table(self): if threading.get_ident() != self._main_thread_id: logger.warning("Table creation attempt from a non-main thread. Aborting.") return if not self.conn: logger.error("No main database connection to create aircraft_details table.") self._connect_safely() if not self.conn: logger.critical("Failed to establish main DB connection for table creation.") return try: with self.conn: cursor = self.conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS aircraft_details ( icao24 TEXT PRIMARY KEY NOT NULL, registration TEXT, manufacturericao TEXT, manufacturername TEXT, model TEXT, typecode TEXT, serialnumber TEXT, operator TEXT, operatorcallsign TEXT, operatoricao TEXT, operatoriata TEXT, owner TEXT, country TEXT, built_year INTEGER, firstflightdate TEXT, categorydescription TEXT, engines TEXT, icaoclass TEXT, linenumber TEXT, modes INTEGER, notes TEXT, status TEXT, timestamp_metadata TEXT ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_aircraft_registration ON aircraft_details(registration);") cursor.execute("CREATE INDEX IF NOT EXISTS idx_aircraft_typecode ON aircraft_details(typecode);") cursor.execute("CREATE INDEX IF NOT EXISTS idx_aircraft_operator ON aircraft_details(operator);") logger.info("Table 'aircraft_details' and indexes ensured in DB (by main thread).") except sqlite3.Error as e: logger.error(f"Error creating 'aircraft_details' table (by main thread): {e}", exc_info=True) def get_aircraft_details(self, icao24: str) -> Optional[Dict[str, Any]]: conn = self._get_thread_safe_connection() if not conn: logger.error(f"Could not get DB connection for get_aircraft_details (ICAO: {icao24}).") return None icao24_clean = icao24.lower().strip() details = None try: with conn: cursor = conn.cursor() cursor.execute("SELECT * FROM aircraft_details WHERE icao24 = ?", (icao24_clean,)) row = cursor.fetchone() if row: details = dict(row) except sqlite3.Error as e: logger.error(f"DB error in get_aircraft_details for {icao24_clean}: {e}", exc_info=True) except Exception as e_generic: logger.error(f"Generic error in get_aircraft_details for {icao24_clean}: {e_generic}", exc_info=True) finally: if conn is not self.conn and conn is not None: try: conn.close(); logger.debug(f"Closed temporary DB connection for get_aircraft_details (Thread: {threading.get_ident()}).") except sqlite3.Error: pass return details def import_from_csv(self, csv_filepath: str, replace_existing: bool = True, progress_callback: Optional[Callable[[int, int, Optional[int]], None]] = None, total_rows_for_callback: Optional[int] = None) -> Tuple[int, int]: conn = self._get_thread_safe_connection() if not conn: logger.error(f"Could not get DB connection for import_from_csv (File: {csv_filepath}).") if progress_callback: try: progress_callback(0, 0, total_rows_for_callback) except Exception as e_cb_fail: logger.error(f"Error in progress_callback during connection failure: {e_cb_fail}") return 0, 0 current_thread_id = threading.get_ident() logger.info(f"Thread ({current_thread_id}) - Starting CSV import: {csv_filepath}. Replace mode: {replace_existing}") # Set field size limit at the beginning of the import process for this thread's CSV reader try: current_csv_limit = csv.field_size_limit() desired_csv_limit = 10 * 1024 * 1024 # 10MB, adjust as needed if desired_csv_limit > current_csv_limit: csv.field_size_limit(desired_csv_limit) logger.info(f"Thread ({current_thread_id}): Increased csv.field_size_limit from {current_csv_limit} to {desired_csv_limit} for this import.") except Exception as e_limit: logger.error(f"Thread ({current_thread_id}): Error attempting to increase csv.field_size_limit: {e_limit}") # Continue, but be aware that import might fail if fields are indeed too large. processed_data_rows = 0 imported_or_updated_rows = 0 db_columns = [ 'icao24', 'registration', 'manufacturericao', 'manufacturername', 'model', 'typecode', 'serialnumber', 'operator', 'operatorcallsign', 'operatoricao', 'operatoriata', 'owner', 'country', 'built_year', 'firstflightdate', 'categorydescription', 'engines', 'icaoclass', 'linenumber', 'modes', 'notes', 'status', 'timestamp_metadata' ] csv_to_db_column_map = { 'icao24': 'icao24', 'registration': 'registration', 'manufacturericao': 'manufacturericao', 'manufacturername': 'manufacturername', 'model': 'model', 'typecode': 'typecode', 'serialnumber': 'serialnumber', 'operator': 'operator', 'operatorcallsign': 'operatorcallsign', 'operatoricao': 'operatoricao', 'operatoriata': 'operatoriata', 'owner': 'owner', 'country': 'country', 'built': 'built_year', 'firstflightdate': 'firstflightdate', 'categorydescription': 'categorydescription', 'engines': 'engines', 'icaoaircraftclass': 'icaoclass', 'linenumber': 'linenumber', 'modes': 'modes', 'notes': 'notes', 'status': 'status', 'timestamp': 'timestamp_metadata' } callback_interval = 500 # Call progress_callback every N processed data rows try: with conn: cursor = conn.cursor() if replace_existing: cursor.execute("DELETE FROM aircraft_details") logger.info(f"Thread ({current_thread_id}): Existing data deleted from aircraft_details.") with open(csv_filepath, 'r', encoding='utf-8-sig') as csvfile: reader = csv.DictReader(csvfile) if not reader.fieldnames: logger.error(f"Thread ({current_thread_id}): CSV file {csv_filepath} has no valid header or is empty.") if progress_callback: progress_callback(0, 0, total_rows_for_callback) return 0,0 for csv_row_index, raw_row_data in enumerate(reader): processed_data_rows += 1 if progress_callback and processed_data_rows % callback_interval == 0: try: progress_callback(processed_data_rows, imported_or_updated_rows, total_rows_for_callback) except Exception as e_cb: logger.error(f"Thread ({current_thread_id}) Error during progress_callback: {e_cb}") row_data_lower_keys = {str(k).lower().strip(): v for k, v in raw_row_data.items() if k} icao24_val = row_data_lower_keys.get('icao24', '').strip().lower() if not icao24_val: logger.warning(f"Thread ({current_thread_id}) CSV Data Row {processed_data_rows}: Missing or empty icao24. Skipping row.") continue data_for_db = {'icao24': icao24_val} for csv_col_lower, db_col_name in csv_to_db_column_map.items(): if csv_col_lower == 'icao24': continue raw_value = row_data_lower_keys.get(csv_col_lower, '') cleaned_value = raw_value.strip() if isinstance(raw_value, str) else raw_value final_value = None if cleaned_value or isinstance(cleaned_value, (int, float, bool)): if db_col_name == 'built_year': final_value = int(cleaned_value) if str(cleaned_value).isdigit() else None elif db_col_name == 'modes': final_value = int(cleaned_value) if str(cleaned_value).isdigit() else None else: final_value = cleaned_value data_for_db[db_col_name] = final_value cols_for_sql = [] vals_for_sql = [] for db_col_key in db_columns: if db_col_key in data_for_db: cols_for_sql.append(db_col_key) vals_for_sql.append(data_for_db[db_col_key]) if not cols_for_sql or 'icao24' not in cols_for_sql: continue placeholders = ', '.join(['?'] * len(cols_for_sql)) sql_cols_str = ', '.join(cols_for_sql) sql = f"INSERT OR REPLACE INTO aircraft_details ({sql_cols_str}) VALUES ({placeholders})" try: cursor.execute(sql, vals_for_sql) imported_or_updated_rows += 1 except sqlite3.Error as e_sql: logger.error(f"Thread ({current_thread_id}) CSV Data Row {processed_data_rows}: SQL error for ICAO24 '{icao24_val}': {e_sql}. Data: {dict(zip(cols_for_sql, vals_for_sql))}") logger.info(f"Thread ({current_thread_id}): CSV import finished. Processed data rows: {processed_data_rows}, DB rows imported/updated: {imported_or_updated_rows}") except FileNotFoundError: logger.error(f"Thread ({current_thread_id}): CSV file not found: {csv_filepath}") except csv.Error as e_csv_format: logger.error(f"Thread ({current_thread_id}): CSV format error in {csv_filepath} (around data row {processed_data_rows+1}): {e_csv_format}", exc_info=True) except Exception as e_general_import: logger.error(f"Thread ({current_thread_id}): General error during CSV import {csv_filepath}: {e_general_import}", exc_info=True) finally: if progress_callback: try: progress_callback(processed_data_rows, imported_or_updated_rows, total_rows_for_callback) except Exception as e_cb_final: logger.error(f"Thread ({current_thread_id}) Error during final progress_callback: {e_cb_final}") if conn is not self.conn and conn is not None: try: conn.close(); logger.debug(f"Closed temporary DB connection for import_from_csv (Thread: {current_thread_id}).") except sqlite3.Error: pass return processed_data_rows, imported_or_updated_rows def close_connection(self): if self.conn and threading.get_ident() == self._main_thread_id: try: if self._is_connection_alive(self.conn): self.conn.close() logger.info(f"Main aircraft database connection closed: {self.db_path}") except sqlite3.Error as e: logger.error(f"Error closing main aircraft DB connection: {e}", exc_info=True) finally: self.conn = None elif self.conn and threading.get_ident() != self._main_thread_id: logger.warning(f"Attempt to close main connection from non-main thread ({threading.get_ident()}). Ignored.") if __name__ == '__main__': # ... (Blocco test standalone come nella versione precedente, assicurati che il logger sia configurato # e che i percorsi dei file di test siano corretti per il tuo ambiente di test) if not logging.getLogger(__name__).hasHandlers(): logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s [%(levelname)s] %(funcName)s (%(threadName)s): %(message)s') logger.info("--- Starting Standalone AircraftDatabaseManager Test (Thread-Safe with Callback) ---") current_script_dir = os.path.dirname(os.path.abspath(__file__)) test_data_base_dir = os.path.join(os.path.dirname(current_script_dir), "test_artefacts") test_data_dir = os.path.join(test_data_base_dir, "aircraft_db_tests_standalone_cb") if not os.path.exists(test_data_dir): os.makedirs(test_data_dir, exist_ok=True) test_db_name = "test_aircraft_db_standalone_cb.sqlite" test_csv_filename = "sample_aircraft_data_standalone_cb.csv" test_csv_path = os.path.join(test_data_dir, test_csv_filename) header = [ 'icao24','timestamp','built','categoryDescription','country','registration','model','manufacturerName' ] # Intestazione ridotta per il test # Crea un CSV con più righe per testare la callback num_test_rows = 1200 sample_data_for_cb_test = [] for i in range(num_test_rows): icao = f"cb{i:04d}" year = 2000 + (i % 20) sample_data_for_cb_test.append( [icao, f'2023-01-01 10:{i//60:02d}:{i%60:02d}', str(year),'Test Category', 'US', f'N{icao.upper()}', f'Model_{i}', f'Manu_{i}'] ) with open(test_csv_path, 'w', newline='', encoding='utf-8-sig') as f_test: writer = csv.writer(f_test) writer.writerow(header) writer.writerows(sample_data_for_cb_test) logger.info(f"Test CSV file with {num_test_rows} data rows created: {test_csv_path}") db_manager_main = AircraftDatabaseManager(db_name=test_db_name, db_root_dir=test_data_dir) def import_job_with_cb(csv_p, db_mgr_instance, replace): thread_id = threading.get_ident() logger.info(f"Starting import in THREAD: {thread_id}") # Definisci la callback qui dentro o passala come argomento def my_test_progress_callback(processed, imported, total): logger.info(f"THREAD Progress CB ({thread_id}): Processed={processed}, Imported={imported}, Total={total if total is not None else 'N/A'}") # Per il test, calcoliamo il totale qui, come farebbe AppController total_rows_in_csv = 0 try: with open(csv_p, 'r', encoding='utf-8-sig') as f_count: r = csv.reader(f_count) next(r) # skip header total_rows_in_csv = sum(1 for _ in r) except: pass p, i = db_mgr_instance.import_from_csv( csv_p, replace_existing=replace, progress_callback=my_test_progress_callback, total_rows_for_callback=total_rows_in_csv ) logger.info(f"THREAD ({thread_id}) Import finished. Processed: {p}, Imported: {i}") assert p == total_rows_in_csv, f"Thread {thread_id}: Processed rows ({p}) non corrisponde al totale ({total_rows_in_csv})" assert i == total_rows_in_csv, f"Thread {thread_id}: Imported rows ({i}) non corrisponde al totale ({total_rows_in_csv})" if db_manager_main.conn is not None : # Assicura che la connessione principale sia stata creata logger.info("--- Starting CSV Import Test (replace=True) from THREAD with Callback ---") import_thread_cb = threading.Thread(target=import_job_with_cb, args=(test_csv_path, db_manager_main, True)) import_thread_cb.start() import_thread_cb.join() # Verifica alcuni dati dopo l'importazione details_cb_test = db_manager_main.get_aircraft_details("cb0000") assert details_cb_test is not None, "cb0000 non trovato dopo import" if details_cb_test: assert details_cb_test['manufacturername'] == 'Manu_0' details_last_cb_test = db_manager_main.get_aircraft_details(f"cb{num_test_rows-1:04d}") assert details_last_cb_test is not None, f"cb{num_test_rows-1:04d} non trovato" db_manager_main.close_connection() logger.info("Standalone AircraftDatabaseManager tests (with callback) completed successfully.") else: logger.error("Initial main DB connection failed in standalone test. Aborting tests.")