From 2bcead2eab1df5a16eb245a3b2499f9b9d2acb14 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:08:36 +0100 Subject: [PATCH 1/8] Add new metadata-precheck module --- relecov_tools/metadata_precheck.py | 847 +++++++++++++++++++++++++++++ 1 file changed, 847 insertions(+) create mode 100644 relecov_tools/metadata_precheck.py diff --git a/relecov_tools/metadata_precheck.py b/relecov_tools/metadata_precheck.py new file mode 100644 index 00000000..dc845540 --- /dev/null +++ b/relecov_tools/metadata_precheck.py @@ -0,0 +1,847 @@ +#!/usr/bin/env python +from __future__ import annotations + +import json +import os +import tempfile +from collections import defaultdict +from datetime import datetime +from typing import Any, Iterable + +import rich.console +from rich.table import Table +from jsonschema import Draft202012Validator +import re + +import relecov_tools.utils +from relecov_tools.base_module import BaseModule +from relecov_tools.config_json import ConfigJson +from relecov_tools.sftp_client import SftpClient +import relecov_tools.assets.schema_utils.custom_validators + +stderr = rich.console.Console( + stderr=True, + style="dim", + highlight=False, + force_terminal=relecov_tools.utils.rich_force_colors(), +) + + +class MetadataPrecheckError(RuntimeError): + """Custom error raised when a metadata file cannot be processed.""" + + +class MetadataPrecheck(BaseModule): + def __init__( + self, + user: str | None = None, + password: str | None = None, + conf_file: str | None = None, + output_dir: str | None = None, + target_folders: Iterable[str] | str | None = None, + export_excel: bool = False, + ) -> None: + super().__init__(output_dir=output_dir, called_module=__name__) + self.log.info("Initiating metadata precheck process") + + self.config = ConfigJson(extra_config=True) + self.core_config = ConfigJson() + + generic_topic = self.core_config.get_topic_data("generic", "relecov_schema") + schema_path = os.path.join( + os.path.dirname(os.path.realpath(__file__)), "schema", generic_topic + ) + self.schema = relecov_tools.utils.read_json_file(schema_path) + + self.metadata_processing = self.core_config.get_topic_data( + "sftp_handle", "metadata_processing" + ) + self.metadata_lab_heading = self.config.get_topic_data( + "read_lab_metadata", "metadata_lab_heading" + ) + self.heading_aliases = ( + self.config.get_topic_data("read_lab_metadata", "alt_heading_equivalences") + or {} + ) + + self.required_properties = self.schema.get("required", []) + self.schema_properties = self.schema.get("properties", {}) + self.label_to_prop = {} + self.prop_to_label = {} + for prop, definition in self.schema_properties.items(): + label = definition.get("label") + if label: + self.label_to_prop[label] = prop + self.prop_to_label[prop] = label + + self.required_labels = [ + self.prop_to_label[prop] + for prop in self.required_properties + if self.prop_to_label.get(prop) + ] + + default_sample_label = self.metadata_processing.get("sample_id_col") + self.sample_id_property = self.label_to_prop.get( + default_sample_label, default_sample_label + ) + self.not_provided_field = self.core_config.get_topic_data( + "generic", "not_provided_field" + ) + starting_date = self.core_config.get_topic_data("generic", "starting_date") + try: + start_date = datetime.strptime(starting_date, "%Y-%m-%d").date() + except (TypeError, ValueError): + start_date = datetime(2020, 1, 1).date() + end_date = datetime.now().date() + self._date_checker = ( + relecov_tools.assets.schema_utils.custom_validators.make_date_checker( + start_date, end_date + ) + ) + self.validator = Draft202012Validator( + self.schema, format_checker=self._date_checker + ) + + self.sheet_options = self._build_sheet_options() + + self.export_excel = export_excel + + parsed_targets, prompt_flag = self._parse_target_folders(target_folders) + self.target_folders = parsed_targets + self.prompt_for_targets = prompt_flag + + if user is None: + user = relecov_tools.utils.prompt_text(msg="Enter the user id") + if password is None: + password = relecov_tools.utils.prompt_password(msg="Enter your password") + + self.sftp_client = SftpClient(conf_file, user, password) + + self.logsum = self.parent_log_summary(output_dir=self.basemod_outdir) + self.folder_reports: dict[str, list[dict[str, Any]]] = defaultdict(list) + self.lab_summary: dict[str, dict[str, Any]] = defaultdict( + lambda: { + "total_samples": 0, + "valid_files": [], + "invalid_files": [], + "files": [], + "invalid_sample_count": 0, + } + ) + self.generated_at = datetime.utcnow().isoformat(timespec="seconds") + "Z" + + self.set_batch_id(datetime.utcnow().strftime("%Y%m%d%H%M%S")) + + def _build_sheet_options(self) -> list[dict[str, str]]: + opts = [] + primary = { + "sheet": self.metadata_processing.get("excel_sheet"), + "header_flag": self.metadata_processing.get("header_flag"), + "sample_col": self.metadata_processing.get("sample_id_col"), + } + alternative = { + "sheet": self.metadata_processing.get("alternative_sheet"), + "header_flag": self.metadata_processing.get("alternative_flag"), + "sample_col": self.metadata_processing.get("alternative_sample_id_col"), + } + for option in (primary, alternative): + if option["sheet"] and option["header_flag"]: + opts.append(option) + return opts + + def _parse_target_folders( + self, target_folders: Iterable[str] | str | None + ) -> tuple[list[str] | None, bool]: + if target_folders is None: + return None, False + if isinstance(target_folders, str): + if target_folders == "ALL": + return None, True + clean = [ + f.strip() for f in target_folders.strip("[]").split(",") if f.strip() + ] + return clean or None, False + clean_list = [f.strip() for f in target_folders if str(f).strip()] + if clean_list and clean_list[0] == "ALL": + return None, True + return clean_list or None, False + + def execute_process(self) -> None: + if not self.sftp_client.open_connection(): + msg = "Unable to establish sftp connection" + self.log.error(msg) + stderr.print(f"[red]{msg}") + raise ConnectionError(msg) + try: + metadata_targets = self._discover_metadata_targets() + if not metadata_targets: + stderr.print("[yellow]No metadata Excel files found in remote folders") + self.log.warning("No metadata Excel files discovered") + for folder, meta_files in metadata_targets.items(): + self._process_folder(folder, meta_files) + finally: + self.sftp_client.close_connection() + + if self.logsum.logs: + self.parent_create_error_summary(to_excel=self.export_excel) + + self._export_report() + self._print_summary() + + def _discover_metadata_targets(self) -> dict[str, list[str]]: + directory_list = self.sftp_client.list_remote_folders(".", recursive=True) + clean_dirs = sorted( + {d.replace("./", "", 1) for d in directory_list if d and d != "."} + ) + + selected_dirs = self._select_target_directories(clean_dirs) + metadata_targets: dict[str, list[str]] = {} + for directory in selected_dirs: + if not directory or directory.endswith("_tmp_processing"): + continue + path_parts = [part for part in directory.split("/") if part] + if any( + part.lower() == "invalid_samples" + or part.lower().endswith("_invalid_samples") + for part in path_parts + ): + self.log.info( + "Skipping %s because it points to an *_invalid_samples folder", + directory, + ) + continue + try: + file_list = self.sftp_client.get_file_list(directory) + except FileNotFoundError: + self.log.warning("Folder %s not found during listing", directory) + continue + meta_files = [ + f + for f in file_list + if f.lower().endswith(".xlsx") + and not os.path.basename(f).startswith((".~lock", "~$")) + ] + if meta_files: + metadata_targets[directory] = sorted(meta_files) + return metadata_targets + + def _select_target_directories(self, clean_dirs: list[str]) -> list[str]: + if self.prompt_for_targets: + choices = sorted(clean_dirs) + if not choices: + return [] + selected = relecov_tools.utils.prompt_checkbox( + msg="Select the folders to validate", choices=choices + ) + return selected + if self.target_folders is None: + return clean_dirs + missing = sorted(set(self.target_folders) - set(clean_dirs)) + for folder in missing: + self.log.warning("Target folder %s not present in remote tree", folder) + stderr.print(f"[yellow]Target folder {folder} not present in remote tree") + return [folder for folder in self.target_folders if folder in clean_dirs] + + def _process_folder(self, folder: str, meta_files: list[str]) -> None: + stderr.print(f"[blue]Processing folder {folder}") + self.log.info("Processing folder %s", folder) + self.logsum.feed_key(key=folder) + for remote_file in meta_files: + file_summary = self._validate_remote_metadata(folder, remote_file) + lab_code = folder.split("/")[0] if folder else "root" + self._update_lab_summary(lab_code, folder, file_summary) + self.folder_reports[folder].append(file_summary) + + def _validate_remote_metadata( + self, folder: str, remote_file: str + ) -> dict[str, Any]: + file_errors: list[dict[str, Any]] = [] + file_warnings: list[dict[str, Any]] = [] + sample_count = 0 + seen_samples: set[str] = set() + row_contexts: list[dict[str, Any]] = [] + + with tempfile.TemporaryDirectory() as tmp_dir: + local_target = os.path.join(tmp_dir, os.path.basename(remote_file)) + try: + self.sftp_client.get_from_sftp(remote_file, local_target, exist_ok=True) + except Exception as exc: # pragma: no cover - passthrough for runtime + message = f"Could not download metadata file {remote_file}: {exc}" + self._record_error(folder, file_errors, message) + self.log.error(message) + return { + "remote_file": remote_file, + "samples": 0, + "valid": False, + "errors": file_errors, + "warnings": file_warnings, + } + + try: + sheet_data = self._read_metadata_sheet(local_target) + except MetadataPrecheckError as exc: + message = f"Unable to read metadata sheet {remote_file}: {exc}" + self._record_error(folder, file_errors, message) + self.log.error(message) + return { + "remote_file": remote_file, + "samples": 0, + "valid": False, + "errors": file_errors, + "warnings": file_warnings, + } + + rows = sheet_data["rows"] + header_flag = sheet_data["header_flag"] + raw_header = sheet_data["header"] + sample_column_raw = sheet_data["sample_column"] + header_row_index = sheet_data["heading_row"] + + if not rows: + message = f"Metadata sheet {remote_file} contains no data rows" + self._record_error(folder, file_errors, message) + self.log.error(message) + return { + "remote_file": remote_file, + "samples": 0, + "valid": False, + "errors": file_errors, + "warnings": file_warnings, + } + + canonical_header = [self._canonical_label(col) for col in raw_header] + if canonical_header and canonical_header[0] == header_flag: + data_columns = canonical_header[1:] + else: + data_columns = canonical_header + + missing_columns = [ + column for column in self.metadata_lab_heading if column not in data_columns + ] + if missing_columns: + message = "Missing columns in metadata header: " + ", ".join( + sorted(missing_columns) + ) + self._record_warning(folder, file_warnings, message) + self.log.warning(message) + + extra_columns = [ + column for column in data_columns if column not in self.metadata_lab_heading + ] + if extra_columns: + message = "Unexpected columns found: " + ", ".join( + sorted(set(extra_columns)) + ) + self._record_warning(folder, file_warnings, message) + self.log.warning(message) + + sample_column = self._canonical_label(sample_column_raw) + required_labels_in_sheet = [ + self._canonical_label(label) + for label in self.required_labels + if self._canonical_label(label) in data_columns + ] + + available_schema_keys = { + self.label_to_prop.get(column, column) + for column in data_columns + if self.label_to_prop.get(column, column) in self.schema_properties + } + + for idx, row in enumerate(rows): + row_number = header_row_index + 1 + idx + canonical_row = self._canonicalize_row(row, header_flag) + sample_value_raw = canonical_row.get(sample_column) + sample_value = self._normalise_sample_id(canonical_row, sample_value_raw) + + if not sample_value: + message = f"Missing sequencing sample identifier at row {row_number}" + self._record_error( + folder, + file_errors, + message, + sample=None, + row=row_number, + ) + else: + if sample_value in seen_samples: + message = ( + f"Duplicated sequencing sample identifier {sample_value} " + f"at row {row_number}" + ) + self._record_warning( + folder, + file_warnings, + message, + sample=sample_value, + row=row_number, + ) + continue + seen_samples.add(sample_value) + sample_count += 1 + + schema_payload = self._row_to_schema_payload(canonical_row) + row_contexts.append( + { + "row": row_number, + "sample": sample_value if sample_value else None, + "payload": schema_payload, + "available_properties": available_schema_keys, + } + ) + + self._apply_schema_validation(row_contexts, folder, file_errors) + + invalid_samples_details: dict[str, list[dict[str, Any]]] = defaultdict(list) + for err in file_errors: + sample_label = err.get("sample") + row = err.get("row") + if not sample_label and row is not None: + sample_label = f"row {row}" + if not sample_label: + continue + invalid_samples_details[sample_label].append( + {"message": err["message"], "row": row} + ) + + invalid_samples = [] + for sample_label, details in invalid_samples_details.items(): + unique_rows = sorted({d["row"] for d in details if d["row"] is not None}) + unique_messages = [] + seen_messages = set() + for detail in details: + msg = detail["message"] + if msg not in seen_messages: + unique_messages.append(msg) + seen_messages.add(msg) + invalid_samples.append( + { + "sample": sample_label, + "rows": unique_rows, + "messages": unique_messages, + "summary": "; ".join(unique_messages), + } + ) + + file_valid = not file_errors + return { + "remote_file": remote_file, + "samples": sample_count, + "valid": file_valid, + "errors": file_errors, + "warnings": file_warnings, + "invalid_samples": invalid_samples, + } + + def _read_metadata_sheet(self, local_path: str) -> dict[str, Any]: + errors: list[str] = [] + for option in self.sheet_options: + try: + rows, heading_row = relecov_tools.utils.read_excel_file( + local_path, + option["sheet"], + option["header_flag"], + leave_empty=True, + ) + header = list(rows[0].keys()) if rows else [] + return { + "rows": rows, + "header": header, + "header_flag": option["header_flag"], + "sample_column": option["sample_col"], + "heading_row": heading_row, + } + except Exception as exc: # pragma: no cover - passthrough for runtime + errors.append(f"{option['sheet']}: {exc}") + if errors: + raise MetadataPrecheckError("; ".join(errors)) + raise MetadataPrecheckError("No readable sheet found in metadata Excel") + + def _canonicalize_row( + self, row: dict[str, Any], header_flag: str + ) -> dict[str, Any]: + canonical = {} + for key, value in row.items(): + if key == header_flag: + continue + canonical_key = self._canonical_label(key) + canonical[canonical_key] = value + return canonical + + def _row_to_schema_payload(self, canonical_row: dict[str, Any]) -> dict[str, Any]: + payload: dict[str, Any] = {} + for label, value in canonical_row.items(): + schema_key = self.label_to_prop.get(label, label) + if schema_key not in self.schema_properties: + continue + prepared = self._prepare_schema_value(schema_key, value) + if prepared is None: + continue + payload[schema_key] = prepared + return payload + + def _prepare_schema_value(self, schema_key: str, value: Any) -> Any: + cleaned = self._clean_cell(value) + if cleaned is None: + return None + if isinstance(cleaned, str) and not cleaned.strip(): + return None + schema_def = self.schema_properties.get(schema_key, {}) + schema_type = schema_def.get("type") + if isinstance(schema_type, list): + schema_type = [t for t in schema_type if t != "null"] + schema_type = schema_type[0] if schema_type else None + if schema_type == "string": + string_value = self._normalise_string_value(schema_key, cleaned) + return self._align_to_enum(schema_key, string_value) + if schema_type in {"integer", "number", "boolean"}: + return relecov_tools.utils.cast_value_to_schema_type(cleaned, schema_type) + if schema_type == "array": + if isinstance(cleaned, list): + raw_items = cleaned + elif isinstance(cleaned, str): + delimiter = schema_def.get("delimiter", ";") + raw_items = [ + part.strip() for part in cleaned.split(delimiter) if part.strip() + ] + else: + return cleaned + normalised_items = [] + for item in raw_items: + normalised_item = self._normalise_string_value(schema_key, item) + aligned_item = self._align_to_enum( + schema_key, normalised_item, is_array_item=True + ) + normalised_items.append(aligned_item) + return normalised_items + return cleaned + + def _normalise_string_value(self, schema_key: str, value: Any) -> str: + if isinstance(value, datetime): + if "date" in schema_key.lower(): + return value.date().isoformat() + return value.isoformat() + if isinstance(value, str): + text = value.strip() + if "date" in schema_key.lower(): + return self._normalise_date_string(text) + return text + if isinstance(value, (int, float)): + if "date" in schema_key.lower(): + return self._normalise_date_string(str(int(value))) + return str(value) + return str(value) + + def _normalise_date_string(self, value: str) -> str: + if value is None: + return value + value = str(value) + if not value: + return value + low_value = value.lower() + if self.not_provided_field and low_value == self.not_provided_field.lower(): + return value + value = value.replace("/", "-").replace(".", "-") + pattern_match = re.match(r"^\d{4}-\d{2}-\d{2}", value) + if pattern_match: + return pattern_match.group(0) + try: + parsed = datetime.strptime(value, "%Y%m%d").date() + return parsed.isoformat() + except ValueError: + return value + + def _strip_ontology(self, value: str) -> str: + if not isinstance(value, str): + return str(value) + if "[" in value and value.strip().endswith("]"): + return value.split("[", 1)[0].strip() + return value.strip() + + def _align_to_enum( + self, schema_key: str, value: Any, *, is_array_item: bool = False + ) -> Any: + schema_def = self.schema_properties.get(schema_key, {}) + enum_values = schema_def.get("enum") + if is_array_item and not enum_values: + enum_values = schema_def.get("items", {}).get("enum", []) + if not enum_values or value is None: + return value + if value in enum_values: + return value + stripped_map = { + self._strip_ontology(enum_val).lower(): enum_val for enum_val in enum_values + } + value_stripped = self._strip_ontology(str(value)).lower() + matched = stripped_map.get(value_stripped) + if matched: + return matched + # fall back to case insensitive direct comparison + for enum_val in enum_values: + if enum_val.lower() == str(value).lower(): + return enum_val + return value + + def _apply_schema_validation( + self, + contexts: list[dict[str, Any]], + folder: str, + file_errors: list[dict[str, Any]], + ) -> None: + schema_props = self.schema_properties + seen_errors: set[tuple[str | None, int | None, str]] = set() + for idx, context in enumerate(contexts): + payload = context.get("payload") or {} + if not payload: + continue + available_props = set(context.get("available_properties") or []) + validation_errors = list(self.validator.iter_errors(payload)) + validation_errors = relecov_tools.assets.schema_utils.custom_validators.validate_with_exceptions( + self.schema, payload, validation_errors + ) + if not validation_errors: + continue + sample_label = context.get("sample") + schema_sample = ( + payload.get(self.sample_id_property) + if self.sample_id_property + else None + ) + if not sample_label and schema_sample: + sample_label = schema_sample + row_number = context.get("row") + display_label = ( + sample_label + if sample_label + else (f"row {row_number}" if row_number is not None else None) + ) + for error in validation_errors: + if error.cause: + error.message = str(error.cause) + if error.validator == "required": + try: + missing_field = list(error.message.split("'"))[1] + except Exception: + missing_field = None + if missing_field and missing_field not in available_props: + continue + error_text = self._format_validation_error(error, schema_props) + error_key = (display_label, row_number, error_text) + if error_key in seen_errors: + continue + seen_errors.add(error_key) + self._record_error( + folder, + file_errors, + error_text, + sample=display_label, + row=row_number, + ) + + def _format_validation_error(self, error, schema_props: dict[str, Any]) -> str: + def get_property_label(prop_key: str) -> str: + prop_def = schema_props.get(prop_key, {}) + return prop_def.get("label", prop_key) + + try: + if error.validator == "required": + error_field = list(error.message.split("'"))[1] + elif error.validator == "anyOf": + multi_errdict = {} + for suberror in error.context: + error_type = suberror.validator + failing_field = suberror.validator_value[0] + sub_label = get_property_label(failing_field) + label_message = suberror.message.replace(failing_field, sub_label) + multi_errdict.setdefault(error_type, []).append( + (sub_label, label_message) + ) + error_field = "" + multi_message = {} + for errtype, fieldtups in multi_errdict.items(): + failed_fields = " or ".join([t[0] for t in fieldtups]) + clean_message = ( + fieldtups[0][1].replace(fieldtups[0][0], "").strip("'") + ) + if error_field: + error_field = error_field + " and" + error_field = error_field + failed_fields + multi_message[errtype] = f"{failed_fields}: {clean_message}" + error.message = "Any of the following: " + " --- ".join( + multi_message.values() + ) + elif error.absolute_path: + error_field = str(error.absolute_path[0]) + else: + error_field = error.validator_value + except Exception: + return f"Validation error: {error.message}" + + field_label = get_property_label(error_field) + message = error.message.replace(str(error_field), field_label) + return f"Error in column {field_label}: {message}" + + def _canonical_label(self, label: Any) -> str: + if label is None: + return "" + label_str = str(label).strip() + return self.heading_aliases.get(label_str, label_str) + + def _normalise_sample_id( + self, row: dict[str, Any], sample_value_raw: Any + ) -> str | None: + candidates = [sample_value_raw] + fallback_keys = [ + "Sample ID given by the submitting laboratory", + "Sample ID given by originating laboratory", + "Sample ID given in the microbiology lab", + "Sample ID given if multiple rna-extraction or passages", + "Sequence file R1", + "Sequence file R1 fastq", + ] + for key in fallback_keys: + candidates.append(row.get(key)) + for value in candidates: + value_clean = self._clean_cell(value) + if self._value_is_present(value_clean): + return str(value_clean).strip() + return None + + def _clean_cell(self, value: Any) -> Any: + if isinstance(value, float) and value != value: + return None + if isinstance(value, str): + stripped = value.strip() + return stripped + return value + + def _value_is_present(self, value: Any) -> bool: + if value is None: + return False + if isinstance(value, str): + stripped = value.strip() + if not stripped: + return False + if "not provided" in stripped.lower(): + return False + return True + + def _record_error( + self, + folder: str, + file_errors: list[dict[str, Any]], + message: str, + sample: str | None = None, + row: int | None = None, + ) -> None: + if sample: + self.logsum.add_error(entry=message, key=folder, sample=sample) + else: + self.logsum.add_error(entry=message, key=folder) + file_errors.append({"message": message, "sample": sample, "row": row}) + + def _record_warning( + self, + folder: str, + file_warnings: list[dict[str, Any]], + message: str, + sample: str | None = None, + row: int | None = None, + ) -> None: + if sample: + self.logsum.add_warning(entry=message, key=folder, sample=sample) + else: + self.logsum.add_warning(entry=message, key=folder) + file_warnings.append({"message": message, "sample": sample, "row": row}) + + def _update_lab_summary( + self, lab_code: str, folder: str, file_summary: dict[str, Any] + ) -> None: + lab_entry = self.lab_summary[lab_code] + lab_entry["total_samples"] += file_summary.get("samples", 0) + lab_entry["invalid_sample_count"] += len( + file_summary.get("invalid_samples", []) + ) + if file_summary.get("valid"): + lab_entry["valid_files"].append( + remote_path := file_summary.get("remote_file") + ) + else: + lab_entry["invalid_files"].append( + remote_path := file_summary.get("remote_file") + ) + lab_entry["files"].append( + { + "folder": folder, + "remote_file": remote_path, + "samples": file_summary.get("samples", 0), + "valid": file_summary.get("valid", False), + "errors": file_summary.get("errors", []), + "warnings": file_summary.get("warnings", []), + "invalid_samples": file_summary.get("invalid_samples", []), + } + ) + + def _export_report(self) -> None: + report = { + "generated_at": self.generated_at, + "total_labs": len(self.lab_summary), + "labs": self.lab_summary, + "folders": self.folder_reports, + } + report_path = self.tag_filename( + os.path.join(self.basemod_outdir, "metadata_precheck_report.json") + ) + try: + os.makedirs(os.path.dirname(report_path), exist_ok=True) + with open(report_path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, ensure_ascii=False) + stderr.print(f"[green]Metadata precheck report saved at {report_path}") + except OSError as exc: + self.log.error("Could not write report %s: %s", report_path, exc) + stderr.print(f"[red]Could not write metadata precheck report: {exc}") + + def _print_summary(self) -> None: + if not self.lab_summary: + return + table = Table(title="Metadata precheck summary", show_lines=False) + table.add_column("Lab", justify="left", no_wrap=True) + table.add_column("Valid files", justify="right") + table.add_column("Invalid files", justify="right") + table.add_column("Samples", justify="right") + table.add_column("Invalid samples", justify="right") + table.add_column("Top issues", justify="left") + + def _collect_lab_messages(files: list[dict[str, Any]], limit: int = 3): + messages = [] + for fdata in files: + for inv in fdata.get("invalid_samples", []): + messages.extend(inv.get("messages", [])) + unique_msgs = [] + for msg in messages: + if msg not in unique_msgs: + unique_msgs.append(msg) + if len(unique_msgs) >= limit: + break + if len(unique_msgs) < len(set(messages)): + unique_msgs.append("...") + truncated = [] + for msg in unique_msgs: + if len(msg) > 120: + truncated.append(msg[:117] + "...") + else: + truncated.append(msg) + return truncated + + for lab_code, entry in sorted(self.lab_summary.items()): + lab_messages = _collect_lab_messages(entry.get("files", [])) + summary_text = "\n".join(lab_messages) if lab_messages else "—" + table.add_row( + lab_code, + str(len(entry.get("valid_files", []))), + str(len(entry.get("invalid_files", []))), + str(entry.get("total_samples", 0)), + str(entry.get("invalid_sample_count", 0)), + summary_text, + ) + stderr.print(table) From ef26b0449e6ff5900e5253a2639a8723cff786dc Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:08:55 +0100 Subject: [PATCH 2/8] Add metadata-precheck module to __main__.py --- relecov_tools/__main__.py | 65 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/relecov_tools/__main__.py b/relecov_tools/__main__.py index 3c3f4e7a..e403d337 100755 --- a/relecov_tools/__main__.py +++ b/relecov_tools/__main__.py @@ -28,6 +28,7 @@ import relecov_tools.ena_upload import relecov_tools.pipeline_manager import relecov_tools.build_schema +import relecov_tools.metadata_precheck import relecov_tools.wrapper import relecov_tools.upload_results import relecov_tools.base_module @@ -343,6 +344,70 @@ def download( sys.exit(1) +@relecov_tools_cli.command(help_priority=3) +@click.option("-u", "--user", help="User name for login to sftp server") +@click.option("-p", "--password", help="Password for the user to login") +@click.option( + "-f", + "--conf_file", + help="Configuration file (not params file)", +) +@click.option( + "-o", + "--output_dir", + "--output-dir", + "--output_folder", + "--out-folder", + "--output_location", + "--output_path", + "--out_dir", + "--output", + "output_dir", + type=click.Path(file_okay=False, resolve_path=True), + help="Directory where the generated output and logs will be saved", +) +@click.option( + "-t", + "--target_folders", + is_flag=False, + flag_value="ALL", + default=None, + help='Flag: Select which folders will be targeted giving [paths] or via prompt. For multiple folders use ["folder1", "folder2"]', +) +@click.option( + "--export-excel/--no-export-excel", + default=False, + help="Generate an Excel summary alongside the JSON report", +) +@click.pass_context +def metadata_precheck( + ctx, + user, + password, + conf_file, + output_dir, + target_folders, + export_excel, +): + """Inspect remote metadata Excels and report missing required data.""" + debug = ctx.obj.get("debug", False) + args_merged = merge_with_extra_config( + ctx=ctx, + add_extra_config=True, + ) + try: + precheck = relecov_tools.metadata_precheck.MetadataPrecheck(**args_merged) + precheck.execute_process() + except Exception as e: + if debug: + log.exception(f"EXCEPTION FOUND: {e}") + raise + else: + log.exception(f"EXCEPTION FOUND: {e}") + stderr.print(f"EXCEPTION FOUND: {e}") + sys.exit(1) + + # metadata @relecov_tools_cli.command(help_priority=3) @click.option( From 83385ae2cedde69ddce32549e92395b7a4968a06 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:09:10 +0100 Subject: [PATCH 3/8] Update README with new metadata-precheck module --- README.md | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/README.md b/README.md index bf308a75..1228e16e 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ relecov-tools is a set of helper tools for the assembly of the different element - [send-mail](#send-mail) - [read-bioinfo-metadata](#read-bioinfo-metadata) - [Configuration of module `read-bioinfo-metadata`](#configuration-of-module-read-bioinfo-metadata) + - [metadata-precheck](#metadata-precheck) - [validate](#validate) - [map](#map) - [upload-to-ena](#upload-to-ena) @@ -132,6 +133,7 @@ Commands: upload-to-gisaid parsed data to create files to upload to gisaid update-db upload the information included in json file to... read-bioinfo-metadata Create the json compliant from the Bioinfo... + metadata-precheck Scan metadata excels in the SFTP before download. metadata-homogeneizer Parse institution metadata lab to the one used... pipeline-manager Create the symbolic links for the samples which... build-schema Generates and updates JSON Schema files from... @@ -271,6 +273,45 @@ Options: ##### Configuration of module `read-bioinfo-metadata` +#### metadata-precheck + +Use this module at the very beginning of the SFTP workflow to inspect every metadata Excel uploaded by the labs before launching the heavy download/processing steps. It reads the files in place (without deleting them), verifies that the template columns are present, and performs a deep JSON-schema validation on each row so you immediately know which lab/sample must be corrected. + +``` +$ relecov-tools metadata-precheck --help +Usage: relecov-tools metadata-precheck [OPTIONS] + + Inspect remote metadata Excels and report missing required data. + +Options: + -u, --user TEXT User name for login to sftp server. + -p, --password TEXT Password for the user to login. + -f, --conf_file TEXT Configuration file (not params file). If omitted, + default values are taken from configuration.json and + extra_config.json. + -o, --output_dir TEXT Directory where logs and reports will be saved. Falls + back to the default logs path when not provided. + -t, --target_folders TEXT Target remote folders. Accepts a JSON-like list + (e.g. ["LAB001/batch1","LAB002"]). Use "ALL" to open an + interactive selector or leave empty to scan every folder. + --export-excel / --no-export-excel + Generate an Excel summary alongside the JSON report. + Disabled by default. + --help Show this message and exit. +``` + +**Highlights** + +- Recursively scans the SFTP tree (skipping folders tagged as `_invalid_samples`) and only downloads metadata `.xlsx` files to a temporary location for validation. +- Checks Excel structure (headers, duplicated samples, missing IDs) and runs the same JSON-schema validation used by `validate`, including enum/type/anyOf rules. +- Outputs per-lab/per-file reports under the selected `output_dir`, including: + - `_metadata_precheck.log` + `*_log_summary.json`. + - `metadata_precheck_report__.json` with the list of labs, files, number of samples per status, and detailed invalid samples/errors. + - Optional Excel summary when `--export-excel` is enabled. +- Prints a Rich table summarising, per lab, the number of valid/invalid files, samples, invalid samples, and the top validation issues so you can spot problems at a glance. + +Run this module before `download`/`wrapper` to ensure the labs fix their metadata first; once the report is clean, `validate` should succeed on the first try. + The [`bioinfo_config.json`](relecov_tools/conf/bioinfo_config.json) file is a configuration file used by the `read-bioinfo-metadata` module. Its purpose is to specify **which files to search for** and **how to extract relevant information** from a folder containing bioinformatics results. With this configuration, the module identifies parameters and results for each sample and returns them in a standardized JSON format. Structure: From 01203c5ded055b755a9707689a4fbbf5fe769795 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:27:47 +0100 Subject: [PATCH 4/8] Update metadata_precheck.py --- relecov_tools/metadata_precheck.py | 485 ++++++++++++++--------------- 1 file changed, 240 insertions(+), 245 deletions(-) diff --git a/relecov_tools/metadata_precheck.py b/relecov_tools/metadata_precheck.py index dc845540..8cc48cd1 100644 --- a/relecov_tools/metadata_precheck.py +++ b/relecov_tools/metadata_precheck.py @@ -31,6 +31,188 @@ class MetadataPrecheckError(RuntimeError): """Custom error raised when a metadata file cannot be processed.""" +class SchemaMapper: + """Handle Excel header normalisation and schema-aware casting.""" + + SAMPLE_FALLBACKS = [ + "Sample ID given by the submitting laboratory", + "Sample ID given by originating laboratory", + "Sample ID given in the microbiology lab", + "Sample ID given if multiple rna-extraction or passages", + "Sequence file R1", + "Sequence file R1 fastq", + ] + + def __init__( + self, + schema_properties: dict[str, Any], + label_to_prop: dict[str, str], + heading_aliases: dict[str, str] | None, + not_provided_field: str | None, + ) -> None: + self.schema_properties = schema_properties + self.label_to_prop = label_to_prop + self.heading_aliases = heading_aliases or {} + self.not_provided_field = (not_provided_field or "").lower() + self.sample_fallbacks = [ + self.canonical_label(label) for label in self.SAMPLE_FALLBACKS + ] + + def canonical_label(self, label: Any) -> str: + if label is None: + return "" + label_str = str(label).strip() + return self.heading_aliases.get(label_str, label_str) + + def canonicalize_row( + self, excel_row: dict[str, Any], header_flag: str + ) -> dict[str, Any]: + """Normalise header names and drop the metadata flag column.""" + canon_row = {} + for key, value in excel_row.items(): + if key == header_flag: + continue + canon_row[self.canonical_label(key)] = value + return canon_row + + def normalise_sample_id( + self, canonical_row: dict[str, Any], sample_column: str + ) -> str | None: + """Try to extract the sequencing sample id with sensible fallbacks.""" + lookup_order = [sample_column] + self.sample_fallbacks + for key in lookup_order: + candidate = canonical_row.get(key) + cleaned = self._clean_cell(candidate) + if cleaned: + return str(cleaned).strip() + return None + + def row_to_payload(self, canonical_row: dict[str, Any]) -> dict[str, Any]: + """Convert a canonical Excel row into a schema-ready dict.""" + payload: dict[str, Any] = {} + for label, value in canonical_row.items(): + schema_key = self.label_to_prop.get(label, label) + if schema_key not in self.schema_properties: + continue + prepared = self._prepare_value(schema_key, value) + if prepared is not None: + payload[schema_key] = prepared + return payload + + # ── Helpers ────────────────────────────────────────────────────────── + def _prepare_value(self, schema_key: str, value: Any) -> Any: + cleaned = self._clean_cell(value) + if cleaned is None: + return None + if isinstance(cleaned, str) and not cleaned.strip(): + return None + + schema_def = self.schema_properties.get(schema_key, {}) + schema_type = schema_def.get("type") + if isinstance(schema_type, list): + schema_type = [s for s in schema_type if s != "null"] + schema_type = schema_type[0] if schema_type else None + + if schema_type == "string": + string_value = self._normalise_string(schema_key, cleaned) + return self._align_to_enum(schema_key, string_value) + + if schema_type in {"integer", "number", "boolean"}: + return relecov_tools.utils.cast_value_to_schema_type(cleaned, schema_type) + + if schema_type == "array": + return self._prepare_array(schema_key, cleaned) + + return cleaned + + def _prepare_array(self, schema_key: str, value: Any) -> list[Any] | Any: + if isinstance(value, list): + raw_items = value + elif isinstance(value, str): + delimiter = self.schema_properties.get(schema_key, {}).get("delimiter", ";") + raw_items = [ + part.strip() for part in value.split(delimiter) if part.strip() + ] + else: + return value + + normalised = [] + for item in raw_items: + normalised_item = self._normalise_string(schema_key, item) + aligned_item = self._align_to_enum( + schema_key, normalised_item, is_array_item=True + ) + normalised.append(aligned_item) + return normalised + + def _normalise_string(self, schema_key: str, value: Any) -> str: + if isinstance(value, datetime): + return ( + value.date().isoformat() + if "date" in schema_key.lower() + else value.isoformat() + ) + if isinstance(value, (int, float)): + text = str(int(value)) if float(value).is_integer() else str(value) + else: + text = str(value) + text = text.strip() + if "date" in schema_key.lower(): + return self._normalise_date(text) + return text + + def _normalise_date(self, value: str) -> str: + if not value: + return value + if value.lower() == self.not_provided_field: + return value + clean = value.replace("/", "-").replace(".", "-") + match = re.match(r"^\d{4}-\d{2}-\d{2}", clean) + if match: + return match.group(0) + try: + parsed = datetime.strptime(clean, "%Y%m%d").date() + return parsed.isoformat() + except ValueError: + return value + + def _align_to_enum( + self, schema_key: str, value: Any, *, is_array_item: bool = False + ) -> Any: + schema_def = self.schema_properties.get(schema_key, {}) + enum_values = schema_def.get("enum") + if is_array_item and not enum_values: + enum_values = schema_def.get("items", {}).get("enum", []) + if not enum_values or value is None: + return value + if value in enum_values: + return value + stripped_map = { + self._strip_ontology(enum_val).lower(): enum_val for enum_val in enum_values + } + candidate = self._strip_ontology(value).lower() + if candidate in stripped_map: + return stripped_map[candidate] + for enum_val in enum_values: + if enum_val.lower() == str(value).lower(): + return enum_val + return value + + @staticmethod + def _strip_ontology(value: Any) -> str: + text = str(value) + if "[" in text and text.strip().endswith("]"): + return text.split("[", 1)[0].strip() + return text.strip() + + def _clean_cell(self, value: Any) -> Any: + if isinstance(value, float) and value != value: # NaN + return None + if isinstance(value, str): + return value.strip() + return value + + class MetadataPrecheck(BaseModule): def __init__( self, @@ -74,12 +256,6 @@ def __init__( self.label_to_prop[label] = prop self.prop_to_label[prop] = label - self.required_labels = [ - self.prop_to_label[prop] - for prop in self.required_properties - if self.prop_to_label.get(prop) - ] - default_sample_label = self.metadata_processing.get("sample_id_col") self.sample_id_property = self.label_to_prop.get( default_sample_label, default_sample_label @@ -102,6 +278,13 @@ def __init__( self.schema, format_checker=self._date_checker ) + self.mapper = SchemaMapper( + self.schema_properties, + self.label_to_prop, + self.heading_aliases, + self.not_provided_field, + ) + self.sheet_options = self._build_sheet_options() self.export_excel = export_excel @@ -259,7 +442,6 @@ def _validate_remote_metadata( file_warnings: list[dict[str, Any]] = [] sample_count = 0 seen_samples: set[str] = set() - row_contexts: list[dict[str, Any]] = [] with tempfile.TemporaryDirectory() as tmp_dir: local_target = os.path.join(tmp_dir, os.path.basename(remote_file)) @@ -309,7 +491,7 @@ def _validate_remote_metadata( "warnings": file_warnings, } - canonical_header = [self._canonical_label(col) for col in raw_header] + canonical_header = [self.mapper.canonical_label(col) for col in raw_header] if canonical_header and canonical_header[0] == header_flag: data_columns = canonical_header[1:] else: @@ -335,12 +517,7 @@ def _validate_remote_metadata( self._record_warning(folder, file_warnings, message) self.log.warning(message) - sample_column = self._canonical_label(sample_column_raw) - required_labels_in_sheet = [ - self._canonical_label(label) - for label in self.required_labels - if self._canonical_label(label) in data_columns - ] + sample_column = self.mapper.canonical_label(sample_column_raw) available_schema_keys = { self.label_to_prop.get(column, column) @@ -350,9 +527,9 @@ def _validate_remote_metadata( for idx, row in enumerate(rows): row_number = header_row_index + 1 + idx - canonical_row = self._canonicalize_row(row, header_flag) + canonical_row = self.mapper.canonicalize_row(row, header_flag) sample_value_raw = canonical_row.get(sample_column) - sample_value = self._normalise_sample_id(canonical_row, sample_value_raw) + sample_value = self.mapper.normalise_sample_id(canonical_row, sample_column) if not sample_value: message = f"Missing sequencing sample identifier at row {row_number}" @@ -380,18 +557,16 @@ def _validate_remote_metadata( seen_samples.add(sample_value) sample_count += 1 - schema_payload = self._row_to_schema_payload(canonical_row) - row_contexts.append( - { - "row": row_number, - "sample": sample_value if sample_value else None, - "payload": schema_payload, - "available_properties": available_schema_keys, - } + schema_payload = self.mapper.row_to_payload(canonical_row) + self._validate_payload( + payload=schema_payload, + sample_label=sample_value if sample_value else None, + row_number=row_number, + available_props=available_schema_keys, + folder=folder, + file_errors=file_errors, ) - self._apply_schema_validation(row_contexts, folder, file_errors) - invalid_samples_details: dict[str, list[dict[str, Any]]] = defaultdict(list) for err in file_errors: sample_label = err.get("sample") @@ -457,186 +632,51 @@ def _read_metadata_sheet(self, local_path: str) -> dict[str, Any]: raise MetadataPrecheckError("; ".join(errors)) raise MetadataPrecheckError("No readable sheet found in metadata Excel") - def _canonicalize_row( - self, row: dict[str, Any], header_flag: str - ) -> dict[str, Any]: - canonical = {} - for key, value in row.items(): - if key == header_flag: - continue - canonical_key = self._canonical_label(key) - canonical[canonical_key] = value - return canonical - - def _row_to_schema_payload(self, canonical_row: dict[str, Any]) -> dict[str, Any]: - payload: dict[str, Any] = {} - for label, value in canonical_row.items(): - schema_key = self.label_to_prop.get(label, label) - if schema_key not in self.schema_properties: - continue - prepared = self._prepare_schema_value(schema_key, value) - if prepared is None: - continue - payload[schema_key] = prepared - return payload - - def _prepare_schema_value(self, schema_key: str, value: Any) -> Any: - cleaned = self._clean_cell(value) - if cleaned is None: - return None - if isinstance(cleaned, str) and not cleaned.strip(): - return None - schema_def = self.schema_properties.get(schema_key, {}) - schema_type = schema_def.get("type") - if isinstance(schema_type, list): - schema_type = [t for t in schema_type if t != "null"] - schema_type = schema_type[0] if schema_type else None - if schema_type == "string": - string_value = self._normalise_string_value(schema_key, cleaned) - return self._align_to_enum(schema_key, string_value) - if schema_type in {"integer", "number", "boolean"}: - return relecov_tools.utils.cast_value_to_schema_type(cleaned, schema_type) - if schema_type == "array": - if isinstance(cleaned, list): - raw_items = cleaned - elif isinstance(cleaned, str): - delimiter = schema_def.get("delimiter", ";") - raw_items = [ - part.strip() for part in cleaned.split(delimiter) if part.strip() - ] - else: - return cleaned - normalised_items = [] - for item in raw_items: - normalised_item = self._normalise_string_value(schema_key, item) - aligned_item = self._align_to_enum( - schema_key, normalised_item, is_array_item=True - ) - normalised_items.append(aligned_item) - return normalised_items - return cleaned - - def _normalise_string_value(self, schema_key: str, value: Any) -> str: - if isinstance(value, datetime): - if "date" in schema_key.lower(): - return value.date().isoformat() - return value.isoformat() - if isinstance(value, str): - text = value.strip() - if "date" in schema_key.lower(): - return self._normalise_date_string(text) - return text - if isinstance(value, (int, float)): - if "date" in schema_key.lower(): - return self._normalise_date_string(str(int(value))) - return str(value) - return str(value) - - def _normalise_date_string(self, value: str) -> str: - if value is None: - return value - value = str(value) - if not value: - return value - low_value = value.lower() - if self.not_provided_field and low_value == self.not_provided_field.lower(): - return value - value = value.replace("/", "-").replace(".", "-") - pattern_match = re.match(r"^\d{4}-\d{2}-\d{2}", value) - if pattern_match: - return pattern_match.group(0) - try: - parsed = datetime.strptime(value, "%Y%m%d").date() - return parsed.isoformat() - except ValueError: - return value - - def _strip_ontology(self, value: str) -> str: - if not isinstance(value, str): - return str(value) - if "[" in value and value.strip().endswith("]"): - return value.split("[", 1)[0].strip() - return value.strip() - - def _align_to_enum( - self, schema_key: str, value: Any, *, is_array_item: bool = False - ) -> Any: - schema_def = self.schema_properties.get(schema_key, {}) - enum_values = schema_def.get("enum") - if is_array_item and not enum_values: - enum_values = schema_def.get("items", {}).get("enum", []) - if not enum_values or value is None: - return value - if value in enum_values: - return value - stripped_map = { - self._strip_ontology(enum_val).lower(): enum_val for enum_val in enum_values - } - value_stripped = self._strip_ontology(str(value)).lower() - matched = stripped_map.get(value_stripped) - if matched: - return matched - # fall back to case insensitive direct comparison - for enum_val in enum_values: - if enum_val.lower() == str(value).lower(): - return enum_val - return value - - def _apply_schema_validation( + def _validate_payload( self, - contexts: list[dict[str, Any]], + *, + payload: dict[str, Any], + sample_label: str | None, + row_number: int | None, + available_props: set[str], folder: str, file_errors: list[dict[str, Any]], ) -> None: + if not payload: + return + validation_errors = list(self.validator.iter_errors(payload)) + validation_errors = relecov_tools.assets.schema_utils.custom_validators.validate_with_exceptions( + self.schema, payload, validation_errors + ) + if not validation_errors: + return + schema_props = self.schema_properties - seen_errors: set[tuple[str | None, int | None, str]] = set() - for idx, context in enumerate(contexts): - payload = context.get("payload") or {} - if not payload: - continue - available_props = set(context.get("available_properties") or []) - validation_errors = list(self.validator.iter_errors(payload)) - validation_errors = relecov_tools.assets.schema_utils.custom_validators.validate_with_exceptions( - self.schema, payload, validation_errors - ) - if not validation_errors: - continue - sample_label = context.get("sample") - schema_sample = ( - payload.get(self.sample_id_property) - if self.sample_id_property - else None - ) - if not sample_label and schema_sample: - sample_label = schema_sample - row_number = context.get("row") - display_label = ( - sample_label - if sample_label - else (f"row {row_number}" if row_number is not None else None) - ) - for error in validation_errors: - if error.cause: - error.message = str(error.cause) - if error.validator == "required": - try: - missing_field = list(error.message.split("'"))[1] - except Exception: - missing_field = None - if missing_field and missing_field not in available_props: - continue - error_text = self._format_validation_error(error, schema_props) - error_key = (display_label, row_number, error_text) - if error_key in seen_errors: + schema_sample = ( + payload.get(self.sample_id_property) if self.sample_id_property else None + ) + display_label = sample_label or schema_sample + if not display_label and row_number is not None: + display_label = f"row {row_number}" + + for error in validation_errors: + if error.cause: + error.message = str(error.cause) + if error.validator == "required": + try: + missing_field = list(error.message.split("'"))[1] + except Exception: + missing_field = None + if missing_field and missing_field not in available_props: continue - seen_errors.add(error_key) - self._record_error( - folder, - file_errors, - error_text, - sample=display_label, - row=row_number, - ) + error_text = self._format_validation_error(error, schema_props) + self._record_error( + folder, + file_errors, + error_text, + sample=display_label, + row=row_number, + ) def _format_validation_error(self, error, schema_props: dict[str, Any]) -> str: def get_property_label(prop_key: str) -> str: @@ -681,51 +721,6 @@ def get_property_label(prop_key: str) -> str: message = error.message.replace(str(error_field), field_label) return f"Error in column {field_label}: {message}" - def _canonical_label(self, label: Any) -> str: - if label is None: - return "" - label_str = str(label).strip() - return self.heading_aliases.get(label_str, label_str) - - def _normalise_sample_id( - self, row: dict[str, Any], sample_value_raw: Any - ) -> str | None: - candidates = [sample_value_raw] - fallback_keys = [ - "Sample ID given by the submitting laboratory", - "Sample ID given by originating laboratory", - "Sample ID given in the microbiology lab", - "Sample ID given if multiple rna-extraction or passages", - "Sequence file R1", - "Sequence file R1 fastq", - ] - for key in fallback_keys: - candidates.append(row.get(key)) - for value in candidates: - value_clean = self._clean_cell(value) - if self._value_is_present(value_clean): - return str(value_clean).strip() - return None - - def _clean_cell(self, value: Any) -> Any: - if isinstance(value, float) and value != value: - return None - if isinstance(value, str): - stripped = value.strip() - return stripped - return value - - def _value_is_present(self, value: Any) -> bool: - if value is None: - return False - if isinstance(value, str): - stripped = value.strip() - if not stripped: - return False - if "not provided" in stripped.lower(): - return False - return True - def _record_error( self, folder: str, From b74665758ad88a6b26a37abff1e14bdaf03c6445 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:31:59 +0100 Subject: [PATCH 5/8] Add no readable sheet error --- relecov_tools/metadata_precheck.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/relecov_tools/metadata_precheck.py b/relecov_tools/metadata_precheck.py index 8cc48cd1..a4013181 100644 --- a/relecov_tools/metadata_precheck.py +++ b/relecov_tools/metadata_precheck.py @@ -27,10 +27,6 @@ ) -class MetadataPrecheckError(RuntimeError): - """Custom error raised when a metadata file cannot be processed.""" - - class SchemaMapper: """Handle Excel header normalisation and schema-aware casting.""" @@ -461,7 +457,7 @@ def _validate_remote_metadata( try: sheet_data = self._read_metadata_sheet(local_target) - except MetadataPrecheckError as exc: + except RuntimeError as exc: message = f"Unable to read metadata sheet {remote_file}: {exc}" self._record_error(folder, file_errors, message) self.log.error(message) @@ -629,8 +625,8 @@ def _read_metadata_sheet(self, local_path: str) -> dict[str, Any]: except Exception as exc: # pragma: no cover - passthrough for runtime errors.append(f"{option['sheet']}: {exc}") if errors: - raise MetadataPrecheckError("; ".join(errors)) - raise MetadataPrecheckError("No readable sheet found in metadata Excel") + raise RuntimeError("; ".join(errors)) + raise RuntimeError("No readable sheet found in metadata Excel") def _validate_payload( self, From 22d93a0a8ce2826ba22beaf846bf8228c906c577 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:37:24 +0100 Subject: [PATCH 6/8] Enrich code with better descriptions --- relecov_tools/metadata_precheck.py | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/relecov_tools/metadata_precheck.py b/relecov_tools/metadata_precheck.py index a4013181..ab7b1621 100644 --- a/relecov_tools/metadata_precheck.py +++ b/relecov_tools/metadata_precheck.py @@ -28,7 +28,7 @@ class SchemaMapper: - """Handle Excel header normalisation and schema-aware casting.""" + """Keep all header normalisation + casting logic encapsulated for reuse and readability.""" SAMPLE_FALLBACKS = [ "Sample ID given by the submitting laboratory", @@ -274,6 +274,7 @@ def __init__( self.schema, format_checker=self._date_checker ) + # Mapper centralises Excel→schema conversions so the flow below stays compact. self.mapper = SchemaMapper( self.schema_properties, self.label_to_prop, @@ -312,6 +313,7 @@ def __init__( self.set_batch_id(datetime.utcnow().strftime("%Y%m%d%H%M%S")) def _build_sheet_options(self) -> list[dict[str, str]]: + """Collect sheet/header settings (primary + alternative) from config.""" opts = [] primary = { "sheet": self.metadata_processing.get("excel_sheet"), @@ -331,6 +333,7 @@ def _build_sheet_options(self) -> list[dict[str, str]]: def _parse_target_folders( self, target_folders: Iterable[str] | str | None ) -> tuple[list[str] | None, bool]: + """Normalise CLI/extra_config target folders and detect interactive mode.""" if target_folders is None: return None, False if isinstance(target_folders, str): @@ -346,6 +349,7 @@ def _parse_target_folders( return clean_list or None, False def execute_process(self) -> None: + """Connect to SFTP, validate every metadata Excel, and emit reports.""" if not self.sftp_client.open_connection(): msg = "Unable to establish sftp connection" self.log.error(msg) @@ -368,6 +372,7 @@ def execute_process(self) -> None: self._print_summary() def _discover_metadata_targets(self) -> dict[str, list[str]]: + """Walk the remote tree and collect folders that contain metadata files.""" directory_list = self.sftp_client.list_remote_folders(".", recursive=True) clean_dirs = sorted( {d.replace("./", "", 1) for d in directory_list if d and d != "."} @@ -384,6 +389,7 @@ def _discover_metadata_targets(self) -> dict[str, list[str]]: or part.lower().endswith("_invalid_samples") for part in path_parts ): + # Skip already-analysed folders to avoid generating duplicate reports. self.log.info( "Skipping %s because it points to an *_invalid_samples folder", directory, @@ -405,6 +411,7 @@ def _discover_metadata_targets(self) -> dict[str, list[str]]: return metadata_targets def _select_target_directories(self, clean_dirs: list[str]) -> list[str]: + """Return the final list of folders to scan (after filters/prompts).""" if self.prompt_for_targets: choices = sorted(clean_dirs) if not choices: @@ -422,6 +429,7 @@ def _select_target_directories(self, clean_dirs: list[str]) -> list[str]: return [folder for folder in self.target_folders if folder in clean_dirs] def _process_folder(self, folder: str, meta_files: list[str]) -> None: + """Run validation for each metadata file inside a remote folder.""" stderr.print(f"[blue]Processing folder {folder}") self.log.info("Processing folder %s", folder) self.logsum.feed_key(key=folder) @@ -434,6 +442,7 @@ def _process_folder(self, folder: str, meta_files: list[str]) -> None: def _validate_remote_metadata( self, folder: str, remote_file: str ) -> dict[str, Any]: + """Download, parse, and validate a single metadata Excel.""" file_errors: list[dict[str, Any]] = [] file_warnings: list[dict[str, Any]] = [] sample_count = 0 @@ -605,6 +614,7 @@ def _validate_remote_metadata( } def _read_metadata_sheet(self, local_path: str) -> dict[str, Any]: + """Load the Excel using configured sheet/header flag (primary or fallback).""" errors: list[str] = [] for option in self.sheet_options: try: @@ -638,6 +648,7 @@ def _validate_payload( folder: str, file_errors: list[dict[str, Any]], ) -> None: + """Run JSON-schema validation for one row and log any issues.""" if not payload: return validation_errors = list(self.validator.iter_errors(payload)) @@ -675,6 +686,8 @@ def _validate_payload( ) def _format_validation_error(self, error, schema_props: dict[str, Any]) -> str: + """Translate jsonschema errors into human-friendly messages.""" + def get_property_label(prop_key: str) -> str: prop_def = schema_props.get(prop_key, {}) return prop_def.get("label", prop_key) @@ -725,6 +738,7 @@ def _record_error( sample: str | None = None, row: int | None = None, ) -> None: + """Store an error both in log summary and in the per-file error list.""" if sample: self.logsum.add_error(entry=message, key=folder, sample=sample) else: @@ -739,6 +753,7 @@ def _record_warning( sample: str | None = None, row: int | None = None, ) -> None: + """Store a warning both in log summary and in the per-file warning list.""" if sample: self.logsum.add_warning(entry=message, key=folder, sample=sample) else: @@ -748,6 +763,7 @@ def _record_warning( def _update_lab_summary( self, lab_code: str, folder: str, file_summary: dict[str, Any] ) -> None: + """Aggregate per-lab stats so later reports/table can be generated.""" lab_entry = self.lab_summary[lab_code] lab_entry["total_samples"] += file_summary.get("samples", 0) lab_entry["invalid_sample_count"] += len( @@ -774,6 +790,7 @@ def _update_lab_summary( ) def _export_report(self) -> None: + """Persist the merged JSON report with lab/folder breakdown.""" report = { "generated_at": self.generated_at, "total_labs": len(self.lab_summary), @@ -793,6 +810,7 @@ def _export_report(self) -> None: stderr.print(f"[red]Could not write metadata precheck report: {exc}") def _print_summary(self) -> None: + """Render an at-a-glance Rich table with lab counts and top issues.""" if not self.lab_summary: return table = Table(title="Metadata precheck summary", show_lines=False) @@ -804,6 +822,7 @@ def _print_summary(self) -> None: table.add_column("Top issues", justify="left") def _collect_lab_messages(files: list[dict[str, Any]], limit: int = 3): + """Return up to `limit` distinct error messages for a lab.""" messages = [] for fdata in files: for inv in fdata.get("invalid_samples", []): From caac43c2b51de0627a13f271660ae030a7a70d4a Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:41:26 +0100 Subject: [PATCH 7/8] Update CHANGELOG --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b78aa4bb..b46a5c29 100755 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Add input_directories to upload-results [#808](https://github.com/BU-ISCIII/relecov-tools/pull/808) - update ERROR handling in mail [#808](https://github.com/BU-ISCIII/relecov-tools/pull/808) - Add labs to laboratory_adress.json [#813](https://github.com/BU-ISCIII/relecov-tools/pull/813) +- Add metadata-precheck module to validate SFTP metadata excels upfront [#816](https://github.com/BU-ISCIII/relecov-tools/pull/816) #### Fixes From b10c7936c31ad322c1ec59dee1602617c357f864 Mon Sep 17 00:00:00 2001 From: Alejandro Date: Wed, 12 Nov 2025 11:48:25 +0100 Subject: [PATCH 8/8] Fix flake8 --- relecov_tools/metadata_precheck.py | 1 - 1 file changed, 1 deletion(-) diff --git a/relecov_tools/metadata_precheck.py b/relecov_tools/metadata_precheck.py index ab7b1621..542b62e6 100644 --- a/relecov_tools/metadata_precheck.py +++ b/relecov_tools/metadata_precheck.py @@ -533,7 +533,6 @@ def _validate_remote_metadata( for idx, row in enumerate(rows): row_number = header_row_index + 1 + idx canonical_row = self.mapper.canonicalize_row(row, header_flag) - sample_value_raw = canonical_row.get(sample_column) sample_value = self.mapper.normalise_sample_id(canonical_row, sample_column) if not sample_value: