999 lines
36 KiB
Python
999 lines
36 KiB
Python
from pathlib import Path
|
|
from datetime import datetime
|
|
import json
|
|
import mimetypes
|
|
from tempfile import NamedTemporaryFile
|
|
|
|
from fastapi import Body, Depends, FastAPI, File, Form, HTTPException, UploadFile
|
|
from fastapi import Response
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy import or_, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.database import Base, engine, get_db
|
|
from app.models import Consultation, Document, Patient, Photo
|
|
from app.schemas import ConsultationCreate, ConsultationOut, ConsultationUpdate, DocumentCreate, DocumentOut, DocumentUpdate, ImmichAssetOut, ImmichPersonOut, PatientCreate, PatientOut, PatientUpdate, PhotoOut, PhotoUpdate
|
|
from app.services.immich import ImmichClient
|
|
from app.services.face_transform import render_face_guides, render_face_transform
|
|
|
|
app = FastAPI(title="Aesthetic Photo Planner API", version="0.3.7")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # À durcir plus tard si accès public.
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
def _ensure_sqlite_patient_columns() -> None:
|
|
"""Mini migrations SQLite locales.
|
|
|
|
Pour une vraie prod, on passera à Alembic/PostgreSQL. Ici, on évite de casser
|
|
ta base locale existante juste parce qu'on ajoute trois champs, cette manie humaine.
|
|
"""
|
|
if not settings.database_url.startswith("sqlite"):
|
|
return
|
|
|
|
optional_columns = {
|
|
"phone": "TEXT",
|
|
"email": "TEXT",
|
|
"procedure": "TEXT",
|
|
"status": "VARCHAR(40) DEFAULT 'actif'",
|
|
"consent_signed": "VARCHAR(20) DEFAULT 'non'",
|
|
}
|
|
|
|
with engine.begin() as conn:
|
|
existing = {row[1] for row in conn.execute(text("PRAGMA table_info(patients)")).fetchall()}
|
|
for column_name, column_type in optional_columns.items():
|
|
if column_name not in existing:
|
|
conn.execute(text(f"ALTER TABLE patients ADD COLUMN {column_name} {column_type}"))
|
|
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS photo_face_landmarks (
|
|
photo_id INTEGER PRIMARY KEY,
|
|
manual_landmarks TEXT NOT NULL,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""))
|
|
|
|
|
|
@app.on_event("startup")
|
|
def on_startup() -> None:
|
|
Base.metadata.create_all(bind=engine)
|
|
_ensure_sqlite_patient_columns()
|
|
|
|
|
|
@app.get("/")
|
|
def root() -> dict:
|
|
return {
|
|
"app": "Aesthetic Photo Planner",
|
|
"version": "0.3.7",
|
|
"status": "running",
|
|
"docs": "/docs",
|
|
"health": "/health",
|
|
"patients": "/patients",
|
|
"documents": "/patients/{patient_id}/documents",
|
|
}
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict:
|
|
return {
|
|
"status": "ok",
|
|
"database": settings.database_url.split(":", 1)[0],
|
|
"immich_configured": bool(settings.immich_api_key),
|
|
}
|
|
|
|
|
|
def _clean_text(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
value = value.strip()
|
|
return value or None
|
|
|
|
|
|
|
|
|
|
def _validate_photo_kind(kind: str | None, default: str = "original") -> str:
|
|
safe_kind = (kind or default).strip().lower() or default
|
|
allowed = {"original", "annotated", "before", "after", "planning", "planification", "face", "simulation"}
|
|
if safe_kind not in allowed:
|
|
raise HTTPException(status_code=400, detail="kind invalide")
|
|
return safe_kind
|
|
|
|
|
|
|
|
|
|
|
|
def _sanitize_manual_landmarks(raw: object) -> dict[str, list[float]]:
|
|
if not isinstance(raw, dict):
|
|
raise HTTPException(status_code=400, detail="manual_landmarks invalide")
|
|
|
|
allowed_keys = {
|
|
"left_eye", "right_eye", "nose", "mouth", "chin", "face_top", "face_left", "face_right"
|
|
}
|
|
cleaned: dict[str, list[float]] = {}
|
|
for key, value in raw.items():
|
|
if key not in allowed_keys:
|
|
continue
|
|
if not isinstance(value, list | tuple) or len(value) < 2:
|
|
raise HTTPException(status_code=400, detail=f"repère invalide: {key}")
|
|
try:
|
|
x = float(value[0])
|
|
y = float(value[1])
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(status_code=400, detail=f"repère invalide: {key}")
|
|
cleaned[key] = [max(0.0, min(1.0, x)), max(0.0, min(1.0, y))]
|
|
|
|
required = {"left_eye", "right_eye", "nose", "mouth", "chin", "face_top", "face_left", "face_right"}
|
|
missing = sorted(required - set(cleaned))
|
|
if missing:
|
|
raise HTTPException(status_code=400, detail=f"repères manquants: {', '.join(missing)}")
|
|
return cleaned
|
|
|
|
|
|
def _get_saved_manual_landmarks(db: Session, photo_id: int) -> dict[str, list[float]] | None:
|
|
if not settings.database_url.startswith("sqlite"):
|
|
return None
|
|
row = db.execute(
|
|
text("SELECT manual_landmarks FROM photo_face_landmarks WHERE photo_id = :photo_id"),
|
|
{"photo_id": photo_id},
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
data = json.loads(row[0])
|
|
return _sanitize_manual_landmarks(data)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _payload_with_saved_landmarks(payload: dict, db: Session, photo_id: int) -> dict:
|
|
if not isinstance(payload, dict):
|
|
payload = {}
|
|
output = dict(payload)
|
|
if not output.get("manual_landmarks"):
|
|
saved = _get_saved_manual_landmarks(db, photo_id)
|
|
if saved:
|
|
output["manual_landmarks"] = saved
|
|
return output
|
|
|
|
def _validate_document_status(status: str | None, default: str = "brouillon") -> str:
|
|
safe_status = (status or default).strip().lower() or default
|
|
allowed = {"brouillon", "à signer", "signe", "signé", "archive", "archivé"}
|
|
if safe_status not in allowed:
|
|
raise HTTPException(status_code=400, detail="status document invalide")
|
|
if safe_status == "signe":
|
|
return "signé"
|
|
if safe_status == "archive":
|
|
return "archivé"
|
|
return safe_status
|
|
|
|
|
|
def _validate_document_type(document_type: str | None, default: str = "consentement_photo") -> str:
|
|
safe_type = (document_type or default).strip().lower() or default
|
|
allowed = {
|
|
"consentement_photo",
|
|
"consentement_intervention",
|
|
"consentement_simulation",
|
|
"compte_rendu_consultation",
|
|
"note_interne",
|
|
"autre",
|
|
}
|
|
if safe_type not in allowed:
|
|
raise HTTPException(status_code=400, detail="type document invalide")
|
|
return safe_type
|
|
|
|
|
|
def _pdf_text_bytes(text: str) -> bytes:
|
|
data = text.encode("cp1252", errors="replace")
|
|
data = data.replace(b"\\", b"\\\\").replace(b"(", b"\\(").replace(b")", b"\\)")
|
|
return b"(" + data + b")"
|
|
|
|
|
|
def _wrap_pdf_line(line: str, width: int = 88) -> list[str]:
|
|
line = line.replace("\t", " ").strip()
|
|
if not line:
|
|
return [""]
|
|
words = line.split()
|
|
rows: list[str] = []
|
|
current = ""
|
|
for word in words:
|
|
candidate = word if not current else f"{current} {word}"
|
|
if len(candidate) <= width:
|
|
current = candidate
|
|
else:
|
|
if current:
|
|
rows.append(current)
|
|
current = word
|
|
if current:
|
|
rows.append(current)
|
|
return rows or [""]
|
|
|
|
|
|
def _build_simple_pdf(title: str, lines: list[str]) -> bytes:
|
|
wrapped: list[str] = []
|
|
for line in lines:
|
|
wrapped.extend(_wrap_pdf_line(line))
|
|
|
|
max_lines = 45
|
|
pages = [wrapped[i:i + max_lines] for i in range(0, len(wrapped), max_lines)] or [[""]]
|
|
objects: list[bytes | None] = [None, None]
|
|
catalog_id = 1
|
|
pages_id = 2
|
|
|
|
def add(obj: bytes) -> int:
|
|
objects.append(obj)
|
|
return len(objects)
|
|
|
|
font_id = add(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>")
|
|
page_ids: list[int] = []
|
|
|
|
for page_index, page_lines in enumerate(pages, start=1):
|
|
stream = bytearray()
|
|
stream.extend(b"BT\n/F1 16 Tf\n50 800 Td\n")
|
|
stream.extend(_pdf_text_bytes(title))
|
|
stream.extend(b" Tj\n/F1 10 Tf\n0 -22 Td\n")
|
|
stream.extend(_pdf_text_bytes(f"Page {page_index}/{len(pages)}"))
|
|
stream.extend(b" Tj\n/F1 11 Tf\n0 -24 Td\n")
|
|
for row in page_lines:
|
|
stream.extend(_pdf_text_bytes(row))
|
|
stream.extend(b" Tj\n0 -16 Td\n")
|
|
stream.extend(b"ET\n")
|
|
content_id = add(b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + bytes(stream) + b"endstream")
|
|
page_obj = (
|
|
f"<< /Type /Page /Parent {pages_id} 0 R /MediaBox [0 0 595 842] "
|
|
f"/Resources << /Font << /F1 {font_id} 0 R >> >> /Contents {content_id} 0 R >>"
|
|
).encode()
|
|
page_ids.append(add(page_obj))
|
|
|
|
objects[catalog_id - 1] = f"<< /Type /Catalog /Pages {pages_id} 0 R >>".encode()
|
|
kids = " ".join(f"{pid} 0 R" for pid in page_ids)
|
|
objects[pages_id - 1] = f"<< /Type /Pages /Kids [{kids}] /Count {len(page_ids)} >>".encode()
|
|
|
|
output = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
|
|
offsets = [0]
|
|
for index, obj in enumerate(objects, start=1):
|
|
assert obj is not None
|
|
offsets.append(len(output))
|
|
output.extend(f"{index} 0 obj\n".encode())
|
|
output.extend(obj)
|
|
output.extend(b"\nendobj\n")
|
|
xref_offset = len(output)
|
|
output.extend(f"xref\n0 {len(objects) + 1}\n".encode())
|
|
output.extend(b"0000000000 65535 f \n")
|
|
for offset in offsets[1:]:
|
|
output.extend(f"{offset:010d} 00000 n \n".encode())
|
|
output.extend(
|
|
f"trailer\n<< /Size {len(objects) + 1} /Root {catalog_id} 0 R >>\nstartxref\n{xref_offset}\n%%EOF\n".encode()
|
|
)
|
|
return bytes(output)
|
|
|
|
|
|
def _document_pdf_lines(patient: Patient, document: Document) -> list[str]:
|
|
return [
|
|
f"Patient: {patient.last_name.upper()} {patient.first_name}",
|
|
f"Date de naissance: {patient.birth_date or 'Non renseignée'}",
|
|
f"Téléphone: {patient.phone or 'Non renseigné'}",
|
|
f"Email: {patient.email or 'Non renseigné'}",
|
|
f"Acte envisagé: {patient.procedure or 'Non renseigné'}",
|
|
"",
|
|
f"Document: {document.title}",
|
|
f"Type: {document.document_type}",
|
|
f"Statut: {document.status}",
|
|
f"Signataire: {document.signed_by or 'Non renseigné'}",
|
|
f"Date signature: {document.signed_at or 'Non renseignée'}",
|
|
f"Généré le: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
|
|
"",
|
|
"Contenu:",
|
|
"",
|
|
*(document.content or "").splitlines(),
|
|
]
|
|
|
|
def _media_type_suffix(media_type: str) -> str:
|
|
if media_type == "image/png":
|
|
return ".png"
|
|
if media_type in {"image/webp", "image/x-webp"}:
|
|
return ".webp"
|
|
return mimetypes.guess_extension(media_type.split(";", 1)[0].strip()) or ".jpg"
|
|
|
|
def _person_name(person: dict) -> str:
|
|
name = (person.get("name") or "").strip()
|
|
return name or "Sans nom"
|
|
|
|
def _person_to_out(person: dict) -> dict:
|
|
person_id = person.get("id")
|
|
if not person_id:
|
|
raise HTTPException(status_code=502, detail="Réponse Immich personne sans id")
|
|
return {
|
|
"id": person_id,
|
|
"name": person.get("name") or None,
|
|
"thumbnail_url": f"/immich/people/{person_id}/thumbnail",
|
|
"is_hidden": person.get("isHidden"),
|
|
}
|
|
|
|
|
|
def _asset_filename(asset: dict) -> str:
|
|
filename = asset.get("originalFileName") or asset.get("fileName") or asset.get("filename")
|
|
if filename:
|
|
return str(filename)
|
|
original_path = asset.get("originalPath") or ""
|
|
if original_path:
|
|
return Path(str(original_path)).name
|
|
asset_id = str(asset.get("id") or "asset")
|
|
return f"immich_{asset_id[:8]}.jpg"
|
|
|
|
def _asset_to_out(asset: dict) -> dict:
|
|
asset_id = asset.get("id")
|
|
if not asset_id:
|
|
raise HTTPException(status_code=502, detail="Réponse Immich asset sans id")
|
|
return {
|
|
"id": asset_id,
|
|
"filename": _asset_filename(asset),
|
|
"thumbnail_url": f"/immich/assets/{asset_id}/thumbnail",
|
|
"file_url": f"/immich/assets/{asset_id}/file",
|
|
"taken_at": asset.get("localDateTime") or asset.get("fileCreatedAt") or asset.get("fileModifiedAt"),
|
|
"type": asset.get("type"),
|
|
}
|
|
|
|
|
|
@app.get("/immich/people", response_model=list[ImmichPersonOut])
|
|
async def list_immich_people(q: str | None = None, with_hidden: bool = False) -> list[dict]:
|
|
immich = ImmichClient()
|
|
people = await immich.list_people(q=q, with_hidden=with_hidden, limit=100)
|
|
return [_person_to_out(person) for person in people if person.get("id")]
|
|
|
|
|
|
@app.get("/immich/people/{person_id}/thumbnail")
|
|
async def get_immich_person_thumbnail(person_id: str) -> Response:
|
|
immich = ImmichClient()
|
|
content, media_type = await immich.get_person_thumbnail(person_id)
|
|
return Response(content=content, media_type=media_type)
|
|
|
|
|
|
|
|
|
|
@app.get("/immich/people/{person_id}/assets", response_model=list[ImmichAssetOut])
|
|
async def list_immich_person_assets(person_id: str, page: int = 1, size: int = 100) -> list[dict]:
|
|
immich = ImmichClient()
|
|
assets = await immich.list_person_assets(person_id, page=page, size=size)
|
|
return [_asset_to_out(asset) for asset in assets if asset.get("id")]
|
|
|
|
|
|
@app.get("/immich/assets/{asset_id}/thumbnail")
|
|
async def get_immich_asset_thumbnail(asset_id: str) -> Response:
|
|
immich = ImmichClient()
|
|
content, media_type = await immich.get_asset_thumbnail(asset_id, size="thumbnail")
|
|
return Response(content=content, media_type=media_type)
|
|
|
|
|
|
@app.get("/immich/assets/{asset_id}/file")
|
|
async def get_immich_asset_file(asset_id: str) -> Response:
|
|
immich = ImmichClient()
|
|
asset = await immich.get_asset_info(asset_id)
|
|
content, media_type = await immich.get_asset_original(asset_id)
|
|
filename = _asset_filename(asset)
|
|
headers = {"Content-Disposition": f'inline; filename="{filename}"'}
|
|
return Response(content=content, media_type=media_type, headers=headers)
|
|
|
|
|
|
@app.post("/patients/{patient_id}/photos/from-immich-asset/{asset_id}", response_model=PhotoOut)
|
|
async def import_photo_from_immich_asset(
|
|
patient_id: int,
|
|
asset_id: str,
|
|
db: Session = Depends(get_db),
|
|
) -> Photo:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
immich = ImmichClient()
|
|
asset = await immich.get_asset_info(asset_id)
|
|
album_name = f"{settings.immich_album_prefix} - {patient.last_name.upper()} {patient.first_name}"
|
|
album = await immich.get_or_create_album(
|
|
album_name,
|
|
description=f"Dossier esthétique patient #{patient.id}",
|
|
)
|
|
album_id = album["id"]
|
|
await immich.add_assets_to_album(album_id, [asset_id])
|
|
|
|
if not patient.immich_album_id:
|
|
patient.immich_album_id = album_id
|
|
|
|
filename = _asset_filename(asset)
|
|
photo = Photo(
|
|
patient_id=patient.id,
|
|
immich_asset_id=asset_id,
|
|
immich_album_id=album_id,
|
|
kind="original",
|
|
filename=filename,
|
|
notes="Photo complète importée depuis une personne reconnue Immich",
|
|
)
|
|
db.add(photo)
|
|
db.commit()
|
|
db.refresh(photo)
|
|
return photo
|
|
|
|
@app.post("/patients/{patient_id}/photos/from-immich-person/{person_id}", response_model=PhotoOut)
|
|
async def import_photo_from_immich_person(
|
|
patient_id: int,
|
|
person_id: str,
|
|
db: Session = Depends(get_db),
|
|
) -> Photo:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
immich = ImmichClient()
|
|
person = await immich.get_person(person_id)
|
|
content, media_type = await immich.get_person_thumbnail(person_id)
|
|
suffix = _media_type_suffix(media_type)
|
|
person_label = _person_name(person).replace("/", "-").replace("\\", "-")
|
|
|
|
with NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:
|
|
temp_path = Path(temp_file.name)
|
|
temp_file.write(content)
|
|
|
|
try:
|
|
album_name = f"{settings.immich_album_prefix} - {patient.last_name.upper()} {patient.first_name}"
|
|
album = await immich.get_or_create_album(
|
|
album_name,
|
|
description=f"Dossier esthétique patient #{patient.id}",
|
|
)
|
|
album_id = album["id"]
|
|
|
|
filename = f"visage_immich_{person_label}_{person_id[:8]}{suffix}"
|
|
asset = await immich.upload_asset(temp_path, filename)
|
|
asset_id = asset["id"]
|
|
await immich.add_assets_to_album(album_id, [asset_id])
|
|
|
|
if not patient.immich_album_id:
|
|
patient.immich_album_id = album_id
|
|
|
|
photo = Photo(
|
|
patient_id=patient.id,
|
|
immich_asset_id=asset_id,
|
|
immich_album_id=album_id,
|
|
kind="face",
|
|
filename=filename,
|
|
notes=f"Visage reconnu Immich: {_person_name(person)}",
|
|
)
|
|
db.add(photo)
|
|
db.commit()
|
|
db.refresh(photo)
|
|
return photo
|
|
finally:
|
|
temp_path.unlink(missing_ok=True)
|
|
|
|
@app.post("/patients", response_model=PatientOut)
|
|
def create_patient(payload: PatientCreate, db: Session = Depends(get_db)) -> Patient:
|
|
patient = Patient(
|
|
first_name=payload.first_name.strip(),
|
|
last_name=payload.last_name.strip(),
|
|
birth_date=_clean_text(payload.birth_date),
|
|
phone=_clean_text(payload.phone),
|
|
email=_clean_text(payload.email),
|
|
procedure=_clean_text(payload.procedure),
|
|
status=_clean_text(payload.status) or "actif",
|
|
consent_signed=_clean_text(payload.consent_signed) or "non",
|
|
notes=_clean_text(payload.notes),
|
|
)
|
|
db.add(patient)
|
|
db.commit()
|
|
db.refresh(patient)
|
|
return patient
|
|
|
|
|
|
@app.get("/patients", response_model=list[PatientOut])
|
|
def list_patients(q: str | None = None, db: Session = Depends(get_db)) -> list[Patient]:
|
|
query = db.query(Patient)
|
|
if q and q.strip():
|
|
needle = f"%{q.strip()}%"
|
|
query = query.filter(
|
|
or_(
|
|
Patient.first_name.ilike(needle),
|
|
Patient.last_name.ilike(needle),
|
|
Patient.phone.ilike(needle),
|
|
Patient.email.ilike(needle),
|
|
Patient.procedure.ilike(needle),
|
|
Patient.notes.ilike(needle),
|
|
)
|
|
)
|
|
return query.order_by(Patient.created_at.desc()).all()
|
|
|
|
|
|
@app.get("/patients/{patient_id}", response_model=PatientOut)
|
|
def get_patient(patient_id: int, db: Session = Depends(get_db)) -> Patient:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
return patient
|
|
|
|
|
|
@app.patch("/patients/{patient_id}", response_model=PatientOut)
|
|
def update_patient(patient_id: int, payload: PatientUpdate, db: Session = Depends(get_db)) -> Patient:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
data = payload.model_dump(exclude_unset=True)
|
|
for key, value in data.items():
|
|
if isinstance(value, str):
|
|
value = value.strip() or None
|
|
setattr(patient, key, value)
|
|
|
|
if not patient.status:
|
|
patient.status = "actif"
|
|
if not patient.consent_signed:
|
|
patient.consent_signed = "non"
|
|
|
|
db.commit()
|
|
db.refresh(patient)
|
|
return patient
|
|
|
|
|
|
|
|
|
|
@app.get("/patients/{patient_id}/consultations", response_model=list[ConsultationOut])
|
|
def list_patient_consultations(patient_id: int, db: Session = Depends(get_db)) -> list[Consultation]:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
return (
|
|
db.query(Consultation)
|
|
.filter(Consultation.patient_id == patient_id)
|
|
.order_by(Consultation.created_at.desc())
|
|
.all()
|
|
)
|
|
|
|
|
|
@app.post("/patients/{patient_id}/consultations", response_model=ConsultationOut)
|
|
def create_patient_consultation(
|
|
patient_id: int,
|
|
payload: ConsultationCreate,
|
|
db: Session = Depends(get_db),
|
|
) -> Consultation:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
consultation = Consultation(
|
|
patient_id=patient.id,
|
|
consultation_date=_clean_text(payload.consultation_date),
|
|
procedure=_clean_text(payload.procedure) or patient.procedure,
|
|
reason=_clean_text(payload.reason),
|
|
plan=_clean_text(payload.plan),
|
|
follow_up=_clean_text(payload.follow_up),
|
|
status=_clean_text(payload.status) or "prévu",
|
|
notes=_clean_text(payload.notes),
|
|
)
|
|
db.add(consultation)
|
|
db.commit()
|
|
db.refresh(consultation)
|
|
return consultation
|
|
|
|
|
|
@app.patch("/consultations/{consultation_id}", response_model=ConsultationOut)
|
|
def update_consultation(
|
|
consultation_id: int,
|
|
payload: ConsultationUpdate,
|
|
db: Session = Depends(get_db),
|
|
) -> Consultation:
|
|
consultation = db.get(Consultation, consultation_id)
|
|
if not consultation:
|
|
raise HTTPException(status_code=404, detail="Consultation introuvable")
|
|
|
|
data = payload.model_dump(exclude_unset=True)
|
|
for key, value in data.items():
|
|
if isinstance(value, str):
|
|
value = value.strip() or None
|
|
setattr(consultation, key, value)
|
|
|
|
if not consultation.status:
|
|
consultation.status = "prévu"
|
|
|
|
db.commit()
|
|
db.refresh(consultation)
|
|
return consultation
|
|
|
|
|
|
@app.delete("/consultations/{consultation_id}")
|
|
def delete_consultation(consultation_id: int, db: Session = Depends(get_db)) -> dict:
|
|
consultation = db.get(Consultation, consultation_id)
|
|
if not consultation:
|
|
raise HTTPException(status_code=404, detail="Consultation introuvable")
|
|
db.delete(consultation)
|
|
db.commit()
|
|
return {"status": "deleted", "id": consultation_id}
|
|
|
|
|
|
@app.get("/patients/{patient_id}/documents", response_model=list[DocumentOut])
|
|
def list_patient_documents(patient_id: int, db: Session = Depends(get_db)) -> list[Document]:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
return (
|
|
db.query(Document)
|
|
.filter(Document.patient_id == patient_id)
|
|
.order_by(Document.created_at.desc())
|
|
.all()
|
|
)
|
|
|
|
|
|
@app.post("/patients/{patient_id}/documents", response_model=DocumentOut)
|
|
def create_patient_document(
|
|
patient_id: int,
|
|
payload: DocumentCreate,
|
|
db: Session = Depends(get_db),
|
|
) -> Document:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
if payload.consultation_id is not None:
|
|
consultation = db.get(Consultation, payload.consultation_id)
|
|
if not consultation or consultation.patient_id != patient_id:
|
|
raise HTTPException(status_code=400, detail="Consultation invalide pour ce patient")
|
|
|
|
title = _clean_text(payload.title)
|
|
if not title:
|
|
raise HTTPException(status_code=400, detail="Titre document obligatoire")
|
|
|
|
document = Document(
|
|
patient_id=patient.id,
|
|
consultation_id=payload.consultation_id,
|
|
document_type=_validate_document_type(payload.document_type),
|
|
title=title,
|
|
status=_validate_document_status(payload.status),
|
|
content=_clean_text(payload.content),
|
|
signed_by=_clean_text(payload.signed_by),
|
|
signed_at=_clean_text(payload.signed_at),
|
|
)
|
|
db.add(document)
|
|
db.commit()
|
|
db.refresh(document)
|
|
return document
|
|
|
|
|
|
@app.patch("/documents/{document_id}", response_model=DocumentOut)
|
|
def update_document(
|
|
document_id: int,
|
|
payload: DocumentUpdate,
|
|
db: Session = Depends(get_db),
|
|
) -> Document:
|
|
document = db.get(Document, document_id)
|
|
if not document:
|
|
raise HTTPException(status_code=404, detail="Document introuvable")
|
|
|
|
data = payload.model_dump(exclude_unset=True)
|
|
if "consultation_id" in data and data["consultation_id"] is not None:
|
|
consultation = db.get(Consultation, data["consultation_id"])
|
|
if not consultation or consultation.patient_id != document.patient_id:
|
|
raise HTTPException(status_code=400, detail="Consultation invalide pour ce patient")
|
|
|
|
for key, value in data.items():
|
|
if key == "document_type" and value is not None:
|
|
value = _validate_document_type(value, default=document.document_type)
|
|
elif key == "status" and value is not None:
|
|
value = _validate_document_status(value, default=document.status)
|
|
elif isinstance(value, str):
|
|
value = value.strip() or None
|
|
setattr(document, key, value)
|
|
|
|
if not document.title:
|
|
raise HTTPException(status_code=400, detail="Titre document obligatoire")
|
|
if not document.document_type:
|
|
document.document_type = "consentement_photo"
|
|
if not document.status:
|
|
document.status = "brouillon"
|
|
|
|
db.commit()
|
|
db.refresh(document)
|
|
return document
|
|
|
|
|
|
@app.delete("/documents/{document_id}")
|
|
def delete_document(document_id: int, db: Session = Depends(get_db)) -> dict:
|
|
document = db.get(Document, document_id)
|
|
if not document:
|
|
raise HTTPException(status_code=404, detail="Document introuvable")
|
|
db.delete(document)
|
|
db.commit()
|
|
return {"status": "deleted", "id": document_id}
|
|
|
|
|
|
@app.get("/documents/{document_id}/pdf")
|
|
def get_document_pdf(document_id: int, db: Session = Depends(get_db)) -> Response:
|
|
document = db.get(Document, document_id)
|
|
if not document:
|
|
raise HTTPException(status_code=404, detail="Document introuvable")
|
|
patient = db.get(Patient, document.patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
pdf = _build_simple_pdf(document.title, _document_pdf_lines(patient, document))
|
|
safe_name = "_".join((document.title or "document").split())[:80] or "document"
|
|
headers = {"Content-Disposition": f'inline; filename="{safe_name}.pdf"'}
|
|
return Response(content=pdf, media_type="application/pdf", headers=headers)
|
|
|
|
|
|
@app.get("/patients/{patient_id}/photos", response_model=list[PhotoOut])
|
|
def list_patient_photos(patient_id: int, db: Session = Depends(get_db)) -> list[Photo]:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
return db.query(Photo).filter(Photo.patient_id == patient_id).order_by(Photo.created_at.desc()).all()
|
|
|
|
|
|
@app.get("/photos/{photo_id}/thumbnail")
|
|
async def get_photo_thumbnail(photo_id: int, db: Session = Depends(get_db)) -> Response:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
immich = ImmichClient()
|
|
content, media_type = await immich.get_asset_thumbnail(photo.immich_asset_id, size="thumbnail")
|
|
return Response(content=content, media_type=media_type)
|
|
|
|
|
|
@app.get("/photos/{photo_id}/file")
|
|
async def get_photo_file(photo_id: int, db: Session = Depends(get_db)) -> Response:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
immich = ImmichClient()
|
|
content, media_type = await immich.get_asset_original(photo.immich_asset_id)
|
|
headers = {"Content-Disposition": f'inline; filename="{photo.filename}"'}
|
|
return Response(content=content, media_type=media_type, headers=headers)
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/photos/{photo_id}/face-landmarks")
|
|
def get_photo_face_landmarks(photo_id: int, db: Session = Depends(get_db)) -> dict:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
row = db.execute(
|
|
text("SELECT manual_landmarks, updated_at FROM photo_face_landmarks WHERE photo_id = :photo_id"),
|
|
{"photo_id": photo_id},
|
|
).fetchone()
|
|
if not row:
|
|
return {"photo_id": photo_id, "manual_landmarks": None, "updated_at": None}
|
|
return {"photo_id": photo_id, "manual_landmarks": json.loads(row[0]), "updated_at": str(row[1]) if row[1] else None}
|
|
|
|
|
|
@app.put("/photos/{photo_id}/face-landmarks")
|
|
def save_photo_face_landmarks(
|
|
photo_id: int,
|
|
payload: dict = Body(default_factory=dict),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
landmarks = _sanitize_manual_landmarks(payload.get("manual_landmarks") or payload)
|
|
encoded = json.dumps(landmarks, ensure_ascii=False, separators=(",", ":"))
|
|
db.execute(
|
|
text("""
|
|
INSERT INTO photo_face_landmarks (photo_id, manual_landmarks, updated_at)
|
|
VALUES (:photo_id, :manual_landmarks, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(photo_id) DO UPDATE SET
|
|
manual_landmarks = excluded.manual_landmarks,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
"""),
|
|
{"photo_id": photo_id, "manual_landmarks": encoded},
|
|
)
|
|
db.commit()
|
|
return {"photo_id": photo_id, "manual_landmarks": landmarks, "saved": True}
|
|
|
|
|
|
@app.delete("/photos/{photo_id}/face-landmarks")
|
|
def delete_photo_face_landmarks(photo_id: int, db: Session = Depends(get_db)) -> dict:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
db.execute(text("DELETE FROM photo_face_landmarks WHERE photo_id = :photo_id"), {"photo_id": photo_id})
|
|
db.commit()
|
|
return {"photo_id": photo_id, "deleted": True}
|
|
|
|
|
|
@app.post("/photos/{photo_id}/face-guides")
|
|
async def get_photo_face_guides(
|
|
photo_id: int,
|
|
payload: dict = Body(default_factory=dict),
|
|
db: Session = Depends(get_db),
|
|
) -> Response:
|
|
"""Retourne une image PNG avec repères de détection visage.
|
|
|
|
Sert à vérifier si le backend cible bien le visage avant de lancer une simulation.
|
|
"""
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
payload = _payload_with_saved_landmarks(payload, db, photo_id)
|
|
immich = ImmichClient()
|
|
content, _media_type = await immich.get_asset_original(photo.immich_asset_id)
|
|
rendered, meta = render_face_guides(content, payload)
|
|
headers = {
|
|
"X-Face-Detected": "true" if meta.get("detected") else "false",
|
|
"X-Face-Method": str(meta.get("method") or "unknown"),
|
|
"X-Face-Landmark-Count": str(meta.get("landmark_count") or 0),
|
|
}
|
|
return Response(content=rendered, media_type="image/png", headers=headers)
|
|
|
|
|
|
@app.post("/photos/{photo_id}/face-transform/preview")
|
|
async def preview_photo_face_transform(
|
|
photo_id: int,
|
|
payload: dict = Body(default_factory=dict),
|
|
db: Session = Depends(get_db),
|
|
) -> Response:
|
|
"""Retourne un aperçu PNG transformé côté backend.
|
|
|
|
La transformation utilise OpenCV côté serveur, ce qui donne une image plus stable
|
|
que le rendu Flutter local, cette petite aventure de pixels approximatifs.
|
|
"""
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
payload = _payload_with_saved_landmarks(payload, db, photo_id)
|
|
immich = ImmichClient()
|
|
content, _media_type = await immich.get_asset_original(photo.immich_asset_id)
|
|
rendered, meta = render_face_transform(content, payload, preview=False)
|
|
headers = {
|
|
"X-Face-Detected": "true" if meta.get("detected") else "false",
|
|
"X-Face-Method": str(meta.get("method") or "unknown"),
|
|
"X-Face-Landmark-Count": str(meta.get("landmark_count") or 0),
|
|
}
|
|
return Response(content=rendered, media_type="image/png", headers=headers)
|
|
|
|
|
|
@app.post("/photos/{photo_id}/face-transform", response_model=PhotoOut)
|
|
async def save_photo_face_transform(
|
|
photo_id: int,
|
|
payload: dict = Body(default_factory=dict),
|
|
db: Session = Depends(get_db),
|
|
) -> Photo:
|
|
"""Génère une simulation visage côté backend, l'envoie dans Immich et l'ajoute au dossier patient."""
|
|
source_photo = db.get(Photo, photo_id)
|
|
if not source_photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
patient = db.get(Patient, source_photo.patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
payload = _payload_with_saved_landmarks(payload, db, photo_id)
|
|
immich = ImmichClient()
|
|
content, _media_type = await immich.get_asset_original(source_photo.immich_asset_id)
|
|
rendered, meta = render_face_transform(content, payload, preview=False)
|
|
|
|
with NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
|
|
temp_path = Path(temp_file.name)
|
|
temp_file.write(rendered)
|
|
|
|
try:
|
|
album_name = f"{settings.immich_album_prefix} - {patient.last_name.upper()} {patient.first_name}"
|
|
album = await immich.get_or_create_album(
|
|
album_name,
|
|
description=f"Dossier esthétique patient #{patient.id}",
|
|
)
|
|
album_id = album["id"]
|
|
filename = f"simulation_backend_{source_photo.id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
|
|
asset = await immich.upload_asset(temp_path, filename)
|
|
asset_id = asset["id"]
|
|
await immich.add_assets_to_album(album_id, [asset_id])
|
|
|
|
if not patient.immich_album_id:
|
|
patient.immich_album_id = album_id
|
|
|
|
status = "visage détecté" if meta.get("detected") else "repères approximatifs"
|
|
photo = Photo(
|
|
patient_id=patient.id,
|
|
immich_asset_id=asset_id,
|
|
immich_album_id=album_id,
|
|
kind="simulation",
|
|
filename=filename,
|
|
notes=f"Simulation visage backend OpenCV non contractuelle depuis {source_photo.filename} ({status}).",
|
|
)
|
|
db.add(photo)
|
|
db.commit()
|
|
db.refresh(photo)
|
|
return photo
|
|
finally:
|
|
temp_path.unlink(missing_ok=True)
|
|
|
|
|
|
@app.patch("/photos/{photo_id}", response_model=PhotoOut)
|
|
def update_photo(photo_id: int, payload: PhotoUpdate, db: Session = Depends(get_db)) -> Photo:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
data = payload.model_dump(exclude_unset=True)
|
|
if "kind" in data and data["kind"] is not None:
|
|
photo.kind = _validate_photo_kind(data["kind"], default=photo.kind)
|
|
if "notes" in data:
|
|
photo.notes = _clean_text(data["notes"])
|
|
|
|
db.commit()
|
|
db.refresh(photo)
|
|
return photo
|
|
|
|
|
|
@app.delete("/photos/{photo_id}")
|
|
def delete_photo(photo_id: int, db: Session = Depends(get_db)) -> dict:
|
|
photo = db.get(Photo, photo_id)
|
|
if not photo:
|
|
raise HTTPException(status_code=404, detail="Photo introuvable")
|
|
|
|
# On supprime seulement l'entrée du dossier patient local.
|
|
# L'asset Immich n'est pas détruit, pour éviter d'effacer une photo source utilisée ailleurs.
|
|
db.delete(photo)
|
|
db.commit()
|
|
return {"status": "deleted", "id": photo_id, "immich_asset_deleted": False}
|
|
|
|
|
|
@app.post("/patients/{patient_id}/photos", response_model=PhotoOut)
|
|
async def upload_patient_photo(
|
|
patient_id: int,
|
|
kind: str = Form("original"),
|
|
notes: str | None = Form(None),
|
|
file: UploadFile = File(...),
|
|
db: Session = Depends(get_db),
|
|
) -> Photo:
|
|
patient = db.get(Patient, patient_id)
|
|
if not patient:
|
|
raise HTTPException(status_code=404, detail="Patient introuvable")
|
|
|
|
safe_kind = _validate_photo_kind(kind)
|
|
|
|
if not file.filename:
|
|
raise HTTPException(status_code=400, detail="Nom de fichier manquant")
|
|
|
|
suffix = Path(file.filename).suffix or ".jpg"
|
|
with NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:
|
|
temp_path = Path(temp_file.name)
|
|
temp_file.write(await file.read())
|
|
|
|
try:
|
|
album_name = f"{settings.immich_album_prefix} - {patient.last_name.upper()} {patient.first_name}"
|
|
immich = ImmichClient()
|
|
album = await immich.get_or_create_album(
|
|
album_name,
|
|
description=f"Dossier esthétique patient #{patient.id}",
|
|
)
|
|
album_id = album["id"]
|
|
|
|
asset = await immich.upload_asset(temp_path, file.filename)
|
|
asset_id = asset["id"]
|
|
await immich.add_assets_to_album(album_id, [asset_id])
|
|
|
|
if not patient.immich_album_id:
|
|
patient.immich_album_id = album_id
|
|
|
|
photo = Photo(
|
|
patient_id=patient.id,
|
|
immich_asset_id=asset_id,
|
|
immich_album_id=album_id,
|
|
kind=safe_kind,
|
|
filename=file.filename,
|
|
notes=_clean_text(notes),
|
|
)
|
|
db.add(photo)
|
|
db.commit()
|
|
db.refresh(photo)
|
|
return photo
|
|
finally:
|
|
temp_path.unlink(missing_ok=True)
|