# RenoValid Back Office — lignes de code à appliquer dans le MCP

Objectif : transformer le bundle actuel en vrais fichiers applicatifs dès que le MCP permet l’écriture `.py` / `.csv`, ou copier ces blocs manuellement dans les bons fichiers.

---

# 1. Créer app/db.py

```python
import os
from collections.abc import Generator

from sqlmodel import SQLModel, Session, create_engine

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./renovalid_backoffice.db")
connect_args = {"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
engine = create_engine(DATABASE_URL, echo=False, connect_args=connect_args)


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)


def get_session() -> Generator[Session, None, None]:
    with Session(engine) as session:
        yield session
```

---

# 2. Créer app/models.py

```python
from datetime import datetime
from enum import Enum
from typing import Optional

from sqlmodel import Field, SQLModel


class ReferentialType(str, Enum):
    TARIFF = "tariff"
    SCALE = "scale"
    ELIGIBILITY_RULE = "eligibility_rule"
    MATERIAL = "material"
    WORK_TYPE = "work_type"


class ProjectStatus(str, Enum):
    DRAFT = "draft"
    SIMULATION = "simulation"
    PROJECT_SAVED = "project_saved"
    ARCHIVED = "archived"


class ReferentialItem(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    type: ReferentialType = Field(index=True)
    code: str = Field(index=True)
    label: str
    value_numeric: Optional[float] = None
    value_text: Optional[str] = None
    unit: Optional[str] = None
    active: bool = Field(default=True, index=True)
    version: str = Field(default="v1", index=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)


class PostalCodeCommune(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    postal_code: str = Field(index=True)
    commune: str = Field(index=True)
    insee_code: Optional[str] = Field(default=None, index=True)
    department_code: Optional[str] = Field(default=None, index=True)
    region: Optional[str] = None
    active: bool = Field(default=True, index=True)
    updated_at: datetime = Field(default_factory=datetime.utcnow)


class SavedProject(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    public_id: str = Field(index=True, unique=True)
    status: ProjectStatus = Field(default=ProjectStatus.DRAFT, index=True)
    applicant_type: Optional[str] = Field(default=None, index=True)
    postal_code: Optional[str] = Field(default=None, index=True)
    commune: Optional[str] = Field(default=None, index=True)
    work_type: Optional[str] = Field(default=None, index=True)
    estimated_amount_ht: Optional[float] = None
    estimated_amount_ttc: Optional[float] = None
    aid_estimate: Optional[float] = None
    source: Optional[str] = Field(default="field", index=True)
    payload_json: str = Field(default="{}")
    created_at: datetime = Field(default_factory=datetime.utcnow, index=True)
    updated_at: datetime = Field(default_factory=datetime.utcnow, index=True)


class FieldInputEvent(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    project_public_id: Optional[str] = Field(default=None, index=True)
    field_name: str = Field(index=True)
    event_type: str = Field(index=True)
    value_length: Optional[int] = None
    error_code: Optional[str] = Field(default=None, index=True)
    step: Optional[str] = Field(default=None, index=True)
    source: Optional[str] = Field(default="field", index=True)
    created_at: datetime = Field(default_factory=datetime.utcnow, index=True)


class AuditLog(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    actor: str = Field(default="system", index=True)
    action: str = Field(index=True)
    entity_type: str = Field(index=True)
    entity_id: Optional[str] = Field(default=None, index=True)
    before_json: Optional[str] = None
    after_json: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.utcnow, index=True)
```

---

# 3. Créer app/schemas.py

