API Reference

Pipeline Entrypoints

Unified runner for GDELT and configured HTML/Scooper scrapers.

The orchestrator owns cross-pipeline CLI concerns so operators can run small smoke tests or larger backfills from one command while each pipeline keeps its source-specific defaults and implementation details.

src.orchestrator.chunk_list(items, num_chunks)

Split a list of items into a specified number of chunks, as evenly as possible.

Parameters:

• items – The list of items to split.

• num_chunks – The number of chunks to create.

Returns:

A list of lists, where each sublist is a chunk of the original items.

src.orchestrator.main(argv=None)

Parse CLI options, run selected pipeline stages, and report summaries.

Parameters:

argv (list[str] | None) – Optional argument list for tests and programmatic callers. When omitted, argparse reads from the process command line.

Returns:

Process exit code. A successful orchestrated run returns 0.

Return type:

int

Provide shared utility functions for file validation, data processing, and URL handling in the application.

This module provides utility functions and constants that assist in processing and managing data for the application. It includes functions for fetching news articles, extracting key information, and classifying them into different categories. Including page fetching, validation of files, JSON and CSV outputs, and URL construction. The shared utilities aim to simplify and streamline repetitive tasks or operations across the project.

- `AI_MODEL`

The specific model that the AI will use for processing.

- `_PROJECT_ROOT`

Specifies the project’s root directory.

- `READY_FOR_RAG_DIR`

Directory designated for resources ready for retrieval-augmented generation (RAG).

- `NOISE_DIR`

Directory for storing noise data.

- `VULNERABILITIES_DIR`

Directory for storing vulnerabilities data.

- `HEADERS`

Headers for HTTP-related tasks.

- `VULN_CSV_HEADER`

Header for the vulnerabilities CSV file.

- `NOISE_CSV_HEADER`

Header for the noise CSV file.

- `SUBSECTOR_FIELDS`

A dictionary that maps subsectors to their specific fields.

Functions:

• get_config_value: Retrieves a configuration value by name, with an optional default.

• get_config_bool: Retrieves a boolean configuration value by name, with an optional default.

• get_config_int: Retrieves an integer configuration value by name, with an optional default.

• get_config_date: Retrieves an ISO-formatted date configuration value by name, with an optional default.

• get_page: Retrieves web page content for a given URL, handling HTTP requests.

• _site_filename: Generates or retrieves specific filename associated with a site.

• check_valid_file: Validates files against specific criteria.

• json_output: Outputs data in JSON format.

• vuln_output: Processes and generates output related to vulnerabilities.

• noise_output: Processes and generates output related to noise.

• build_page_url: Constructs URLs for web pages based on given parameters.

• ai_check_validation: Parses and verifies whether a healthcare-related article describes an ongoing operational disruption or confirmed breach

  at a named healthcare entity based on strict, predefined criteria.

• get_extraction_template: Builds a typed JSON extraction template scoped to a single subsector, mapping each field to its expected primitive type.

• extract_fields: Extracts subsector-specific metadata from a confirmed disruption article by prompting the LLM with a typed, subsector-scoped template.

• install_handler: Installs a signal handler for graceful shutdown.

• pause_if_shutdown: Checks if a shutdown has been requested and pauses processing if so.

• request_pause: Requests a pause in processing, setting the shutdown event.

• collect_as_completed: Collects results from futures as they complete, handling shutdown requests.

• shutdown_executor: Shuts down a concurrent executor, optionally canceling pending futures if a shutdown has been requested.

• shutdown_requested: Checks if a shutdown has been requested.

Possible subsectors:

• “drug_shortage”: A confirmed shortage of a named drug patients need now.

• “medical_device_shortage”: A confirmed inability to supply a specific named medical device.

• “cyber_attack”: A confirmed breach or attack involving a named healthcare entity.

• “natural_disaster”: Operational shutdowns due to fire, flood, storm, or other physical events.

• “other”: Other confirmed operational disruptions that do not fit the previous categories.

• “none”: Used when no operational disruption or breach is confirmed.

src.shared_utils.get_config_value(name, default=None)

Retrieve a configuration value by name, with an optional default.

Parameters:

• name (str) – The name of the configuration variable to retrieve.

• default (str | None) – An optional default value to return if the configuration variable is not set or is empty. Defaults to None.

Returns:

The value of the configuration variable if it exists and is not empty; otherwise, returns the provided default value.

Return type:

str | None

src.shared_utils.get_config_bool(name, default=False)

Retrieve a boolean configuration value by name, with an optional default.

Parameters:

• name (str) – The name of the configuration variable to retrieve.

• default (bool) – An optional default value to return if the configuration variable is not set or is empty. Defaults to False.

Returns:

The boolean value of the configuration variable if it exists and is not empty; otherwise, returns the provided default value. The function interprets “true” and “yes” (case-insensitive) as True, and any other non-empty value as False.

Return type:

bool

src.shared_utils.get_config_int(name, default=None)

Retrieve an integer configuration value by name, with an optional default.

Parameters:

• name (str) – The name of the configuration variable to retrieve.

• default (int | None) – An optional default value to return if the configuration variable is not set or is empty. Defaults to None.

Returns:

The integer value of the configuration variable if it exists and is not empty; otherwise, returns the provided default value.

Return type:

int | None

src.shared_utils.get_config_date(name, default=None)

Retrieve an ISO-formatted date configuration value by name, with an optional default.

Parameters:

• name (str) – The name of the configuration variable to retrieve.

• default (datetime.date | None) – An optional default value to return if the configuration variable is not set, empty, or not a valid ISO date. Defaults to None.

Returns:

The parsed date if the configuration variable exists and is a valid YYYY-MM-DD string; otherwise, returns the provided default value.

Return type:

datetime.date | None

src.shared_utils.get_page(url, connect_timeout=10, read_timeout=15, absolute_timeout=45)

Fetches the content of a web page for the given URL.

Parameters:

• url (str) – The URL of the web page to fetch.

• connect_timeout (int) – Timeout for connecting to the server.

• read_timeout (int) – Idle timeout for reading data from the server.

• absolute_timeout (int) – Maximum total time allowed for the request.

Returns:

The response object containing the web page content.

Return type:

requests.Response

Raises:

• requests.exceptions.HTTPError – If the HTTP request returned an unsuccessful status code.

• requests.exceptions.RequestException – For any issues during the HTTP request such as timeouts or connection errors.

src.shared_utils.check_valid_file(site_name)

Checks for the existence of required files and directories for the given site name. If the required files do not exist, creates them with appropriate initial content.

Parameters:

site_name (str) – The name of the site used to generate file names and structure.

Function Logic:

• Ensures the directories READY_FOR_RAG_DIR, NOISE_DIR, and VULNERABILITIES_DIR exist by creating them if necessary.

• Constructs a file stem using the supplied site_name with the help of the _site_filename function.

• Checks if a .json file for the site exists in READY_FOR_RAG_DIR. If not, creates the file with a default JSON structure.

• Checks if a .csv file for the site exists in NOISE_DIR. If not, creates an empty file with a header row defined by NOISE_CSV_HEADER.

