feat: add analytics
This commit is contained in:
438
src/plesna_gerance/api/routes/analytics.py
Normal file
438
src/plesna_gerance/api/routes/analytics.py
Normal file
@@ -0,0 +1,438 @@
|
||||
"""Analytics routes - Data analysis and reporting endpoints."""
|
||||
|
||||
from datetime import date
|
||||
from collections import defaultdict
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import select, func, distinct, extract
|
||||
|
||||
from ...database import get_session
|
||||
from ...database.models import Depense, Document, Immeuble, Lot, Tag
|
||||
from ..schemas.models import (
|
||||
ImmeubleResponse,
|
||||
LotResponse,
|
||||
DepenseDetail,
|
||||
DepensesSummary,
|
||||
CategorySummary,
|
||||
TagSummary,
|
||||
MonthlySummary,
|
||||
FournisseurSummary,
|
||||
FournisseurResponse,
|
||||
TagResponse,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["analytics"])
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Reference data endpoints (for filters)
|
||||
# ============================================================
|
||||
|
||||
|
||||
@router.get("/immeubles", response_model=list[ImmeubleResponse])
|
||||
async def list_immeubles(
|
||||
session: Session = Depends(get_session),
|
||||
) -> list[ImmeubleResponse]:
|
||||
"""Liste tous les immeubles avec statistiques.
|
||||
|
||||
Retourne la liste des immeubles avec le nombre de lots et de depenses.
|
||||
"""
|
||||
# Get immeubles with counts
|
||||
stmt = (
|
||||
select(
|
||||
Immeuble,
|
||||
func.count(distinct(Lot.id)).label("nb_lots"),
|
||||
func.count(distinct(Depense.id)).label("nb_depenses"),
|
||||
)
|
||||
.outerjoin(Lot, Lot.immeuble_id == Immeuble.id)
|
||||
.outerjoin(Depense, Depense.immeuble_id == Immeuble.id)
|
||||
.group_by(Immeuble.id)
|
||||
.order_by(Immeuble.code)
|
||||
)
|
||||
|
||||
result = session.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
return [
|
||||
ImmeubleResponse(
|
||||
id=row.Immeuble.id,
|
||||
code=row.Immeuble.code,
|
||||
adresse=row.Immeuble.adresse,
|
||||
ville=row.Immeuble.ville,
|
||||
code_postal=row.Immeuble.code_postal,
|
||||
nb_lots=row.nb_lots or 0,
|
||||
nb_depenses=row.nb_depenses or 0,
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
@router.get("/lots", response_model=list[LotResponse])
|
||||
async def list_lots(
|
||||
immeuble_id: int | None = Query(None, description="Filtrer par immeuble"),
|
||||
session: Session = Depends(get_session),
|
||||
) -> list[LotResponse]:
|
||||
"""Liste tous les lots, optionnellement filtres par immeuble.
|
||||
|
||||
- **immeuble_id**: ID de l'immeuble pour filtrer (optionnel)
|
||||
"""
|
||||
stmt = (
|
||||
select(Lot, Immeuble.code.label("immeuble_code"))
|
||||
.join(Immeuble, Lot.immeuble_id == Immeuble.id)
|
||||
.order_by(Immeuble.code, Lot.numero)
|
||||
)
|
||||
|
||||
if immeuble_id is not None:
|
||||
stmt = stmt.where(Lot.immeuble_id == immeuble_id)
|
||||
|
||||
result = session.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
return [
|
||||
LotResponse(
|
||||
id=row.Lot.id,
|
||||
numero=row.Lot.numero,
|
||||
type=row.Lot.type,
|
||||
immeuble_id=row.Lot.immeuble_id,
|
||||
immeuble_code=row.immeuble_code,
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
@router.get("/fournisseurs", response_model=list[FournisseurResponse])
|
||||
async def list_fournisseurs(
|
||||
session: Session = Depends(get_session),
|
||||
) -> list[FournisseurResponse]:
|
||||
"""Liste tous les fournisseurs distincts avec statistiques.
|
||||
|
||||
Retourne la liste des fournisseurs avec le nombre de depenses et total.
|
||||
"""
|
||||
stmt = (
|
||||
select(
|
||||
Depense.fournisseur,
|
||||
func.count(Depense.id).label("nb_depenses"),
|
||||
func.sum(Depense.debit).label("total_debit"),
|
||||
)
|
||||
.where(Depense.fournisseur.isnot(None))
|
||||
.where(Depense.fournisseur != "")
|
||||
.group_by(Depense.fournisseur)
|
||||
.order_by(func.sum(Depense.debit).desc())
|
||||
)
|
||||
|
||||
result = session.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
return [
|
||||
FournisseurResponse(
|
||||
nom=row.fournisseur,
|
||||
nb_depenses=row.nb_depenses or 0,
|
||||
total_debit=row.total_debit or 0.0,
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
@router.get("/tags/stats", response_model=list[TagResponse])
|
||||
async def list_tags_with_stats(
|
||||
session: Session = Depends(get_session),
|
||||
) -> list[TagResponse]:
|
||||
"""Liste tous les tags avec statistiques.
|
||||
|
||||
Retourne la liste des tags avec le nombre de depenses associees.
|
||||
"""
|
||||
stmt = (
|
||||
select(
|
||||
Tag,
|
||||
func.count(Depense.id).label("nb_depenses"),
|
||||
)
|
||||
.outerjoin(Depense, Depense.tag_id == Tag.id)
|
||||
.group_by(Tag.id)
|
||||
.order_by(Tag.nom)
|
||||
)
|
||||
|
||||
result = session.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
return [
|
||||
TagResponse(
|
||||
id=row.Tag.id,
|
||||
nom=row.Tag.nom,
|
||||
nb_depenses=row.nb_depenses or 0,
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
# ============================================================
|
||||
# Analytics endpoints
|
||||
# ============================================================
|
||||
|
||||
|
||||
@router.get("/analytics/depenses", response_model=list[DepenseDetail])
|
||||
async def get_depenses(
|
||||
immeuble_id: int | None = Query(None, description="Filtrer par immeuble"),
|
||||
lot_id: int | None = Query(None, description="Filtrer par lot"),
|
||||
tag_id: int | None = Query(None, description="Filtrer par tag"),
|
||||
categorie: str | None = Query(None, description="Filtrer par categorie"),
|
||||
fournisseur: str | None = Query(None, description="Filtrer par fournisseur"),
|
||||
date_debut: date | None = Query(None, description="Date de debut (YYYY-MM-DD)"),
|
||||
date_fin: date | None = Query(None, description="Date de fin (YYYY-MM-DD)"),
|
||||
limit: int = Query(500, description="Nombre maximum de resultats"),
|
||||
offset: int = Query(0, description="Decalage pour la pagination"),
|
||||
session: Session = Depends(get_session),
|
||||
) -> list[DepenseDetail]:
|
||||
"""Retourne la liste des depenses filtrees avec details.
|
||||
|
||||
Filtres disponibles:
|
||||
- **immeuble_id**: ID de l'immeuble
|
||||
- **lot_id**: ID du lot
|
||||
- **tag_id**: ID du tag
|
||||
- **categorie**: Categorie de depense
|
||||
- **fournisseur**: Nom du fournisseur
|
||||
- **date_debut**: Date de debut (incluse)
|
||||
- **date_fin**: Date de fin (incluse)
|
||||
- **limit**: Nombre max de resultats (defaut: 500)
|
||||
- **offset**: Decalage pour pagination
|
||||
"""
|
||||
stmt = (
|
||||
select(
|
||||
Depense,
|
||||
Document.date.label("document_date"),
|
||||
Document.reference.label("document_reference"),
|
||||
Immeuble.code.label("immeuble_code"),
|
||||
Immeuble.adresse.label("immeuble_adresse"),
|
||||
Lot.numero.label("lot_numero"),
|
||||
Tag.nom.label("tag_nom"),
|
||||
)
|
||||
.join(Document, Depense.document_id == Document.id)
|
||||
.join(Immeuble, Depense.immeuble_id == Immeuble.id)
|
||||
.outerjoin(Lot, Depense.lot_id == Lot.id)
|
||||
.outerjoin(Tag, Depense.tag_id == Tag.id)
|
||||
.order_by(Document.date.desc(), Depense.id.desc())
|
||||
)
|
||||
|
||||
# Apply filters
|
||||
if immeuble_id is not None:
|
||||
stmt = stmt.where(Depense.immeuble_id == immeuble_id)
|
||||
if lot_id is not None:
|
||||
stmt = stmt.where(Depense.lot_id == lot_id)
|
||||
if tag_id is not None:
|
||||
stmt = stmt.where(Depense.tag_id == tag_id)
|
||||
if categorie is not None:
|
||||
stmt = stmt.where(Depense.categorie == categorie)
|
||||
if fournisseur is not None:
|
||||
stmt = stmt.where(Depense.fournisseur.ilike(f"%{fournisseur}%"))
|
||||
if date_debut is not None:
|
||||
stmt = stmt.where(Document.date >= date_debut)
|
||||
if date_fin is not None:
|
||||
stmt = stmt.where(Document.date <= date_fin)
|
||||
|
||||
stmt = stmt.limit(limit).offset(offset)
|
||||
|
||||
result = session.execute(stmt)
|
||||
rows = result.all()
|
||||
|
||||
return [
|
||||
DepenseDetail(
|
||||
id=row.Depense.id,
|
||||
document_id=row.Depense.document_id,
|
||||
document_date=row.document_date,
|
||||
document_reference=row.document_reference,
|
||||
immeuble_id=row.Depense.immeuble_id,
|
||||
immeuble_code=row.immeuble_code,
|
||||
immeuble_adresse=row.immeuble_adresse,
|
||||
lot_id=row.Depense.lot_id,
|
||||
lot_numero=row.lot_numero,
|
||||
tag_id=row.Depense.tag_id,
|
||||
tag_nom=row.tag_nom,
|
||||
categorie=row.Depense.categorie,
|
||||
sous_categorie=row.Depense.sous_categorie,
|
||||
fournisseur=row.Depense.fournisseur,
|
||||
description=row.Depense.description,
|
||||
debit=row.Depense.debit or 0.0,
|
||||
credit=row.Depense.credit or 0.0,
|
||||
tva=row.Depense.tva or 0.0,
|
||||
locatif=row.Depense.locatif or 0.0,
|
||||
deductible=row.Depense.deductible or 0.0,
|
||||
)
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
@router.get("/analytics/depenses/summary", response_model=DepensesSummary)
|
||||
async def get_depenses_summary(
|
||||
immeuble_id: int | None = Query(None, description="Filtrer par immeuble"),
|
||||
lot_id: int | None = Query(None, description="Filtrer par lot"),
|
||||
tag_id: int | None = Query(None, description="Filtrer par tag"),
|
||||
categorie: str | None = Query(None, description="Filtrer par categorie"),
|
||||
fournisseur: str | None = Query(None, description="Filtrer par fournisseur"),
|
||||
date_debut: date | None = Query(None, description="Date de debut (YYYY-MM-DD)"),
|
||||
date_fin: date | None = Query(None, description="Date de fin (YYYY-MM-DD)"),
|
||||
session: Session = Depends(get_session),
|
||||
) -> DepensesSummary:
|
||||
"""Retourne un resume agrege des depenses pour les graphiques.
|
||||
|
||||
Memes filtres que /analytics/depenses.
|
||||
Retourne:
|
||||
- Totaux globaux
|
||||
- Repartition par categorie
|
||||
- Repartition par tag
|
||||
- Evolution mensuelle
|
||||
- Top fournisseurs
|
||||
"""
|
||||
# Base query with filters
|
||||
base_stmt = select(Depense).join(Document, Depense.document_id == Document.id)
|
||||
|
||||
if immeuble_id is not None:
|
||||
base_stmt = base_stmt.where(Depense.immeuble_id == immeuble_id)
|
||||
if lot_id is not None:
|
||||
base_stmt = base_stmt.where(Depense.lot_id == lot_id)
|
||||
if tag_id is not None:
|
||||
base_stmt = base_stmt.where(Depense.tag_id == tag_id)
|
||||
if categorie is not None:
|
||||
base_stmt = base_stmt.where(Depense.categorie == categorie)
|
||||
if fournisseur is not None:
|
||||
base_stmt = base_stmt.where(Depense.fournisseur.ilike(f"%{fournisseur}%"))
|
||||
if date_debut is not None:
|
||||
base_stmt = base_stmt.where(Document.date >= date_debut)
|
||||
if date_fin is not None:
|
||||
base_stmt = base_stmt.where(Document.date <= date_fin)
|
||||
|
||||
# Get all matching depenses
|
||||
result = session.execute(base_stmt)
|
||||
depenses = result.scalars().all()
|
||||
|
||||
# Calculate totals
|
||||
total_count = len(depenses)
|
||||
total_debit = sum(d.debit or 0 for d in depenses)
|
||||
total_credit = sum(d.credit or 0 for d in depenses)
|
||||
total_tva = sum(d.tva or 0 for d in depenses)
|
||||
total_locatif = sum(d.locatif or 0 for d in depenses)
|
||||
total_deductible = sum(d.deductible or 0 for d in depenses)
|
||||
|
||||
# Get document dates for monthly aggregation
|
||||
doc_dates = {}
|
||||
for d in depenses:
|
||||
if d.document_id not in doc_dates:
|
||||
doc = session.get(Document, d.document_id)
|
||||
doc_dates[d.document_id] = doc.date if doc else None
|
||||
|
||||
# Aggregate by category
|
||||
by_category_dict = defaultdict(
|
||||
lambda: {
|
||||
"count": 0,
|
||||
"debit": 0.0,
|
||||
"credit": 0.0,
|
||||
"tva": 0.0,
|
||||
"locatif": 0.0,
|
||||
"deductible": 0.0,
|
||||
}
|
||||
)
|
||||
for d in depenses:
|
||||
cat = d.categorie or "NON_CATEGORISE"
|
||||
by_category_dict[cat]["count"] += 1
|
||||
by_category_dict[cat]["debit"] += d.debit or 0
|
||||
by_category_dict[cat]["credit"] += d.credit or 0
|
||||
by_category_dict[cat]["tva"] += d.tva or 0
|
||||
by_category_dict[cat]["locatif"] += d.locatif or 0
|
||||
by_category_dict[cat]["deductible"] += d.deductible or 0
|
||||
|
||||
by_category = [
|
||||
CategorySummary(
|
||||
categorie=cat,
|
||||
count=data["count"],
|
||||
total_debit=data["debit"],
|
||||
total_credit=data["credit"],
|
||||
total_tva=data["tva"],
|
||||
total_locatif=data["locatif"],
|
||||
total_deductible=data["deductible"],
|
||||
)
|
||||
for cat, data in sorted(by_category_dict.items(), key=lambda x: -x[1]["debit"])
|
||||
]
|
||||
|
||||
# Aggregate by tag
|
||||
by_tag_dict = defaultdict(lambda: {"tag_nom": None, "count": 0, "debit": 0.0})
|
||||
for d in depenses:
|
||||
tag_key = d.tag_id or 0 # 0 for untagged
|
||||
by_tag_dict[tag_key]["count"] += 1
|
||||
by_tag_dict[tag_key]["debit"] += d.debit or 0
|
||||
if d.tag_id and d.tag:
|
||||
by_tag_dict[tag_key]["tag_nom"] = d.tag.nom
|
||||
|
||||
by_tag = [
|
||||
TagSummary(
|
||||
tag_id=tag_id if tag_id != 0 else None,
|
||||
tag_nom=data["tag_nom"] if tag_id != 0 else "Non taggue",
|
||||
count=data["count"],
|
||||
total_debit=data["debit"],
|
||||
)
|
||||
for tag_id, data in sorted(by_tag_dict.items(), key=lambda x: -x[1]["debit"])
|
||||
]
|
||||
|
||||
# Aggregate by month
|
||||
by_month_dict = defaultdict(lambda: {"count": 0, "debit": 0.0, "credit": 0.0})
|
||||
for d in depenses:
|
||||
doc_date = doc_dates.get(d.document_id)
|
||||
if doc_date:
|
||||
month_key = (doc_date.year, doc_date.month)
|
||||
by_month_dict[month_key]["count"] += 1
|
||||
by_month_dict[month_key]["debit"] += d.debit or 0
|
||||
by_month_dict[month_key]["credit"] += d.credit or 0
|
||||
|
||||
by_month = [
|
||||
MonthlySummary(
|
||||
year=year,
|
||||
month=month,
|
||||
count=data["count"],
|
||||
total_debit=data["debit"],
|
||||
total_credit=data["credit"],
|
||||
)
|
||||
for (year, month), data in sorted(by_month_dict.items())
|
||||
]
|
||||
|
||||
# Aggregate by fournisseur (top 20)
|
||||
by_fournisseur_dict = defaultdict(lambda: {"count": 0, "debit": 0.0})
|
||||
for d in depenses:
|
||||
fournisseur_key = d.fournisseur or "Non specifie"
|
||||
by_fournisseur_dict[fournisseur_key]["count"] += 1
|
||||
by_fournisseur_dict[fournisseur_key]["debit"] += d.debit or 0
|
||||
|
||||
by_fournisseur = [
|
||||
FournisseurSummary(
|
||||
fournisseur=fournisseur if fournisseur != "Non specifie" else None,
|
||||
count=data["count"],
|
||||
total_debit=data["debit"],
|
||||
)
|
||||
for fournisseur, data in sorted(
|
||||
by_fournisseur_dict.items(), key=lambda x: -x[1]["debit"]
|
||||
)[:20]
|
||||
]
|
||||
|
||||
return DepensesSummary(
|
||||
total_count=total_count,
|
||||
total_debit=total_debit,
|
||||
total_credit=total_credit,
|
||||
total_tva=total_tva,
|
||||
total_locatif=total_locatif,
|
||||
total_deductible=total_deductible,
|
||||
by_category=by_category,
|
||||
by_tag=by_tag,
|
||||
by_month=by_month,
|
||||
by_fournisseur=by_fournisseur,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/analytics/categories")
|
||||
async def list_categories(
|
||||
session: Session = Depends(get_session),
|
||||
) -> list[str]:
|
||||
"""Liste toutes les categories de depenses distinctes."""
|
||||
stmt = (
|
||||
select(distinct(Depense.categorie))
|
||||
.where(Depense.categorie.isnot(None))
|
||||
.order_by(Depense.categorie)
|
||||
)
|
||||
result = session.execute(stmt)
|
||||
return [row[0] for row in result.all() if row[0]]
|
||||
Reference in New Issue
Block a user