147 lines
4.7 KiB
Python
147 lines
4.7 KiB
Python
"""
|
|
Flask adapter for the Gitea-backed suggestions module.
|
|
|
|
Wire it into the app with:
|
|
|
|
from suggestions.routes_flask import suggestions_bp
|
|
app.register_blueprint(suggestions_bp)
|
|
|
|
Then a logged-in user can open /suggestions to file feedback. The blueprint
|
|
exposes both the HTML page and the JSON API the page's JS talks to.
|
|
|
|
AUTH: replace `current_user()` below with however this app identifies the
|
|
logged-in person. It must return an object with `.name` and `.email`, or None.
|
|
If the app has no auth at all, you can return a fixed identity (see the stub).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from flask import Blueprint, jsonify, render_template, request
|
|
|
|
from .gitea import (
|
|
ALLOWED_IMAGE_TYPES,
|
|
MAX_UPLOAD_BYTES,
|
|
GiteaClient,
|
|
GiteaError,
|
|
footer,
|
|
gitea_config,
|
|
save_upload,
|
|
sniff_image,
|
|
)
|
|
|
|
suggestions_bp = Blueprint(
|
|
"suggestions",
|
|
__name__,
|
|
template_folder="templates",
|
|
static_folder="static",
|
|
static_url_path="/suggestions-static",
|
|
)
|
|
|
|
# Where uploaded images land. Lives under the mounted /data volume so backups
|
|
# and restarts keep them. Served at /uploads/suggestions/<name> (see below).
|
|
UPLOAD_DIR = Path("/data/uploads/suggestions")
|
|
|
|
|
|
def current_user():
|
|
"""Resolve the logged-in user. ADAPT THIS to the app's real auth.
|
|
|
|
Must return an object exposing `.name` and `.email`, or None if anonymous.
|
|
"""
|
|
# --- Example: Flask-Login ---
|
|
# from flask_login import current_user as u
|
|
# return u if getattr(u, "is_authenticated", False) else None
|
|
#
|
|
# --- Example: session dict ---
|
|
# from flask import session
|
|
# if "user_email" in session:
|
|
# return type("U", (), {"name": session.get("user_name", "User"),
|
|
# "email": session["user_email"]})()
|
|
# return None
|
|
#
|
|
# --- Stub: single-household app with no real auth ---
|
|
return type("U", (), {"name": "Household", "email": "household@local"})()
|
|
|
|
|
|
def _guard():
|
|
return current_user() is not None
|
|
|
|
|
|
@suggestions_bp.route("/suggestions")
|
|
def suggestions_page():
|
|
return render_template("suggestions.html", configured=gitea_config() is not None)
|
|
|
|
|
|
@suggestions_bp.route("/api/suggestions", methods=["GET", "POST"])
|
|
def suggestions_api():
|
|
if not _guard():
|
|
return jsonify({"error": "unauthorized"}), 401
|
|
cfg = gitea_config()
|
|
if not cfg:
|
|
return jsonify({"error": "not_configured"}), 503
|
|
client = GiteaClient(cfg)
|
|
|
|
if request.method == "GET":
|
|
state = request.args.get("state", "open")
|
|
try:
|
|
return jsonify({"issues": client.list_issues(state)})
|
|
except GiteaError as e:
|
|
return jsonify({"error": str(e)}), 502
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
title = (data.get("title") or "").strip()
|
|
if len(title) < 3:
|
|
return jsonify({"error": "title_required"}), 400
|
|
body = (data.get("body") or "").strip()
|
|
user = current_user()
|
|
if user is not None:
|
|
body += footer(user.name, user.email)
|
|
try:
|
|
return jsonify({"issue": client.create_issue(title, body)}), 201
|
|
except GiteaError as e:
|
|
return jsonify({"error": str(e)}), 502
|
|
|
|
|
|
@suggestions_bp.route("/api/suggestions/<int:number>/comments", methods=["GET", "POST"])
|
|
def suggestion_comments(number: int):
|
|
if not _guard():
|
|
return jsonify({"error": "unauthorized"}), 401
|
|
cfg = gitea_config()
|
|
if not cfg:
|
|
return jsonify({"error": "not_configured"}), 503
|
|
client = GiteaClient(cfg)
|
|
|
|
if request.method == "GET":
|
|
try:
|
|
return jsonify({"comments": client.list_comments(number)})
|
|
except GiteaError as e:
|
|
return jsonify({"error": str(e)}), 502
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
text = (data.get("body") or "").strip()
|
|
if not text:
|
|
return jsonify({"error": "body_required"}), 400
|
|
user = current_user()
|
|
if user is not None:
|
|
text += f"\n\n_— {user.name}_"
|
|
try:
|
|
return jsonify({"comment": client.create_comment(number, text)}), 201
|
|
except GiteaError as e:
|
|
return jsonify({"error": str(e)}), 502
|
|
|
|
|
|
@suggestions_bp.route("/api/suggestions/upload", methods=["POST"])
|
|
def suggestions_upload():
|
|
if not _guard():
|
|
return jsonify({"error": "unauthorized"}), 401
|
|
if "file" not in request.files:
|
|
return jsonify({"error": "file_required"}), 400
|
|
f = request.files["file"]
|
|
data = f.read(MAX_UPLOAD_BYTES + 1)
|
|
if len(data) > MAX_UPLOAD_BYTES:
|
|
return jsonify({"error": "too_large"}), 413
|
|
if sniff_image(data) not in ALLOWED_IMAGE_TYPES:
|
|
return jsonify({"error": "unsupported_type"}), 415
|
|
url = save_upload(f.filename or "image", data, UPLOAD_DIR, request.host_url)
|
|
return jsonify({"url": url}), 201
|