• Checks if a .csv file for the site exists in VULNERABILITIES_DIR. If not, creates an empty file with a header row defined by VULN_CSV_HEADER.

• Prints messages to indicate the creation of new files when applicable.

src.shared_utils.is_known_article(site_name, title, body_snippet)

Not documenting on purpose, this function will likely be deleted in the near future

Parameters:

• site_name (str)

• title (str)

• body_snippet (str)

Return type:

bool

src.shared_utils.prepend_vuln_csv(site_name, new_rows)

Not documenting on purpose, this function will likely be deleted in the near future

Parameters:

• site_name (str)

• new_rows (list[list[str]])

Return type:

None

src.shared_utils.prepend_noise_csv(site_name, new_rows)

Not documenting on purpose, this function will likely be deleted in the near future

Parameters:

• site_name (str)

• new_rows (list[list[str]])

Return type:

None

src.shared_utils.prepend_json_sources(site_name, new_vulns)

Not documenting on purpose, this function will likely be deleted in the near future

Parameters:

• site_name (str)

• new_vulns (list[Vulnerability])

Return type:

None

src.shared_utils.get_title(url)

Fetch and return the page title for a URL.

Extracts the HTML <title> tag and strips common site-name suffixes (e.g. “ | Reuters”, “ - NBC News”). Falls back to the raw URL if no usable <title> tag is found or the request fails.

Parameters:

url (str) – The URL to fetch. https:// is prepended if the scheme is missing.

Returns:

Cleaned page title, or the URL on failure.

Return type:

str

src.shared_utils.get_body(url)

Fetch and return the main article text for a URL.

The function performs a simple HTML scrape using requests and BeautifulSoup, removes common non-content tags and noisy selectors (ads, sidebars, footers), and returns the concatenated paragraph text when available. Network errors or missing content return an empty string.

Parameters:

url (str) – The URL to fetch. If the scheme is missing, https:// is prepended.

Returns:

Cleaned article text, or an empty string on error or if no body

content is found.

Return type:

str

Notes

• This is a heuristic extractor and may not work for all sites.

• The returned body may be long; callers should truncate if needed.

src.shared_utils.get_body_and_title(url)

Fetch a URL once and return both the article body and cleaned title.

Combines the work of get_body() and get_title() into a single HTTP request, halving outbound traffic when both values are needed for the same page.

The title is extracted before the body because _extract_body_from_soup() mutates the soup tree by decomposing non-content elements (which could remove the <title> tag).

Parameters:

url (str) – The URL to fetch. https:// is prepended when the scheme is missing.

Returns:

A (body, title) tuple. On network errors the body is "" and the title falls back to the (normalised) URL. On empty / missing content the body is "" while the title may still be valid.

Return type:

tuple[str, str]

src.shared_utils.ai_check_validation(title, body, use_bert=False, verbose=False, port=11434)

Parses and verifies whether a healthcare-related article describes an ongoing operational disruption or confirmed breach at a named healthcare entity based on strict, predefined criteria.

Parameters:

• title (str) – The title of the article being analyzed.

• body (str) – The main content or excerpt of the article.

• use_bert (bool) – False by default, calls bert before calling the llm to save time

• port (int) – The port on which the ollama server is running

• verbose (bool)

Return type:

tuple[bool, str]

Returns: A tuple:

• A boolean indicating whether the article is flagged as a threat (True if operational disruption or confirmed breach).

• A string providing further details: the subsector if flagged as a disruption or the reason for rejection if not flagged.

This function sends the article’s title and body to an AI system for evaluation. The AI follows explicit rules to assess disruptions or breaches in healthcare. If an operational disruption is identified, the response will specify the subsector such as ‘cyber_attack’, ‘drug_shortage’, etc. If not, the output will explain why the article was rejected.

Exceptions: If an error occurs during the request or response parsing, the function catches the error, logs it, and returns False with “Parsing Error”.

src.shared_utils.get_extraction_template(subsector)

Builds a typed JSON extraction template for the LLM prompt.

Inspects the dataclass annotations for the given subsector and maps each field to a stringified type hint (e.g., “string”, “boolean”, “integer”, “list of strings”) to constrain the LLM output and prevent type hallucination.

Parameters:

subsector (str) – The classification name of the healthcare subsector.

Returns:

A mapping of required field names to their expected primitive types.

Return type:

dict

exception src.shared_utils.MissingSubsectorFieldsError

Raised when a subsector has no configured extraction fields.

src.shared_utils.extract_fields(subsector, title, body, port)

Extract universal and subsector fields for a validated article.

This function is called after an article classifies as a true vulnerability. It sends the article title and body to Ollama to populate the universal sector fields and the fields specific to the selected subsector.

Parameters:

• subsector – Subsector returned by ai_check_validation.

• title – Title of the current article.

• body – Full body text of the current article.

• port – The port on which the ollama server is running.

Returns:

A tuple with the universal LLM_SECTOR_FIELDS values first and the matching SUBSECTOR_FIELDS values second.

Raises:

MissingSubsectorFieldsError – If the subsector is unknown or has no configured extraction fields.

Return type:

tuple[dict, dict]

Note

The AI currently decides which values can be extracted from the article. That keeps extraction flexible, but it is not ideal as a long-term structured-data contract.

exception src.shared_utils.model_unavailable_error

Raised when configured Ollama model is unavailable

exception src.shared_utils.LLMUnavailableError

Raised when an LLM HTTP call fails (timeout/connection/HTTP error).

Distinct from a negative classification: signals the call never produced a usable answer, so callers should skip and retry rather than treat the article as noise.

src.shared_utils.ensure_model_available(model='llama3.2:latest')

Parameters:

model (str)

Return type:

None

src.shared_utils.run_clean()

Clean all GDELT-generated data so the pipeline starts fresh.

Delegates to scripts.clean_gdelt.run_clean() which owns the canonical implementation. This wrapper exists for backward compatibility.

src.shared_utils.clear_directory(directory)

Delete all files and subdirectories inside a directory.

Parameters:

directory (Path) – The path to the directory to clear.

Return type:

None

src.shared_utils.df_dup(dfs, verbose=False)

Deduplicates a lits of dataframes, used to make one single csv write by scooper

Parameters:

• dfs (list[pandas.DataFrame]) – list of dataframes with the same schema

• verbose (bool) – prints messages when True

Returns:

One unique DataFrame a list of duplicate titles

Return type:

tuple[pandas.DataFrame, list[str]]

src.shared_utils.update_csv(df, path, verbose=False)

Appends a csv with a given DataFrame. This is used in scooper after calling ‘df_dup’

Parameters:

• df (pandas.DataFrame) – DataFrame to append

• path (Path) – CSV path

• verbose (bool) – used to print messages if true

Return type:

None

src.shared_utils.update_json(vulns, path)

Used to write a list of Vulnerabilities into a given path

Parameters:

• vulns (list[Vulnerability]) – vulnerabilities to add to the file.

• path (str) – destination *.json file (created if it does not exist)

Return type:

None

class src.shared_utils.NoiseCollector(output_path)

Accumulate rejected-article records and flush them to a JSON file.

Used when the --debug / -d flag is passed to capture every article the pipeline skips or rejects so operators can evaluate false-negative rates.

