monregistraire/app/main.py
Daniel Allaire be7a64a6cc Lancement v1.0 — Gestionnaire d'immobilisations fiscales (Chezlepro Inc.)
Application web locale (FastAPI + SQLite + React SPA) pour la gestion des
immobilisations fiscales d'une PME québécoise (T2 / CO-17).

─── Modèle de données ───────────────────────────────────────────────────────
• Deux niveaux : PARENT (immobilisation fiscale) + PIÈCE (inventaire physique)
• Trois vues par actif : Propriétaire, CPA, Auditeur
• Porteur d'étiquette d'assemblage (unique par immobilisation)
• Piste d'audit journalisée sur chaque modification

─── Mécanique fiscale ───────────────────────────────────────────────────────
• Pool DPA par catégorie (Cat. 8, 10, 12, 46, 50) — amortissement dégressif
• Règle de la demi-année paramétrable (coeff. 0..1 par exercice)
• Perte finale et récupération d'amortissement calculées automatiquement
• Annexe 8 / Schedule 8 multi-catégories (JSON + CSV téléchargeable)
• Rapport fiscal HTML imprimable — lignes Schedule 1 (T2 : 104/107/410/414 ;
  CO-17 : 130/135/150/155)
• Export immobilisations par catégorie (JSON + CSV)

─── Fonctionnalités ─────────────────────────────────────────────────────────
• CRUD complet : parents et pièces, rattachement/détachement (libre/stock/rebut)
• Écriture comptable générée au détachement
• Registre de pool DPA multi-exercices avec override manuel
• Sauvegarde/restauration (SQLite native backup, horodatée)
• Factures jointes aux pièces (PDF/JPG/PNG, max 20 Mo, UUID + regex)
• UI mobile-first : bottom nav, FAB, bottom sheet, safe-area iPhone
• Guide utilisateur complet in-app (aide.html, 11 sections)
• Données de démonstration riches (6 immos, 5 catégories, perte finale +
  récupération incluses)

─── Qualité ─────────────────────────────────────────────────────────────────
• 50 tests pytest — 100 % verts (calculs DPA, export, backup, factures, seed)
• Lancement en 1 commande : ./lancer.sh (Mac/Linux) ou lancer.bat (Windows)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-10 10:30:53 -04:00

