# -*- coding: utf-8 -*- """ НЕОН CRM — доски заметок/задач (HTTP-хендлеры). Маршруты (все требуют X-Auth-Token): GET /api/boards — доступные доски POST /api/boards — приватная доска «владелец ↔ админ» PATCH /api/boards/ — переименовать (owner/admin) DELETE /api/boards/ — удалить (только owner) GET /api/boards//notes — заметки (?since= — только изменённые) POST /api/boards//notes — создать заметку GET /api/boards//activity — лента событий доски PATCH /api/notes/ — правка заметки DELETE /api/notes/ — удаление (owner/admin/автор) Хендлеры собраны в NotesRoutes — миксин для Handler из server.py, чтобы точка интеграции маршрутов была одной строкой на метод. Доступ к доске проверяет storage.board_guard(): чужая приватная доска неотличима от несуществующей — оба случая отдаются как 404 {"error": "not found"}. """ import re from datetime import datetime from urllib.parse import parse_qs, urlsplit import storage from storage import clean_guest_text # NF-14: staff-тексты без NUL/bidi-мусора BOARD_ID_RE = re.compile(r"^/api/boards/(\d+)$") BOARD_NOTES_RE = re.compile(r"^/api/boards/(\d+)/notes$") BOARD_ACTIVITY_RE = re.compile(r"^/api/boards/(\d+)/activity$") NOTE_ID_RE = re.compile(r"^/api/notes/(\d+)$") NOTE_STATUSES = ("open", "in_progress", "done") # роли с правами администратора — как ADMIN_ROLES в server.py # (оттуда импортировать нельзя: циклический импорт) ADMIN_ROLES = ("owner", "admin") DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") TITLE_MAX = 200 BODY_MAX = 4000 ACTIVITY_LIMIT_MAX = 200 class NotesRoutes: """Хендлеры досок; подключаются к Handler в server.py.""" def _empty(self, code: int = 204): self._drain_body() # NF-13: 204 без чтения тела ломал бы keep-alive self.send_response(code) self.send_header("Content-Length", "0") self.end_headers() # ---------------- доски ---------------- def _handle_boards_get(self, user: dict): self._json({"boards": storage.boards_list(user)}) def _handle_board_create(self, user: dict): if user["role"] != "owner": self._json({"error": "owner only"}, 403) return body = self._read_body() or {} title = clean_guest_text(str(body.get("title") or ""), TITLE_MAX).strip() admin_login = str(body.get("admin_login") or "").strip() if not title: self._json({"error": "укажите название доски"}, 400) return users = {u["login"]: u for u in storage.list_users()} admin = users.get(admin_login) if admin is None or admin["role"] != "admin": self._json({"error": "выберите администратора — участника доски"}, 400) return board = storage.board_create(title, admin_login, user["login"]) self._json({"board": board}, 201) def _handle_board_patch(self, user: dict, board_id: int): board = storage.board_guard(board_id, user) if board is None: self._json({"error": "not found"}, 404) return if user["role"] not in ADMIN_ROLES: self._json({"error": "admin only"}, 403) return body = self._read_body() or {} title = clean_guest_text(str(body.get("title") or ""), TITLE_MAX).strip() if not title: self._json({"error": "укажите название доски"}, 400) return self._json({"board": storage.board_rename(board_id, title)}) def _handle_board_delete(self, user: dict, board_id: int): board = storage.board_guard(board_id, user) if board is None: self._json({"error": "not found"}, 404) return if user["role"] != "owner": self._json({"error": "owner only"}, 403) return if not storage.board_delete(board_id): self._json({"error": "not found"}, 404) return self._json({"ok": True}) # ---------------- заметки ---------------- def _handle_board_notes_get(self, user: dict, board_id: int): if storage.board_guard(board_id, user) is None: self._json({"error": "not found"}, 404) return since = None raw_since = (parse_qs(urlsplit(self.path).query).get("since") or [""])[0] if raw_since: # '+' в зоне вида +05:00 при разборе query превращается в пробел raw_since = raw_since.replace(" ", "+") try: since = datetime.fromisoformat(raw_since) except ValueError: self._json({"error": "bad since"}, 400) return self._json(storage.notes_list(board_id, since)) def _handle_note_create(self, user: dict, board_id: int): board = storage.board_guard(board_id, user) if board is None: self._json({"error": "not found"}, 404) return body = self._read_body() or {} title = clean_guest_text(str(body.get("title") or ""), TITLE_MAX).strip() if not title: self._json({"error": "укажите заголовок задачи"}, 400) return text = clean_guest_text(str(body.get("body") or ""), BODY_MAX) important = bool(body.get("important")) assignee = body.get("assignee") if assignee is not None: assignee = str(assignee).strip() or None if assignee and not storage.note_assignee_ok(board, assignee): self._json({"error": "исполнитель — не участник этой доски"}, 400) return due = None if body.get("due_date"): if not DATE_RE.match(str(body["due_date"])): self._json({"error": "срок в формате ГГГГ-ММ-ДД"}, 400) return due = body["due_date"] note = storage.note_create(board, user, title, text, assignee, important, due) self._json({"note": note}, 201) def _handle_note_patch(self, user: dict, note_id: int): note = storage.note_get(note_id) board = storage.board_guard(note["board_id"], user) if note else None if note is None or board is None: self._json({"error": "not found"}, 404) return body = self._read_body() or {} if not body: self._json({"error": "нечего изменять"}, 400) return fields: dict = {} if "title" in body: title = clean_guest_text(str(body["title"] or ""), TITLE_MAX).strip() if not title: self._json({"error": "укажите заголовок задачи"}, 400) return fields["title"] = title if "body" in body: fields["body"] = clean_guest_text(str(body["body"] or ""), BODY_MAX) if "important" in body: if not isinstance(body["important"], bool): self._json({"error": "important (bool) required"}, 400) return fields["important"] = body["important"] if "status" in body: if body["status"] not in NOTE_STATUSES: self._json({"error": "статус: open, in_progress или done"}, 400) return fields["status"] = body["status"] if "due_date" in body: if body["due_date"] is None: fields["due_date"] = None elif DATE_RE.match(str(body["due_date"])): fields["due_date"] = str(body["due_date"]) else: self._json({"error": "срок в формате ГГГГ-ММ-ДД"}, 400) return if "assignee" in body: assignee = body["assignee"] if assignee is not None: assignee = str(assignee).strip() or None if assignee and not storage.note_assignee_ok(board, assignee): self._json({"error": "исполнитель — не участник этой доски"}, 400) return fields["assignee_login"] = assignee updated = storage.note_update(note_id, user, board, fields) if updated is None: self._json({"error": "not found"}, 404) return self._json({"note": updated}) def _handle_note_delete(self, user: dict, note_id: int): note = storage.note_get(note_id) board = storage.board_guard(note["board_id"], user) if note else None if note is None or board is None: self._json({"error": "not found"}, 404) return if user["role"] not in ADMIN_ROLES and note["author"]["id"] != user["login"]: self._json({"error": "удалять может автор, админ или владелец"}, 403) return if storage.note_delete(note_id, user["login"]) is None: self._json({"error": "not found"}, 404) return self._empty(204) # ---------------- журнал ---------------- def _handle_board_activity(self, user: dict, board_id: int): if storage.board_guard(board_id, user) is None: self._json({"error": "not found"}, 404) return qs = parse_qs(urlsplit(self.path).query) try: limit = min(ACTIVITY_LIMIT_MAX, max(1, int((qs.get("limit") or ["50"])[0]))) except ValueError: limit = 50 self._json({"events": storage.board_activity(board_id, limit)})