# SPDX-FileCopyrightText: 2023-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Compatibility-focused edit-entry route refactor.
This follows the add-entry route structure so both tab1/tab2 keep a shared
mental model while preserving edit-specific behavior (versioning + diffs).
"""
from __future__ import annotations
import logging
from flask import (
Blueprint,
current_app,
flash,
g,
redirect,
render_template,
request,
session,
url_for,
)
from flask_login import login_required
from wtforms import SelectMultipleField
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 (
form_owned_field_names as _form_owned_field_names,
)
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.helper_functions import compare_field_diffs, create_new_version
from labfrog.helpers.choices import ordered_unique_strings
from labfrog.helpers.demo import demo_limit_reached, demo_user_filter
from labfrog.helpers.target import _normalize_target_payload
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_shot_number_list_for_set,
get_total_number_of_shots,
int_current_number_of_sets,
update_indiv_count_MODIFY_SHOT,
)
blueprint = Blueprint("edit_entry", __name__)
[docs]
def identify_matching_documents(old_id):
"""Follow an edit chain backwards so the UI can compare historical versions."""
if not old_id:
return []
shots = get_db()["shots"]
matching_documents = []
first_match = shots.find_one(demo_user_filter({"_id_OLD": old_id}))
if isinstance(first_match, dict):
matching_documents.append(first_match)
current_id = old_id
while current_id is not None:
current_match = shots.find_one(demo_user_filter({"_id": current_id}))
if current_match is None:
break
matching_documents.append(current_match)
current_id = current_match.get("_id_OLD")
return matching_documents
def _prefill_form_and_diffs_from_current_doc(
form,
current_doc,
*,
parameters_db,
options_collection,
mode,
visible_fields,
custom_field_lookup,
multi_select_fields,
):
if current_doc is None:
return form, [], {}
matching_documents = identify_matching_documents(current_doc.get("_id_OLD"))
if len(matching_documents) > 1:
diff_fields, diff_multiselect_choices = compare_field_diffs(
matching_documents,
multi_value_fields=multi_select_fields,
)
else:
diff_fields, diff_multiselect_choices = [], {}
if request.method != "POST":
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,
)
return form, diff_fields, diff_multiselect_choices
def _validate_required_float_fields(form, custom_field_docs, visible_fields):
for field in custom_field_docs:
field_name = field.get("field_name")
if field_name not in visible_fields:
continue
validators = field.get("validators") or []
custom_field_type = (field.get("custom_field_type") or "").lower()
if "float" not in custom_field_type or "InputRequired" not in validators:
continue
try:
float(getattr(form, field_name).data)
except (TypeError, ValueError):
return {field_name: "Invalid value. Please enter a valid float value."}
return {}
[docs]
@blueprint.route("/tab2", methods=["GET", "POST"]) # legacy alias
@blueprint.route("/edit_shot", methods=["GET", "POST"]) # legacy alias
@blueprint.route("/edit_entry", methods=["GET", "POST"]) # canonical public path
@login_required
def edit_entry():
init_request_state()
ensure_runtime_indexes()
if request.path in {"/tab2", "/edit_shot"}:
logging.getLogger(__name__).warning(
"Route alias %s is legacy and will be removed in a future release; "
"use /edit_entry.",
request.path,
)
target_dict = get_cached_mediawiki_targets()
campaign_dict = get_cached_mediawiki_campaigns()
db = get_db()
parameters_db = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
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()
)
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", []),
])
custom_field_types, custom_fields_by_section = collect_visible_custom_fields(
form, visible_fields, custom_field_docs
)
multi_select_fields = {
name
for name, field_obj in getattr(form, "_fields", {}).items()
if isinstance(field_obj, SelectMultipleField)
}
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
form, diff_fields, diff_multiselect_choices = (
_prefill_form_and_diffs_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,
multi_select_fields=multi_select_fields,
)
)
shot_number_list = []
if mode == "set" and current_doc:
set_number_at_index = int(current_doc.get("set_number", 0) or 0)
if set_number_at_index:
shot_number_list = get_shot_number_list_for_set(set_number_at_index)
previous_source_values = {}
if mode == "shot" and isinstance(current_doc, dict):
shot_number_for_previous = int(current_doc.get("shot_number", 0) or 0)
if shot_number_for_previous > 0:
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
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
flash("Form cleared!", "success")
can_jump_to_first = has_entries and (clear_viewing or not at_oldest_entry)
can_go_previous = has_entries and (not clear_viewing) and (not at_oldest_entry)
can_go_next = has_entries and (not clear_viewing) and (not at_latest_entry)
can_jump_to_last = has_entries and (clear_viewing or not at_latest_entry)
custom_error = {}
if request.method == "POST":
if "date_time" in form.data:
try:
form.date_time.data = _coerce_datetime_field(
request.form.get("date_time")
)
except Exception as exc:
logging.error("Error updating date_time field: %s", exc)
flash(f"Error updating date_time field: {exc}", "error")
if "submit" in request.form and form.validate_on_submit():
user = current_username()
custom_error = _validate_required_float_fields(
form, custom_field_docs, visible_fields
)
if custom_error:
session.pop("_flashes", None)
field_name = next(iter(custom_error.keys()))
flash(
f"ValueError, please enter a valid float for field {field_name}",
"error",
)
else:
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
)
_submit_ok = True
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")
_submit_ok = False
if _submit_ok and current_doc is None:
session.pop("_flashes", None)
flash("No entry is selected for editing.", "error")
return redirect(url_for("edit_entry.edit_entry"))
if _submit_ok:
updated_fields = {"user": user, **data, "status": "active"}
if field_value_meta:
updated_fields[FIELD_VALUE_META_KEY] = field_value_meta
updated_fields[INCLUDE_DIAG_DETAILS_FIELD] = True
if _submit_ok and demo_limit_reached(shots, user):
flash_demo_limit_reached()
_submit_ok = False
if _submit_ok:
doc_id = str(current_doc["_id"])
new_doc = create_new_version(
doc_id,
updated_fields,
form_owned_fields=_form_owned_field_names(
form,
visible_fields=visible_fields,
always_include=always_include,
diagnostic_names=set(g.get("DIAG_NAMES", {}).keys()),
),
)
if not new_doc:
logging.error("Failed to update shot")
flash("Failed to update shot.", "error")
_submit_ok = False
if _submit_ok:
clear_navigation_caches()
if mode == "set":
_total, _before_set, shot_number_list = (
update_indiv_count_MODIFY_SHOT(
data.get("set_number"),
data.get("set_length"),
)
)
matching_documents = identify_matching_documents(
new_doc.get("_id_OLD")
)
if len(matching_documents) > 1:
diff_fields, diff_multiselect_choices = compare_field_diffs(
matching_documents,
multi_value_fields=multi_select_fields,
)
else:
diff_fields, diff_multiselect_choices = [], {}
flash("Set Edited." if mode == "set" else "Shot Edited.", "success")
else:
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"] if has_entries else 0
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,
)
if mode == "set" and current_doc:
set_number_at_index = int(current_doc.get("set_number", 0) or 0)
if set_number_at_index and not shot_number_list:
shot_number_list = get_shot_number_list_for_set(set_number_at_index)
shot_numbers = [int(n) for n in shot_number_list if n not in (None, "")]
if shot_numbers:
nav_help["start_shot"] = min(shot_numbers)
nav_help["end_shot"] = max(shot_numbers)
else:
set_length = int(current_doc.get("set_length", 0) or 0)
upcoming = int(nav_help.get("UPCOMING_SHOT_NUMBER", 1) or 1)
nav_help["start_shot"] = max(1, upcoming - set_length)
nav_help["end_shot"] = max(1, upcoming - 1)
kafka_data = current_doc.get("kafka") if isinstance(current_doc, dict) else None
return render_template(
"edit_entry.html",
custom_error=custom_error,
kafka_data=kafka_data,
mode=mode,
selected_layout=selected_layout,
diff_fields=diff_fields,
diff_multiselect_choices=diff_multiselect_choices,
nav_help=nav_help,
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,
can_jump_to_first=can_jump_to_first,
can_go_previous=can_go_previous,
can_go_next=can_go_next,
can_jump_to_last=can_jump_to_last,
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"],
shot_number_list=shot_number_list,
field_aliases=page_context["field_aliases"],
previous_source_values=previous_source_values,
empty_campaign_message=(
"No shots in this campaign yet." if not has_entries else None
),
)