initial site
This commit is contained in:
merrow committed
commit
da5b259d06
167 files changed
+34157
No files matched your search
@@ -0,0 +1,227 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
НЕОН CRM — доски заметок/задач (HTTP-хендлеры).
|
||||
|
||||
Маршруты (все требуют X-Auth-Token):
|
||||
GET /api/boards — доступные доски
|
||||
POST /api/boards — приватная доска «владелец ↔ админ»
|
||||
PATCH /api/boards/<id> — переименовать (owner/admin)
|
||||
DELETE /api/boards/<id> — удалить (только owner)
|
||||
GET /api/boards/<id>/notes — заметки (?since= — только изменённые)
|
||||
POST /api/boards/<id>/notes — создать заметку
|
||||
GET /api/boards/<id>/activity — лента событий доски
|
||||
PATCH /api/notes/<id> — правка заметки
|
||||
DELETE /api/notes/<id> — удаление (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)})
|
||||
Reference in new issue
Block a user