108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
# playlists/playlist_model.py
|
||
from pydantic import BaseModel, Field, validator, field_validator
|
||
from typing import List, Optional, Literal
|
||
from datetime import date, datetime
|
||
import json
|
||
import db
|
||
|
||
|
||
from typing import Optional, List, Literal
|
||
from pydantic import BaseModel, Field
|
||
import json
|
||
|
||
class RuleSet(BaseModel):
|
||
include_labels: List[str] = []
|
||
exclude_labels: List[str] = []
|
||
include_playlists: List[int] = []
|
||
exclude_playlists: List[int] = []
|
||
date_after: Optional[str] = None
|
||
date_before: Optional[str] = None
|
||
date_delta_days: Optional[int] = None
|
||
difficulty: Optional[str] = None
|
||
day_of_week: Optional[str] = None
|
||
address_keyword: Optional[str] = None
|
||
label_logic: Literal["AND", "OR"] = "AND"
|
||
logic: Literal["AND", "OR"] = "AND"
|
||
|
||
# --- Normalisation des dates (entrée) ---
|
||
@field_validator("date_after", "date_before", mode="before")
|
||
def normalize_date(cls, v):
|
||
"""Convertit automatiquement date/datetime en str ISO avant stockage."""
|
||
if isinstance(v, (date, datetime)):
|
||
return v.isoformat()
|
||
return v
|
||
|
||
def model_dump(self, *args, **kwargs):
|
||
"""Force la sortie JSON-safe (dates converties en str)."""
|
||
data = super().model_dump(*args, **kwargs)
|
||
for key in ["date_after", "date_before"]:
|
||
v = data.get(key)
|
||
if isinstance(v, (date, datetime)):
|
||
data[key] = v.isoformat()
|
||
return data
|
||
|
||
@validator("date_after", "date_before", pre=True, always=True)
|
||
def convert_date(cls, v):
|
||
"""Convertit automatiquement les objets date/datetime en ISO string."""
|
||
if isinstance(v, (date, datetime)):
|
||
return v.isoformat()
|
||
return v
|
||
|
||
def dict(self, *args, **kwargs):
|
||
"""S’assure que toutes les dates sont des chaînes sérialisables."""
|
||
data = super().dict(*args, **kwargs)
|
||
for key in ["date_after", "date_before"]:
|
||
v = data.get(key)
|
||
if isinstance(v, (date, datetime)):
|
||
data[key] = v.isoformat()
|
||
return data
|
||
|
||
def to_json(self) -> str:
|
||
return json.dumps(self.dict(), ensure_ascii=False, indent=2)
|
||
|
||
@classmethod
|
||
def from_json(cls, raw):
|
||
if not raw:
|
||
return cls()
|
||
if isinstance(raw, dict):
|
||
return cls(**raw)
|
||
try:
|
||
return cls(**json.loads(raw))
|
||
except Exception:
|
||
return cls()
|
||
|
||
|
||
class Playlist(BaseModel):
|
||
id: Optional[int] = None
|
||
name: str
|
||
description: Optional[str] = ""
|
||
type: Literal["manual", "dynamic"] = "manual"
|
||
rules: RuleSet = Field(default_factory=RuleSet)
|
||
created_at: Optional[str] = None
|
||
updated_at: Optional[str] = None
|
||
|
||
@validator("rules", pre=True, always=True)
|
||
def ensure_rules(cls, v):
|
||
"""Transforme la valeur de rules_json (None, str ou dict) en RuleSet."""
|
||
if isinstance(v, RuleSet):
|
||
return v
|
||
return RuleSet.from_json(v)
|
||
|
||
def save(self):
|
||
"""Insert or update playlist"""
|
||
with db.get_conn() as conn:
|
||
cur = conn.cursor()
|
||
if self.id:
|
||
cur.execute("""
|
||
UPDATE playlists
|
||
SET name=?, description=?, type=?, rules_json=?, updated_at=CURRENT_TIMESTAMP
|
||
WHERE id=?
|
||
""", (self.name, self.description, self.type, self.rules.to_json(), self.id))
|
||
else:
|
||
cur.execute("""
|
||
INSERT INTO playlists (name, description, type, rules_json)
|
||
VALUES (?, ?, ?, ?)
|
||
""", (self.name, self.description, self.type, self.rules.to_json()))
|
||
self.id = cur.lastrowid
|
||
conn.commit()
|