Daniel Allaire
691537400f
Remplace les mentions textuelles par des URLs directes dans les docstrings de cca_pool.py, cca_registry.py et le commentaire CCA_CATEGORIES dans store.py. - laws-lois.justice.gc.ca : LIR art. 13, 20 ; Règl. art. 1100 - canada.ca : Guide T2 (T4012) pour l'Annexe 8 / Schedule 8 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
580 lines
28 KiB
Python
580 lines
28 KiB
Python
"""
|
|
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)
|