From 6bb8595ed658a692fb85c242fe10f52fe82f0e37 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Thu, 16 Jul 2026 11:00:28 -0400 Subject: [PATCH 1/3] Introduce datetime_overrides table to db --- pipeline/models/__init__.py | 3 + pipeline/models/datetime_overrides.py | 238 ++++++++++++++++++++++++++ 2 files changed, 241 insertions(+) create mode 100644 pipeline/models/datetime_overrides.py diff --git a/pipeline/models/__init__.py b/pipeline/models/__init__.py index 4e50483..6254d79 100644 --- a/pipeline/models/__init__.py +++ b/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,6 +91,7 @@ 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(), ] @@ -97,6 +99,7 @@ def init_db(config_file: Path): 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(), diff --git a/pipeline/models/datetime_overrides.py b/pipeline/models/datetime_overrides.py new file mode 100644 index 0000000..f5b4fee --- /dev/null +++ b/pipeline/models/datetime_overrides.py @@ -0,0 +1,238 @@ +#!/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, so a + previously-consumed override that gets corrected is picked up again + on the next crawler pass. + """ + 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 and not yet + consumed by a prior crawler pass. + + 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 + (unconsumed) 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}' AND do_consumed_at IS NULL; + """ + 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: + """ + Marks a datetime override as consumed, once a crawler pass has + successfully used it to import the file. Best-effort and a no-op if no + matching (unconsumed) 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!") From 5091fd5f3dbc8da32ffe03828e16b5f3d7ecccb0 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Thu, 16 Jul 2026 14:04:17 -0400 Subject: [PATCH 2/3] Utilize datetime_overrides on file parse failure --- .../ampscz/2_import_interview_files.py | 88 ++++++++++++++----- 1 file changed, 64 insertions(+), 24 deletions(-) diff --git a/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py b/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py index c288d44..3173384 100755 --- a/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py +++ b/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: """ From 9ab87204037451661a79c31539fdcf9435de89ba Mon Sep 17 00:00:00 2001 From: rtmcard Date: Mon, 20 Jul 2026 16:16:46 -0400 Subject: [PATCH 3/3] Remove override_consumed check for subsequent crawler runs --- pipeline/models/datetime_overrides.py | 36 ++++++++++++++++++--------- 1 file changed, 24 insertions(+), 12 deletions(-) diff --git a/pipeline/models/datetime_overrides.py b/pipeline/models/datetime_overrides.py index f5b4fee..2ccc3dd 100644 --- a/pipeline/models/datetime_overrides.py +++ b/pipeline/models/datetime_overrides.py @@ -115,9 +115,11 @@ def drop_table_query() -> str: 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, so a - previously-consumed override that gets corrected is picked up again - on the next crawler pass. + 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") @@ -150,8 +152,17 @@ def to_sql(self) -> str: 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 and not yet - consumed by a prior crawler pass. + 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. @@ -160,14 +171,14 @@ def get_override_datetime(config_file: Path, identifier: str) -> Optional[dateti failure). Returns: - Optional[datetime]: The overridden datetime, or None if no - (unconsumed) override has been recorded for this identifier. + 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}' AND do_consumed_at IS NULL; + WHERE do_identifier = '{identifier_sql}'; """ result = db.fetch_record(config_file=config_file, query=query) if result is None: @@ -177,10 +188,11 @@ def get_override_datetime(config_file: Path, identifier: str) -> Optional[dateti def mark_consumed(config_file: Path, identifier: str) -> None: """ - Marks a datetime override as consumed, once a crawler pass has - successfully used it to import the file. Best-effort and a no-op if no - matching (unconsumed) override row exists, mirroring - db.resolve_failure()'s call-on-every-successful-import semantics. + 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"""