rtm44-comments

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