import re from datetime import UTC, datetime from app.library.log import get_logger NAME_WHITESPACE_PATTERN = re.compile(r"\s+") LOG = get_logger() async def seed_defaults(repo) -> None: from .defaults import DEFAULT_PRESETS for preset in DEFAULT_PRESETS: try: name = preset.get("name") if not name or not isinstance(name, str): LOG.warning("Skipping default preset with invalid name: %s", preset) continue if not (name := preset_name(name)): LOG.warning("Skipping default preset with empty normalized name: %s", name) continue payload = {**preset} payload.pop("id", None) payload["default"] = True payload["name"] = name updated_at_value: datetime | None = None updated_at_raw = payload.get("updated_at") if isinstance(updated_at_raw, str): updated_at_value = datetime.fromisoformat(updated_at_raw) elif isinstance(updated_at_raw, datetime): updated_at_value = updated_at_raw if isinstance(updated_at_value, datetime) and updated_at_value.tzinfo is None: updated_at_value = updated_at_value.replace(tzinfo=UTC) payload["updated_at"] = updated_at_value existing = await repo.get_by_name(name) if None is existing: await repo.create(payload) continue if existing.updated_at and updated_at_value and existing.updated_at >= updated_at_value: if False is existing.default: await repo.update(existing.id, {"default": True}) continue await repo.update(existing.id, payload) except Exception as exc: LOG.exception( "Failed to seed default preset '%s'.", preset.get("name"), extra={"preset": preset.get("name"), "exception_type": type(exc).__name__}, ) def preset_name(value: str) -> str: return NAME_WHITESPACE_PATTERN.sub("_", value.strip().lower())