# SPDX-FileCopyrightText: 2026 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Shared helpers for campaign layout setup workflows."""
import datetime
from flask import flash, redirect, session, url_for
from labfrog.db import get_db
from labfrog.helpers.layout import (
allow_default_for_campaign,
build_field_layout_doc,
field_layout_filter,
find_mode_layout_doc,
paired_mode,
resolve_layout_selections,
)
def _persist_default_layout_mapping(
parameters_collection, mode: str, campaign: str
) -> bool:
"""Attach a campaign to the mode-specific DEFAULT layout so it survives restart."""
if not campaign:
return False
default_doc = parameters_collection.find_one({
"layout_name": "DEFAULT",
"mode": mode,
})
if not default_doc:
default_doc = parameters_collection.find_one({"layout_name": "DEFAULT"})
if not default_doc:
return False
campaigns = [c for c in (default_doc.get("campaigns") or []) if c]
if campaign in campaigns:
return True
campaigns.append(campaign)
parameters_collection.update_one(
{"_id": default_doc["_id"]},
{
"$set": {
"campaigns": campaigns,
"updated_at": datetime.datetime.now(datetime.timezone.utc),
}
},
)
return True
def _persist_default_layout_mapping_pair(
parameters_collection, mode: str, campaign: str
):
"""Attach a campaign to the DEFAULT layout in both shot/set modes."""
persisted = False
for target_mode in {mode, paired_mode(mode)}:
allow_default_for_campaign(target_mode, campaign)
persisted = (
_persist_default_layout_mapping(
parameters_collection, target_mode, campaign
)
or persisted
)
return persisted
def _ensure_counterpart_campaign_mapping(
parameters_collection,
*,
mode: str,
layout_doc: dict,
campaign: str,
) -> None:
"""Ensure the sibling shot/set layout exists and maps the campaign too."""
layout_name = (layout_doc or {}).get("layout_name")
if not layout_name or not campaign:
return
counterpart = paired_mode(mode)
sibling_doc = find_mode_layout_doc(parameters_collection, layout_name, counterpart)
if sibling_doc and sibling_doc.get("_id"):
campaigns = [c for c in (sibling_doc.get("campaigns") or []) if c]
if campaign in campaigns:
return
campaigns.append(campaign)
parameters_collection.update_one(
{"_id": sibling_doc["_id"]},
{
"$set": {
"campaigns": campaigns,
"updated_at": datetime.datetime.now(datetime.timezone.utc),
}
},
)
return
selected_parameters, selected_diagnostics, _ = resolve_layout_selections(layout_doc)
counterpart_doc = build_field_layout_doc(
parameters_collection,
layout_name=layout_name,
mode=counterpart,
description=layout_doc.get("description"),
responsible_person=layout_doc.get("responsible_person"),
campaigns=[*(layout_doc.get("campaigns") or []), campaign],
selected_parameters=selected_parameters,
selected_diagnostics=selected_diagnostics,
)
parameters_collection.update_one(
{"layout_name": layout_name, "mode": counterpart, **field_layout_filter()},
{"$set": counterpart_doc},
upsert=True,
)
[docs]
def mapped_layout_sources(parameters_collection, mode: str) -> list[dict]:
"""Return campaign-mapped layouts with light metadata for setup helpers."""
sources = []
cursor = parameters_collection.find({
"$and": [
{"layout_name": {"$ne": "DEFAULT"}},
{"$or": [{"mode": mode}, {"mode": {"$exists": False}}]},
field_layout_filter(),
]
}).sort("layout_name", 1)
for doc in cursor:
campaigns = [c for c in (doc.get("campaigns") or []) if c]
if not campaigns:
continue
selected_parameters, selected_diagnostics, selected_fields = (
resolve_layout_selections(doc)
)
for campaign in campaigns:
sources.append({
"campaign": campaign,
"layout_name": doc.get("layout_name"),
"description": doc.get("description", ""),
"responsible_person": doc.get("responsible_person", ""),
"selected_count": len(selected_fields),
"diagnostic_count": len(selected_diagnostics),
"header_name": doc.get("layout_name", ""),
"option_key": f"{campaign}::{doc.get('layout_name')}",
})
return sources
[docs]
def build_campaign_layout_setup_context(
parameters_collection, mode: str, campaign: str, next_url: str
) -> dict:
"""Build the template context for the missing-campaign-layout chooser."""
return {
"campaign": campaign,
"mode": mode,
"next_url": next_url,
"source_layouts": mapped_layout_sources(parameters_collection, mode),
"suggested_layout_name": f"{campaign} Fields" if campaign else "New Fields",
}
[docs]
def handle_campaign_layout_setup_submission(
parameters_collection,
mode: str,
campaign: str,
next_url: str,
form_data,
):
"""Handle setup chooser submissions and return the next response."""
setup_method = (form_data.get("setup_method") or "").strip()
source_campaign = (form_data.get("source_campaign") or "").strip()
source_layout_name = (form_data.get("source_layout") or "").strip()
new_layout_name = (form_data.get("new_layout_name") or "").strip()
if setup_method == "use_default":
if _persist_default_layout_mapping_pair(parameters_collection, mode, campaign):
flash(
f"Using the DEFAULT layout for campaign {campaign}.",
"success",
)
else:
flash(
f"Using the DEFAULT layout for campaign {campaign} in this session.",
"success",
)
return redirect(next_url)
if setup_method == "use_existing":
if not source_layout_name:
flash("Please choose a source campaign layout.", "error")
return None
source_doc = parameters_collection.find_one({
"layout_name": source_layout_name,
"$or": [{"mode": mode}, {"mode": {"$exists": False}}],
})
if not source_doc:
flash("Selected source layout was not found.", "error")
return None
campaigns = [c for c in (source_doc.get("campaigns") or []) if c]
if campaign not in campaigns:
campaigns.append(campaign)
parameters_collection.update_one(
{"_id": source_doc["_id"]},
{
"$set": {
"campaigns": campaigns,
"updated_at": datetime.datetime.now(datetime.timezone.utc),
}
},
)
_ensure_counterpart_campaign_mapping(
parameters_collection,
mode=mode,
layout_doc={**source_doc, "campaigns": campaigns},
campaign=campaign,
)
if source_campaign:
session[f"selected_headers_layout_{mode}"] = source_doc.get(
"layout_name", "DEFAULT"
)
session["selected_layout"] = source_doc.get("layout_name", "DEFAULT")
flash(
f"Assigned campaign {campaign} to layout {source_doc.get('layout_name')}.",
"success",
)
return redirect(next_url)
if setup_method == "clone_adjust":
if not source_layout_name:
flash("Please choose a source campaign layout to clone.", "error")
return None
if not new_layout_name:
new_layout_name = f"{campaign} Fields"
if parameters_collection.find_one({
"layout_name": new_layout_name,
"mode": mode,
}):
flash(
f"Layout {new_layout_name} already exists. Choose a different name.",
"error",
)
return None
return redirect(
url_for(
"field_selection.field_selection",
draft_mode="clone",
draft_layout_name=new_layout_name,
draft_campaign=campaign,
draft_source_layout=source_layout_name,
)
)
if setup_method == "start_fresh":
if not new_layout_name:
new_layout_name = f"{campaign} Fields"
return redirect(
url_for(
"field_selection.field_selection",
draft_mode="fresh",
draft_layout_name=new_layout_name,
draft_campaign=campaign,
)
)
flash("Please choose a setup method.", "error")
return None