feat(cli): chemistry LIMS ingestion with Google Drive sync (BDMS-1034)

.env.example

@@ -55,6 +55,13 @@ TRANSFER_NMW_MIRROR=True             # load the NMW_* 1:1 staging mirror
 GCS_BUCKET_NAME=
 GOOGLE_APPLICATION_CREDENTIALS=/path/to/gcs_credentials.json
 
+# chemistry ingestion: Google Drive folder an engineer ingests LIMS .xlsx
+# workbooks from, on demand (no polling/scheduling; service account / ADC must
+# have read access to this folder). The ingested-file manifest is stored in
+# GCS_BUCKET_NAME at CHEMISTRY_INGEST_MANIFEST_PATH.
+CHEMISTRY_DRIVE_FOLDER_ID=
+CHEMISTRY_INGEST_MANIFEST_PATH=chemistry-ingest/manifest.json
+
 # set to development for lexicon and parameter to be populated and enable the enums to work
 MODE=development
 

.github/workflows/CD_production.yml

@@ -99,16 +99,6 @@ jobs:
         run: |
           uv run --no-dev alembic upgrade head
 
-      - name: Refresh materialized views on production database
-        env:
-          DB_DRIVER: "cloudsql"
-          CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
-          CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
-          CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
-          CLOUD_SQL_IAM_AUTH: true
-        run: |
-          uv run --no-dev python -m cli.cli refresh-materialized-views
-
       - name: Ensure envsubst is available
         run: |
           if ! command -v envsubst >/dev/null 2>&1; then

.github/workflows/CD_staging.yml

@@ -59,16 +59,6 @@ jobs:
         run: |
           uv run --no-dev alembic upgrade head
 
-      - name: Refresh materialized views on staging database
-        env:
-          DB_DRIVER: "cloudsql"
-          CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
-          CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
-          CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
-          CLOUD_SQL_IAM_AUTH: true
-        run: |
-          uv run --no-dev python -m cli.cli refresh-materialized-views
-
       - name: Ensure envsubst is available
         run: |
           if ! command -v envsubst >/dev/null 2>&1; then

.github/workflows/CD_testing.yml

@@ -59,16 +59,6 @@ jobs:
         run: |
           uv run --no-dev alembic upgrade head
 
-      - name: Refresh materialized views on staging database
-        env:
-          DB_DRIVER: "cloudsql"
-          CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
-          CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
-          CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
-          CLOUD_SQL_IAM_AUTH: true
-        run: |
-          uv run --no-dev python -m cli.cli refresh-materialized-views
-
       - name: Ensure envsubst is available
         run: |
           if ! command -v envsubst >/dev/null 2>&1; then

.github/workflows/jira_codex_pr.yml

@@ -71,7 +71,7 @@ jobs:
       - name: Sync dependencies (pyproject/uv.lock)
         run: |
           set -euo pipefail
-          uv sync --all-extras --dev
+          uv sync --all-extras --dev --group cli
 
       - name: Verify tooling exists
         run: |

.github/workflows/tests.yml

@@ -82,7 +82,7 @@ jobs:
           key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('uv.lock') }}
 
       - name: Install the project
-        run: uv sync --locked --all-extras --dev
+        run: uv sync --locked --all-extras --dev --group cli
 
       - name: Show Alembic heads
         run: uv run alembic heads
@@ -174,7 +174,7 @@ jobs:
           key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('uv.lock') }}
 
       - name: Install the project
-        run: uv sync --locked --all-extras --dev
+        run: uv sync --locked --all-extras --dev --group cli
 
       - name: Show Alembic heads
         run: uv run alembic heads

cli/cli.py

@@ -32,8 +32,10 @@
 
 cli = typer.Typer(help="Command line interface for managing the application.")
 water_levels = typer.Typer(help="Water-level utilities")
+water_chemistry = typer.Typer(help="Water-chemistry utilities")
 data_migrations = typer.Typer(help="Data migration utilities")
 cli.add_typer(water_levels, name="water-levels")
+cli.add_typer(water_chemistry, name="water-chemistry")
 cli.add_typer(data_migrations, name="data-migrations")
 
 
