nappMATLAB-toolbox/openBIS/nappSyncToOpenbis.py

"""
nappSyncToOpenbis.py

Sync napp* beamtime metadata rows into openBIS objects: update an existing
object if one already exists for a given file name, otherwise create a new
one. One object per table row. Fields already filled in openBIS (e.g. by
hand during the beamtime) are never overwritten by automated values.

Called from MATLAB via:
    py.importlib.import_module('nappSyncToOpenbis')
    resultJson = py.nappSyncToOpenbis.sync_rows(rowsJson, url, space, ...)

============================================================================
HOW THIS FILE IS ORGANIZED
============================================================================
Two sections:

  CONFIGURATION -- everything you're likely to want to change: which
  FileList column goes to which openBIS property, fixed values, how text
  fields get combined, etc. All plain data (dicts/lists/strings) -- no
  logic. This is what you edit for:
    - a new column -> property mapping
    - changing which column(s) feed a text field (e.g. Notes, Sample Notes)
    - a new fixed/constant value
    - adapting this file for a DIFFERENT openBIS object type entirely:
      replace this whole section, leave the engine below alone.

  ENGINE -- generic code that reads the configuration above and does the
  actual work (connecting, matching existing objects, uploading files,
  embedding images, etc.). You shouldn't need to touch this for ordinary
  changes -- only if you need a genuinely new KIND of derivation that
  doesn't fit the existing patterns (direct copy, numeric conversion,
  timestamp, text-join, category-dependent source).

Several things below are workarounds for confirmed pybis (v1.37.5) quirks,
found by reading pybis's own source directly rather than guessing -- see
the comments at each one for what was actually wrong and how it was
confirmed. These live in the engine, not the config, since they're not
something a schema change would ever need to touch.
"""

import json
import re
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from pandas import DataFrame
from pybis import Openbis

LOCAL_TZ = ZoneInfo("Europe/Zurich")   # <-- change if a future beamtime runs elsewhere


# ============================================================================
# CONFIGURATION -- edit this section for ordinary changes. See file docstring.
# ============================================================================

OBJECT_TYPE = "LAC_EXPERIMENTAL_STEP_BEAMTIME"
MEASUREMENT_TYPE_VOCABULARY = "LAC_EXPERIMENTALSTEPTYPE_BEALINE"
RAW_DATASET_TYPE = "RAW_DATA"
RESULTS_PROPERTY = "default_experiment.experimental_results"
FILENUMBER_PROPERTY = "filenumber"
CATEGORY_PROPERTY = "lac_steptype"
KE_RANGE_PROPERTY = "ke_range_center"
KE_RANGE_SOURCE_COLUMN = "scientaEkinRange_eV"

# --- File category, derived from filename extension -- drives lac_steptype
# and which photon-energy source column is used below. Add new extensions
# / categories here if a future beamtime uses different file types.
CATEGORY_BY_EXTENSION = {
    "ibw": "XPS",
    "h5":  "NEXAFS",
    "bmp": "PHOTO", "jpg": "PHOTO", "jpeg": "PHOTO", "png": "PHOTO",
    "wmv": "PHOTO", "mov": "PHOTO",
}

# --- Direct column -> property mappings ------------------------------------
# Simple 1:1 copies, no conversion needed (text properties, or numeric
# properties where the JSON value's native type already matches -- e.g.
# ke_step is REAL and a plain float is exactly right as-is).
# Property codes are lowercase -- confirmed openBIS wants that at the wire
# level regardless of what the admin UI/xlsx export displays.
DIRECT_MAP = {
    "sample":                 "sample_name",
    "regionName":             "region",
    "scientaLensMode":        "lens_mode",
    "scientaAcquisitionMode": "acq_mode",
    "scientaEkinStep_eV":     "ke_step",
}