Parameters:

output_path (Path) – Destination JSON file (e.g. data/noise/debug_noise_gdelt.json).

add(*, url, title, source, reason, body_preview='', stage='')

Append one noise record.

Parameters:

• url (str) – The article URL that was rejected.

• title (str) – The article title.

• source (str) – Pipeline source name (e.g. "GDELT", "CyberScoop").

• reason (str) – Human-readable rejection reason.

• body_preview (str) – First 250 characters of the article body.

• stage (str) – Pipeline stage that rejected the article (e.g. "already_seen", "validation", "extraction").

Return type:

None

flush()

Write accumulated records to the JSON file and return the path.

Creates the parent directory if it does not exist. Returns None when there are no records to write.

Return type:

Path | None

src.shared_utils.shutdown_requested()

Check if a shutdown event has been triggered.

Return type:

bool

src.shared_utils.install_handler()

Install a SIGINT handler that sets a global shutdown event.

Return type:

None

src.shared_utils.pause_if_shutdown(stats)

Check if a shutdown event has been triggered and pause the pipeline if so.

Parameters:

stats – An object with a ‘paused’ attribute to indicate the paused state.

Returns:

True if the pipeline was paused due to a shutdown event, False otherwise.

Return type:

bool

src.shared_utils.request_pause(stats)

Request a pause in the pipeline by setting a global shutdown event.

Parameters:

stats – An object with a ‘paused’ attribute to indicate the paused state.

Return type:

None

src.shared_utils.collect_as_completed(futures, on_result)

Collect future results, returning promptly once shutdown is requested.

Parameters:

• futures (Iterable[Future[_T]]) – An iterable of Future objects to collect results from.

• on_result (Callable[[_T], None]) – A callback function to process each result as it becomes available.

Return type:

None

src.shared_utils.shutdown_executor(executor)

Shut down an executor without waiting when shutdown was requested.

Return type:

None

src.shared_utils.exit_if_shutdown(code=0)

Exit the process immediately after shutdown cleanup, without joining workers.

Parameters:

code (int) – Exit code to return. Defaults to 0.

Returns:

The exit code, which can be used for logging or further processing.

Return type:

int

GDELT Pipeline

GDELT end-to-end runner.

Pipeline:

gdelt_seeds.backfill_seeds – collect candidate seeds from GDELT GKG src.shared_utils.get_body_and_title – scrape page body + title in one request src.shared_utils.ai_check_validation – LLM validates as active disruption src.shared_utils.extract_fields – LLM extracts schema-specific fields

Constants:

BODY_CHAR_LIMIT: Number of characters to include from the article body when passing to the LLM for validation and extraction.

Functions:

stable_id(url): Generate a stable ID for a given URL using SHA-256 hashing. fmt_dt(value): Format a date string into YYYY-MM-DD HH:MM format. Tries multiple input formats and returns the original value if parsing fails. ensure_raw_dirs(): Ensure that the raw directories for seeds, validated, and enriched data exist. Creates them if they don’t. ensure_cache_dir(): Ensure that the GDELT zip cache directory exists. Creates it if it doesn’t. save_json(path, data): Save a dictionary as JSON to the specified path, creating parent directories if needed. persist_raw_seeds(raw_seeds): Persist raw seeds to the seeds directory, using stable IDs for filenames. persist_stage(directory, article_id, stage, url, data): Persist data for a specific stage (validated, enriched) using a stable ID for the filename. load_staged_payloads(stage, reporter, stats): Load staged payloads for the requested GDELT stitch stage. stitch_staged_records(output_path, stage, seen_urls_file, use_bert, reporter, stats, verbose): Recover final output from a staged GDELT pipeline stage. load_seen(seen_file): Load seen URLs from file. Returns a set of URLs that have been seen and processed. If the file does not exist or cannot be read, returns an empty set. save_seen(seen, seen_file): Save seen URLs to file. process_seed(seed, seen, use_bert, reporter, stats): Run a single seed through validation + extraction. Returns a Vulnerability if validated as a disruption, else None. run(num_files, limit, subsectors, output_path, start_date, end_date, seen_urls_file, use_bert, verbose, reporter, stats): Main function to run the GDELT pipeline end-to-end.

src.GDELT.runner.stable_id(url)

Generate a stable ID for a given URL using SHA-256 hashing.

Parameters:

url (str)

Return type:

str

src.GDELT.runner.fmt_dt(value)

Format a date string into YYYY-MM-DD HH:MM format. Tries multiple input formats and returns the original value if parsing fails.

Parameters:

value (str) – The input date string to format.

Returns:

MM format, or the original value if parsing fails.

Return type:

A formatted date string in YYYY-MM-DD HH

src.GDELT.runner.ensure_raw_dirs()

Ensure that the raw directories for seeds, validated, and enriched data exist. Creates them if they don’t.

Return type:

None

src.GDELT.runner.ensure_cache_dir()

Ensure that the GDELT zip cache directory exists. Creates it if it doesn’t.

The cache directory GDELT_CACHE_DIR is never cleared by the pipeline so that zip files downloaded in one run are reused in subsequent runs covering the same date range. Only remove files from this directory manually when you want to force a fresh download.

Return type:

None

src.GDELT.runner.save_json(path, data)

Save a dictionary as JSON to the specified path, creating parent directories if needed.

Parameters:

• path (Path) – The path to the JSON file to save.

• data (dict) – The dictionary to save as JSON.

Return type:

None

src.GDELT.runner.dedupe_raw_seeds(raw_seeds)

Deduplicate raw GDELT seeds by exact URL while preserving subsector evidence.

The first seed for a URL stays canonical so existing single-subsector seed metadata remains backward compatible. Later duplicates only contribute unique subsector labels to the canonical seed’s detected_subsectors list.

Parameters:

raw_seeds (list[dict]) – The raw seed dictionaries to deduplicate by URL.

Returns:

A list of unique seed dictionaries with all distinct subsector labels preserved in detected_subsectors.

Return type:

list[dict]

src.GDELT.runner.persist_raw_seeds(raw_seeds)

Persist raw seeds to the seeds directory, using stable IDs for filenames.

Parameters:

raw_seeds (list[dict]) – A list of seed dictionaries to persist.

Return type:

None

src.GDELT.runner.persist_stage(directory, article_id, stage, url, data)

Persist data for a specific stage (validated, enriched) using a stable ID for the filename.

Parameters: - directory: The directory to save the file in (e.g., validated, enriched). - article_id: The stable ID for the article, used as the filename. - stage: The processing stage (e.g., “validated”, “enriched”) to include in the saved data. - url: The URL of the article, included in the saved data for reference. - data: The dictionary of data to save for this stage.

Parameters:

• directory (Path)

• article_id (str)

• stage (str)

• url (str)

• data (dict)

Return type:

None

src.GDELT.runner.load_seen(seen_file=None)

Load seen URLs from file. Returns a set of URLs that have been seen and processed. If the file does not exist or cannot be read, returns an empty set.

Parameters:

seen_file (Path | None) – Optional path to the JSON file containing seen URLs. If None, defaults to data/seen_urls.json in the project root.

Returns:

A set of URLs that have been seen and processed. Returns an empty set if the file does not exist or cannot be read.

Return type:

set

src.GDELT.runner.save_seen(seen, seen_file=None)

Save seen URLs to file.

Parameters:

• seen (set) – A set of URLs that have been seen and processed.

• seen_file (Path | None) – Optional path to the JSON file to save seen URLs. If None, defaults to data/seen_urls.json in the project root.

Return type:

None

src.GDELT.runner.write_output_records(records, output_path, reporter, stats)

Write completed GDELT records to the configured processed JSON output.

This helper centralizes the final output write so normal completion and graceful interrupt handling use the same merge behavior. Records are converted to dictionaries, publication dates are normalized through fmt_dt, and existing output files are merged when they already contain a sources list or legacy list-shaped output.

Parameters:

• records (list[Vulnerability | dict]) – Completed vulnerability records ready for processed output.

• output_path (str | None) – Optional JSON file or directory path. Directory paths write GDELT.json inside the directory; None writes to the default processed GDELT output.

• reporter (CliReporter) – Reporter used to print the write summary.

• stats (PipelineStats) – Pipeline statistics updated with the number of newly written output records.

Returns:

The resolved output file path that was written.

Return type:

Path

src.GDELT.runner.process_staged_seeds(seeds, seen, use_bert=False, reporter=None, stats=None, debug_noise=None, port=None)

Process staged GDELT seeds through validation and extraction while preserving progress if processing is interrupted.

Parameters:

• seeds (list[dict]) – The staged seed dictionaries to process.

• seen (set) – A set of URLs that have already been processed.

• use_bert (bool) – Whether to run a BERT pre-filter before LLM validation.

• reporter (CliReporter | None) – Optional CliReporter for logging progress and details.

• stats (PipelineStats | None) – Optional PipelineStats for tracking processing statistics.

• debug_noise (NoiseCollector | None) – Optional NoiseCollector for recording rejected articles.

• port (int | None) – Optional port for the LLM validation service.

Returns:

A list of vulnerabilities completed from the staged seeds.

Return type:

list[Vulnerability]

src.GDELT.runner.load_staged_payloads(stage='enriched', reporter=None, stats=None, directory=None)

Load valid staged GDELT payloads for the requested recovery stage, skipping files that are malformed or missing the expected payload.

Parameters:

• stage (str) – The recovery stage to load: seeds, validated, or enriched.

• reporter (CliReporter | None) – Optional CliReporter for reporting malformed staged files.

• stats (PipelineStats | None) – Optional PipelineStats updated when staged files are skipped.

• directory (Path | None) – Optional staging directory override for the requested stage.

Returns:

A list of payload dictionaries loaded from the staging directory.

Return type:

list[dict]

src.GDELT.runner.stitch_staged_records(output_path=None, stage='enriched', seen=None, use_bert=False, reporter=None, stats=None, verbose=False)

Recover records from a staged GDELT pipeline stage and write the deduplicated records to the final output file. Seed recovery resumes validation and extraction for seeds without completed staged records.

Parameters:

• output_path (str | None) – Optional JSON file or directory path for the final output.

• stage (str) – The recovery stage to stitch: seeds, validated, or enriched.

• seen (set | None) – Optional set of URLs that have already been processed.

• use_bert (bool) – Whether to run a BERT pre-filter during seed recovery.

• reporter (CliReporter | None) – Optional CliReporter for logging progress and details.

• stats (PipelineStats | None) – Optional PipelineStats for tracking recovery statistics.

• verbose (bool) – Whether to show detailed per-article output.

Returns:

A deduplicated list of recovered records with formatted publication dates.

Return type:

list[dict]

src.GDELT.runner.process_seed(seed, seen, use_bert=False, reporter=None, stats=None, debug_noise=None, port=None)

Run a single seed through validation + extraction. Returns a Vulnerability if validated as a disruption, else None.

Parameters: - seed: The seed dictionary containing at least a “url” key. - seen: A set of URLs that have already been processed, used to skip duplicates. - use_bert: Whether to run a BERT pre-filter before LLM validation. - reporter: Optional CliReporter for logging progress and details. - stats: Optional PipelineStats for tracking statistics. - debug_noise: Optional NoiseCollector for recording rejected articles.

Returns: - A Vulnerability object if the seed is validated as a disruption, or None if it is skipped or rejected.

Parameters:

• seed (dict)

• seen (set)

• use_bert (bool)

• reporter (CliReporter | None)

• stats (PipelineStats | None)

• debug_noise (NoiseCollector | None)

Return type:

Vulnerability | None

src.GDELT.runner.run(num_files, limit, output_path=None, start_date=None, end_date=None, seen=None, use_bert=False, verbose=False, reporter=None, stats=None, raw_seeds=None, debug_noise=None, port=11434, sector='health', clear_seeds=True)

Main function to run the GDELT pipeline end-to-end.

Parameters:

num_files: Number of GDELT GKG files to scan for seeds. limit: Optional cap on the number of seeds to process, useful for testing. subsectors: Comma-separated list of subsectors to include, or “all”. output_path: Optional path to output JSON file or directory. start_date: Optional earliest date for GDELT files to include (YYYYMMDD or ISO format). end_date: Optional latest date for GDELT files to include (YYYYMMDD or ISO format). seen: Optional set of URLs that have already been processed. use_bert: Whether to run a BERT pre-filter before LLM validation. verbose: Whether to show detailed per-article output. reporter: Optional CliReporter for logging progress and details. stats: Optional PipelineStats for tracking statistics. clean: Whether to clear modified directories and files before running. raw_seeds: Raw seed dictionaries to process debug_noise: Optional NoiseCollector for recording rejected articles. port: Where to run the ollama server sector: The sector to filter for (default: “health”). Currently only “health” is supported. clear_seeds: When True (default), clear the shared seed staging directory

after a successful run. Set False when a parent (e.g. orchestrator) clears it once after all workers finish.

Returns:

A list of validated and enriched vulnerability records as dictionaries.

Interrupt behavior:

Pressing Ctrl-C while a seed is being processed marks the run as paused, saves seen URLs, writes any completed records through write_output_records, preserves seed staging, and returns the completed records collected before the interrupt.

Parameters:

• num_files (int)

• limit (int | None)

• output_path (str | None)

• start_date (str | None)

• end_date (str | None)

• seen (set | None)

• use_bert (bool)

• verbose (bool)

• reporter (CliReporter | None)

• stats (PipelineStats | None)

• raw_seeds (list[dict] | None)

• debug_noise (NoiseCollector | None)

• port (int | None)

• sector (str)

• clear_seeds (bool)

Return type:

tuple[PipelineStats, list[dict]]

GDELT seed discovery and backfill logic. This module provides functions to fetch and process GDELT GKG files, filter them for relevant themes and US locations, and extract candidate seed records for the healthcare subsectors of interest. It also includes a backfill function to collect seeds from recent GDELT files based on date bounds or a specified number of recent files.