```python
from datetime import datetime
from typing import Optional

from pydantic import BaseModel

from app.models import ProjectStatus, ReferentialType


class ReferentialItemCreate(BaseModel):
    type: ReferentialType
    code: str
    label: str
    value_numeric: Optional[float] = None
    value_text: Optional[str] = None
    unit: Optional[str] = None
    active: bool = True
    version: str = "v1"


class ReferentialItemRead(ReferentialItemCreate):
    id: int
    created_at: datetime
    updated_at: datetime


class PostalCodeCreate(BaseModel):
    postal_code: str
    commune: str
    insee_code: Optional[str] = None
    department_code: Optional[str] = None
    region: Optional[str] = None
    active: bool = True


class PostalCodeRead(PostalCodeCreate):
    id: int
    updated_at: datetime


class SavedProjectCreate(BaseModel):
    public_id: str
    status: ProjectStatus = ProjectStatus.DRAFT
    applicant_type: Optional[str] = None
    postal_code: Optional[str] = None
    commune: Optional[str] = None
    work_type: Optional[str] = None
    estimated_amount_ht: Optional[float] = None
    estimated_amount_ttc: Optional[float] = None
    aid_estimate: Optional[float] = None
    source: Optional[str] = "field"
    payload_json: str = "{}"


class SavedProjectUpdate(BaseModel):
    status: Optional[ProjectStatus] = None
    applicant_type: Optional[str] = None
    postal_code: Optional[str] = None
    commune: Optional[str] = None
    work_type: Optional[str] = None
    estimated_amount_ht: Optional[float] = None
    estimated_amount_ttc: Optional[float] = None
    aid_estimate: Optional[float] = None
    source: Optional[str] = None
    payload_json: Optional[str] = None


class SavedProjectRead(SavedProjectCreate):
    id: int
    created_at: datetime
    updated_at: datetime


class FieldInputEventCreate(BaseModel):
    project_public_id: Optional[str] = None
    field_name: str
    event_type: str
    value_length: Optional[int] = None
    error_code: Optional[str] = None
    step: Optional[str] = None
    source: Optional[str] = "field"


class StatsSummary(BaseModel):
    total_projects: int
    drafts: int
    simulations: int
    saved_projects: int
    archived: int
    total_input_events: int
    top_error_field: Optional[str] = None
```

---

# 4. Créer app/services/import_export.py

```python
from io import BytesIO

import pandas as pd

SUPPORTED_FORMATS = {"csv", "xlsx", "xls"}


def read_tabular_file(file_bytes: bytes, filename: str) -> pd.DataFrame:
    extension = filename.lower().split(".")[-1]
    if extension not in SUPPORTED_FORMATS:
        raise ValueError("Format non supporté. Utiliser csv, xls ou xlsx.")
    if extension == "csv":
        return pd.read_csv(BytesIO(file_bytes))
    return pd.read_excel(BytesIO(file_bytes))


def dataframe_to_records(df: pd.DataFrame) -> list[dict]:
    df = df.where(pd.notnull(df), None)
    return df.to_dict(orient="records")


def records_to_csv_bytes(records: list[dict]) -> bytes:
    df = pd.DataFrame(records)
    return df.to_csv(index=False).encode("utf-8")


def records_to_xlsx_bytes(records: list[dict]) -> bytes:
    output = BytesIO()
    df = pd.DataFrame(records)
    with pd.ExcelWriter(output, engine="openpyxl") as writer:
        df.to_excel(writer, index=False, sheet_name="export")
    return output.getvalue()


def validate_required_columns(df: pd.DataFrame, required_columns: set[str]) -> None:
    missing = required_columns - set(df.columns)
    if missing:
        raise ValueError(f"Colonnes manquantes: {', '.join(sorted(missing))}")
```

---

# 5. Créer app/services/statistics.py

```python
from sqlmodel import Session, func, select

from app.models import FieldInputEvent, ProjectStatus, SavedProject
from app.schemas import StatsSummary


def compute_stats_summary(session: Session) -> StatsSummary:
    total_projects = session.exec(select(func.count(SavedProject.id))).one()
    drafts = session.exec(select(func.count(SavedProject.id)).where(SavedProject.status == ProjectStatus.DRAFT)).one()
    simulations = session.exec(select(func.count(SavedProject.id)).where(SavedProject.status == ProjectStatus.SIMULATION)).one()
    saved_projects = session.exec(select(func.count(SavedProject.id)).where(SavedProject.status == ProjectStatus.PROJECT_SAVED)).one()
    archived = session.exec(select(func.count(SavedProject.id)).where(SavedProject.status == ProjectStatus.ARCHIVED)).one()
    total_input_events = session.exec(select(func.count(FieldInputEvent.id))).one()
    top_error_row = session.exec(
        select(FieldInputEvent.field_name, func.count(FieldInputEvent.id))
        .where(FieldInputEvent.event_type == "error")
        .group_by(FieldInputEvent.field_name)
        .order_by(func.count(FieldInputEvent.id).desc())
    ).first()
    return StatsSummary(
        total_projects=total_projects or 0,
        drafts=drafts or 0,
        simulations=simulations or 0,
        saved_projects=saved_projects or 0,
        archived=archived or 0,
        total_input_events=total_input_events or 0,
        top_error_field=top_error_row[0] if top_error_row else None,
    )
```

