# SPDX-FileCopyrightText: 2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Parameter options management."""
import datetime
import logging
from flask import (
Blueprint,
current_app,
flash,
g,
jsonify,
render_template,
request,
session,
url_for,
)
from flask_login import current_user, login_required
from wtforms import SelectField, SelectMultipleField
from labfrog.db import get_db
from labfrog.form_factory import create_input_form
from labfrog.forms import DynamicForm
from labfrog.helpers.aliases import invalidate_field_alias_cache
from labfrog.helpers.choices import normalize_choice_value
from labfrog.helpers.layout import available_campaign_names, paired_mode
from labfrog.helpers.urls import get_management_return_url
blueprint = Blueprint("parameter_options", __name__)
_MAX_PARAMETER_OPTION_ROWS = 500
def _current_username() -> str:
"""Return current username or a sensible fallback."""
try:
if current_user and not current_user.is_anonymous:
return getattr(current_user, "username", None) or getattr(
current_user, "id", "unknown"
)
except Exception: # pragma: no cover - defensive
# This name is stored as provenance on the option, so an unexpected
# failure here silently attributes someone's edit to "Anonymous".
logging.warning(
"Could not resolve the current username; "
"recording this change as Anonymous.",
exc_info=True,
)
return "Anonymous"
def _normalize_option_scope(field: str | None, campaign: str | None) -> str | None:
"""Campaign chooser options must be global, not campaign-scoped."""
if (field or "").strip() == "Campaign":
return None
return campaign
def _get_parameter_fields(mode: str) -> tuple[list[str], dict]:
"""Return (fields, field_sections_dict) for the default layout."""
collection_name = current_app.config.get("MONGODB_COLLECTION_FOR_SETTINGS")
db = get_db()
default_doc = db[str(collection_name)].find_one({
"layout_name": "DEFAULT",
"mode": mode,
})
sections = default_doc.get("field_sections_dict") if default_doc else {}
fields = []
for section_fields in sections.values():
fields.extend(section_fields)
# include custom + diagnostics from get_all_fields (we'll filter diagnostics later)
all_fields = DynamicForm.get_all_fields(collection=collection_name, mode=mode)
for f in all_fields:
if f not in fields:
fields.append(f)
return list(dict.fromkeys(fields)), (sections or {})
def _get_selectable_fields(mode: str) -> list[str]:
"""Return fields available for parameter option and alias management."""
param_fields, _sections = _get_parameter_fields(mode)
# include custom select fields even if not in layout
custom_selects = get_db()["custom_fields_app"].find(
{
"custom_field_type": {
"$in": [
"quality_select",
"select_custom",
"select_multi_custom",
"SelectField", # legacy values
"SelectMultipleField",
]
}
},
{"field_name": 1},
)
extra_fields = []
for doc in custom_selects:
fname = doc.get("field_name")
if fname:
extra_fields.append(fname)
# Build a form instance so the fields exist with choices
form = create_input_form(
collection=current_app.config.get("MONGODB_COLLECTION_FOR_SETTINGS"),
mode=mode,
visible_fields=param_fields,
)
selectable = []
DIAG_NAMES = g.get("DIAG_NAMES", {})
for field in list(param_fields) + extra_fields:
if field in DIAG_NAMES:
continue # diagnostics managed elsewhere
field_obj = getattr(form, field, None)
if isinstance(field_obj, (SelectField, SelectMultipleField)):
selectable.append(field)
# Fallback: include well-known managed fields.
# `target` must remain alias-editable even when runtime wiring makes it non-select.
for fallback in ("Campaign", "target"):
field_obj = getattr(form, fallback, None)
if fallback == "target" and field_obj is not None:
selectable.append(fallback)
continue
if isinstance(field_obj, (SelectField, SelectMultipleField)):
selectable.append(fallback)
return sorted(set(selectable))
def _get_field_alias(field: str, campaign: str | None) -> str:
normalized_campaign = normalize_choice_value(campaign) or None
alias_collection = get_db()["field_aliases"]
if normalized_campaign:
campaign_values = [
normalized_campaign,
f'"{normalized_campaign}"',
f"'{normalized_campaign}'",
]
doc = alias_collection.find_one(
{"field_name": field, "campaign": {"$in": campaign_values}},
{"alias": 1, "_id": 0},
)
if doc:
return (doc.get("alias") or "").strip()
fallback = alias_collection.find_one(
{
"field_name": field,
"$or": [
{"campaign": None},
{"campaign": ""},
{"campaign": {"$exists": False}},
],
},
{"alias": 1, "_id": 0},
)
return (fallback.get("alias") or "").strip() if fallback else ""
def _normalize_option_records(option_rows) -> list[dict]:
"""Input: raw option rows. Output: normalized rows safe for API + form rendering."""
normalized = []
seen_values = set()
rows = option_rows if isinstance(option_rows, list) else []
for raw_index, row in enumerate(rows):
if isinstance(row, dict):
value = normalize_choice_value(row.get("value"))
label = str(row.get("label") or "").strip()
active = bool(row.get("active", True))
raw_order = row.get("order", raw_index)
elif isinstance(row, (list, tuple)) and len(row) >= 2:
value = normalize_choice_value(row[0])
label = str(row[1] or "").strip()
active = True
raw_order = raw_index
elif isinstance(row, str):
value = normalize_choice_value(row)
label = value or ""
active = True
raw_order = raw_index
else:
continue
if not value and not label:
continue
value = value or normalize_choice_value(label)
if not value:
continue
label = label or value
dedupe_key = value.casefold()
if dedupe_key in seen_values:
continue
seen_values.add(dedupe_key)
try:
order = int(raw_order)
except (TypeError, ValueError):
order = raw_index
normalized.append({
"value": value,
"label": label,
"active": active,
"order": order,
})
if len(normalized) >= _MAX_PARAMETER_OPTION_ROWS:
break
return sorted(normalized, key=lambda option: option["order"])
def _get_base_options(field: str, mode: str) -> list[dict]:
"""Return base options for a field from the current form configuration."""
collection_name = current_app.config.get("MONGODB_COLLECTION_FOR_SETTINGS")
param_fields, _sections = _get_parameter_fields(mode)
if field not in param_fields:
param_fields.append(field)
form = create_input_form(
collection=collection_name,
mode=mode,
visible_fields=param_fields,
)
field_obj = getattr(form, field, None)
if not field_obj or not hasattr(field_obj, "choices"):
return []
choices = []
for idx, choice in enumerate(field_obj.choices or []):
if isinstance(choice, (list, tuple)) and len(choice) >= 2:
value, label = choice[0], choice[1]
else:
value = choice
label = choice
value = (value or "").strip()
label = (label or "").strip()
if not value and not label:
continue
choices.append({
"value": value or label,
"label": label or value,
"active": True,
"order": idx,
})
return _normalize_option_records(choices)
def _merge_base_and_custom(base: list[dict], custom: list[dict]) -> list[dict]:
"""Merge custom overrides onto base choices without removing base entries."""
normalized_base = _normalize_option_records(base)
normalized_custom = _normalize_option_records(custom)
base_map = {}
base_order = []
for idx, opt in enumerate(normalized_base):
value = (opt.get("value") or "").strip()
if not value or value in base_map:
continue
base_map[value] = {
"value": value,
"label": (opt.get("label") or value).strip(),
"active": True,
"order": idx,
}
base_order.append(value)
overrides = {}
extras = []
for opt in normalized_custom:
value = (opt.get("value") or "").strip()
if not value:
continue
if value in base_map:
overrides[value] = opt
else:
extras.append({
"value": value,
"label": (opt.get("label") or value).strip(),
"active": bool(opt.get("active", True)),
"order": opt.get("order", 0),
})
merged = []
for value in base_order:
base_opt = base_map[value]
override = overrides.get(value)
if override:
base_opt = {
"value": value,
"label": (override.get("label") or base_opt["label"]).strip(),
"active": bool(override.get("active", True)),
"order": base_opt["order"],
}
merged.append(base_opt)
merged.extend(sorted(extras, key=lambda x: x.get("order", 0)))
return merged
[docs]
@blueprint.route("/parameter_options", methods=["GET"]) # canonical alias
@blueprint.route("/set_parameter_options", methods=["GET"]) # current public path
@login_required
def parameter_options():
"""Render the parameter options manager."""
mode = session.get("mode", "shot")
selected_layout = (
session.get("selected_layout")
or session.get(f"selected_source_layout_{mode}")
or session.get(f"selected_source_layout_{paired_mode(mode)}")
or "DEFAULT"
)
selectable_fields = _get_selectable_fields(mode)
selected_campaign = session.get("selected_campaign")
campaigns = available_campaign_names([selected_campaign])
initial_field = (request.args.get("field") or "").strip()
return_url = get_management_return_url(
request.values.get("next"),
url_for("field_selection.field_selection"),
)
if initial_field and initial_field not in selectable_fields:
selectable_fields.append(initial_field)
return render_template(
"parameter_options.html",
selected_layout=selected_layout,
active_tab="parameter_options",
selectable_fields=selectable_fields,
campaigns=campaigns,
selected_campaign=selected_campaign,
initial_field=initial_field,
return_url=return_url,
)
[docs]
@blueprint.route("/parameter_options/list", methods=["GET"]) # canonical alias
@blueprint.route("/set_parameter_options/list", methods=["GET"]) # current public path
@login_required
def list_parameter_options():
"""Return options for a field (shared across layouts/modes)."""
field = request.args.get("field")
requested_campaign = normalize_choice_value(request.args.get("campaign")) or None
if not field:
return jsonify({"error": "field is required"}), 400
options_campaign = _normalize_option_scope(field, requested_campaign)
alias = _get_field_alias(field, requested_campaign)
mode = session.get("mode", "shot")
base_options = _get_base_options(field, mode)
options = list(
get_db()["campaign_layouts"].find({
"field_name": field,
"$or": [
{"campaign": options_campaign},
{"campaign": None},
{"campaign": {"$exists": False}},
],
})
)
if options:
doc = next((d for d in options if d.get("campaign") == options_campaign), None)
if doc is None:
doc = next(
(d for d in options if d.get("campaign") in (None, "")), options[0]
)
custom_options = _normalize_option_records(doc.get("options"))
merged = _merge_base_and_custom(base_options, custom_options)
return jsonify({"options": merged, "alias": alias})
# No options configured yet; fall back to options_text from custom field definition
custom = get_db()["custom_fields_app"].find_one(
{"field_name": field}, {"options_text": 1}
)
if custom:
raw = (custom.get("options_text") or "").strip()
if raw:
split_opts = [opt.strip() for opt in raw.split(",") if opt.strip()]
fallback = [
{"value": opt, "label": opt, "active": True, "order": idx}
for idx, opt in enumerate(split_opts)
]
if fallback:
return jsonify({"options": fallback, "alias": alias})
# No defaults found
if base_options:
return jsonify({"options": base_options, "alias": alias})
return jsonify({"options": [], "alias": alias})
[docs]
@blueprint.route("/parameter_options/save", methods=["POST"]) # canonical alias
@blueprint.route("/set_parameter_options/save", methods=["POST"]) # current public path
@login_required
def save_parameter_options():
"""Save option set for a field (shared across layouts/modes)."""
payload = request.get_json(silent=True) or {}
field = payload.get("field_name")
options = payload.get("options", [])
alias = (payload.get("alias") or "").strip()
requested_campaign = normalize_choice_value(payload.get("campaign")) or None
if not field:
return jsonify({"error": "field_name is required"}), 400
options_campaign = _normalize_option_scope(field, requested_campaign)
alias_campaign = requested_campaign
if not isinstance(options, list):
return jsonify({"error": "options must be a list"}), 400
mode = session.get("mode", "shot")
_get_base_options(field, mode) # still computed for compatibility/validation
# Clean/validate options
cleaned = []
seen_values = set()
for idx, opt in enumerate(options):
if not isinstance(opt, dict):
continue
value = (opt.get("value") or "").strip()
label = (opt.get("label") or "").strip()
if not value and not label:
continue # skip empty rows
value = normalize_choice_value(value) or normalize_choice_value(label)
label = label or value
if not value:
continue
dedupe_key = value.casefold()
if dedupe_key in seen_values:
return jsonify({"error": f"duplicate value '{value}'"}), 400
seen_values.add(dedupe_key)
cleaned.append({
"value": value,
"label": label,
"active": bool(opt.get("active", True)), # keep for compatibility
"order": opt.get("order", idx),
})
if len(cleaned) >= _MAX_PARAMETER_OPTION_ROWS:
break
doc = {
"field_name": field,
"options": cleaned,
"campaign": options_campaign,
"updated_by": _current_username(),
"updated_at": datetime.datetime.now(datetime.timezone.utc),
}
coll = get_db()["campaign_layouts"]
try:
coll.update_one(
{"field_name": field, "campaign": options_campaign},
{"$set": doc},
upsert=True,
)
alias_collection = get_db()["field_aliases"]
if alias:
alias_collection.update_one(
{"field_name": field, "campaign": alias_campaign},
{
"$set": {
"field_name": field,
"alias": alias,
"campaign": alias_campaign,
"updated_by": _current_username(),
"updated_at": datetime.datetime.now(datetime.timezone.utc),
}
},
upsert=True,
)
if field == "Campaign" and alias_campaign:
# Campaign option rows are global, so older saves could put the
# Campaign alias in the global bucket. Remove that duplicate
# when the same alias is now saved to an explicit campaign.
alias_collection.delete_many({
"field_name": field,
"alias": alias,
"$or": [
{"campaign": None},
{"campaign": ""},
{"campaign": {"$exists": False}},
],
})
else:
alias_collection.delete_one({
"field_name": field,
"campaign": alias_campaign,
})
invalidate_field_alias_cache()
except Exception as exc: # pragma: no cover
logging.error("Failed to save parameter options: %s", exc)
return jsonify({"error": "save_failed"}), 500
flash(f"Saved options for {field}", "success")
return jsonify({"ok": True})