""" Couche de persistance SQLite — modèle à deux niveaux. Stockage : un fichier immos.db. Chaque actif est une ligne ; la structure JSON 3-vues (view_admin / view_cpa / view_auditor) est sérialisée en colonne JSON. Les colonnes plates (id, type, name, parent_id) servent aux requêtes. - PARENT : immobilisation fiscale. Base amortissable CALCULÉE = somme des coûts des pièces rattachées (jamais éditée directement). - PIÈCE : élément d'inventaire. Coût éditable ; il alimente la base du parent. Édition : tout champ éditable sauf la base consolidée du parent. Chaque modification d'un champ d'un PARENT est journalisée dans sa piste d'audit (ancien -> nouveau + date) pour défendabilité fiscale. """ from __future__ import annotations import copy import json import os import sqlite3 from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from threading import Lock _db_env = os.environ.get("IMMOS_DB_PATH") DB_PATH = Path(_db_env) if _db_env else Path(__file__).resolve().parent.parent / "immos.db" PARENT_TYPE = "parent" PIECE_TYPE = "piece" # Taux de DPA par catégorie — Règlement de l'impôt sur le revenu, Annexe II # (C.R.C., ch. 945). Art. 1100(1) Règl. : méthode du solde dégressif. # Ces taux sont les taux ordinaires ; aucun taux d'accélération temporaire # (p. ex. Règl. 1100(1)(a.1), 1100(1)(a.2)) n'est appliqué automatiquement. # https://laws-lois.justice.gc.ca/fra/reglements/C.R.C.,_ch._945/section-1100.html CCA_CATEGORIES = [ {"id": 50, "description": "Matériel informatique et logiciels de systèmes", "dpa_rate": 0.55, "default_useful_life_years": 4}, {"id": 8, "description": "Mobilier, équipement et matériel divers", "dpa_rate": 0.20, "default_useful_life_years": 10}, {"id": 10, "description": "Matériel informatique (avant mars 2007), véhicules","dpa_rate": 0.30, "default_useful_life_years": 5}, {"id": 12, "description": "Logiciels d'application, petits outils", "dpa_rate": 1.00, "default_useful_life_years": 2}, {"id": 46, "description": "Matériel d'infrastructure de réseau de données", "dpa_rate": 0.30, "default_useful_life_years": 7}, ] CCA_BY_ID = {c["id"]: c for c in CCA_CATEGORIES} def _now_iso() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") def _compute_linear_deprec( cost: float, isd: str, useful_life_years: float ) -> tuple[float, float, float]: """ Amortissement linéaire indicatif (comptable, NON fiscal). Retourne (accum_deprec_cad, nbv_cad, remaining_life_years). """ if isd and len(isd) >= 10: try: y_, m_, d_ = int(isd[:4]), int(isd[5:7]), int(isd[8:10]) isd_dt = datetime(y_, m_, d_, tzinfo=timezone.utc) today_dt = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) years_svc = max(0.0, (today_dt - isd_dt).days / 365.25) except (ValueError, IndexError): years_svc = 0.0 annual = cost / useful_life_years if useful_life_years > 0 else cost accum = round(min(annual * years_svc, cost), 2) nbv = round(max(cost - accum, 0.0), 2) remaining = round(max(useful_life_years - years_svc, 0.0), 2) else: accum, nbv, remaining = 0.0, round(cost, 2), useful_life_years return accum, nbv, remaining def _compute_piece_fnacc(cost: float, acquisition_year: int, cca_rate: float) -> tuple[float, int]: """ FNACC théorique d'une pièce au 31 décembre du dernier exercice fiscal complété. Méthode : amortissement dégressif, règle de la demi-année, DPA maximale chaque année. Retourne (fnacc_cad, reference_year). Indicatif uniquement : la FNACC fiscale légalement opposable appartient au solde de catégorie (pool), pas à la pièce individuelle. """ reference_year = datetime.now(timezone.utc).year - 1 if acquisition_year > reference_year: # Pièce acquise dans l'exercice courant — aucune DPA réclamée encore. return (round(cost, 2), reference_year) fnacc = cost * (1.0 - cca_rate * 0.5) # Règl. 1100(2) — demi-année for _ in range(reference_year - acquisition_year): # Règl. 1100(1) — solde dégressif fnacc *= (1.0 - cca_rate) return (round(max(fnacc, 0.0), 2), reference_year) class AssetStore: def __init__(self, db_path: Path = DB_PATH) -> None: self._lock = Lock() self._db_path = db_path self._conn = sqlite3.connect(str(db_path), check_same_thread=False) self._conn.row_factory = sqlite3.Row self._init_schema() # ---- Schéma ---- def _init_schema(self) -> None: self._conn.execute(""" CREATE TABLE IF NOT EXISTS assets ( id INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT NOT NULL, name TEXT NOT NULL, parent_id INTEGER, data TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (parent_id) REFERENCES assets(id) ON DELETE SET NULL ) """) self._conn.commit() # ---- Helpers ---- def _row_to_asset(self, row: sqlite3.Row) -> dict: asset = json.loads(row["data"]) asset["id"] = row["id"] asset["type"] = row["type"] asset["name"] = row["name"] asset["parent_id"] = row["parent_id"] return asset def _raw_get(self, asset_id: int) -> dict | None: cur = self._conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)) row = cur.fetchone() return self._row_to_asset(row) if row else None def _raw_children(self, parent_id: int) -> list[dict]: cur = self._conn.execute("SELECT * FROM assets WHERE parent_id = ?", (parent_id,)) return [self._row_to_asset(r) for r in cur.fetchall()] def _sum_pieces_cost(self, parent_id: int) -> Decimal: total = Decimal("0") for child in self._raw_children(parent_id): total += Decimal(str(child["view_cpa"]["capitalizable_base_cad"])) return total def _with_computed_base(self, asset: dict) -> dict: snap = copy.deepcopy(asset) if snap["type"] == PARENT_TYPE: children = self._raw_children(snap["id"]) cost_total = Decimal("0") fnacc_total = Decimal("0") replacement_total = Decimal("0") nbv_linear_total = Decimal("0") for child in children: enriched = self._with_computed_base(child) cost_total += Decimal(str(enriched["view_cpa"]["capitalizable_base_cad"])) fnacc_total += Decimal(str(enriched["view_cpa"].get("fnacc_cad", 0))) replacement_total += Decimal(str(enriched["view_admin"].get("replacement_cost_cad", 0))) nbv_linear_total += Decimal(str(enriched["view_cpa"].get( "nbv_linear_cad", enriched["view_cpa"]["capitalizable_base_cad"]))) snap["view_cpa"]["capitalizable_base_cad"] = float(cost_total) snap["view_cpa"]["cost_pre_tax_cad"] = float(cost_total) snap["view_cpa"]["net_book_value_cad"] = float(fnacc_total) snap["view_admin"]["pieces_count"] = len(children) snap["view_admin"]["total_replacement_cost_cad"] = float(replacement_total) snap["view_cpa"]["nbv_linear_total_cad"] = float(nbv_linear_total) # L'étiquette du parent est celle du porteur désigné. bearer = next((c for c in children if c["view_admin"].get("is_tag_bearer")), None) if bearer: snap["view_admin"]["inventory_tag"] = bearer["view_admin"].get("inventory_tag", "") elif snap["type"] == PIECE_TYPE: cost = float(snap["view_cpa"].get("capitalizable_base_cad", 0)) isd = snap["view_cpa"].get("in_service_date") or "" rate = float(snap["view_cpa"]["cca_class"].get("dpa_rate", 0.55)) override = snap["view_cpa"].get("fnacc_override_cad") ref_year = datetime.now(timezone.utc).year - 1 if isd and isd[:4].isdigit(): # En service : amortissement dégressif depuis l'année de mise en service. fnacc_auto, ref_year = _compute_piece_fnacc(cost, int(isd[:4]), rate) else: # Stock (jamais rattachée) : FNACC = coût intégral, zéro amortissement. fnacc_auto = round(cost, 2) snap["view_cpa"]["fnacc_auto_cad"] = fnacc_auto snap["view_cpa"]["fnacc_reference_year"] = ref_year snap["view_cpa"]["fnacc_is_override"] = override is not None snap["view_cpa"]["fnacc_cad"] = float(override) if override is not None else fnacc_auto # Amortissement linéaire indicatif (comptable, non fiscal). useful_life_years = float(snap["view_admin"].get("useful_life_years") or CCA_BY_ID.get(snap["view_cpa"]["cca_class"]["id"], {}).get("default_useful_life_years", 4)) snap["view_admin"]["useful_life_years"] = useful_life_years accum_l, nbv_l, remaining_l = _compute_linear_deprec(cost, isd, useful_life_years) snap["view_cpa"]["accum_deprec_linear_cad"] = accum_l snap["view_cpa"]["nbv_linear_cad"] = nbv_l snap["view_cpa"]["remaining_useful_life_years"] = remaining_l return snap def _persist(self, asset: dict) -> None: """Écrit un actif existant (UPDATE) à partir de sa structure complète.""" data = copy.deepcopy(asset) for k in ("id", "type", "name", "parent_id"): data.pop(k, None) self._conn.execute( "UPDATE assets SET type=?, name=?, parent_id=?, data=?, updated_at=? WHERE id=?", (asset["type"], asset["name"], asset["parent_id"], json.dumps(data, ensure_ascii=False), _now_iso(), asset["id"]), ) self._conn.commit() # ---- Lectures ---- def _inject_bearer_tags(self, assets: list[dict]) -> None: """Propage l'étiquette du porteur à ses siblings et marque tag_inherited.""" bearer_tag: dict[int, str] = {} for a in assets: if a["type"] == PIECE_TYPE and a["parent_id"] is not None and a["view_admin"].get("is_tag_bearer"): bearer_tag[a["parent_id"]] = a["view_admin"].get("inventory_tag", "") for a in assets: if a["type"] == PIECE_TYPE and a["parent_id"] is not None: is_bearer = bool(a["view_admin"].get("is_tag_bearer")) a["view_admin"]["tag_inherited"] = not is_bearer if not is_bearer: a["view_admin"]["inventory_tag"] = bearer_tag.get(a["parent_id"], "") def all(self) -> list[dict]: cur = self._conn.execute("SELECT * FROM assets ORDER BY id") result = [self._with_computed_base(self._row_to_asset(r)) for r in cur.fetchall()] self._inject_bearer_tags(result) return result def get(self, asset_id: int) -> dict | None: a = self._raw_get(asset_id) if a is None: return None result = self._with_computed_base(a) if result["type"] == PIECE_TYPE and result["parent_id"] is not None: siblings = self._raw_children(result["parent_id"]) bearer = next((s for s in siblings if s["view_admin"].get("is_tag_bearer")), None) is_bearer = bool(result["view_admin"].get("is_tag_bearer")) result["view_admin"]["tag_inherited"] = not is_bearer if not is_bearer: result["view_admin"]["inventory_tag"] = bearer["view_admin"].get("inventory_tag", "") if bearer else "" return result # ---- Création ---- def create_asset(self, payload: dict) -> dict: with self._lock: atype = payload.get("type") if atype not in (PARENT_TYPE, PIECE_TYPE): raise ValueError("Type invalide (parent|piece).") name = payload.get("name") or "Immo" cost = float(payload.get("cost_cad") or 0) currency = payload.get("currency") or "CAD" now = _now_iso() data = { "path": "", "view_admin": { "serial_number": payload.get("serial_number") or "", "inventory_tag": payload.get("inventory_tag") or "", "manufacturer": payload.get("manufacturer") or "", "model": payload.get("model") or "", "form_factor": payload.get("form_factor") or "", # Propre au PARENT : assembleur de l'immobilisation. "assembler": payload.get("assembler") or ("Chezlepro Inc." if atype == PARENT_TYPE else ""), "canvas": {"x": 0, "y": 0, "width": 240, "height": 140, "z_index": 0}, "operational_status": payload.get("operational_status") or "Stock", # Coût estimé pour remplacer à neuf (assurances, budget). Propre aux PIÈCES. "replacement_cost_cad": float(payload.get("replacement_cost_cad") or 0), # True sur exactement une pièce par immobilisation : source de l'étiquette partagée. "is_tag_bearer": False, }, "view_cpa": { "original_cost": {"amount": cost, "currency": currency}, "exchange_rate_applied": float(payload.get("exchange_rate") or 1.0), "exchange_rate_date": payload.get("acquisition_date") or now[:10], "cost_pre_tax_cad": cost, "non_recoverable_taxes_and_duties": float(payload.get("non_recoverable_taxes_and_duties") or 0), "capitalizable_base_cad": cost, # FNACC saisie manuellement (override). Null = utiliser le calcul auto. "fnacc_override_cad": payload.get("fnacc_override_cad"), "cca_class": { "id": int(payload.get("cca_class_id") or 50), "description": CCA_BY_ID.get(int(payload.get("cca_class_id") or 50), {}).get("description", "Matériel informatique"), "dpa_rate": CCA_BY_ID.get(int(payload.get("cca_class_id") or 50), {}).get("dpa_rate", 0.55), }, "acquisition_date": payload.get("acquisition_date") or now[:10], # Parent : date du jour par défaut. Pièce libre : vide — ne s'amortit # pas tant qu'elle n'est pas rattachée (in_service_date posée par attach). "in_service_date": payload.get("in_service_date") or (now[:10] if atype == PARENT_TYPE else ""), "half_year_rule_applies": bool(payload.get("half_year_rule_applies", True)), }, "view_auditor": { "supplier": payload.get("supplier") or "", "invoice_pdf_file_id": payload.get("invoice_pdf_file_id") or "", "invoice_number": payload.get("invoice_number") or "", "audit_trail": [{ "action": "CREATED", "description": f"{'Immobilisation' if atype==PARENT_TYPE else 'Pièce'} créée", "timestamp": now, "previous_value": None, "new_value": {"type": atype, "parent_id": None}, }], }, } cur = self._conn.execute( "INSERT INTO assets (type, name, parent_id, data, created_at, updated_at) VALUES (?,?,?,?,?,?)", (atype, name, None, json.dumps(data, ensure_ascii=False), now, now), ) self._conn.commit() new_id = cur.lastrowid self._set_path(new_id) return self.get(new_id) def _set_path(self, asset_id: int) -> None: a = self._raw_get(asset_id) if not a: return if a["parent_id"]: a["path"] = f"parent_{a['parent_id']}.piece_{asset_id}" else: a["path"] = (f"parent_{asset_id}" if a["type"] == PARENT_TYPE else f"libre_{asset_id}") self._persist(a) # ---- Édition ---- # Champs éditables par chemin "view.cle". La base consolidée du parent est exclue. def update_asset(self, asset_id: int, changes: dict) -> dict: with self._lock: asset = self._raw_get(asset_id) if asset is None: raise KeyError(f"Actif {asset_id} introuvable.") is_parent = asset["type"] == PARENT_TYPE audit_changes = [] for key, new_value in changes.items(): # Interdit : éditer la base consolidée d'un parent (calculée). if is_parent and key in ("view_cpa.capitalizable_base_cad", "view_cpa.cost_pre_tax_cad"): raise ValueError("La base consolidée d'un parent est calculée et non éditable.") old_value = self._get_path(asset, key) if old_value == new_value: continue self._set_path_value(asset, key, new_value) audit_changes.append((key, old_value, new_value)) # Nom de premier niveau si modifié via view ? On gère 'name' à part. if "name" in changes and changes["name"] != asset["name"]: audit_changes.append(("name", asset["name"], changes["name"])) asset["name"] = changes["name"] # Si la catégorie DPA change, le taux et la description suivent (catalogue). if "view_cpa.cca_class.id" in changes: cid = int(changes["view_cpa.cca_class.id"]) cat = CCA_BY_ID.get(cid) if cat: asset["view_cpa"]["cca_class"]["dpa_rate"] = cat["dpa_rate"] asset["view_cpa"]["cca_class"]["description"] = cat["description"] # Journalisation : toujours pour un parent (données fiscales). if is_parent and audit_changes: for (field, old, new) in audit_changes: asset["view_auditor"]["audit_trail"].append({ "action": "EDITED", "description": f"Champ « {field} » modifié", "timestamp": _now_iso(), "previous_value": {field: old}, "new_value": {field: new}, }) self._persist(asset) return self.get(asset_id) def _get_path(self, obj: dict, dotted: str): cur = obj for part in dotted.split("."): if not isinstance(cur, dict) or part not in cur: return None cur = cur[part] return cur def _set_path_value(self, obj: dict, dotted: str, value) -> None: parts = dotted.split(".") cur = obj for part in parts[:-1]: cur = cur.setdefault(part, {}) cur[parts[-1]] = value # ---- Attache / Détache ---- def attach(self, piece_id: int, parent_id: int) -> dict: with self._lock: piece = self._raw_get(piece_id) parent = self._raw_get(parent_id) if piece is None: raise KeyError(f"Pièce {piece_id} introuvable.") if parent is None: raise KeyError(f"Parent {parent_id} introuvable.") if parent["type"] != PARENT_TYPE: raise ValueError("La cible n'est pas une immobilisation parent.") if piece["type"] != PIECE_TYPE: raise ValueError("Seule une pièce peut être rattachée.") old_parent_id = piece["parent_id"] piece["parent_id"] = parent_id piece["path"] = f"parent_{parent_id}.piece_{piece_id}" piece["view_admin"]["operational_status"] = "Actif" now = _now_iso() # Pose la date de mise en service au premier rattachement seulement. # Re-rattachement (pièce déjà en service) : date conservée, DPA continue. if not piece["view_cpa"].get("in_service_date"): today = now[:10] piece["view_cpa"]["in_service_date"] = today piece["view_auditor"]["audit_trail"].append({ "action": "IN_SERVICE", "description": f"Mise en service posée automatiquement au rattachement au parent {parent_id} ({today})", "timestamp": now, "previous_value": {"in_service_date": None}, "new_value": {"in_service_date": today}, }) piece["view_auditor"]["audit_trail"].append({ "action": "ATTACHED", "description": f"Rattachée au parent {parent_id}", "timestamp": now, "previous_value": {"parent_id": old_parent_id}, "new_value": {"parent_id": parent_id}, }) self._persist(piece) return self.get(parent_id) def detach(self, piece_id: int, disposition: str) -> dict: with self._lock: piece = self._raw_get(piece_id) if piece is None: raise KeyError(f"Pièce {piece_id} introuvable.") old_parent_id = piece["parent_id"] piece["parent_id"] = None piece["path"] = f"libre_{piece_id}" status = "Rebut" if disposition == "scrap" else "Stock" piece["view_admin"]["operational_status"] = status piece["view_admin"]["inventory_tag"] = "" piece["view_admin"]["is_tag_bearer"] = False piece["view_auditor"]["audit_trail"].append({ "action": "DETACHED", "description": f"Détachée du parent {old_parent_id} — disposition {disposition}", "timestamp": _now_iso(), "previous_value": {"parent_id": old_parent_id}, "new_value": {"parent_id": None, "operational_status": status}, }) self._persist(piece) return self.get(piece_id) # ---- Porteur d'étiquette ---- def set_tag_bearer(self, piece_id: int) -> list[dict]: """Désigne une pièce comme porteur d'étiquette pour son immobilisation parente. - Tous les siblings perdent is_tag_bearer. - L'étiquette affichée sur tous les siblings et sur le parent est déduite du porteur à la lecture (aucun stockage sur les non-porteurs). - Migration rétroactive : si la pièce n'a pas d'étiquette et le parent en a une, celle-ci est copiée sur la pièce porteur. """ with self._lock: piece = self._raw_get(piece_id) if piece is None: raise KeyError(f"Pièce {piece_id} introuvable.") if piece["parent_id"] is None: raise ValueError("La pièce doit être rattachée à une immobilisation pour être porteur d'étiquette.") parent_id = piece["parent_id"] parent = self._raw_get(parent_id) if parent is None: raise KeyError(f"Parent {parent_id} introuvable.") now = _now_iso() # Migration rétroactive : hérite l'étiquette du parent si la pièce n'en a pas. if not piece["view_admin"].get("inventory_tag") and parent["view_admin"].get("inventory_tag"): piece["view_admin"]["inventory_tag"] = parent["view_admin"]["inventory_tag"] # Retirer le statut porteur des autres pièces. for sibling in self._raw_children(parent_id): if sibling["id"] == piece_id: continue if sibling["view_admin"].get("is_tag_bearer"): sibling["view_admin"]["is_tag_bearer"] = False sibling["view_auditor"]["audit_trail"].append({ "action": "TAG_BEARER_REMOVED", "description": f"Statut porteur retiré (nouveau porteur : pièce {piece_id})", "timestamp": now, "previous_value": {"is_tag_bearer": True}, "new_value": {"is_tag_bearer": False}, }) self._persist(sibling) # Désigner cette pièce comme porteur. tag = piece["view_admin"].get("inventory_tag", "") piece["view_admin"]["is_tag_bearer"] = True piece["view_auditor"]["audit_trail"].append({ "action": "TAG_BEARER_SET", "description": f"Désignée porteur d'étiquette pour l'immo. {parent_id} — étiquette : {tag!r}", "timestamp": now, "previous_value": {"is_tag_bearer": False}, "new_value": {"is_tag_bearer": True}, }) self._persist(piece) # Journaliser sur le parent. parent["view_auditor"]["audit_trail"].append({ "action": "TAG_BEARER_CHANGED", "description": f"Porteur d'étiquette → pièce {piece_id} ({piece['name']}) — étiquette : {tag!r}", "timestamp": now, "previous_value": None, "new_value": {"tag_bearer_piece_id": piece_id, "inventory_tag": tag}, }) self._persist(parent) return self.all() # ---- Suppression ---- def delete_asset(self, asset_id: int) -> None: with self._lock: # Détache d'abord les pièces enfants (deviennent libres). for child in self._raw_children(asset_id): child["parent_id"] = None child["path"] = f"libre_{child['id']}" child["view_admin"]["operational_status"] = "Stock" self._persist(child) self._conn.execute("DELETE FROM assets WHERE id = ?", (asset_id,)) self._conn.commit() # ---- Export fiscal ---- def fiscal_export(self) -> dict: """Regroupe les PARENTS par catégorie DPA avec base consolidée.""" parents = [a for a in self.all() if a["type"] == PARENT_TYPE] by_class: dict[int, dict] = {} for p in parents: cca = p["view_cpa"]["cca_class"] cid = cca["id"] entry = by_class.setdefault(cid, { "cca_class_id": cid, "description": cca["description"], "dpa_rate": cca["dpa_rate"], "items": [], "total_base_cad": 0.0, }) base = float(p["view_cpa"]["capitalizable_base_cad"]) entry["items"].append({ "id": p["id"], "name": p["name"], "inventory_tag": p["view_admin"]["inventory_tag"], "base_cad": base, "in_service_date": p["view_cpa"]["in_service_date"], "half_year_rule": p["view_cpa"]["half_year_rule_applies"], "pieces_count": p["view_admin"].get("pieces_count", 0), }) entry["total_base_cad"] = round(entry["total_base_cad"] + base, 2) return {"classes": list(by_class.values()), "generated_at": _now_iso()} def reset(self) -> None: with self._lock: self._conn.execute("DELETE FROM assets") self._conn.execute("DELETE FROM cca_pool_years") self._conn.commit() def backup(self, dest_path: Path) -> None: dest_path.parent.mkdir(parents=True, exist_ok=True) dest_conn = sqlite3.connect(str(dest_path)) with self._lock: self._conn.backup(dest_conn) dest_conn.close() def restore_from(self, src_path: Path) -> None: src_conn = sqlite3.connect(str(src_path)) with self._lock: src_conn.backup(self._conn) src_conn.close() store = AssetStore() # Registre fiscal (pool de catégorie) branché sur le même store/connexion. from .cca_registry import CcaRegistry registry = CcaRegistry(store._conn, store, store._lock)