Constants: GKG_COLS: Mapping of column indices to their corresponding field names in the GDELT GKG files. NOISE_THEMES: Set of themes considered noise. SUBSECTOR_THEMES: Mapping of subsectors to their required theme sets. US_TLDS: Set of top-level domains associated with US-based websites. BLOCKED_TLDS: Set of top-level domains associated with non-US-based websites that should be blocked. URL_DENY_PATTERNS: Regular expression pattern to identify URLs that should be denied based on their path. URL_REQUIRE_PATTERNS: Regular expression pattern to identify URLs that should be required based on their content.

Functions: is_us_located(location_str): Check if a location string from GDELT indicates a US location. themes_match(theme_str, subsector=”all”): Check if themes in a theme string match the required themes for a given subsector or any supported subsector. detect_subsector(theme_str): Detect the specific subsector for a theme string based on the presence of required themes. themes_match_noise(theme_str): Check if themes in a theme string match any of the noise themes. url_passes_quality(url): Check if a URL passes quality checks based on its domain and path. process_gkg_file(link, subsector=”all”, reporter=None, stats=None): Download and process a GDELT GKG file, filtering for relevant themes, US locations, and quality URLs, and return candidate seed records. backfill_seeds(num_files=20, subsector=”all”, start_date=None, end_date=None, reporter=None, stats=None): Collect recent or date-bounded GDELT seeds for the requested subsector by scanning the master file list and processing relevant GKG files.

src.GDELT.gdelt_seeds.is_us_located(location_str)

Check if a GDELT location string indicates a US location.

Parameters:

location_str – The location string from GDELT’s V1Locations field.

Returns:

True if the location string indicates a US location, False otherwise.

src.GDELT.gdelt_seeds.themes_match(theme_str, sector='health')

Check if themes match a requested subsector.

Parameters:

• theme_str – The theme string from GDELT’s V1Themes field.

• sector – Sector key into SECTOR_THEMES (default: health).

Returns:

True if the themes match the requested sector, False otherwise.

src.GDELT.gdelt_seeds.detect_subsectors(theme_str, sector='health')

Return all matching subsectors for a theme string.

Parameters:

• theme_str – The theme string from GDELT’s V1Themes field.

• sector – Sector key into SECTOR_THEMES (default: health).

Returns:

A list of detected subsector names, or an empty list if no specific subsectors can be detected.

src.GDELT.gdelt_seeds.detect_subsector(theme_str, sector='health')

Return the first matching subsector for a theme string, or None.

Parameters:

• theme_str – The theme string from GDELT’s V1Themes field.

• sector – Sector key into SECTOR_THEMES (default: health).

Returns:

The detected subsector name if a match is found, or None if no specific subsector can be detected.

src.GDELT.gdelt_seeds.themes_match_noise(theme_str)

Check if themes match any noise patterns.

Parameters:

theme_str – The theme string from GDELT’s V1Themes field.

Returns:

True if any noise theme is present in the theme string, False otherwise.

src.GDELT.gdelt_seeds.url_passes_quality(url)

Check if a URL passes quality checks. The URL must start with http, have a domain in US_TLDS and not in BLOCKED_TLDS, and its path must not match URL_DENY_PATTERNS while matching URL_REQUIRE_PATTERNS.

Parameters:

url – The URL to check.

Returns:

True if the URL passes quality checks, False otherwise.

src.GDELT.gdelt_seeds.process_gkg_file(link, cache_dir=None, reporter=None, stats=None, sector='health')

Download and filter one GDELT GKG file into candidate seed records.

Parameters:

• link – The URL to the GDELT GKG file (a .zip containing a .csv).

• subsector – The subsector to filter for, or “all” for any supported subsector.

• cache_dir (Path | None) – Optional directory for caching downloaded zip files.

• reporter (CliReporter | None) – Optional CliReporter for logging progress and warnings.

• stats (PipelineStats | None) – Optional PipelineStats for recording statistics.

• sector (str) – Sector key into SECTOR_THEMES (default: health).

Returns:

A tuple containing a list of candidate seed records (dicts) and the total number of rows processed from the GKG file. Each seed record includes the URL, source, themes, subsector, date, and file name.

src.GDELT.gdelt_seeds.fetch_gkg_links(num_files=20, start_date=None, end_date=None)

Fetch and filter GDELT GKG zip URLs from the master file list.

Parameters:

• num_files – When no date bounds are set, return this many most-recent files.

• start_date – Optional inclusive lower bound on file timestamps.

• end_date – Optional inclusive upper bound on file timestamps.

Returns:

A list of GKG zip URLs matching the requested bounds.

src.GDELT.gdelt_seeds.backfill_seeds(num_files=20, start_date=None, end_date=None, links=None, cache_dir=None, reporter=None, stats=None, instance_name=None, sector='health')

Collect recent or date-bounded GDELT seeds for the requested subsector.

Parameters:

• num_files – The number of most recent GDELT files to scan if no date bounds are provided. Ignored if start_date or end_date is specified.

• subsector – The subsector to filter for, or “all” for any supported subsector.

• start_date – Optional start date bound (inclusive) in formats like YYYY-MM-DD or YYYYMMDD. If provided, only files with timestamps on or after this date will be processed.

• end_date – Optional end date bound (inclusive) in formats like YYYY-MM-DD or YYYYMMDD. If provided, only files with timestamps on or before this date will be processed.

• links – Optional pre-filtered GKG zip URLs to process instead of fetching the master file list.

• reporter (CliReporter | None) – Optional CliReporter for logging progress and warnings.

• stats (PipelineStats | None) – Optional PipelineStats for recording statistics.

• sector (str) – Sector key into SECTOR_THEMES (default: health).

• cache_dir (Path | None)

• instance_name (str | None)

Returns:

A list of unique seed records (dicts) that match the specified subsector and date bounds, extracted from the relevant GDELT GKG files. Each seed record includes the URL, source, themes, subsector, date, and file name.

HTML Pipeline

Scooper scraper that

Functions

• ‘_bert_status()’ : returns a human-readable description of the optional BERT pre-filter

• ‘_unseen_df()’ : returns a raw df of unclassified data

• ‘_setup_csv’ : helper function that set ups CSVs and makes sure they exist,

mostly used on the first run - ‘_scrape_page’ : scrapes 1 individual page of a cite. This has catching logic that stops once we hit a known cite - ‘_raw_data’ : Calls ‘_scrape_page’ and iterates through all pages in a cite. This mends everything into 1 DataFrame - ‘_update_raw_csv’ : Helper function that updates the Raw CSV with new found information - ‘setup_scooper’ : function to be called from the orchastrator level. This set up the scraper by scrapping before hand, and leaving the classifying step (most expensive) to be determined by the user - ‘run_scooper’ : classifies all the raw data by handing it off DataFrames to run_scooper - ‘_process_site’ : takes a dataframe and goes row by row to classify the entire df

src.HTML.scooper.setup_scooper(sb_only=False)

Sets up the pipeline for future runs. This updates and scrapes all new data, saves it to a CSV, and makes sure things like paths exist for future runs.

Parameters:

sb_only (bool)

Return type:

None

src.HTML.scooper.run_scooper(use_bert=False, verbose=False, start_date=None, end_date=None, reporter=None, stats=None, sb_only=False, site_split=False, port=11434, max_workers=None)

Runs the scooper using the raw and unclassified data. If site_split is True, it breaks off and makes a thread per unique site_name.