# --- Numeric columns needing a type conversion -----------------------------
# value = (openBIS property code, "int" or "float").
# "int"   -> rounded to a genuine Python int (for INTEGER-typed properties;
#            MATLAB/JSON numbers arrive as floats, e.g. 57.0, even for
#            whole-number fields).
# "float" -> passed through as-is (REAL-typed properties -- JSON already
#            gives a plain float, nothing to convert).
# Note: sequence_iterations_real / region_iterations_real are misleadingly
# named -- their CODEs say "_real" but their actual openBIS data type is
# INTEGER, confirmed from the schema export. Named per the actual type
# needed here, not per the (inconsistent) code name.
NUMERIC_MAP = {
    "sampleTemp_dC":             ("temp_reading",            "float"),
    "iceTemp_dC":                ("temp_calibrated",          "float"),
    "cellPressure_mbar":         ("cell_pressure_real",       "float"),
    "scientaDwellTime_ms":       ("dwell_time_integer",       "int"),
    "scientaPassEnergy_eV":      ("passenergy_integer",       "int"),
    "scientaSequenceIterations": ("sequence_iterations_real", "int"),
    "scientaRegionIterations":   ("region_iterations_real",   "int"),
    "smplX_mm":                  ("position_x_real",          "float"),
    "smplY_mm":                  ("position_y_real",          "float"),
    "smplZ_mm":                  ("position_z_real",          "float"),
}

# --- Timestamp columns ------------------------------------------------------
# value = (openBIS property code, MATLAB source format).
# "datevec" = a 6-element [Y M D H MI S] list (e.g. creationDate)
# "datenum" = a MATLAB serial datenum (e.g. lastModifiedDatenum)
TIMESTAMP_MAP = {
    "creationDate":        ("start_date", "datevec"),
    "lastModifiedDatenum": ("end_date",   "datenum"),
}

# --- Text fields built by joining one or more columns with newlines --------
# key = target property code, value = list of source columns, in order.
# A column that's empty/missing for a given row is simply skipped; if NONE
# of a field's source columns have anything, the property is left unset
# entirely (never written as a blank string).
#   This is exactly where to make a change like "comment should go to
# Notes, not Sample Notes" -- edit the list(s) below, nothing else.
TEXT_JOIN_MAP = {
    "notes": ["comment"],
}

# --- Photon energy: which column to use depends on the file category -------
# XPS uses a single fixed excitation energy; NEXAFS genuinely scans a
# range. "scalar" reads one number (unwrapping a 1-element list if needed);
# "range" expects a 2-element [min, max] list.
PHOTON_ENERGY_PROPERTY = "photon_energy"
PHOTON_ENERGY_SOURCE_BY_CATEGORY = {
    "XPS":    ("xRayEkin_eV",      "scalar"),
    "NEXAFS": ("xRayEkinRange_eV", "range"),
}

# --- Fixed/constant values, written once per sync (not derived from any
# row's own data) ------------------------------------------------------
# Note: these are VOCABULARY TERM values, not property codes -- a
# different namespace, genuinely uppercase per the LAC_ENDSTATIONS
# vocabulary definition (confirmed from the schema export).
CONSTANT_MAP = {
    "endstation_name": "SOLID-GAS",   # <-- change if a different campaign/instrument uses this script
}

# --- Campaign-level values, from FileList.Properties.CustomProperties,
# same value applied to every row this run ------------------------------
CAMPAIGN_MAP = {
    "beamline":        "campaignLocation",
    "team":            "team",
    "researchproject": "projectName",
}

# Deliberately not represented anywhere in this openBIS object type
# (confirmed OK to leave unmapped): contact, contactORCID, funding,
# publicationDoi, publicationDataDoi, rawDataDoi, dataLevel, sourceDirectory,
# campaignName.


# ============================================================================
# ENGINE -- generic logic reading the configuration above. See file docstring.
# ============================================================================

