234 lines
7.6 KiB
Python
234 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.db import models, schemas
|
|
|
|
|
|
"""
|
|
CRUD minimal pour manipuler les produits dans la base SQLite.
|
|
"""
|
|
|
|
|
|
def get_product(db: Session, product_id: int) -> models.Product | None:
|
|
return db.query(models.Product).filter(models.Product.id == product_id).first()
|
|
|
|
|
|
def list_products(db: Session, skip: int = 0, limit: int = 100) -> list[models.Product]:
|
|
return db.query(models.Product).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create_product(db: Session, data: schemas.ProductCreate) -> models.Product:
|
|
# Convertir les HttpUrl en strings pour SQLite
|
|
data_dict = data.model_dump()
|
|
if data_dict.get("url"):
|
|
data_dict["url"] = str(data_dict["url"])
|
|
if data_dict.get("url_image"):
|
|
data_dict["url_image"] = str(data_dict["url_image"])
|
|
|
|
product = models.Product(**data_dict)
|
|
db.add(product)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise
|
|
db.refresh(product)
|
|
return product
|
|
|
|
|
|
def create_product_with_snapshot(
|
|
db: Session, data: schemas.ProductCreateWithSnapshot
|
|
) -> models.Product:
|
|
"""Crée un produit avec un snapshot initial (données issues de la prévisualisation)."""
|
|
data_dict = data.model_dump()
|
|
|
|
# Extraire les champs du snapshot
|
|
snapshot_fields = [
|
|
"prix_actuel",
|
|
"prix_conseille",
|
|
"prix_min_30j",
|
|
"etat_stock",
|
|
"en_stock",
|
|
"note",
|
|
"nombre_avis",
|
|
"prime",
|
|
"choix_amazon",
|
|
"offre_limitee",
|
|
"exclusivite_amazon",
|
|
"a_propos",
|
|
"description",
|
|
"carateristique",
|
|
"details",
|
|
"categorie_amazon",
|
|
]
|
|
snapshot_data = {k: data_dict.pop(k) for k in snapshot_fields if k in data_dict}
|
|
|
|
# Sérialiser les champs JSON pour le stockage
|
|
if snapshot_data.get("a_propos") is not None:
|
|
snapshot_data["a_propos"] = json.dumps(snapshot_data["a_propos"], ensure_ascii=False)
|
|
if snapshot_data.get("carateristique") is not None:
|
|
snapshot_data["carateristique"] = json.dumps(snapshot_data["carateristique"], ensure_ascii=False)
|
|
if snapshot_data.get("details") is not None:
|
|
snapshot_data["details"] = json.dumps(snapshot_data["details"], ensure_ascii=False)
|
|
|
|
# Convertir les HttpUrl en strings pour SQLite
|
|
if data_dict.get("url"):
|
|
data_dict["url"] = str(data_dict["url"])
|
|
if data_dict.get("url_image"):
|
|
data_dict["url_image"] = str(data_dict["url_image"])
|
|
|
|
# Créer le produit
|
|
product = models.Product(**data_dict)
|
|
db.add(product)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise
|
|
db.refresh(product)
|
|
|
|
# Créer le snapshot initial si on a des données
|
|
has_snapshot_data = any(v is not None for v in snapshot_data.values())
|
|
if has_snapshot_data:
|
|
snapshot = models.ProductSnapshot(
|
|
produit_id=product.id,
|
|
statut_scrap="preview",
|
|
**snapshot_data,
|
|
)
|
|
db.add(snapshot)
|
|
db.commit()
|
|
|
|
return product
|
|
|
|
|
|
def update_product(db: Session, product: models.Product, changes: schemas.ProductUpdate) -> models.Product:
|
|
for field, value in changes.dict(exclude_unset=True).items():
|
|
setattr(product, field, value)
|
|
db.add(product)
|
|
db.commit()
|
|
db.refresh(product)
|
|
return product
|
|
|
|
|
|
def remove_product(db: Session, product: models.Product) -> None:
|
|
db.delete(product)
|
|
db.commit()
|
|
|
|
|
|
def list_snapshots(
|
|
db: Session, product_id: int, days: int | None = None, limit: int = 1000
|
|
) -> list[models.ProductSnapshot]:
|
|
"""Retourne les snapshots d'un produit, filtrés par nombre de jours si spécifié."""
|
|
from datetime import datetime, timedelta
|
|
|
|
query = db.query(models.ProductSnapshot).filter(
|
|
models.ProductSnapshot.produit_id == product_id
|
|
)
|
|
|
|
# Filtrer par date si days est spécifié
|
|
if days is not None and days > 0:
|
|
cutoff_date = datetime.utcnow() - timedelta(days=days)
|
|
query = query.filter(models.ProductSnapshot.scrape_le >= cutoff_date)
|
|
|
|
return query.order_by(models.ProductSnapshot.scrape_le.desc()).limit(limit).all()
|
|
|
|
|
|
def get_latest_snapshot(db: Session, product_id: int) -> models.ProductSnapshot | None:
|
|
return (
|
|
db.query(models.ProductSnapshot)
|
|
.filter(models.ProductSnapshot.produit_id == product_id)
|
|
.order_by(models.ProductSnapshot.scrape_le.desc())
|
|
.first()
|
|
)
|
|
|
|
|
|
def get_product_with_snapshot(db: Session, product_id: int) -> dict | None:
|
|
"""Retourne un produit enrichi avec les données du dernier snapshot."""
|
|
product = get_product(db, product_id)
|
|
if not product:
|
|
return None
|
|
return _enrich_product_with_snapshot(db, product)
|
|
|
|
|
|
def list_products_with_snapshots(db: Session, skip: int = 0, limit: int = 100) -> list[dict]:
|
|
"""Retourne la liste des produits enrichis avec leurs derniers snapshots."""
|
|
products = list_products(db, skip=skip, limit=limit)
|
|
return [_enrich_product_with_snapshot(db, p) for p in products]
|
|
|
|
|
|
def _enrich_product_with_snapshot(db: Session, product: models.Product) -> dict:
|
|
"""Ajoute les données du dernier snapshot au produit."""
|
|
snapshot = get_latest_snapshot(db, product.id)
|
|
|
|
result = {
|
|
"id": product.id,
|
|
"boutique": product.boutique,
|
|
"url": str(product.url),
|
|
"asin": product.asin,
|
|
"titre": product.titre,
|
|
"url_image": str(product.url_image) if product.url_image else None,
|
|
"categorie": product.categorie,
|
|
"type": product.type,
|
|
"actif": product.actif,
|
|
"cree_le": product.cree_le,
|
|
"modifie_le": product.modifie_le,
|
|
}
|
|
|
|
if snapshot:
|
|
# Calcul de la réduction en pourcentage
|
|
reduction = None
|
|
if snapshot.prix_actuel and snapshot.prix_conseille:
|
|
reduction = round((1 - snapshot.prix_actuel / snapshot.prix_conseille) * 100)
|
|
|
|
# Désérialiser les champs JSON
|
|
a_propos = None
|
|
if snapshot.a_propos:
|
|
try:
|
|
a_propos = json.loads(snapshot.a_propos)
|
|
except json.JSONDecodeError:
|
|
a_propos = None
|
|
|
|
carateristique = None
|
|
if snapshot.carateristique:
|
|
try:
|
|
carateristique = json.loads(snapshot.carateristique)
|
|
except json.JSONDecodeError:
|
|
carateristique = None
|
|
|
|
details = None
|
|
if snapshot.details:
|
|
try:
|
|
details = json.loads(snapshot.details)
|
|
except json.JSONDecodeError:
|
|
details = None
|
|
|
|
result.update(
|
|
{
|
|
"prix_actuel": snapshot.prix_actuel,
|
|
"prix_conseille": snapshot.prix_conseille,
|
|
"prix_min_30j": snapshot.prix_min_30j,
|
|
"reduction_pourcent": reduction,
|
|
"etat_stock": snapshot.etat_stock,
|
|
"en_stock": snapshot.en_stock,
|
|
"note": snapshot.note,
|
|
"nombre_avis": snapshot.nombre_avis,
|
|
"prime": snapshot.prime,
|
|
"choix_amazon": snapshot.choix_amazon,
|
|
"offre_limitee": snapshot.offre_limitee,
|
|
"exclusivite_amazon": snapshot.exclusivite_amazon,
|
|
"a_propos": a_propos,
|
|
"description": snapshot.description,
|
|
"carateristique": carateristique,
|
|
"details": details,
|
|
"categorie_amazon": snapshot.categorie_amazon,
|
|
"dernier_scrape": snapshot.scrape_le,
|
|
"statut_scrap": snapshot.statut_scrap,
|
|
}
|
|
)
|
|
|
|
return result
|