before gemiin

This commit is contained in:
2026-02-22 22:18:32 +01:00
parent fb33540bb0
commit 9db5cbf236
147 changed files with 7948 additions and 531 deletions

View File

@@ -40,6 +40,7 @@ EXPECTED_COLUMNS: dict[str, list[tuple[str, str, str | None]]] = {
"tool": [
("photo_url", "TEXT", None),
("video_url", "TEXT", None),
("notice_texte", "TEXT", None),
("notice_fichier_url", "TEXT", None),
("boutique_nom", "TEXT", None),
("boutique_url", "TEXT", None),

View File

@@ -10,6 +10,7 @@ class Tool(SQLModel, table=True):
categorie: Optional[str] = None # beche|fourche|griffe|arrosage|taille|autre
photo_url: Optional[str] = None
video_url: Optional[str] = None
notice_texte: Optional[str] = None
notice_fichier_url: Optional[str] = None
boutique_nom: Optional[str] = None
boutique_url: Optional[str] = None

View File

@@ -1,8 +1,9 @@
import os
import unicodedata
import uuid
from typing import List, Optional
from fastapi import APIRouter, Body, Depends, File, HTTPException, Query, UploadFile, status
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
from pydantic import BaseModel
from sqlmodel import Session, select
@@ -16,9 +17,91 @@ class MediaPatch(BaseModel):
entity_id: Optional[int] = None
titre: Optional[str] = None
CANONICAL_ENTITY_TYPES = {
"jardin",
"plante",
"adventice",
"outil",
"plantation",
"bibliotheque",
}
ENTITY_TYPE_ALIASES = {
# Canonique
"jardin": "jardin",
"plante": "plante",
"adventice": "adventice",
"outil": "outil",
"plantation": "plantation",
"bibliotheque": "bibliotheque",
# Variantes FR
"jardins": "jardin",
"plantes": "plante",
"adventices": "adventice",
"outils": "outil",
"plantations": "plantation",
"bibliotheques": "bibliotheque",
"bibliotheque_media": "bibliotheque",
# Variantes EN (courantes via API)
"garden": "jardin",
"gardens": "jardin",
"plant": "plante",
"plants": "plante",
"weed": "adventice",
"weeds": "adventice",
"tool": "outil",
"tools": "outil",
"planting": "plantation",
"plantings": "plantation",
"library": "bibliotheque",
"media_library": "bibliotheque",
}
router = APIRouter(tags=["media"])
def _normalize_token(value: str) -> str:
token = (value or "").strip().lower()
token = unicodedata.normalize("NFKD", token).encode("ascii", "ignore").decode("ascii")
return token.replace("-", "_").replace(" ", "_")
def _normalize_entity_type(value: str, *, strict: bool = True) -> str:
token = _normalize_token(value)
canonical = ENTITY_TYPE_ALIASES.get(token, token)
if canonical in CANONICAL_ENTITY_TYPES:
return canonical
if strict:
allowed = ", ".join(sorted(CANONICAL_ENTITY_TYPES))
raise HTTPException(
status_code=422,
detail=f"entity_type invalide: '{value}'. Valeurs autorisees: {allowed}",
)
return value
def _entity_type_candidates(value: str) -> set[str]:
canonical = _normalize_entity_type(value, strict=True)
candidates = {canonical}
for alias, target in ENTITY_TYPE_ALIASES.items():
if target == canonical:
candidates.add(alias)
return candidates
def _canonicalize_rows(rows: List[Media], session: Session) -> None:
changed = False
for media in rows:
normalized = _normalize_entity_type(media.entity_type, strict=False)
if normalized in CANONICAL_ENTITY_TYPES and normalized != media.entity_type:
media.entity_type = normalized
session.add(media)
changed = True
if changed:
session.commit()
def _save_webp(data: bytes, max_px: int) -> str:
try:
from PIL import Image
@@ -47,12 +130,12 @@ async def upload_file(file: UploadFile = File(...)):
name = _save_webp(data, 1200)
thumb = _save_webp(data, 300)
return {"url": f"/uploads/{name}", "thumbnail_url": f"/uploads/{thumb}"}
else:
name = f"{uuid.uuid4()}_{file.filename}"
path = os.path.join(UPLOAD_DIR, name)
with open(path, "wb") as f:
f.write(data)
return {"url": f"/uploads/{name}", "thumbnail_url": None}
name = f"{uuid.uuid4()}_{file.filename}"
path = os.path.join(UPLOAD_DIR, name)
with open(path, "wb") as f:
f.write(data)
return {"url": f"/uploads/{name}", "thumbnail_url": None}
@router.get("/media/all", response_model=List[Media])
@@ -63,8 +146,10 @@ def list_all_media(
"""Retourne tous les médias, filtrés optionnellement par entity_type."""
q = select(Media).order_by(Media.created_at.desc())
if entity_type:
q = q.where(Media.entity_type == entity_type)
return session.exec(q).all()
q = q.where(Media.entity_type.in_(_entity_type_candidates(entity_type)))
rows = session.exec(q).all()
_canonicalize_rows(rows, session)
return rows
@router.get("/media", response_model=List[Media])
@@ -73,15 +158,19 @@ def list_media(
entity_id: int = Query(...),
session: Session = Depends(get_session),
):
return session.exec(
rows = session.exec(
select(Media).where(
Media.entity_type == entity_type, Media.entity_id == entity_id
Media.entity_type.in_(_entity_type_candidates(entity_type)),
Media.entity_id == entity_id,
)
).all()
_canonicalize_rows(rows, session)
return rows
@router.post("/media", response_model=Media, status_code=status.HTTP_201_CREATED)
def create_media(m: Media, session: Session = Depends(get_session)):
m.entity_type = _normalize_entity_type(m.entity_type, strict=True)
session.add(m)
session.commit()
session.refresh(m)
@@ -93,7 +182,12 @@ def update_media(id: int, payload: MediaPatch, session: Session = Depends(get_se
m = session.get(Media, id)
if not m:
raise HTTPException(404, "Media introuvable")
for k, v in payload.model_dump(exclude_none=True).items():
updates = payload.model_dump(exclude_none=True)
if "entity_type" in updates:
updates["entity_type"] = _normalize_entity_type(updates["entity_type"], strict=True)
for k, v in updates.items():
setattr(m, k, v)
session.add(m)
session.commit()

View File

@@ -12,7 +12,7 @@ def list_plants(
categorie: Optional[str] = Query(None),
session: Session = Depends(get_session),
):
q = select(Plant)
q = select(Plant).order_by(Plant.nom_commun, Plant.variete, Plant.id)
if categorie:
q = q.where(Plant.categorie == categorie)
return session.exec(q).all()

View File

@@ -1,18 +1,28 @@
import os
import shutil
import time
import json
import tempfile
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from fastapi import APIRouter, Depends
from fastapi.responses import FileResponse
from starlette.background import BackgroundTask
from sqlmodel import Session, select
from app.database import get_session
from app.models.settings import UserSettings
from app.config import UPLOAD_DIR
from app.config import DATABASE_URL, UPLOAD_DIR
router = APIRouter(tags=["réglages"])
_PREV_CPU_USAGE_USEC: int | None = None
_PREV_CPU_TS: float | None = None
_TEXT_EXTENSIONS = {
".txt", ".md", ".markdown", ".json", ".csv", ".log", ".ini", ".yaml", ".yml", ".xml"
}
def _read_int_from_paths(paths: list[str]) -> int | None:
@@ -113,6 +123,68 @@ def _disk_stats() -> dict[str, Any]:
}
def _safe_remove(path: str) -> None:
try:
os.remove(path)
except OSError:
pass
def _resolve_sqlite_db_path() -> Path | None:
prefix = "sqlite:///"
if not DATABASE_URL.startswith(prefix):
return None
raw = DATABASE_URL[len(prefix):]
if not raw:
return None
db_path = Path(raw)
if db_path.is_absolute():
return db_path
return (Path.cwd() / db_path).resolve()
def _zip_directory(zipf: zipfile.ZipFile, source_dir: Path, arc_prefix: str) -> int:
count = 0
if not source_dir.is_dir():
return count
for root, _, files in os.walk(source_dir):
root_path = Path(root)
for name in files:
file_path = root_path / name
if not file_path.is_file():
continue
rel = file_path.relative_to(source_dir)
arcname = str(Path(arc_prefix) / rel)
zipf.write(file_path, arcname=arcname)
count += 1
return count
def _zip_data_text_files(
zipf: zipfile.ZipFile,
data_root: Path,
db_path: Path | None,
uploads_dir: Path,
) -> int:
count = 0
if not data_root.is_dir():
return count
for root, _, files in os.walk(data_root):
root_path = Path(root)
for name in files:
file_path = root_path / name
if db_path and file_path == db_path:
continue
if uploads_dir in file_path.parents:
continue
if file_path.suffix.lower() not in _TEXT_EXTENSIONS:
continue
rel = file_path.relative_to(data_root)
zipf.write(file_path, arcname=str(Path("data_text") / rel))
count += 1
return count
@router.get("/settings")
def get_settings(session: Session = Depends(get_session)):
rows = session.exec(select(UserSettings)).all()
@@ -161,3 +233,51 @@ def get_debug_system_stats() -> dict[str, Any]:
"memory": _memory_stats(),
"disk": _disk_stats(),
}
@router.get("/settings/backup/download")
def download_backup_zip() -> FileResponse:
now = datetime.now(timezone.utc)
ts = now.strftime("%Y%m%d_%H%M%S")
db_path = _resolve_sqlite_db_path()
uploads_dir = Path(UPLOAD_DIR).resolve()
data_root = db_path.parent if db_path else uploads_dir.parent
fd, tmp_zip_path = tempfile.mkstemp(prefix=f"jardin_backup_{ts}_", suffix=".zip")
os.close(fd)
tmp_zip = Path(tmp_zip_path)
stats = {
"database_files": 0,
"upload_files": 0,
"text_files": 0,
}
with zipfile.ZipFile(tmp_zip, mode="w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as zipf:
if db_path and db_path.is_file():
zipf.write(db_path, arcname=f"db/{db_path.name}")
stats["database_files"] = 1
stats["upload_files"] = _zip_directory(zipf, uploads_dir, "uploads")
stats["text_files"] = _zip_data_text_files(zipf, data_root, db_path, uploads_dir)
manifest = {
"generated_at_utc": now.isoformat(),
"database_url": DATABASE_URL,
"paths": {
"database_path": str(db_path) if db_path else None,
"uploads_path": str(uploads_dir),
"data_root": str(data_root),
},
"included": stats,
"text_extensions": sorted(_TEXT_EXTENSIONS),
}
zipf.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2))
download_name = f"jardin_backup_{ts}.zip"
return FileResponse(
path=str(tmp_zip),
media_type="application/zip",
filename=download_name,
background=BackgroundTask(_safe_remove, str(tmp_zip)),
)

View File

@@ -12,6 +12,7 @@ router = APIRouter(tags=["tâches"])
def list_tasks(
statut: Optional[str] = None,
garden_id: Optional[int] = None,
planting_id: Optional[int] = None,
session: Session = Depends(get_session),
):
q = select(Task)
@@ -19,6 +20,9 @@ def list_tasks(
q = q.where(Task.statut == statut)
if garden_id:
q = q.where(Task.garden_id == garden_id)
if planting_id:
q = q.where(Task.planting_id == planting_id)
q = q.order_by(Task.echeance, Task.created_at.desc())
return session.exec(q).all()

View File

@@ -0,0 +1,26 @@
def test_create_media_normalizes_english_entity_type(client):
r = client.post(
"/api/media",
json={
"entity_type": "plant",
"entity_id": 12,
"url": "/uploads/test.webp",
},
)
assert r.status_code == 201
assert r.json()["entity_type"] == "plante"
def test_list_media_accepts_alias_entity_type_filter(client):
client.post(
"/api/media",
json={
"entity_type": "plante",
"entity_id": 99,
"url": "/uploads/test2.webp",
},
)
r = client.get("/api/media", params={"entity_type": "plant", "entity_id": 99})
assert r.status_code == 200
assert len(r.json()) == 1
assert r.json()[0]["entity_type"] == "plante"

View File

@@ -12,6 +12,16 @@ def test_list_plants(client):
assert len(r.json()) == 2
def test_allow_same_common_name_with_different_varieties(client):
client.post("/api/plants", json={"nom_commun": "Tomate", "variete": "Roma"})
client.post("/api/plants", json={"nom_commun": "Tomate", "variete": "Andine Cornue"})
r = client.get("/api/plants")
assert r.status_code == 200
tomates = [p for p in r.json() if p["nom_commun"] == "Tomate"]
assert len(tomates) == 2
assert {p.get("variete") for p in tomates} == {"Roma", "Andine Cornue"}
def test_get_plant(client):
r = client.post("/api/plants", json={"nom_commun": "Basilic"})
id = r.json()["id"]

View File

@@ -26,3 +26,13 @@ def test_update_task_statut(client):
r2 = client.put(f"/api/tasks/{id}", json={"titre": "À faire", "statut": "fait"})
assert r2.status_code == 200
assert r2.json()["statut"] == "fait"
def test_filter_tasks_by_planting_id(client):
client.post("/api/tasks", json={"titre": "Template arrosage", "statut": "template"})
client.post("/api/tasks", json={"titre": "Arroser rang 1", "statut": "a_faire", "planting_id": 10})
client.post("/api/tasks", json={"titre": "Arroser rang 2", "statut": "a_faire", "planting_id": 11})
r = client.get("/api/tasks?planting_id=10")
assert r.status_code == 200
assert len(r.json()) == 1
assert r.json()[0]["planting_id"] == 10

View File

@@ -28,3 +28,15 @@ def test_tool_with_video_url(client):
)
assert r.status_code == 201
assert r.json()["video_url"] == "/uploads/demo-outil.mp4"
def test_tool_with_notice_texte(client):
r = client.post(
"/api/tools",
json={
"nom": "Sécateur",
"notice_texte": "Aiguiser la lame tous les 3 mois.",
},
)
assert r.status_code == 201
assert r.json()["notice_texte"] == "Aiguiser la lame tous les 3 mois."