From b53ef72d97044c8009825a2250af2fde19906931 Mon Sep 17 00:00:00 2001 From: Ryan McArdle Date: Tue, 7 Jul 2026 09:53:54 -0400 Subject: [PATCH 01/15] Prevent double-progress-bars from swallowing query executions --- .../ampscz/4_import_audio_journals.py | 5 +++- pipeline/helpers/db.py | 25 ++++++++++++++++--- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py index 0d13c75..46e713c 100755 --- a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py +++ b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py @@ -366,12 +366,15 @@ def import_journals(config_file: Path, study_id: str, progress: Progress) -> Non ) # Execute the queries + # Note: no show_progress here - this runs inside the outer per-study progress + # bar started in __main__, and rich only allows one active progress display + # at a time. db.execute_queries( queries=sql_queries, config_file=config_file, show_commands=False, - show_progress=True, on_failure=lambda: (logger.error("Error executing queries")), + # This will hide which queries failed/why; consider updated logging here? ) diff --git a/pipeline/helpers/db.py b/pipeline/helpers/db.py index d586d91..9646c27 100644 --- a/pipeline/helpers/db.py +++ b/pipeline/helpers/db.py @@ -12,6 +12,7 @@ import pandas as pd import psycopg2 import sqlalchemy +from rich.errors import LiveError from pipeline import orchestrator from pipeline.helpers import cli, utils @@ -192,11 +193,27 @@ def execute_query(query: str): pass if show_progress: - with utils.get_progress_bar() as progress: - task = progress.add_task("Executing SQL queries...", total=len(queries)) - + try: + with utils.get_progress_bar() as progress: + task = progress.add_task( + "Executing SQL queries...", total=len(queries) + ) + + for command in queries: + progress.update(task, advance=1) + execute_query(command) + except LiveError: + # A progress bar (rich.live.Live) is already active elsewhere in the + # call stack - rich only allows one at a time. Fall back to running + # the queries without a progress bar instead of letting this bubble + # up into the except below, which would abort the whole batch before + # a single query has run. + logger.warning( + "[yellow]show_progress=True requested but a progress display is " + "already active; running without a progress bar.", + extra={"markup": True}, + ) for command in queries: - progress.update(task, advance=1) execute_query(command) else: From 122d778aa67f0e318cc640e9360b819b97537adc Mon Sep 17 00:00:00 2001 From: Ryan McArdle Date: Tue, 7 Jul 2026 10:56:51 -0400 Subject: [PATCH 02/15] Increased logging verbosity for tracking dropped/errored files --- pipeline/core/__init__.py | 4 ++++ pipeline/core/load_openface.py | 6 +++++- pipeline/core/wipe.py | 21 ++++++++++++------- .../ampscz/4_import_audio_journals.py | 9 ++++++-- pipeline/helpers/db.py | 20 ++++++++++++++++++ pipeline/helpers/sftp.py | 10 +++++++++ pipeline/models/ffprobe_metadata.py | 6 ++++-- pipeline/report/main.py | 17 +++++++++++---- pipeline/report/video/qc.py | 20 +++++++++++++++--- pipeline/runners/01_fetch_video.py | 5 ++++- pipeline/runners/08_load_openface.py | 6 +++++- pipeline/runners/21_fetch_audio.py | 5 ++++- pipeline/runners/99_wiper.py | 5 ++++- .../22_openface_role_validation.py | 6 +++++- .../23_llm_speaker_identification.py | 14 +++++++++++-- .../24_llm_language_identification.py | 16 +++++++++++--- .../transcription/25_transcribeme_pull.py | 5 ++++- .../22_transcribeme_push_interview_audio.py | 5 +++++ pipeline/scripts/wipe_exported_assets.py | 1 + 19 files changed, 151 insertions(+), 30 deletions(-) diff --git a/pipeline/core/__init__.py b/pipeline/core/__init__.py index 036925d..8da5f22 100644 --- a/pipeline/core/__init__.py +++ b/pipeline/core/__init__.py @@ -176,6 +176,10 @@ def get_openface_path( try: of_path = Path(openface_path) except TypeError: + logger.warning( + f"openface path is not set (got {openface_path!r}) for interview " + f"{interview_name}, subject {subject_id}, study {study_id}, role {role}" + ) return None if not of_path.exists() and redirect_to_exported_assets: diff --git a/pipeline/core/load_openface.py b/pipeline/core/load_openface.py index 582f7a1..6e6d01a 100644 --- a/pipeline/core/load_openface.py +++ b/pipeline/core/load_openface.py @@ -292,7 +292,11 @@ def construct_insert_queries( case _: pass except ValueError as e: - print(f"Error casting {col} with value {df[col]} to {datatype}: {e}") + logger.error( + f"Error casting column {col} to {datatype} while building " + f"openface_features insert queries for interview {interview_name}, " + f"role {role}, csv_file {csv_file}: {e}" + ) queries: List[str] = [] diff --git a/pipeline/core/wipe.py b/pipeline/core/wipe.py index 8394536..50115eb 100644 --- a/pipeline/core/wipe.py +++ b/pipeline/core/wipe.py @@ -149,9 +149,10 @@ def get_interview_files( interview_name=interview_name, role=role, ) - except FileNotFoundError: - stream = None - except ValueError: + except (FileNotFoundError, ValueError) as e: + logger.debug( + f"No stream found for interview {interview_name}, role {role}: {e}" + ) stream = None if stream is not None: @@ -163,9 +164,11 @@ def get_interview_files( interview_name=interview_name, role=role, ) - except FileNotFoundError: - of_path = None - except ValueError: + except (FileNotFoundError, ValueError) as e: + logger.debug( + f"No OpenFace path found for interview {interview_name}, " + f"role {role}: {e}" + ) of_path = None if of_path is not None: @@ -177,7 +180,11 @@ def get_interview_files( interview_name=interview_name, report_version=version, ) - except FileNotFoundError: + except FileNotFoundError as e: + logger.debug( + f"No PDF report found for interview {interview_name}, " + f"version {version}: {e}" + ) report_path = None related_files.extend(decrypted_files) diff --git a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py index 46e713c..b9648de 100755 --- a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py +++ b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py @@ -373,8 +373,13 @@ def import_journals(config_file: Path, study_id: str, progress: Progress) -> Non queries=sql_queries, config_file=config_file, show_commands=False, - on_failure=lambda: (logger.error("Error executing queries")), - # This will hide which queries failed/why; consider updated logging here? + on_failure=lambda: logger.error( + f"Failed to import audio journals for study {study_id}: " + f"none of the {len(sql_queries)} pending journal-insert query(ies) for " + f"this study were persisted. See the query/error logged above for detail; " + f"per-journal failure detail is not yet available (batch is not isolated " + f"per-record)." + ), ) diff --git a/pipeline/helpers/db.py b/pipeline/helpers/db.py index 9646c27..dd8103a 100644 --- a/pipeline/helpers/db.py +++ b/pipeline/helpers/db.py @@ -153,6 +153,7 @@ def execute_queries( """ command = None output = [] + executed_count = 0 if backup: repo_root = cli.get_repo_root_from_config(config_file=config_file) @@ -183,6 +184,7 @@ def execute_queries( cur = conn.cursor() def execute_query(query: str): + nonlocal executed_count if show_commands: logger.debug("Executing query:") logger.debug(f"[bold blue]{query}", extra={"markup": True}) @@ -191,6 +193,7 @@ def execute_query(query: str): output.append(cur.fetchall()) except psycopg2.ProgrammingError: pass + executed_count += 1 if show_progress: try: @@ -233,6 +236,23 @@ def execute_query(query: str): if command is not None: logger.error(f"[red]For query: {command}", extra={"markup": True}) logger.error(e) + if len(queries) > 1: + # execute_queries() commits once, after every query in the batch has + # run - the connection is closed without ever calling commit() here, + # so Postgres rolls back the whole transaction. That means none of + # this batch's queries were persisted, including the ones that ran + # fine before the failure above, not just the ones after it. We + # can't say here which individual files/records those queries + # belonged to (that requires the caller to isolate per-record), so + # surface the blast radius as a count instead of silently dropping + # them. + logger.warning( + f"[yellow]Batch aborted: {executed_count}/{len(queries)} queries in " + f"this call ran before the failure above, but since this batch " + f"never reached commit(), none of the {len(queries)} query(ies) " + f"were persisted.", + extra={"markup": True}, + ) if on_failure is not None: on_failure() else: diff --git a/pipeline/helpers/sftp.py b/pipeline/helpers/sftp.py index 54e21a9..2da5bff 100644 --- a/pipeline/helpers/sftp.py +++ b/pipeline/helpers/sftp.py @@ -96,6 +96,16 @@ def sftp_move_file( logger.info( f"Destination directory does not exist, creating: {remote_destination_path.parent}" ) + else: + # Any other OSError (permissions, disk full, etc.) still falls through + # to the mkdir+retry below, which is unlikely to be the right recovery + # for a non-"missing directory" error. Log the real error so a + # follow-up failure at the retried rename isn't a total mystery. + logger.warning( + f"Unexpected error (errno={e.errno}) renaming [remote] " + f"{remote_source_path} to [remote] {remote_destination_path}: {e}. " + f"Attempting the missing-directory recovery anyway." + ) sftp.mkdir(str(remote_destination_path.parent)) # Retry the rename operation after creating the directory sftp.rename(str(remote_source_path), str(remote_destination_path)) diff --git a/pipeline/models/ffprobe_metadata.py b/pipeline/models/ffprobe_metadata.py index 78ca124..3e9526a 100644 --- a/pipeline/models/ffprobe_metadata.py +++ b/pipeline/models/ffprobe_metadata.py @@ -393,7 +393,7 @@ def stream_to_sql( ) ON CONFLICT (fma_source_path) DO NOTHING; """ except KeyError as e: - logger.error(f"Key error: {e}") + logger.error(f"Key error building ffprobe stream row for {source_path}: {e}") logger.debug(f"Stream: {stream}") raise e else: @@ -450,7 +450,9 @@ def to_sql(self) -> List[str]: try: streams = self.metadata["streams"] except KeyError as e: - logger.error(f"Metadata does not have 'streams' key: {e}") + logger.error( + f"Metadata does not have 'streams' key for {self.source_path}: {e}" + ) logger.debug(f"Metadata: {self.metadata}") return [ f""" diff --git a/pipeline/report/main.py b/pipeline/report/main.py index ebcdfaa..12c39e5 100644 --- a/pipeline/report/main.py +++ b/pipeline/report/main.py @@ -2,6 +2,7 @@ Generates a report for the Interview. """ +import logging import tempfile from datetime import timedelta from pathlib import Path @@ -21,6 +22,8 @@ from pipeline.models.lite.interview_metadata import InterviewMetadata from pipeline.report import common, header, video +logger = logging.getLogger(__name__) + def generate_report( interview_name: str, @@ -88,8 +91,11 @@ def generate_report( of_pt_session["timestamp"] = of_pt_session["timestamp"].apply( utils.datetime_time_to_float ) - except AttributeError: - pass + except AttributeError as e: + logger.debug( + f"Could not convert subject timestamp column to float for " + f"{interview_name}, leaving as-is: {e}" + ) if interview_metadata.has_interviewer_stream: status.update("Fetching OpenFace features for interviewer...") @@ -105,8 +111,11 @@ def generate_report( of_int_session["timestamp"] = of_int_session["timestamp"].apply( utils.datetime_time_to_float ) - except AttributeError: - pass + except AttributeError as e: + logger.debug( + f"Could not convert interviewer timestamp column to float for " + f"{interview_name}, leaving as-is: {e}" + ) temp_files_common: List[tempfile.NamedTemporaryFile] = [] # type: ignore diff --git a/pipeline/report/video/qc.py b/pipeline/report/video/qc.py index c03c493..85a02d4 100644 --- a/pipeline/report/video/qc.py +++ b/pipeline/report/video/qc.py @@ -2,6 +2,7 @@ Quality Control Section for Apperance and Behavior Section """ +import logging import sys import tempfile from datetime import timedelta @@ -18,6 +19,7 @@ from pipeline.models.lite.video_metadata import VideoMetadata console = utils.get_console() +logger = logging.getLogger(__name__) def draw_sample_image( @@ -361,7 +363,11 @@ def draw_qc_metrics_by_role( frames_color = orange_color else: frames_color = red_color - except TypeError: + except TypeError as e: + logger.debug( + f"successful_frames_percentage missing/invalid for {interview_name} " + f"role {role}, defaulting QC color to red: {e}" + ) frames_color = red_color try: @@ -371,7 +377,11 @@ def draw_qc_metrics_by_role( confidence_color = orange_color else: confidence_color = red_color - except TypeError: + except TypeError as e: + logger.debug( + f"successful_frames_confidence_mean missing/invalid for " + f"{interview_name} role {role}, defaulting QC color to red: {e}" + ) confidence_color = red_color match role: @@ -388,7 +398,11 @@ def draw_qc_metrics_by_role( try: confidence_qc_text = f"{qc_metrics.successful_frames_confidence_mean * 100:.2f}% OF confidence mean" - except TypeError: + except TypeError as e: + logger.debug( + f"successful_frames_confidence_mean missing/invalid for " + f"{interview_name} role {role}, defaulting QC text to 0%: {e}" + ) confidence_qc_text = "0% OF confidence mean" pdf.draw_text( diff --git a/pipeline/runners/01_fetch_video.py b/pipeline/runners/01_fetch_video.py index 57f9a78..692a809 100755 --- a/pipeline/runners/01_fetch_video.py +++ b/pipeline/runners/01_fetch_video.py @@ -169,7 +169,10 @@ def on_failure(): global COUNTER # pylint: disable=global-statement COUNTER -= 1 - logger.info("Decryption request failed. Ignoring file.") + logger.info( + f"Decryption request failed. Ignoring file: " + f"{file_to_decrypt_path}" + ) sql_query = InterviewFile.ignore_file(file_to_decrypt_path) db.execute_queries(config_file=config_file, queries=[sql_query]) diff --git a/pipeline/runners/08_load_openface.py b/pipeline/runners/08_load_openface.py index 8a77e0e..97391cb 100755 --- a/pipeline/runners/08_load_openface.py +++ b/pipeline/runners/08_load_openface.py @@ -83,6 +83,7 @@ study_id = studies[0] logger.info(f"Starting with study: {study_id}") + interview_name = None try: while True: interview_name = load_openface.get_interview_to_process( @@ -127,7 +128,10 @@ load_openface.log_load_openface(config_file=config_file, lof=lof) except Exception as e: - logger.error(f"Error: {e}") + logger.error( + f"Error loading OpenFace features (last interview attempted: " + f"{interview_name}): {e}" + ) notifications.send_notification( notify_type="failure", title=f"{MODULE_NAME} failed", diff --git a/pipeline/runners/21_fetch_audio.py b/pipeline/runners/21_fetch_audio.py index 20af83e..cc7b6a8 100755 --- a/pipeline/runners/21_fetch_audio.py +++ b/pipeline/runners/21_fetch_audio.py @@ -149,7 +149,10 @@ def on_failure(): global COUNTER # pylint: disable=global-statement COUNTER -= 1 - logger.info("Decryption request failed. Ignoring file.") + logger.info( + f"Decryption request failed. Ignoring file: " + f"{file_to_decrypt_path}" + ) sql_query = InterviewFile.ignore_file(file_to_decrypt_path) db.execute_queries(config_file=config_file, queries=[sql_query]) diff --git a/pipeline/runners/99_wiper.py b/pipeline/runners/99_wiper.py index a1f1dc4..08debae 100755 --- a/pipeline/runners/99_wiper.py +++ b/pipeline/runners/99_wiper.py @@ -202,7 +202,10 @@ def get_interviews_to_wipe(config_file: Path, study_id: str) -> List[str]: show_commands=True, ) except Exception as e: - logger.error(f"Error: {e}") + logger.error( + f"Error dropping DB rows for interview " + f"{interview_to_wipe}: {e}" + ) logger.error("Continuing...") logger.info( f"Wiped interview: [bold blue]{interview_to_wipe} in {timer.duration} seconds.", diff --git a/pipeline/runners/study_specific/22_openface_role_validation.py b/pipeline/runners/study_specific/22_openface_role_validation.py index fafac3d..a0f4714 100755 --- a/pipeline/runners/study_specific/22_openface_role_validation.py +++ b/pipeline/runners/study_specific/22_openface_role_validation.py @@ -289,7 +289,11 @@ def check_match(transcript_qqc: Dict[str, Any], fau_data: Dict[str, Any]) -> boo try: if transcript_qqc[speaker]["role"] != stats["role"]: return False - except KeyError: + except KeyError as e: + logger.debug( + f"Missing 'role' key while comparing speaker {speaker!r} " + f"(transcript_qqc vs FAU-derived stats): {e}. Treating as a mismatch." + ) return False return True diff --git a/pipeline/runners/study_specific/23_llm_speaker_identification.py b/pipeline/runners/study_specific/23_llm_speaker_identification.py index d1d165c..35099f9 100755 --- a/pipeline/runners/study_specific/23_llm_speaker_identification.py +++ b/pipeline/runners/study_specific/23_llm_speaker_identification.py @@ -134,10 +134,17 @@ def parse_transcript_to_df(transcript: Path) -> pd.DataFrame: pd.to_datetime(time, format="%H:%M:%S.%f") except ValueError: # add text to the previous line - print(line) + logger.debug( + f"Line does not start with a timestamp in {transcript}, " + f"treating as a continuation of the previous line: {line!r}" + ) data[-1]["transcript"] += " " + line.strip() continue except ValueError: + logger.debug( + f"Could not split line into speaker/time/text in {transcript}, " + f"skipping line: {line!r}" + ) continue text = text.strip() @@ -261,7 +268,10 @@ def process_transcript( template=template, ) except ValueError as e: - logger.error(f"Error: {e}") + logger.error( + f"Error building LLM prompt for {transcript_path} " + f"(role {role}): {e}" + ) subject_result = LlmSpeakerIdentification( llm_source_transcript=transcript_path, ollama_model_identifier="default", diff --git a/pipeline/runners/study_specific/24_llm_language_identification.py b/pipeline/runners/study_specific/24_llm_language_identification.py index 9256388..db7e992 100755 --- a/pipeline/runners/study_specific/24_llm_language_identification.py +++ b/pipeline/runners/study_specific/24_llm_language_identification.py @@ -126,10 +126,17 @@ def parse_transcript_to_df(transcript: Path) -> pd.DataFrame: pd.to_datetime(time, format="%H:%M:%S.%f") except ValueError: # add text to the previous line - print(line) + logger.debug( + f"Line does not start with a timestamp in {transcript}, " + f"treating as a continuation of the previous line: {line!r}" + ) data[-1]["transcript"] += " " + line.strip() continue except ValueError: + logger.debug( + f"Could not split line into speaker/time/text in {transcript}, " + f"skipping line: {line!r}" + ) continue text = text.strip() @@ -242,7 +249,7 @@ def process_transcript( template=template, ) except ValueError as e: - logger.error(f"Error: {e}") + logger.error(f"Error building LLM prompt for {transcript_path}: {e}") return LlmLanguageIdentification( llm_source_transcript=transcript_path, ollama_model_identifier="default", @@ -402,7 +409,10 @@ def log_language_identification_result( transcript_path=file_to_process, config_file=config_file, ) - except ValueError: + except ValueError as e: + logger.error( + f"Skipping language identification for {file_to_process}: {e}" + ) continue language_identification_duration = timer.duration diff --git a/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py b/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py index 98e7185..e082480 100755 --- a/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py +++ b/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py @@ -338,7 +338,10 @@ def get_match( logger.info(f"[remote] Deleting {sftp_upload_path}") sftp.sftp_delete_file(sftp_client, sftp_upload_path) except paramiko.SSHException as e: - logger.error(f"SSH connection error: {e}") + logger.error( + f"SSH connection error while pulling transcript " + f"{transcript_path}: {e}" + ) logger.info(f"Retrying in {retry_timeout} seconds...") time.sleep(retry_timeout) diff --git a/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py b/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py index 410112d..8a91579 100755 --- a/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py +++ b/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py @@ -154,6 +154,11 @@ def get_interview_session_number( session_number = result_df["interview_name"].tolist().index(interview_name) + 1 return session_number except ValueError: + logger.warning( + f"Interview {interview_name} (subject {subject_id}, type " + f"{interview_type}) not found among its own study's interviews " + f"list; cannot compute a session number." + ) return None diff --git a/pipeline/scripts/wipe_exported_assets.py b/pipeline/scripts/wipe_exported_assets.py index 54617b9..6fbd80b 100755 --- a/pipeline/scripts/wipe_exported_assets.py +++ b/pipeline/scripts/wipe_exported_assets.py @@ -124,6 +124,7 @@ def get_items_to_delete( cli.remove(item) COUNTER += 1 except FileNotFoundError: + logger.debug(f"Item already removed, skipping: {item}") SKIP_COUNTER += 1 continue From b9900c80389fbb830f85d7f65f98e96e5f6a14cd Mon Sep 17 00:00:00 2001 From: Ryan McArdle Date: Tue, 7 Jul 2026 11:33:59 -0400 Subject: [PATCH 03/15] Introduce persistent error ledger into db --- pipeline/helpers/db.py | 134 ++++++++++++++++++++- pipeline/models/__init__.py | 3 + pipeline/models/pipeline_failures.py | 170 +++++++++++++++++++++++++++ 3 files changed, 306 insertions(+), 1 deletion(-) create mode 100644 pipeline/models/pipeline_failures.py diff --git a/pipeline/helpers/db.py b/pipeline/helpers/db.py index dd8103a..c9f23d1 100644 --- a/pipeline/helpers/db.py +++ b/pipeline/helpers/db.py @@ -7,7 +7,7 @@ import sys from datetime import datetime from pathlib import Path -from typing import Callable, Dict, Literal, Optional +from typing import Callable, Dict, Literal, Optional, Union import pandas as pd import psycopg2 @@ -132,6 +132,9 @@ def execute_queries( db: str = "postgresql", backup: bool = False, on_failure: Optional[Callable] = on_failure, + failure_stage: Optional[str] = None, + failure_identifier: Optional[str] = None, + failure_identifier_type: str = "batch", ) -> list: """ Executes a list of SQL queries on a PostgreSQL database. @@ -147,6 +150,12 @@ def execute_queries( db (str, optional): The section of the configuration file to use. Defaults to "postgresql". backup (bool, optional): Whether to sace all executed queries to a file. + failure_stage (str, optional): If set (together with failure_identifier), + a failure will also be recorded in the pipeline_failures ledger via + record_failure(). Defaults to None (no ledger entry). + failure_identifier (str, optional): What failed - see failure_stage. + failure_identifier_type (str, optional): The kind of thing + failure_identifier is (e.g. "study", "file_path"). Defaults to "batch". Returns: list: A list of tuples containing the results of the executed queries. @@ -253,6 +262,15 @@ def execute_query(query: str): f"were persisted.", extra={"markup": True}, ) + if failure_stage is not None and failure_identifier is not None: + record_failure( + config_file=config_file, + stage=failure_stage, + identifier=failure_identifier, + identifier_type=failure_identifier_type, + error=e, + db=db, + ) if on_failure is not None: on_failure() else: @@ -264,6 +282,120 @@ def execute_query(query: str): return output +def record_failure( + config_file: Path, + stage: str, + identifier: str, + error: Union[str, Exception], + identifier_type: Literal["file_path", "study", "batch", "other"] = "file_path", + db: str = "postgresql", +) -> None: + """ + Records (or bumps the occurrence count / last_seen_at of) a failure in the + durable pipeline_failures ledger, for turning "what's stuck and why" into a + query instead of grepping log files. Distinct from the generic 'logs' table. + + Best-effort and never raises: a bug in this bookkeeping call must not crash + or mask the caller's real failure. + + Args: + config_file (Path): The path to the configuration file. + stage (str): The pipeline stage/module the failure occurred in. + identifier (str): What failed - a file path when known, otherwise the + most specific thing available (study_id, a batch description, etc). + error (Union[str, Exception]): The error. If an Exception is passed, + its class name is recorded as the error type. + identifier_type (str, optional): The kind of thing `identifier` is. + Defaults to "file_path". + db (str, optional): The section of the configuration file to use. + Defaults to "postgresql". + """ + try: + # Local import: pipeline.models.pipeline_failures imports pipeline.helpers.db + # (like every model file), so importing it at module load time here would + # be circular. + from pipeline.models.pipeline_failures import PipelineFailure + + error_type = type(error).__name__ if isinstance(error, Exception) else None + failure = PipelineFailure( + stage=stage, + identifier=identifier, + identifier_type=identifier_type, + error=str(error), + error_type=error_type, + ) + execute_queries( + config_file=config_file, + queries=[failure.to_sql()], + show_commands=False, + silent=True, + db=db, + on_failure=lambda: logger.error( + f"[yellow]Could not record failure-ledger row for " + f"stage={stage!r} identifier={identifier!r} (see error logged above).", + extra={"markup": True}, + ), + ) + except Exception: + logger.exception( + f"Unexpected error recording pipeline failure ledger row for " + f"stage={stage!r} identifier={identifier!r}; continuing without it." + ) + + +def resolve_failure( + config_file: Path, + stage: str, + identifier: str, + note: Optional[str] = None, + db: str = "postgresql", +) -> None: + """ + Marks a pipeline_failures row as resolved. Companion to record_failure() - + not called automatically anywhere; call sites opt in explicitly. + + Args: + config_file (Path): The path to the configuration file. + stage (str): The pipeline stage/module, matching a prior record_failure() call. + identifier (str): What was fixed, matching a prior record_failure() call. + note (str, optional): A note on how/why this was resolved. + db (str, optional): The section of the configuration file to use. + Defaults to "postgresql". + """ + try: + stage_sql = santize_string(stage) + identifier_sql = santize_string(identifier) + note_clause = ( + f", pf_resolved_note = '{santize_string(note)}'" if note is not None else "" + ) + + query = f""" + UPDATE pipeline_failures + SET pf_resolved = TRUE, + pf_resolved_at = CURRENT_TIMESTAMP + {note_clause} + WHERE pf_stage = '{stage_sql}' AND pf_identifier = '{identifier_sql}'; + """ + + execute_queries( + config_file=config_file, + queries=[query], + show_commands=False, + silent=True, + db=db, + on_failure=lambda: logger.error( + f"[yellow]Could not mark failure resolved for " + f"stage={stage!r} identifier={identifier!r}.", + extra={"markup": True}, + ), + ) + except Exception: + logger.exception( + f"Unexpected error resolving pipeline failure ledger row for " + f"stage={stage!r} identifier={identifier!r}." + ) + + def get_db_connection( config_file: Path, db: str = "postgresql" ) -> sqlalchemy.engine.base.Engine: diff --git a/pipeline/models/__init__.py b/pipeline/models/__init__.py index f8cc197..4e50483 100644 --- a/pipeline/models/__init__.py +++ b/pipeline/models/__init__.py @@ -15,6 +15,7 @@ from pipeline.models.interview_files import InterviewFile from pipeline.models.key_store import KeyStore from pipeline.models.logs import Log +from pipeline.models.pipeline_failures import PipelineFailure from pipeline.models.decrypted_files import DecryptedFile from pipeline.models.video_qqc import VideoQuickQc from pipeline.models.interview_roles import InterviewRole @@ -88,12 +89,14 @@ def init_db(config_file: Path): Study.drop_table_query(), KeyStore.drop_table_query(), Log.drop_table_query(), + PipelineFailure.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(), Study.init_table_query(), Subject.init_table_query(), FormData.init_table_query(), diff --git a/pipeline/models/pipeline_failures.py b/pipeline/models/pipeline_failures.py new file mode 100644 index 0000000..95efd85 --- /dev/null +++ b/pipeline/models/pipeline_failures.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python +""" +PipelineFailure Model + +A durable, queryable ledger of pipeline failures - distinct from the generic +'logs' table. Deduplicates repeated failures of the same thing at the same +stage into a single row (bumping an occurrence count / last-seen timestamp) +instead of growing unboundedly, and preserves when a failure was first seen +even as it recurs. +""" + +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 + + +from typing import Literal, Optional + +from pipeline.helpers import cli, db, utils + +console = utils.get_console() + +IdentifierType = Literal["file_path", "study", "batch", "other"] + + +class PipelineFailure: + """ + Represents a row in the 'pipeline_failures' table. + + Attributes: + stage (str): The pipeline stage/module the failure occurred in + (e.g. matches the module_name used for [logging] config sections). + identifier (str): What failed - a file path when known, otherwise the + most specific thing available (study_id, a batch description, etc). + error (str): The error message. + identifier_type (IdentifierType): What kind of thing `identifier` is. + error_type (Optional[str]): The exception class name, if known. + """ + + def __init__( + self, + stage: str, + identifier: str, + error: str, + identifier_type: IdentifierType = "file_path", + error_type: Optional[str] = None, + ) -> None: + self.stage = stage + self.identifier = identifier + self.identifier_type = identifier_type + self.error = error + self.error_type = error_type + + def __str__(self) -> str: + return f"PipelineFailure({self.stage}, {self.identifier}, {self.error})" + + def __repr__(self) -> str: + return self.__str__() + + @staticmethod + def init_table_query() -> str: + """ + Return the SQL query to create the 'pipeline_failures' table. + """ + sql_query = """ + CREATE TABLE IF NOT EXISTS pipeline_failures ( + pf_id SERIAL PRIMARY KEY, + pf_stage TEXT NOT NULL, + pf_identifier_type TEXT NOT NULL, + pf_identifier TEXT NOT NULL, + pf_error TEXT NOT NULL, + pf_error_type TEXT, + pf_occurrence_count INTEGER NOT NULL DEFAULT 1, + pf_first_seen_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + pf_last_seen_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + pf_resolved BOOLEAN NOT NULL DEFAULT FALSE, + pf_resolved_at TIMESTAMP, + pf_resolved_note TEXT, + UNIQUE (pf_stage, pf_identifier) + ); + """ + + return sql_query + + @staticmethod + def drop_table_query() -> str: + """ + Return the SQL query to drop the 'pipeline_failures' table. + """ + sql_query = """ + DROP TABLE IF EXISTS pipeline_failures; + """ + + return sql_query + + def to_sql(self) -> str: + """ + Return the SQL query to insert (or bump the occurrence count of) this + PipelineFailure in the 'pipeline_failures' table. + + Note: `pf_first_seen_at` is deliberately absent from the `DO UPDATE SET` + clause below - it must only ever be set once, by the INSERT's DEFAULT. + Recurrence flips `pf_resolved` back to FALSE: a failure that was marked + resolved and then happens again is not resolved anymore. + """ + stage = db.santize_string(self.stage) + identifier_type = db.santize_string(self.identifier_type) + identifier = db.santize_string(self.identifier) + error = db.santize_string(self.error) + error_type_sql = ( + f"'{db.santize_string(self.error_type)}'" + if self.error_type is not None + else "NULL" + ) + + sql_query = f""" + INSERT INTO pipeline_failures ( + pf_stage, pf_identifier_type, pf_identifier, pf_error, pf_error_type + ) VALUES ( + '{stage}', '{identifier_type}', '{identifier}', '{error}', {error_type_sql} + ) ON CONFLICT (pf_stage, pf_identifier) DO UPDATE SET + pf_identifier_type = EXCLUDED.pf_identifier_type, + pf_error = EXCLUDED.pf_error, + pf_error_type = EXCLUDED.pf_error_type, + pf_occurrence_count = pipeline_failures.pf_occurrence_count + 1, + pf_last_seen_at = CURRENT_TIMESTAMP, + pf_resolved = FALSE, + pf_resolved_at = NULL; + """ + + return sql_query + + +if __name__ == "__main__": + config_file = utils.get_config_file_path() + + console.log("Initializing 'pipeline_failures' table...") + + create_queries = [PipelineFailure.init_table_query()] # CREATE TABLE IF NOT EXISTS + + if cli.confirm_action( + "This table accumulates durable failure history (occurrence counts, " + "first/last-seen timestamps). Drop and recreate 'pipeline_failures', " + "destroying all existing failure history?" + ): + console.log("[red]Dropping 'pipeline_failures' table if it exists...") + sql_queries = [PipelineFailure.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 7af2f60ee1c5edf6ad10bb0a6a51c1da06cd0ad7 Mon Sep 17 00:00:00 2001 From: Ryan McArdle Date: Tue, 7 Jul 2026 11:55:48 -0400 Subject: [PATCH 04/15] Wire error ledger into pipeline --- pipeline/core/load_openface.py | 7 +++++++ pipeline/core/metadata.py | 15 +++++++++++++++ .../ampscz/2_import_interview_files.py | 9 ++++++++- .../study_specific/ampscz/3_import_transcripts.py | 7 +++++++ .../ampscz/4_import_audio_journals.py | 3 +++ .../ampscz/5_import_journal_transcripts.py | 12 ++++++++++-- .../ampscz/8_import_expected_interviews.py | 3 +++ .../ampscz/prescient/7_import_form_data.py | 11 ++++++++++- .../ampscz/pronet/7_import_form_data.py | 7 ++++++- pipeline/helpers/db.py | 4 +++- pipeline/models/pipeline_failures.py | 4 +++- pipeline/runners/01_fetch_video.py | 7 +++++++ pipeline/runners/08_load_openface.py | 9 ++++++++- pipeline/runners/21_fetch_audio.py | 7 +++++++ pipeline/runners/99_wiper.py | 3 +++ .../23_llm_speaker_identification.py | 7 +++++++ .../24_llm_language_identification.py | 14 ++++++++++++++ .../ampscz/transcription/25_transcribeme_pull.py | 7 +++++++ .../22_transcribeme_push_interview_audio.py | 8 ++++++++ 19 files changed, 136 insertions(+), 8 deletions(-) diff --git a/pipeline/core/load_openface.py b/pipeline/core/load_openface.py index 6e6d01a..ee6ddaa 100644 --- a/pipeline/core/load_openface.py +++ b/pipeline/core/load_openface.py @@ -297,6 +297,13 @@ def construct_insert_queries( f"openface_features insert queries for interview {interview_name}, " f"role {role}, csv_file {csv_file}: {e}" ) + db.record_failure( + config_file=config_file, + stage="load_openface", + identifier=interview_name, + error=e, + identifier_type="interview_name", + ) queries: List[str] = [] diff --git a/pipeline/core/metadata.py b/pipeline/core/metadata.py index 67b07ec..8998b09 100644 --- a/pipeline/core/metadata.py +++ b/pipeline/core/metadata.py @@ -81,6 +81,21 @@ def log_metadata( source_path=source, metadata=metadata, requested_by=requested_by ) + if "streams" not in metadata: + # FfprobeMetadata.to_sql() falls back to inserting a placeholder row + # (source_path/requested_by only, no stream data) when this happens - + # that placeholder permanently satisfies get_file_to_process()'s + # "not already in ffprobe_metadata" check above, so this file is never + # retried. Record it so it's queryable instead of silently stuck. + db.record_failure( + config_file=config_file, + stage="metadata", + identifier=str(source), + error="ffprobe metadata has no 'streams' key; inserting placeholder " + "row only - this file will not be retried", + identifier_type="file_path", + ) + sql_queries = ffprobe_metadata.to_sql() logger.info("Logging metadata...", extra={"markup": True}) 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 5a3236d..b553277 100755 --- a/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py +++ b/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py @@ -531,7 +531,14 @@ def import_interviews(config_file: Path, study_id: str, progress: Progress) -> N ) # Execute the SQL queries - db.execute_queries(config_file=config_file, queries=sql_queries, show_commands=False) + db.execute_queries( + config_file=config_file, + queries=sql_queries, + show_commands=False, + failure_stage="import_interview_files:ampscz", + failure_identifier=study_id, + failure_identifier_type="study", + ) def mark_unique_interviews_as_primary(config_file: Path, study_id: str) -> None: diff --git a/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py b/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py index 439cc30..d75975e 100755 --- a/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py +++ b/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py @@ -98,6 +98,13 @@ def transcripts_to_models( except IndexError as e: logger.error(f"Error processing transcript {filename}: {e}") logger.error("Skipping.") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(transcript), + error=e, + identifier_type="file_path", + ) continue # interview_path = core.get_interview_path( diff --git a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py index b9648de..8ddf93c 100755 --- a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py +++ b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py @@ -380,6 +380,9 @@ def import_journals(config_file: Path, study_id: str, progress: Progress) -> Non f"per-journal failure detail is not yet available (batch is not isolated " f"per-record)." ), + failure_stage=MODULE_NAME, + failure_identifier=study_id, + failure_identifier_type="study", ) diff --git a/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py b/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py index c59d5ee..236da3c 100755 --- a/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py +++ b/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py @@ -85,13 +85,14 @@ def get_diary_name_from_transcript(transcript_filename: str) -> str: def transcripts_to_models( - transcripts: List[Path] + transcripts: List[Path], config_file: Path ) -> Tuple[List[File], List[TranscriptFile]]: """ Converts the transcripts into File and InterviewFile models. Args: transcripts (List[Path]): The list of transcripts. + config_file (Path): The path to the config file. Returns: Tuple[List[File], List[TranscriptFile]]: The list of File and InterviewFile models. @@ -111,6 +112,13 @@ def transcripts_to_models( except IndexError as e: logger.error(f"Error processing transcript {filename}: {e}") logger.error("Skipping.") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(transcript), + error=e, + identifier_type="file_path", + ) continue file = File(file_path=transcript) @@ -180,7 +188,7 @@ def import_transcripts(data_root: Path, study: str, config_file: Path) -> None: logger.info(f"Found {len(transcripts)} transcripts.") files, transcript_files = transcripts_to_models( - transcripts=transcripts + transcripts=transcripts, config_file=config_file ) logger.info( diff --git a/pipeline/crawlers/study_specific/ampscz/8_import_expected_interviews.py b/pipeline/crawlers/study_specific/ampscz/8_import_expected_interviews.py index 9a4d1cc..9a2ed41 100755 --- a/pipeline/crawlers/study_specific/ampscz/8_import_expected_interviews.py +++ b/pipeline/crawlers/study_specific/ampscz/8_import_expected_interviews.py @@ -246,6 +246,9 @@ def models_to_db( queries=sql_queries, config_file=config_file, show_commands=False, + failure_stage="import_expected_interviews", + failure_identifier=study_id, + failure_identifier_type="study", ) diff --git a/pipeline/crawlers/study_specific/ampscz/prescient/7_import_form_data.py b/pipeline/crawlers/study_specific/ampscz/prescient/7_import_form_data.py index 6b04ac8..c676029 100755 --- a/pipeline/crawlers/study_specific/ampscz/prescient/7_import_form_data.py +++ b/pipeline/crawlers/study_specific/ampscz/prescient/7_import_form_data.py @@ -225,13 +225,18 @@ def parse_form_csv( return form_data_list -def models_to_db(form_data_list: List[FormData], config_file: Path) -> None: +def models_to_db( + form_data_list: List[FormData], config_file: Path, batch_identifier: str +) -> None: """ Imports the FormData models into the database. Args: form_data_list (List[FormData]): The list of FormData models. config_file (Path): The path to the config file. + batch_identifier (str): Identifies this batch (spans multiple studies - + unlike the pronet variant, prescient form data isn't imported + per-study, so there's no single study_id to key the failure on). Returns: None @@ -245,6 +250,9 @@ def models_to_db(form_data_list: List[FormData], config_file: Path) -> None: queries=sql_queries, config_file=config_file, show_commands=False, + failure_stage="import_form_data:prescient", + failure_identifier=batch_identifier, + failure_identifier_type="batch", ) @@ -287,6 +295,7 @@ def import_form_data( models_to_db( form_data_list=form_data_list, config_file=config_file, + batch_identifier=str(data_root), ) diff --git a/pipeline/crawlers/study_specific/ampscz/pronet/7_import_form_data.py b/pipeline/crawlers/study_specific/ampscz/pronet/7_import_form_data.py index 0dc91e2..0450d96 100755 --- a/pipeline/crawlers/study_specific/ampscz/pronet/7_import_form_data.py +++ b/pipeline/crawlers/study_specific/ampscz/pronet/7_import_form_data.py @@ -133,7 +133,7 @@ def parse_form_json(subject_form_json: Path) -> List[FormData]: def models_to_db( - form_data_list: List[FormData], config_file: Path + form_data_list: List[FormData], config_file: Path, study_id: str ) -> None: """ Imports the FormData models into the database. @@ -141,6 +141,7 @@ def models_to_db( Args: form_data_list (List[FormData]): The list of FormData models. config_file (Path): The path to the config file. + study_id (str): The study the form data belongs to. Returns: None @@ -154,6 +155,9 @@ def models_to_db( queries=sql_queries, config_file=config_file, show_commands=False, + failure_stage="import_form_data:pronet", + failure_identifier=study_id, + failure_identifier_type="study", ) @@ -199,6 +203,7 @@ def import_form_data( models_to_db( form_data_list=form_data_list, config_file=config_file, + study_id=study, ) diff --git a/pipeline/helpers/db.py b/pipeline/helpers/db.py index c9f23d1..e85d6eb 100644 --- a/pipeline/helpers/db.py +++ b/pipeline/helpers/db.py @@ -287,7 +287,9 @@ def record_failure( stage: str, identifier: str, error: Union[str, Exception], - identifier_type: Literal["file_path", "study", "batch", "other"] = "file_path", + identifier_type: Literal[ + "file_path", "study", "interview_name", "subject", "batch", "other" + ] = "file_path", db: str = "postgresql", ) -> None: """ diff --git a/pipeline/models/pipeline_failures.py b/pipeline/models/pipeline_failures.py index 95efd85..7bf865f 100644 --- a/pipeline/models/pipeline_failures.py +++ b/pipeline/models/pipeline_failures.py @@ -33,7 +33,9 @@ console = utils.get_console() -IdentifierType = Literal["file_path", "study", "batch", "other"] +IdentifierType = Literal[ + "file_path", "study", "interview_name", "subject", "batch", "other" +] class PipelineFailure: diff --git a/pipeline/runners/01_fetch_video.py b/pipeline/runners/01_fetch_video.py index 692a809..c8bbb75 100755 --- a/pipeline/runners/01_fetch_video.py +++ b/pipeline/runners/01_fetch_video.py @@ -173,6 +173,13 @@ def on_failure(): f"Decryption request failed. Ignoring file: " f"{file_to_decrypt_path}" ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(file_to_decrypt_path), + error="Decryption request failed; file marked ignored", + identifier_type="file_path", + ) sql_query = InterviewFile.ignore_file(file_to_decrypt_path) db.execute_queries(config_file=config_file, queries=[sql_query]) diff --git a/pipeline/runners/08_load_openface.py b/pipeline/runners/08_load_openface.py index 97391cb..b562e39 100755 --- a/pipeline/runners/08_load_openface.py +++ b/pipeline/runners/08_load_openface.py @@ -29,7 +29,7 @@ from pipeline import orchestrator from pipeline.core import load_openface -from pipeline.helpers import cli, utils, notifications +from pipeline.helpers import cli, db, utils, notifications MODULE_NAME = "load_openface" @@ -132,6 +132,13 @@ f"Error loading OpenFace features (last interview attempted: " f"{interview_name}): {e}" ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=interview_name or "unknown", + error=e, + identifier_type="interview_name", + ) notifications.send_notification( notify_type="failure", title=f"{MODULE_NAME} failed", diff --git a/pipeline/runners/21_fetch_audio.py b/pipeline/runners/21_fetch_audio.py index cc7b6a8..54165ca 100755 --- a/pipeline/runners/21_fetch_audio.py +++ b/pipeline/runners/21_fetch_audio.py @@ -153,6 +153,13 @@ def on_failure(): f"Decryption request failed. Ignoring file: " f"{file_to_decrypt_path}" ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(file_to_decrypt_path), + error="Decryption request failed; file marked ignored", + identifier_type="file_path", + ) sql_query = InterviewFile.ignore_file(file_to_decrypt_path) db.execute_queries(config_file=config_file, queries=[sql_query]) diff --git a/pipeline/runners/99_wiper.py b/pipeline/runners/99_wiper.py index 08debae..236d277 100755 --- a/pipeline/runners/99_wiper.py +++ b/pipeline/runners/99_wiper.py @@ -200,6 +200,9 @@ def get_interviews_to_wipe(config_file: Path, study_id: str) -> List[str]: config_file=config_file, queries=drop_queries, show_commands=True, + failure_stage=MODULE_NAME, + failure_identifier=interview_to_wipe, + failure_identifier_type="interview_name", ) except Exception as e: logger.error( diff --git a/pipeline/runners/study_specific/23_llm_speaker_identification.py b/pipeline/runners/study_specific/23_llm_speaker_identification.py index 35099f9..806c6c8 100755 --- a/pipeline/runners/study_specific/23_llm_speaker_identification.py +++ b/pipeline/runners/study_specific/23_llm_speaker_identification.py @@ -272,6 +272,13 @@ def process_transcript( f"Error building LLM prompt for {transcript_path} " f"(role {role}): {e}" ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(transcript_path), + error=e, + identifier_type="file_path", + ) subject_result = LlmSpeakerIdentification( llm_source_transcript=transcript_path, ollama_model_identifier="default", diff --git a/pipeline/runners/study_specific/24_llm_language_identification.py b/pipeline/runners/study_specific/24_llm_language_identification.py index db7e992..b6cfe53 100755 --- a/pipeline/runners/study_specific/24_llm_language_identification.py +++ b/pipeline/runners/study_specific/24_llm_language_identification.py @@ -250,6 +250,13 @@ def process_transcript( ) except ValueError as e: logger.error(f"Error building LLM prompt for {transcript_path}: {e}") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(transcript_path), + error=e, + identifier_type="file_path", + ) return LlmLanguageIdentification( llm_source_transcript=transcript_path, ollama_model_identifier="default", @@ -413,6 +420,13 @@ def log_language_identification_result( logger.error( f"Skipping language identification for {file_to_process}: {e}" ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(file_to_process), + error=e, + identifier_type="file_path", + ) continue language_identification_duration = timer.duration diff --git a/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py b/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py index e082480..2e9f03f 100755 --- a/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py +++ b/pipeline/runners/study_specific/ampscz/transcription/25_transcribeme_pull.py @@ -342,6 +342,13 @@ def get_match( f"SSH connection error while pulling transcript " f"{transcript_path}: {e}" ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(transcript_path), + error=e, + identifier_type="file_path", + ) logger.info(f"Retrying in {retry_timeout} seconds...") time.sleep(retry_timeout) diff --git a/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py b/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py index 8a91579..830be6a 100755 --- a/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py +++ b/pipeline/runners/study_specific/ampscz/transcription/interviews/22_transcribeme_push_interview_audio.py @@ -159,6 +159,14 @@ def get_interview_session_number( f"{interview_type}) not found among its own study's interviews " f"list; cannot compute a session number." ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=interview_name, + error="Interview not found in its own study's interview list; " + "cannot compute a session number", + identifier_type="interview_name", + ) return None From df862a3b086a000e0e897162bd0db655095f243f Mon Sep 17 00:00:00 2001 From: Ryan McArdle Date: Tue, 7 Jul 2026 11:57:27 -0400 Subject: [PATCH 05/15] Additional comments/concerns for discussion --- pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py | 1 + .../study_specific/ampscz/5_import_journal_transcripts.py | 1 + pipeline/models/interview_parts.py | 2 ++ 3 files changed, 4 insertions(+) diff --git a/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py b/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py index d75975e..4ac2818 100755 --- a/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py +++ b/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py @@ -67,6 +67,7 @@ def get_interview_name_from_transcript(transcript_filename: str) -> str: # make all day as positive if day < 0: day = -day + # Could this cause name clashes? eg day -5 and day 5 end up the same interview_name = f"{study_id}-{subject_id}-{data_type}Interview-day{day:04d}" diff --git a/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py b/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py index 236da3c..10d6bb0 100755 --- a/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py +++ b/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py @@ -78,6 +78,7 @@ def get_diary_name_from_transcript(transcript_filename: str) -> str: # make all session as positive if session < 0: session = -session + # the same concern from import_transcripts; possible name collisions? journal_name = f"{study_id}-{subject_id}-{data_type}-day{day:04d}-session{session:04d}" diff --git a/pipeline/models/interview_parts.py b/pipeline/models/interview_parts.py index 00db61b..997d2ef 100644 --- a/pipeline/models/interview_parts.py +++ b/pipeline/models/interview_parts.py @@ -96,6 +96,8 @@ def to_sql(self): {self.is_primary}, {self.is_duplicate} ) ON CONFLICT (interview_path) DO NOTHING; """ + # On conflict, we do not override (does not match pattern for subjects) + # Could result in stale interview parts data in db? return sql_query From 8a91a184632962a0b3cc097c77c7100624f16897 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 11:11:34 -0400 Subject: [PATCH 06/15] Separated failure ledger schema and additional logging in crawlers --- .../ampscz/1_import_study_metadata.py | 32 ++++++++++++++--- .../ampscz/6_import_data_dictionary.py | 35 ++++++++++++------- pipeline/helpers/db.py | 2 +- pipeline/helpers/utils.py | 2 ++ pipeline/models/pipeline_failures.py | 28 ++++++++++----- pyproject.toml | 1 + uv.lock | 27 ++++++++++++-- 7 files changed, 97 insertions(+), 30 deletions(-) diff --git a/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py b/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py index 548e979..8af9908 100755 --- a/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py +++ b/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py @@ -67,7 +67,13 @@ def insert_study(config_file: Path, study_id: str) -> None: logger.info(f"Inserting study: {study_id}") query = study.to_sql() - db.execute_queries(config_file=config_file, queries=[query]) + db.execute_queries( + config_file=config_file, + queries=[query], + failure_stage=MODULE_NAME, + failure_identifier=study_id, + failure_identifier_type="study", + ) def get_study_metadata(config_file: Path, study_id: str) -> pd.DataFrame: @@ -95,8 +101,16 @@ def get_study_metadata(config_file: Path, study_id: str) -> pd.DataFrame: # Check if study_metadata exists if not study_metadata.exists(): + error = FileNotFoundError(f"could not read file: {study_metadata}") logger.error(f'Study metadata file "{study_metadata}" not found.') - raise FileNotFoundError(f"could not read file: {study_metadata}") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=study_id, + error=error, + identifier_type="study", + ) + raise error else: insert_study(config_file=config_file, study_id=study_id) @@ -162,18 +176,26 @@ def fetch_subjects(config_file: Path, study_id: str) -> List[Subject]: return subjects -def insert_subjects(config_file: Path, subjects: List[Subject]): +def insert_subjects(config_file: Path, subjects: List[Subject], study_id: str): """ Inserts the subjects into the database. Args: config_file (Path): The path to the configuration file. subjects (List[Subject]): The list of subjects to insert. + study_id (str): The ID of the study the subjects belong to. """ queries = [subject.to_sql() for subject in subjects] - db.execute_queries(config_file=config_file, queries=queries, show_commands=False) + db.execute_queries( + config_file=config_file, + queries=queries, + show_commands=False, + failure_stage=MODULE_NAME, + failure_identifier=study_id, + failure_identifier_type="study", + ) if __name__ == "__main__": @@ -210,6 +232,6 @@ def insert_subjects(config_file: Path, subjects: List[Subject]): for study_id in studies: logger.info(f"Study ID: {study_id}") subjects = fetch_subjects(config_file=config_file, study_id=study_id) - insert_subjects(config_file=config_file, subjects=subjects) + insert_subjects(config_file=config_file, subjects=subjects, study_id=study_id) logger.info("[bold green]Done!", extra={"markup": True}) diff --git a/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py b/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py index d86e11e..64cb6e8 100755 --- a/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py +++ b/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py @@ -73,17 +73,28 @@ def remove_html_tags(input_string: str) -> str: logger.info(f"Reading updated data dictionary from {updated_data_dictionary_path}") - data_dictionary = pd.read_csv(updated_data_dictionary_path) - - # Remove HTML tags from all columns - for col in data_dictionary.columns: - data_dictionary[col] = data_dictionary[col].apply(remove_html_tags) - - db.df_to_table( - config_file=config_file, - df=data_dictionary, - table_name="data_dictionary", - if_exists="replace", - ) + try: + data_dictionary = pd.read_csv(updated_data_dictionary_path) + + # Remove HTML tags from all columns + for col in data_dictionary.columns: + data_dictionary[col] = data_dictionary[col].apply(remove_html_tags) + + db.df_to_table( + config_file=config_file, + df=data_dictionary, + table_name="data_dictionary", + if_exists="replace", + ) + except Exception as e: + logger.error(f"Error importing data dictionary: {e}") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(updated_data_dictionary_path), + error=e, + identifier_type="file_path", + ) + sys.exit(1) logger.info("Data dictionary imported successfully") diff --git a/pipeline/helpers/db.py b/pipeline/helpers/db.py index e85d6eb..9671cf7 100644 --- a/pipeline/helpers/db.py +++ b/pipeline/helpers/db.py @@ -372,7 +372,7 @@ def resolve_failure( ) query = f""" - UPDATE pipeline_failures + UPDATE pipeline_ledger.pipeline_failures SET pf_resolved = TRUE, pf_resolved_at = CURRENT_TIMESTAMP {note_clause} diff --git a/pipeline/helpers/utils.py b/pipeline/helpers/utils.py index df2e262..79de10e 100644 --- a/pipeline/helpers/utils.py +++ b/pipeline/helpers/utils.py @@ -93,6 +93,8 @@ def configure_logging(config_file: Path, module_name: str, logger: logging.Logge archive_file.parent.mkdir(parents=True, exist_ok=True) log_file.rename(archive_file) + if not log_file.parent.exists(): + log_file.parent.mkdir(parents=True, exist_ok=True) file_handler = logging.FileHandler(log_file, mode="a") file_handler.setLevel(logging.DEBUG) file_handler.setFormatter( diff --git a/pipeline/models/pipeline_failures.py b/pipeline/models/pipeline_failures.py index 7bf865f..1edc74c 100644 --- a/pipeline/models/pipeline_failures.py +++ b/pipeline/models/pipeline_failures.py @@ -37,10 +37,15 @@ "file_path", "study", "interview_name", "subject", "batch", "other" ] +# Kept out of 'public' so a ledger row can never collide with (or be mistaken +# for) an application table, and so it can be permissioned/retained separately. +SCHEMA_NAME = "pipeline_ledger" +TABLE_NAME = f"{SCHEMA_NAME}.pipeline_failures" + class PipelineFailure: """ - Represents a row in the 'pipeline_failures' table. + Represents a row in the 'pipeline_ledger.pipeline_failures' table. Attributes: stage (str): The pipeline stage/module the failure occurred in @@ -75,10 +80,13 @@ def __repr__(self) -> str: @staticmethod def init_table_query() -> str: """ - Return the SQL query to create the 'pipeline_failures' table. + Return the SQL query to create the 'pipeline_ledger' schema and, within + it, the 'pipeline_failures' table. """ - sql_query = """ - CREATE TABLE IF NOT EXISTS pipeline_failures ( + sql_query = f""" + CREATE SCHEMA IF NOT EXISTS {SCHEMA_NAME}; + + CREATE TABLE IF NOT EXISTS {TABLE_NAME} ( pf_id SERIAL PRIMARY KEY, pf_stage TEXT NOT NULL, pf_identifier_type TEXT NOT NULL, @@ -100,10 +108,12 @@ def init_table_query() -> str: @staticmethod def drop_table_query() -> str: """ - Return the SQL query to drop the 'pipeline_failures' table. + Return the SQL query to drop the 'pipeline_failures' table. Leaves the + 'pipeline_ledger' schema itself in place (idempotent CREATE SCHEMA IF + NOT EXISTS on re-init doesn't need it gone first). """ - sql_query = """ - DROP TABLE IF EXISTS pipeline_failures; + sql_query = f""" + DROP TABLE IF EXISTS {TABLE_NAME}; """ return sql_query @@ -129,7 +139,7 @@ def to_sql(self) -> str: ) sql_query = f""" - INSERT INTO pipeline_failures ( + INSERT INTO {TABLE_NAME} ( pf_stage, pf_identifier_type, pf_identifier, pf_error, pf_error_type ) VALUES ( '{stage}', '{identifier_type}', '{identifier}', '{error}', {error_type_sql} @@ -137,7 +147,7 @@ def to_sql(self) -> str: pf_identifier_type = EXCLUDED.pf_identifier_type, pf_error = EXCLUDED.pf_error, pf_error_type = EXCLUDED.pf_error_type, - pf_occurrence_count = pipeline_failures.pf_occurrence_count + 1, + pf_occurrence_count = {TABLE_NAME}.pf_occurrence_count + 1, pf_last_seen_at = CURRENT_TIMESTAMP, pf_resolved = FALSE, pf_resolved_at = NULL; diff --git a/pyproject.toml b/pyproject.toml index f085582..8fd1441 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "flask-wtf>=1.2.2", "opencv-python-headless>=4.12.0.88", "pandas>=2.3.3", + "pandas-stubs~=2.3.3", "psycopg2-binary>=2.9.11", "pydantic>=2.12.4", "rich>=14.2.0", diff --git a/uv.lock b/uv.lock index 0956386..36d287a 100644 --- a/uv.lock +++ b/uv.lock @@ -219,6 +219,7 @@ dependencies = [ { name = "flask-wtf" }, { name = "opencv-python-headless" }, { name = "pandas" }, + { name = "pandas-stubs" }, { name = "psycopg2-binary" }, { name = "pydantic" }, { name = "rich" }, @@ -235,6 +236,7 @@ requires-dist = [ { name = "flask-wtf", specifier = ">=1.2.2" }, { name = "opencv-python-headless", specifier = ">=4.12.0.88" }, { name = "pandas", specifier = ">=2.3.3" }, + { name = "pandas-stubs", specifier = "~=2.3.3" }, { name = "psycopg2-binary", specifier = ">=2.9.11" }, { name = "pydantic", specifier = ">=2.12.4" }, { name = "rich", specifier = ">=14.2.0" }, @@ -294,7 +296,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/44/69/9b804adb5fd0671f367781560eb5eb586c4d495277c93bde4307b9e28068/greenlet-3.2.4-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:3b67ca49f54cede0186854a008109d6ee71f66bd57bb36abd6d0a0267b540cdd", size = 274079, upload-time = "2025-08-07T13:15:45.033Z" }, { url = "https://files.pythonhosted.org/packages/46/e9/d2a80c99f19a153eff70bc451ab78615583b8dac0754cfb942223d2c1a0d/greenlet-3.2.4-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:ddf9164e7a5b08e9d22511526865780a576f19ddd00d62f8a665949327fde8bb", size = 640997, upload-time = "2025-08-07T13:42:56.234Z" }, { url = "https://files.pythonhosted.org/packages/3b/16/035dcfcc48715ccd345f3a93183267167cdd162ad123cd93067d86f27ce4/greenlet-3.2.4-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:f28588772bb5fb869a8eb331374ec06f24a83a9c25bfa1f38b6993afe9c1e968", size = 655185, upload-time = "2025-08-07T13:45:27.624Z" }, - { url = "https://files.pythonhosted.org/packages/31/da/0386695eef69ffae1ad726881571dfe28b41970173947e7c558d9998de0f/greenlet-3.2.4-cp312-cp312-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:5c9320971821a7cb77cfab8d956fa8e39cd07ca44b6070db358ceb7f8797c8c9", size = 649926, upload-time = "2025-08-07T13:53:15.251Z" }, { url = "https://files.pythonhosted.org/packages/68/88/69bf19fd4dc19981928ceacbc5fd4bb6bc2215d53199e367832e98d1d8fe/greenlet-3.2.4-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:c60a6d84229b271d44b70fb6e5fa23781abb5d742af7b808ae3f6efd7c9c60f6", size = 651839, upload-time = "2025-08-07T13:18:30.281Z" }, { url = "https://files.pythonhosted.org/packages/19/0d/6660d55f7373b2ff8152401a83e02084956da23ae58cddbfb0b330978fe9/greenlet-3.2.4-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:3b3812d8d0c9579967815af437d96623f45c0f2ae5f04e366de62a12d83a8fb0", size = 607586, upload-time = "2025-08-07T13:18:28.544Z" }, { url = "https://files.pythonhosted.org/packages/8e/1a/c953fdedd22d81ee4629afbb38d2f9d71e37d23caace44775a3a969147d4/greenlet-3.2.4-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:abbf57b5a870d30c4675928c37278493044d7c14378350b3aa5d484fa65575f0", size = 1123281, upload-time = "2025-08-07T13:42:39.858Z" }, @@ -305,7 +306,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/49/e8/58c7f85958bda41dafea50497cbd59738c5c43dbbea5ee83d651234398f4/greenlet-3.2.4-cp313-cp313-macosx_11_0_universal2.whl", hash = "sha256:1a921e542453fe531144e91e1feedf12e07351b1cf6c9e8a3325ea600a715a31", size = 272814, upload-time = "2025-08-07T13:15:50.011Z" }, { url = "https://files.pythonhosted.org/packages/62/dd/b9f59862e9e257a16e4e610480cfffd29e3fae018a68c2332090b53aac3d/greenlet-3.2.4-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:cd3c8e693bff0fff6ba55f140bf390fa92c994083f838fece0f63be121334945", size = 641073, upload-time = "2025-08-07T13:42:57.23Z" }, { url = "https://files.pythonhosted.org/packages/f7/0b/bc13f787394920b23073ca3b6c4a7a21396301ed75a655bcb47196b50e6e/greenlet-3.2.4-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:710638eb93b1fa52823aa91bf75326f9ecdfd5e0466f00789246a5280f4ba0fc", size = 655191, upload-time = "2025-08-07T13:45:29.752Z" }, - { url = "https://files.pythonhosted.org/packages/f2/d6/6adde57d1345a8d0f14d31e4ab9c23cfe8e2cd39c3baf7674b4b0338d266/greenlet-3.2.4-cp313-cp313-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:c5111ccdc9c88f423426df3fd1811bfc40ed66264d35aa373420a34377efc98a", size = 649516, upload-time = "2025-08-07T13:53:16.314Z" }, { url = "https://files.pythonhosted.org/packages/7f/3b/3a3328a788d4a473889a2d403199932be55b1b0060f4ddd96ee7cdfcad10/greenlet-3.2.4-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:d76383238584e9711e20ebe14db6c88ddcedc1829a9ad31a584389463b5aa504", size = 652169, upload-time = "2025-08-07T13:18:32.861Z" }, { url = "https://files.pythonhosted.org/packages/ee/43/3cecdc0349359e1a527cbf2e3e28e5f8f06d3343aaf82ca13437a9aa290f/greenlet-3.2.4-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:23768528f2911bcd7e475210822ffb5254ed10d71f4028387e5a99b4c6699671", size = 610497, upload-time = "2025-08-07T13:18:31.636Z" }, { url = "https://files.pythonhosted.org/packages/b8/19/06b6cf5d604e2c382a6f31cafafd6f33d5dea706f4db7bdab184bad2b21d/greenlet-3.2.4-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:00fadb3fedccc447f517ee0d3fd8fe49eae949e1cd0f6a611818f4f6fb7dc83b", size = 1121662, upload-time = "2025-08-07T13:42:41.117Z" }, @@ -316,7 +316,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/22/5c/85273fd7cc388285632b0498dbbab97596e04b154933dfe0f3e68156c68c/greenlet-3.2.4-cp314-cp314-macosx_11_0_universal2.whl", hash = "sha256:49a30d5fda2507ae77be16479bdb62a660fa51b1eb4928b524975b3bde77b3c0", size = 273586, upload-time = "2025-08-07T13:16:08.004Z" }, { url = "https://files.pythonhosted.org/packages/d1/75/10aeeaa3da9332c2e761e4c50d4c3556c21113ee3f0afa2cf5769946f7a3/greenlet-3.2.4-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:299fd615cd8fc86267b47597123e3f43ad79c9d8a22bebdce535e53550763e2f", size = 686346, upload-time = "2025-08-07T13:42:59.944Z" }, { url = "https://files.pythonhosted.org/packages/c0/aa/687d6b12ffb505a4447567d1f3abea23bd20e73a5bed63871178e0831b7a/greenlet-3.2.4-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:c17b6b34111ea72fc5a4e4beec9711d2226285f0386ea83477cbb97c30a3f3a5", size = 699218, upload-time = "2025-08-07T13:45:30.969Z" }, - { url = "https://files.pythonhosted.org/packages/dc/8b/29aae55436521f1d6f8ff4e12fb676f3400de7fcf27fccd1d4d17fd8fecd/greenlet-3.2.4-cp314-cp314-manylinux2014_s390x.manylinux_2_17_s390x.whl", hash = "sha256:b4a1870c51720687af7fa3e7cda6d08d801dae660f75a76f3845b642b4da6ee1", size = 694659, upload-time = "2025-08-07T13:53:17.759Z" }, { url = "https://files.pythonhosted.org/packages/92/2e/ea25914b1ebfde93b6fc4ff46d6864564fba59024e928bdc7de475affc25/greenlet-3.2.4-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:061dc4cf2c34852b052a8620d40f36324554bc192be474b9e9770e8c042fd735", size = 695355, upload-time = "2025-08-07T13:18:34.517Z" }, { url = "https://files.pythonhosted.org/packages/72/60/fc56c62046ec17f6b0d3060564562c64c862948c9d4bc8aa807cf5bd74f4/greenlet-3.2.4-cp314-cp314-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:44358b9bf66c8576a9f57a590d5f5d6e72fa4228b763d0e43fee6d3b06d3a337", size = 657512, upload-time = "2025-08-07T13:18:33.969Z" }, { url = "https://files.pythonhosted.org/packages/23/6e/74407aed965a4ab6ddd93a7ded3180b730d281c77b765788419484cdfeef/greenlet-3.2.4-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:2917bdf657f5859fbf3386b12d68ede4cf1f04c90c3a6bc1f013dd68a22e2269", size = 1612508, upload-time = "2025-11-04T12:42:23.427Z" }, @@ -558,6 +557,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/70/44/5191d2e4026f86a2a109053e194d3ba7a31a2d10a9c2348368c63ed4e85a/pandas-2.3.3-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:3869faf4bd07b3b66a9f462417d0ca3a9df29a9f6abd5d0d0dbab15dac7abe87", size = 13202175, upload-time = "2025-09-29T23:31:59.173Z" }, ] +[[package]] +name = "pandas-stubs" +version = "2.3.3.260113" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy" }, + { name = "types-pytz" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/92/5d/be23854a73fda69f1dbdda7bc10fbd6f930bd1fa87aaec389f00c901c1e8/pandas_stubs-2.3.3.260113.tar.gz", hash = "sha256:076e3724bcaa73de78932b012ec64b3010463d377fa63116f4e6850643d93800", size = 116131, upload-time = "2026-01-13T22:30:16.704Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/c6/df1fe324248424f77b89371116dab5243db7f052c32cc9fe7442ad9c5f75/pandas_stubs-2.3.3.260113-py3-none-any.whl", hash = "sha256:ec070b5c576e1badf12544ae50385872f0631fc35d99d00dc598c2954ec564d3", size = 168246, upload-time = "2026-01-13T22:30:15.244Z" }, +] + [[package]] name = "psycopg2-binary" version = "2.9.11" @@ -840,6 +852,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9c/5e/6a29fa884d9fb7ddadf6b69490a9d45fded3b38541713010dad16b77d015/sqlalchemy-2.0.44-py3-none-any.whl", hash = "sha256:19de7ca1246fbef9f9d1bff8f1ab25641569df226364a0e40457dc5457c54b05", size = 1928718, upload-time = "2025-10-10T15:29:45.32Z" }, ] +[[package]] +name = "types-pytz" +version = "2026.2.0.20260518" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9f/d9/9fa4019d2235bd374293e1fd4153879b28b6ae1d2bae98addd352c9713f2/types_pytz-2026.2.0.20260518.tar.gz", hash = "sha256:e5d254329e9c4e91f0781b22c43a4bb2d10bb044d97b24c4b05d45567b0eae16", size = 10871, upload-time = "2026-05-18T06:02:45.789Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/62/89/41e80670779a223d8bc8bc83019a619988cfa5c432cedac5cec23884fbc4/types_pytz-2026.2.0.20260518-py3-none-any.whl", hash = "sha256:3a12eaa38f476bd650902a9c9bb442f03f3c7dee2be5c5848bce61bd708d205a", size = 10125, upload-time = "2026-05-18T06:02:44.968Z" }, +] + [[package]] name = "typing-extensions" version = "4.15.0" From 8dd9599ff055bc6a9d054aee6c5eae6c8d413a7f Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 11:37:54 -0400 Subject: [PATCH 07/15] Check for active subjects and record genuine failures --- pipeline/core/__init__.py | 26 ++++++ .../ampscz/2_import_interview_files.py | 89 ++++++++++++++++--- 2 files changed, 102 insertions(+), 13 deletions(-) diff --git a/pipeline/core/__init__.py b/pipeline/core/__init__.py index 8da5f22..0036b2b 100644 --- a/pipeline/core/__init__.py +++ b/pipeline/core/__init__.py @@ -71,6 +71,32 @@ def get_subject_ids(config_file: Path, study_id: str) -> List[str]: return subject_ids +def get_subject_active_status(config_file: Path, study_id: str) -> Dict[str, bool]: + """ + Gets the is_active flag for every subject in a study, in a single query. + + Each DB call opens (and disposes of) its own connection, so callers that + need this per-subject should fetch the whole study's status map once + up front and look up subject_id in it, rather than querying per subject. + + Args: + config_file (Path): The path to the configuration file. + study_id (str): The study ID. + + Returns: + Dict[str, bool]: Mapping of subject_id to is_active. + """ + query = f""" + SELECT subject_id, is_active + FROM subjects + WHERE study_id = '{study_id}'; + """ + + results = db.execute_sql(config_file=config_file, query=query) + + return dict(zip(results["subject_id"], results["is_active"])) + + def get_all_cols(csv_file: Path) -> List[str]: """ Returns a list of all column names in a CSV file. 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 b553277..565e04f 100755 --- a/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py +++ b/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py @@ -32,7 +32,7 @@ import multiprocessing import re from datetime import date, datetime, time -from typing import Dict, List, Tuple +from typing import Dict, List, Optional, Tuple from rich.logging import RichHandler from rich.progress import Progress @@ -141,12 +141,15 @@ def catogorize_audio_files( return files -def fetch_interview_files(interview_part: InterviewParts) -> List[InterviewFile]: +def fetch_interview_files( + interview_part: InterviewParts, config_file: Path +) -> List[InterviewFile]: """ Fetches the interview files for a given interview. Args: interview (Interview): The interview object. + config_file (Path): The path to the configuration file. Returns: List[InterviewFile]: A list of InterviewFile objects. @@ -159,10 +162,20 @@ def fetch_interview_files(interview_part: InterviewParts) -> List[InterviewFile] ) subject_id = dp_dash_dict["subject"] if not isinstance(subject_id, str): + error = ValueError( + f"Could not parse subject ID from {interview_part.interview_name}" + ) logger.error(f"Could not parse subject ID from {interview_part.interview_name}") logger.error(f"dp_dash_dict: {dp_dash_dict}") logger.error(f"subject_id: {subject_id} - {type(subject_id)}") - raise ValueError(f"Could not parse subject ID from {interview_part.interview_name}") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=interview_part.interview_name, + error=error, + identifier_type="interview_name", + ) + raise error interview_path = interview_part.interview_path if interview_path.is_file(): @@ -262,7 +275,7 @@ def handle_multi_part_interviews( def fetch_interviews( - config_file: Path, subject_id: str, study_id: str + config_file: Path, subject_id: str, study_id: str, is_active: Optional[bool] ) -> List[InterviewParts]: """ Fetches the interviews for a given subject ID. @@ -270,10 +283,17 @@ def fetch_interviews( Args: config_file (Path): The path to the config file. subject_id (str): The subject ID. + is_active (Optional[bool]): Whether the subject is active, from + core.get_subject_active_status() - fetched once per study by the + caller rather than queried per subject here. Returns: List[Interview]: A list of Interview objects. """ + if is_active is not None and not is_active: + logger.debug(f"{subject_id}: subject is not active, skipping interview fetch.") + return [] + config_params = config(path=config_file, section="general") data_root = Path(config_params["data_root"]) @@ -287,7 +307,12 @@ def fetch_interviews( ) if not interview_type_path.exists(): - logger.warning( + # Routine: an active subject just hasn't had this interview type + # yet. Not warning-worthy - the genuinely actionable version of + # this ("runsheet says it happened, we don't have it") is already + # surfaced by dpinterview-web's Missing issues dashboard, which + # compares expected_interviews against what actually got imported. + logger.debug( f"{subject_id}: Could not find {interview_type.value} interviews: \ {interview_type_path} does not exist." ) @@ -308,17 +333,32 @@ def fetch_interviews( time_dt = time.fromisoformat("00:00:00") interview_datetime = datetime.combine(date_dt, time_dt) actual_interview_datetime = datetime.combine(date_dt, actual_time_dt) - except (ValueError, IndexError): - logger.warning( + except (ValueError, IndexError) as e: + 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, + identifier=str(interview_dir), + error=e, + identifier_type="file_path", + ) continue consent_date_s = core.get_consent_date_from_subject_id( config_file=config_file, subject_id=subject_id, study_id=study_id ) if consent_date_s is None: - logger.warning(f"Could not find consent date for {subject_id}") + error = ValueError(f"Could not find consent date for {subject_id}") + logger.error(str(error)) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=subject_id, + error=error, + identifier_type="subject", + ) continue consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") @@ -358,10 +398,17 @@ def fetch_interviews( interview_datetime = actual_interview_datetime.replace( hour=0, minute=0, second=0, microsecond=0 ) - except ValueError: - logger.warning( + except ValueError as e: + logger.error( f"Could not parse date and time from {wav_file}. Skipping..." ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=str(wav_file), + error=e, + identifier_type="file_path", + ) continue consent_date_s = core.get_consent_date_from_subject_id( @@ -369,7 +416,15 @@ def fetch_interviews( ) if consent_date_s is None: - logger.warning(f"Could not find consent date for {subject_id}") + error = ValueError(f"Could not find consent date for {subject_id}") + logger.error(str(error)) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + identifier=subject_id, + error=error, + identifier_type="subject", + ) continue consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") @@ -493,6 +548,9 @@ def import_interviews(config_file: Path, study_id: str, progress: Progress) -> N # Get the subjects subjects = core.get_subject_ids(config_file=config_file, study_id=study_id) + subject_active_status = core.get_subject_active_status( + config_file=config_file, study_id=study_id + ) # Get the interviews logger.info(f"Fetching interviews for {study_id}") @@ -505,7 +563,10 @@ def import_interviews(config_file: Path, study_id: str, progress: Progress) -> N ) interview_parts.extend( fetch_interviews( - config_file=config_file, subject_id=subject_id, study_id=study_id + config_file=config_file, + subject_id=subject_id, + study_id=study_id, + is_active=subject_active_status.get(subject_id), ) ) progress.remove_task(task) @@ -519,7 +580,9 @@ def import_interviews(config_file: Path, study_id: str, progress: Progress) -> N for interview_part in interview_parts: interview_counter += 1 progress.update(task, advance=1) - interview_files.extend(fetch_interview_files(interview_part=interview_part)) + interview_files.extend( + fetch_interview_files(interview_part=interview_part, config_file=config_file) + ) progress.remove_task(task) # Generate the SQL queries to import the interview files From f9653d1692b728ba64e524c3e8194eefcc52ab2b Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 11:50:16 -0400 Subject: [PATCH 08/15] Config update --- pipeline/models/pipeline_failures.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/pipeline/models/pipeline_failures.py b/pipeline/models/pipeline_failures.py index 1edc74c..4777fb1 100644 --- a/pipeline/models/pipeline_failures.py +++ b/pipeline/models/pipeline_failures.py @@ -27,6 +27,7 @@ pass +import argparse from typing import Literal, Optional from pipeline.helpers import cli, db, utils @@ -157,7 +158,23 @@ def to_sql(self) -> str: if __name__ == "__main__": - config_file = utils.get_config_file_path() + parser = argparse.ArgumentParser( + prog="pipeline_failures", + description="Initialize the 'pipeline_ledger.pipeline_failures' 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 'pipeline_failures' table...") From 69e22e77ea39c79bfd48a7f7bc6e68930cbb6fa9 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 12:25:09 -0400 Subject: [PATCH 09/15] Improved failure ledger schema --- .../ampscz/1_import_study_metadata.py | 1 + .../ampscz/2_import_interview_files.py | 19 ++++- .../ampscz/3_import_transcripts.py | 7 +- .../ampscz/5_import_journal_transcripts.py | 7 +- .../ampscz/6_import_data_dictionary.py | 1 + pipeline/helpers/db.py | 38 +++++++++ pipeline/models/pipeline_failures.py | 84 +++++++++++++++++-- 7 files changed, 146 insertions(+), 11 deletions(-) diff --git a/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py b/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py index 8af9908..421bd30 100755 --- a/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py +++ b/pipeline/crawlers/study_specific/ampscz/1_import_study_metadata.py @@ -106,6 +106,7 @@ def get_study_metadata(config_file: Path, study_id: str) -> pd.DataFrame: db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="missing_file", identifier=study_id, error=error, identifier_type="study", 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 565e04f..4a23c7c 100755 --- a/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py +++ b/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py @@ -142,7 +142,7 @@ def catogorize_audio_files( def fetch_interview_files( - interview_part: InterviewParts, config_file: Path + interview_part: InterviewParts, config_file: Path, study_id: str ) -> List[InterviewFile]: """ Fetches the interview files for a given interview. @@ -150,6 +150,7 @@ def fetch_interview_files( Args: interview (Interview): The interview object. config_file (Path): The path to the configuration file. + study_id (str): The study ID, for ledger context if parsing fails. Returns: List[InterviewFile]: A list of InterviewFile objects. @@ -171,9 +172,11 @@ def fetch_interview_files( db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="subject_id_parse", identifier=interview_part.interview_name, error=error, identifier_type="interview_name", + study_id=study_id, ) raise error interview_path = interview_part.interview_path @@ -340,9 +343,12 @@ def fetch_interviews( 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 @@ -355,9 +361,11 @@ def fetch_interviews( db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="consent_date_missing", identifier=subject_id, error=error, identifier_type="subject", + study_id=study_id, ) continue consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") @@ -405,9 +413,12 @@ def fetch_interviews( 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 @@ -421,9 +432,11 @@ def fetch_interviews( db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="consent_date_missing", identifier=subject_id, error=error, identifier_type="subject", + study_id=study_id, ) continue consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") @@ -581,7 +594,9 @@ def import_interviews(config_file: Path, study_id: str, progress: Progress) -> N interview_counter += 1 progress.update(task, advance=1) interview_files.extend( - fetch_interview_files(interview_part=interview_part, config_file=config_file) + fetch_interview_files( + interview_part=interview_part, config_file=config_file, study_id=study_id + ) ) progress.remove_task(task) diff --git a/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py b/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py index 4ac2818..265640a 100755 --- a/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py +++ b/pipeline/crawlers/study_specific/ampscz/3_import_transcripts.py @@ -75,13 +75,14 @@ def get_interview_name_from_transcript(transcript_filename: str) -> str: def transcripts_to_models( - transcripts: List[Path], config_file: Path + transcripts: List[Path], config_file: Path, study_id: str ) -> Tuple[List[File], List[TranscriptFile]]: """ Converts the transcripts into File and InterviewFile models. Args: transcripts (List[Path]): The list of transcripts. + study_id (str): The study ID, for ledger context if parsing fails. Returns: Tuple[List[File], List[TranscriptFile]]: The list of File and InterviewFile models. @@ -102,9 +103,11 @@ def transcripts_to_models( db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="filename_parse", identifier=str(transcript), error=e, identifier_type="file_path", + study_id=study_id, ) continue @@ -185,7 +188,7 @@ def import_transcripts(data_root: Path, study: str, config_file: Path) -> None: logger.info(f"Found {len(transcripts)} transcripts.") files, transcript_files = transcripts_to_models( - transcripts=transcripts, config_file=config_file + transcripts=transcripts, config_file=config_file, study_id=study ) logger.info( diff --git a/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py b/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py index 10d6bb0..4720869 100755 --- a/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py +++ b/pipeline/crawlers/study_specific/ampscz/5_import_journal_transcripts.py @@ -86,7 +86,7 @@ def get_diary_name_from_transcript(transcript_filename: str) -> str: def transcripts_to_models( - transcripts: List[Path], config_file: Path + transcripts: List[Path], config_file: Path, study_id: str ) -> Tuple[List[File], List[TranscriptFile]]: """ Converts the transcripts into File and InterviewFile models. @@ -94,6 +94,7 @@ def transcripts_to_models( Args: transcripts (List[Path]): The list of transcripts. config_file (Path): The path to the config file. + study_id (str): The study ID, for ledger context if parsing fails. Returns: Tuple[List[File], List[TranscriptFile]]: The list of File and InterviewFile models. @@ -116,9 +117,11 @@ def transcripts_to_models( db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="filename_parse", identifier=str(transcript), error=e, identifier_type="file_path", + study_id=study_id, ) continue @@ -189,7 +192,7 @@ def import_transcripts(data_root: Path, study: str, config_file: Path) -> None: logger.info(f"Found {len(transcripts)} transcripts.") files, transcript_files = transcripts_to_models( - transcripts=transcripts, config_file=config_file + transcripts=transcripts, config_file=config_file, study_id=study ) logger.info( diff --git a/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py b/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py index 64cb6e8..4d82c19 100755 --- a/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py +++ b/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py @@ -91,6 +91,7 @@ def remove_html_tags(input_string: str) -> str: db.record_failure( config_file=config_file, stage=MODULE_NAME, + error_code="data_dictionary_import_failed", identifier=str(updated_data_dictionary_path), error=e, identifier_type="file_path", diff --git a/pipeline/helpers/db.py b/pipeline/helpers/db.py index 9671cf7..3c3d03c 100644 --- a/pipeline/helpers/db.py +++ b/pipeline/helpers/db.py @@ -133,8 +133,11 @@ def execute_queries( backup: bool = False, on_failure: Optional[Callable] = on_failure, failure_stage: Optional[str] = None, + failure_error_code: str = "db_write_failure", failure_identifier: Optional[str] = None, failure_identifier_type: str = "batch", + failure_study_id: Optional[str] = None, + failure_subject_id: Optional[str] = None, ) -> list: """ Executes a list of SQL queries on a PostgreSQL database. @@ -153,9 +156,19 @@ def execute_queries( failure_stage (str, optional): If set (together with failure_identifier), a failure will also be recorded in the pipeline_failures ledger via record_failure(). Defaults to None (no ledger entry). + failure_error_code (str, optional): A short, stable code for why this + batch failed - see PipelineFailure's ErrorCode. Defaults to + "db_write_failure", the right default for the common case of "this + SQL batch raised an exception". failure_identifier (str, optional): What failed - see failure_stage. failure_identifier_type (str, optional): The kind of thing failure_identifier is (e.g. "study", "file_path"). Defaults to "batch". + failure_study_id (str, optional): The study this batch was for, if + known, for ledger filtering/reporting. Auto-filled from + failure_identifier when failure_identifier_type == "study". + failure_subject_id (str, optional): The subject this batch was for, if + known, for ledger filtering/reporting. Auto-filled from + failure_identifier when failure_identifier_type == "subject". Returns: list: A list of tuples containing the results of the executed queries. @@ -266,8 +279,11 @@ def execute_query(query: str): record_failure( config_file=config_file, stage=failure_stage, + error_code=failure_error_code, identifier=failure_identifier, identifier_type=failure_identifier_type, + study_id=failure_study_id, + subject_id=failure_subject_id, error=e, db=db, ) @@ -285,11 +301,14 @@ def execute_query(query: str): def record_failure( config_file: Path, stage: str, + error_code: str, identifier: str, error: Union[str, Exception], identifier_type: Literal[ "file_path", "study", "interview_name", "subject", "batch", "other" ] = "file_path", + study_id: Optional[str] = None, + subject_id: Optional[str] = None, db: str = "postgresql", ) -> None: """ @@ -303,12 +322,23 @@ def record_failure( Args: config_file (Path): The path to the configuration file. stage (str): The pipeline stage/module the failure occurred in. + error_code (str): A short, stable code for *why* this failed (e.g. + "datetime_parse"), shared across every occurrence of the same kind + of failure regardless of identifier or exact message - see + pipeline.models.pipeline_failures.ErrorCode for the known set. identifier (str): What failed - a file path when known, otherwise the most specific thing available (study_id, a batch description, etc). error (Union[str, Exception]): The error. If an Exception is passed, its class name is recorded as the error type. identifier_type (str, optional): The kind of thing `identifier` is. Defaults to "file_path". + study_id (str, optional): The study this failure occurred in, if known, + for ledger filtering/reporting. Auto-filled from `identifier` when + identifier_type == "study" and this isn't passed explicitly. + subject_id (str, optional): The subject this failure relates to, if + known, for ledger filtering/reporting. Auto-filled from + `identifier` when identifier_type == "subject" and this isn't + passed explicitly. db (str, optional): The section of the configuration file to use. Defaults to "postgresql". """ @@ -318,11 +348,19 @@ def record_failure( # be circular. from pipeline.models.pipeline_failures import PipelineFailure + if study_id is None and identifier_type == "study": + study_id = identifier + if subject_id is None and identifier_type == "subject": + subject_id = identifier + error_type = type(error).__name__ if isinstance(error, Exception) else None failure = PipelineFailure( stage=stage, + error_code=error_code, identifier=identifier, identifier_type=identifier_type, + study_id=study_id, + subject_id=subject_id, error=str(error), error_type=error_type, ) diff --git a/pipeline/models/pipeline_failures.py b/pipeline/models/pipeline_failures.py index 4777fb1..42b996f 100644 --- a/pipeline/models/pipeline_failures.py +++ b/pipeline/models/pipeline_failures.py @@ -38,6 +38,31 @@ "file_path", "study", "interview_name", "subject", "batch", "other" ] +# A short, stable code for *why* something failed, independent of the specific +# file/subject/message involved - e.g. every unparseable date string becomes +# "datetime_parse" instead of a one-off "invalid isoformat string: 'XYZ'" that +# can't be grouped with any other row. This is what reports/dashboards should +# group and filter by. Add new codes here as new failure sites get wired up, +# so call sites stay consistent instead of inventing near-duplicate strings. +ErrorCode = Literal[ + "datetime_parse", + "subject_id_parse", + "filename_parse", + "consent_date_missing", + "missing_file", + "db_write_failure", + "data_dictionary_import_failed", + "ffprobe_streams_missing", + "openface_datatype_cast_failed", + "openface_load_failed", + "decryption_failed", + "llm_prompt_build_failed", + "llm_language_identification_failed", + "transcribeme_pull_failed", + "interview_not_in_study_list", + "other", +] + # Kept out of 'public' so a ledger row can never collide with (or be mistaken # for) an application table, and so it can be permissioned/retained separately. SCHEMA_NAME = "pipeline_ledger" @@ -51,29 +76,43 @@ class PipelineFailure: Attributes: stage (str): The pipeline stage/module the failure occurred in (e.g. matches the module_name used for [logging] config sections). + error_code (ErrorCode): A short, stable code identifying *why* this + failed (e.g. "datetime_parse"), for grouping/reporting across + occurrences that have different identifiers or error messages. identifier (str): What failed - a file path when known, otherwise the most specific thing available (study_id, a batch description, etc). error (str): The error message. identifier_type (IdentifierType): What kind of thing `identifier` is. + study_id (Optional[str]): The study the failure occurred in, if known. + subject_id (Optional[str]): The subject the failure relates to, if known. error_type (Optional[str]): The exception class name, if known. """ def __init__( self, stage: str, + error_code: ErrorCode, identifier: str, error: str, identifier_type: IdentifierType = "file_path", + study_id: Optional[str] = None, + subject_id: Optional[str] = None, error_type: Optional[str] = None, ) -> None: self.stage = stage + self.error_code = error_code self.identifier = identifier self.identifier_type = identifier_type + self.study_id = study_id + self.subject_id = subject_id self.error = error self.error_type = error_type def __str__(self) -> str: - return f"PipelineFailure({self.stage}, {self.identifier}, {self.error})" + return ( + f"PipelineFailure({self.stage}, {self.error_code}, " + f"{self.identifier}, {self.error})" + ) def __repr__(self) -> str: return self.__str__() @@ -81,8 +120,12 @@ def __repr__(self) -> str: @staticmethod def init_table_query() -> str: """ - Return the SQL query to create the 'pipeline_ledger' schema and, within - it, the 'pipeline_failures' table. + Return the SQL to create the 'pipeline_ledger' schema and, within it, + the 'pipeline_failures' table - and to bring an already-existing table + (from before pf_error_code/pf_study_id/pf_subject_id existed) up to + the current shape via ADD COLUMN IF NOT EXISTS, so re-running this + against an already-initialized deployment upgrades it in place instead + of requiring a destructive drop/recreate. """ sql_query = f""" CREATE SCHEMA IF NOT EXISTS {SCHEMA_NAME}; @@ -90,8 +133,11 @@ def init_table_query() -> str: CREATE TABLE IF NOT EXISTS {TABLE_NAME} ( pf_id SERIAL PRIMARY KEY, pf_stage TEXT NOT NULL, + pf_error_code TEXT NOT NULL DEFAULT 'unknown', pf_identifier_type TEXT NOT NULL, pf_identifier TEXT NOT NULL, + pf_study_id TEXT, + pf_subject_id TEXT, pf_error TEXT NOT NULL, pf_error_type TEXT, pf_occurrence_count INTEGER NOT NULL DEFAULT 1, @@ -102,6 +148,18 @@ def init_table_query() -> str: pf_resolved_note TEXT, UNIQUE (pf_stage, pf_identifier) ); + + ALTER TABLE {TABLE_NAME} + ADD COLUMN IF NOT EXISTS pf_error_code TEXT NOT NULL DEFAULT 'unknown'; + ALTER TABLE {TABLE_NAME} ADD COLUMN IF NOT EXISTS pf_study_id TEXT; + ALTER TABLE {TABLE_NAME} ADD COLUMN IF NOT EXISTS pf_subject_id TEXT; + + CREATE INDEX IF NOT EXISTS pipeline_failures_study_id_idx + ON {TABLE_NAME} (pf_study_id); + CREATE INDEX IF NOT EXISTS pipeline_failures_subject_id_idx + ON {TABLE_NAME} (pf_subject_id); + CREATE INDEX IF NOT EXISTS pipeline_failures_error_code_idx + ON {TABLE_NAME} (pf_error_code); """ return sql_query @@ -130,6 +188,7 @@ def to_sql(self) -> str: resolved and then happens again is not resolved anymore. """ stage = db.santize_string(self.stage) + error_code = db.santize_string(self.error_code) identifier_type = db.santize_string(self.identifier_type) identifier = db.santize_string(self.identifier) error = db.santize_string(self.error) @@ -138,14 +197,29 @@ def to_sql(self) -> str: if self.error_type is not None else "NULL" ) + 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} ( - pf_stage, pf_identifier_type, pf_identifier, pf_error, pf_error_type + pf_stage, pf_error_code, pf_identifier_type, pf_identifier, + pf_study_id, pf_subject_id, pf_error, pf_error_type ) VALUES ( - '{stage}', '{identifier_type}', '{identifier}', '{error}', {error_type_sql} + '{stage}', '{error_code}', '{identifier_type}', '{identifier}', + {study_id_sql}, {subject_id_sql}, '{error}', {error_type_sql} ) ON CONFLICT (pf_stage, pf_identifier) DO UPDATE SET + pf_error_code = EXCLUDED.pf_error_code, pf_identifier_type = EXCLUDED.pf_identifier_type, + pf_study_id = EXCLUDED.pf_study_id, + pf_subject_id = EXCLUDED.pf_subject_id, pf_error = EXCLUDED.pf_error, pf_error_type = EXCLUDED.pf_error_type, pf_occurrence_count = {TABLE_NAME}.pf_occurrence_count + 1, From eec4d83ba9e2f713a3f190cdae97ca3eec72fddc Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 12:35:57 -0400 Subject: [PATCH 10/15] Introduce batch crawler --- .../ampscz/run_pronet_crawlers.py | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 pipeline/crawlers/study_specific/ampscz/run_pronet_crawlers.py diff --git a/pipeline/crawlers/study_specific/ampscz/run_pronet_crawlers.py b/pipeline/crawlers/study_specific/ampscz/run_pronet_crawlers.py new file mode 100644 index 0000000..7704b10 --- /dev/null +++ b/pipeline/crawlers/study_specific/ampscz/run_pronet_crawlers.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python +""" +Runs the ProNET AMPSCZ crawlers in sequence against a single config file. + +Each crawler below currently runs as its own independent cron entry. This +just chains the same scripts, in their dependency order (subjects before +interviews/journals before form data before expected interviews), stopping +at the first failure instead of letting downstream stages run against stale +or incomplete data. + +Each stage runs as its own subprocess rather than being imported in-process: +every crawler script does its own module-level logging setup (logging. +basicConfig) and sys.path bootstrap, so importing several of them into one +process risks duplicate/conflicting logging handlers. A fresh process per +stage also matches how these already run under cron today. +""" + +import subprocess +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 +import logging + +from rich.logging import RichHandler + +from pipeline.helpers import db, utils + +MODULE_NAME = "run_pronet_crawlers" + +logger = logging.getLogger(MODULE_NAME) +logargs = { + "level": logging.DEBUG, + "format": "%(message)s", + "handlers": [RichHandler(rich_tracebacks=True)], +} +logging.basicConfig(**logargs) + +console = utils.get_console() + +CRAWLERS_DIR = Path(__file__).resolve().parent + +CRAWLER_SCRIPTS = [ + CRAWLERS_DIR / "1_import_study_metadata.py", + CRAWLERS_DIR / "2_import_interview_files.py", + CRAWLERS_DIR / "3_import_transcripts.py", + CRAWLERS_DIR / "4_import_audio_journals.py", + CRAWLERS_DIR / "5_import_journal_transcripts.py", + CRAWLERS_DIR / "6_import_data_dictionary.py", + CRAWLERS_DIR / "pronet" / "7_import_form_data.py", + CRAWLERS_DIR / "8_import_expected_interviews.py", +] + + +def run_crawler(script_path: Path, config_file: Path) -> int: + """ + Runs a single crawler script as a subprocess against config_file. + + Args: + script_path (Path): The crawler script to run. + config_file (Path): The config file to pass to it via -c. + + Returns: + int: The subprocess's exit code. + """ + logger.info(f"[bold cyan]Running {script_path.name}...", extra={"markup": True}) + + result = subprocess.run( + [sys.executable, str(script_path), "-c", str(config_file)], + check=False, + ) + + return result.returncode + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + prog=MODULE_NAME, + description="Run the ProNET AMPSCZ crawlers in sequence.", + ) + parser.add_argument( + "-c", "--config", type=str, help="Path to the config file.", required=False, default='pronet.ampscz.config.ini' + ) + + args = parser.parse_args() + + if args.config: + config_file = Path(args.config).resolve() + if not config_file.exists(): + logger.error(f"Error: Config file '{config_file}' does not exist.") + sys.exit(1) + else: + config_file = utils.get_config_file_path() + + console.rule(f"[bold red]{MODULE_NAME}") + logger.info(f"Using config file: {config_file}") + + for script_path in CRAWLER_SCRIPTS: + if not script_path.exists(): + logger.error(f"[bold red]Crawler script not found: {script_path}") + sys.exit(1) + + returncode = run_crawler(script_path=script_path, config_file=config_file) + + if returncode != 0: + logger.error( + f"[bold red]{script_path.name} failed (exit code {returncode}); " + f"stopping before running downstream crawlers.", + extra={"markup": True}, + ) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + error_code="crawler_stage_failed", + identifier=script_path.stem, + error=f"{script_path.name} exited with code {returncode}", + identifier_type="batch", + ) + sys.exit(returncode) + + logger.info(f"[bold green]{script_path.name} completed.", extra={"markup": True}) + + logger.info( + "[bold green]All ProNET crawlers completed successfully.", + extra={"markup": True}, + ) From 9551397dad0ff6ef7f6f4179f91f7a5fe8b6c7c5 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 12:36:20 -0400 Subject: [PATCH 11/15] Config for data dictionary --- .../ampscz/6_import_data_dictionary.py | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py b/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py index 4d82c19..c9b5582 100755 --- a/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py +++ b/pipeline/crawlers/study_specific/ampscz/6_import_data_dictionary.py @@ -19,13 +19,14 @@ except ValueError: pass +import argparse import logging import re import pandas as pd from rich.logging import RichHandler -from pipeline.helpers import db, utils +from pipeline.helpers import cli, db, utils MODULE_NAME = "crawlers.import_data_dictionary" @@ -59,9 +60,27 @@ def remove_html_tags(input_string: str) -> str: if __name__ == "__main__": - console.rule(f"[bold red]{MODULE_NAME}") + parser = argparse.ArgumentParser( + prog=MODULE_NAME, description="Import the data dictionary into the database." + ) + parser.add_argument( + "-c", "--config", type=str, help="Path to the config file.", required=False + ) + + args = parser.parse_args() - config_file = utils.get_config_file_path() + 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: + if cli.confirm_action("Using default config file."): + config_file = utils.get_config_file_path() + else: + sys.exit(1) + + console.rule(f"[bold red]{MODULE_NAME}") console.print(f"Using config file: {config_file}") utils.configure_logging( From d413231c1409ffbda3bc48b793244c06d26e3da6 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 12:36:29 -0400 Subject: [PATCH 12/15] Config for data dictionary --- pipeline/models/pipeline_failures.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pipeline/models/pipeline_failures.py b/pipeline/models/pipeline_failures.py index 42b996f..6163594 100644 --- a/pipeline/models/pipeline_failures.py +++ b/pipeline/models/pipeline_failures.py @@ -60,6 +60,7 @@ "llm_language_identification_failed", "transcribeme_pull_failed", "interview_not_in_study_list", + "crawler_stage_failed", "other", ] From 359892657c59fee0793b745498777d81029b6b2b Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 14:04:02 -0400 Subject: [PATCH 13/15] Add OH timezone and timezone fail logging --- .../ampscz/4_import_audio_journals.py | 29 ++++++++++++++----- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py index 8ddf93c..7f4f6e4 100755 --- a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py +++ b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py @@ -70,6 +70,7 @@ "PronetNL": "America/New_York", "PronetNN": "America/Chicago", "PronetOR": "America/Los_Angeles", + "PronetOH": "America/New_York", "PronetPA": "America/New_York", "PronetPI": "America/New_York", "PronetPV": "Europe/Rome", @@ -131,7 +132,7 @@ def get_journal_timestamp_from_mindlamp_json( def fetch_journals( - config_file: Path, subject_id: str, study_id: str + config_file: Path, subject_id: str, study_id: str, study_timezone: str ) -> List[AudioJournal]: """ Fetches the AudioJournals for a given subject ID. @@ -139,6 +140,9 @@ def fetch_journals( Args: config_file (Path): The path to the config file. subject_id (str): The subject ID. + study_timezone (str): The IANA timezone for the study (see + study_timezones) - resolved once per study by the caller so a + missing entry is only recorded/skipped once, not once per subject. Returns: List[AudioJournal]: A list of AudioJournal objects. @@ -158,11 +162,6 @@ def fetch_journals( audio_journal_path = list(audio_journal_root_path.glob("*_sound_*.mp3")) audio_journals: List[AudioJournal] = [] - study_timezone = study_timezones.get(study_id, None) - if study_timezone is None: - logger.error(f"No timezone found for {study_id}") - sys.exit(1) - subject_consent_date = Subject.get_consent_date( study_id=study_id, subject_id=subject_id, config_file=config_file ) @@ -339,6 +338,19 @@ def import_journals(config_file: Path, study_id: str, progress: Progress) -> Non config_file (Path): The path to the configuration file. """ + study_timezone = study_timezones.get(study_id, None) + if study_timezone is None: + logger.error(f"No timezone found for {study_id} - skipping study") + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + error_code="crawler_stage_failed", + identifier=study_id, + identifier_type="study", + error=f"No timezone configured for study {study_id} in study_timezones", + ) + return + # Get the subjects subjects = core.get_subject_ids(config_file=config_file, study_id=study_id) @@ -353,7 +365,10 @@ def import_journals(config_file: Path, study_id: str, progress: Progress) -> Non ) journals.extend( fetch_journals( - config_file=config_file, subject_id=subject_id, study_id=study_id + config_file=config_file, + subject_id=subject_id, + study_id=study_id, + study_timezone=study_timezone, ) ) progress.remove_task(task) From cb50ae0689ea29939a45b54693716456e2d5cae4 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 15:09:21 -0400 Subject: [PATCH 14/15] Deletes stale records for audio_journals on insert (avoids collision when journal upload is delayed compared to timestamp) --- pipeline/models/audio_journals.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/pipeline/models/audio_journals.py b/pipeline/models/audio_journals.py index d307916..fc04a9c 100644 --- a/pipeline/models/audio_journals.py +++ b/pipeline/models/audio_journals.py @@ -106,12 +106,27 @@ def to_sql(self) -> str: """ sanitized_path = db.santize_string(str(self.aj_path)) - + sanitized_subject_id = db.santize_string(self.subject_id) + sanitized_study_id = db.santize_string(self.study_id) + + # audio_journals also has UNIQUE (aj_day, aj_session, subject_id, study_id), + # separate from the aj_path primary key. Session numbers are recomputed + # from scratch on every crawl (see fetch_journals), so a newly-discovered + # file can "steal" the (day, session) slot a different, already-persisted + # file used to hold. Since a single INSERT...ON CONFLICT can only name one + # arbiter index, clear any other file's stale claim on this slot first; + # that stale file gets its own correct slot re-inserted later in the same + # batch, since every on-disk file is re-upserted every run. sql_query = f""" + DELETE FROM audio_journals + WHERE aj_day = {self.aj_day} AND aj_session = {self.aj_session} + AND subject_id = '{sanitized_subject_id}' AND study_id = '{sanitized_study_id}' + AND aj_path <> '{sanitized_path}'; + INSERT INTO audio_journals (aj_path, aj_name, aj_datetime, aj_day, aj_session, subject_id, study_id) VALUES ('{sanitized_path}', '{self.aj_name}', '{self.aj_datetime}', {self.aj_day}, - {self.aj_session}, '{self.subject_id}', '{self.study_id}') + {self.aj_session}, '{sanitized_subject_id}', '{sanitized_study_id}') ON CONFLICT (aj_path) DO UPDATE SET aj_name = excluded.aj_name, aj_datetime = excluded.aj_datetime, From 890b1b03d97c1a37402565cf42642f98c2ea6475 Mon Sep 17 00:00:00 2001 From: rtmcard Date: Tue, 14 Jul 2026 15:49:01 -0400 Subject: [PATCH 15/15] DB connection optimizations --- .../ampscz/2_import_interview_files.py | 60 +++++++------------ .../ampscz/4_import_audio_journals.py | 25 ++++---- 2 files changed, 38 insertions(+), 47 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 4a23c7c..c288d44 100755 --- a/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py +++ b/pipeline/crawlers/study_specific/ampscz/2_import_interview_files.py @@ -297,6 +297,29 @@ def fetch_interviews( logger.debug(f"{subject_id}: subject is not active, skipping interview fetch.") return [] + # Fetched once per subject rather than once per interview file - it's the + # same value for every file below, and each call previously opened its own + # fresh DB connection (db.get_db_connection() creates a new SQLAlchemy + # engine per call), which was the dominant cost for subjects with many + # interview files. + consent_date_s = core.get_consent_date_from_subject_id( + config_file=config_file, subject_id=subject_id, study_id=study_id + ) + if consent_date_s is None: + error = ValueError(f"Could not find consent date for {subject_id}") + logger.error(str(error)) + db.record_failure( + config_file=config_file, + stage=MODULE_NAME, + error_code="consent_date_missing", + identifier=subject_id, + error=error, + identifier_type="subject", + study_id=study_id, + ) + return [] + consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") + config_params = config(path=config_file, section="general") data_root = Path(config_params["data_root"]) @@ -352,24 +375,6 @@ def fetch_interviews( ) continue - consent_date_s = core.get_consent_date_from_subject_id( - config_file=config_file, subject_id=subject_id, study_id=study_id - ) - if consent_date_s is None: - error = ValueError(f"Could not find consent date for {subject_id}") - logger.error(str(error)) - db.record_failure( - config_file=config_file, - stage=MODULE_NAME, - error_code="consent_date_missing", - identifier=subject_id, - error=error, - identifier_type="subject", - study_id=study_id, - ) - continue - consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") - interview_name = dpdash.get_dpdash_name( study=study_id, subject=subject_id, @@ -422,25 +427,6 @@ def fetch_interviews( ) continue - consent_date_s = core.get_consent_date_from_subject_id( - config_file=config_file, subject_id=subject_id, study_id=study_id - ) - - if consent_date_s is None: - error = ValueError(f"Could not find consent date for {subject_id}") - logger.error(str(error)) - db.record_failure( - config_file=config_file, - stage=MODULE_NAME, - error_code="consent_date_missing", - identifier=subject_id, - error=error, - identifier_type="subject", - study_id=study_id, - ) - continue - consent_date = datetime.strptime(consent_date_s, "%Y-%m-%d") - interview_name = dpdash.get_dpdash_name( study=study_id, subject=subject_id, diff --git a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py index 7f4f6e4..591c089 100755 --- a/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py +++ b/pipeline/crawlers/study_specific/ampscz/4_import_audio_journals.py @@ -162,16 +162,6 @@ def fetch_journals( audio_journal_path = list(audio_journal_root_path.glob("*_sound_*.mp3")) audio_journals: List[AudioJournal] = [] - subject_consent_date = Subject.get_consent_date( - study_id=study_id, subject_id=subject_id, config_file=config_file - ) - subject_consent_date = pytz.timezone(study_timezone).localize( - subject_consent_date # type: ignore - ) # convert to timezone aware datetime - if subject_consent_date is None: - logger.error(f"No consent date found for {subject_id} - skipping...") - return [] - if len(audio_journal_path) == 0: logger.debug( f"No audio journal found for {subject_id} at {audio_journal_root_path}" @@ -180,6 +170,21 @@ def fetch_journals( else: logger.info(f"Found {len(audio_journal_path)} audio journals for {subject_id}") + # Fetched only once we know there's actually work to do - this call opens + # its own fresh DB connection (db.get_db_connection() creates a new + # SQLAlchemy engine per call), and most subjects in a study have no audio + # journals at all, so checking audio_journal_path first avoids paying for + # a connection on every subject that has nothing to process. + subject_consent_date = Subject.get_consent_date( + study_id=study_id, subject_id=subject_id, config_file=config_file + ) + if subject_consent_date is None: + logger.error(f"No consent date found for {subject_id} - skipping...") + return [] + subject_consent_date = pytz.timezone(study_timezone).localize( + subject_consent_date # type: ignore + ) # convert to timezone aware datetime + for audio_journal in audio_journal_path: audio_journal_basename = ( audio_journal.name