Returns:

• PipelineStats

• list of new vulns (used to make JSON)

• vulnerabilities dataframe (used to write CSV vulnerabilities)

• noise dataframes (used to write CSV noise)

Parameters:

• use_bert (bool)

• verbose (bool)

• start_date (date | None)

• end_date (date | None)

• reporter (CliReporter | None)

• stats (PipelineStats | None)

• sb_only (bool)

• site_split (bool)

• port (int)

• max_workers (int | None)

Return type:

tuple[PipelineStats, list[Vulnerability], pandas.DataFrame, pandas.DataFrame]

src.HTML.scooper.save_results(vuln_list, vuln_dfs, noise_dfs, sb_only=False)

Persist classified scooper results. Dedups across the given frames, appends to the vuln/noise CSVs, and writes the JSON corpus.

Accepts lists of frames so it serves both the single-pass/site-split route (wrap one frame in a list) and the date-split route (one frame per window). Empty frames/lists are handled gracefully. When sb_only is set, local CSV writes are skipped (Supabase path TODO); the JSON corpus is always written.

Parameters:

• vuln_list (list[Vulnerability]) – validated vulnerabilities used to build the JSON corpus

• vuln_dfs (list[pandas.DataFrame]) – list of vulnerability frames to dedupe + append to the vuln CSV

• noise_dfs (list[pandas.DataFrame]) – list of noise frames to dedupe + append to the noise CSV

• sb_only (bool) – skip local CSV writes when True

Return type:

None

Data Models

This module defines the Vulnerability class and related data classes for representing healthcare vulnerabilities across various subsectors. Each vulnerability includes required fields such as id, title, source_name, direct_link, subsector, date_accessed, date_published, and content. Optional fields include exec_summary, confidence_level, risk_level, geography_scope, start_date, end_date, resilience_or_mitigation_observed, and subsector_data.

Functions: - Vulnerability.to_dict(): Convert a Vulnerability instance to a dictionary. - Vulnerability.from_dict(data: dict): Create a Vulnerability instance from a dictionary.

class src.classes.vulnerability.SubsectorData

to_dict()

Return type:

dict

classmethod from_dict(data)

Parameters:

data (dict)

Return type:

SubsectorData

class src.classes.vulnerability.DrugShortageData(drug_name=None, generic_name=None, manufacturer=None, dosage_form=None, shortage_reason=None, estimated_resolution_date=None, affected_regions=None, domestic_vs_foreign_dependency=None)

Class for representing drug shortage-related healthcare vulnerabilities.

Parameters:

• drug_name (str | None)

• generic_name (str | None)

• manufacturer (str | None)

• dosage_form (str | None)

• shortage_reason (str | None)

• estimated_resolution_date (str | None)

• affected_regions (list[str] | None)

• domestic_vs_foreign_dependency (str | None)

drug_name: str | None = None

generic_name: str | None = None

manufacturer: str | None = None

dosage_form: str | None = None

shortage_reason: str | None = None

estimated_resolution_date: str | None = None

affected_regions: list[str] | None = None

domestic_vs_foreign_dependency: str | None = None

to_dict()

Convert the DrugShortageData instance to a dictionary.

Return type:

dict

classmethod from_dict(data)

Create a DrugShortageData instance from a dictionary.

Parameters:

data (dict)

Return type:

DrugShortageData

class src.classes.vulnerability.MedicalDeviceShortageData(device_name=None, device_category=None, manufacturer=None, manufacturer_country=None, shortage_reason=None, fda_recall_number=None, recall_class=None, affected_specialties=<factory>, alternatives_available=None, estimated_resolution_date=None, domestic_vs_foreign_dependency=None)

Class for representing medical device shortage-related healthcare vulnerabilities.

Parameters:

• device_name (str | None)

• device_category (str | None)

• manufacturer (str | None)

• manufacturer_country (str | None)

• shortage_reason (str | None)

• fda_recall_number (str | None)

• recall_class (str | None)

• affected_specialties (list[str])

• alternatives_available (bool | None)

• estimated_resolution_date (str | None)

• domestic_vs_foreign_dependency (str | None)

device_name: str | None = None

device_category: str | None = None

manufacturer: str | None = None

manufacturer_country: str | None = None

shortage_reason: str | None = None

fda_recall_number: str | None = None

recall_class: str | None = None

affected_specialties: list[str]

alternatives_available: bool | None = None

estimated_resolution_date: str | None = None

domestic_vs_foreign_dependency: str | None = None

to_dict()

Convert the MedicalDeviceShortageData instance to a dictionary.

Return type:

dict

classmethod from_dict(data)

Create a MedicalDeviceShortageData instance from a dictionary.

Parameters:

data (dict)

Return type:

MedicalDeviceShortageData

class src.classes.vulnerability.CyberAttackData(attack_type=None, threat_actor=None, individuals_affected=None, data_types_exposed=<factory>, systems_affected=<factory>, ransom_demanded_usd=None, ransom_paid=None, downtime_days=None, services_disrupted=<factory>, law_enforcement_involved=None, hhs_breach_portal_listed=None)

Class for representing cyber attack-related healthcare vulnerabilities.

Parameters:

• attack_type (str | None)

• threat_actor (str | None)

• individuals_affected (int | None)

• data_types_exposed (list[str])

• systems_affected (list[str])

• ransom_demanded_usd (float | None)

• ransom_paid (bool | None)

• downtime_days (int | None)

• services_disrupted (list[str])

• law_enforcement_involved (bool | None)

• hhs_breach_portal_listed (bool | None)

attack_type: str | None = None

threat_actor: str | None = None

individuals_affected: int | None = None

data_types_exposed: list[str]

systems_affected: list[str]

ransom_demanded_usd: float | None = None

ransom_paid: bool | None = None

downtime_days: int | None = None

services_disrupted: list[str]

law_enforcement_involved: bool | None = None

hhs_breach_portal_listed: bool | None = None

to_dict()

Convert the CyberAttackData instance to a dictionary.

Return type:

dict

classmethod from_dict(data)

Create a CyberAttackData instance from a dictionary.

Parameters:

data (dict)

Return type:

CyberAttackData

class src.classes.vulnerability.NaturalDisasterData(disaster_type=None, disaster_name=None, fema_declaration_id=None, category_magnitude=None, affected_facilities_count=None, evacuation_ordered=None, field_hospitals=None, beds_offline=None, facility_status=None, estimated_damage_usd=None, infrastructure_damage=<factory>, services_disrupted=<factory>)

Class for representing natural disaster-related healthcare vulnerabilities.

Parameters:

• disaster_type (str | None)

• disaster_name (str | None)

• fema_declaration_id (str | None)

• category_magnitude (str | None)

• affected_facilities_count (int | None)

• evacuation_ordered (bool | None)

• field_hospitals (int | None)

• beds_offline (int | None)

• facility_status (str | None)

• estimated_damage_usd (float | None)

• infrastructure_damage (list[str])

• services_disrupted (list[str])

disaster_type: str | None = None

disaster_name: str | None = None

fema_declaration_id: str | None = None

category_magnitude: str | None = None

affected_facilities_count: int | None = None

