# SPDX-FileCopyrightText: 2023-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Compatibility-focused add-entry route refactor.
This pass keeps the route close to the current LabFrog helper/template contract
while still using the newer typed field/payload helpers where they fit cleanly.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from flask import (
Blueprint,
current_app,
flash,
g,
redirect,
render_template,
request,
session,
url_for,
)
from flask_login import login_required
from labfrog.campaign_layout_setup import (
build_campaign_layout_setup_context,
handle_campaign_layout_setup_submission,
)
from labfrog.common import (
get_cached_mediawiki_campaigns,
get_cached_mediawiki_targets,
init_request_state,
)
from labfrog.db import (
clear_navigation_caches,
ensure_runtime_indexes,
fetch_shot_navigation_state,
get_db,
)
from labfrog.entry_route_common import (
add_utc_datetime_metadata as _add_utc_datetime_metadata,
)
from labfrog.entry_route_common import (
apply_clear_link as _apply_clear_link,
)
from labfrog.entry_route_common import (
build_context_counters as _build_context_counters,
)
from labfrog.entry_route_common import (
build_entry_page_context as _build_entry_page_context,
)
from labfrog.entry_route_common import (
coerce_datetime_field as _coerce_datetime_field,
)
from labfrog.entry_route_common import (
prefill_form_from_current_doc as _prefill_form_from_current_doc,
)
from labfrog.entry_route_common import (
prepare_submission_payload_for_entry as _prepare_submission_payload_for_entry,
)
from labfrog.entry_route_common import (
resolve_mode_and_campaign as _resolve_mode_and_campaign,
)
from labfrog.field_specs import (
FIELD_VALUE_META_KEY,
INCLUDE_DIAG_DETAILS_FIELD,
normalize_custom_field_type,
)
from labfrog.form_factory import create_dynamic_form
from labfrog.form_feedback import (
current_username,
flash_demo_limit_reached,
flash_form_errors,
)
from labfrog.helpers.choices import ordered_unique_strings
from labfrog.helpers.demo import demo_limit_reached, demo_user_filter
from labfrog.helpers.layout import campaign_layout_missing, default_allowed_for_campaign
from labfrog.helpers.target import _normalize_target_payload
from labfrog.kafka_listener import get_latest_kafka_snapshot
from labfrog.offline_spool import record_unsaved_entry
from labfrog.request_state import (
build_route_form_state,
choose_default_campaign_name,
collect_visible_custom_fields,
ensure_campaign_layout_session_state,
load_custom_field_documents,
)
from labfrog.shot_details import (
get_total_number_of_shots,
int_current_number_of_sets,
update_indiv_count_NEW_SHOT,
)
from labfrog.target_series import (
build_series_fields_for_doc,
clear_series,
finish_series,
series_context_for_template,
start_series,
)
[docs]
@dataclass(frozen=True)
class SubmitResult:
handled: bool = False
response: object | None = None
blueprint = Blueprint("add_entry", __name__)
ENTRY_RECORDED_EVENT_SESSION_KEY = "labfrog_entry_recorded_event"
def _json_safe_event_value(value):
if value is None:
return None
if hasattr(value, "isoformat"):
return value.isoformat()
if isinstance(value, (str, int, float, bool)):
return value
return str(value)
def _entry_recorded_event_payload(data: dict, *, mode: str, campaign: str) -> dict:
"""Return a small browser event payload for newly recorded entries."""
number_field = "set_number" if mode == "set" else "shot_number"
return {
"type": "entry-recorded",
"mode": mode,
"campaign": campaign,
"number": _json_safe_event_value(data.get(number_field)),
"number_field": number_field,
"date_time": _json_safe_event_value(data.get("date_time")),
}
def _handle_submit(
*,
form,
mode: str,
visible_fields: list[str],
always_include: list[str],
custom_field_lookup: dict[str, dict],
shots,
nav_help: dict,
target_dict: dict,
campaign_dict: dict,
):
if mode == "set" and request.form.get("set_number") not in [None, ""]:
set_number_int = int(request.form.get("set_number"))
nav_help["set_number"] = set_number_int - 1
session["set_number"] = set_number_int - 1
nav_help["UPCOMING_SET_NUMBER"] = set_number_int
session["UPCOMING_SET_NUMBER"] = set_number_int
if mode == "shot" and request.form.get("shot_number") not in [None, ""]:
shot_number_int = int(request.form.get("shot_number"))
nav_help["shot_number"] = shot_number_int - 1
nav_help["UPCOMING_SHOT_NUMBER"] = shot_number_int
session["shot_number"] = shot_number_int - 1
session["UPCOMING_SHOT_NUMBER"] = shot_number_int
if not (
"submit" in request.form
and form.validate_on_submit()
and request.form.to_dict().get("date_time") not in [None, ""]
):
return SubmitResult()
user = current_username()
data, field_value_meta = _prepare_submission_payload_for_entry(
form,
mode=mode,
visible_fields=visible_fields,
always_include=always_include,
custom_field_lookup=custom_field_lookup,
diagnostic_names=set(g.get("DIAG_NAMES", {}).keys()),
shots_collection=shots,
campaign_choices=campaign_dict,
)
data, field_value_meta = _normalize_target_payload(
data, field_value_meta, target_choices=target_dict
)
try:
data["date_time"] = _coerce_datetime_field(request.form.get("date_time"))
_add_utc_datetime_metadata(data)
except Exception as exc:
logging.error("Error updating date_time field: %s", exc)
session.pop("_flashes", None)
flash(f"Error updating date_time field: {exc}", "error")
return SubmitResult(handled=True)
series_fields = build_series_fields_for_doc(session)
new_doc = {"version": 0, "user": user, **data, **series_fields, "status": "active"}
if field_value_meta:
new_doc[FIELD_VALUE_META_KEY] = field_value_meta
new_doc[INCLUDE_DIAG_DETAILS_FIELD] = True
if demo_limit_reached(shots, user):
flash_demo_limit_reached()
return SubmitResult(handled=True)
try:
kafka_snap = get_latest_kafka_snapshot(current_app._get_current_object())
if kafka_snap is not None:
new_doc["kafka"] = kafka_snap
except Exception as exc:
logging.warning("Could not attach Kafka snapshot to shot record: %s", exc)
try:
shots.insert_one(new_doc)
clear_navigation_caches()
if mode == "set":
shot_number_list = update_indiv_count_NEW_SHOT(
data.get("set_number"),
nav_help.get("UPCOMING_SHOT_NUMBER"),
data.get("set_length"),
)
nav_help["set_number"] = int(data.get("set_number"))
session["set_number"] = int(data.get("set_number")) + 1
nav_help["UPCOMING_SET_NUMBER"] = int(data.get("set_number")) + 1
session["UPCOMING_SET_NUMBER"] = int(data.get("set_number")) + 2
nav_help["THIS_SHOT_NUMBER_LIST"] = shot_number_list
nav_help["UPCOMING_SHOT_NUMBER"] = (
max(shot_number_list) + 1 if shot_number_list else 1
)
else:
nav_help["shot_number"] = int(data.get("shot_number"))
session["shot_number"] = int(data.get("shot_number")) + 1
nav_help["UPCOMING_SHOT_NUMBER"] = int(data.get("shot_number")) + 1
session["UPCOMING_SHOT_NUMBER"] = int(data.get("shot_number")) + 2
session.pop("_flashes", None)
flash("Set Recorded." if mode == "set" else "Shot Recorded.", "success")
session["clear_date_time_on_next_add"] = True
session[ENTRY_RECORDED_EVENT_SESSION_KEY] = _entry_recorded_event_payload(
data,
mode=mode,
campaign=data.get("Campaign") or session.get("selected_campaign", ""),
)
return SubmitResult(
handled=True, response=redirect(url_for("add_entry.add_entry"))
)
except Exception as exc:
# Log the detail, show the operator a plain message: a driver error can
# carry the host, port and auth details of the database connection.
logging.exception("Error saving to mongodb: %s", exc)
session.pop("_flashes", None)
try:
spooled = record_unsaved_entry(new_doc, mode=mode, reason=str(exc))
except Exception:
# Nowhere left to put it. Say so rather than implying it was kept.
logging.exception("Could not spool the unsaved entry")
flash(
"Failed to record entry, and it could not be saved to this "
"computer either. Your input is still on screen -- copy it "
"somewhere safe before navigating away.",
"error",
)
return SubmitResult(handled=True)
# Deliberately no redirect: the operator keeps the values on screen so
# they can retry or correct them. The exception is never interpolated
# here, because a driver error carries the database host and password.
flash(
f"Failed to record entry: the database could not be reached. Your "
f"{'set' if mode == 'set' else 'shot'} was saved on this computer "
f"as {spooled.name} and nothing has been lost. Open the offline "
"queue to send it once the connection is back.",
"warning",
)
return SubmitResult(handled=True)
[docs]
@blueprint.route("/tab1", methods=["GET", "POST"]) # legacy alias
@blueprint.route("/add_shot", methods=["GET", "POST"]) # legacy alias
@blueprint.route("/add_entry", methods=["GET", "POST"]) # canonical public path
@login_required
def add_entry():
init_request_state()
ensure_runtime_indexes()
if request.path in {"/tab1", "/add_shot"}:
logging.getLogger(__name__).warning(
"Route alias %s is legacy and will be removed in a future release; "
"use /add_entry.",
request.path,
)
target_dict = get_cached_mediawiki_targets()
campaign_dict = get_cached_mediawiki_campaigns()
parameters_db = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
db = get_db()
parameters_collection = db[parameters_db]
custom_fields_collection = db["custom_fields_app"]
shots = db["shots"]
mode, campaign = _resolve_mode_and_campaign(
default_campaign=choose_default_campaign_name()
)
campaign_setup_needed = campaign_layout_missing(
mode, campaign, parameters_collection
) and not default_allowed_for_campaign(mode, campaign)
if request.method == "POST" and request.form.get("setup_method"):
response = handle_campaign_layout_setup_submission(
parameters_collection,
mode,
campaign,
request.path,
request.form,
)
if response is not None:
return response
if not campaign_setup_needed:
ensure_campaign_layout_session_state(
mode, campaign, parameters_collection=parameters_collection
)
route_state = build_route_form_state(
mode, parameters_collection, apply_demo_defaults_now=True
)
visible_fields = list(route_state["visible_fields"])
selected_layout = route_state["selected_layout"]
always_include = route_state["always_include"]
options_collection = route_state["options_collection"]
custom_field_docs, custom_field_lookup = load_custom_field_documents(
custom_fields_collection
)
form = create_dynamic_form(
collection=parameters_db,
options_collection=options_collection,
mode=mode,
visible_fields=visible_fields,
)
visible_fields = ordered_unique_strings([
*(
field_name
for fields in getattr(form, "FIELD_SECTIONS", {}).values()
for field_name in (fields or [])
),
*g.get("DIAG_NAMES_VISIBLE", []),
])
visible_fields = [
field_name
for field_name in visible_fields
if field_name not in {"shot_day", "shot_group"}
]
custom_field_types, custom_fields_by_section = collect_visible_custom_fields(
form, visible_fields, custom_field_docs
)
current_index = int(request.args.get("index", default="0"))
all_docs_query = demo_user_filter({
"status": "active",
"mode": mode,
"Campaign": campaign,
})
nav_state = fetch_shot_navigation_state(
shots, all_docs_query, current_index=current_index, mode=mode
)
total_entries = nav_state["total_entries"]
current_index = nav_state["current_index"]
prev_index = nav_state["prev_index"]
next_index = nav_state["next_index"]
first_entry = nav_state["first_entry"]
last_entry = nav_state["last_entry"]
latest_doc = nav_state["latest_doc"]
current_doc = nav_state["current_doc"]
current_doc_id = str(current_doc["_id"]) if current_doc is not None else ""
has_entries = total_entries > 0
at_latest_entry = (not has_entries) or current_index <= 0
at_oldest_entry = (not has_entries) or current_index >= (total_entries - 1)
counters = _build_context_counters(
mode=mode,
latest_doc=latest_doc,
current_doc=current_doc,
has_entries=has_entries,
)
nav_help = counters["nav_help"]
view_current_entry = counters["view_current_entry"]
next_fire_entry = counters["next_fire_entry"]
if getattr(form, "Campaign", None) is not None:
form.Campaign.data = campaign
if getattr(form, "mode", None) is not None:
form.mode.data = mode
if request.method != "POST" and current_doc is not None:
should_respect_copy_forward = "index" not in request.args
form = _prefill_form_from_current_doc(
form,
current_doc,
parameters_db=parameters_db,
options_collection=options_collection,
mode=mode,
visible_fields=visible_fields,
custom_field_lookup=custom_field_lookup,
respect_copy_forward=should_respect_copy_forward,
)
previous_source_values = {}
if mode == "shot":
previous_doc = None
# On default Add Entry (no index), compare against the latest shot.
# When browsing by index, compare against the shot immediately before that shot.
is_browsing_history = "index" in request.args
if not is_browsing_history and current_doc is not None:
previous_doc = current_doc
else:
shot_number_for_previous = None
if current_doc is not None:
shot_number_for_previous = int(current_doc.get("shot_number", 0) or 0)
elif latest_doc is not None:
shot_number_for_previous = (
int(latest_doc.get("shot_number", 0) or 0) + 1
)
if shot_number_for_previous:
previous_doc = shots.find_one(
{
"status": "active",
"mode": "shot",
"Campaign": campaign,
"shot_number": {"$lt": shot_number_for_previous},
},
sort=[("shot_number", -1), ("date_time", -1), ("_id", -1)],
)
if isinstance(previous_doc, dict):
for field_name, field_doc in custom_field_lookup.items():
if (
normalize_custom_field_type(field_doc.get("custom_field_type"))
!= "calculated_prev_shot"
):
continue
source_field_name = str(field_doc.get("calculated_from") or "").strip()
if not source_field_name:
continue
source_value = previous_doc.get(source_field_name)
if source_value not in (None, ""):
previous_source_values[field_name] = source_value
if (
session.pop("clear_date_time_on_next_add", False)
and getattr(form, "date_time", None) is not None
):
form.date_time.data = None
clear_viewing = False
if request.args.get("link") == "CLEAR":
form = _apply_clear_link(
parameters_db=parameters_db,
options_collection=options_collection,
mode=mode,
visible_fields=visible_fields,
)
clear_viewing = True
view_current_entry = None
session.pop("_flashes", None)
flash("Form cleared!", "success")
if request.method == "POST":
submit_result = _handle_submit(
form=form,
mode=mode,
visible_fields=visible_fields,
always_include=always_include,
custom_field_lookup=custom_field_lookup,
shots=shots,
nav_help=nav_help,
target_dict=target_dict,
campaign_dict=campaign_dict,
)
if submit_result.response is not None:
return submit_result.response
if not submit_result.handled:
flash_form_errors(form, logger=logging.getLogger(__name__))
if mode == "set":
total_number_of_shots = get_total_number_of_shots()
current_number_of_sets = int_current_number_of_sets()
else:
total_number_of_shots = counters["latest_shot_number"]
current_number_of_sets = None
page_context = _build_entry_page_context(
shots_collection=shots,
all_docs_query=all_docs_query,
current_doc=current_doc,
campaign=campaign,
)
series_ctx = series_context_for_template(session)
kafka_data = current_doc.get("kafka") if isinstance(current_doc, dict) else None
return render_template(
"add_entry.html",
custom_error={},
kafka_data=kafka_data,
mode=mode,
nav_help=nav_help,
selected_layout=selected_layout,
form=form,
section_colors=getattr(form, "SECTION_COLORS", {}),
current_index=current_index,
current_doc_id=current_doc_id,
prev_index=prev_index,
next_index=next_index,
first_entry=first_entry,
last_entry=last_entry,
has_entries=has_entries,
at_latest_entry=at_latest_entry,
at_oldest_entry=at_oldest_entry,
mediawiki_status=page_context["mediawiki_status"],
view_current_entry=view_current_entry,
next_fire_entry=next_fire_entry,
clear_viewing=clear_viewing,
dynamic_fields_length=page_context["dynamic_fields_length"],
dynamic_field_names=page_context["dynamic_field_names"],
target_dict=target_dict,
campaign_dict=campaign_dict,
campaign=campaign,
visible_fields=visible_fields,
always_include=always_include,
custom_fields_by_section=custom_fields_by_section,
custom_field_types=custom_field_types,
total_number_of_shots=total_number_of_shots,
current_number_of_sets=current_number_of_sets,
campaign_days=page_context["campaign_days"],
current_shot_day=page_context["current_shot_day"],
current_shot_day_label=page_context["current_shot_day_label"],
current_shot_day_scope=page_context["current_shot_day_scope"],
current_shot_day_badge=page_context["current_shot_day_badge"],
field_aliases=page_context["field_aliases"],
previous_source_values=previous_source_values,
entry_recorded_event=session.pop(ENTRY_RECORDED_EVENT_SESSION_KEY, None),
campaign_setup_needed=campaign_setup_needed,
campaign_setup_context=(
build_campaign_layout_setup_context(
parameters_collection, mode, campaign, request.path
)
if campaign_setup_needed
else None
),
**series_ctx,
)
[docs]
@blueprint.route("/add_entry/series/start", methods=["POST"])
@login_required
def start_target_series():
"""Start a new same-target series and store it in the session."""
from flask import jsonify
label = request.form.get("label", "").strip()
sample = request.form.get("sample", "").strip()
notes = request.form.get("notes", "").strip()
planned_count_raw = request.form.get("planned_count", "").strip()
planned_count = None
if planned_count_raw:
try:
planned_count = int(planned_count_raw)
except ValueError:
return (
jsonify({"ok": False, "error": "planned_count must be an integer"}),
400,
)
state = start_series(
session,
label=label,
sample=sample,
planned_count=planned_count,
notes=notes,
)
return jsonify({"ok": True, "series_id": state["id"], "state": state})
[docs]
@blueprint.route("/add_entry/series/finish", methods=["POST"])
@login_required
def finish_target_series():
"""Close the active same-target series."""
from flask import jsonify
finish_series(session)
return jsonify({"ok": True})
[docs]
@blueprint.route("/add_entry/series/clear", methods=["POST"])
@login_required
def clear_target_series():
"""Remove the active series state from the session."""
from flask import jsonify
clear_series(session)
return jsonify({"ok": True})