---

# 6. Créer app/routers/health.py

```python
from fastapi import APIRouter

router = APIRouter(prefix="/health", tags=["health"])


@router.get("")
def health_check():
    return {"status": "ok", "service": "renovalid-backoffice"}
```

---

# 7. Créer app/routers/referentials.py

```python
from datetime import datetime

from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile
from sqlmodel import Session, select

from app.db import get_session
from app.models import ReferentialItem
from app.schemas import ReferentialItemCreate, ReferentialItemRead
from app.services.import_export import dataframe_to_records, read_tabular_file, records_to_csv_bytes, records_to_xlsx_bytes, validate_required_columns

router = APIRouter(prefix="/referentials", tags=["referentials"])


@router.post("", response_model=ReferentialItemRead)
def create_referential_item(payload: ReferentialItemCreate, session: Session = Depends(get_session)):
    item = ReferentialItem(**payload.model_dump())
    session.add(item)
    session.commit()
    session.refresh(item)
    return item


@router.get("", response_model=list[ReferentialItemRead])
def list_referential_items(type: str | None = None, active: bool | None = None, session: Session = Depends(get_session)):
    query = select(ReferentialItem)
    if type:
        query = query.where(ReferentialItem.type == type)
    if active is not None:
        query = query.where(ReferentialItem.active == active)
    return session.exec(query.order_by(ReferentialItem.type, ReferentialItem.code)).all()


@router.post("/import")
async def import_referentials(file: UploadFile = File(...), session: Session = Depends(get_session)):
    try:
        content = await file.read()
        df = read_tabular_file(content, file.filename or "")
        validate_required_columns(df, {"type", "code", "label"})
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    imported = 0
    for record in dataframe_to_records(df):
        item = ReferentialItem(
            type=record["type"],
            code=str(record["code"]),
            label=str(record["label"]),
            value_numeric=record.get("value_numeric"),
            value_text=record.get("value_text"),
            unit=record.get("unit"),
            active=bool(record.get("active", True)),
            version=str(record.get("version", "v1")),
            updated_at=datetime.utcnow(),
        )
        session.add(item)
        imported += 1
    session.commit()
    return {"imported": imported}


@router.get("/export")
def export_referentials(format: str = "csv", session: Session = Depends(get_session)):
    items = session.exec(select(ReferentialItem)).all()
    records = [
        {
            "id": item.id,
            "type": item.type,
            "code": item.code,
            "label": item.label,
            "value_numeric": item.value_numeric,
            "value_text": item.value_text,
            "unit": item.unit,
            "active": item.active,
            "version": item.version,
        }
        for item in items
    ]
    if format == "csv":
        return Response(content=records_to_csv_bytes(records), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=referentials.csv"})
    if format == "xlsx":
        return Response(content=records_to_xlsx_bytes(records), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=referentials.xlsx"})
    raise HTTPException(status_code=400, detail="Format attendu: csv ou xlsx")
```

---

# 8. Créer app/routers/postal_codes.py

