feat: datetime-fail-matching

pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py

@@ -40,6 +40,7 @@
 from pipeline import core, orchestrator
 from pipeline.helpers import cli, db, dpdash, utils
 from pipeline.helpers.config import config
+from pipeline.models import datetime_overrides
 from pipeline.models.files import File
 from pipeline.models.interview_files import InterviewFile
 from pipeline.models.interview_parts import InterviewParts
@@ -360,20 +361,34 @@ def fetch_interviews(
                 interview_datetime = datetime.combine(date_dt, time_dt)
                 actual_interview_datetime = datetime.combine(date_dt, actual_time_dt)
             except (ValueError, IndexError) as e:
-                logger.error(
-                    f"{subject_id}: Could not parse date and time from {base_name}. Skipping..."
+                override_datetime = datetime_overrides.get_override_datetime(
+                    config_file=config_file, identifier=str(interview_dir)
                 )
-                db.record_failure(
-                    config_file=config_file,
-                    stage=MODULE_NAME,
-                    error_code="datetime_parse",
-                    identifier=str(interview_dir),
-                    error=e,
-                    identifier_type="file_path",
-                    study_id=study_id,
-                    subject_id=subject_id,
+                if override_datetime is None:
+                    logger.error(
+                        f"{subject_id}: Could not parse date and time from {base_name}. Skipping..."
+                    )
+                    db.record_failure(
+                        config_file=config_file,
+                        stage=MODULE_NAME,
+                        error_code="datetime_parse",
+                        identifier=str(interview_dir),
+                        error=e,
+                        identifier_type="file_path",
+                        study_id=study_id,
+                        subject_id=subject_id,
+                    )
+                    continue
+
+                logger.info(
+                    f"{subject_id}: Using staff-confirmed datetime override for {base_name}"
+                )
+                actual_interview_datetime = override_datetime
+                # Ignore time information, to get accurate day - mirrors the
+                # normal-parse path above.
+                interview_datetime = override_datetime.replace(
+                    hour=0, minute=0, second=0, microsecond=0
                 )
-                continue
 
             interview_name = dpdash.get_dpdash_name(
                 study=study_id,
@@ -412,20 +427,33 @@ def fetch_interviews(
                     hour=0, minute=0, second=0, microsecond=0
                 )
             except ValueError as e:
-                logger.error(
-                    f"Could not parse date and time from {wav_file}. Skipping..."
+                override_datetime = datetime_overrides.get_override_datetime(
+                    config_file=config_file, identifier=str(wav_file)
                 )
-                db.record_failure(
-                    config_file=config_file,
-                    stage=MODULE_NAME,
-                    error_code="datetime_parse",
-                    identifier=str(wav_file),
-                    error=e,
-                    identifier_type="file_path",
-                    study_id=study_id,
-                    subject_id=subject_id,
+                if override_datetime is None:
+                    logger.error(
+                        f"Could not parse date and time from {wav_file}. Skipping..."
+                    )
+                    db.record_failure(
+                        config_file=config_file,
+                        stage=MODULE_NAME,
+                        error_code="datetime_parse",
+                        identifier=str(wav_file),
+                        error=e,
+                        identifier_type="file_path",
+                        study_id=study_id,
+                        subject_id=subject_id,
+                    )
+                    continue
+
+                logger.info(
+                    f"{subject_id}: Using staff-confirmed datetime override for {wav_file}"
+                )
+                actual_interview_datetime = override_datetime
+                # truncate time, mirrors the normal-parse path above
+                interview_datetime = override_datetime.replace(
+                    hour=0, minute=0, second=0, microsecond=0
                 )
-                continue
 
             interview_name = dpdash.get_dpdash_name(
                 study=study_id,
@@ -604,6 +632,18 @@ def import_interviews(config_file: Path, study_id: str, progress: Progress) -> N
         failure_identifier_type="study",
     )
 
+    # Successfully-imported parts may have previously failed to date-parse
+    # (recorded in pipeline_failures, then matched to a runsheet entry via a
+    # staff-confirmed datetime_overrides row) - close the loop on both now
+    # that the import actually succeeded. No-ops for normally-parsed files:
+    # there's no matching pipeline_failures/datetime_overrides row to update.
+    for interview_part in interview_parts:
+        identifier = str(interview_part.interview_path)
+        db.resolve_failure(
+            config_file=config_file, stage=MODULE_NAME, identifier=identifier
+        )
+        datetime_overrides.mark_consumed(config_file=config_file, identifier=identifier)
+
 
 def mark_unique_interviews_as_primary(config_file: Path, study_id: str) -> None:
     """

pipeline/models/__init__.py

@@ -16,6 +16,7 @@
 from pipeline.models.key_store import KeyStore
 from pipeline.models.logs import Log
 from pipeline.models.pipeline_failures import PipelineFailure
+from pipeline.models.datetime_overrides import DatetimeOverride
 from pipeline.models.decrypted_files import DecryptedFile
 from pipeline.models.video_qqc import VideoQuickQc
 from pipeline.models.interview_roles import InterviewRole
@@ -90,13 +91,15 @@ def init_db(config_file: Path):
         KeyStore.drop_table_query(),
         Log.drop_table_query(),
         PipelineFailure.drop_table_query(),
+        DatetimeOverride.drop_table_query(),
         FfprobeMetadata.drop_table_query(),
     ]
 
     create_queries_l: List[Union[str, List[str]]] = [
         KeyStore.init_table_query(),
         Log.init_table_query(),
         PipelineFailure.init_table_query(),
+        DatetimeOverride.init_table_query(),
         Study.init_table_query(),
         Subject.init_table_query(),
         FormData.init_table_query(),

pipeline/models/datetime_overrides.py

@@ -0,0 +1,250 @@
+#!/usr/bin/env python
+"""
+DatetimeOverride Model
+
+Staff-confirmed event datetime for a raw interview file/directory whose name
+could not be date-parsed by the crawler (recorded as a 'datetime_parse'
+pipeline_failures row). Written by dpinterview-web's "Runsheet Match"
+remediation page after a human matches the malformed file to a runsheet
+entry; consumed by 2_import_interview_files.py as a fallback when the
+on-disk name still fails to parse, so a manually-matched file gets imported
+through the exact same downstream code path (categorization, hashing,
+Interview/InterviewParts/InterviewFile creation) as a normally-named one.
+"""
+
+import sys
+from pathlib import Path
+
+file = Path(__file__).resolve()
+parent = file.parent
+ROOT = None
+for parent in file.parents:
+    if parent.name == "dpinterview":
+        ROOT = parent
+sys.path.append(str(ROOT))
+
+# remove current directory from path
+try:
+    sys.path.remove(str(parent))
+except ValueError:
+    pass
+
+import argparse
+from datetime import datetime
+from typing import Optional
+
+from pipeline.helpers import cli, db, utils
+
+console = utils.get_console()
+
+# Kept alongside pipeline_failures in the same non-'public' schema - an
+# override only ever exists in reference to a pipeline_failures row.
+SCHEMA_NAME = "pipeline_ledger"
+TABLE_NAME = f"{SCHEMA_NAME}.datetime_overrides"
+
+
+class DatetimeOverride:
+    """
+    Represents a row in the 'pipeline_ledger.datetime_overrides' table.
+
+    Attributes:
+        identifier (str): The raw file/directory path that failed to
+            date-parse - matches pipeline_failures.pf_identifier for the
+            corresponding 'datetime_parse' failure.
+        override_datetime (datetime): The staff-confirmed actual event
+            datetime for this file.
+        study_id (Optional[str]): The study this override applies to, if known.
+        subject_id (Optional[str]): The subject this override applies to, if known.
+    """
+
+    def __init__(
+        self,
+        identifier: str,
+        override_datetime: datetime,
+        study_id: Optional[str] = None,
+        subject_id: Optional[str] = None,
+    ) -> None:
+        self.identifier = identifier
+        self.override_datetime = override_datetime
+        self.study_id = study_id
+        self.subject_id = subject_id
+
+    def __str__(self) -> str:
+        return f"DatetimeOverride({self.identifier}, {self.override_datetime})"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+    @staticmethod
+    def init_table_query() -> str:
+        """
+        Return the SQL to create the 'datetime_overrides' table (schema is
+        shared with, and already created by, pipeline_failures).
+        """
+        sql_query = f"""
+        CREATE SCHEMA IF NOT EXISTS {SCHEMA_NAME};
+
+        CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
+            do_id SERIAL PRIMARY KEY,
+            do_identifier TEXT NOT NULL UNIQUE,
+            do_study_id TEXT,
+            do_subject_id TEXT,
+            do_override_datetime TIMESTAMP NOT NULL,
+            do_created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            do_consumed_at TIMESTAMP
+        );
+
+        CREATE INDEX IF NOT EXISTS datetime_overrides_identifier_idx
+            ON {TABLE_NAME} (do_identifier);
+        """
+
+        return sql_query
+
+    @staticmethod
+    def drop_table_query() -> str:
+        """
+        Return the SQL query to drop the 'datetime_overrides' table. Leaves
+        the shared 'pipeline_ledger' schema in place.
+        """
+        sql_query = f"""
+        DROP TABLE IF EXISTS {TABLE_NAME};
+        """
+
+        return sql_query
+
+    def to_sql(self) -> str:
+        """
+        Return the SQL query to insert (or replace) this override. Re-linking
+        an identifier to a new datetime clears do_consumed_at - it no longer
+        gates whether the override applies (see get_override_datetime()), but
+        clearing it keeps the field meaningful as "first picked up under the
+        current value" rather than carrying a stale timestamp from a
+        since-corrected date.
+        """
+        identifier = db.santize_string(self.identifier)
+        override_datetime = self.override_datetime.strftime("%Y-%m-%d %H:%M:%S")
+        study_id_sql = (
+            f"'{db.santize_string(self.study_id)}'"
+            if self.study_id is not None
+            else "NULL"
+        )
+        subject_id_sql = (
+            f"'{db.santize_string(self.subject_id)}'"
+            if self.subject_id is not None
+            else "NULL"
+        )
+
+        sql_query = f"""
+        INSERT INTO {TABLE_NAME} (
+            do_identifier, do_study_id, do_subject_id, do_override_datetime
+        ) VALUES (
+            '{identifier}', {study_id_sql}, {subject_id_sql}, '{override_datetime}'
+        ) ON CONFLICT (do_identifier) DO UPDATE SET
+            do_study_id = EXCLUDED.do_study_id,
+            do_subject_id = EXCLUDED.do_subject_id,
+            do_override_datetime = EXCLUDED.do_override_datetime,
+            do_consumed_at = NULL;
+        """
+
+        return sql_query
+
+
+def get_override_datetime(config_file: Path, identifier: str) -> Optional[datetime]:
+    """
+    Looks up a staff-confirmed datetime override for a raw file/directory
+    path that failed to date-parse, if one has been recorded.
+
+    Deliberately does NOT filter on do_consumed_at: the crawler has no
+    "already imported, skip" check and re-parses every raw file/directory
+    from scratch on every pass, so a one-shot override would stop matching
+    after its first successful use - silently un-resolving the same
+    datetime_parse pipeline_failures row (and re-erroring instead of
+    re-logging the override) on every subsequent run. The override must
+    keep applying on every pass for as long as the raw name still fails to
+    parse. do_consumed_at is retained purely as a record of when the
+    override was first successfully picked up - see mark_consumed().
+
+    Args:
+        config_file (Path): The path to the configuration file.
+        identifier (str): The raw path that failed to parse (matches
+            pipeline_failures.pf_identifier for the same 'datetime_parse'
+            failure).
+
+    Returns:
+        Optional[datetime]: The overridden datetime, or None if no override
+        has been recorded for this identifier.
+    """
+    identifier_sql = db.santize_string(identifier)
+    query = f"""
+        SELECT do_override_datetime
+        FROM {TABLE_NAME}
+        WHERE do_identifier = '{identifier_sql}';
+    """
+    result = db.fetch_record(config_file=config_file, query=query)
+    if result is None:
+        return None
+    return datetime.fromisoformat(result)
+
+
+def mark_consumed(config_file: Path, identifier: str) -> None:
+    """
+    Records the first time a crawler pass successfully used this override to
+    import the file - informational only, does not gate get_override_datetime().
+    Best-effort and a no-op if no matching (not-yet-stamped) override row
+    exists, mirroring db.resolve_failure()'s call-on-every-successful-import
+    semantics.
+    """
+    identifier_sql = db.santize_string(identifier)
+    query = f"""
+        UPDATE {TABLE_NAME}
+        SET do_consumed_at = CURRENT_TIMESTAMP
+        WHERE do_identifier = '{identifier_sql}' AND do_consumed_at IS NULL;
+    """
+    db.execute_queries(
+        config_file=config_file,
+        queries=[query],
+        show_commands=False,
+        silent=True,
+    )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="datetime_overrides",
+        description="Initialize the 'pipeline_ledger.datetime_overrides' table.",
+    )
+    parser.add_argument(
+        "-c", "--config", type=str, help="Path to the config file.", required=False
+    )
+
+    args = parser.parse_args()
+
+    if args.config:
+        config_file = Path(args.config).resolve()
+        if not config_file.exists():
+            console.log(f"[red]Error: Config file '{config_file}' does not exist.")
+            sys.exit(1)
+    else:
+        config_file = utils.get_config_file_path()
+
+    console.log("Initializing 'datetime_overrides' table...")
+
+    create_queries = [DatetimeOverride.init_table_query()]  # CREATE TABLE IF NOT EXISTS
+
+    if cli.confirm_action(
+        "This table accumulates staff-confirmed datetime overrides for "
+        "manually-matched files. Drop and recreate 'datetime_overrides', "
+        "destroying all existing overrides?"
+    ):
+        console.log("[red]Dropping 'datetime_overrides' table if it exists...")
+        sql_queries = [DatetimeOverride.drop_table_query()] + create_queries
+    else:
+        console.log(
+            "Skipping drop. Creating the table only if it doesn't already exist "
+            "(existing data, if any, is preserved)."
+        )
+        sql_queries = create_queries
+
+    db.execute_queries(config_file=config_file, queries=sql_queries)
+
+    console.log("[green]Done!")