def _connect(url, token=None, username=None, password=None, verify_certificates=True):
    o = Openbis(url, verify_certificates=verify_certificates)
    if token:
        # NOT o.set_token(token) -- that calls is_token_valid(), which hits
        # openBIS's isSessionActive RPC. Confirmed (by reading pybis
        # 1.37.5's source) that set_token()/the token property setter/
        # is_session_active() all route through that one same check, and
        # confirmed separately that PAT authentication is disabled at the
        # server level on this instance (community.openbis.ch/t/134) --
        # tokens are creatable/listable but rejected on real use. Writing
        # directly to __dict__ bypasses the property setter (and that
        # check) without affecting anything else -- every other pybis
        # method reads self.token via the plain getter, no validation on
        # read. Kept for whenever PAT auth gets enabled server-side.
        o.__dict__["token"] = token
    elif username and password:
        o.login(username, password)
    else:
        raise ValueError("Provide either a token, or both username and password.")
    return o


def _clean_str_or_none(value):
    """Coerce a value that may have crossed the MATLAB<->Python boundary
    (via pyargs, not JSON) into a clean native str, or None if it
    represents 'nothing'. Deliberately avoids `if value:` -- MATLAB's []
    can arrive as a NumPy array, and NumPy raises its own error ("ambiguous
    truth value") for exactly that kind of implicit boolean check, rather
    than just being falsy the way None/'' are."""
    if value is None:
        return None
    if hasattr(value, "size"):   # NumPy array or similar
        return None if value.size == 0 else str(value)
    text = str(value).strip()
    return text if text else None


def _to_clean_int(value):
    """Round a MATLAB-crossed numeric value (arrives as a JSON float, e.g.
    57.0) to a genuine Python int for an INTEGER-typed property. Returns
    None if it can't be cleanly interpreted as a number."""
    if value is None:
        return None
    try:
        return int(round(float(value)))
    except (ValueError, TypeError):
        return None


def _matlab_datenum_to_str(datenum):
    """MATLAB datenum (days since 0-Jan-0000) -> 'YYYY-MM-DD HH:MM:SS +ZZZZ',
    matching the format openBIS itself displays for these TIMESTAMP
    properties (confirmed from a real exported entry). The -366 offset is
    the standard MATLAB<->Python datenum conversion constant (Python's
    ordinal day 1 = 1-Jan-0001). Localized to LOCAL_TZ so the UTC offset is
    correct across the CET/CEST boundary."""
    if datenum is None:
        return None
    try:
        days = int(datenum)
        frac = datenum - days
        dt = datetime.fromordinal(days - 366) + timedelta(days=frac)
        return dt.replace(tzinfo=LOCAL_TZ).strftime("%Y-%m-%d %H:%M:%S %z")
    except (ValueError, TypeError, OverflowError):
        return None


def _matlab_datevec_to_str(datevec):
    """[Y, M, D, H, MI, S] -> 'YYYY-MM-DD HH:MM:SS +ZZZZ' (see above)."""
    if not isinstance(datevec, list) or len(datevec) < 6:
        return None
    try:
        y, mo, d, h, mi, s = datevec[:6]
        dt = datetime(int(y), int(mo), int(d), int(h), int(mi), int(s))
        return dt.replace(tzinfo=LOCAL_TZ).strftime("%Y-%m-%d %H:%M:%S %z")
    except (ValueError, TypeError):
        return None