```python
from datetime import datetime

from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile
from sqlmodel import Session, select

from app.db import get_session
from app.models import PostalCodeCommune
from app.schemas import PostalCodeCreate, PostalCodeRead
from app.services.import_export import dataframe_to_records, read_tabular_file, records_to_csv_bytes, records_to_xlsx_bytes, validate_required_columns

router = APIRouter(prefix="/postal-codes", tags=["postal-codes"])


@router.post("", response_model=PostalCodeRead)
def create_postal_code(payload: PostalCodeCreate, session: Session = Depends(get_session)):
    row = PostalCodeCommune(**payload.model_dump(), updated_at=datetime.utcnow())
    session.add(row)
    session.commit()
    session.refresh(row)
    return row


@router.get("", response_model=list[PostalCodeRead])
def search_postal_codes(postal_code: str | None = None, commune: str | None = None, session: Session = Depends(get_session)):
    query = select(PostalCodeCommune)
    if postal_code:
        query = query.where(PostalCodeCommune.postal_code == postal_code)
    if commune:
        query = query.where(PostalCodeCommune.commune.ilike(f"%{commune}%"))
    return session.exec(query.order_by(PostalCodeCommune.postal_code, PostalCodeCommune.commune)).all()


@router.post("/import")
async def import_postal_codes(file: UploadFile = File(...), session: Session = Depends(get_session)):
    try:
        content = await file.read()
        df = read_tabular_file(content, file.filename or "")
        validate_required_columns(df, {"postal_code", "commune"})
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    imported = 0
    for record in dataframe_to_records(df):
        row = PostalCodeCommune(
            postal_code=str(record["postal_code"]).zfill(5),
            commune=str(record["commune"]),
            insee_code=record.get("insee_code"),
            department_code=record.get("department_code"),
            region=record.get("region"),
            active=bool(record.get("active", True)),
            updated_at=datetime.utcnow(),
        )
        session.add(row)
        imported += 1
    session.commit()
    return {"imported": imported}


@router.get("/export")
def export_postal_codes(format: str = "csv", session: Session = Depends(get_session)):
    rows = session.exec(select(PostalCodeCommune)).all()
    records = [
        {
            "id": row.id,
            "postal_code": row.postal_code,
            "commune": row.commune,
            "insee_code": row.insee_code,
            "department_code": row.department_code,
            "region": row.region,
            "active": row.active,
        }
        for row in rows
    ]
    if format == "csv":
        return Response(content=records_to_csv_bytes(records), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=postal_codes.csv"})
    if format == "xlsx":
        return Response(content=records_to_xlsx_bytes(records), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=postal_codes.xlsx"})
    raise HTTPException(status_code=400, detail="Format attendu: csv ou xlsx")
```

---

# 9. Créer app/routers/simulations.py

```python
from datetime import datetime

from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select

from app.db import get_session
from app.models import SavedProject
from app.schemas import SavedProjectCreate, SavedProjectRead, SavedProjectUpdate

router = APIRouter(prefix="/projects", tags=["projects"])


@router.post("", response_model=SavedProjectRead)
def create_saved_project(payload: SavedProjectCreate, session: Session = Depends(get_session)):
    existing = session.exec(select(SavedProject).where(SavedProject.public_id == payload.public_id)).first()
    if existing:
        raise HTTPException(status_code=409, detail="public_id déjà existant")
    project = SavedProject(**payload.model_dump())
    session.add(project)
    session.commit()
    session.refresh(project)
    return project


@router.get("", response_model=list[SavedProjectRead])
def list_saved_projects(status: str | None = None, postal_code: str | None = None, work_type: str | None = None, session: Session = Depends(get_session)):
    query = select(SavedProject)
    if status:
        query = query.where(SavedProject.status == status)
    if postal_code:
        query = query.where(SavedProject.postal_code == postal_code)
    if work_type:
        query = query.where(SavedProject.work_type == work_type)
    return session.exec(query.order_by(SavedProject.updated_at.desc())).all()


@router.get("/{public_id}", response_model=SavedProjectRead)
def get_saved_project(public_id: str, session: Session = Depends(get_session)):
    project = session.exec(select(SavedProject).where(SavedProject.public_id == public_id)).first()
    if not project:
        raise HTTPException(status_code=404, detail="Projet introuvable")
    return project


@router.patch("/{public_id}", response_model=SavedProjectRead)
def update_saved_project(public_id: str, payload: SavedProjectUpdate, session: Session = Depends(get_session)):
    project = session.exec(select(SavedProject).where(SavedProject.public_id == public_id)).first()
    if not project:
        raise HTTPException(status_code=404, detail="Projet introuvable")
    for key, value in payload.model_dump(exclude_unset=True).items():
        setattr(project, key, value)
    project.updated_at = datetime.utcnow()
    session.add(project)
    session.commit()
    session.refresh(project)
    return project
```