@@ -959,6 +961,201 @@ def water_levels_bulk_upload(
     raise typer.Exit(result.exit_code)
 
 
+@water_chemistry.command("bulk-upload")
+def water_chemistry_bulk_upload(
+    file_path: str = typer.Option(
+        ...,
+        "--file",
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        help="Path to LIMS .xlsx workbook containing chemistry results.",
+    ),
+    theme: ThemeMode = typer.Option(
+        ThemeMode.auto, "--theme", help="Color theme: auto, light, dark."
+    ),
+):
+    """
+    parse a LIMS chemistry workbook and load it into the NMA Major/Minor
+    chemistry tables. Each distinct lab sample (WCLab_ID) is appended as a new
+    lettered sample point; a lab sample already recorded for the well is
+    skipped. A row that fails to map or references an unknown well aborts the
+    whole file.
+    """
+    from cli.service_adapter import chemistry_lims_xlsx
+
+    colors = _palette(theme)
+    result = chemistry_lims_xlsx(file_path)
+
+    payload = result.payload if isinstance(result.payload, dict) else {}
+    summary = payload.get("summary", {})
+    validation_errors = payload.get("validation_errors", [])
+    created_samples = payload.get("created_samples", [])
+    skipped_duplicates = payload.get("skipped_duplicates", [])
+
+    if result.exit_code == 0:
+        typer.secho("[WATER CHEMISTRY IMPORT] SUCCESS", fg=colors["ok"], bold=True)
+    else:
+        typer.secho(
+            "[WATER CHEMISTRY IMPORT] COMPLETED WITH ISSUES",
+            fg=colors["issue"],
+            bold=True,
+        )
+    typer.secho("=" * 72, fg=colors["accent"])
+
+    if summary:
+        processed = summary.get("total_rows_processed", 0)
+        imported = summary.get("total_rows_imported", 0)
+        rows_with_issues = summary.get("validation_errors_or_warnings", 0)
+        typer.secho("SUMMARY", fg=colors["accent"], bold=True)
+        label_width = 16
+        value_width = 8
+        typer.secho("  " + "-" * (label_width + 3 + value_width), fg=colors["muted"])
+        typer.secho(
+            f"  {'processed':<{label_width}} | {processed:>{value_width}}",
+            fg=colors["accent"],
+        )
+        typer.secho(
+            f"  {'imported':<{label_width}} | {imported:>{value_width}}",
+            fg=colors["ok"],
+        )
+        issue_color = colors["issue"] if rows_with_issues else colors["ok"]
+        typer.secho(
+            f"  {'rows_with_issues':<{label_width}} | {rows_with_issues:>{value_width}}",
+            fg=issue_color,
+        )
+        typer.echo()
+
+    if created_samples:
+        typer.secho("CREATED SAMPLES", fg=colors["ok"], bold=True)
+        for sample in created_samples:
+            typer.secho(
+                f"  - {sample['sample_point_id']} "
+                f"(WCLab_ID {sample.get('wclab_id')}): {sample.get('rows', 0)} row(s)",
+                fg=colors["ok"],
+            )
+        typer.echo()
+
+    if skipped_duplicates:
+        typer.secho("SKIPPED (already ingested)", fg=colors["muted"], bold=True)
+        for dupe in skipped_duplicates:
+            typer.secho(
+                f"  - {dupe['pointid']} (WCLab_ID {dupe.get('wclab_id')})",
+                fg=colors["field"],
+            )
+        typer.echo()
+
+    if validation_errors:
+        typer.secho("VALIDATION", fg=colors["accent"], bold=True)
+        typer.secho(
+            f"Validation errors: {len(validation_errors)}",
+            fg=colors["issue"],
+            bold=True,
+        )
+        for entry in validation_errors[:25]:
+            typer.secho(f"  - {entry}", fg=colors["issue"])
+        if len(validation_errors) > 25:
+            typer.secho(
+                f"... and {len(validation_errors) - 25} more validation errors",
+                fg=colors["issue"],
+            )
+
+    typer.secho("=" * 72, fg=colors["accent"])
+    raise typer.Exit(result.exit_code)
+
+
+@water_chemistry.command("sync-drive")
+def water_chemistry_sync_drive(
+    folder_id: str = typer.Option(
+        None,
+        "--folder-id",
+        help="Google Drive folder id to scan. Defaults to $CHEMISTRY_DRIVE_FOLDER_ID.",
+    ),
+    dry_run: bool = typer.Option(
+        False,
+        "--dry-run",
+        help="List new files without downloading, ingesting, or updating the manifest.",
+    ),
+    theme: ThemeMode = typer.Option(
+        ThemeMode.auto, "--theme", help="Color theme: auto, light, dark."
+    ),
+):
+    """
+    ingest LIMS chemistry workbooks from a Google Drive folder. Run on demand
+    by an engineer -- this does no polling or scheduling. New or changed files
+    are ingested; a manifest of ingested files is kept in GCS so
+    already-processed files are skipped.
+    """
+    from services.chemistry_drive import ChemistryDriveConfigError, sync_and_ingest
+
+    colors = _palette(theme)
+    try:
+        result = sync_and_ingest(folder_id=folder_id, dry_run=dry_run)
+    except ChemistryDriveConfigError as exc:
+        typer.secho(str(exc), fg=colors["issue"], bold=True, err=True)
+        raise typer.Exit(1) from exc
+
+    summary = result.to_payload()["summary"]
+    header = (
+        "[CHEMISTRY DRIVE SYNC] DRY RUN" if result.dry_run else "[CHEMISTRY DRIVE SYNC]"
+    )
+    header_color = colors["ok"] if result.exit_code == 0 else colors["issue"]
+    typer.secho(header, fg=header_color, bold=True)
+    typer.secho("=" * 72, fg=colors["accent"])
+    typer.secho(f"Folder: {result.folder_id}", fg=colors["accent"])
+    typer.echo()
+
+    typer.secho("SUMMARY", fg=colors["accent"], bold=True)
+    for label, value, color in (
+        ("files_seen", summary["files_seen"], colors["accent"]),
+        ("new_files", summary["new_files"], colors["accent"]),
+        ("ingested", summary["ingested"], colors["ok"]),
+        ("skipped", summary["skipped"], colors["muted"]),
+        (
+            "failed",
+            summary["failed"],
+            colors["issue"] if summary["failed"] else colors["ok"],
+        ),
+    ):
+        typer.secho(f"  {label:<12} | {value:>6}", fg=color)
+    typer.echo()
+
+    if result.dry_run and result.new_files:
+        typer.secho("NEW FILES (not ingested)", fg=colors["accent"], bold=True)
+        for name in result.new_files:
+            typer.secho(f"  - {name}", fg=colors["field"])
+        typer.echo()
+
+    if result.ingested:
+        typer.secho("INGESTED", fg=colors["ok"], bold=True)
+        for record in result.ingested:
+            typer.secho(
+                f"  - {record['name']}: {record.get('rows_imported', 0)} row(s)",
+                fg=colors["ok"],
+            )
+        typer.echo()
+
+    if result.failed:
+        typer.secho("FAILED", fg=colors["issue"], bold=True)
+        for record in result.failed:
+            detail = record.get("error")
+            if not detail:
+                payload = record.get("payload", {})
+                errors = payload.get("validation_errors") or []
+                if errors:
+                    detail = errors[0]
+                    if len(errors) > 1:
+                        detail += f" (+{len(errors) - 1} more)"
+                else:
+                    detail = "ingestion aborted"
+            typer.secho(f"  - {record['name']}: {detail}", fg=colors["issue"])
+        typer.echo()
+
+    typer.secho("=" * 72, fg=colors["accent"])
+    raise typer.Exit(result.exit_code)
+
+
 @data_migrations.command("list")
 def data_migrations_list(
     theme: ThemeMode = typer.Option(

cli/service_adapter.py

@@ -89,6 +89,18 @@ def water_levels_csv(source_file: Path | str, *, pretty_json: bool = False):
     return result
 
 
+def chemistry_lims_xlsx(source_file: Path | str):
+    from services.chemistry_lims import bulk_upload_chemistry
+
+    if isinstance(source_file, str):
+        source_file = Path(source_file)
+
+    result = bulk_upload_chemistry(source_file)
+    if result.stderr:
+        print(result.stderr, file=sys.stderr)
+    return result
+
+
 def associate_assets(source_directory: Path | str) -> list[str]:
     """
     given a directory