mcardle-features-runners

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.
@@ -176,6 +202,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:

pipeline/core/audio_qc_override.py

@@ -0,0 +1,135 @@
+"""
+Manual audio QC override.
+
+Reviewers can mark a file that failed automated audio QC
+(transcribeme.audio_qc.aqc_passed = FALSE) as manually approved
+(aqc_override = TRUE) from the dashboard. This module relocates such a file
+from `rejected_audio/` back into `pending_audio/` the next time a
+transcribeme push runner looks for work, so it flows through the rest of the
+pipeline exactly as if it had passed QC.
+
+The file's path is a primary/foreign key chain across `files`,
+`transcribeme.wav_conversion`, and `transcribeme.audio_qc`, and none of
+those foreign keys are DEFERRABLE - so the path can't be renamed in place
+across tables in one transaction. Instead this deletes the old rows and
+re-inserts fresh ones at the new path, carrying forward the original QC
+metrics/fail_reasons/timestamps (aqc_passed stays FALSE - it's a factual
+record of what the automated check found; aqc_override is what lets the
+file through).
+"""
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+import pandas as pd
+
+from pipeline.helpers import db
+from pipeline.models.files import File
+from pipeline.models.transcribeme.audio_qc import AudioQC
+from pipeline.models.transcribeme.wav_conversion import WavConversion
+
+logger = logging.getLogger(__name__)
+
+
+def _fetch_current_row(wav_path: Path, config_file: Path) -> Optional[pd.Series]:
+    query = f"""
+        SELECT
+            wc.wc_source_path, wc.wc_duration_s, wc.wc_timestamp,
+            aqc.aqc_passed, aqc.aqc_metrics, aqc.aqc_fail_reasons,
+            aqc.aqc_duration_s, aqc.aqc_timestamp, aqc.aqc_override
+        FROM transcribeme.wav_conversion wc
+        JOIN transcribeme.audio_qc aqc ON aqc.aqc_source_path = wc.wc_destination_path
+        WHERE wc.wc_destination_path = '{db.santize_string(wav_path)}'
+    """
+    result_df = db.execute_sql(config_file=config_file, query=query)
+    if result_df.empty:
+        return None
+    return result_df.iloc[0]
+
+
+def relocate_if_overridden(wav_path: Path, config_file: Path) -> Path:
+    """
+    If `wav_path` is a manually-overridden failed-QC file still sitting in
+    `rejected_audio/`, move it to `pending_audio/` and re-point the
+    files/wav_conversion/audio_qc rows at the new path, preserving the
+    original QC metrics/fail_reasons/timestamps.
+
+    Idempotent: files that already passed QC, aren't overridden, or have
+    already been relocated by a previous run are returned unchanged.
+
+    Args:
+        wav_path (Path): Current path of the converted WAV file, i.e. the
+            current transcribeme.audio_qc.aqc_source_path /
+            transcribeme.wav_conversion.wc_destination_path.
+        config_file (Path): Path to the config file.
+
+    Returns:
+        Path: The file's path after relocation (or `wav_path` unchanged if
+            no relocation was needed).
+    """
+    row = _fetch_current_row(wav_path=wav_path, config_file=config_file)
+    if row is None:
+        return wav_path
+
+    if bool(row["aqc_passed"]) or not bool(row["aqc_override"]):
+        return wav_path
+
+    if wav_path.parent.name != "rejected_audio":
+        # Already relocated by a previous run.
+        return wav_path
+
+    new_path = wav_path.parent.parent / "pending_audio" / wav_path.name
+    new_path.parent.mkdir(parents=True, exist_ok=True)
+
+    if new_path.exists():
+        raise FileExistsError(
+            f"Cannot relocate overridden audio QC file: {new_path} already exists."
+        )
+
+    logger.info(
+        f"Relocating manually-overridden audio QC file {wav_path} -> {new_path}",
+        extra={"markup": True},
+    )
+    wav_path.rename(new_path)
+
+    new_file = File(file_path=new_path)
+    new_wav_conversion = WavConversion(
+        wc_source_path=Path(row["wc_source_path"]),
+        wc_destination_path=new_path,
+        wc_duration_s=row["wc_duration_s"],
+    )
+    new_wav_conversion.wc_timestamp = row["wc_timestamp"]
+    new_audio_qc = AudioQC(
+        aqc_source_path=new_path,
+        aqc_passed=False,
+        aqc_metrics=row["aqc_metrics"],
+        aqc_fail_reasons=row["aqc_fail_reasons"],
+        aqc_duration_s=row["aqc_duration_s"],
+        aqc_timestamp=row["aqc_timestamp"],
+    )
+
+    sanitized_old_path = db.santize_string(wav_path)
+    sanitized_new_path = db.santize_string(new_path)
+
+    queries = [
+        f"DELETE FROM transcribeme.audio_qc WHERE aqc_source_path = '{sanitized_old_path}';",
+        f"DELETE FROM transcribeme.wav_conversion WHERE wc_destination_path = '{sanitized_old_path}';",
+        f"DELETE FROM files WHERE file_path = '{sanitized_old_path}';",
+        new_file.to_sql(),
+        new_wav_conversion.to_sql(),
+        new_audio_qc.to_sql(),
+        f"UPDATE transcribeme.audio_qc SET aqc_override = TRUE "
+        f"WHERE aqc_source_path = '{sanitized_new_path}';",
+    ]
+
+    db.execute_queries(
+        config_file=config_file,
+        queries=queries,
+        show_commands=False,
+        failure_stage="pipeline.core.audio_qc_override",
+        failure_identifier=str(new_path),
+        failure_identifier_type="file_path",
+    )
+
+    return new_path

pipeline/core/load_openface.py

@@ -292,7 +292,18 @@ 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}"
+            )
+            db.record_failure(
+                config_file=config_file,
+                stage="load_openface",
+                identifier=interview_name,
+                error=e,
+                identifier_type="interview_name",
+            )
 
     queries: List[str] = []
 

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})

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)

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,17 @@ 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,
+            error_code="missing_file",
+            identifier=study_id,
+            error=error,
+            identifier_type="study",
+        )
+        raise error
     else:
         insert_study(config_file=config_file, study_id=study_id)
 
@@ -162,18 +177,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 +233,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})