398 lines
13 KiB
Python
398 lines
13 KiB
Python
import json
|
||
import re
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Protocol
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy.orm import selectinload
|
||
|
||
from app.extensions import db
|
||
|
||
from .models import TASK_PRIORITIES, TASK_STATUSES, Subtask, Task
|
||
|
||
|
||
BACKUP_SCHEMA_VERSION = 1
|
||
MAX_MARKDOWN_TASKS = 200
|
||
MAX_BACKUP_TASKS = 1000
|
||
MARKDOWN_TASK = re.compile(
|
||
r"^\s*(?:[-*+]\s+|\d+[.)]\s+)?"
|
||
r"(?:\[(?P<checked>[ xX])\]\s*)?(?P<title>.+?)\s*$"
|
||
)
|
||
|
||
|
||
class TransferValidationError(ValueError):
|
||
pass
|
||
|
||
|
||
class BackupUser(Protocol):
|
||
id: int
|
||
theme: str
|
||
completion_chime: bool
|
||
dyslexia_font: bool
|
||
notification_frequency: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ImportedTask:
|
||
title: str
|
||
status: str = "not_started"
|
||
|
||
|
||
def parse_markdown(content: bytes) -> list[ImportedTask]:
|
||
try:
|
||
text = content.decode("utf-8-sig")
|
||
except UnicodeDecodeError as exc:
|
||
raise TransferValidationError("The text file must use UTF-8 encoding.") from exc
|
||
|
||
tasks: list[ImportedTask] = []
|
||
in_code_block = False
|
||
for line_number, raw_line in enumerate(text.splitlines(), start=1):
|
||
line = raw_line.strip()
|
||
if line.startswith("```"):
|
||
in_code_block = not in_code_block
|
||
continue
|
||
if not line or in_code_block or line.startswith("#"):
|
||
continue
|
||
|
||
match = MARKDOWN_TASK.match(line)
|
||
if match is None:
|
||
continue
|
||
title = match.group("title").strip()
|
||
if not title:
|
||
continue
|
||
if len(title) > 200:
|
||
raise TransferValidationError(
|
||
f"Line {line_number} exceeds the 200-character task-title limit."
|
||
)
|
||
status = "done" if (match.group("checked") or "").lower() == "x" else "not_started"
|
||
tasks.append(ImportedTask(title=title, status=status))
|
||
if len(tasks) > MAX_MARKDOWN_TASKS:
|
||
raise TransferValidationError(
|
||
f"A single text import may contain at most {MAX_MARKDOWN_TASKS} tasks."
|
||
)
|
||
|
||
if not tasks:
|
||
raise TransferValidationError("No task lines were found in the text file.")
|
||
return tasks
|
||
|
||
|
||
def import_markdown(user_id: int, content: bytes) -> int:
|
||
imported = parse_markdown(content)
|
||
completed_at = datetime.now(timezone.utc)
|
||
tasks = [
|
||
Task(
|
||
user_id=user_id,
|
||
title=item.title,
|
||
status=item.status,
|
||
priority="normal",
|
||
completed_at=completed_at if item.status == "done" else None,
|
||
)
|
||
for item in imported
|
||
]
|
||
db.session.add_all(tasks)
|
||
try:
|
||
db.session.commit()
|
||
except SQLAlchemyError as exc:
|
||
db.session.rollback()
|
||
raise TransferValidationError(
|
||
"The task list could not be imported safely. No changes were saved."
|
||
) from exc
|
||
return len(tasks)
|
||
|
||
|
||
def export_backup(user: BackupUser) -> bytes:
|
||
tasks = list(
|
||
db.session.scalars(
|
||
select(Task)
|
||
.where(Task.user_id == user.id, Task.cleared_at.is_(None))
|
||
.options(selectinload(Task.subtasks))
|
||
.order_by(Task.created_at, Task.id)
|
||
)
|
||
)
|
||
document = {
|
||
"schema_version": BACKUP_SCHEMA_VERSION,
|
||
"exported_at": datetime.now(timezone.utc).isoformat(),
|
||
"settings": {
|
||
"theme": user.theme,
|
||
"completion_chime": user.completion_chime,
|
||
"dyslexia_font": user.dyslexia_font,
|
||
"notification_frequency": user.notification_frequency,
|
||
},
|
||
"tasks": [_serialize_task(task) for task in tasks],
|
||
}
|
||
return json.dumps(
|
||
document,
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
sort_keys=True,
|
||
).encode("utf-8")
|
||
|
||
|
||
def import_backup(
|
||
user: BackupUser,
|
||
content: bytes,
|
||
*,
|
||
replace_existing: bool,
|
||
) -> int:
|
||
document = _parse_backup(content)
|
||
tasks = [_build_task(user.id, item) for item in document["tasks"]]
|
||
|
||
if replace_existing:
|
||
existing_tasks = list(
|
||
db.session.scalars(
|
||
select(Task)
|
||
.where(Task.user_id == user.id)
|
||
.options(selectinload(Task.subtasks))
|
||
)
|
||
)
|
||
for task in existing_tasks:
|
||
db.session.delete(task)
|
||
|
||
_apply_settings(user, document["settings"])
|
||
db.session.add_all(tasks)
|
||
try:
|
||
db.session.commit()
|
||
except SQLAlchemyError as exc:
|
||
db.session.rollback()
|
||
raise TransferValidationError(
|
||
"The backup could not be restored safely. No changes were saved."
|
||
) from exc
|
||
return len(tasks)
|
||
|
||
|
||
def _serialize_task(task: Task) -> dict[str, Any]:
|
||
return {
|
||
"title": task.title,
|
||
"description": task.description,
|
||
"status": task.status,
|
||
"priority": task.priority,
|
||
"created_at": _isoformat(task.created_at),
|
||
"due_date": _isoformat(task.due_date),
|
||
"completed_at": _isoformat(task.completed_at),
|
||
"timer_seconds": task.timer_seconds,
|
||
"context": task.context,
|
||
"chunk_sessions": task.chunk_sessions,
|
||
"subtasks": [
|
||
{"title": subtask.title, "completed": subtask.completed}
|
||
for subtask in task.subtasks
|
||
],
|
||
}
|
||
|
||
|
||
def _parse_backup(content: bytes) -> dict[str, Any]:
|
||
try:
|
||
document = json.loads(content.decode("utf-8-sig"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||
raise TransferValidationError("The backup is not valid UTF-8 JSON.") from exc
|
||
if not isinstance(document, dict):
|
||
raise TransferValidationError("The backup root must be a JSON object.")
|
||
if document.get("schema_version") != BACKUP_SCHEMA_VERSION:
|
||
raise TransferValidationError(
|
||
f"Only backup schema version {BACKUP_SCHEMA_VERSION} is supported."
|
||
)
|
||
raw_tasks = document.get("tasks")
|
||
if not isinstance(raw_tasks, list):
|
||
raise TransferValidationError("The backup must contain a tasks list.")
|
||
if len(raw_tasks) > MAX_BACKUP_TASKS:
|
||
raise TransferValidationError(
|
||
f"A backup may contain at most {MAX_BACKUP_TASKS} tasks."
|
||
)
|
||
return {
|
||
"settings": _validate_settings(document.get("settings", {})),
|
||
"tasks": [
|
||
_validate_task(item, index)
|
||
for index, item in enumerate(raw_tasks, start=1)
|
||
],
|
||
}
|
||
|
||
|
||
def _validate_task(item: Any, index: int) -> dict[str, Any]:
|
||
if not isinstance(item, dict):
|
||
raise TransferValidationError(f"Task {index} must be a JSON object.")
|
||
title = _bounded_string(item.get("title"), f"Task {index} title", 200)
|
||
description = _optional_string(
|
||
item.get("description"), f"Task {index} description", 5000
|
||
)
|
||
status = item.get("status", "not_started")
|
||
priority = item.get("priority", "normal")
|
||
if status not in TASK_STATUSES:
|
||
raise TransferValidationError(f"Task {index} has an unknown status.")
|
||
if priority not in TASK_PRIORITIES:
|
||
raise TransferValidationError(f"Task {index} has an unknown priority.")
|
||
|
||
timer_seconds = item.get("timer_seconds", 0)
|
||
if isinstance(timer_seconds, bool) or not isinstance(timer_seconds, int):
|
||
raise TransferValidationError(f"Task {index} timer must be whole seconds.")
|
||
if timer_seconds < 0 or timer_seconds > 10_800:
|
||
raise TransferValidationError(
|
||
f"Task {index} timer is outside 0–10800 seconds."
|
||
)
|
||
|
||
raw_subtasks = item.get("subtasks", [])
|
||
if not isinstance(raw_subtasks, list) or len(raw_subtasks) > 200:
|
||
raise TransferValidationError(f"Task {index} has an invalid subtask list.")
|
||
subtasks = [
|
||
_validate_subtask(subtask, index, subtask_index)
|
||
for subtask_index, subtask in enumerate(raw_subtasks, start=1)
|
||
]
|
||
chunks = _validate_chunks(item.get("chunk_sessions", []), index)
|
||
created_at = _optional_datetime(
|
||
item.get("created_at"), f"Task {index} created_at"
|
||
)
|
||
due_date = _optional_datetime(item.get("due_date"), f"Task {index} due_date")
|
||
completed_at = _optional_datetime(
|
||
item.get("completed_at"), f"Task {index} completed_at"
|
||
)
|
||
if status == "done" and completed_at is None:
|
||
raise TransferValidationError(
|
||
f"Task {index} is done but has no completion date."
|
||
)
|
||
if status != "done" and completed_at is not None:
|
||
raise TransferValidationError(
|
||
f"Task {index} is active but has a completion date."
|
||
)
|
||
|
||
return {
|
||
"title": title,
|
||
"description": description,
|
||
"status": status,
|
||
"priority": priority,
|
||
"created_at": created_at,
|
||
"due_date": due_date,
|
||
"completed_at": completed_at,
|
||
"timer_seconds": timer_seconds,
|
||
"context": _optional_string(
|
||
item.get("context"), f"Task {index} context", 100
|
||
),
|
||
"chunk_sessions": chunks,
|
||
"subtasks": subtasks,
|
||
}
|
||
|
||
|
||
def _validate_subtask(item: Any, task_index: int, index: int) -> dict[str, Any]:
|
||
if not isinstance(item, dict):
|
||
raise TransferValidationError(
|
||
f"Subtask {index} of task {task_index} must be an object."
|
||
)
|
||
completed = item.get("completed", False)
|
||
if not isinstance(completed, bool):
|
||
raise TransferValidationError(
|
||
f"Subtask {index} of task {task_index} has invalid completion state."
|
||
)
|
||
return {
|
||
"title": _bounded_string(
|
||
item.get("title"),
|
||
f"Subtask {index} of task {task_index} title",
|
||
100,
|
||
),
|
||
"completed": completed,
|
||
}
|
||
|
||
|
||
def _validate_chunks(value: Any, task_index: int) -> list[dict[str, int | bool]]:
|
||
if not isinstance(value, list) or len(value) > 12:
|
||
raise TransferValidationError(
|
||
f"Task {task_index} has an invalid focus-block list."
|
||
)
|
||
chunks = []
|
||
for index, item in enumerate(value, start=1):
|
||
if not isinstance(item, dict):
|
||
raise TransferValidationError(
|
||
f"Focus block {index} of task {task_index} must be an object."
|
||
)
|
||
minutes = item.get("minutes")
|
||
completed = item.get("completed", False)
|
||
if (
|
||
isinstance(minutes, bool)
|
||
or not isinstance(minutes, int)
|
||
or minutes < 1
|
||
or minutes > 120
|
||
or not isinstance(completed, bool)
|
||
):
|
||
raise TransferValidationError(
|
||
f"Focus block {index} of task {task_index} is invalid."
|
||
)
|
||
chunks.append({"minutes": minutes, "completed": completed})
|
||
return chunks
|
||
|
||
|
||
def _validate_settings(value: Any) -> dict[str, Any]:
|
||
if not isinstance(value, dict):
|
||
raise TransferValidationError("Backup settings must be a JSON object.")
|
||
settings: dict[str, Any] = {}
|
||
if "theme" in value:
|
||
if value["theme"] not in {"auto", "light", "dark"}:
|
||
raise TransferValidationError("The backup contains an unknown theme.")
|
||
settings["theme"] = value["theme"]
|
||
for name in ("completion_chime", "dyslexia_font"):
|
||
if name in value:
|
||
if not isinstance(value[name], bool):
|
||
raise TransferValidationError(f"The {name} setting must be true or false.")
|
||
settings[name] = value[name]
|
||
if "notification_frequency" in value:
|
||
if value["notification_frequency"] not in {"quiet", "moderate", "alert"}:
|
||
raise TransferValidationError(
|
||
"The backup contains an unknown notification frequency."
|
||
)
|
||
settings["notification_frequency"] = value["notification_frequency"]
|
||
return settings
|
||
|
||
|
||
def _build_task(user_id: int, item: dict[str, Any]) -> Task:
|
||
task = Task(
|
||
user_id=user_id,
|
||
title=item["title"],
|
||
description=item["description"],
|
||
status=item["status"],
|
||
priority=item["priority"],
|
||
due_date=item["due_date"],
|
||
completed_at=item["completed_at"],
|
||
timer_seconds=item["timer_seconds"],
|
||
context=item["context"],
|
||
chunk_sessions=item["chunk_sessions"],
|
||
)
|
||
if item["created_at"] is not None:
|
||
task.created_at = item["created_at"]
|
||
task.subtasks = [Subtask(**subtask) for subtask in item["subtasks"]]
|
||
return task
|
||
|
||
|
||
def _apply_settings(user: BackupUser, settings: dict[str, Any]) -> None:
|
||
for name, value in settings.items():
|
||
setattr(user, name, value)
|
||
|
||
|
||
def _bounded_string(value: Any, label: str, maximum: int) -> str:
|
||
if not isinstance(value, str) or not value.strip():
|
||
raise TransferValidationError(f"{label} must be a non-empty string.")
|
||
cleaned = value.strip()
|
||
if len(cleaned) > maximum:
|
||
raise TransferValidationError(f"{label} exceeds {maximum} characters.")
|
||
return cleaned
|
||
|
||
|
||
def _optional_string(value: Any, label: str, maximum: int) -> str | None:
|
||
if value is None or value == "":
|
||
return None
|
||
if not isinstance(value, str):
|
||
raise TransferValidationError(f"{label} must be text or null.")
|
||
if len(value) > maximum:
|
||
raise TransferValidationError(f"{label} exceeds {maximum} characters.")
|
||
return value
|
||
|
||
|
||
def _optional_datetime(value: Any, label: str) -> datetime | None:
|
||
if value is None:
|
||
return None
|
||
if not isinstance(value, str):
|
||
raise TransferValidationError(f"{label} must be an ISO date-time or null.")
|
||
try:
|
||
return datetime.fromisoformat(value)
|
||
except ValueError as exc:
|
||
raise TransferValidationError(f"{label} is not a valid ISO date-time.") from exc
|
||
|
||
|
||
def _isoformat(value: datetime | None) -> str | None:
|
||
return value.isoformat() if value is not None else None
|