| """Initialisation de company.db à partir des CSV de ventes.""" |
| from __future__ import annotations |
|
|
| import csv |
| import logging |
| import sqlite3 |
| from pathlib import Path |
|
|
| from app.core.config import get_settings |
|
|
| logger = logging.getLogger(__name__) |
|
|
| CREATE_SALES = """ |
| CREATE TABLE IF NOT EXISTS sales ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| client TEXT NOT NULL, |
| produit TEXT NOT NULL, |
| region TEXT NOT NULL, |
| mois TEXT NOT NULL, |
| trimestre TEXT NOT NULL, |
| chiffre_affaires REAL NOT NULL, |
| marge REAL NOT NULL, |
| source_doc TEXT NOT NULL |
| ); |
| """ |
|
|
|
|
| def _csv_paths(data_dir: Path) -> list[Path]: |
| reports = data_dir / "reports" |
| return sorted(reports.glob("sales_*.csv")) |
|
|
|
|
| def seed_company_db(db_path: Path, data_dir: Path | None = None) -> Path: |
| """(Re)crée la base SQLite lecture-métier à partir des CSV.""" |
| settings = get_settings() |
| root = Path(data_dir or settings.local_data_dir) |
| db_path.parent.mkdir(parents=True, exist_ok=True) |
|
|
| if db_path.exists(): |
| db_path.unlink() |
|
|
| conn = sqlite3.connect(str(db_path)) |
| try: |
| conn.execute(CREATE_SALES) |
| insert = ( |
| "INSERT INTO sales " |
| "(client, produit, region, mois, trimestre, " |
| "chiffre_affaires, marge, source_doc) " |
| "VALUES (?, ?, ?, ?, ?, ?, ?, ?)" |
| ) |
| total = 0 |
| for csv_path in _csv_paths(root): |
| source_doc = csv_path.stem |
| with csv_path.open(encoding="utf-8", newline="") as handle: |
| reader = csv.DictReader(handle, delimiter=";") |
| for row in reader: |
| conn.execute( |
| insert, |
| ( |
| row["client"], |
| row["produit"], |
| row["region"], |
| row["mois"], |
| row.get("trimestre", ""), |
| float(row["chiffre_affaires"]), |
| float(row["marge"]), |
| source_doc, |
| ), |
| ) |
| total += 1 |
| conn.commit() |
| logger.info("SQLite seed: %s (%d lignes)", db_path, total) |
| finally: |
| conn.close() |
| return db_path |
|
|
|
|
| def ensure_company_db(db_path: Path | None = None) -> Path: |
| settings = get_settings() |
| path = Path(db_path or settings.sqlite_db_path) |
| if path.exists() and path.stat().st_size > 0: |
| return path |
| return seed_company_db(path) |
|
|