---

# 10. Créer app/routers/stats.py

```python
from fastapi import APIRouter, Depends
from sqlmodel import Session

from app.db import get_session
from app.models import FieldInputEvent
from app.schemas import FieldInputEventCreate, StatsSummary
from app.services.statistics import compute_stats_summary

router = APIRouter(prefix="/stats", tags=["stats"])


@router.post("/field-events")
def create_field_input_event(payload: FieldInputEventCreate, session: Session = Depends(get_session)):
    event = FieldInputEvent(**payload.model_dump())
    session.add(event)
    session.commit()
    session.refresh(event)
    return {"id": event.id}


@router.get("/summary", response_model=StatsSummary)
def get_stats_summary(session: Session = Depends(get_session)):
    return compute_stats_summary(session)
```

---

# 11. Créer app/main.py

```python
from fastapi import FastAPI

from app.db import create_db_and_tables
from app.routers import health, postal_codes, referentials, simulations, stats

app = FastAPI(
    title="RenoValid Back Office",
    version="0.1.0",
    description="Back-office MVP pour référentiels, barèmes, codes postaux, projets et statistiques terrain.",
)


@app.on_event("startup")
def on_startup() -> None:
    create_db_and_tables()


app.include_router(health.router)
app.include_router(referentials.router)
app.include_router(postal_codes.router)
app.include_router(simulations.router)
app.include_router(stats.router)
```

---

# 12. Créer app/seed.py

```python
from sqlmodel import Session

from app.db import create_db_and_tables, engine
from app.models import PostalCodeCommune, ReferentialItem, ReferentialType


def seed() -> None:
    create_db_and_tables()
    with Session(engine) as session:
        session.add(ReferentialItem(type=ReferentialType.TARIFF, code="ISOL_COMBLES_M2", label="Isolation des combles - prix indicatif au m²", value_numeric=45.0, unit="EUR_HT_M2", version="mvp"))
        session.add(ReferentialItem(type=ReferentialType.SCALE, code="AIDE_BASE_PERCENT", label="Taux d’aide de base", value_numeric=30.0, unit="PERCENT", version="mvp"))
        session.add(PostalCodeCommune(postal_code="75001", commune="Paris 1er Arrondissement", insee_code="75101", department_code="75", region="Île-de-France"))
        session.commit()


if __name__ == "__main__":
    seed()
```

---

# 13. Créer tests/test_health.py

```python
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)


def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"
```

---

# 14. Créer tests/test_projects.py

```python
from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)


def test_create_project():
    response = client.post(
        "/projects",
        json={
            "public_id": "test-project-001",
            "status": "draft",
            "postal_code": "75001",
            "commune": "Paris 1er Arrondissement",
            "work_type": "isolation",
        },
    )
    assert response.status_code in (200, 409)
```

---

# 15. Créer data/sample_postal_codes.csv

```csv
postal_code,commune,insee_code,department_code,region,active
75001,Paris 1er Arrondissement,75101,75,Île-de-France,true
69001,Lyon 1er Arrondissement,69381,69,Auvergne-Rhône-Alpes,true
13001,Marseille 1er Arrondissement,13201,13,Provence-Alpes-Côte d'Azur,true
```

---

# 16. Créer data/sample_tariffs.csv

```csv
type,code,label,value_numeric,value_text,unit,active,version
tariff,ISOL_COMBLES_M2,Isolation des combles - prix indicatif au m²,45,,EUR_HT_M2,true,mvp
scale,AIDE_BASE_PERCENT,Taux d’aide de base,30,,PERCENT,true,mvp
work_type,CHAUFFAGE,Chauffage,,,,true,mvp
```

---

# 17. Créer les fichiers vides d’initialisation Python

```txt
app/__init__.py
app/routers/__init__.py
app/services/__init__.py
```

Ces trois fichiers peuvent être vides.

---

# 18. Commandes de test après création des fichiers

```bash
cd renovalid-backoffice
pip install -r requirements.txt
python -m app.seed
pytest
uvicorn app.main:app --reload
```