File size: 2,544 Bytes
98aa813 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """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)
|