828 lines
33 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
Serveur FastAPI v3 — persistance SQLite, édition complète, export fiscal.
Routes :
GET /api/assets -> tous les actifs
GET /api/assets/{id} -> un actif
POST /api/assets -> créer (parent | piece)
PATCH /api/assets/{id} -> éditer (changes: {chemin.pointé: valeur})
POST /api/assets/{id}/attach -> rattacher une pièce à un parent
POST /api/assets/{id}/detach -> détacher (free | stock | scrap)
DELETE /api/assets/{id} -> supprimer (pièces enfants -> libres)
GET /api/fiscal-export -> regroupement par catégorie (JSON)
GET /api/fiscal-export.csv -> même chose en CSV téléchargeable
POST /api/seed-demo -> insère des données d'exemple
POST /api/reset -> vide la base
"""
from __future__ import annotations
import csv
import io
import re
import uuid
from datetime import date, datetime, timezone
from pathlib import Path
from fastapi import FastAPI, File, HTTPException, UploadFile
import html as _html
from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from .accounting import process_asset_detachment
from .store import store, registry, PARENT_TYPE, PIECE_TYPE, CCA_CATEGORIES, CCA_BY_ID
BASE_DIR = Path(__file__).resolve().parent.parent
STATIC_DIR = BASE_DIR / "static"
BACKUP_DIR = store._db_path.parent / "backups"
INVOICES_DIR = store._db_path.parent / "invoices"
_ALLOWED_EXTS = {".pdf", ".jpg", ".jpeg", ".png"}
_INVOICE_RE = re.compile(r"[0-9a-f]{32}\.(pdf|jpg|jpeg|png)")
_MAX_UPLOAD = 20 * 1024 * 1024 # 20 Mo
app = FastAPI(title="Gestionnaire d'immobilisations")
class CreateBody(BaseModel):
type: str
name: str
cost_cad: float = 0
currency: str = "CAD"
serial_number: str | None = None
inventory_tag: str | None = None
manufacturer: str | None = None
model: str | None = None
form_factor: str | None = None
assembler: str | None = None
operational_status: str | None = None
cca_class_id: int = 50
cca_class_desc: str | None = None
dpa_rate: float = 0.55
exchange_rate: float = 1.0
non_recoverable_taxes_and_duties: float = 0
acquisition_date: str | None = None
in_service_date: str | None = None
supplier: str | None = None
invoice_number: str | None = None
invoice_pdf_file_id: str | None = None
replacement_cost_cad: float = 0
fnacc_override_cad: float | None = None
class UpdateBody(BaseModel):
changes: dict # { "view_admin.serial_number": "...", "name": "...", ... }
class AttachBody(BaseModel):
parent_id: int
class DetachBody(BaseModel):
disposition: str
detachment_date: str | None = None
@app.get("/api/cca-categories")
def cca_categories() -> list[dict]:
return CCA_CATEGORIES
@app.get("/api/assets")
def list_assets() -> list[dict]:
return store.all()
@app.get("/api/assets/{asset_id}")
def get_asset(asset_id: int) -> dict:
a = store.get(asset_id)
if a is None:
raise HTTPException(404, "Actif introuvable.")
return a
@app.post("/api/assets")
def create_asset(body: CreateBody) -> dict:
try:
store.create_asset(body.model_dump())
except ValueError as e:
raise HTTPException(400, str(e))
return {"assets": store.all()}
@app.patch("/api/assets/{asset_id}")
def update_asset(asset_id: int, body: UpdateBody) -> dict:
try:
store.update_asset(asset_id, body.changes)
except (KeyError, ValueError) as e:
raise HTTPException(400, str(e))
return {"assets": store.all()}
@app.post("/api/assets/{asset_id}/attach")
def attach_asset(asset_id: int, body: AttachBody) -> dict:
try:
store.attach(asset_id, body.parent_id)
except (KeyError, ValueError) as e:
raise HTTPException(400, str(e))
return {"assets": store.all()}
@app.post("/api/assets/{asset_id}/detach")
def detach_asset(asset_id: int, body: DetachBody) -> dict:
asset = store.get(asset_id)
if asset is None:
raise HTTPException(404, "Pièce introuvable.")
if body.disposition not in ("free", "stock", "scrap"):
raise HTTPException(400, "Disposition invalide (free|stock|scrap).")
accounting = None
if body.disposition in ("stock", "scrap"):
d = date.fromisoformat(body.detachment_date) if body.detachment_date else date.today()
try:
accounting = process_asset_detachment(asset, d, body.disposition).to_dict() # type: ignore[arg-type]
except (ValueError, ArithmeticError) as e:
raise HTTPException(400, str(e))
store.detach(asset_id, body.disposition)
return {"accounting": accounting, "assets": store.all()}
@app.post("/api/assets/{asset_id}/set-tag-bearer")
def set_tag_bearer(asset_id: int) -> dict:
try:
return {"assets": store.set_tag_bearer(asset_id)}
except (KeyError, ValueError) as e:
raise HTTPException(400, str(e))
@app.delete("/api/assets/{asset_id}")
def delete_asset(asset_id: int) -> dict:
if store.get(asset_id) is None:
raise HTTPException(404, "Actif introuvable.")
store.delete_asset(asset_id)
return {"assets": store.all()}
@app.get("/api/fiscal-export")
def fiscal_export() -> dict:
return store.fiscal_export()
@app.get("/api/fiscal-export.csv")
def fiscal_export_csv() -> StreamingResponse:
data = store.fiscal_export()
buf = io.StringIO()
w = csv.writer(buf)
w.writerow(["Categorie DPA", "Taux DPA", "Immobilisation", "Etiquette",
"Base amortissable CAD", "Mise en service", "Demi-annee", "Nb pieces"])
for cls in data["classes"]:
for it in cls["items"]:
w.writerow([
f"Cat. {cls['cca_class_id']} - {cls['description']}",
f"{cls['dpa_rate']*100:.0f}%",
it["name"], it["inventory_tag"],
f"{it['base_cad']:.2f}", it["in_service_date"],
"Oui" if it["half_year_rule"] else "Non", it["pieces_count"],
])
w.writerow([f"TOTAL Cat. {cls['cca_class_id']}", "", "", "",
f"{cls['total_base_cad']:.2f}", "", "", ""])
w.writerow([])
buf.seek(0)
return StreamingResponse(
iter([buf.getvalue()]),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=export_fiscal.csv"},
)
@app.post("/api/invoices")
async def upload_invoice(file: UploadFile = File(...)) -> dict:
ext = Path(file.filename or "").suffix.lower()
if ext not in _ALLOWED_EXTS:
raise HTTPException(400, f"Format non supporté. Acceptés : PDF, JPG, PNG.")
content = await file.read()
if len(content) > _MAX_UPLOAD:
raise HTTPException(400, "Fichier trop volumineux (max 20 Mo).")
INVOICES_DIR.mkdir(parents=True, exist_ok=True)
file_id = uuid.uuid4().hex + ext
(INVOICES_DIR / file_id).write_bytes(content)
return {"file_id": file_id}
@app.get("/api/invoices/{file_id}")
def get_invoice(file_id: str) -> FileResponse:
if not _INVOICE_RE.fullmatch(file_id):
raise HTTPException(400, "Identifiant invalide.")
path = INVOICES_DIR / file_id
if not path.exists():
raise HTTPException(404, "Fichier introuvable.")
mt = ("application/pdf" if file_id.endswith(".pdf")
else "image/png" if file_id.endswith(".png")
else "image/jpeg")
return FileResponse(str(path), media_type=mt)
@app.delete("/api/invoices/{file_id}")
def delete_invoice(file_id: str) -> dict:
if not _INVOICE_RE.fullmatch(file_id):
raise HTTPException(400, "Identifiant invalide.")
path = INVOICES_DIR / file_id
if path.exists():
path.unlink()
return {"ok": True}
@app.get("/api/backups")
def list_backups() -> list[dict]:
BACKUP_DIR.mkdir(parents=True, exist_ok=True)
result = []
for f in sorted(BACKUP_DIR.glob("immos_*.db"), reverse=True):
stat = f.stat()
result.append({
"filename": f.name,
"size_kb": round(stat.st_size / 1024, 1),
"created_at": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
.astimezone().isoformat(timespec="seconds"),
})
return result
@app.post("/api/backup")
def create_backup() -> dict:
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
dest = BACKUP_DIR / f"immos_{ts}.db"
store.backup(dest)
return {"filename": dest.name, "backups": list_backups()}
class RestoreBody(BaseModel):
filename: str
@app.post("/api/restore")
def restore_backup(body: RestoreBody) -> dict:
if not re.fullmatch(r"immos_\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}\.db", body.filename):
raise HTTPException(400, "Nom de fichier invalide.")
src = BACKUP_DIR / body.filename
if not src.exists():
raise HTTPException(404, "Sauvegarde introuvable.")
store.restore_from(src)
return {"assets": store.all()}
@app.post("/api/reset")
def reset() -> dict:
store.reset()
return {"assets": store.all()}
@app.post("/api/seed-demo")
def seed_demo() -> dict:
"""Jeu de données complet — 6 immobilisations, 5 catégories DPA, pièces
actives/stock/rebut, porteurs d'étiquette, registres multi-exercices
couvrant DPA standard, perte finale (Cat. 12) et récupération (Cat. 10)."""
today_str = date.today().isoformat()
# ── 1. Cat. 50 — Serveur Dell PowerEdge R760 (actif, acquis 2024) ─────────
r760 = store.create_asset({
"type": PARENT_TYPE, "name": "Serveur Dell PowerEdge R760",
"manufacturer": "Dell", "model": "PowerEdge R760", "form_factor": "2U",
"cca_class_id": 50, "operational_status": "Actif",
"in_service_date": "2024-02-20", "assembler": "Dell Canada Inc.",
})
r760_pieces = [
("Châssis + carte mère R760", "Dell", "R760 Base", 23_436.05, "SRV-CHS-001", "DC-QC-SRV-001"),
("RAM DDR5 64 Go Slot 1", "Samsung", "M321R8GA0BB0", 420.00, "MEM-001-S1", ""),
("RAM DDR5 64 Go Slot 2", "Samsung", "M321R8GA0BB0", 420.00, "MEM-001-S2", ""),
("SSD NVMe 3,84 To", "Micron", "7450 PRO", 980.00, "SSD-001-NV1", ""),
("GPU NVIDIA A30 PCIe", "NVIDIA", "A30", 4_500.00, "GPU-001-A30", ""),
]
r760_ids = []
for nm, mf, md, cost, sn, tag in r760_pieces:
pc = store.create_asset({
"type": PIECE_TYPE, "name": nm, "manufacturer": mf, "model": md,
"cost_cad": cost, "serial_number": sn, "inventory_tag": tag,
"acquisition_date": "2024-01-15",
"supplier": "Dell Canada Inc.", "invoice_number": "DELL-CA-2024-R760",
})
store.attach(pc["id"], r760["id"])
r760_ids.append(pc["id"])
store.set_tag_bearer(r760_ids[0]) # Châssis = porteur
# ── 2. Cat. 50 — Serveur HP ProLiant DL380 Gen10 (actif, acquis 2023) ─────
dl380 = store.create_asset({
"type": PARENT_TYPE, "name": "Serveur HP ProLiant DL380 Gen10",
"manufacturer": "HP", "model": "ProLiant DL380 Gen10", "form_factor": "2U",
"cca_class_id": 50, "operational_status": "Actif",
"in_service_date": "2023-04-01", "assembler": "HP Inc.",
})
dl380_pieces = [
("Châssis HP DL380 Gen10", "HP", "DL380 Gen10", 18_200.00, "SRV-HP-001", "DC-QC-SRV-002"),
("RAM DDR4 128 Go", "Samsung", "M393A8G40AB2", 890.00, "MEM-HP-001", ""),
("SSD SATA 7,68 To", "Micron", "5400 MAX", 650.00, "SSD-HP-SATA-1", ""),
]
dl380_ids = []
for nm, mf, md, cost, sn, tag in dl380_pieces:
pc = store.create_asset({
"type": PIECE_TYPE, "name": nm, "manufacturer": mf, "model": md,
"cost_cad": cost, "serial_number": sn, "inventory_tag": tag,
"acquisition_date": "2023-03-10",
"supplier": "HP Inc.", "invoice_number": "HP-CA-2023-DL380",
})
store.attach(pc["id"], dl380["id"])
dl380_ids.append(pc["id"])
store.set_tag_bearer(dl380_ids[0]) # Châssis HP = porteur
# ── 3. Cat. 8 — Onduleur APC Smart-UPS RT 10 kVA (actif, acquis 2024) ─────
ups = store.create_asset({
"type": PARENT_TYPE, "name": "Onduleur APC Smart-UPS RT 10 kVA",
"manufacturer": "APC", "model": "SURT10000XLI", "form_factor": "Rack 6U",
"cca_class_id": 8, "operational_status": "Actif",
"in_service_date": "2024-07-01", "assembler": "APC by Schneider Electric",
})
ups_pieces = [
("Boîtier UPS SURT10000XLI", "APC", "SURT10000XLI", 8_450.00, "UPS-001-BOITIER", "DC-QC-UPS-001"),
("Pack batteries SURT192XLBP", "APC", "SURT192XLBP", 1_200.00, "UPS-001-BAT", ""),
("Carte réseau UPS AP9631", "APC", "AP9631", 320.00, "UPS-001-NMC", ""),
]
ups_ids = []
for nm, mf, md, cost, sn, tag in ups_pieces:
pc = store.create_asset({
"type": PIECE_TYPE, "name": nm, "manufacturer": mf, "model": md,
"cost_cad": cost, "serial_number": sn, "inventory_tag": tag,
"cca_class_id": 8, # pièces héritent la catégorie du parent pour le pool
"acquisition_date": "2024-06-15",
"supplier": "APC by Schneider Electric", "invoice_number": "APC-QC-2024-UPS",
})
store.attach(pc["id"], ups["id"])
ups_ids.append(pc["id"])
store.set_tag_bearer(ups_ids[0]) # Boîtier = porteur
# ── 4. Cat. 46 — Switch Cisco Catalyst 9300-24T (actif, acquis 2025) ───────
sw = store.create_asset({
"type": PARENT_TYPE, "name": "Switch Cisco Catalyst 9300-24T",
"manufacturer": "Cisco", "model": "C9300-24T", "form_factor": "Rack 1U",
"cca_class_id": 46, "operational_status": "Actif",
"in_service_date": "2025-02-01", "assembler": "Cisco Systems Canada",
})
sw_pieces = [
("Switch Cisco C9300-24T", "Cisco", "C9300-24T", 12_500.00, "SW-001-CISCO", "DC-QC-SW-001"),
("Module SFP+ 10G #1", "Cisco", "SFP-10G-SR", 350.00, "SFP-001-1", ""),
("Module SFP+ 10G #2", "Cisco", "SFP-10G-SR", 350.00, "SFP-001-2", ""),
("Module SFP+ 10G #3", "Cisco", "SFP-10G-SR", 350.00, "SFP-001-3", ""),
]
sw_ids = []
for nm, mf, md, cost, sn, tag in sw_pieces:
pc = store.create_asset({
"type": PIECE_TYPE, "name": nm, "manufacturer": mf, "model": md,
"cost_cad": cost, "serial_number": sn, "inventory_tag": tag,
"cca_class_id": 46, # pièces héritent la catégorie du parent pour le pool
"acquisition_date": "2025-01-10",
"supplier": "Cisco Systems Canada", "invoice_number": "CISCO-CA-2025-SW001",
})
store.attach(pc["id"], sw["id"])
sw_ids.append(pc["id"])
store.set_tag_bearer(sw_ids[0]) # Switch = porteur
# ── 5. Cat. 12 — Licence Veeam Backup Enterprise (non renouvelée 2025) ─────
# Démontre la perte finale : 100 % DPA, pool physiquement vide en 2025.
veeam = store.create_asset({
"type": PARENT_TYPE, "name": "Licence Veeam Backup Enterprise",
"manufacturer": "Veeam", "model": "VBR Enterprise Plus", "form_factor": "Logiciel",
"cca_class_id": 12, "operational_status": "Rebut",
"in_service_date": "2024-03-01", "assembler": "Veeam Software Group",
})
lic_veeam = store.create_asset({
"type": PIECE_TYPE, "name": "Licence Veeam Enterprise Plus (3 ans)",
"manufacturer": "Veeam", "model": "VBR-ENT-PLUS-3Y",
"cost_cad": 3_000.00, "serial_number": "LIC-VBR-2024-001",
"cca_class_id": 12, # logiciel → Cat. 12, pool distinct de l'équipement
"acquisition_date": "2024-02-15",
"supplier": "Veeam Software Group", "invoice_number": "VBR-QC-2024-001",
})
store.attach(lic_veeam["id"], veeam["id"])
store.set_tag_bearer(lic_veeam["id"])
store.detach(lic_veeam["id"], "scrap") # Licence non renouvelée → rebut
# ── 6. Cat. 10 — Serveur IBM x3650 M4 (fin de vie, vendu 2024) ───────────
# Démontre la récupération : produit de cession > FNACC résiduelle.
# Actif vendu avant l'adoption de l'application ; suivi via opening_ucc_override.
store.create_asset({
"type": PARENT_TYPE, "name": "Serveur IBM x3650 M4 (fin de vie)",
"manufacturer": "IBM", "model": "System x3650 M4", "form_factor": "2U",
"cca_class_id": 10, "operational_status": "Rebut",
"in_service_date": "2019-06-01", "acquisition_date": "2019-05-15",
"assembler": "IBM Canada",
})
# ── Pièces libres (stock de rechange) ─────────────────────────────────────
store.create_asset({
"type": PIECE_TYPE, "name": "RAM DDR5 32 Go pièce de rechange",
"manufacturer": "Samsung", "model": "M321R4GA0BB0",
"cost_cad": 210.00, "serial_number": "MEM-SPARE-001",
"acquisition_date": today_str, "supplier": "Dell Canada Inc.",
})
store.create_asset({
"type": PIECE_TYPE, "name": "SSD NVMe 1,92 To pièce de rechange",
"manufacturer": "Micron", "model": "7450 PRO 1.92T",
"cost_cad": 540.00, "serial_number": "SSD-SPARE-001",
"acquisition_date": today_str, "supplier": "Dell Canada Inc.",
})
# ── Pièce mise au rebut avec piste d'audit complète ───────────────────────
old_nic = store.create_asset({
"type": PIECE_TYPE, "name": "Vieille carte réseau Broadcom BCM5720",
"manufacturer": "Broadcom", "model": "BCM5720", "form_factor": "PCIe",
"cost_cad": 280.00, "serial_number": "NIC-LEGACY-001",
"acquisition_date": "2022-01-10",
"supplier": "Dell Canada Inc.", "invoice_number": "DELL-CA-2022-LEGACY",
})
store.attach(old_nic["id"], dl380["id"]) # Rattachée au DL380
store.detach(old_nic["id"], "scrap") # Puis mise au rebut
# ── Registre fiscal (pool DPA) ────────────────────────────────────────────
# Cat. 50 — 3 exercices 2023-2025 (DPA standard)
registry.upsert_year(2023, 50, dispositions=0, adjustment_coeff=0.5)
registry.upsert_year(2024, 50, dispositions=0, adjustment_coeff=0.5)
registry.upsert_year(2025, 50, dispositions=0, adjustment_coeff=1.0)
# Cat. 8 — 2 exercices 2024-2025
registry.upsert_year(2024, 8, dispositions=0, adjustment_coeff=0.5)
registry.upsert_year(2025, 8, dispositions=0, adjustment_coeff=1.0)
# Cat. 46 — 1 exercice 2025
registry.upsert_year(2025, 46, dispositions=0, adjustment_coeff=0.5)
# Cat. 12 — Veeam : 2024 (DPA 100 %) + 2025 perte finale
registry.upsert_year(2024, 12, dispositions=0, adjustment_coeff=0.5)
registry.upsert_year(2025, 12, dispositions=0, adjustment_coeff=1.0,
pool_physically_empty=True)
# Cat. 10 — IBM x3650 M4 : historique 2022-2024, récupération en 2024
# FNACC au 01-01-2022 = 4 500 $ (dérivé avant utilisation de l'application).
# Vendu 3 000 $ en 2024 → solde négatif → récupération d'amortissement.
registry.upsert_year(2022, 10, additions_override=0.0, dispositions=0,
adjustment_coeff=0.5, opening_ucc_override=4_500.0)
registry.upsert_year(2023, 10, dispositions=0, adjustment_coeff=1.0)
registry.upsert_year(2024, 10, dispositions=3_000.0, adjustment_coeff=1.0,
pool_physically_empty=True)
return {"assets": store.all()}
class PoolYearBody(BaseModel):
year: int
cca_class_id: int = 50
additions_override: float | None = None
dispositions: float = 0
adjustment_coeff: float = 0.5
pool_physically_empty: bool = False
opening_ucc_override: float | None = None
@app.get("/api/cca-registry")
def cca_registry(cca_class_id: int = 50) -> dict:
return registry.compute(cca_class_id)
@app.post("/api/cca-registry/year")
def cca_upsert_year(body: PoolYearBody) -> dict:
registry.upsert_year(
body.year, body.cca_class_id,
additions_override=body.additions_override,
dispositions=body.dispositions,
adjustment_coeff=body.adjustment_coeff,
pool_physically_empty=body.pool_physically_empty,
opening_ucc_override=body.opening_ucc_override,
)
return registry.compute(body.cca_class_id)
@app.delete("/api/cca-registry/year/{year}")
def cca_delete_year(year: int, cca_class_id: int = 50) -> dict:
registry.delete_year(year, cca_class_id)
return registry.compute(cca_class_id)
@app.get("/api/cca-registry.csv")
def cca_registry_csv(cca_class_id: int = 50) -> StreamingResponse:
data = registry.compute(cca_class_id)
buf = io.StringIO()
w = csv.writer(buf)
w.writerow(["Annee", "FNACC ouverture", "Ajouts", "Dispositions",
"Coeff ajust.", "Ajustement", "Base DPA", "Taux",
"DPA", "FNACC cloture", "Recuperation", "Perte finale"])
for y in data["years"]:
w.writerow([y["year"], y["opening_ucc"], y["additions"], y["dispositions"],
y["adjustment_coeff"], y["adjustment"], y["base_for_cca"],
y["cca_rate"], y["cca_claimed"], y["closing_ucc"],
y["recapture"], y["terminal_loss"]])
buf.seek(0)
return StreamingResponse(iter([buf.getvalue()]), media_type="text/csv",
headers={"Content-Disposition": f"attachment; filename=registre_cat{cca_class_id}.csv"})
class AddDispositionBody(BaseModel):
year: int
cca_class_id: int = 50
amount: float = 0
@app.post("/api/cca-registry/year/add-disposition")
def cca_add_disposition(body: AddDispositionBody) -> dict:
"""Cumule un produit de disposition sur une année du pool (sans écraser les autres champs)."""
data = registry.compute(body.cca_class_id)
row = next((y for y in data["years"] if y["year"] == body.year), None)
current_disp = float(row["dispositions"]) if row else 0.0
current_coeff = float(row["adjustment_coeff"]) if row else 0.5
registry.upsert_year(
body.year, body.cca_class_id,
dispositions=current_disp + body.amount,
adjustment_coeff=current_coeff,
)
return registry.compute(body.cca_class_id)
@app.get("/api/annexe8")
def annexe8(year: int = None) -> dict:
"""Données Annexe 8 / Schedule 8 pour toutes les catégories actives, un exercice."""
from decimal import Decimal
from datetime import datetime, timezone
if year is None:
year = date.today().year - 1
class_ids = sorted({a["view_cpa"]["cca_class"]["id"] for a in store.all()})
classes = []
for cid in class_ids:
cat = CCA_BY_ID.get(cid, {"description": "", "dpa_rate": 0.55})
rate = Decimal(str(cat["dpa_rate"]))
pool = registry.compute(cid, cca_rate=rate)
year_data = next((y for y in pool["years"] if y["year"] == year), None)
classes.append({
"cca_class_id": cid,
"description": cat["description"],
"dpa_rate": float(rate),
"physical_count": pool["physical_count"],
"year_data": year_data,
})
return {
"year": year,
"classes": classes,
"generated_at": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
}
@app.get("/api/annexe8.csv")
def annexe8_csv(year: int = None) -> StreamingResponse:
data = annexe8(year)
yr = data["year"]
buf = io.StringIO()
w = csv.writer(buf)
w.writerow([f"ANNEXE 8 — DÉDUCTION POUR AMORTISSEMENT (DPA) — Exercice {yr}"])
w.writerow(["Chezlepro Inc."])
w.writerow([])
w.writerow(["Catégorie", "Description", "Taux", "FNACC ouv.", "Ajouts", "Dispositions",
"Coeff.", "Ajust. $", "Base DPA", "DPA réclamée", "FNACC clôt.",
"Récupération", "Perte finale", "Note"])
total_cca = 0.0
for cls in data["classes"]:
y = cls["year_data"]
if y:
w.writerow([
f"Cat. {cls['cca_class_id']}", cls["description"],
f"{cls['dpa_rate']*100:.0f}%",
y["opening_ucc"], y["additions"], y["dispositions"],
y["adjustment_coeff"], y["adjustment"], y["base_for_cca"],
y["cca_claimed"], y["closing_ucc"],
y["recapture"], y["terminal_loss"], y["note"],
])
total_cca += float(y["cca_claimed"])
else:
w.writerow([f"Cat. {cls['cca_class_id']}", cls["description"],
f"{cls['dpa_rate']*100:.0f}%"] + [""] * 10 + [f"Aucune donnée pour {yr}"])
w.writerow([])
w.writerow(["", "", "", "", "", "", "", "", "DPA TOTALE", f"{total_cca:.2f}"])
w.writerow([])
w.writerow(["Calculs indicatifs — à valider par un CPA avant déclaration fiscale."])
buf.seek(0)
return StreamingResponse(iter([buf.getvalue()]), media_type="text/csv",
headers={"Content-Disposition": f"attachment; filename=annexe8_{yr}.csv"})
def _build_rapport_data(year: int) -> dict:
ann = annexe8(year)
total_cca = sum(float(c["year_data"]["cca_claimed"]) for c in ann["classes"] if c["year_data"])
total_recapture = sum(float(c["year_data"]["recapture"]) for c in ann["classes"] if c["year_data"])
total_terminal = sum(float(c["year_data"]["terminal_loss"]) for c in ann["classes"] if c["year_data"])
book_deprec = 0.0
for a in store.all():
if a["type"] != PIECE_TYPE:
continue
isd = a["view_cpa"].get("in_service_date") or ""
if not isd or isd[:4] > str(year):
continue
cost = float(a["view_cpa"].get("capitalizable_base_cad", 0))
life = float(a["view_admin"].get("useful_life_years") or 4)
if life > 0:
book_deprec += cost / life
return {
"year": year,
"annexe8": ann,
"schedule1": {
"book_deprec_cad": round(book_deprec, 2),
"cca_claimed_cad": round(total_cca, 2),
"recapture_cad": round(total_recapture, 2),
"terminal_loss_cad": round(total_terminal, 2),
},
"fiscal_export": store.fiscal_export(),
"generated_at": ann["generated_at"],
}
def _fmt(n: float) -> str:
return f"{n:,.2f} $".replace(",", " ")
def _render_rapport_html(d: dict) -> str:
yr = d["year"]
s1 = d["schedule1"]
gen = d["generated_at"]
e = _html.escape
def pool_rows() -> str:
rows = []
for cls in d["annexe8"]["classes"]:
y = cls["year_data"]
if not y:
rows.append(f'<tr><td>Cat.{cls["cca_class_id"]}</td><td colspan="9" style="color:#888">Aucune donnée pour {yr}</td></tr>')
continue
rows.append(
f'<tr>'
f'<td>Cat.&nbsp;{cls["cca_class_id"]}</td>'
f'<td>{e(cls["description"])}</td>'
f'<td class="r">{cls["dpa_rate"]*100:.0f}&nbsp;%</td>'
f'<td class="r">{_fmt(float(y["opening_ucc"]))}</td>'
f'<td class="r">{_fmt(float(y["additions"]))}</td>'
f'<td class="r">{_fmt(float(y["dispositions"]))}</td>'
f'<td class="r">{_fmt(float(y["base_for_cca"]))}</td>'
f'<td class="r"><b>{_fmt(float(y["cca_claimed"]))}</b></td>'
f'<td class="r">{_fmt(float(y["closing_ucc"]))}</td>'
f'<td class="r">{_fmt(float(y["recapture"])) if float(y["recapture"]) else ""}</td>'
f'<td class="r">{_fmt(float(y["terminal_loss"])) if float(y["terminal_loss"]) else ""}</td>'
f'</tr>'
)
return "\n".join(rows)
def asset_rows() -> str:
rows = []
for cls in d["fiscal_export"]["classes"]:
rows.append(
f'<tr class="subhead"><td colspan="5">Cat.&nbsp;{cls["cca_class_id"]}{e(cls["description"])} ({cls["dpa_rate"]*100:.0f}&nbsp;%)</td></tr>'
)
for it in cls["items"]:
rows.append(
f'<tr>'
f'<td>{e(it["name"])}</td>'
f'<td>{e(it["inventory_tag"] or "")}</td>'
f'<td class="r">{_fmt(float(it["base_cad"]))}</td>'
f'<td>{e(it["in_service_date"] or "")}</td>'
f'<td>{"Oui" if it["half_year_rule"] else "Non"}</td>'
f'</tr>'
)
rows.append(
f'<tr class="subtotal"><td colspan="2">Sous-total Cat.&nbsp;{cls["cca_class_id"]}</td>'
f'<td class="r">{_fmt(float(cls["total_base_cad"]))}</td><td colspan="2"></td></tr>'
)
return "\n".join(rows)
net = s1["book_deprec_cad"] + s1["recapture_cad"] - s1["cca_claimed_cad"] - s1["terminal_loss_cad"]
net_label = "Augmentation du revenu imposable" if net > 0 else "Réduction du revenu imposable"
return f"""<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8">
<title>Rapport fiscal {yr} — Chezlepro Inc.</title>
<style>
body {{ font-family: Arial, sans-serif; font-size: 12px; color: #111; max-width: 960px; margin: 0 auto; padding: 20px; }}
h1 {{ font-size: 18px; margin-bottom: 2px; }}
h2 {{ font-size: 13px; border-bottom: 2px solid #333; padding-bottom: 3px; margin-top: 28px; }}
table {{ width: 100%; border-collapse: collapse; margin-top: 10px; }}
th {{ background: #e8e8e8; padding: 5px 8px; text-align: left; border: 1px solid #bbb; font-size: 11px; }}
td {{ padding: 4px 8px; border: 1px solid #ddd; }}
.r {{ text-align: right; font-variant-numeric: tabular-nums; }}
tr.subtotal {{ background: #f5f5f5; font-weight: bold; }}
tr.subhead {{ background: #ddeeff; font-weight: bold; }}
.s1-ref {{ color: #555; font-size: 10px; }}
.highlight {{ background: #fffbe6; font-weight: bold; }}
.net {{ background: #e6f4ea; font-weight: bold; font-size: 13px; }}
.disclaimer {{ font-size: 10px; color: #666; border-top: 1px solid #ccc; margin-top: 24px; padding-top: 8px; }}
.print-btn {{ margin-bottom: 16px; padding: 8px 18px; cursor: pointer; font-size: 13px; }}
@media print {{
.print-btn {{ display: none; }}
@page {{ margin: 1.5cm; }}
}}
</style>
</head>
<body>
<button class="print-btn" onclick="window.print()">Imprimer / Enregistrer en PDF</button>
<h1>Rapport d'exercice fiscal {yr}</h1>
<p style="margin:4px 0 0"><b>Chezlepro Inc.</b> &nbsp;|&nbsp; Généré le {gen}</p>
<h2>1. Déduction pour amortissement — Annexe 8 (T2) / Annexe D (CO-17)</h2>
<table>
<thead>
<tr>
<th>Cat.</th><th>Description</th><th>Taux</th>
<th class="r">FNACC ouv.</th><th class="r">Ajouts</th><th class="r">Disp.</th>
<th class="r">Base DPA</th><th class="r">DPA réclamée</th>
<th class="r">FNACC clôt.</th><th class="r">Récupération</th><th class="r">Perte finale</th>
</tr>
</thead>
<tbody>
{pool_rows()}
</tbody>
</table>
<h2>2. Ajustements Annexe 1 / Schedule 1 — impact sur le revenu imposable</h2>
<table>
<thead>
<tr><th>Ajustement</th><th>Ligne T2 Sch.&nbsp;1</th><th>Ligne CO-17 Ann.&nbsp;1</th><th class="r">Montant ($)</th><th>Effet</th></tr>
</thead>
<tbody>
<tr>
<td>Amortissement comptable (add-back) <span class="s1-ref">— estimé selon durées de vie utiles de l'outil</span></td>
<td>104</td><td>130</td>
<td class="r">{_fmt(s1["book_deprec_cad"])}</td>
<td>+ revenu</td>
</tr>
<tr>
<td>Récupération d'amortissement</td>
<td>107</td><td>135</td>
<td class="r">{_fmt(s1["recapture_cad"]) if s1["recapture_cad"] else ""}</td>
<td>+ revenu</td>
</tr>
<tr class="highlight">
<td>DPA réclamée (total toutes catégories)</td>
<td>410</td><td>150</td>
<td class="r">{_fmt(s1["cca_claimed_cad"])}</td>
<td> revenu</td>
</tr>
<tr>
<td>Perte finale</td>
<td>414</td><td>155</td>
<td class="r">{_fmt(s1["terminal_loss_cad"]) if s1["terminal_loss_cad"] else ""}</td>
<td> revenu</td>
</tr>
<tr class="net">
<td colspan="3">{net_label}</td>
<td class="r">{_fmt(abs(net))}</td>
<td>{"+" if net > 0 else ""}</td>
</tr>
</tbody>
</table>
<p style="font-size:10px;color:#666;margin-top:4px">
* Les numéros de lignes T2/CO-17 sont ceux de l'exercice {yr} — vérifier avec le formulaire officiel de l'année courante.
L'amortissement comptable (ligne 104/130) est estimé selon les durées de vie utiles saisies dans l'outil.
Si vos états financiers utilisent une méthode différente, inscrire le chiffre réel de vos livres.
</p>
<h2>3. Inventaire des immobilisations par catégorie DPA</h2>
<table>
<thead>
<tr><th>Désignation</th><th>Étiquette</th><th class="r">Base amortissable</th><th>Mise en service</th><th>Demi-année</th></tr>
</thead>
<tbody>
{asset_rows()}
</tbody>
</table>
<p class="disclaimer">
Calculs structurellement corrects mais indicatifs — à valider avant tout dépôt fiscal.
Outil : Gestionnaire d'immobilisations fiscales, Chezlepro Inc.
</p>
</body>
</html>"""
@app.get("/api/rapport-fiscal")
def rapport_fiscal_json(year: int = None) -> dict:
if year is None:
year = date.today().year - 1
return _build_rapport_data(year)
@app.get("/api/rapport-fiscal.html", response_class=HTMLResponse)
def rapport_fiscal_html(year: int = None) -> HTMLResponse:
if year is None:
year = date.today().year - 1
return HTMLResponse(content=_render_rapport_html(_build_rapport_data(year)))
@app.get("/")
def index() -> FileResponse:
return FileResponse(STATIC_DIR / "index.html")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")