hermes
2a9e0ad15e
Round-2 review findings.
- `_ensure_indexes` swallowed a `PyMongoError` with a bare `pass`. An index
failure does not stop the app serving, but it means the lookup is a
collection scan, which gets slow as the data grows — so it is now logged
as a warning with the traceback.
- The Content-Security-Policy string is now a module-level `CSP` constant
rather than inline in the response handler, and it gained
`object-src 'none'` (the app uses no plugins). The test now also asserts
that neither `unsafe-inline` nor `unsafe-eval` appears — either would
defeat the point of the policy.
- `api_run` returned a bare dict for its 404 while the success path used
`jsonify`. It is now `jsonify({"error": "not found"})` for consistency.
237 lines
8.4 KiB
Python
237 lines
8.4 KiB
Python
"""The brew-process pages: record readings and review the runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from flask import (Blueprint, flash, jsonify, redirect, render_template,
|
|
request, session, url_for)
|
|
from flask_wtf import FlaskForm
|
|
from wtforms import (BooleanField, DateField, IntegerField, StringField,
|
|
SubmitField, TextAreaField)
|
|
from wtforms.validators import DataRequired, Length, NumberRange, Optional
|
|
|
|
from brewing import is_plausible_sg, run_summary
|
|
|
|
bp = Blueprint("brews", __name__)
|
|
|
|
|
|
def repo():
|
|
"""The request-scoped repository (one Mongo client per request)."""
|
|
from flask import current_app
|
|
|
|
return current_app.extensions["beer_repo"]()
|
|
|
|
|
|
def login_required(fn):
|
|
"""Redirect to the login page when there is no session."""
|
|
from functools import wraps
|
|
|
|
@wraps(fn)
|
|
def wrapper(*args, **kwargs):
|
|
if not session.get("logged_in"):
|
|
return redirect(url_for("auth.login", next=request.path))
|
|
return fn(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def _last_reading_date(run) -> str | None:
|
|
"""The ISO date of the most recent reading in a run."""
|
|
last = run.last_reading()
|
|
if last is None or last.date is None:
|
|
return None
|
|
return last.date.isoformat()
|
|
|
|
|
|
class ReadingForm(FlaskForm):
|
|
"""Add or update a gravity reading."""
|
|
|
|
beer_run_id = IntegerField("Run number", validators=[DataRequired()])
|
|
beer_run_type = StringField("Beer / cider name",
|
|
validators=[Optional(), Length(max=120)])
|
|
Date = DateField("Reading date", validators=[DataRequired()])
|
|
sg = IntegerField("Gravity reading (points)",
|
|
validators=[DataRequired(), NumberRange(min=900, max=1500)])
|
|
final_reading = BooleanField("This is the final reading")
|
|
comment = TextAreaField("Comment", validators=[Optional(), Length(max=2000)])
|
|
submit = SubmitField("Save reading")
|
|
|
|
|
|
class RunMetaForm(FlaskForm):
|
|
"""Edit the canonical metadata of a run.
|
|
|
|
These are the details that cannot be derived from the readings: the name
|
|
the brew is actually known by, which tap it is on, and its original
|
|
gravity.
|
|
"""
|
|
|
|
name = StringField("Beer / cider name", validators=[Optional(), Length(max=120)])
|
|
tap_number = IntegerField("Tap number",
|
|
validators=[Optional(), NumberRange(min=1, max=99)])
|
|
og_points = IntegerField("Original gravity (points)",
|
|
validators=[Optional(), NumberRange(min=900, max=1500)])
|
|
notes = TextAreaField("Brew notes", validators=[Optional(), Length(max=4000)])
|
|
submit = SubmitField("Save run")
|
|
|
|
|
|
@bp.route("/brews", methods=["GET", "POST"])
|
|
@login_required
|
|
def index():
|
|
"""The brew-process page: record a reading, then review the runs."""
|
|
form = ReadingForm()
|
|
|
|
if form.validate_on_submit():
|
|
sg_points = int(form.sg.data)
|
|
if not is_plausible_sg(sg_points):
|
|
flash(
|
|
f"{sg_points} is outside the usual hydrometer range "
|
|
"(990-1200). It was saved anyway — double-check the reading.",
|
|
"warning",
|
|
)
|
|
repo().upsert_reading(
|
|
beer_run_id=str(form.beer_run_id.data),
|
|
beer_type=(form.beer_run_type.data or "").strip(),
|
|
sg=sg_points,
|
|
reading_date=form.Date.data.isoformat(),
|
|
final_reading=bool(form.final_reading.data),
|
|
comment=(form.comment.data or "").strip(),
|
|
)
|
|
flash("Reading saved.", "success")
|
|
return redirect(url_for("brews.index"))
|
|
|
|
runs = repo().runs()
|
|
metadata = repo().all_run_metadata()
|
|
|
|
summaries = []
|
|
for run in runs:
|
|
meta = metadata.get(run.beer_run_id, {})
|
|
summary = run_summary(run, og_override=meta.get("og_points"))
|
|
summary["canonical_name"] = meta.get("name") or run.display_name()
|
|
summary["tap_number"] = meta.get("tap_number")
|
|
summary["notes"] = meta.get("notes", "")
|
|
summary["last_date"] = _last_reading_date(run)
|
|
summaries.append(summary)
|
|
|
|
# Active runs: most recent activity first — that is what is fermenting now.
|
|
active = sorted([s for s in summaries if s["active"]],
|
|
key=lambda s: s["last_date"] or "", reverse=True)
|
|
# Finished runs: most recently finished first.
|
|
completed = sorted([s for s in summaries if not s["active"]],
|
|
key=lambda s: s["final_date"] or "", reverse=True)
|
|
|
|
return render_template(
|
|
"brews.html",
|
|
form=form,
|
|
summaries=summaries,
|
|
active_runs=active,
|
|
completed_runs=completed,
|
|
next_tap=repo().next_free_tap_number(),
|
|
)
|
|
|
|
|
|
@bp.route("/runs/<run_id>", methods=["GET", "POST"])
|
|
@login_required
|
|
def run_detail(run_id: str):
|
|
"""One run in detail, with its readings and its editable metadata."""
|
|
run = repo().run(run_id)
|
|
if run is None:
|
|
flash(f"No run numbered {run_id}.", "error")
|
|
return redirect(url_for("brews.index"))
|
|
|
|
meta = repo().run_metadata(run_id)
|
|
meta_form = RunMetaForm()
|
|
|
|
if meta_form.validate_on_submit():
|
|
repo().upsert_run_metadata(
|
|
run_id,
|
|
name=(meta_form.name.data or "").strip() or None,
|
|
tap_number=meta_form.tap_number.data,
|
|
og_points=meta_form.og_points.data,
|
|
notes=(meta_form.notes.data or "").strip() or None,
|
|
)
|
|
flash("Run details saved.", "success")
|
|
return redirect(url_for("brews.run_detail", run_id=run_id))
|
|
|
|
if not meta_form.is_submitted():
|
|
meta_form.name.data = meta.get("name") or run.display_name()
|
|
meta_form.tap_number.data = meta.get("tap_number")
|
|
meta_form.og_points.data = meta.get("og_points")
|
|
meta_form.notes.data = meta.get("notes", "")
|
|
|
|
summary = run_summary(run, og_override=meta.get("og_points"))
|
|
summary["canonical_name"] = meta.get("name") or run.display_name()
|
|
summary["tap_number"] = meta.get("tap_number")
|
|
summary["notes"] = meta.get("notes", "")
|
|
|
|
return render_template("run_detail.html", run=run, summary=summary,
|
|
meta_form=meta_form, meta=meta)
|
|
|
|
|
|
@bp.route("/api/runs")
|
|
@login_required
|
|
def api_runs():
|
|
"""JSON feed of the run summaries, used by the charts on the analysis
|
|
pages.
|
|
|
|
The charts fetch this rather than having the data inlined into the page,
|
|
so one fetch serves every chart on a page.
|
|
"""
|
|
runs = repo().runs()
|
|
metadata = repo().all_run_metadata()
|
|
|
|
payload = []
|
|
for run in runs:
|
|
meta = metadata.get(run.beer_run_id, {})
|
|
summary = run_summary(run, og_override=meta.get("og_points"))
|
|
summary["canonical_name"] = meta.get("name") or run.display_name()
|
|
summary["tap_number"] = meta.get("tap_number")
|
|
summary["last_date"] = _last_reading_date(run)
|
|
payload.append(summary)
|
|
|
|
return jsonify(payload)
|
|
|
|
|
|
@bp.route("/api/runs/<run_id>")
|
|
@login_required
|
|
def api_run(run_id: str):
|
|
"""JSON detail of a single run."""
|
|
run = repo().run(run_id)
|
|
if run is None:
|
|
return jsonify({"error": "not found"}), 404
|
|
|
|
meta = repo().run_metadata(run_id)
|
|
summary = run_summary(run, og_override=meta.get("og_points"))
|
|
summary["canonical_name"] = meta.get("name") or run.display_name()
|
|
summary["tap_number"] = meta.get("tap_number")
|
|
return jsonify(summary)
|
|
|
|
|
|
@bp.route("/runs/<run_id>/final", methods=["POST"])
|
|
@login_required
|
|
def mark_final(run_id: str):
|
|
"""Flag a reading as the final one for its run."""
|
|
reading_date = request.form.get("date", "")
|
|
if not reading_date:
|
|
flash("No reading date supplied.", "error")
|
|
return redirect(url_for("brews.run_detail", run_id=run_id))
|
|
|
|
changed = repo().set_final_reading(run_id, reading_date, True)
|
|
if changed:
|
|
flash("Final reading recorded.", "success")
|
|
else:
|
|
flash("That reading could not be updated.", "error")
|
|
return redirect(url_for("brews.run_detail", run_id=run_id))
|
|
|
|
|
|
@bp.route("/runs/<run_id>/unfinal", methods=["POST"])
|
|
@login_required
|
|
def unfinal(run_id: str):
|
|
"""Un-flag a final reading, putting the run back into the active list."""
|
|
reading_date = request.form.get("date", "")
|
|
if not reading_date:
|
|
flash("No reading date supplied.", "error")
|
|
return redirect(url_for("brews.run_detail", run_id=run_id))
|
|
|
|
repo().set_final_reading(run_id, reading_date, False)
|
|
flash("Reading is no longer marked final.", "info")
|
|
return redirect(url_for("brews.run_detail", run_id=run_id))
|