""" WriteToAccess — mattoncino riutilizzabile per scrivere un DataFrame in Access. Metodo: csv_transfer 1. DataFrame → CSV temporaneo (temp dir), separatore virgola, float con ',' come decimale (locale italiano) e quotati per evitare ambiguità 2. schema.ini con tipi espliciti (evita auto-rilevamento errato di TransferText) 3. CSV → Access via Access.Application.DoCmd.TransferText 4. Pulizia file temporanei Molto più veloce di pyodbc INSERT per tabelle grandi. Richiede Microsoft Access installato sulla macchina. Uso: from WriteToAccess import df_to_access n = df_to_access(df, r'C:\\path\\to\\db.accdb', 'NomeTabella') """ import csv import tempfile from pathlib import Path import pandas as pd import pyodbc # Mappa tipo ODBC → dichiarazione schema.ini # Garantisce che TransferText usi Long invece di Short Integer _ODBC_TO_SCHEMA = { 'INTEGER': 'Long', # Long Integer (4 byte) — evita overflow Short 'DOUBLE': 'Double', 'DATETIME': 'DateTime', 'BIT': 'Bit', } def _build_schema_ini(db_path: Path, table: str, csv_name: str, csv_columns: list | None = None) -> str: """ Genera il contenuto dello schema.ini con tipi espliciti per ogni colonna. I numeri Col{i} corrispondono alla posizione nel CSV (= ordine colonne del DataFrame), NON all'ordine delle colonne nella tabella Access. Se csv_columns è fornito, viene usato per determinare l'ordine corretto. """ conn_str = ( 'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};' f'DBQ={db_path};' ) lines = [ f'[{csv_name}]', 'Format=CSVDelimited', 'ColNameHeader=True', ] try: with pyodbc.connect(conn_str) as conn: odbc_cols = list(conn.cursor().columns(table=table)) # Mappa nome_colonna → tipo schema.ini type_map = {} for col in odbc_cols: schema_type = _ODBC_TO_SCHEMA.get(col.type_name.upper()) if schema_type: type_map[col.column_name] = schema_type if csv_columns: # Usa l'ordine reale del CSV (= ordine del DataFrame) for i, col_name in enumerate(csv_columns, start=1): schema_type = type_map.get(col_name) if schema_type: lines.append(f'Col{i}={col_name} {schema_type}') else: # Fallback: usa l'ordine della tabella Access (meno preciso) for i, col in enumerate(odbc_cols, start=1): schema_type = _ODBC_TO_SCHEMA.get(col.type_name.upper()) if schema_type: lines.append(f'Col{i}={col.column_name} {schema_type}') # VARCHAR/TEXT: lascia senza spec → TransferText usa Text automaticamente except Exception: pass # fallback: nessun tipo esplicito (comportamento precedente) return '\n'.join(lines) + '\n' def _fix_float_decimals(df: pd.DataFrame, db_path: Path | None = None, table: str | None = None) -> pd.DataFrame: """ Converte valori float in stringhe con ',' come separatore decimale. Copre due casi: 1. Colonne pandas di tipo float64/float32 — conversione diretta 2. Colonne pandas di tipo object che mappano a DOUBLE in Access — stringhe come "4.58" diventano "4,58" Il driver ISAM di Access su locale italiano usa ',' come decimale: "4.58" viene letto come 458 (punto = separatore migliaia). """ df = df.copy() # Legge i tipi Access una volta sola (serve per Caso 1 + 2 + 3) integer_cols: set = set() double_cols: set = set() if db_path and table: try: conn_str = ( 'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};' f'DBQ={db_path};' ) with pyodbc.connect(conn_str) as conn: odbc_cols = list(conn.cursor().columns(table=table)) double_cols = {c.column_name for c in odbc_cols if c.type_name.upper() == 'DOUBLE'} integer_cols = {c.column_name for c in odbc_cols if c.type_name.upper() == 'INTEGER'} except Exception: pass # Caso 1: colonne float pandas # - se la colonna è INTEGER in Access → stringa intera pura ("420741.0"→"420741") # (evita che il driver ISAM italiano legga il punto come sep. migliaia) # - altrimenti → stringa con ',' come decimale ("4.69"→"4,69") float_cols = df.select_dtypes(include=['float64', 'float32', 'float16']).columns for col in float_cols: if col in integer_cols: df[col] = df[col].apply( lambda x: str(int(x)) if pd.notna(x) and x == int(x) else (str(x).replace('.', ',') if pd.notna(x) else '') ) else: df[col] = df[col].apply( lambda x: str(x).replace('.', ',') if pd.notna(x) else '' ) # Caso 2+3: colonne object che mappano a DOUBLE o INTEGER in Access str_cols = df.select_dtypes(include=['object']).columns for col in str_cols: if col in double_cols: df[col] = df[col].apply(_comma_decimal_if_float) elif col in integer_cols: df[col] = df[col].apply(_int_from_float_str) return df def _comma_decimal_if_float(x) -> str: """Sostituisce '.' con ',' solo se il valore è una stringa numerica float.""" if x is None or x == '': return x or '' s = str(x) try: float(s) # verifica che sia un numero return s.replace('.', ',') except ValueError: return s def _int_from_float_str(x) -> str: """ Converte una stringa float-intera in stringa intera pura. "420741.0" → "420741" (evita che il driver ISAM italiano legga il punto come separatore migliaia e produca "4207410"). Lascia invariate stringhe non numeriche o già intere. """ if x is None or x == '': return x or '' s = str(x) try: f = float(s) if f == int(f): return str(int(f)) return s # non è un intero — lascia invariato except (ValueError, OverflowError): return s def df_to_access( df: pd.DataFrame, db_path: str | Path, table: str, clear: bool = True, ) -> int: """ Scrive un DataFrame in una tabella Access. Parametri: df — DataFrame da scrivere db_path — path al file .accdb table — nome della tabella Access di destinazione clear — se True (default), svuota la tabella prima dell'import Restituisce il numero di righe importate. Lancia eccezione in caso di errore. """ db_path = Path(db_path) if df is None or df.empty: return 0 # 1. Svuota la tabella via pyodbc (se richiesto) if clear: conn_str = ( 'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};' f'DBQ={db_path};' ) with pyodbc.connect(conn_str, autocommit=True) as conn: conn.execute(f'DELETE * FROM [{table}]') # 2. DataFrame → CSV temporaneo tmp_dir = Path(tempfile.gettempdir()) csv_path = tmp_dir / f'_wta_{table}.csv' ini_path = tmp_dir / 'schema.ini' try: # Converti float → stringa con ',' decimale (locale italiano) df_out = _fix_float_decimals(df, db_path=db_path, table=table) # QUOTE_NONNUMERIC: quota tutte le stringhe (incluse "4,69") per evitare # ambiguità con il separatore di campo ',' df_out.to_csv(csv_path, index=False, encoding='cp1252', quoting=csv.QUOTE_NONNUMERIC, errors='replace') ini_path.write_text( _build_schema_ini(db_path, table, csv_path.name, csv_columns=list(df.columns)), encoding='utf-8', ) # 3. Import via Access.Application.DoCmd.TransferText import win32com.client app = win32com.client.Dispatch('Access.Application') try: app.OpenCurrentDatabase(str(db_path)) # acImportDelim=0, no spec, table, csv, has_header=True app.DoCmd.TransferText(0, None, table, str(csv_path), True) app.CloseCurrentDatabase() finally: app.Quit() del app finally: # 4. Pulizia file temporanei for f in (csv_path, ini_path): if f.exists(): try: f.unlink() except Exception: pass return len(df)