Davide Scaini
284 lines
9.5 KiB
Python
284 lines
9.5 KiB
Python
"""Ideas and feedback endpoints (/api/ideas/*, /api/feedback)."""
|
|
from __future__ import annotations
|
|
|
|
import fcntl as _fcntl
|
|
import json
|
|
import secrets
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Cookie, File, Form, HTTPException, UploadFile
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from bincio.serve import deps
|
|
from bincio.serve.models import IdeaBody, IdeaCommentBody
|
|
|
|
router = APIRouter()
|
|
|
|
_FEEDBACK_IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".heic"}
|
|
_FEEDBACK_MAX_IMAGES = 3
|
|
_FEEDBACK_MAX_IMAGE_BYTES = 2 * 1024 * 1024 # 2 MB
|
|
|
|
|
|
def _ideas_dir(data_dir: Path) -> Path:
|
|
d = data_dir / "_ideas"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
@router.get("/api/ideas")
|
|
async def list_ideas(
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
dd = deps._get_data_dir()
|
|
ideas = []
|
|
for path in sorted(_ideas_dir(dd).glob("*.json")):
|
|
try:
|
|
idea = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
votes = idea.get("votes", [])
|
|
idea["vote_count"] = len(votes)
|
|
idea["my_vote"] = user.handle in votes
|
|
ideas.append(idea)
|
|
def _sort_key(x: dict):
|
|
s = x.get("status") or "open"
|
|
order = {"awaiting": 0, "open": 1, "done": 2, "declined": 3}
|
|
return (order.get(s, 1), -x["vote_count"], -x["created_at"])
|
|
ideas.sort(key=_sort_key)
|
|
return JSONResponse({"ideas": ideas})
|
|
|
|
|
|
@router.post("/api/ideas")
|
|
async def create_idea(
|
|
data: IdeaBody,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
title = data.title.strip()[:200]
|
|
body = data.body.strip()[:2000]
|
|
if not title:
|
|
raise HTTPException(400, "Title required")
|
|
dd = deps._get_data_dir()
|
|
idea_id = secrets.token_hex(8)
|
|
idea = {
|
|
"id": idea_id,
|
|
"title": title,
|
|
"body": body,
|
|
"author": user.handle,
|
|
"created_at": int(time.time()),
|
|
"votes": [],
|
|
}
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
path.write_text(json.dumps(idea, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return JSONResponse({"id": idea_id})
|
|
|
|
|
|
@router.post("/api/ideas/{idea_id}/vote")
|
|
async def toggle_idea_vote(
|
|
idea_id: str,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
dd = deps._get_data_dir()
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
if not path.exists():
|
|
raise HTTPException(404, "Not found")
|
|
with open(path, "r+", encoding="utf-8") as f:
|
|
_fcntl.flock(f, _fcntl.LOCK_EX)
|
|
idea = json.load(f)
|
|
votes: list = idea.get("votes", [])
|
|
if user.handle in votes:
|
|
votes.remove(user.handle)
|
|
voted = False
|
|
else:
|
|
votes.append(user.handle)
|
|
voted = True
|
|
idea["votes"] = votes
|
|
f.seek(0)
|
|
f.truncate()
|
|
json.dump(idea, f, ensure_ascii=False, indent=2)
|
|
return JSONResponse({"voted": voted, "votes": len(votes)})
|
|
|
|
|
|
@router.post("/api/ideas/{idea_id}/status")
|
|
async def toggle_idea_status(
|
|
idea_id: str,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
if not user.is_admin:
|
|
raise HTTPException(403, "Forbidden")
|
|
dd = deps._get_data_dir()
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
if not path.exists():
|
|
raise HTTPException(404, "Not found")
|
|
with open(path, "r+", encoding="utf-8") as f:
|
|
_fcntl.flock(f, _fcntl.LOCK_EX)
|
|
idea = json.load(f)
|
|
cycle = {"open": "awaiting", "awaiting": "done", "done": "open"}
|
|
idea["status"] = cycle.get(idea.get("status") or "open", "awaiting")
|
|
f.seek(0)
|
|
f.truncate()
|
|
json.dump(idea, f, ensure_ascii=False, indent=2)
|
|
return JSONResponse({"status": idea["status"]})
|
|
|
|
|
|
@router.post("/api/ideas/{idea_id}/decline")
|
|
async def decline_idea(
|
|
idea_id: str,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
if not user.is_admin:
|
|
raise HTTPException(403, "Forbidden")
|
|
dd = deps._get_data_dir()
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
if not path.exists():
|
|
raise HTTPException(404, "Not found")
|
|
with open(path, "r+", encoding="utf-8") as f:
|
|
_fcntl.flock(f, _fcntl.LOCK_EX)
|
|
idea = json.load(f)
|
|
idea["status"] = "open" if idea.get("status") == "declined" else "declined"
|
|
f.seek(0)
|
|
f.truncate()
|
|
json.dump(idea, f, ensure_ascii=False, indent=2)
|
|
return JSONResponse({"status": idea["status"]})
|
|
|
|
|
|
@router.post("/api/ideas/{idea_id}/comment")
|
|
async def set_idea_comment(
|
|
idea_id: str,
|
|
data: IdeaCommentBody,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
if not user.is_admin:
|
|
raise HTTPException(403, "Forbidden")
|
|
dd = deps._get_data_dir()
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
if not path.exists():
|
|
raise HTTPException(404, "Not found")
|
|
comment = data.comment.strip()[:1000]
|
|
with open(path, "r+", encoding="utf-8") as f:
|
|
_fcntl.flock(f, _fcntl.LOCK_EX)
|
|
idea = json.load(f)
|
|
if comment:
|
|
idea["admin_comment"] = comment
|
|
else:
|
|
idea.pop("admin_comment", None)
|
|
f.seek(0)
|
|
f.truncate()
|
|
json.dump(idea, f, ensure_ascii=False, indent=2)
|
|
return JSONResponse({"ok": True, "admin_comment": comment or None})
|
|
|
|
|
|
@router.patch("/api/ideas/{idea_id}")
|
|
async def edit_idea(
|
|
idea_id: str,
|
|
data: IdeaBody,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
dd = deps._get_data_dir()
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
if not path.exists():
|
|
raise HTTPException(404, "Not found")
|
|
title = data.title.strip()[:200]
|
|
body = data.body.strip()[:2000]
|
|
if not title:
|
|
raise HTTPException(400, "Title required")
|
|
with open(path, "r+", encoding="utf-8") as f:
|
|
_fcntl.flock(f, _fcntl.LOCK_EX)
|
|
idea = json.load(f)
|
|
if not user.is_admin and idea.get("author") != user.handle:
|
|
raise HTTPException(403, "Forbidden")
|
|
idea["title"] = title
|
|
idea["body"] = body
|
|
f.seek(0)
|
|
f.truncate()
|
|
json.dump(idea, f, ensure_ascii=False, indent=2)
|
|
return JSONResponse({"ok": True, "title": title, "body": body})
|
|
|
|
|
|
@router.delete("/api/ideas/{idea_id}")
|
|
async def delete_idea(
|
|
idea_id: str,
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
dd = deps._get_data_dir()
|
|
path = _ideas_dir(dd) / f"{idea_id}.json"
|
|
if not path.exists():
|
|
raise HTTPException(404, "Not found")
|
|
try:
|
|
idea = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
raise HTTPException(500, "Could not read idea")
|
|
if not user.is_admin and idea.get("author") != user.handle:
|
|
raise HTTPException(403, "Forbidden")
|
|
path.unlink()
|
|
return JSONResponse({"ok": True})
|
|
|
|
|
|
# ── Feedback ──────────────────────────────────────────────────────────────────
|
|
|
|
@router.post("/api/feedback")
|
|
async def submit_feedback(
|
|
text: str = Form(""),
|
|
images: list[UploadFile] = File(default=[]),
|
|
bincio_session: Optional[str] = Cookie(default=None),
|
|
) -> JSONResponse:
|
|
user = deps._require_user(bincio_session)
|
|
|
|
text = text.strip()
|
|
if not text and not any(f.filename for f in images):
|
|
raise HTTPException(400, "Feedback must include text or at least one image")
|
|
if len(images) > _FEEDBACK_MAX_IMAGES:
|
|
raise HTTPException(400, f"Maximum {_FEEDBACK_MAX_IMAGES} images per submission")
|
|
|
|
feedback_dir = deps._get_data_dir() / "_feedback"
|
|
feedback_dir.mkdir(exist_ok=True)
|
|
images_dir = feedback_dir / user.handle
|
|
images_dir.mkdir(exist_ok=True)
|
|
|
|
now = int(time.time())
|
|
submission_id = f"{now}_{secrets.token_hex(4)}"
|
|
saved_images: list[str] = []
|
|
|
|
for img in images:
|
|
if not img.filename:
|
|
continue
|
|
suffix = Path(img.filename).suffix.lower()
|
|
if suffix not in _FEEDBACK_IMAGE_SUFFIXES:
|
|
raise HTTPException(400, f"Unsupported image type '{suffix}'")
|
|
contents = await img.read()
|
|
if len(contents) > _FEEDBACK_MAX_IMAGE_BYTES:
|
|
raise HTTPException(413, f"Image '{img.filename}' exceeds 2 MB limit")
|
|
safe_name = f"{submission_id}_{Path(img.filename).name}"
|
|
(images_dir / safe_name).write_bytes(contents)
|
|
saved_images.append(safe_name)
|
|
|
|
from datetime import datetime, timezone
|
|
entry = {
|
|
"id": submission_id,
|
|
"handle": user.handle,
|
|
"submitted_at": datetime.now(timezone.utc).isoformat(),
|
|
"text": text,
|
|
"images": saved_images,
|
|
}
|
|
|
|
log_file = feedback_dir / f"{user.handle}.json"
|
|
existing: list[dict] = []
|
|
if log_file.exists():
|
|
try:
|
|
existing = json.loads(log_file.read_text())
|
|
except (OSError, json.JSONDecodeError):
|
|
existing = []
|
|
existing.append(entry)
|
|
log_file.write_text(json.dumps(existing, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
|
|
return JSONResponse({"ok": True, "id": submission_id})
|