Compare commits
2 Commits
16e4905dcb
...
0893c713e4
| Author | SHA1 | Date | |
|---|---|---|---|
| 0893c713e4 | |||
|
|
635372d7a2 |
27
README.md
27
README.md
@ -154,9 +154,12 @@ apps/board/
|
||||
auth.py Gitea OAuth2 (confidential client) and the membership gate
|
||||
members.py roles, provisioning, ownership transfer, GDPR erasure/export
|
||||
board.py threads and comments
|
||||
content.py the editor — writes posts to Gitea's contents API
|
||||
gitea.py the only module that talks to Gitea
|
||||
tokens.py per-member access tokens, refreshed before they expire
|
||||
render.py markdown with raw HTML disabled
|
||||
schema.sql three tables and the one-owner index
|
||||
tests/ pytest, 41 checks — `python3 -m pytest apps/board/tests`
|
||||
schema.sql the tables, including the one-owner index
|
||||
tests/ pytest, 97 checks — `python3 -m pytest apps/board/tests`
|
||||
```
|
||||
|
||||
Three things about it are load-bearing and easy to undo by accident:
|
||||
@ -178,6 +181,26 @@ Its SQLite database is the only state on the server that git does not hold.
|
||||
`scripts/backup-board.sh` takes a consistent snapshot nightly — see
|
||||
`docs/server-setup.md` §11.
|
||||
|
||||
### Writing posts
|
||||
|
||||
`/comunidad/contenido/` replaces Decap CMS for admins: a form that commits a
|
||||
file through Gitea's contents API, so Woodpecker sees an ordinary push and
|
||||
translate → build → deploy runs unchanged. Commits carry the author's own
|
||||
account, not a bot's.
|
||||
|
||||
It exists because Decap has no supported way to be themed — its maintainer's
|
||||
answer is override its CSS and accept that class names move, or fork it — so
|
||||
`/admin/` would always look like a different product bolted on.
|
||||
|
||||
**Decap is still running.** Both editors write the same files, and the pipeline
|
||||
cannot tell them apart, so there is no cutover to get wrong. `docs/server-setup.md`
|
||||
§11.8 covers removing Decap when the new editor has earned it.
|
||||
|
||||
Filenames are the contract, not a detail: `YYYY-MM-DD-slug.es.md` is what
|
||||
`split_lang()` parses and what keeps two posts off one URL. The tests import
|
||||
`scripts/translate.py` and run its parser over what the editor writes, because
|
||||
a file it cannot parse publishes in Spanish and is never translated, silently.
|
||||
|
||||
## GEO/SEO surfaces
|
||||
|
||||
- `hreflang` + `og:locale(:alternate)` + JSON-LD `BlogPosting`/`Blog` with
|
||||
|
||||
@ -50,6 +50,11 @@ def create_app(overrides: dict | None = None) -> Flask:
|
||||
ADMIN_TOKEN=os.environ.get("GITEA_ADMIN_TOKEN", ""),
|
||||
OWNER_LOGIN=os.environ.get("BOARD_OWNER", ""),
|
||||
BASE_URL=os.environ.get("BOARD_BASE_URL", "https://vienalatina.com"),
|
||||
# The repository the editor commits to — the same one Woodpecker builds,
|
||||
# which is what makes publishing from here indistinguishable from
|
||||
# publishing from Decap as far as the pipeline is concerned.
|
||||
CONTENT_REPO=os.environ.get("CONTENT_REPO", "pablo/vienalatina"),
|
||||
CONTENT_BRANCH=os.environ.get("CONTENT_BRANCH", "main"),
|
||||
URL_PREFIX=URL_PREFIX,
|
||||
COOLDOWN_SECONDS=int(os.environ.get("BOARD_COOLDOWN_SECONDS", "20")),
|
||||
SESSION_COOKIE_HTTPONLY=True,
|
||||
@ -59,7 +64,11 @@ def create_app(overrides: dict | None = None) -> Flask:
|
||||
# types the address without https.
|
||||
SESSION_COOKIE_SECURE=_env_flag("BOARD_COOKIE_SECURE", True),
|
||||
SESSION_COOKIE_NAME="vl_board",
|
||||
MAX_CONTENT_LENGTH=256 * 1024,
|
||||
# Generous enough for a photo off a phone. The board itself needs a
|
||||
# fraction of this; the editor uploads images, and a writer whose
|
||||
# picture is silently refused has no way to tell what went wrong.
|
||||
MAX_CONTENT_LENGTH=10 * 1024 * 1024,
|
||||
UPLOAD_MAX_BYTES=int(os.environ.get("BOARD_UPLOAD_MAX_BYTES", 8 * 1024 * 1024)),
|
||||
)
|
||||
if overrides:
|
||||
app.config.update(overrides)
|
||||
@ -69,9 +78,10 @@ def create_app(overrides: dict | None = None) -> Flask:
|
||||
# restart, which is a confusing way to find out the variable is unset.
|
||||
raise RuntimeError("BOARD_SECRET_KEY is required (generate one with `openssl rand -hex 32`).")
|
||||
|
||||
from . import auth, board, members
|
||||
from . import auth, board, content, members
|
||||
app.register_blueprint(auth.bp, url_prefix=URL_PREFIX)
|
||||
app.register_blueprint(members.bp, url_prefix=URL_PREFIX)
|
||||
app.register_blueprint(content.bp, url_prefix=URL_PREFIX)
|
||||
app.register_blueprint(board.bp, url_prefix=URL_PREFIX)
|
||||
|
||||
app.teardown_appcontext(close_db)
|
||||
|
||||
@ -14,7 +14,7 @@ import secrets
|
||||
from flask import (Blueprint, current_app, flash, g, redirect, render_template,
|
||||
request, session, url_for)
|
||||
|
||||
from . import gitea
|
||||
from . import gitea, tokens
|
||||
from .db import get_db
|
||||
|
||||
bp = Blueprint("auth", __name__)
|
||||
@ -77,8 +77,8 @@ def callback():
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
try:
|
||||
token = gitea.exchange_code(code, redirect_uri())
|
||||
profile = gitea.fetch_user(token)
|
||||
credentials = gitea.exchange_code(code, redirect_uri())
|
||||
profile = gitea.fetch_user(credentials["access_token"])
|
||||
except gitea.GiteaError as exc:
|
||||
current_app.logger.warning("OAuth failed: %s", exc)
|
||||
flash(str(exc), "error")
|
||||
@ -117,6 +117,10 @@ def callback():
|
||||
session.clear()
|
||||
session["member_id"] = member["id"]
|
||||
|
||||
# Kept so the editor can commit as this person rather than as a bot. Stored
|
||||
# in the database, never in the cookie — see apps/board/tokens.py.
|
||||
tokens.save(member["id"], credentials)
|
||||
|
||||
target = request.args.get("next") or session.pop("oauth_next", "") or ""
|
||||
# Only ever redirect within this app: an absolute URL here would make the
|
||||
# login page an open redirect that phishing can point anywhere.
|
||||
@ -127,6 +131,10 @@ def callback():
|
||||
|
||||
@bp.route("/logout", methods=["POST"])
|
||||
def logout():
|
||||
# Drop the Gitea token too. Signing out should stop the server being able to
|
||||
# act as you, not just stop the browser being able to ask it to.
|
||||
if g.member is not None:
|
||||
tokens.forget(g.member["id"])
|
||||
session.clear()
|
||||
flash("Sesión cerrada.", "ok")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
397
apps/board/content.py
Normal file
397
apps/board/content.py
Normal file
@ -0,0 +1,397 @@
|
||||
"""Writing the public site from inside the members area.
|
||||
|
||||
Replaces what Decap CMS does at /admin/, for one reason: Decap has no supported
|
||||
way to be restyled, so it will always look like a different product bolted onto
|
||||
the side of this one.
|
||||
|
||||
Posts are files in a git repository, so this is a form that commits a file.
|
||||
There is no working copy, no queue and no new state — `apps/board/gitea.py`
|
||||
talks to Gitea's contents API, Gitea's push webhook fires Woodpecker, and the
|
||||
translate → build → deploy pipeline runs exactly as it does for a Decap commit.
|
||||
Nothing downstream can tell the difference, which is the property that makes
|
||||
this safe to switch on while Decap is still running.
|
||||
|
||||
The filename and frontmatter rules below are not this module's to invent: they
|
||||
are the contract `scripts/translate.py` reads. `<basename>.es.md` is what
|
||||
`split_lang()` parses, the date prefix is what stops two posts with one title
|
||||
colliding on a URL, and the absence of `translated_from` is what marks a file as
|
||||
something a person wrote.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import secrets
|
||||
import unicodedata
|
||||
from datetime import date as date_type
|
||||
from datetime import datetime
|
||||
|
||||
import yaml
|
||||
from flask import (Blueprint, abort, current_app, flash, redirect,
|
||||
render_template, request, url_for)
|
||||
|
||||
from . import gitea, tokens
|
||||
from .db import get_db
|
||||
from .render import to_html
|
||||
from .security import admin_required
|
||||
|
||||
bp = Blueprint("content", __name__)
|
||||
|
||||
COLLECTIONS = {
|
||||
"post": {
|
||||
"label": "Artículos", "singular": "Artículo",
|
||||
"folder": "content/post", "dated": True,
|
||||
},
|
||||
"page": {
|
||||
"label": "Páginas", "singular": "Página",
|
||||
"folder": "content/page", "dated": False,
|
||||
},
|
||||
}
|
||||
|
||||
CATEGORIES = ["Turismo", "Cultura", "Gastronomía", "Comunidad", "Comercio"]
|
||||
|
||||
UPLOAD_FOLDER = "static/uploads"
|
||||
IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "webp", "gif", "avif"}
|
||||
|
||||
TITLE_MAX = 140
|
||||
BODY_MAX = 100_000
|
||||
|
||||
# Anything arriving in a URL is checked against this before it reaches a path.
|
||||
# Without it, `../../` in a filename would let the editor read and overwrite any
|
||||
# file in the repository — the pipeline and the Caddyfile included.
|
||||
SAFE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*\.es\.md$")
|
||||
|
||||
|
||||
# --- the contract with translate.py --------------------------------------
|
||||
|
||||
def slugify(text: str) -> str:
|
||||
"""Accents folded rather than dropped, so "Gastronomía" is not "gastronom"."""
|
||||
folded = unicodedata.normalize("NFKD", text or "")
|
||||
ascii_only = "".join(ch for ch in folded if not unicodedata.combining(ch))
|
||||
return re.sub(r"[^a-z0-9]+", "-", ascii_only.lower()).strip("-") or "sin-titulo"
|
||||
|
||||
|
||||
def filename_for(collection: str, title: str, when: date_type | None) -> str:
|
||||
slug = slugify(title)
|
||||
if COLLECTIONS[collection]["dated"]:
|
||||
return f"{when:%Y-%m-%d}-{slug}.es.md"
|
||||
return f"{slug}.es.md"
|
||||
|
||||
|
||||
def split_frontmatter(text: str) -> tuple[dict, str]:
|
||||
match = re.match(r"\A---\n(.*?)\n---\n?(.*)\Z", text, re.DOTALL)
|
||||
if not match:
|
||||
return {}, text
|
||||
return yaml.safe_load(match.group(1)) or {}, match.group(2)
|
||||
|
||||
|
||||
def build_document(fields: dict, body: str) -> str:
|
||||
"""Key order matches what the pipeline already writes, so a file edited here
|
||||
and a file written by translate.py read the same in a diff."""
|
||||
front = yaml.safe_dump(fields, allow_unicode=True, sort_keys=False,
|
||||
default_flow_style=False)
|
||||
return f"---\n{front}---\n\n{body.strip()}\n"
|
||||
|
||||
|
||||
def frontmatter_for(collection: str, form: dict) -> dict:
|
||||
fields = {"title": form["title"]}
|
||||
if COLLECTIONS[collection]["dated"]:
|
||||
fields["date"] = form["date"]
|
||||
fields["lang"] = "es"
|
||||
fields["manual_translation"] = form["manual_translation"]
|
||||
if collection == "post":
|
||||
if form["categories"]:
|
||||
fields["categories"] = form["categories"]
|
||||
if form["description"]:
|
||||
fields["description"] = form["description"]
|
||||
if form["image"]:
|
||||
fields["image"] = form["image"]
|
||||
return fields
|
||||
|
||||
|
||||
# --- listing, with the cache doing the work ------------------------------
|
||||
|
||||
def _cache_read(path: str, sha: str):
|
||||
return get_db().execute(
|
||||
"SELECT * FROM content_cache WHERE path = ? AND sha = ?", (path, sha)
|
||||
).fetchone()
|
||||
|
||||
|
||||
def _cache_write(path: str, sha: str, fm: dict) -> None:
|
||||
get_db().execute(
|
||||
"""INSERT INTO content_cache (path, sha, title, date, categories, generated)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(path) DO UPDATE SET
|
||||
sha = excluded.sha, title = excluded.title, date = excluded.date,
|
||||
categories = excluded.categories, generated = excluded.generated,
|
||||
updated_at = datetime('now')""",
|
||||
(path, sha, str(fm.get("title", "")), str(fm.get("date", "")),
|
||||
", ".join(fm.get("categories") or []), 1 if fm.get("translated_from") else 0),
|
||||
)
|
||||
|
||||
|
||||
def listing(collection: str) -> list[dict]:
|
||||
folder = COLLECTIONS[collection]["folder"]
|
||||
entries = tokens.with_token(gitea.list_directory, folder)
|
||||
|
||||
items = []
|
||||
for entry in entries:
|
||||
name = entry.get("name", "")
|
||||
if not name.endswith(".es.md"):
|
||||
continue # generated German and Portuguese siblings are not edited here
|
||||
path, sha = entry["path"], entry["sha"]
|
||||
|
||||
row = _cache_read(path, sha)
|
||||
if row is None:
|
||||
text, _ = tokens.with_token(gitea.read_file, path)
|
||||
fm, _body = split_frontmatter(text)
|
||||
_cache_write(path, sha, fm)
|
||||
row = _cache_read(path, sha)
|
||||
|
||||
if row["generated"]:
|
||||
continue # written by the pipeline; editing it would be overwritten
|
||||
items.append({
|
||||
"name": name, "path": path, "sha": sha,
|
||||
"title": row["title"] or name,
|
||||
"date": row["date"], "categories": row["categories"],
|
||||
})
|
||||
|
||||
items.sort(key=lambda item: (item["date"], item["name"]), reverse=True)
|
||||
return items
|
||||
|
||||
|
||||
# --- form handling -------------------------------------------------------
|
||||
|
||||
def _collection_or_404(collection: str) -> dict:
|
||||
if collection not in COLLECTIONS:
|
||||
abort(404)
|
||||
return COLLECTIONS[collection]
|
||||
|
||||
|
||||
def _name_or_404(name: str) -> str:
|
||||
if not SAFE_NAME.match(name):
|
||||
abort(404)
|
||||
return name
|
||||
|
||||
|
||||
def _read_form(collection: str) -> tuple[dict, str, list[str]]:
|
||||
"""Returns (fields, body, errors). Always returns something renderable, so a
|
||||
rejected form comes back with the writer's text still in it."""
|
||||
errors = []
|
||||
title = request.form.get("title", "").strip()[:TITLE_MAX]
|
||||
body = request.form.get("body", "").strip()[:BODY_MAX]
|
||||
|
||||
raw_date = request.form.get("date", "").strip()
|
||||
when = None
|
||||
if COLLECTIONS[collection]["dated"]:
|
||||
try:
|
||||
when = datetime.strptime(raw_date, "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
errors.append("La fecha debe tener el formato AAAA-MM-DD.")
|
||||
|
||||
categories = [c for c in request.form.getlist("categories") if c in CATEGORIES]
|
||||
image = request.form.get("image", "").strip()
|
||||
if image and not re.match(r"^/uploads/[A-Za-z0-9._-]+$", image):
|
||||
errors.append("La imagen no es válida.")
|
||||
image = ""
|
||||
|
||||
if not title:
|
||||
errors.append("El título no puede estar vacío.")
|
||||
if not body:
|
||||
errors.append("El cuerpo no puede estar vacío.")
|
||||
|
||||
fields = {
|
||||
"title": title, "date": when, "categories": categories,
|
||||
"description": request.form.get("description", "").strip(),
|
||||
"image": image,
|
||||
"manual_translation": request.form.get("manual_translation") == "on",
|
||||
}
|
||||
return fields, body, errors
|
||||
|
||||
|
||||
def _upload_image() -> str:
|
||||
"""Commit an uploaded picture and return the path the frontmatter uses.
|
||||
|
||||
Deliberately a second commit rather than part of the post's. Gitea's
|
||||
contents API writes one file per request, and batching both into a single
|
||||
commit means the lower-level git trees API — noticeably more code to get
|
||||
wrong, for a benefit nobody sees beyond one fewer pipeline run.
|
||||
"""
|
||||
upload = request.files.get("picture")
|
||||
if not upload or not upload.filename:
|
||||
return ""
|
||||
|
||||
extension = upload.filename.rsplit(".", 1)[-1].lower()
|
||||
if extension not in IMAGE_EXTENSIONS:
|
||||
raise gitea.GiteaError(
|
||||
f"Formato de imagen no admitido. Usa: {', '.join(sorted(IMAGE_EXTENSIONS))}."
|
||||
)
|
||||
|
||||
data = upload.read()
|
||||
maximum = current_app.config["UPLOAD_MAX_BYTES"]
|
||||
if len(data) > maximum:
|
||||
raise gitea.GiteaError(
|
||||
f"La imagen pesa {len(data) // 1024}KB y el máximo es {maximum // 1024}KB."
|
||||
)
|
||||
|
||||
stem = slugify(upload.filename.rsplit(".", 1)[0])[:60]
|
||||
# A random suffix rather than a counter: two people uploading "foto.jpg"
|
||||
# in the same minute must not race for the same path.
|
||||
name = f"{stem}-{secrets.token_hex(3)}.{extension}"
|
||||
tokens.with_token(gitea.write_file, f"{UPLOAD_FOLDER}/{name}", data,
|
||||
f"content: subir {name}")
|
||||
return f"/uploads/{name}"
|
||||
|
||||
|
||||
# --- routes --------------------------------------------------------------
|
||||
|
||||
@bp.route("/contenido")
|
||||
@bp.route("/contenido/<collection>")
|
||||
@admin_required
|
||||
def index(collection: str = "post"):
|
||||
meta = _collection_or_404(collection)
|
||||
try:
|
||||
items = listing(collection)
|
||||
except tokens.NeedsSignIn:
|
||||
return redirect(url_for("auth.login", next=request.path))
|
||||
except gitea.GiteaError as exc:
|
||||
flash(str(exc), "error")
|
||||
items = []
|
||||
return render_template("content_list.html", collection=collection, meta=meta,
|
||||
collections=COLLECTIONS, items=items)
|
||||
|
||||
|
||||
@bp.route("/contenido/<collection>/nuevo", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def new(collection: str):
|
||||
meta = _collection_or_404(collection)
|
||||
if request.method == "GET":
|
||||
return render_template("content_form.html", collection=collection, meta=meta,
|
||||
categories=CATEGORIES, item=None, fields=None,
|
||||
body="", today=date_type.today().isoformat())
|
||||
|
||||
fields, body, errors = _read_form(collection)
|
||||
if errors:
|
||||
return _back_to_form(collection, meta, fields, body, errors, None)
|
||||
|
||||
try:
|
||||
picture = _upload_image()
|
||||
if picture:
|
||||
fields["image"] = picture
|
||||
name = filename_for(collection, fields["title"], fields["date"])
|
||||
path = f"{meta['folder']}/{name}"
|
||||
document = build_document(frontmatter_for(collection, fields), body)
|
||||
tokens.with_token(gitea.write_file, path, document.encode("utf-8"),
|
||||
f"content: publicar «{fields['title']}»")
|
||||
except tokens.NeedsSignIn:
|
||||
return redirect(url_for("auth.login", next=request.path))
|
||||
except gitea.GiteaError as exc:
|
||||
return _back_to_form(collection, meta, fields, body, [str(exc)], None)
|
||||
|
||||
flash("Publicado. La traducción tarda un par de minutos.", "ok")
|
||||
return redirect(url_for("content.index", collection=collection))
|
||||
|
||||
|
||||
@bp.route("/contenido/<collection>/editar/<name>", methods=["GET", "POST"])
|
||||
@admin_required
|
||||
def edit(collection: str, name: str):
|
||||
meta = _collection_or_404(collection)
|
||||
name = _name_or_404(name)
|
||||
path = f"{meta['folder']}/{name}"
|
||||
|
||||
if request.method == "GET":
|
||||
try:
|
||||
text, sha = tokens.with_token(gitea.read_file, path)
|
||||
except tokens.NeedsSignIn:
|
||||
return redirect(url_for("auth.login", next=request.path))
|
||||
except gitea.GiteaError as exc:
|
||||
flash(str(exc), "error")
|
||||
return redirect(url_for("content.index", collection=collection))
|
||||
|
||||
fm, body = split_frontmatter(text)
|
||||
if fm.get("translated_from"):
|
||||
flash("Ese archivo lo genera la traducción automática; no se edita aquí.",
|
||||
"error")
|
||||
return redirect(url_for("content.index", collection=collection))
|
||||
|
||||
fields = {
|
||||
"title": fm.get("title", ""),
|
||||
"date": fm.get("date"),
|
||||
"categories": fm.get("categories") or [],
|
||||
"description": fm.get("description", ""),
|
||||
"image": fm.get("image", ""),
|
||||
"manual_translation": bool(fm.get("manual_translation")),
|
||||
}
|
||||
return render_template("content_form.html", collection=collection, meta=meta,
|
||||
categories=CATEGORIES, item={"name": name, "sha": sha},
|
||||
fields=fields, body=body,
|
||||
today=date_type.today().isoformat())
|
||||
|
||||
fields, body, errors = _read_form(collection)
|
||||
sha = request.form.get("sha", "")
|
||||
item = {"name": name, "sha": sha}
|
||||
if errors:
|
||||
return _back_to_form(collection, meta, fields, body, errors, item)
|
||||
|
||||
try:
|
||||
picture = _upload_image()
|
||||
if picture:
|
||||
fields["image"] = picture
|
||||
document = build_document(frontmatter_for(collection, fields), body)
|
||||
tokens.with_token(gitea.write_file, path, document.encode("utf-8"),
|
||||
f"content: actualizar «{fields['title']}»", sha=sha)
|
||||
except tokens.NeedsSignIn:
|
||||
return redirect(url_for("auth.login", next=request.path))
|
||||
except gitea.GiteaError as exc:
|
||||
return _back_to_form(collection, meta, fields, body, [str(exc)], item)
|
||||
|
||||
flash("Guardado.", "ok")
|
||||
return redirect(url_for("content.index", collection=collection))
|
||||
|
||||
|
||||
@bp.route("/contenido/<collection>/eliminar/<name>", methods=["POST"])
|
||||
@admin_required
|
||||
def delete(collection: str, name: str):
|
||||
meta = _collection_or_404(collection)
|
||||
name = _name_or_404(name)
|
||||
path = f"{meta['folder']}/{name}"
|
||||
try:
|
||||
_text, sha = tokens.with_token(gitea.read_file, path)
|
||||
tokens.with_token(gitea.delete_file, path, sha, f"content: eliminar {name}")
|
||||
except tokens.NeedsSignIn:
|
||||
return redirect(url_for("auth.login", next=request.path))
|
||||
except gitea.GiteaError as exc:
|
||||
flash(str(exc), "error")
|
||||
return redirect(url_for("content.index", collection=collection))
|
||||
|
||||
get_db().execute("DELETE FROM content_cache WHERE path = ?", (path,))
|
||||
# The German and Portuguese siblings go too, but not from here: the next
|
||||
# pipeline run reaps them (translate.py's orphaned_siblings).
|
||||
flash("Eliminado. Las traducciones se borran en la siguiente publicación.", "ok")
|
||||
return redirect(url_for("content.index", collection=collection))
|
||||
|
||||
|
||||
@bp.route("/contenido/<collection>/vista-previa", methods=["POST"])
|
||||
@admin_required
|
||||
def preview(collection: str):
|
||||
"""Rendered on the server and returned as a whole page.
|
||||
|
||||
No JavaScript and no fetch: the Content-Security-Policy forbids inline
|
||||
script, and a preview is not worth a second way of talking to the server.
|
||||
"""
|
||||
meta = _collection_or_404(collection)
|
||||
fields, body, _errors = _read_form(collection)
|
||||
sha = request.form.get("sha", "")
|
||||
item = {"name": request.form.get("name", ""), "sha": sha} if sha else None
|
||||
return render_template("content_form.html", collection=collection, meta=meta,
|
||||
categories=CATEGORIES, item=item, fields=fields, body=body,
|
||||
today=date_type.today().isoformat(),
|
||||
preview_html=to_html(body))
|
||||
|
||||
|
||||
def _back_to_form(collection, meta, fields, body, errors, item):
|
||||
for message in errors:
|
||||
flash(message, "error")
|
||||
return render_template("content_form.html", collection=collection, meta=meta,
|
||||
categories=CATEGORIES, item=item, fields=fields, body=body,
|
||||
today=date_type.today().isoformat()), 400
|
||||
@ -1,7 +1,6 @@
|
||||
"""The only place that talks to Gitea.
|
||||
|
||||
Two unrelated conversations happen here and are worth keeping apart in your
|
||||
head:
|
||||
Three unrelated conversations happen here, worth keeping apart in your head:
|
||||
|
||||
* **Sign-in** uses OAuth2 on behalf of the person at the keyboard. The app is
|
||||
registered as a *confidential* client with a secret, which it can hold
|
||||
@ -12,13 +11,19 @@ head:
|
||||
* **Creating an account** uses a site-admin token belonging to the instance,
|
||||
not to any member. That token can create and modify any Gitea user, so the
|
||||
environment holding it is as sensitive as Gitea's own admin password.
|
||||
|
||||
* **Reading and writing content** uses the signed-in member's *own* access
|
||||
token. Commits are then attributed to the person who actually wrote the post,
|
||||
and Gitea's permissions apply unchanged — the editor cannot grant write access
|
||||
to somebody who does not already have it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import secrets
|
||||
import string
|
||||
from urllib.parse import urlencode
|
||||
from urllib.parse import quote, urlencode
|
||||
|
||||
import requests
|
||||
from flask import current_app
|
||||
@ -30,6 +35,15 @@ class GiteaError(RuntimeError):
|
||||
"""Gitea refused a request. The message is safe to show a member."""
|
||||
|
||||
|
||||
class StaleFile(GiteaError):
|
||||
"""The file changed since it was loaded into the form.
|
||||
|
||||
Gitea rejects a write whose `sha` no longer matches the branch, which is
|
||||
what makes two people editing one post a visible conflict rather than a
|
||||
silent overwrite of whoever saved first.
|
||||
"""
|
||||
|
||||
|
||||
def _base() -> str:
|
||||
return current_app.config["GITEA_URL"].rstrip("/")
|
||||
|
||||
@ -38,6 +52,16 @@ def _api(path: str) -> str:
|
||||
return f"{_base()}/api/v1{path}"
|
||||
|
||||
|
||||
def _repo() -> str:
|
||||
return current_app.config["CONTENT_REPO"].strip("/")
|
||||
|
||||
|
||||
def _branch() -> str:
|
||||
return current_app.config["CONTENT_BRANCH"]
|
||||
|
||||
|
||||
# --- sign-in -------------------------------------------------------------
|
||||
|
||||
def authorize_url(state: str, redirect_uri: str) -> str:
|
||||
query = urlencode({
|
||||
"client_id": current_app.config["OAUTH_CLIENT_ID"],
|
||||
@ -48,24 +72,40 @@ def authorize_url(state: str, redirect_uri: str) -> str:
|
||||
return f"{_base()}/login/oauth/authorize?{query}"
|
||||
|
||||
|
||||
def exchange_code(code: str, redirect_uri: str) -> str:
|
||||
def _token_request(payload: dict) -> dict:
|
||||
response = requests.post(
|
||||
f"{_base()}/login/oauth/access_token",
|
||||
json={
|
||||
"client_id": current_app.config["OAUTH_CLIENT_ID"],
|
||||
"client_secret": current_app.config["OAUTH_CLIENT_SECRET"],
|
||||
"code": code,
|
||||
"grant_type": "authorization_code",
|
||||
"redirect_uri": redirect_uri,
|
||||
**payload,
|
||||
},
|
||||
timeout=TIMEOUT,
|
||||
)
|
||||
if response.status_code != 200:
|
||||
raise GiteaError("No se pudo completar el inicio de sesión.")
|
||||
token = response.json().get("access_token")
|
||||
if not token:
|
||||
data = response.json()
|
||||
if not data.get("access_token"):
|
||||
raise GiteaError("Gitea no devolvió un token de acceso.")
|
||||
return token
|
||||
return data
|
||||
|
||||
|
||||
def exchange_code(code: str, redirect_uri: str) -> dict:
|
||||
"""Returns the whole token response, not just the access token.
|
||||
|
||||
The refresh token matters: Gitea's access tokens last about an hour, and
|
||||
without refreshing, saving a post would start failing partway through an
|
||||
afternoon's work for no reason the writer could understand.
|
||||
"""
|
||||
return _token_request({
|
||||
"code": code,
|
||||
"grant_type": "authorization_code",
|
||||
"redirect_uri": redirect_uri,
|
||||
})
|
||||
|
||||
|
||||
def refresh_token(token: str) -> dict:
|
||||
return _token_request({"refresh_token": token, "grant_type": "refresh_token"})
|
||||
|
||||
|
||||
def fetch_user(token: str) -> dict:
|
||||
@ -79,6 +119,8 @@ def fetch_user(token: str) -> dict:
|
||||
return response.json()
|
||||
|
||||
|
||||
# --- account creation (site-admin token) ---------------------------------
|
||||
|
||||
def generate_password() -> str:
|
||||
# Shown once to the admin, then changed by the member on first login.
|
||||
# Punctuation is left out on purpose: this gets read aloud or copied by
|
||||
@ -113,3 +155,80 @@ def admin_create_user(login: str, email: str, full_name: str, password: str) ->
|
||||
if response.status_code in (401, 403):
|
||||
raise GiteaError("El token de administración de Gitea no es válido.")
|
||||
raise GiteaError(f"Gitea rechazó la creación del usuario ({response.status_code}).")
|
||||
|
||||
|
||||
# --- content (the member's own token) ------------------------------------
|
||||
|
||||
def _contents_url(path: str) -> str:
|
||||
# quote() with no safe characters: a path segment is data, not structure.
|
||||
return _api(f"/repos/{_repo()}/contents/{quote(path, safe='/')}")
|
||||
|
||||
|
||||
def _content_request(method: str, url: str, token: str, **kwargs):
|
||||
response = requests.request(
|
||||
method, url,
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=TIMEOUT,
|
||||
**kwargs,
|
||||
)
|
||||
if response.status_code in (401, 403):
|
||||
raise PermissionError("gitea-unauthorised") # caller refreshes and retries
|
||||
return response
|
||||
|
||||
|
||||
def list_directory(path: str, token: str) -> list[dict]:
|
||||
"""Names and shas only — the contents API does not return file bodies here."""
|
||||
response = _content_request("GET", _contents_url(path), token,
|
||||
params={"ref": _branch()})
|
||||
if response.status_code == 404:
|
||||
return [] # an empty content folder is normal, not an error
|
||||
if response.status_code != 200:
|
||||
raise GiteaError(f"Gitea no devolvió la lista de archivos ({response.status_code}).")
|
||||
payload = response.json()
|
||||
return [item for item in payload if item.get("type") == "file"]
|
||||
|
||||
|
||||
def read_file(path: str, token: str) -> tuple[str, str]:
|
||||
"""(text, sha). The sha comes back so a later write can prove it is current."""
|
||||
response = _content_request("GET", _contents_url(path), token,
|
||||
params={"ref": _branch()})
|
||||
if response.status_code == 404:
|
||||
raise GiteaError("Ese archivo ya no existe.")
|
||||
if response.status_code != 200:
|
||||
raise GiteaError(f"No se pudo leer el archivo ({response.status_code}).")
|
||||
payload = response.json()
|
||||
text = base64.b64decode(payload.get("content", "")).decode("utf-8")
|
||||
return text, payload.get("sha", "")
|
||||
|
||||
|
||||
def write_file(path: str, data: bytes, message: str, token: str,
|
||||
sha: str | None = None) -> str:
|
||||
"""Create when `sha` is None, update otherwise. Returns the new sha."""
|
||||
body = {
|
||||
"content": base64.b64encode(data).decode("ascii"),
|
||||
"message": message,
|
||||
"branch": _branch(),
|
||||
}
|
||||
if sha:
|
||||
body["sha"] = sha
|
||||
response = _content_request("PUT" if sha else "POST", _contents_url(path),
|
||||
token, json=body)
|
||||
if response.status_code in (200, 201):
|
||||
return response.json().get("content", {}).get("sha", "")
|
||||
if response.status_code in (409, 422):
|
||||
raise StaleFile(
|
||||
"Alguien más guardó este archivo mientras lo editabas. "
|
||||
"Vuelve a abrirlo para no perder su trabajo."
|
||||
)
|
||||
raise GiteaError(f"Gitea rechazó el guardado ({response.status_code}).")
|
||||
|
||||
|
||||
def delete_file(path: str, sha: str, message: str, token: str) -> None:
|
||||
response = _content_request("DELETE", _contents_url(path), token, json={
|
||||
"sha": sha, "message": message, "branch": _branch(),
|
||||
})
|
||||
if response.status_code in (200, 204):
|
||||
return
|
||||
if response.status_code in (409, 422):
|
||||
raise StaleFile("El archivo cambió desde que lo abriste. Recarga la lista.")
|
||||
raise GiteaError(f"Gitea rechazó el borrado ({response.status_code}).")
|
||||
|
||||
@ -10,3 +10,6 @@ gunicorn==23.0.0
|
||||
markdown-it-py==4.2.0
|
||||
linkify-it-py==2.2.0
|
||||
requests==2.33.1
|
||||
# Same serialiser the pipeline uses, so frontmatter written by the editor and
|
||||
# by scripts/translate.py is byte-identical in a diff.
|
||||
pyyaml==6.0.2
|
||||
|
||||
@ -54,3 +54,38 @@ CREATE TABLE IF NOT EXISTS comments (
|
||||
|
||||
CREATE INDEX IF NOT EXISTS comments_thread
|
||||
ON comments(thread_id, created_at) WHERE deleted_at IS NULL;
|
||||
|
||||
-- Gitea access tokens for the editor.
|
||||
--
|
||||
-- Kept here rather than in the session cookie. Flask signs cookies but does not
|
||||
-- encrypt them, so a live token sitting in one is readable by anything that can
|
||||
-- read the cookie — and a token is enough to commit to the repository as its
|
||||
-- owner. ON DELETE CASCADE ties the token to the membership: erasing a member
|
||||
-- takes their token with it, with nothing to remember.
|
||||
CREATE TABLE IF NOT EXISTS gitea_tokens (
|
||||
member_id INTEGER PRIMARY KEY REFERENCES members(id) ON DELETE CASCADE,
|
||||
access_token TEXT NOT NULL,
|
||||
refresh_token TEXT NOT NULL DEFAULT '',
|
||||
expires_at TEXT,
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
-- Frontmatter of content files, keyed by the git blob sha.
|
||||
--
|
||||
-- Listing a folder through Gitea's contents API returns names and shas but no
|
||||
-- bodies, so showing titles and dates means fetching every file. Caching on the
|
||||
-- sha turns that from one request per post on every page load into one request
|
||||
-- in total, because a sha changes only when the file does. Nothing needs
|
||||
-- invalidating: a row is only ever read for a path the listing still returns.
|
||||
CREATE TABLE IF NOT EXISTS content_cache (
|
||||
path TEXT PRIMARY KEY,
|
||||
sha TEXT NOT NULL,
|
||||
title TEXT NOT NULL DEFAULT '',
|
||||
date TEXT NOT NULL DEFAULT '',
|
||||
categories TEXT NOT NULL DEFAULT '',
|
||||
-- Files carrying `translated_from` are the pipeline's output, not anyone's
|
||||
-- draft. Recorded here so the listing can skip them without re-reading
|
||||
-- every file to find out what it already knew.
|
||||
generated INTEGER NOT NULL DEFAULT 0 CHECK (generated IN (0, 1)),
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
@ -177,7 +177,8 @@ label { display: block; margin: 1rem 0 0.35rem; font-weight: 600; font-size: 0.9
|
||||
label.check { display: flex; align-items: center; gap: 0.5rem; font-weight: 400; }
|
||||
label.check input { margin: 0; }
|
||||
|
||||
input[type="text"], input[type="email"], input:not([type]), select, textarea {
|
||||
input[type="text"], input[type="email"], input[type="date"],
|
||||
input:not([type]), select, textarea {
|
||||
width: 100%;
|
||||
padding: 0.6rem 0.75rem;
|
||||
border: 1px solid var(--border-light);
|
||||
@ -273,3 +274,35 @@ textarea { resize: vertical; line-height: 1.55; }
|
||||
.feed-head { flex-direction: column; align-items: flex-start; }
|
||||
.table { display: block; overflow-x: auto; }
|
||||
}
|
||||
|
||||
/* --- editor ----------------------------------------------------------- */
|
||||
|
||||
.tabs {
|
||||
display: flex;
|
||||
gap: 0.5rem;
|
||||
margin-bottom: 1.25rem;
|
||||
}
|
||||
.tab {
|
||||
padding: 0.4rem 1rem;
|
||||
border-radius: 999px;
|
||||
text-decoration: none;
|
||||
color: var(--muted);
|
||||
background: var(--surface-soft);
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
.tab--on { background: var(--brand-soft); color: var(--brand-dark); font-weight: 600; }
|
||||
|
||||
.fieldset {
|
||||
border: 1px solid var(--border-light);
|
||||
border-radius: var(--radius-md);
|
||||
padding: 0.85rem 1rem 1rem;
|
||||
margin: 1rem 0 0;
|
||||
}
|
||||
.fieldset legend { padding: 0 0.4rem; font-weight: 600; font-size: 0.9rem; }
|
||||
.fieldset .check { margin: 0.35rem 0; }
|
||||
|
||||
input[type="file"] {
|
||||
width: 100%;
|
||||
padding: 0.5rem 0;
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
@ -30,6 +30,9 @@
|
||||
<nav class="site-bar__pages" aria-label="Secciones">
|
||||
<ul class="page-nav">
|
||||
<li><a href="{{ url_for('board.threads') }}">Mensajes</a></li>
|
||||
{% if g.member.role in ('owner', 'admin') %}
|
||||
<li><a href="{{ url_for('content.index') }}">Contenido</a></li>
|
||||
{% endif %}
|
||||
<li><a href="{{ url_for('members.index') }}">Miembros</a></li>
|
||||
</ul>
|
||||
</nav>
|
||||
|
||||
75
apps/board/templates/content_form.html
Normal file
75
apps/board/templates/content_form.html
Normal file
@ -0,0 +1,75 @@
|
||||
{% extends "base.html" %}
|
||||
{% block title %}{{ 'Editar' if item else 'Nuevo' }} {{ meta.singular|lower }}{% endblock %}
|
||||
|
||||
{% set f = fields or {} %}
|
||||
|
||||
{% block main %}
|
||||
{% if preview_html %}
|
||||
<article class="card">
|
||||
<div class="card__meta"><span class="tag tag--quiet">Vista previa</span></div>
|
||||
<h1 class="card__title">{{ f.title }}</h1>
|
||||
<div class="prose">{{ preview_html }}</div>
|
||||
</article>
|
||||
{% endif %}
|
||||
|
||||
<form class="card" method="post" enctype="multipart/form-data"
|
||||
action="{{ url_for('content.edit', collection=collection, name=item.name) if item
|
||||
else url_for('content.new', collection=collection) }}">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
|
||||
{% if item %}
|
||||
<input type="hidden" name="sha" value="{{ item.sha }}">
|
||||
<input type="hidden" name="name" value="{{ item.name }}">
|
||||
{% endif %}
|
||||
|
||||
<h1>{{ 'Editar' if item else 'Nuevo' }} {{ meta.singular|lower }}</h1>
|
||||
|
||||
<label for="title">Título</label>
|
||||
<input id="title" name="title" maxlength="140" required value="{{ f.title or '' }}">
|
||||
|
||||
{% if meta.dated %}
|
||||
<label for="date">Fecha</label>
|
||||
<input id="date" name="date" type="date" required value="{{ f.date or today }}">
|
||||
{% endif %}
|
||||
|
||||
{% if collection == 'post' %}
|
||||
<fieldset class="fieldset">
|
||||
<legend>Categorías</legend>
|
||||
{% for category in categories %}
|
||||
<label class="check">
|
||||
<input type="checkbox" name="categories" value="{{ category }}"
|
||||
{{ 'checked' if category in (f.categories or []) }}>
|
||||
{{ category }}
|
||||
</label>
|
||||
{% endfor %}
|
||||
</fieldset>
|
||||
|
||||
<label for="description">Resumen</label>
|
||||
<textarea id="description" name="description" rows="2">{{ f.description or '' }}</textarea>
|
||||
|
||||
<label for="picture">Imagen destacada</label>
|
||||
{% if f.image %}
|
||||
<p class="muted small">Actual: <span class="mono">{{ f.image }}</span> — sube otra para reemplazarla.</p>
|
||||
<input type="hidden" name="image" value="{{ f.image }}">
|
||||
{% endif %}
|
||||
<input id="picture" name="picture" type="file" accept="image/*">
|
||||
{% endif %}
|
||||
|
||||
<label for="body">Cuerpo</label>
|
||||
<textarea id="body" name="body" rows="18" required
|
||||
placeholder="Markdown: **negrita**, ## títulos, listas, [enlaces](https://…).">{{ body }}</textarea>
|
||||
|
||||
<label class="check">
|
||||
<input type="checkbox" name="manual_translation" {{ 'checked' if f.manual_translation }}>
|
||||
Traducción manual — no generar alemán ni portugués para esto
|
||||
</label>
|
||||
|
||||
<div class="actions">
|
||||
<button class="btn" type="submit">{{ 'Guardar' if item else 'Publicar' }}</button>
|
||||
<button class="linkish" type="submit"
|
||||
formaction="{{ url_for('content.preview', collection=collection) }}">
|
||||
Vista previa
|
||||
</button>
|
||||
<a class="linkish" href="{{ url_for('content.index', collection=collection) }}">Cancelar</a>
|
||||
</div>
|
||||
</form>
|
||||
{% endblock %}
|
||||
57
apps/board/templates/content_list.html
Normal file
57
apps/board/templates/content_list.html
Normal file
@ -0,0 +1,57 @@
|
||||
{% extends "base.html" %}
|
||||
{% block title %}{{ meta.label }}{% endblock %}
|
||||
|
||||
{% block main %}
|
||||
<div class="feed-head">
|
||||
<h1>Contenido</h1>
|
||||
<a class="btn" href="{{ url_for('content.new', collection=collection) }}">
|
||||
Nuevo {{ meta.singular|lower }}
|
||||
</a>
|
||||
</div>
|
||||
|
||||
<nav class="tabs" aria-label="Tipo de contenido">
|
||||
{% for key, other in collections.items() %}
|
||||
<a class="tab {{ 'tab--on' if key == collection }}"
|
||||
href="{{ url_for('content.index', collection=key) }}">{{ other.label }}</a>
|
||||
{% endfor %}
|
||||
</nav>
|
||||
|
||||
{% if not items %}
|
||||
<article class="card">
|
||||
<p class="muted">Todavía no hay nada aquí.</p>
|
||||
</article>
|
||||
{% else %}
|
||||
<table class="table">
|
||||
<thead>
|
||||
<tr><th>Título</th>{% if meta.dated %}<th>Fecha</th>{% endif %}<th>Categorías</th><th></th></tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for item in items %}
|
||||
<tr>
|
||||
<td>
|
||||
<a href="{{ url_for('content.edit', collection=collection, name=item.name) }}">{{ item.title }}</a>
|
||||
<div class="muted small mono">{{ item.name }}</div>
|
||||
</td>
|
||||
{% if meta.dated %}<td>{{ item.date }}</td>{% endif %}
|
||||
<td>{{ item.categories }}</td>
|
||||
<td class="actions actions--row">
|
||||
<a class="linkish" href="{{ url_for('content.edit', collection=collection, name=item.name) }}">Editar</a>
|
||||
<form method="post"
|
||||
action="{{ url_for('content.delete', collection=collection, name=item.name) }}"
|
||||
data-confirm="Se eliminará «{{ item.title }}» del sitio público. ¿Seguro?">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
|
||||
<button class="linkish linkish--danger" type="submit">Eliminar</button>
|
||||
</form>
|
||||
</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
{% endif %}
|
||||
|
||||
<p class="muted small">
|
||||
Se escribe en español. El alemán y el portugués los genera la traducción
|
||||
automática un par de minutos después de publicar, y no se editan aquí: para
|
||||
corregir una traducción a mano, marca «traducción manual» en ese archivo.
|
||||
</p>
|
||||
{% endblock %}
|
||||
@ -23,7 +23,11 @@ def test_anonymous_is_sent_to_login(client, path):
|
||||
|
||||
|
||||
def _stub_gitea(monkeypatch, login):
|
||||
monkeypatch.setattr(gitea, "exchange_code", lambda code, uri: "token")
|
||||
# exchange_code returns the whole token response now, because the editor
|
||||
# needs the refresh token to keep working past Gitea's one-hour expiry.
|
||||
monkeypatch.setattr(gitea, "exchange_code", lambda code, uri: {
|
||||
"access_token": "token", "refresh_token": "refresh", "expires_in": 3600,
|
||||
})
|
||||
monkeypatch.setattr(gitea, "fetch_user", lambda token: {
|
||||
"login": login, "full_name": login.title(), "email": f"{login}@example.com",
|
||||
})
|
||||
|
||||
310
apps/board/tests/test_content.py
Normal file
310
apps/board/tests/test_content.py
Normal file
@ -0,0 +1,310 @@
|
||||
"""The editor.
|
||||
|
||||
Gitea is replaced by a dictionary. What matters here is not that HTTP works —
|
||||
`requests` can be trusted for that — but that the files this produces are the
|
||||
files `scripts/translate.py` expects, because a post the pipeline cannot parse
|
||||
publishes in Spanish and is never translated, silently. Several tests therefore
|
||||
import the real translate.py and run its parser over what the editor wrote.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import io
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from apps.board import content, gitea
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[3]
|
||||
|
||||
|
||||
def _load_translate():
|
||||
"""Import scripts/translate.py directly — it is a script, not a package."""
|
||||
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"vl_translate", REPO_ROOT / "scripts" / "translate.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
translate = _load_translate()
|
||||
|
||||
|
||||
class FakeRepo:
|
||||
"""A repository in a dict, with shas that change when content does."""
|
||||
|
||||
def __init__(self):
|
||||
self.files: dict[str, bytes] = {}
|
||||
self.commits: list[str] = []
|
||||
self.reads = 0
|
||||
|
||||
@staticmethod
|
||||
def _sha(data: bytes) -> str:
|
||||
return hashlib.sha1(data).hexdigest()
|
||||
|
||||
def list_directory(self, path, token=None):
|
||||
out = []
|
||||
for name, data in self.files.items():
|
||||
if name.startswith(path + "/") and "/" not in name[len(path) + 1:]:
|
||||
out.append({"name": name.rsplit("/", 1)[1], "path": name,
|
||||
"type": "file", "sha": self._sha(data)})
|
||||
return out
|
||||
|
||||
def read_file(self, path, token=None):
|
||||
self.reads += 1
|
||||
if path not in self.files:
|
||||
raise gitea.GiteaError("Ese archivo ya no existe.")
|
||||
return self.files[path].decode("utf-8"), self._sha(self.files[path])
|
||||
|
||||
def write_file(self, path, data, message, token=None, sha=None):
|
||||
if sha and self.files.get(path) is not None and self._sha(self.files[path]) != sha:
|
||||
raise gitea.StaleFile("Alguien más guardó este archivo mientras lo editabas.")
|
||||
self.files[path] = data
|
||||
self.commits.append(message)
|
||||
return self._sha(data)
|
||||
|
||||
def delete_file(self, path, sha, message, token=None):
|
||||
self.files.pop(path, None)
|
||||
self.commits.append(message)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo(monkeypatch):
|
||||
fake = FakeRepo()
|
||||
for name in ("list_directory", "read_file", "write_file", "delete_file"):
|
||||
monkeypatch.setattr(gitea, name, getattr(fake, name))
|
||||
return fake
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def editor(db, make_member, sign_in):
|
||||
"""An admin with a stored Gitea token, which every content route needs."""
|
||||
member_id = make_member("editora", role="admin")
|
||||
db.execute(
|
||||
"""INSERT INTO gitea_tokens (member_id, access_token, refresh_token, expires_at)
|
||||
VALUES (?, 'tok', 'ref', NULL)""",
|
||||
(member_id,),
|
||||
)
|
||||
sign_in(member_id)
|
||||
return member_id
|
||||
|
||||
|
||||
def publish(post, title="Sabores del barrio", body="Texto del artículo.", **extra):
|
||||
data = {"title": title, "body": body, "date": "2026-09-25"}
|
||||
data.update(extra)
|
||||
return post("/comunidad/contenido/post/nuevo", data)
|
||||
|
||||
|
||||
# --- the contract with the pipeline --------------------------------------
|
||||
|
||||
def test_the_filename_is_one_translate_py_can_parse(repo, editor, post):
|
||||
publish(post)
|
||||
path, = repo.files
|
||||
assert path == "content/post/2026-09-25-sabores-del-barrio.es.md"
|
||||
assert translate.split_lang(Path(path)) == ("2026-09-25-sabores-del-barrio", "es")
|
||||
|
||||
|
||||
def test_accents_are_folded_not_dropped():
|
||||
assert content.slugify("Gastronomía en Viena") == "gastronomia-en-viena"
|
||||
assert content.slugify("¿Qué comer?") == "que-comer"
|
||||
assert content.slugify("—") == "sin-titulo"
|
||||
|
||||
|
||||
def test_the_frontmatter_marks_an_authored_source(repo, editor, post):
|
||||
publish(post, categories="Gastronomía", description="Un resumen.")
|
||||
path, = repo.files
|
||||
fm, body = translate.split_frontmatter(repo.files[path].decode())
|
||||
assert fm["lang"] == "es"
|
||||
assert fm["manual_translation"] is False
|
||||
assert fm["categories"] == ["Gastronomía"]
|
||||
assert "translated_from" not in fm # what makes it a source, not output
|
||||
assert not translate.is_generated(fm)
|
||||
assert body.strip() == "Texto del artículo."
|
||||
|
||||
|
||||
def test_two_posts_with_one_title_land_on_different_days(repo, editor, post):
|
||||
publish(post, title="Resumen del mes", date="2026-09-25")
|
||||
publish(post, title="Resumen del mes", date="2026-10-25")
|
||||
assert sorted(repo.files) == [
|
||||
"content/post/2026-09-25-resumen-del-mes.es.md",
|
||||
"content/post/2026-10-25-resumen-del-mes.es.md",
|
||||
]
|
||||
|
||||
|
||||
def test_the_freeze_toggle_round_trips(repo, editor, post, client):
|
||||
publish(post, manual_translation="on")
|
||||
path, = repo.files
|
||||
fm, _ = translate.split_frontmatter(repo.files[path].decode())
|
||||
assert translate.is_frozen(fm)
|
||||
|
||||
name = path.rsplit("/", 1)[1]
|
||||
body = client.get(f"/comunidad/contenido/post/editar/{name}").get_data(as_text=True)
|
||||
assert 'name="manual_translation" checked' in body.replace(" ", " ")
|
||||
|
||||
|
||||
def test_pages_are_not_dated(repo, editor, post):
|
||||
post("/comunidad/contenido/page/nuevo", {"title": "Acerca", "body": "Quiénes somos."})
|
||||
assert list(repo.files) == ["content/page/acerca.es.md"]
|
||||
|
||||
|
||||
def test_the_commit_message_does_not_skip_translation(repo, editor, post):
|
||||
publish(post)
|
||||
assert repo.commits and all("[skip-translate]" not in m for m in repo.commits)
|
||||
|
||||
|
||||
# --- concurrency and safety ----------------------------------------------
|
||||
|
||||
def test_a_stale_sha_is_refused_with_something_readable(repo, editor, post, client):
|
||||
publish(post)
|
||||
path, = repo.files
|
||||
name = path.rsplit("/", 1)[1]
|
||||
repo.files[path] = "---\ntitle: Otra cosa\n---\n\nAlguien llegó antes.\n".encode()
|
||||
|
||||
response = post(f"/comunidad/contenido/post/editar/{name}",
|
||||
{"title": "Mi versión", "body": "Texto", "date": "2026-09-25",
|
||||
"sha": "unasha-vieja"})
|
||||
assert response.status_code == 400
|
||||
assert "Alguien más guardó" in response.get_data(as_text=True)
|
||||
assert b"Alguien" in repo.files[path] # the other edit survived
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", [
|
||||
"..%2F..%2Fetc%2Fpasswd", ".woodpecker.yml", "config.yaml",
|
||||
"hola.de.md", "hola.es.md.bak", "a/b.es.md",
|
||||
])
|
||||
def test_only_spanish_source_filenames_are_reachable(repo, editor, client, name):
|
||||
assert client.get(f"/comunidad/contenido/post/editar/{name}").status_code == 404
|
||||
|
||||
|
||||
def test_a_generated_sibling_is_hidden_and_refused(repo, editor, client):
|
||||
repo.files["content/post/hola.es.md"] = (
|
||||
b"---\ntitle: Hola\nlang: es\ntranslated_from: de\n---\n\nGenerado.\n")
|
||||
assert "Hola" not in client.get("/comunidad/contenido").get_data(as_text=True)
|
||||
|
||||
response = client.get("/comunidad/contenido/post/editar/hola.es.md",
|
||||
follow_redirects=True)
|
||||
assert "traducción automática" in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_deleting_leaves_the_siblings_to_the_pipeline(repo, editor, post):
|
||||
publish(post)
|
||||
path, = repo.files
|
||||
repo.files["content/post/2026-09-25-sabores-del-barrio.de.md"] = b"generado"
|
||||
|
||||
post(f"/comunidad/contenido/post/eliminar/{path.rsplit('/', 1)[1]}")
|
||||
assert path not in repo.files
|
||||
# Reaping the German file is translate.py's job, on the next run.
|
||||
assert "content/post/2026-09-25-sabores-del-barrio.de.md" in repo.files
|
||||
|
||||
|
||||
# --- who may publish -----------------------------------------------------
|
||||
|
||||
CONTENT_ROUTES = [
|
||||
"/comunidad/contenido",
|
||||
"/comunidad/contenido/post",
|
||||
"/comunidad/contenido/post/nuevo",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", CONTENT_ROUTES)
|
||||
def test_a_plain_member_cannot_publish(repo, client, make_member, sign_in, path):
|
||||
"""Posting to the board is not the same permission as publishing to the
|
||||
public site, so the editor is admins and the owner only."""
|
||||
sign_in(make_member("vecina"))
|
||||
assert client.get(path).status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", CONTENT_ROUTES)
|
||||
def test_anonymous_is_sent_to_login(repo, client, path):
|
||||
response = client.get(path)
|
||||
assert response.status_code == 302
|
||||
assert "/comunidad/login" in response.headers["Location"]
|
||||
|
||||
|
||||
def test_a_member_without_a_token_is_sent_back_through_gitea(
|
||||
repo, client, make_member, sign_in):
|
||||
sign_in(make_member("sintoken", role="admin"))
|
||||
response = client.get("/comunidad/contenido")
|
||||
assert response.status_code == 302
|
||||
assert "/comunidad/login" in response.headers["Location"]
|
||||
|
||||
|
||||
# --- images --------------------------------------------------------------
|
||||
|
||||
def _image(name="foto.jpg", size=32):
|
||||
return (io.BytesIO(b"x" * size), name)
|
||||
|
||||
|
||||
def test_an_uploaded_image_is_committed_and_referenced(repo, editor, post):
|
||||
publish(post, picture=_image())
|
||||
uploads = [p for p in repo.files if p.startswith("static/uploads/")]
|
||||
assert len(uploads) == 1
|
||||
document = repo.files["content/post/2026-09-25-sabores-del-barrio.es.md"].decode()
|
||||
fm, _ = translate.split_frontmatter(document)
|
||||
assert fm["image"].startswith("/uploads/")
|
||||
assert fm["image"].endswith(".jpg")
|
||||
|
||||
|
||||
def test_two_uploads_of_one_filename_do_not_collide(repo, editor, post):
|
||||
publish(post, title="Uno", picture=_image())
|
||||
publish(post, title="Dos", picture=_image())
|
||||
uploads = [p for p in repo.files if p.startswith("static/uploads/")]
|
||||
assert len(uploads) == 2
|
||||
|
||||
|
||||
def test_a_script_disguised_as_an_image_is_refused(repo, editor, post):
|
||||
response = publish(post, picture=_image("payload.svg"))
|
||||
assert response.status_code == 400
|
||||
assert "Formato de imagen no admitido" in response.get_data(as_text=True)
|
||||
assert not repo.files
|
||||
|
||||
|
||||
def test_an_oversized_image_says_so_instead_of_failing(app, repo, editor, post):
|
||||
app.config["UPLOAD_MAX_BYTES"] = 1024
|
||||
response = publish(post, picture=_image(size=4096))
|
||||
assert response.status_code == 400
|
||||
assert "máximo" in response.get_data(as_text=True)
|
||||
|
||||
|
||||
# --- listing, preview, cache ---------------------------------------------
|
||||
|
||||
def test_the_listing_reads_each_file_once(repo, editor, client):
|
||||
publish_paths = {
|
||||
"content/post/2026-09-01-uno.es.md": b"---\ntitle: Uno\ndate: 2026-09-01\nlang: es\n---\n\nA\n",
|
||||
"content/post/2026-09-02-dos.es.md": b"---\ntitle: Dos\ndate: 2026-09-02\nlang: es\n---\n\nB\n",
|
||||
}
|
||||
repo.files.update(publish_paths)
|
||||
|
||||
client.get("/comunidad/contenido")
|
||||
after_first = repo.reads
|
||||
assert after_first == 2
|
||||
|
||||
body = client.get("/comunidad/contenido").get_data(as_text=True)
|
||||
assert repo.reads == after_first # served from content_cache
|
||||
assert "Uno" in body and "Dos" in body
|
||||
|
||||
|
||||
def test_the_listing_refreshes_when_a_file_changes(repo, editor, client):
|
||||
repo.files["content/post/2026-09-01-uno.es.md"] = (
|
||||
b"---\ntitle: Uno\ndate: 2026-09-01\nlang: es\n---\n\nA\n")
|
||||
client.get("/comunidad/contenido")
|
||||
repo.files["content/post/2026-09-01-uno.es.md"] = (
|
||||
b"---\ntitle: Uno corregido\ndate: 2026-09-01\nlang: es\n---\n\nA\n")
|
||||
body = client.get("/comunidad/contenido").get_data(as_text=True)
|
||||
assert "Uno corregido" in body
|
||||
|
||||
|
||||
def test_the_preview_renders_markdown_and_escapes_html(repo, editor, post):
|
||||
response = post("/comunidad/contenido/post/vista-previa", {
|
||||
"title": "Prueba", "date": "2026-09-25",
|
||||
"body": "**fuerte** y <script>alert(1)</script>",
|
||||
})
|
||||
page = response.get_data(as_text=True)
|
||||
assert "<strong>fuerte</strong>" in page
|
||||
assert "<script>alert(1)</script>" not in page
|
||||
assert not repo.files # a preview must not publish anything
|
||||
97
apps/board/tokens.py
Normal file
97
apps/board/tokens.py
Normal file
@ -0,0 +1,97 @@
|
||||
"""Keeping the member's Gitea token usable for as long as they are signed in.
|
||||
|
||||
Gitea's OAuth access tokens expire after about an hour. A writer who opened the
|
||||
editor after lunch and saved at three would otherwise get a failure with no
|
||||
explanation and no way to act on it, so this refreshes ahead of expiry and
|
||||
retries once when Gitea rejects a token anyway.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from flask import g
|
||||
|
||||
from . import gitea
|
||||
from .db import get_db
|
||||
|
||||
# Refresh this far before the stated expiry. A token that dies mid-request is
|
||||
# indistinguishable to the writer from the app being broken.
|
||||
EARLY = timedelta(minutes=5)
|
||||
|
||||
|
||||
class NeedsSignIn(RuntimeError):
|
||||
"""The token is gone or unrefreshable — send them through Gitea again."""
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def save(member_id: int, payload: dict) -> None:
|
||||
expires_in = payload.get("expires_in")
|
||||
expires_at = (
|
||||
(_now() + timedelta(seconds=int(expires_in))).isoformat()
|
||||
if expires_in else None
|
||||
)
|
||||
get_db().execute(
|
||||
"""INSERT INTO gitea_tokens (member_id, access_token, refresh_token, expires_at)
|
||||
VALUES (?, ?, ?, ?)
|
||||
ON CONFLICT(member_id) DO UPDATE SET
|
||||
access_token = excluded.access_token,
|
||||
refresh_token = excluded.refresh_token,
|
||||
expires_at = excluded.expires_at,
|
||||
updated_at = datetime('now')""",
|
||||
(member_id, payload["access_token"], payload.get("refresh_token", ""), expires_at),
|
||||
)
|
||||
|
||||
|
||||
def forget(member_id: int) -> None:
|
||||
get_db().execute("DELETE FROM gitea_tokens WHERE member_id = ?", (member_id,))
|
||||
|
||||
|
||||
def _stored(member_id: int):
|
||||
return get_db().execute(
|
||||
"SELECT * FROM gitea_tokens WHERE member_id = ?", (member_id,)
|
||||
).fetchone()
|
||||
|
||||
|
||||
def _refresh(row) -> str:
|
||||
if not row["refresh_token"]:
|
||||
raise NeedsSignIn()
|
||||
try:
|
||||
payload = gitea.refresh_token(row["refresh_token"])
|
||||
except gitea.GiteaError as exc:
|
||||
raise NeedsSignIn() from exc
|
||||
save(row["member_id"], payload)
|
||||
return payload["access_token"]
|
||||
|
||||
|
||||
def access_token(member_id: int) -> str:
|
||||
row = _stored(member_id)
|
||||
if row is None:
|
||||
raise NeedsSignIn()
|
||||
if row["expires_at"]:
|
||||
expires = datetime.fromisoformat(row["expires_at"])
|
||||
if _now() + EARLY >= expires:
|
||||
return _refresh(row)
|
||||
return row["access_token"]
|
||||
|
||||
|
||||
def with_token(call, *args, **kwargs):
|
||||
"""Run a Gitea call with the current member's token, refreshing once if it
|
||||
is rejected.
|
||||
|
||||
The retry exists because expiry is not the only reason a token stops
|
||||
working — it can be revoked in Gitea, or invalidated by a password change —
|
||||
and in those cases the clock says the token is still fine.
|
||||
"""
|
||||
member_id = g.member["id"]
|
||||
token = access_token(member_id)
|
||||
try:
|
||||
return call(*args, token=token, **kwargs)
|
||||
except PermissionError:
|
||||
row = _stored(member_id)
|
||||
if row is None:
|
||||
raise NeedsSignIn()
|
||||
return call(*args, token=_refresh(row), **kwargs)
|
||||
@ -434,3 +434,70 @@ Members' names, emails and writing are personal data under GDPR.
|
||||
- **Retention:** soft-deleted posts stay in the database until removed by hand.
|
||||
If you want a real retention limit, that is a `DELETE ... WHERE deleted_at <`
|
||||
in this same cron slot — and a decision to take deliberately, not by default.
|
||||
|
||||
### 11.8 The content editor (`/comunidad/contenido/`)
|
||||
|
||||
Admins and the owner can write, edit and delete posts and pages from inside the
|
||||
members area, instead of Decap at `/admin/`.
|
||||
|
||||
**Decap is still there and still works.** Nothing was removed. Use the new
|
||||
editor for a few real posts first; if something turns out to be missing, switch
|
||||
tabs. Removing Decap is a separate decision — see below.
|
||||
|
||||
Nothing extra to install or configure: it runs in the container already serving
|
||||
`/comunidad/`, and commits through the Gitea OAuth application registered in
|
||||
§11.1. Two settings exist if the repository is ever renamed:
|
||||
|
||||
```
|
||||
CONTENT_REPO=pablo/vienalatina # owner/repo inside Gitea
|
||||
CONTENT_BRANCH=main
|
||||
```
|
||||
|
||||
**How publishing works.** The editor is a form that commits a file through
|
||||
Gitea's contents API. Gitea's webhook fires Woodpecker, and translate → build →
|
||||
deploy runs exactly as it does for a Decap commit — the pipeline cannot tell
|
||||
which editor wrote the file, which is what makes running both at once safe.
|
||||
|
||||
**Commits are made with your own account**, not a bot's, so `git log` shows who
|
||||
wrote each post and Gitea's permissions apply unchanged. Your access token is
|
||||
stored in the members-area database (never in a cookie) and refreshed
|
||||
automatically; Gitea expires them after about an hour, and without refreshing,
|
||||
saving would start failing mid-afternoon for no visible reason.
|
||||
|
||||
**Filenames follow the same rules Decap used**, because `scripts/translate.py`
|
||||
reads them: `YYYY-MM-DD-slug.es.md` for posts, `slug.es.md` for pages. A file
|
||||
whose name breaks that contract publishes in Spanish and is never translated,
|
||||
with nothing reported anywhere — which is why the tests import translate.py and
|
||||
run its parser over what the editor writes.
|
||||
|
||||
**Two people editing one post** is a visible conflict, not a silent overwrite:
|
||||
the form carries the file's git sha and Gitea rejects a write whose sha has
|
||||
moved on. You are asked to reopen the post rather than losing the other edit.
|
||||
|
||||
**Images** are committed as a second, separate commit before the post itself,
|
||||
so publishing with a picture produces two pipeline runs. Harmless, and the
|
||||
alternative — batching both into one commit via the git trees API — is
|
||||
considerably more code for something nobody sees.
|
||||
|
||||
**What it deliberately does not do:** rich-text editing (markdown with a
|
||||
preview button instead), a media library, drafts, or editing the generated
|
||||
German and Portuguese files. Those stay the pipeline's, and a hand-written
|
||||
translation is still frozen with `manual_translation: true`.
|
||||
|
||||
#### Worth tightening later
|
||||
|
||||
The OAuth application requests no explicit scope, so Gitea grants the default —
|
||||
full access to the account, which is more than the editor needs. Narrowing it to
|
||||
`read:user write:repository` is a one-line change in `apps/board/gitea.py`'s
|
||||
`authorize_url()`, but it invalidates existing authorisations: everyone has to
|
||||
approve the app again. Worth doing while the member list is short, and worth
|
||||
testing on a throwaway account first, since a wrong scope string breaks sign-in
|
||||
for everybody.
|
||||
|
||||
#### Removing Decap, once you are confident
|
||||
|
||||
Not urgent — leaving it costs a folder and one `wget` in the pipeline:
|
||||
|
||||
1. `rm -rf static/admin/`
|
||||
2. Delete the `wget … decap-cms.js` line from `.woodpecker.yml`
|
||||
3. Delete the `decap-cms` OAuth application in Gitea
|
||||
|
||||
Loading…
Reference in New Issue
Block a user