evacuation_ordered: bool | None = None

field_hospitals: int | None = None

beds_offline: int | None = None

facility_status: str | None = None

estimated_damage_usd: float | None = None

infrastructure_damage: list[str]

services_disrupted: list[str]

to_dict()

Convert the NaturalDisasterData instance to a dictionary.

Return type:

dict

classmethod from_dict(data)

Create a NaturalDisasterData instance from a dictionary.

Parameters:

data (dict)

Return type:

NaturalDisasterData

class src.classes.vulnerability.OtherData(event_type=None, event_description=None, severity=None, departments_affected=<factory>, staff_type_affected=None, beds_offline=None, services_disrupted=<factory>, regulatory_response=None)

Class for representing other types of healthcare vulnerabilities that don’t fit into predefined categories.

Parameters:

• event_type (str | None)

• event_description (str | None)

• severity (str | None)

• departments_affected (list[str])

• staff_type_affected (str | None)

• beds_offline (int | None)

• services_disrupted (list[str])

• regulatory_response (str | None)

event_type: str | None = None

event_description: str | None = None

severity: str | None = None

departments_affected: list[str]

staff_type_affected: str | None = None

beds_offline: int | None = None

services_disrupted: list[str]

regulatory_response: str | None = None

to_dict()

Convert the OtherData instance to a dictionary.

Return type:

dict

classmethod from_dict(data)

Create an OtherData instance from a dictionary.

Parameters:

data (dict)

Return type:

OtherData

class src.classes.vulnerability.Vulnerability(id, title, source_name, direct_link, subsector, date_accessed, date_published, content, exec_summary='', confidence_level=None, risk_level=None, geography_scope=None, start_date=None, end_date=None, resilience_or_mitigation_observed=None, subsector_data=None)

Class representing a healthcare vulnerability with both required and optional fields.

Parameters:

• id (str)

• title (str)

• source_name (str)

• direct_link (str)

• subsector (str)

• date_accessed (str)

• date_published (str)

• content (str)

• exec_summary (str)

• confidence_level (str | None)

• risk_level (str | None)

• geography_scope (str | None)

• start_date (str | None)

• end_date (str | None)

• resilience_or_mitigation_observed (str | None)

• subsector_data (SubsectorData | None)

id: str

title: str

source_name: str

direct_link: str

subsector: str

date_accessed: str

date_published: str

content: str

exec_summary: str = ''

confidence_level: str | None = None

risk_level: str | None = None

geography_scope: str | None = None

start_date: str | None = None

end_date: str | None = None

resilience_or_mitigation_observed: str | None = None

subsector_data: SubsectorData | None = None

to_dict()

Convert the Vulnerability instance to a dictionary.

Return type:

dict

classmethod from_dict(data)

Create a Vulnerability instance from a dictionary.

Parameters: - data (dict): A dictionary containing the vulnerability data.

Returns: - Vulnerability: An instance of the Vulnerability class populated with the provided data.

Parameters:

data (dict)

Return type:

Vulnerability

BERT Filtering (Deprecated)

Classify healthcare-related news items using a zero-shot BERT model.

This module is a library that provides helpers to run zero-shot classification on article text in raw text format (headline + excerpt). It is intended to be called via run_bert_inference from the pipeline’s ask_llm module.

Dependencies:

transformers (pipeline) torch (lazy-loaded on first device detection)

Example

from src.GDELT.BERT_filter import load_model, run_bert_inference

classifier = load_model() result = run_bert_inference({“title”: “Hospital breach”, “body”: “…”}, classifier)

src.GDELT.BERT_filter.get_device()

Detect the best available device for model inference.

Returns:

0 for CUDA, “mps” for Apple Silicon, or -1 for CPU.

Return type:

int | str

src.GDELT.BERT_filter.describe_model()

Return the BERT model identifier and device label without loading weights.

Return type:

tuple[str, str]

src.GDELT.BERT_filter.load_model(verbose=False)

Load the zero-shot classification pipeline.

Checks for a local finetuned model at FINETUNE_BERT_PATH. If found, loads it; otherwise warns and falls back to FALLBACK_MODEL_ID.

Returns:

Loaded zero-shot classification pipeline.

Return type:

transformers.Pipeline

Parameters:

verbose (bool)

src.GDELT.BERT_filter.run_bert_inference(data, classifier=None, verbose=False)

Classify a single article as a potential healthcare-related hit.

If the finetuned model is present, uses the model’s trained candidate labels and returns the predicted subsector directly. If only the base model is available, falls back to a threshold-based approach and returns “potential_hit” or “none”.

Parameters:

• data (dict) – Article payload. Expected keys: - “title” (str): Article headline. Missing/None treated as “”. - “body” (str): Article body/text. Missing/None treated as “”.

• classifier – Loaded transformers pipeline instance for inference. - If None, load_model() is called automatically (recommended for most callers). Pass explicitly only to override the default model, during testing or experimentation.

• verbose (bool)

Returns:

If finetuned model is present, one of: “drug_shortage”,

”medical_device_shortage”, “cyber_attack”, “natural_disaster”, “other”, or “none”. If falling back to base model, one of: “potential_hit” or “none”.

Return type:

str

RAG Server (Deprecated)

FastAPI RAG server for querying healthcare disruption incident reports.

This module provides a REST API that answers questions about healthcare crises (cyberattacks, supply shortages, natural disasters, staffing issues) using a retrieval-augmented generation (RAG) pipeline backed by ChromaDB and Ollama LLM.

Endpoints:

GET / : Serve the web UI (index.html). POST /chat : Ask a question and get an answer with source citations. GET /status : Check system readiness and database statistics.

The RAG chain uses HuggingFace embeddings for semantic search and Ollama for generating answers from retrieved incident reports.

src.RAG.server.get_chain()

Initialize and return the RAG chain and retriever (singleton pattern).

Lazily initializes the embedding model, ChromaDB connection, and LLM on first call, then caches them as module globals to avoid re-creation. Subsequent calls return the cached instances.

Returns:

(chain, retriever) where chain is a LangChain runnable that

takes a question and returns an answer, and retriever is a LangChain retriever that fetches the top-K similar documents.

Return type:

tuple

Raises:

RuntimeError – If ChromaDB has not been initialized (run ingest.py first).

class src.RAG.server.ChatRequest(*args, **kwargs)

Request model for the /chat endpoint.

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

question

The user’s question about healthcare disruptions.

Type:

str

class src.RAG.server.SourceDoc(*args, **kwargs)

Metadata for a source document retrieved by the RAG pipeline.

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

id

Unique identifier for the document.

Type:

str

title

Article title.

Type:

str

source_name

Source news outlet or publication.

Type:

str

direct_link

URL to the original article.

Type:

str

class src.RAG.server.ChatResponse(*args, **kwargs)

Response model for the /chat endpoint.

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

answer

The LLM’s generated answer to the question.

Type:

str

sources

Metadata for retrieved source documents.

Type:

list[SourceDoc]

model

Name of the LLM used (e.g., “gemma4:e4b”).

Type:

str

chunks_retrieved

Number of document chunks returned by the retriever.

Type:

int

src.RAG.server.serve_ui()

