117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
"""
|
|
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}_"
|