Étape 3 : import CSV/Excel des bénévoles et disponibilités
- Fichier unique, processus en deux temps : POST /editions/{id}/import/analyze
(colonnes détectées, aperçu, mapping suggéré, demi-journées de l'édition)
puis POST /editions/{id}/import/execute (mapping validé)
- Parser CSV (délimiteur et encodage devinés, BOM) et XLSX (openpyxl),
détection XLSX par nombre magique
- Vocabulaire des valeurs de disponibilité (tri-état, accents/casse) et des
tailles de t-shirt ; compétences créées à la volée dans le référentiel
- Rapport ligne par ligne : codes de traduction + paramètres, jamais de
phrase. Doublons : upsert sur email, signalement sur (nom, prénom) sans
email, doublons intra-fichier. Mode simulation (rollback, rien de persisté)
- 40 tests : parser, valeurs, service (tous les cas), endpoints
- Dépendances : python-multipart, openpyxl
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
This commit is contained in:
AdamThomas
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
cb2e84adf0
commit
bb919554ff
@@ -0,0 +1,6 @@
|
||||
"""Import CSV/Excel des bénévoles et de leurs disponibilités.
|
||||
|
||||
Fichier unique (une ligne par bénévole, disponibilités en colonnes). Processus
|
||||
en deux temps : `analyze` (colonnes détectées + aperçu) puis `execute` (mapping
|
||||
validé). Voir CLAUDE.md § 4.2.
|
||||
"""
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Lecture d'un fichier importé (CSV ou XLSX) vers une table normalisée."""
|
||||
|
||||
import csv
|
||||
import io
|
||||
from dataclasses import dataclass
|
||||
|
||||
from openpyxl import load_workbook
|
||||
|
||||
_ENCODINGS = ("utf-8-sig", "utf-8", "cp1252", "latin-1")
|
||||
_DELIMITERS = (";", ",", "\t")
|
||||
_XLSX_MAGIC = b"PK\x03\x04"
|
||||
|
||||
|
||||
class ImportParseError(Exception):
|
||||
"""Fichier illisible. `code` est une clé de traduction."""
|
||||
|
||||
def __init__(self, code: str, **params: object) -> None:
|
||||
super().__init__(code)
|
||||
self.code = code
|
||||
self.params = params
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedTable:
|
||||
headers: list[str]
|
||||
rows: list[list[str]]
|
||||
|
||||
def as_dicts(self) -> list[dict[str, str]]:
|
||||
return [
|
||||
{h: v for h, v in zip(self.headers, row, strict=False)}
|
||||
for row in self.rows
|
||||
]
|
||||
|
||||
|
||||
def parse_upload(filename: str, content: bytes) -> ParsedTable:
|
||||
if not content:
|
||||
raise ImportParseError("import.error.empty_file")
|
||||
name = (filename or "").lower()
|
||||
if content[:4] == _XLSX_MAGIC or name.endswith((".xlsx", ".xlsm")):
|
||||
return _parse_xlsx(content)
|
||||
return _parse_csv(content)
|
||||
|
||||
|
||||
def _decode(content: bytes) -> str:
|
||||
for enc in _ENCODINGS:
|
||||
try:
|
||||
return content.decode(enc)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
raise ImportParseError("import.error.encoding")
|
||||
|
||||
|
||||
def _sniff_delimiter(sample: str) -> str:
|
||||
lines = [ln for ln in sample.splitlines() if ln.strip()]
|
||||
header = lines[0] if lines else ""
|
||||
counts = {d: header.count(d) for d in _DELIMITERS}
|
||||
best = max(counts, key=lambda d: counts[d])
|
||||
return best if counts[best] else ","
|
||||
|
||||
|
||||
def _fit(row: list[str], width: int) -> list[str]:
|
||||
if len(row) < width:
|
||||
return row + [""] * (width - len(row))
|
||||
return row[:width]
|
||||
|
||||
|
||||
def _parse_csv(content: bytes) -> ParsedTable:
|
||||
text = _decode(content)
|
||||
delimiter = _sniff_delimiter(text[:4096])
|
||||
reader = csv.reader(io.StringIO(text), delimiter=delimiter)
|
||||
rows = [[(cell or "").strip() for cell in row] for row in reader]
|
||||
rows = [row for row in rows if any(row)]
|
||||
if not rows:
|
||||
raise ImportParseError("import.error.empty_file")
|
||||
headers = rows[0]
|
||||
width = len(headers)
|
||||
return ParsedTable(headers=headers, rows=[_fit(r, width) for r in rows[1:]])
|
||||
|
||||
|
||||
def _cell_to_str(value: object) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, str):
|
||||
return value.strip()
|
||||
if isinstance(value, float) and value.is_integer():
|
||||
return str(int(value))
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def _parse_xlsx(content: bytes) -> ParsedTable:
|
||||
try:
|
||||
workbook = load_workbook(
|
||||
io.BytesIO(content), read_only=True, data_only=True
|
||||
)
|
||||
except Exception as exc:
|
||||
raise ImportParseError("import.error.unreadable_file") from exc
|
||||
worksheet = workbook.active
|
||||
if worksheet is None:
|
||||
workbook.close()
|
||||
raise ImportParseError("import.error.empty_file")
|
||||
rows = [
|
||||
[_cell_to_str(cell) for cell in row]
|
||||
for row in worksheet.iter_rows(values_only=True)
|
||||
]
|
||||
workbook.close()
|
||||
rows = [row for row in rows if any(row)]
|
||||
if not rows:
|
||||
raise ImportParseError("import.error.empty_file")
|
||||
headers = rows[0]
|
||||
width = len(headers)
|
||||
return ParsedTable(headers=headers, rows=[_fit(r, width) for r in rows[1:]])
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Schémas d'entrée/sortie de l'import.
|
||||
|
||||
Le rapport ne contient que des **codes** de traduction + des paramètres
|
||||
structurés, jamais de phrase (voir `locales/fr.json`).
|
||||
"""
|
||||
|
||||
import datetime as dt
|
||||
import uuid
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
Severity = Literal["ERROR", "WARNING"]
|
||||
RowAction = Literal["CREATED", "UPDATED", "SKIPPED"]
|
||||
|
||||
|
||||
class ColumnMapping(BaseModel):
|
||||
first_name: str
|
||||
last_name: str
|
||||
email: str | None = None
|
||||
phone: str | None = None
|
||||
tshirt_size: str | None = None
|
||||
diet: str | None = None
|
||||
comment: str | None = None
|
||||
skills: str | None = None
|
||||
|
||||
|
||||
class AvailabilityColumnMapping(BaseModel):
|
||||
header: str
|
||||
date: dt.date
|
||||
day_part_code: str
|
||||
|
||||
|
||||
class ImportMapping(BaseModel):
|
||||
dry_run: bool = True
|
||||
columns: ColumnMapping
|
||||
availability_columns: list[AvailabilityColumnMapping] = Field(
|
||||
default_factory=list
|
||||
)
|
||||
|
||||
|
||||
class Issue(BaseModel):
|
||||
code: str
|
||||
severity: Severity
|
||||
field: str | None = None
|
||||
params: dict[str, str | int | None] = Field(default_factory=dict)
|
||||
|
||||
|
||||
class RowReport(BaseModel):
|
||||
line: int
|
||||
action: RowAction
|
||||
volunteer_id: uuid.UUID | None = None
|
||||
issues: list[Issue] = Field(default_factory=list)
|
||||
|
||||
|
||||
class ImportReport(BaseModel):
|
||||
dry_run: bool
|
||||
total_rows: int
|
||||
created: int = 0
|
||||
updated: int = 0
|
||||
skipped: int = 0
|
||||
availabilities_written: int = 0
|
||||
skills_created: list[str] = Field(default_factory=list)
|
||||
rows: list[RowReport] = Field(default_factory=list)
|
||||
|
||||
|
||||
class PreviewRow(BaseModel):
|
||||
line: int
|
||||
values: dict[str, str]
|
||||
|
||||
|
||||
class DayPartInfo(BaseModel):
|
||||
code: str
|
||||
label: str
|
||||
start_time: dt.time
|
||||
end_time: dt.time
|
||||
order: int
|
||||
|
||||
|
||||
class AnalyzeResponse(BaseModel):
|
||||
filename: str
|
||||
format: Literal["csv", "xlsx"]
|
||||
headers: list[str]
|
||||
row_count: int
|
||||
preview: list[PreviewRow]
|
||||
suggested_mapping: dict[str, str]
|
||||
edition_dates: list[dt.date]
|
||||
day_parts: list[DayPartInfo]
|
||||
@@ -0,0 +1,395 @@
|
||||
"""Cœur de l'import : suggestion de mapping, validation, exécution."""
|
||||
|
||||
import re
|
||||
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.importing.parser import ParsedTable
|
||||
from app.importing.schemas import (
|
||||
ImportMapping,
|
||||
ImportReport,
|
||||
Issue,
|
||||
RowReport,
|
||||
)
|
||||
from app.importing.text import normalize, slug_code, split_skills
|
||||
from app.importing.values import parse_availability_cell, parse_tshirt_size
|
||||
from app.models import (
|
||||
Availability,
|
||||
DayPart,
|
||||
Edition,
|
||||
Skill,
|
||||
Volunteer,
|
||||
VolunteerSkill,
|
||||
)
|
||||
|
||||
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||||
|
||||
_FIELD_HINTS: dict[str, tuple[str, ...]] = {
|
||||
"email": ("mail", "courriel"),
|
||||
"first_name": ("prenom", "first name", "firstname", "given name"),
|
||||
"last_name": ("nom", "last name", "lastname", "surname", "famille"),
|
||||
"phone": ("tel", "phone", "portable", "mobile", "gsm"),
|
||||
"tshirt_size": ("tshirt", "t shirt", "taille", "shirt"),
|
||||
"diet": ("regime", "alimentaire", "diet", "allergie"),
|
||||
"comment": ("commentaire", "remarque", "note", "precision"),
|
||||
"skills": ("competence", "skill", "habilitation", "brevet", "diplome"),
|
||||
}
|
||||
|
||||
|
||||
def suggest_mapping(headers: list[str]) -> dict[str, str]:
|
||||
"""Association approximative en-tête → champ, à confirmer côté client."""
|
||||
result: dict[str, str] = {}
|
||||
used: set[str] = set()
|
||||
for field, hints in _FIELD_HINTS.items():
|
||||
for header in headers:
|
||||
if header in used:
|
||||
continue
|
||||
if any(hint in normalize(header) for hint in hints):
|
||||
result[field] = header
|
||||
used.add(header)
|
||||
break
|
||||
return result
|
||||
|
||||
|
||||
def validate_mapping(
|
||||
table: ParsedTable,
|
||||
mapping: ImportMapping,
|
||||
edition: Edition,
|
||||
day_parts: list[DayPart],
|
||||
) -> list[Issue]:
|
||||
"""Contrôles préalables. Retourne une liste vide si le mapping est correct."""
|
||||
errors: list[Issue] = []
|
||||
headers = set(table.headers)
|
||||
columns = mapping.columns
|
||||
|
||||
for field in ("first_name", "last_name"):
|
||||
header = getattr(columns, field)
|
||||
if not header:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.required_field_missing",
|
||||
severity="ERROR",
|
||||
field=field,
|
||||
)
|
||||
)
|
||||
elif header not in headers:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.column_not_found",
|
||||
severity="ERROR",
|
||||
field=field,
|
||||
params={"column": header},
|
||||
)
|
||||
)
|
||||
|
||||
for field in ("email", "phone", "tshirt_size", "diet", "comment", "skills"):
|
||||
header = getattr(columns, field)
|
||||
if header and header not in headers:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.column_not_found",
|
||||
severity="ERROR",
|
||||
field=field,
|
||||
params={"column": header},
|
||||
)
|
||||
)
|
||||
|
||||
valid_codes = {dp.code for dp in day_parts}
|
||||
seen: set[str] = set()
|
||||
for column in mapping.availability_columns:
|
||||
if column.header not in headers:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.column_not_found",
|
||||
severity="ERROR",
|
||||
field="availability",
|
||||
params={"column": column.header},
|
||||
)
|
||||
)
|
||||
if column.header in seen:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.duplicate_availability_column",
|
||||
severity="ERROR",
|
||||
params={"column": column.header},
|
||||
)
|
||||
)
|
||||
seen.add(column.header)
|
||||
if column.day_part_code not in valid_codes:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.unknown_day_part",
|
||||
severity="ERROR",
|
||||
params={
|
||||
"column": column.header,
|
||||
"day_part_code": column.day_part_code,
|
||||
},
|
||||
)
|
||||
)
|
||||
if not edition.start_date <= column.date <= edition.end_date:
|
||||
errors.append(
|
||||
Issue(
|
||||
code="import.mapping.date_outside_edition",
|
||||
severity="ERROR",
|
||||
params={
|
||||
"column": column.header,
|
||||
"date": column.date.isoformat(),
|
||||
},
|
||||
)
|
||||
)
|
||||
return errors
|
||||
|
||||
|
||||
def run_import(
|
||||
*,
|
||||
session: Session,
|
||||
edition: Edition,
|
||||
table: ParsedTable,
|
||||
mapping: ImportMapping,
|
||||
) -> ImportReport:
|
||||
"""Exécute l'import. `mapping.dry_run` → aucune persistance (rollback final).
|
||||
|
||||
Le mapping est supposé déjà validé par `validate_mapping`.
|
||||
"""
|
||||
columns = mapping.columns
|
||||
day_part_by_code = {
|
||||
dp.code: dp
|
||||
for dp in session.exec(
|
||||
select(DayPart).where(DayPart.edition_id == edition.id)
|
||||
)
|
||||
}
|
||||
|
||||
existing = list(
|
||||
session.exec(
|
||||
select(Volunteer).where(Volunteer.edition_id == edition.id)
|
||||
)
|
||||
)
|
||||
by_email = {v.email.lower(): v for v in existing if v.email}
|
||||
by_name: dict[tuple[str, str], Volunteer] = {}
|
||||
for volunteer in existing:
|
||||
if not volunteer.email:
|
||||
key = (normalize(volunteer.last_name), normalize(volunteer.first_name))
|
||||
by_name.setdefault(key, volunteer)
|
||||
|
||||
skills_by_code = {
|
||||
s.code: s
|
||||
for s in session.exec(
|
||||
select(Skill).where(Skill.edition_id == edition.id)
|
||||
)
|
||||
}
|
||||
created_skills: set[str] = set()
|
||||
|
||||
report = ImportReport(dry_run=mapping.dry_run, total_rows=len(table.rows))
|
||||
email_seen: dict[str, int] = {}
|
||||
name_seen: dict[tuple[str, str], int] = {}
|
||||
|
||||
for index, row in enumerate(table.as_dicts()):
|
||||
line = index + 2 # +1 en-tête, +1 pour une numérotation à partir de 1
|
||||
entry = RowReport(line=line, action="SKIPPED")
|
||||
|
||||
first_name = (row.get(columns.first_name) or "").strip()
|
||||
last_name = (row.get(columns.last_name) or "").strip()
|
||||
email_raw = (
|
||||
(row.get(columns.email) or "").strip() if columns.email else ""
|
||||
)
|
||||
email = email_raw.lower() or None
|
||||
|
||||
if not first_name or not last_name:
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.missing_required_field",
|
||||
severity="ERROR",
|
||||
field="first_name" if not first_name else "last_name",
|
||||
)
|
||||
)
|
||||
report.rows.append(entry)
|
||||
report.skipped += 1
|
||||
continue
|
||||
|
||||
if email and not _EMAIL_RE.match(email):
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.invalid_email",
|
||||
severity="ERROR",
|
||||
field="email",
|
||||
params={"value": email_raw},
|
||||
)
|
||||
)
|
||||
report.rows.append(entry)
|
||||
report.skipped += 1
|
||||
continue
|
||||
|
||||
if email:
|
||||
if email in email_seen:
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.duplicate_email_in_file",
|
||||
severity="ERROR",
|
||||
field="email",
|
||||
params={"value": email_raw, "first_line": email_seen[email]},
|
||||
)
|
||||
)
|
||||
report.rows.append(entry)
|
||||
report.skipped += 1
|
||||
continue
|
||||
email_seen[email] = line
|
||||
else:
|
||||
key = (normalize(last_name), normalize(first_name))
|
||||
if key in name_seen:
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.duplicate_name_in_file",
|
||||
severity="WARNING",
|
||||
params={"first_line": name_seen[key]},
|
||||
)
|
||||
)
|
||||
else:
|
||||
name_seen[key] = line
|
||||
|
||||
volunteer = by_email.get(email) if email else None
|
||||
if volunteer is not None:
|
||||
entry.action = "UPDATED"
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.updated_existing",
|
||||
severity="WARNING",
|
||||
params={"volunteer_id": str(volunteer.id)},
|
||||
)
|
||||
)
|
||||
else:
|
||||
if not email:
|
||||
match = by_name.get((normalize(last_name), normalize(first_name)))
|
||||
if match is not None:
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.possible_duplicate_name",
|
||||
severity="WARNING",
|
||||
params={"volunteer_id": str(match.id)},
|
||||
)
|
||||
)
|
||||
volunteer = Volunteer(
|
||||
edition_id=edition.id,
|
||||
first_name=first_name,
|
||||
last_name=last_name,
|
||||
)
|
||||
session.add(volunteer)
|
||||
entry.action = "CREATED"
|
||||
|
||||
volunteer.first_name = first_name
|
||||
volunteer.last_name = last_name
|
||||
if columns.email:
|
||||
volunteer.email = email_raw or None
|
||||
if columns.phone:
|
||||
volunteer.phone = (row.get(columns.phone) or "").strip() or None
|
||||
if columns.diet:
|
||||
volunteer.diet = (row.get(columns.diet) or "").strip() or None
|
||||
if columns.comment:
|
||||
volunteer.comment = (row.get(columns.comment) or "").strip() or None
|
||||
if columns.tshirt_size:
|
||||
raw_size = (row.get(columns.tshirt_size) or "").strip()
|
||||
if raw_size:
|
||||
size = parse_tshirt_size(raw_size)
|
||||
if size is None:
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.invalid_tshirt_size",
|
||||
severity="WARNING",
|
||||
field="tshirt_size",
|
||||
params={"value": raw_size},
|
||||
)
|
||||
)
|
||||
else:
|
||||
volunteer.tshirt_size = size
|
||||
|
||||
session.flush()
|
||||
|
||||
if columns.skills:
|
||||
linked = set(
|
||||
session.exec(
|
||||
select(VolunteerSkill.skill_id).where(
|
||||
VolunteerSkill.volunteer_id == volunteer.id
|
||||
)
|
||||
)
|
||||
)
|
||||
for token in split_skills(row.get(columns.skills) or ""):
|
||||
code = slug_code(token)
|
||||
if not code:
|
||||
continue
|
||||
skill = skills_by_code.get(code)
|
||||
if skill is None:
|
||||
skill = Skill(
|
||||
edition_id=edition.id, code=code, label=token.strip()
|
||||
)
|
||||
session.add(skill)
|
||||
session.flush()
|
||||
skills_by_code[code] = skill
|
||||
created_skills.add(code)
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.skill_created",
|
||||
severity="WARNING",
|
||||
params={"code": code, "label": skill.label},
|
||||
)
|
||||
)
|
||||
if skill.id not in linked:
|
||||
session.add(
|
||||
VolunteerSkill(
|
||||
volunteer_id=volunteer.id, skill_id=skill.id
|
||||
)
|
||||
)
|
||||
linked.add(skill.id)
|
||||
|
||||
for column in mapping.availability_columns:
|
||||
status, recognized = parse_availability_cell(
|
||||
row.get(column.header, "") or ""
|
||||
)
|
||||
if not recognized:
|
||||
entry.issues.append(
|
||||
Issue(
|
||||
code="import.row.unknown_availability_value",
|
||||
severity="WARNING",
|
||||
field=column.header,
|
||||
params={
|
||||
"column": column.header,
|
||||
"value": (row.get(column.header, "") or "").strip(),
|
||||
},
|
||||
)
|
||||
)
|
||||
continue
|
||||
if status is None:
|
||||
continue
|
||||
day_part = day_part_by_code[column.day_part_code]
|
||||
current = session.exec(
|
||||
select(Availability)
|
||||
.where(Availability.volunteer_id == volunteer.id)
|
||||
.where(Availability.date == column.date)
|
||||
.where(Availability.day_part_id == day_part.id)
|
||||
).first()
|
||||
if current is not None:
|
||||
current.status = status
|
||||
else:
|
||||
session.add(
|
||||
Availability(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
date=column.date,
|
||||
day_part_id=day_part.id,
|
||||
status=status,
|
||||
)
|
||||
)
|
||||
report.availabilities_written += 1
|
||||
|
||||
entry.volunteer_id = volunteer.id
|
||||
report.rows.append(entry)
|
||||
if entry.action == "CREATED":
|
||||
report.created += 1
|
||||
else:
|
||||
report.updated += 1
|
||||
|
||||
report.skills_created = sorted(created_skills)
|
||||
|
||||
if mapping.dry_run:
|
||||
session.rollback()
|
||||
else:
|
||||
session.commit()
|
||||
return report
|
||||
@@ -0,0 +1,26 @@
|
||||
"""Normalisation de texte partagée par l'import (accents, casse, découpage)."""
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
|
||||
_SKILL_SEPARATORS = re.compile(r"[;,/\n]+")
|
||||
_MULTI_UNDERSCORE = re.compile(r"_+")
|
||||
|
||||
|
||||
def normalize(value: str) -> str:
|
||||
"""Minuscule, sans accents, espaces réduits."""
|
||||
lowered = unicodedata.normalize("NFKD", (value or "").strip().lower())
|
||||
stripped = "".join(c for c in lowered if not unicodedata.combining(c))
|
||||
return " ".join(stripped.split())
|
||||
|
||||
|
||||
def slug_code(value: str) -> str:
|
||||
"""Code stable pour une compétence : MAJUSCULES, séparateurs en `_`."""
|
||||
base = normalize(value).upper()
|
||||
chars = [c if c.isalnum() else "_" for c in base]
|
||||
return _MULTI_UNDERSCORE.sub("_", "".join(chars)).strip("_")
|
||||
|
||||
|
||||
def split_skills(value: str) -> list[str]:
|
||||
parts = (p.strip() for p in _SKILL_SEPARATORS.split(value or ""))
|
||||
return [p for p in parts if p]
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Interprétation des valeurs de cellule : disponibilités et tailles de t-shirt."""
|
||||
|
||||
from app.importing.text import normalize
|
||||
from app.models import AvailabilityStatus, TshirtSize
|
||||
|
||||
_AVAILABLE = {
|
||||
"x", "o", "oui", "y", "yes", "ok", "1", "vrai", "true",
|
||||
"dispo", "disponible", "present", "presente",
|
||||
"matin", "aprem", "apres-midi", "apres midi", "midi", "soir", "soiree",
|
||||
"journee", "jour",
|
||||
}
|
||||
_IF_NEEDED = {
|
||||
"si besoin", "sib", "peut-etre", "peut etre", "?", "2",
|
||||
"backup", "reserve", "au besoin",
|
||||
}
|
||||
_UNAVAILABLE = {
|
||||
"non", "n", "no", "0", "faux", "false", "indispo", "indisponible", "absent",
|
||||
"-", "x-",
|
||||
}
|
||||
_EMPTY = {"", "na", "n/a", "nc"}
|
||||
|
||||
_TSHIRT = {
|
||||
"XS": TshirtSize.XS,
|
||||
"S": TshirtSize.S,
|
||||
"M": TshirtSize.M,
|
||||
"L": TshirtSize.L,
|
||||
"XL": TshirtSize.XL,
|
||||
"XXL": TshirtSize.XXL,
|
||||
"2XL": TshirtSize.XXL,
|
||||
"XXXL": TshirtSize.XXXL,
|
||||
"3XL": TshirtSize.XXXL,
|
||||
}
|
||||
|
||||
|
||||
def parse_availability_cell(value: str) -> tuple[AvailabilityStatus | None, bool]:
|
||||
"""(statut, reconnu).
|
||||
|
||||
- statut `None` + reconnu `True` → cellule vide, pas de déclaration.
|
||||
- statut `None` + reconnu `False` → valeur non comprise, à signaler.
|
||||
"""
|
||||
norm = normalize(value)
|
||||
if norm in _EMPTY:
|
||||
return None, True
|
||||
if norm in _AVAILABLE:
|
||||
return AvailabilityStatus.AVAILABLE, True
|
||||
if norm in _IF_NEEDED:
|
||||
return AvailabilityStatus.IF_NEEDED, True
|
||||
if norm in _UNAVAILABLE:
|
||||
return AvailabilityStatus.UNAVAILABLE, True
|
||||
return None, False
|
||||
|
||||
|
||||
def parse_tshirt_size(value: str) -> TshirtSize | None:
|
||||
key = normalize(value).upper().replace(" ", "").replace("-", "")
|
||||
return _TSHIRT.get(key)
|
||||
Reference in New Issue
Block a user