def _file_number(name):
    """Doubled-digit convention: 'PHOENIX0183183_1.ibw' -> 183 (the leading
    digit run '0183183' splits into '0183' + '183', the same number with
    and without a leading zero). For names that don't match that pattern
    but have a later, short digit run (e.g. 'PHOENIX04042016_001.h5', where
    the leading run is a date, not a real index), that trailing run is used
    instead. Returns None if no clean number can be extracted."""
    if not name:
        return None
    digit_runs = re.findall(r"\d+", name)
    if not digit_runs:
        return None

    first = digit_runs[0]
    n = len(first)
    for split_at in range(-(-n // 2), n):   # ceil(n/2) .. n-1
        a, b = first[:split_at], first[split_at:]
        if a and b and int(a) == int(b):
            try:
                return int(a)
            except ValueError:
                return None

    if len(digit_runs) > 1 and len(digit_runs[-1]) <= 4:
        try:
            return int(digit_runs[-1])
        except ValueError:
            pass

    try:
        return int(first)
    except ValueError:
        return None   # e.g. a date-based name too long/ambiguous to use as-is


def _category_for_name(name):
    if not name or "." not in name:
        return None
    ext = name.lower().rsplit(".", 1)[-1]
    return CATEGORY_BY_EXTENSION.get(ext)


def _row_derived_props(row, category):
    """Everything that isn't a straight column copy: timestamps, file
    number, category, photon/KE energy (which may depend on category),
    and the text-join fields. Driven entirely by the CONFIGURATION section
    above -- add a new derivation *kind* here only if none of the existing
    patterns (direct/numeric/timestamp/text-join/category-dependent) fit."""
    props = {}

    for col, (prop, fmt) in TIMESTAMP_MAP.items():
        val = row.get(col)
        formatted = _matlab_datevec_to_str(val) if fmt == "datevec" else _matlab_datenum_to_str(val)
        if formatted:
            props[prop] = formatted

    fnum = _file_number(row.get("name"))
    if fnum is not None:
        props[FILENUMBER_PROPERTY] = fnum

    if category is not None:
        props[CATEGORY_PROPERTY] = category

    if category in PHOTON_ENERGY_SOURCE_BY_CATEGORY:
        col, kind = PHOTON_ENERGY_SOURCE_BY_CATEGORY[category]
        val = row.get(col)
        if kind == "scalar":
            if isinstance(val, list):
                val = val[0] if len(val) > 0 else None
            if isinstance(val, (int, float)):
                props[PHOTON_ENERGY_PROPERTY] = f"{val:.4g} eV"
        elif kind == "range":
            if isinstance(val, list) and len(val) == 2 and all(v is not None for v in val):
                props[PHOTON_ENERGY_PROPERTY] = f"{val[0]:.4g} - {val[1]:.4g} eV"

    ke_range = row.get(KE_RANGE_SOURCE_COLUMN)
    if isinstance(ke_range, list) and len(ke_range) == 2 and all(v is not None for v in ke_range):
        props[KE_RANGE_PROPERTY] = f"{ke_range[0]:.4g} - {ke_range[1]:.4g} eV"

    for prop, cols in TEXT_JOIN_MAP.items():
        parts = [str(row[c]) for c in cols if row.get(c)]
        if parts:
            props[prop] = "\n".join(parts)

    return props


def attach_result_image(o, obj, image_path, url, verify_certificates=True):
    """Uploads image_path to openBIS's ELN file service and appends it to
    RESULTS_PROPERTY, alongside any existing content there rather than
    replacing it.
    Confirmed working pattern from openBIS support: community.openbis.ch/t/391
    """
    import requests

    upload_url = f"{url}/openbis/openbis/file-service/eln-lims?type=Files&sessionID={o.token}"
    with open(image_path, "rb") as f:
        r = requests.post(upload_url, files={"file": f}, verify=verify_certificates)
    r.raise_for_status()
    file_url = json.loads(r.text)["url"]

    existing = _current_prop_safe(obj, RESULTS_PROPERTY)
    figure_html = f'<figure class="image"><img src="{file_url}" /></figure>'
    obj.props[RESULTS_PROPERTY] = (existing or "") + figure_html


def _current_prop_safe(obj, prop_code):
    try:
        return obj.props[prop_code]
    except Exception:
        return None


def add_photo_vocabulary_term(url, token=None, username=None, password=None,
                               verify_certificates=True):
    """One-time (or idempotent) setup: adds a PHOTO term to
    MEASUREMENT_TYPE_VOCABULARY, so photo rows can set CATEGORY_PROPERTY.
    Safe to call more than once -- checks for the term first.
    Run this deliberately, not as part of routine syncing.
    """
    o = _connect(url, token=token, username=username, password=password,
                 verify_certificates=verify_certificates)
    try:
        existing_terms = o.get_terms(vocabulary=MEASUREMENT_TYPE_VOCABULARY)
        # Read the underlying DataFrame directly rather than iterating
        # existing_terms -- iterating a Things collection internally calls
        # a single-item fetch (get_term) that's missing a required arg in
        # this pybis version (confirmed via a real traceback: TypeError:
        # get_term() missing 'vocabularyCode'), the same class of bug as
        # get_parents() below. .df sidesteps that entirely.
        if existing_terms is not None and len(existing_terms) > 0:
            df = existing_terms.df
            code_col = "code" if "code" in df.columns else df.columns[0]
            existing_codes = list(df[code_col])
        else:
            existing_codes = []
        if "PHOTO" in existing_codes:
            return json.dumps({"action": "already_exists"})

        term = o.new_term(
            code="PHOTO",
            vocabularyCode=MEASUREMENT_TYPE_VOCABULARY,
            label="Photo",
            description="Camera photo of the sample or setup",
        )
        term.save()
        return json.dumps({"action": "created"})
    finally:
        o.logout()


def get_property_assignments(url, object_type=None, token=None, username=None,
                              password=None, verify_certificates=True):
    """Read-only: dumps the complete current property structure of an
    object type -- code, data type, section, order, mandatory -- exactly
    as openBIS has it right now. Use this as ground truth before designing
    a new/modified object type, rather than working from memory of an old
    export. Makes no changes."""
    o = _connect(url, token=token, username=username, password=password,
                 verify_certificates=verify_certificates)
    object_type = object_type or OBJECT_TYPE
    try:
        sample_type = o.get_object_type(object_type)
        pas = sample_type.get_property_assignments()
        df = pas.df if hasattr(pas, "df") else DataFrame(pas)
        cols = [c for c in ["code", "dataType", "section", "ordinal", "mandatory"] if c in df.columns]
        df = df[cols].sort_values("ordinal") if "ordinal" in df.columns else df[cols]
        return df.to_json(orient="records")
    finally:
        o.logout()


def property_usage_report(url, object_type=None, token=None, username=None,
                           password=None, verify_certificates=True):
    """Read-only report: for every property on the given object type,
    counts how many EXISTING entries have a non-empty value for it. Run
    this BEFORE removing or changing a property in openBIS's schema, so
    you know how much real data would actually be affected rather than
    guessing. Makes no changes -- purely a search + count.

    The property list is read live via get_property_assignments() rather
    than hardcoded, so this works unmodified for any object type -- part
    of what makes this file reusable as a template.

    Returns a JSON string: {"total_entries": N, "property_fill_counts":
    {"temp_reading": 12, "cell_pressure_real": 12, ...}} -- a property
    absent from the dict (or at 0) currently has no data anywhere, so
    removing it is safe; anything with a high count is exactly what you'd
    lose."""
    o = _connect(url, token=token, username=username, password=password,
                 verify_certificates=verify_certificates)
    object_type = object_type or OBJECT_TYPE
    try:
        sample_type = o.get_object_type(object_type)
        pas = sample_type.get_property_assignments()
        pa_df = pas.df if hasattr(pas, "df") else DataFrame(pas)
        all_props = list(pa_df["code"]) if "code" in pa_df.columns else []

        df = o.get_objects(type=object_type, props=all_props)
        total = len(df)
        counts = {}
        for col in all_props:
            if col not in df.columns:
                continue
            non_empty = df[col].apply(
                lambda v: v is not None and v == v and v != ""   # v==v excludes NaN
            ).sum()
            if non_empty > 0:
                counts[col] = int(non_empty)
        return json.dumps({"total_entries": int(total), "property_fill_counts": counts})
    finally:
        o.logout()


def sync_rows(rows_json, url, space, collection=None, object_type=None,
              campaign_props_json=None, parent_identifier=None,
              upload_files=True, token=None, username=None, password=None,
              verify_certificates=True):
    """
    rows_json  -- JSON string: array of objects, one per table row
                  (built in MATLAB via jsonencode(table2struct(T))). Each
                  row must include 'name' and 'folder' if upload_files=True,
                  and 'resultImages' (a list of local PNG paths, or an
                  empty list) if you want plots embedded in the Results
                  field. Columns absent from a given table (e.g. a
                  beamtime without smplX_mm/Y/Z) are simply skipped --
                  nothing needs to exist in every row.
    url        -- openBIS instance URL
    space      -- openBIS space code to create/find objects in
    collection -- openBIS experiment/collection identifier,
                  e.g. '/SPACE/PROJECT/COLLECTION' (optional)
    object_type -- openBIS object type code (falls back to OBJECT_TYPE)
    campaign_props_json -- JSON string: {"campaignLocation": "...", "team":
                  "...", "projectName": "..."} -- same value applied to
                  every row this run (falls back to skipping campaign
                  properties if not provided)
    parent_identifier -- identifier (e.g. '/SPACE/BEAMTIME_CODE') of a
                  parent object representing the beamtime as a whole.
                  Applied to every row's object via add_parents() (see the
                  comment at that call for why not set_parents()/
                  get_parents()).
    upload_files -- if True, attaches the original raw file (from
                  fullfile(folder,name)) as a RAW_DATASET_TYPE dataset on
                  each object
    token / username+password -- authentication; provide one or the other

    Update behaviour: an existing object is matched by its $name property
    (not Code, which openBIS auto-generates for this type). Any property
    that already has a non-empty value on an existing object is left
    untouched -- automated sync only fills in blanks, never overwrites
    something entered by hand (0/0.0 counts as a real value, not empty).
    A property that can't be set (e.g. a vocabulary term that doesn't
    exist yet) is skipped and reported in that row's message, without
    preventing the rest of the row from saving.

    Returns a JSON string: one summary entry per row --
        [{"name": "...", "action": "created"|"updated"|"error", "message": "..."}]
    """
    rows = json.loads(rows_json)
    object_type = _clean_str_or_none(object_type) or OBJECT_TYPE
    campaign_props = json.loads(campaign_props_json) if campaign_props_json else {}

    url   = _clean_str_or_none(url)
    space = _clean_str_or_none(space)
    collection = _clean_str_or_none(collection)
    parent_identifier = _clean_str_or_none(parent_identifier)

    o = _connect(url, token=token, username=username, password=password,
                 verify_certificates=verify_certificates)
    results = []

    # Resolve the parent identifier to an actual object once, up front,
    # rather than handing pybis a bare string to resolve internally on
    # every row. Confirmed via a real traceback that passing the string
    # directly works fine at object CREATION time but breaks UPDATE
    # (pybis/attribute.py's internal string->object resolution ends up
    # leaving a raw string where save() later expects a fully-resolved
    # identifier dict, causing "string indices must be integers, not
    # 'str'"). Passing an already-resolved object sidesteps that path.
    parent_obj = None
    parent_warning = None
    if parent_identifier:
        try:
            parent_obj = o.get_object(parent_identifier)
        except Exception as e:
            parent_warning = f"could not resolve parent_identifier '{parent_identifier}': {e}"

    def _is_empty(value):
        """Deliberately not `not value` -- that would treat a genuine 0 or
        0.0 (a perfectly valid REAL/INTEGER reading, e.g. a position or
        pressure) as 'empty' and overwrite it. Only None and '' count."""
        return value is None or value == ""

    def _current_value(obj, prop_code):
        try:
            return obj.props[prop_code]
        except Exception:
            return None

    def _set_if_empty(obj, prop_code, value, warnings):
        """Sets a property only if currently empty. Any failure (e.g. an
        unrecognized property code, or a vocabulary term that doesn't
        exist yet) is recorded as a warning rather than raised, so it
        can't prevent the rest of the row's properties from saving.
        prop_code is force-lowercased -- confirmed (via a real openBIS
        error listing every valid code) that property codes are lowercase
        at the wire level even though the admin UI/xlsx export displays
        them uppercase."""
        prop_code = prop_code.lower()
        if _is_empty(value):
            return
        try:
            if _is_empty(_current_value(obj, prop_code)):
                obj.props[prop_code] = value
        except Exception as e:
            warnings.append(f"{prop_code}: {e}")

    try:
        for row in rows:
            name = row.get("name", "")
            warnings = []
            if parent_warning:
                warnings.append(parent_warning)
            try:
                category = _category_for_name(name)

                existing = o.get_objects(type=object_type, space=space, **{"$name": name})
                if len(existing) > 0:
                    obj = existing[0]
                    action = "updated"
                else:
                    new_kwargs = dict(type=object_type, space=space, props={"$name": name})
                    if collection:
                        # Only included when truthy: explicitly passing
                        # collection=None causes pybis to emit a malformed
                        # empty "experimentId": {} in the request (missing
                        # its required @type discriminator), which the
                        # server then rejects outright -- confirmed via a
                        # real server error. Omitting the key entirely
                        # avoids that path.
                        new_kwargs["collection"] = collection
                    if parent_obj:
                        new_kwargs["parents"] = [parent_obj]
                    obj = o.new_object(**new_kwargs)
                    action = "created"

                if parent_obj and action == "updated":
                    try:
                        # add_parents() (not set_parents(), and NOT
                        # preceded by a get_parents() check): get_parents()
                        # has a confirmed bug where it populates the
                        # internal _parents cache as a flat list of plain
                        # identifier strings (via a DataFrame
                        # .values.flatten()) -- a different shape than
                        # add_parents()/set_parents() produce. Merely
                        # calling it corrupts state that obj.save() later
                        # can't handle ("string indices must be integers,
                        # not 'str'"), even on rows that never touch
                        # parents at all this run. add_parents() sidesteps
                        # that path entirely, and unlike set_parents() it
                        # only adds our parent rather than replacing the
                        # whole list -- so it can't remove a different
                        # parent someone added by hand.
                        obj.add_parents([parent_obj])
                    except Exception as e:
                        warnings.append(f"parent link: {e}")

                for col, prop in DIRECT_MAP.items():
                    if col in row:
                        _set_if_empty(obj, prop, row[col], warnings)

                for col, (prop, kind) in NUMERIC_MAP.items():
                    if col in row:
                        value = row[col]
                        if kind == "int":
                            value = _to_clean_int(value)
                        _set_if_empty(obj, prop, value, warnings)

                for prop, value in _row_derived_props(row, category).items():
                    _set_if_empty(obj, prop, value, warnings)

                for prop, campaign_key in CAMPAIGN_MAP.items():
                    _set_if_empty(obj, prop, campaign_props.get(campaign_key), warnings)

                for prop, value in CONSTANT_MAP.items():
                    _set_if_empty(obj, prop, value, warnings)

                for img_path in (row.get("resultImages") or []):
                    try:
                        attach_result_image(o, obj, img_path, url, verify_certificates=verify_certificates)
                    except Exception as e:
                        warnings.append(f"result image ({img_path}): {e}")

                obj.save()

                if upload_files and row.get("folder"):
                    src_path = f"{row['folder']}/{name}"
                    try:
                        ds = o.new_dataset(type=RAW_DATASET_TYPE, sample=obj, files=[src_path])
                        ds.save()
                    except Exception as e:
                        warnings.append(f"file upload: {e}")

                results.append({"name": name, "action": action, "message": "; ".join(warnings)})
            except Exception as e:
                results.append({"name": name, "action": "error", "message": str(e)})
    finally:
        o.logout()

    return json.dumps(results)


def check_connection(url, token=None, username=None, password=None,
                      verify_certificates=True):
    """Quick standalone connectivity/auth check -- call this before sync_rows
    on a full table, to confirm login/token works.

    Deliberately does NOT use is_session_active()/isSessionActive: that RPC
    has been observed rejecting valid PATs on this openBIS instance even
    though the same token works fine for real operations (get_spaces here,
    and get_objects/new_object/etc. in sync_rows). Space listing is used
    instead as a cheap, harmless, real call that actually proves the
    token/login works end to end."""
    o = _connect(url, token=token, username=username, password=password,
                 verify_certificates=verify_certificates)
    try:
        spaces = o.get_spaces()
        working = spaces is not None
    except Exception as e:
        return json.dumps({"connection_works": False, "message": str(e)})
    finally:
        try:
            o.logout()
        except Exception:
            pass   # harmless either way; not the check we care about here
    return json.dumps({"connection_works": working, "message": ""})