Serve the web UI (index.html) at the root path.

Returns:

The contents of index.html.

Return type:

HTMLResponse

Raises:

HTTPException – 404 if index.html is not found.

src.RAG.server.chat(req)

Retrieve relevant incident reports and generate an answer via RAG.

Uses the RAG chain to retrieve the top-K most relevant documents from ChromaDB and feed them to the LLM with a specialized healthcare disruption prompt. Returns the LLM’s answer and source metadata.

Parameters:

req (ChatRequest) – Request containing a question string.

Returns:

The answer, retrieved sources, model name, and chunk count.

Return type:

ChatResponse

Raises:

HTTPException – 400 if question is empty; 503 if ChromaDB not initialized; 500 for other RAG errors.

src.RAG.server.status()

Return system readiness status and database statistics.

Checks if ChromaDB has been initialized and counts the number of indexed documents. Returns configuration and availability information.

Returns:

A JSON object with keys:

• status (str): “ok” if DB is ready, “no_db” otherwise.

• db_ready (bool): Whether ChromaDB exists.

• records_indexed (int): Number of documents in the collection.

• llm_model (str): Name of the LLM in use.

• embed_model (str): Name of the embedding model in use.

• message (str): Human-readable status message.

Return type:

JSONResponse

Ingest and deduplicate healthcare disruption articles into ChromaDB.

This module loads JSON documents containing healthcare article metadata, converts them to text chunks with embeddings, and stores them in a vector database (ChromaDB) for semantic search. It includes duplicate detection using cosine similarity and optional deep merging of near-duplicate records.

Key features: - Semantic duplicate detection with configurable threshold. - Deep merging of similar records with matching subsectors. - Detailed logging of duplicate events. - Optional classification gate via LLM/BERT validation. - Chunking with configurable overlap for better context preservation.

Main entry point: ingest(filepath, …)

src.RAG.ingest.load_document(filepath)

Load and parse a JSON file containing healthcare article records.

The JSON file must contain a top-level object with a ‘sources’ key that maps to a list of record objects. Each record typically contains fields like title, content, source_name, subsector, etc.

Parameters:

filepath (str) – Path to the JSON file.

Returns:

List of record dictionaries from the ‘sources’ key.

Return type:

list[dict]

Raises:

• ValueError – If the file is not a JSON object or lacks a ‘sources’ key.

• FileNotFoundError – If the file does not exist.

• json.JSONDecodeError – If the file is not valid JSON.

src.RAG.ingest.record_to_text(record)

Convert a record dict to human-readable text for embedding and analysis.

Formats known fields (title, source, link, subsector, dates, body, summary) into labeled lines. Flattens subsector_data into readable key-value pairs. Also captures and warns about any unexpected fields not in the schema.

Parameters:

record (dict) – Article record with fields like title, content, subsector_data, etc.

Returns:

Formatted text with labeled fields, suitable for embeddings or

LLM processing. Empty string if the record has no useful content.

Return type:

str

src.RAG.ingest.build_documents(records)

Convert records to LangChain Document objects with chunking and metadata.

Each record is converted to text via record_to_text, then split into chunks (800 chars, 160 char overlap) for better embedding performance. Metadata (id, title, source, subsector, raw JSON) is attached to each chunk.

Parameters:

records (list[dict]) – List of article records.

Returns:

LangChain Document objects, one per chunk. Records that

yield no text are skipped.

Return type:

list[Document]

src.RAG.ingest.resolve_chroma_dir(diff_dir)

Resolve the ChromaDB directory path.

If diff_dir is provided, validates that it exists and is a directory, then returns the absolute path. If diff_dir is None, returns DEFAULT_CHROMA_DIR (which Chroma will create on first write).

Parameters:

diff_dir (str | None) – Override directory path, or None to use default.

Returns:

Absolute path to the ChromaDB directory.

Return type:

str

Raises:

SystemExit – If diff_dir is provided but doesn’t exist or isn’t a directory.

src.RAG.ingest.merge_records(existing, new)

Deep-merge two healthcare records with the same subsector.

Combines data from an existing record (already in DB) with a new incoming record. Merge strategy: - id: kept from existing (merged record replaces it). - title/content/exec_summary: concatenated with separator. - source_name/direct_link: joined with “ | “ when distinct. - dates: latest value (ISO strings sort lexically). - subsector: kept from existing (precondition: must match new). - subsector_data: deep-merged (lists extended + deduped, scalars favor existing). - Unknown fields: same policy as subsector_data scalars.

Parameters:

• existing (dict) – Record already in the database.

• new (dict) – Incoming record to merge.

Returns:

Merged record with combined data.

Return type:

dict

src.RAG.ingest.find_duplicate(db, record, threshold)

Find the most semantically similar record in the database.

Uses cosine distance (Chroma default, lower = closer) to find the best match. Returns a hit only if: - Distance is at or below the threshold. - The hit has the same subsector as the incoming record (subsector match). - The hit is a different record (different id).

Parameters:

• db (Chroma) – Vector database instance.

• record (dict) – Incoming record to search for duplicates against.

• threshold (float) – Maximum cosine distance to consider a match.

Returns:

(existing_record, distance) if a match is found,

None otherwise (including when the DB is empty).

Return type:

tuple[dict, float] | None

src.RAG.ingest.log_duplicate(chroma_dir, existing, new, merged, distance, threshold, action)

Log a duplicate detection event to <chroma_dir>/duplication_log.txt.

Appends a human-readable JSON block containing the existing record, incoming record, merged/proposed record, distance, threshold, and action. This log allows reviewers to inspect merges and re-ingest corrected data.

Parameters:

• chroma_dir (str) – Path to the ChromaDB directory.

• existing (dict) – The existing record found in the database.

• new (dict) – The incoming record being ingested.

• merged (dict) – The merged/proposed record.

• distance (float) – Cosine distance between the records.

• threshold (float) – The threshold used for duplicate detection.

• action (str) – Action taken (e.g., “merged”, “logged_only_subsector_mismatch”).

Returns:

Writes to the log file as a side effect.

Return type:

None

src.RAG.ingest.ingest(filepath, *, new_db=False, diff_dir=None, force=False, dup_threshold=0.44, use_bert=False)

Main ingestion pipeline: load, chunk, deduplicate, and index documents.

Loads healthcare article records from a JSON file, converts them to chunked LangChain documents, and indexes them in ChromaDB. Optionally detects and merges semantic duplicates. Per-record classification gate (LLM/BERT) can filter out non-disruption articles.

The pipeline loads and parses JSON records, initializes the HuggingFace embedding model, prepares the ChromaDB vector store, and then checks each record for duplicates before either merging or inserting it.

Parameters:

• filepath (str) – Path to input JSON file (must have ‘sources’ key).

• new_db (bool) – If True, delete existing ChromaDB before starting.

• diff_dir (str | None) – Override default ChromaDB directory.

• force (bool) – Skip semantic duplicate checks; insert all records directly.

• dup_threshold (float) – Cosine distance threshold for duplicate detection.

• use_bert (bool) – Enable BERT pre-screening before LLM validation.

Returns:

Prints pipeline progress and status to stdout.

Return type:

None