workflows: Migrate basic digitization scripts to Airflow

.gitignore

@@ -18,3 +18,9 @@ import_xml_files/
 .venv
 
 s3_pdf_issues.json
+
+# Airflow
+logs/
+airflow.cfg
+airflow.db
+webserver_config.py

Dockerfile

@@ -1,21 +1,28 @@
-FROM python:3.12-slim
+FROM registry.cern.ch/cern-sis/base-images/apache/airflow:3.1.7-python3.11
 
-WORKDIR /code
+ENV PYTHONBUFFERED=0
+ENV AIRFLOW__LOGGING__LOGGING_LEVEL=INFO
 
-ENV PATH="/root/.poetry/bin:${PATH}"
+ENV PYTHONPATH="/opt/airflow/refactory:${PYTHONPATH}"
 
+USER root
 RUN apt-get update \
- && apt-get install -y kstart curl \
+ && apt-get install -y kstart \
  && apt-get clean \
- && rm -rf /var/lib/apt/lists/* \
- && curl -sSL https://install.python-poetry.org/ \
-  | python
+ && rm -rf /var/lib/apt/lists/*
 
-ENV PATH="/root/.local/bin:$PATH"
 
-RUN poetry --version
+USER airflow
 
-COPY poetry.lock pyproject.toml ./
-COPY digitization ./digitization
+ARG AIRFLOW_VERSION=3.1.7
+ARG PYTHON_VERSION=3.11
 
-RUN poetry install
+COPY airflow/requirements.txt ./requirements.txt
+RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" -r requirements.txt \
+    --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
+
+
+COPY airflow/dags ./dags
+
+
+COPY refactory ./refactory

airflow/Dockerfile

@@ -0,0 +1,25 @@
+FROM registry.cern.ch/cern-sis/base-images/apache/airflow:3.1.7-python3.11
+
+ENV PYTHONUNBUFFERED=0
+ENV AIRFLOW__LOGGING__LOGGING_LEVEL=INFO
+
+USER root
+
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+    gcc \
+    libffi-dev \
+    libssl-dev \
+    kstart \
+    curl \
+    && apt-get autoremove -yqq --purge \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+USER airflow
+ARG AIRFLOW_VERSION=3.1.7
+ARG PYTHON_VERSION=3.11
+
+COPY airflow/requirements.txt /requirements.txt
+RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" -r /requirements.txt \
+    --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

airflow/README.md

@@ -0,0 +1,108 @@
+# Digitization – Airflow
+
+Airflow DAGs for the digitization workflows:
+
+| DAG | What it does |
+| --- | --- |
+| `validate_files_integrity` | Checks that the PDFs of the selected boxes in S3 can be opened (not corrupted). |
+| `match_and_export` | Matches the records in the box Excel files (`BOITE_O0<n>.xlsx`) with the files in S3, then builds the MARCXML files. |
+
+## Setup
+
+Run these commands from the `airflow/` folder:
+
+```bash
+docker compose up --build
+```
+
+- UI: http://localhost:8080 (user `admin`, password `admin`)
+- New DAGs start **paused**. Unpause a DAG before you trigger it.
+
+### Connections
+
+The DAGs use two Airflow connections. Create them in **Admin → Connections**, or put them in `scripts/connections.json` and the bootstrap service imports them on startup.
+
+| Conn ID | Type | Used for |
+| --- | --- | --- |
+| `aws_s3_cern` | Amazon Web Services (access key, secret key, `endpoint_url` in extra) | Reading and writing files in S3 |
+| `cernbox_conn` | HTTP (login = CERN account, password) | Uploading XMLs and logs to CERNBox |
+
+Public CERNBox links (used to read the box Excel files) need no credentials.
+
+---
+
+## `validate_files_integrity`
+
+Expected S3 layout: `<bucket>/<base_path>/BOITE_O0<n>.../*.pdf`
+
+| Param | Default | Description |
+| --- | --- | --- |
+| `data_source` | *(required)* | The boxes to check. See the formats below. |
+| `bucket` | `digitization-dev` | The S3 bucket. |
+| `base_path` | `cern-archives/raw/PDF/` | The S3 prefix that contains the `BOITE_*` folders. |
+| `upload_reports` | `false` | If `true`, uploads `validation_report.json` and `s3_pdf_issues.log` to `<base_path>`. |
+
+`data_source` formats:
+
+| Format | Example | Boxes |
+| --- | --- | --- |
+| Single number | `42` | 42 |
+| Range (inclusive) | `1..10` | 1 to 10 |
+| List | `[3, 7, 12]` | 3, 7, 12 |
+| CERNBox URL or public link hash | `https://cernbox.cern.ch/s/AbCdEf123456` or `AbCdEf123456` | The box numbers in the names of the `.xlsx` files in that folder |
+
+Examples:
+
+```json
+{ "data_source": "1..50" }
+{ "data_source": "[101, 102]", "upload_reports": true }
+{ "data_source": "AbCdEf123456", "base_path": "cern-archives/raw/PDF/" }
+```
+
+Output: the task logs show a summary with the number of valid PDFs, corrupted PDFs, and missing boxes. The full report is available in the XCom of `validate_s3_pdfs`.
+
+---
+
+## `match_and_export`
+
+Expected S3 layout: `<bucket>/<base_path>/<file_type>/BOITE_O0<n>.../...`
+
+| Param | Default | Description |
+| --- | --- | --- |
+| `data_source` | *(required)* | Where to find the box `.xlsx` files: a CERNBox URL, or a folder path inside the container. |
+| `base_paths` | `cern-archives/raw/` | The S3 root path. To search more than one root, separate them with commas. |
+| `file_types` | `PDF,PDF_LATEX` | The subfolders to match under each base path, separated by commas (for example `PDF`, `PDF_LATEX`, `TIFF`). |
+| `bucket` | `digitization-dev` | The S3 bucket. |
+| `execution_mode` | `Full Export (CERNBox)` | See the table below. |
+| `cernbox_path` | `xml_exports` | The destination folder in your CERNBox (relative to your home folder). |
+
+`execution_mode` options:
+
+| Mode | Matching | Mismatch log upload | XML generation | XML upload |
+| --- | :-: | :-: | :-: | :-: |
+| `Dry Run` | ✅ | ❌ | ❌ | ❌ |
+| `Generate XMLs (Temp Folder Only)` | ✅ | ✅ | ✅ (temp folder in the worker) | ❌ |
+| `Full Export (CERNBox)` | ✅ | ✅ | ✅ | ✅ |
+
+`data_source` examples:
+
+- Public share: `https://cernbox.cern.ch/s/AbCdEf123456`
+- EOS path: `https://cernbox.cern.ch/files/spaces/eos/user/j/jdoe/boites` (requires `cernbox_conn`)
+- Local folder: `/opt/airflow/tests/data/boites` (the path must be mounted in the container; see the `volumes` section in `docker-compose.yaml`)
+
+Examples:
+
+```json
+{ "data_source": "https://cernbox.cern.ch/s/AbCdEf123456", "execution_mode": "Dry Run" }
+
+{
+  "data_source": "https://cernbox.cern.ch/s/AbCdEf123456",
+  "base_paths": "cern-archives/raw/, cern-archives/batch2/",
+  "file_types": "PDF,PDF_LATEX,TIFF",
+  "cernbox_path": "xml_exports/2026-10"
+}
+```
+
+Output in CERNBox (`<cernbox_path>/`):
+- One XML file per box, plus `Boites_combined.xml` (Full Export only)
+- `all_boites_mismatches_<run_id>.json`: the records that were not found in S3, and the S3 files that no record matched

airflow/dags/check_files/utils.py

@@ -0,0 +1,35 @@
+import os
+from pypdf import PdfReader
+from pypdf.errors import PdfReadError
+
+
+def validate_pdf(file_path: str) -> bool:
+    """Checks if a local PDF is structurally valid and readable."""
+    try:
+        file_size = os.path.getsize(file_path)
+        if file_size < 100:
+            return False
+
+        with open(file_path, "rb") as f:
+            header = f.read(8)
+            f.seek(-min(1024, file_size), 2)
+            trailer = f.read()
+
+        if not header.startswith(b"%PDF-"):
+            return False
+        if b"%%EOF" not in trailer:
+            return False
+
+        reader = PdfReader(file_path)
+        if len(reader.pages) == 0:
+            return False
+
+        _ = reader.pages[0]
+
+        return True
+
+    except OSError as e:
+        raise RuntimeError(f"System error when accessing file {file_path}: {e}") from e
+
+    except (PdfReadError, Exception):
+        return False

airflow/dags/check_files/validate_files_integrity.py

@@ -0,0 +1,111 @@
+import json
+import tempfile
+from datetime import datetime
+from airflow.sdk import dag, task, Param
+
+from common.storage_connection import S3Provider
+from common.utils import parse_inventory
+from check_files.validator import resolve_target_boxes, validate_s3_files
+
+
+@dag(
+    dag_id="validate_files_integrity",
+    start_date=datetime(2024, 1, 1),
+    schedule=None,
+    catchup=False,
+    tags=["digitization", "validation"],
+    params={
+        "data_source": Param(
+            type="string",
+            minLength=1,
+            description="Inventory source (e.g., 1..1000, [1,2], CERNBox URL or hash)",
+        ),
+        "bucket": Param(
+            "digitization-dev", type="string", description="Target S3 Bucket name"
+        ),
+        "base_path": Param(
+            "cern-archives/raw/PDF/",
+            type="string",
+            description="Base S3 path for files",
+        ),
+        "upload_reports": Param(
+            False,
+            type="boolean",
+            description="Toggle to upload validation reports back to storage",
+        ),
+    },
+)
+def validation_dag():
+
+    @task(task_id="prepare_target_boxes")
+    def prepare_targets_boxes(**context):
+        """Parses inputs and figures out exactly which boxes to check."""
+        raw_data_source = context["params"]["data_source"]
+
+        inventory_input = parse_inventory(raw_data_source)
+
+        targets = resolve_target_boxes(inventory_input)
+        if not targets:
+            raise ValueError(f"No boxes found for data_source: {raw_data_source}")
+
+        print(f"{len(targets)} boxes for validation.")
+        return list(targets)
+
+    @task(task_id="validate_s3_pdfs")
+    def perform_validation(target_boxes: list, **context):
+        """Iterates through S3 and checks if PDFs are corrupted."""
+        bucket = context["params"]["bucket"]
+        # list_folders uses Delimiter="/", so the prefix must end with "/" to list box folders.
+        base_path = context["params"]["base_path"].rstrip("/") + "/"
+
+        provider = S3Provider(bucket=bucket)
+
+        print(f"Starting validation for {len(target_boxes)} boxes in {base_path}")
+        result_report = validate_s3_files(provider, base_path, set(target_boxes))
+
+        return result_report
+
+    @task(task_id="upload_validation_reports")
+    def upload_reports(report: dict, **context):
+        """Generates the JSON/TXT reports and uploads them securely to S3."""
+        upload_toggle = context["params"]["upload_reports"]
+
+        if not upload_toggle:
+            print("Report upload skipped (Toggle is OFF).")
+            return
+
+        bucket = context["params"]["bucket"]
+        base_path = context["params"]["base_path"]
+        provider = S3Provider(bucket=bucket)
+
+        text_log = f"Validation report for boxes {report['metadata']['target_boxes']}\n"
+        text_log += f"✅ Valid: {report['statistics']['valid_files_count']}\n"
+        text_log += f"❌ Corrupted: {report['statistics']['corrupted_files_count']}\n\n"
+
+        for vf in report["output"]["valid_files"]:
+            text_log += f"✅ Valid PDF: {vf}\n"
+        for cf in report["output"]["corrupted_files"]:
+            text_log += f"❌ Corrupted PDF: {cf}\n"
+
+        with (
+            tempfile.NamedTemporaryFile(mode="w", delete=False) as json_tmp,
+            tempfile.NamedTemporaryFile(mode="w", delete=False) as txt_tmp,
+        ):
+            json.dump(report, json_tmp, indent=4)
+            txt_tmp.write(text_log)
+            json_tmp_path, txt_tmp_path = json_tmp.name, txt_tmp.name
+
+        remote_json_path = f"{base_path.rstrip('/')}/validation_report.json"
+        remote_log_path = f"{base_path.rstrip('/')}/s3_pdf_issues.log"
+
+        provider.upload_file(json_tmp_path, remote_json_path)
+        provider.upload_file(txt_tmp_path, remote_log_path)
+
+        print(f"Uploaded securely to {remote_json_path} and {remote_log_path}")
+
+    targets = prepare_targets_boxes()
+    validation_results = perform_validation(targets)
+    upload_reports(validation_results)
+
+
+validation_dag_instance = validation_dag()