Add suggestions feature: Gitea-backed feedback → issues

suggestions/gitea.py

@@ -0,0 +1,116 @@
+"""
+Gitea-backed suggestions module — framework-agnostic core.
+
+Files user-submitted suggestions as issues in a Gitea repo, mirroring the
+pattern used by Tony's `dms` and `fullmoonbakehouse` apps. No database of its
+own — Gitea issues ARE the storage.
+
+Configure via environment variables:
+    GITEA_URL    e.g. http://192.168.0.5:3022
+    GITEA_TOKEN  a Gitea access token with repo issue read/write scope
+    GITEA_REPO   "owner/repo", e.g. bonna61/moon-household-budget
+
+If any are unset, gitea_config() returns None and callers should degrade
+gracefully (show "feedback isn't set up yet" instead of crashing).
+
+Depends only on `requests` (add it to requirements.txt if not already there).
+"""
+from __future__ import annotations
+
+import os
+import secrets
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+import requests
+
+ALLOWED_IMAGE_TYPES = {
+    "image/jpeg", "image/png", "image/webp",
+    "image/gif", "image/heic", "image/heif",
+}
+MAX_UPLOAD_BYTES = 10 * 1024 * 1024  # 10 MB
+
+
+@dataclass
+class GiteaConfig:
+    base_url: str
+    token: str
+    repo: str  # "owner/repo"
+
+
+def gitea_config() -> Optional[GiteaConfig]:
+    base = os.environ.get("GITEA_URL")
+    token = os.environ.get("GITEA_TOKEN")
+    repo = os.environ.get("GITEA_REPO")
+    if not (base and token and repo):
+        return None
+    return GiteaConfig(base.rstrip("/"), token, repo)
+
+
+class GiteaError(RuntimeError):
+    pass
+
+
+class GiteaClient:
+    def __init__(self, cfg: GiteaConfig):
+        self.cfg = cfg
+
+    def _url(self, path: str) -> str:
+        return f"{self.cfg.base_url}/api/v1/repos/{self.cfg.repo}{path}"
+
+    def _req(self, method: str, path: str, **kw):
+        headers = {
+            "Authorization": f"token {self.cfg.token}",
+            "Content-Type": "application/json",
+        }
+        r = requests.request(method, self._url(path), headers=headers, timeout=15, **kw)
+        if not r.ok:
+            raise GiteaError(f"Gitea {r.status_code}: {r.text[:300]}")
+        return r.json()
+
+    def list_issues(self, state: str = "open"):
+        return self._req("GET", f"/issues?type=issues&state={state}&limit=50")
+
+    def create_issue(self, title: str, body: str):
+        return self._req("POST", "/issues", json={"title": title, "body": body})
+
+    def list_comments(self, number: int):
+        return self._req("GET", f"/issues/{number}/comments")
+
+    def create_comment(self, number: int, body: str):
+        return self._req("POST", f"/issues/{number}/comments", json={"body": body})
+
+
+def sniff_image(buf: bytes) -> Optional[str]:
+    """Return a MIME type by inspecting magic bytes, or None if unrecognized."""
+    if buf[:3] == b"\xff\xd8\xff":
+        return "image/jpeg"
+    if buf[:8] == b"\x89PNG\r\n\x1a\n":
+        return "image/png"
+    if buf[:4] == b"GIF8":
+        return "image/gif"
+    if buf[:4] == b"RIFF" and buf[8:12] == b"WEBP":
+        return "image/webp"
+    if buf[4:8] == b"ftyp":
+        brand = buf[8:12].decode("ascii", "ignore")
+        if brand in ("heic", "heix", "hevc", "hevx", "heim", "heis", "hevm", "hevs"):
+            return "image/heic"
+        if brand in ("mif1", "msf1"):
+            return "image/heif"
+    return None
+
+
+def save_upload(filename: str, data: bytes, upload_dir: Path, origin: str) -> str:
+    """Persist an upload and return an absolute URL for embedding in markdown."""
+    upload_dir.mkdir(parents=True, exist_ok=True)
+    ext = Path(filename).suffix.lower() or ".bin"
+    safe = f"{int(time.time() * 1000)}-{secrets.token_urlsafe(8)}{ext}"
+    (upload_dir / safe).write_bytes(data)
+    return f"{origin.rstrip('/')}/uploads/suggestions/{safe}"
+
+
+def footer(name: str, email: str, kind: str = "user") -> str:
+    """Attribution footer appended to a filed issue body (dms/fmb style)."""
+    return f"\n\n---\n_Filed by {name} ({email}) · {kind}_"