"""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.
Attributes:
- `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.
"""
import csv
import datetime
import json
import os
import re
import signal
import subprocess
import tempfile
import shutil
import time
import threading
import pandas as pd
from collections.abc import Callable, Iterable
from concurrent.futures import FIRST_COMPLETED, Future, wait
from pathlib import Path
from typing import TypeVar
import requests
from bs4 import BeautifulSoup
from .classes import SUBSECTOR_DATA_CLASSES, Vulnerability
from .logging_utils import get_file_logger
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
_CONFIG_PATH = _PROJECT_ROOT / "src" / "config" / "config.cfg"
_T = TypeVar("_T")
_SHUTDOWN_POLL_INTERVAL = 0.1
def _load_config_file() -> dict[str, str]:
"""
Load configuration values from the config.cfg file into a dictionary.
Returns:
dict[str, str]: A dictionary containing configuration key-value pairs. If the config file does not exist, returns an empty dictionary.
"""
values: dict[str, str] = {}
if not _CONFIG_PATH.exists():
return values
for raw_line in _CONFIG_PATH.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value.strip()
return values
_CONFIG = _load_config_file()
[docs]
def get_config_value(name: str, default: str | None = None) -> str | 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:
str | None: The value of the configuration variable if it exists and is not empty; otherwise, returns the provided default value.
"""
value = _CONFIG.get(name)
if value is None or value == "":
return default
return value
[docs]
def get_config_bool(name: str, default: bool = False) -> bool:
"""
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:
bool: 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.
"""
value = get_config_value(name)
if value is None:
return default
return value.lower() in {"true", "yes"}
[docs]
def get_config_int(name: str, default: int | None = None) -> int | 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:
int | None: The integer value of the configuration variable if it exists and is not empty; otherwise, returns the provided default value.
"""
value = get_config_value(name)
if value is None:
return default
try:
return int(value)
except ValueError:
return default
[docs]
def get_config_date(
name: str, default: datetime.date | None = None
) -> datetime.date | 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:
datetime.date | None: The parsed date if the configuration variable exists and is a valid ``YYYY-MM-DD`` string; otherwise, returns the provided default value.
"""
value = get_config_value(name)
if value is None:
return default
try:
return datetime.date.fromisoformat(value)
except ValueError:
return default
AI_URL = get_config_value("AI_URL", "http://localhost:11434/api/generate")
AI_MODEL = get_config_value("AI_MODEL", "llama3.2:latest")
MIN_BODY_CHARS_FOR_LLM = get_config_int("MIN_BODY_CHARS_FOR_LLM", 150) or 150
BODY_CHAR_LIMIT = get_config_int("BODY_CHAR_LIMIT", 4000) or 4000
LOG_FILE = _PROJECT_ROOT / "data" / "logs" / "shared_utils.log"
LOGGER = get_file_logger(__name__, LOG_FILE)
# temporary for now, to be removed later
READY_FOR_RAG_DIR = _PROJECT_ROOT / "data" / "processed"
NOISE_DIR = _PROJECT_ROOT / "data" / "noise"
VULNERABILITIES_DIR = _PROJECT_ROOT / "data" / "vulnerabilities"
_NOISE_PATTERNS = (
"ad",
"advert",
"promo",
"sidebar",
"related",
"newsletter",
"subscribe",
"comment",
"social",
"share",
"cookie",
)
_NOISE_RE = re.compile("|".join(_NOISE_PATTERNS), re.IGNORECASE)
_NOISE_SELECTOR = ",".join(f'[class*="{p}"]' for p in _NOISE_PATTERNS)
_BODY_BOILERPLATE_PATTERNS = re.compile(
r"skip to main content|copyright|terms of use|privacy policy|powered by|"
r"content management system|blox digital|subscribe|sign in|log in|search|menu|close",
re.IGNORECASE,
)
_TITLE_SITE_SUFFIX_RE = re.compile(r"(?:\s*[|–—]\s+[^|–—]+|\s-\s+[^-]+)$")
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
)
}
VULN_CSV_HEADER = [
"date_accessed",
"date_published",
"source_name",
"subsector",
"title",
"direct_link",
"exec_summary",
"content_preview",
]
NOISE_CSV_HEADER = [
"date_accessed",
"source_name",
"title",
"url",
"reason",
"body_preview",
]
LLM_SECTOR_FIELDS = [
"exec_summary",
"geography_scope",
"start_date",
"end_date",
"resilience_or_mitigation_observed",
]
SUBSECTOR_FIELDS = {
"drug_shortage": [
"drug_name",
"generic_name",
"manufacturer",
"dosage_form",
"shortage_reason",
"estimated_resolution_date",
"affected_regions",
"domestic_vs_foreign_dependency",
],
"medical_device_shortage": [
"device_name",
"device_category",
"manufacturer",
"manufacturer_country",
"shortage_reason",
"fda_recall_number",
"recall_class",
"affected_specialties",
"alternatives_available",
"estimated_resolution_date",
"domestic_vs_foreign_dependency",
],
"cyber_attack": [
"attack_type",
"threat_actor",
"individuals_affected",
"data_types_exposed",
"systems_affected",
"ransom_demanded_usd",
"ransom_paid",
"downtime_days",
"services_disrupted",
"law_enforcement_involved",
"hhs_breach_portal_listed",
],
"natural_disaster": [
"disaster_type",
"disaster_name",
"fema_declaration_id",
"category_magnitude",
"affected_facilities_count",
"evacuation_ordered",
"field_hospitals",
"beds_offline",
"facility_status",
"estimated_damage_usd",
"infrastructure_damage",
"services_disrupted",
],
"other": [
"event_type",
"event_description",
"severity",
"departments_affected",
"staff_type_affected",
"beds_offline",
"services_disrupted",
"regulatory_response",
],
}
[docs]
def 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:
requests.Response: The response object containing the web page content.
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.
"""
start_time = time.time()
resp = requests.get(
url, timeout=(connect_timeout, read_timeout), headers=HEADERS, stream=True
)
resp.raise_for_status()
content = bytearray()
for chunk in resp.iter_content(chunk_size=8192):
if time.time() - start_time > absolute_timeout:
resp.close()
raise requests.exceptions.Timeout(
f"Absolute timeout of {absolute_timeout}s exceeded"
)
if chunk:
content.extend(chunk)
resp._content = bytes(content)
LOGGER.debug("Successfully fetched URL: %s", url)
return resp
def _site_filename(site_name: str) -> str:
"""
Generates a sanitized site filename by trimming leading and trailing whitespace from the given site name.
Parameters:
site_name (str): The name of the site as a string.
Returns:
str: The trimmed site name with whitespace removed.
"""
return site_name.strip()
[docs]
def 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.
"""
READY_FOR_RAG_DIR.mkdir(parents=True, exist_ok=True)
NOISE_DIR.mkdir(parents=True, exist_ok=True)
VULNERABILITIES_DIR.mkdir(parents=True, exist_ok=True)
stem = _site_filename(site_name)
json_path = READY_FOR_RAG_DIR / f"{stem}.json"
if not json_path.exists():
json_path.write_text(json.dumps({"sources": []}, indent=4), encoding="utf-8")
LOGGER.debug("Created JSON file for site %s at %s", site_name, json_path)
noise_path = NOISE_DIR / f"{stem}.csv"
if not noise_path.exists():
with open(noise_path, "w", newline="", encoding="utf-8") as f:
csv.writer(f).writerow(NOISE_CSV_HEADER)
LOGGER.debug("Created noise CSV file for site %s at %s", site_name, noise_path)
vulnerabilities_path = VULNERABILITIES_DIR / f"{stem}.csv"
if not vulnerabilities_path.exists():
with open(vulnerabilities_path, "w", newline="", encoding="utf-8") as f:
csv.writer(f).writerow(VULN_CSV_HEADER)
LOGGER.debug(
"Created vulnerabilities CSV file for site %s at %s",
site_name,
vulnerabilities_path,
)
def _content_preview(body: str | None) -> str:
"""
Parameter:
body (str): the entire body of an article
Returns:
First 250 characters of the article (success)
"""
return (body or "")[:250].replace("\n", " ")
def _top_row_matches(
csv_path: Path,
title: str,
body_snippet: str,
preview_column: str,
) -> bool:
"""
Not documenting on purpose, this function will likely be deleted in the near future
"""
if not csv_path.exists():
LOGGER.debug("CSV file %s does not exist, cannot match top row", csv_path)
return False
with open(csv_path, "r", newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
try:
first_row = next(reader)
except StopIteration:
LOGGER.debug("CSV file %s is empty, cannot match top row", csv_path)
return False
if first_row.get("title", "") != title:
LOGGER.debug(
"Top row title %s does not match incoming title %s in file %s",
first_row.get("title", ""),
title,
csv_path,
)
return False
incoming_preview = _content_preview(body_snippet)
if first_row.get(preview_column, "") != incoming_preview:
LOGGER.warning("Body preview differs for title %s", title)
return True
[docs]
def is_known_article(site_name: str, title: str, body_snippet: str) -> bool:
"""
Not documenting on purpose, this function will likely be deleted in the near future
"""
site = _site_filename(site_name)
if _top_row_matches(
VULNERABILITIES_DIR / f"{site}.csv", title, body_snippet, "content_preview"
):
return True
if _top_row_matches(NOISE_DIR / f"{site}.csv", title, body_snippet, "body_preview"):
return True
return False
[docs]
def prepend_vuln_csv(site_name: str, new_rows: list[list[str]]) -> None:
"""
Not documenting on purpose, this function will likely be deleted in the near future
"""
if not new_rows:
return
csv_path = VULNERABILITIES_DIR / f"{_site_filename(site_name)}.csv"
existing_data_rows: list[list[str]] = []
if csv_path.exists():
with open(csv_path, "r", newline="", encoding="utf-8") as f:
reader = csv.reader(f)
try:
next(reader) # drop the existing header
except StopIteration:
pass
existing_data_rows = list(reader)
fd, tmp_path = tempfile.mkstemp(
prefix=f"{_site_filename(site_name)}.",
suffix=".csv.tmp",
dir=str(VULNERABILITIES_DIR),
)
try:
with os.fdopen(fd, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(VULN_CSV_HEADER)
writer.writerows(new_rows)
writer.writerows(existing_data_rows)
os.replace(tmp_path, csv_path)
except Exception:
LOGGER.error(
"Failed to prepend to vulnerabilities CSV for site %s: %s",
site_name,
exc_info=True,
)
if os.path.exists(tmp_path):
os.remove(tmp_path)
raise
[docs]
def prepend_noise_csv(site_name: str, new_rows: list[list[str]]) -> None:
"""
Not documenting on purpose, this function will likely be deleted in the near future
"""
if not new_rows:
return
csv_path = NOISE_DIR / f"{_site_filename(site_name)}.csv"
existing_data_rows: list[list[str]] = []
if csv_path.exists():
with open(csv_path, "r", newline="", encoding="utf-8") as f:
reader = csv.reader(f)
try:
next(reader) # drop the existing header
except StopIteration:
pass
existing_data_rows = list(reader)
fd, tmp_path = tempfile.mkstemp(
prefix=f"{_site_filename(site_name)}.",
suffix=".csv.tmp",
dir=str(NOISE_DIR),
)
try:
with os.fdopen(fd, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(NOISE_CSV_HEADER)
writer.writerows(new_rows)
writer.writerows(existing_data_rows)
os.replace(tmp_path, csv_path)
except Exception:
LOGGER.error(
"Failed to prepend to noise CSV for site %s: %s", site_name, exc_info=True
)
if os.path.exists(tmp_path):
os.remove(tmp_path)
raise
[docs]
def prepend_json_sources(site_name: str, new_vulns: list[Vulnerability]) -> None:
"""
Not documenting on purpose, this function will likely be deleted in the near future
"""
if not new_vulns:
return
json_path = READY_FOR_RAG_DIR / f"{_site_filename(site_name)}.json"
if json_path.exists():
data = json.loads(json_path.read_text(encoding="utf-8"))
else:
data = {"sources": []}
new_dicts = [v.to_dict() for v in new_vulns]
data["sources"] = new_dicts + data.get("sources", [])
fd, tmp_path = tempfile.mkstemp(
prefix=f"{_site_filename(site_name)}.",
suffix=".json.tmp",
dir=str(READY_FOR_RAG_DIR),
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, indent=4, ensure_ascii=False)
os.replace(tmp_path, json_path)
except Exception:
LOGGER.error(
"Failed to prepend to JSON sources for site %s: %s",
site_name,
exc_info=True,
)
if os.path.exists(tmp_path):
os.remove(tmp_path)
raise
def _extract_title_from_soup(soup: BeautifulSoup, fallback_url: str) -> str:
"""Extract and clean the page title from a parsed BeautifulSoup tree.
Looks for the HTML ``<title>`` tag, strips common site-name suffixes
(e.g. " | Reuters", " - NBC News"), and falls back to *fallback_url*
when no usable title is found.
Args:
soup: A parsed BeautifulSoup document.
fallback_url: The value returned when no ``<title>`` tag is present
or when the tag text is empty.
Returns:
str: Cleaned page title, or *fallback_url*.
"""
title_tag = soup.find("title")
if title_tag:
raw = title_tag.get_text(strip=True)
if raw:
cleaned = _TITLE_SITE_SUFFIX_RE.sub("", raw).strip()
return cleaned if cleaned else raw
return fallback_url
def _extract_body_from_soup(soup: BeautifulSoup, url: str) -> str:
"""Extract the main article text from a parsed BeautifulSoup tree.
Removes common non-content tags and noisy selectors (ads, sidebars,
footers), then returns the concatenated paragraph text when available.
Args:
soup: A parsed BeautifulSoup document. **Note:** this function
mutates the tree by decomposing non-content elements.
url: The original URL (used only for log messages).
Returns:
str: Cleaned article text, or an empty string if no body content
is found.
"""
# Strip non-content tags
for tag in soup(
[
"script",
"style",
"noscript",
"iframe",
"form",
"header",
"footer",
"nav",
"aside",
]
):
tag.decompose()
# Strip noise by class
for el in soup.select(_NOISE_SELECTOR):
el.decompose()
# Strip noise by id
for el in soup.find_all(id=_NOISE_RE):
el.decompose()
# Pick the best content container, in order of preference
main = None
for candidate in (
soup.find("article"),
soup.find("main"),
soup.find(attrs={"role": "main"}),
soup.body,
soup,
):
if candidate and candidate.get_text(strip=True):
main = candidate
break
if main is None:
LOGGER.warning("No body found for URL %s", url)
return ""
def _filtered_text_chunks(node) -> list[str]:
"""
Extract text chunks from a BeautifulSoup node, filtering out empty strings and boilerplate.
Parameters:
node (BeautifulSoup): The BeautifulSoup node to extract text from.
Returns:
list[str]: A list of filtered text chunks.
"""
chunks: list[str] = []
# Use stripped_strings to get text without extra whitespace, then filter out empty and boilerplate chunks
for chunk in node.stripped_strings:
normalized = chunk.strip()
if not normalized:
continue
if _BODY_BOILERPLATE_PATTERNS.search(normalized):
continue
chunks.append(normalized)
LOGGER.debug(
"Extracted %d filtered text chunks from node in URL %s", len(chunks), url
)
return chunks
# Prefer paragraph text (gives cleaner article body across sites.)
# Fall back to filtered descendant text when no <p> tags are present.
content_chunks = [
p.get_text(" ", strip=True)
for p in main.find_all("p")
if p.get_text(" ", strip=True)
and not _BODY_BOILERPLATE_PATTERNS.search(p.get_text(" ", strip=True))
]
if not content_chunks:
content_chunks = _filtered_text_chunks(main)
if content_chunks:
LOGGER.debug(
"Extracted %d content chunks from URL %s", len(content_chunks), url
)
return "\n\n".join(content_chunks)
LOGGER.debug("No content chunks found after filtering for URL %s", url)
return ""
[docs]
def get_title(url: str) -> str:
"""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.
Args:
url (str): The URL to fetch. ``https://`` is prepended if the scheme
is missing.
Returns:
str: Cleaned page title, or the URL on failure.
"""
if not url:
return url
if not url.startswith(("http://", "https://")):
url = "https://" + url
try:
resp = requests.get(url, timeout=30, headers=HEADERS)
resp.raise_for_status()
except requests.RequestException as e:
LOGGER.warning("Failed to fetch URL %s: %s", url, e)
return url
# should solve the unicode error
except UnicodeDecodeError as e:
LOGGER.warning("Unable to decode URL (not in UTF 8) %s: %s", url, e)
return url
soup = BeautifulSoup(resp.text, "html.parser")
return _extract_title_from_soup(soup, url)
[docs]
def get_body(url: str) -> str:
"""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.
Args:
url (str): The URL to fetch. If the scheme is missing, `https://` is
prepended.
Returns:
str: Cleaned article text, or an empty string on error or if no body
content is found.
Notes:
- This is a heuristic extractor and may not work for all sites.
- The returned body may be long; callers should truncate if needed.
"""
if not url:
LOGGER.warning("Empty URL provided to get_body()")
return ""
# Normalize URL
if not url.startswith(("http://", "https://")):
url = "https://" + url
# Fetch
try:
resp = requests.get(url, timeout=30, headers=HEADERS)
resp.raise_for_status()
except requests.RequestException as e:
LOGGER.warning("Failed to fetch URL %s: %s", url, e)
return ""
# should solve the unicode error
except UnicodeDecodeError as e:
LOGGER.warning("Unable to decode URL (not in UTF 8) %s: %s", url, e)
return ""
soup = BeautifulSoup(resp.text, "html.parser")
return _extract_body_from_soup(soup, url)
[docs]
def get_body_and_title(url: str) -> tuple[str, str]:
"""Fetch a URL once and return both the article body and cleaned title.
Combines the work of :func:`get_body` and :func:`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
:func:`_extract_body_from_soup` mutates the soup tree by decomposing
non-content elements (which could remove the ``<title>`` tag).
Args:
url: 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.
"""
if not url:
LOGGER.warning("Empty URL provided to get_body_and_title()")
return "", url or ""
# Normalize URL
if not url.startswith(("http://", "https://")):
url = "https://" + url
if shutdown_requested():
return "", url
# Fetch once
try:
resp = requests.get(url, timeout=30, headers=HEADERS)
resp.raise_for_status()
except requests.RequestException as e:
LOGGER.warning("Failed to fetch URL %s: %s", url, e)
return "", url
# should solve the unicode error
except UnicodeDecodeError as e:
LOGGER.warning("Unable to decode URL (not in UTF 8) %s: %s", url, e)
return "", url
soup = BeautifulSoup(resp.text, "html.parser")
# Extract title first — _extract_body_from_soup mutates the tree.
title = _extract_title_from_soup(soup, url)
body = _extract_body_from_soup(soup, url)
return body, title
_classifier = None
def _run_bert(title: str, body: str, verbose: bool = False) -> str:
"""
Run BERT inference on an article. Returns the predicted subsector string or "none".
Loads the classifier once on first call and reuses it for subsequent articles.
"potential_hit" from the base model fallback is normalized to "other".
"""
global _classifier
try:
from src.GDELT.BERT_filter import run_bert_inference, load_model
except ImportError as exc:
LOGGER.error("Failed to import src.GDELT.BERT_filter module: %s", exc)
raise RuntimeError("src.GDELT.BERT_filter module not found") from exc
if _classifier is None:
_classifier = load_model(verbose=verbose)
if _classifier is None:
LOGGER.warning("BERT classifier failed to load, skipping")
return "none"
try:
result = run_bert_inference(
{"title": title, "body": body}, _classifier, verbose=verbose
)
except Exception as e:
LOGGER.warning("BERT inference failed: %s", e)
return "none"
if result == "potential_hit":
return "other"
return result
[docs]
def ai_check_validation(
title, body, use_bert=False, verbose: bool = False, port: int = 11434
) -> tuple[bool, str]:
"""
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
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".
"""
body_text = body or ""
if len(body_text) < MIN_BODY_CHARS_FOR_LLM:
LOGGER.info(
"Skipping LLM validation for title %s because body length %d is below %d characters",
title,
len(body_text),
MIN_BODY_CHARS_FOR_LLM,
)
return False, "Body too short for LLM review"
if shutdown_requested():
raise LLMUnavailableError("shutdown requested")
if use_bert:
bert_subsector = _run_bert(title, body, verbose=verbose)
if bert_subsector == "none":
LOGGER.info("BERT rejected article with title %s", title)
return False, "BERT: unrelated news"
LOGGER.info("BERT flagged article with title %s as %s", title, bert_subsector)
prompt = f"""
You are a strict Healthcare Operations Auditor. Your ONLY job is to flag articles that describe a REAL, ALREADY-OCCURRING healthcare disruption or a CONFIRMED breach at a named healthcare entity.
DEFAULT TO NO. Reject the article unless the evidence is explicit, named, and concrete. The vast majority of healthcare news is NOT a disruption.
===== ACCEPT (mark YES) ONLY IF (A) OR (B) IS TRUE =====
(A) ACTIVE CARE DISRUPTION — the article states that a NAMED facility (hospital, clinic, pharmacy, lab, healthcare network) is CURRENTLY or RECENTLY:
- Diverting ambulances, cancelling surgeries, or turning patients away
- Operating on downtime / paper procedures because EHR is offline
- Suspending services or evacuating due to fire, flood, storm, or other physical event
- Physically out of a specific drug or medical device that patients need now (real supply outage, not pricing or formulary debate)
- Cut off from operations by a workforce strike, power outage, or other concrete event
(B) CONFIRMED HEALTHCARE BREACH / CYBERATTACK — this rule has TWO parts; if BOTH are true, the article is YES.
Part 1: The victim is a NAMED healthcare entity. ALL of these qualify as healthcare entities for this rule:
* hospitals, clinics, health systems, physician groups
* pharmacies (retail or hospital)
* health insurers / payers / PBMs
* MEDICAL DEVICE MANUFACTURERS (e.g. Stryker, TriMed, Medtronic) — yes, they count
* pharma manufacturers
* healthcare-specific software / EHR / billing vendors (e.g. Epic, Change Healthcare)
* clinical labs and diagnostic companies
Part 2: The incident has ALREADY HAPPENED — confirmed by the entity, a regulator, an HHS breach notice, an SEC 8-K, or a public breach disclosure. Any ONE of the following counts:
* ransomware / intrusion / unauthorized access confirmed
* PHI or patient records exposed, exfiltrated, or encrypted
* Data security incident formally disclosed by the entity
* Operational systems impacted by the attack
If Part 1 AND Part 2 are both true → YES, subsector "cyber_attack". This applies even if the article does NOT describe care being stopped. A confirmed PHI breach at a healthcare entity IS the disruption.
===== CONCRETE YES EXAMPLES (these MUST be marked YES) =====
- "Signature Healthcare diverts ambulances amid cyberattack" — named hospital, current diversion -> YES, cyber_attack
- "TriMed (orthopedic implant maker) confirms data breach exposing patient PHI" — named device manufacturer, confirmed breach with PHI -> YES, cyber_attack
- "Acme Pharma halts production of injectable epinephrine; pharmacies report shortage" — named drug, real supply outage -> YES, drug_shortage
- "Hurricane evacuates Memorial Hospital; ER closed" — named facility, current closure -> YES, natural_disaster
- "Nurse strike at St. Jude shuts down elective surgeries" — named facility, current stoppage -> YES, other
===== REJECT (mark NO) — these are ALL noise =====
- Funding rounds, valuations, IPOs, M&A, partnerships, commercial deals, earnings
- Product launches, AI tool / chatbot debuts, software releases, roadmaps, strategy announcements
- Surveys, statistics, annual / quarterly trend reports (e.g. "FBI IC3 annual report", "burnout survey", "AI adoption survey")
- Government policy, legislation, regulation, payment-rate changes, prior-auth rules, value-based-care models
- Lawsuits, court rulings, legal opinions, settlements — UNLESS the article describes an actual ongoing care stoppage caused by them
- Research, clinical trials, drug discovery, efficacy comparisons (e.g. "drug X is healthier than drug Y", "GLP-1 helps migraines")
- Drug pricing, formulary changes, access programs, TrumpRx / Medicare deals — without an actual supply outage
- Cyber THREATS / advisories / vulnerabilities not yet exploited against a named victim ("CISA warns…", "researchers discover bug", "hardening guidance")
- Cyber attacks on entities OUTSIDE healthcare (generic router malware, espionage campaigns, non-healthcare ransomware)
- Workforce / burnout / compensation trends without a current named-facility care stoppage
- Crimes, accidents, or arrests that merely INVOLVE a healthcare object or location but do NOT stop care or operations at a named facility — e.g. a stolen ambulance, theft of medical equipment, a car crash involving an ambulance, a shooting or assault in a hospital parking lot, vandalism. The word "ambulance" or "hospital" appearing in the article is NOT enough; the article must describe care or operations actually being halted.
- Individual human-interest or patient stories (one person's illness, death, long wait, or recovery) without a named facility halting operations
- Interviews, executive profiles, conferences, op-eds, opinion pieces
- Anything hedged with "potential", "could", "may affect", "future risk", "expected to"
===== CONCRETE NO EXAMPLES (these MUST be marked NO) =====
- "Man steals an ambulance and leads police on a chase" — a crime involving a vehicle; no named facility stopped care -> NO
- "Shooting in hospital parking lot; operations continue normally" — crime at a location, care not disrupted -> NO
- "Nurses vote to authorize a possible strike; no date set" — hedged future threat, no current stoppage -> NO
- "Family mourns father who died awaiting a transplant" — individual patient story, no facility disruption -> NO
===== JSON OUTPUT CONTRACT — FOLLOW EXACTLY =====
Respond with EXACTLY this JSON shape and nothing else:
{{
"analysis": "One factual sentence: name the entity and the impact, OR state the reason for rejection.",
"is_operational_disruption": boolean,
"subsector": "drug_shortage" | "medical_device_shortage" | "cyber_attack" | "natural_disaster" | "other" | "none"
}}
HARD RULES on subsector:
- If is_operational_disruption is false → subsector MUST be "none".
- If is_operational_disruption is true → subsector MUST be EXACTLY one of: "drug_shortage", "medical_device_shortage", "cyber_attack", "natural_disaster", "other". NEVER "none". NEVER any other string.
- "other" is reserved for confirmed disruptions that genuinely don't fit the four named categories (e.g. workforce strike at a named hospital, power outage shutting down a named facility). Do NOT use "other" as a fallback for marginal articles — when in doubt, mark NO.
DECISION CHECK before you answer:
If your analysis sentence describes a confirmed cyberattack, ransomware, breach, PHI exposure, drug shortage, device shortage, evacuation, or care stoppage at a NAMED healthcare entity, you MUST set is_operational_disruption to true and pick a non-"none" subsector. Your boolean MUST match the facts in your analysis sentence — never say "confirmed breach" in analysis and false in the boolean.
TITLE: {title}
EXCERPT: {body}
"""
try:
url = f"http://localhost:{port}/api/generate"
resp = requests.post(
url,
json={
"model": AI_MODEL,
"prompt": prompt,
"stream": False,
"format": "json",
"options": {
"temperature": 1,
"top_k": 64,
"top_p": 0.95,
"num_ctx": 4096,
},
},
timeout=60,
)
LOGGER.debug("HTTP status %d", resp.status_code)
resp.raise_for_status()
raw_response = resp.json().get("response", "{}")
LOGGER.debug(
"Validation LLM raw response title=%s raw_response=%s",
title,
raw_response,
)
data = json.loads(raw_response)
LOGGER.debug("Parsed JSON for title %s: %s", title, data)
is_threat = data.get("is_operational_disruption", False)
# Use subsector if it's a threat, otherwise use the analysis as the "reason"
if isinstance(is_threat, str):
is_threat = is_threat.upper() != "NO"
else:
is_threat = bool(is_threat)
detail = (
data.get("subsector", "none")
if is_threat
else data.get("analysis", "No impact detected")
)
return is_threat, detail
# used for timeout / connection refused / HTTP errors
# we should wrap all ai_check_validation calls in a try/catch loop
# to re try on these vs handling them as noise
except requests.exceptions.RequestException as e:
LOGGER.warning("LLM call failed for %s: %s", title, e)
raise LLMUnavailableError(f"Ollama validation call failed: {e}") from e
except Exception as e:
LOGGER.warning("Error parsing AI response for title %s: %s", title, e)
return False, "Parsing Error"
[docs]
class MissingSubsectorFieldsError(ValueError):
"""Raised when a subsector has no configured extraction fields."""
[docs]
class model_unavailable_error(RuntimeError):
"""
Raised when configured Ollama model is unavailable
"""
[docs]
class LLMUnavailableError(RuntimeError):
"""
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.
"""
checked_ollama_models: set[str] = set()
[docs]
def ensure_model_available(model: str = AI_MODEL) -> None:
if model in checked_ollama_models:
return
try:
result = subprocess.run(
["ollama", "list"],
capture_output=True,
check=False,
text=True,
timeout=15,
)
except FileNotFoundError as exc:
raise model_unavailable_error(
"[ERROR] Ollama CLI not found.\nInstall and make sure 'ollama' is on PATH"
) from exc
except subprocess.SubprocessError as exc:
raise model_unavailable_error(
"[ERROR] Could not query Ollama models.\n"
"Make sure Ollama is running, then try again"
) from exc
if result.returncode != 0:
raise model_unavailable_error(
"[ERROR] Could not query Ollama models.\n"
"Make sure Ollama is running, then try again"
)
installed_models = {
line.split()[0]
for line in result.stdout.splitlines()[1:]
if line.strip() and line.split()
}
if model not in installed_models:
raise model_unavailable_error(
f"[ERROR] Model '{model}' not found in Ollama. Make sure Ollama is running.\n"
f"Run: ollama pull {model}"
)
checked_ollama_models.add(model)
[docs]
def run_clean():
"""Clean all GDELT-generated data so the pipeline starts fresh.
Delegates to :func:`scripts.clean_gdelt.run_clean` which owns the
canonical implementation. This wrapper exists for backward compatibility.
"""
# Inline import to avoid circular dependency and keep the scripts package
# out of the default import path for shared_utils.
import importlib
mod = importlib.import_module("scripts.clean_gdelt")
mod.run_clean()
[docs]
def clear_directory(directory: Path) -> None:
"""
Delete all files and subdirectories inside a directory.
Parameters:
directory: The path to the directory to clear.
"""
if not directory.exists():
LOGGER.debug("Directory does not exist, skipping clear: %s", directory)
return
# Iterate over all items in the directory and remove them
for item in directory.iterdir():
try:
if item.is_file() or item.is_symlink():
item.unlink()
elif item.is_dir():
shutil.rmtree(item)
except Exception as exc:
LOGGER.warning("Failed to remove %s: %s", item, exc)
[docs]
def df_dup(
dfs: list[pd.DataFrame], verbose: bool = False
) -> tuple[pd.DataFrame, list[str]]:
"""
Deduplicates a lits of dataframes, used to make one single csv write by scooper
Args:
dfs: list of dataframes with the same schema
verbose: prints messages when True
Returns:
One unique DataFrame a list of duplicate titles
"""
valid_dfs = [df for df in dfs if df is not None and not df.empty]
if not valid_dfs:
if verbose:
print("[WARNING]: All frames were None")
return pd.DataFrame(), []
combined = pd.concat(valid_dfs, ignore_index=True)
key = [c for c in ("source_name", "title") if c in combined.columns]
if not key:
return combined.drop_duplicates(ignore_index=True), []
dup_titles: list[str] = []
if "title" in combined.columns:
dup_mask = combined.duplicated(subset=key, keep="first")
dup_titles = combined.loc[dup_mask, "title"].dropna().unique().tolist()
deduped = combined.drop_duplicates(subset=key, keep="first", ignore_index=True)
return deduped, dup_titles
[docs]
def update_csv(df: pd.DataFrame, path: Path, verbose: bool = False) -> None:
"""
Appends a csv with a given DataFrame. This is used in scooper after calling 'df_dup'
Args:
df: DataFrame to append
path: CSV path
verbose: used to print messages if true
"""
if df is None or df.empty:
if verbose:
print("[WARNING]df is not usable (it is empty or None)")
return
path.parent.mkdir(parents=True, exist_ok=True)
write_header = not path.exists() or path.stat().st_size == 0
if write_header and verbose:
print("Writing headers, file was empty or did not exist")
df.to_csv(
path,
mode="a",
header=write_header,
index=False,
date_format="%Y-%m-%d",
)
[docs]
def update_json(vulns: list[Vulnerability], path: str) -> None:
"""
Used to write a list of Vulnerabilities into a given path
Args:
vulns: vulnerabilities to add to the file.
path: destination *.json file (created if it does not exist)
"""
clean_vulns: list[Vulnerability] = []
seen: set[tuple[str, str]] = set()
for vuln in vulns:
inst = (vuln.title, vuln.source_name)
if inst in seen:
continue
seen.add(inst)
clean_vulns.append(vuln)
file_path = Path(path)
if file_path.exists():
sources = json.loads(file_path.read_text(encoding="utf-8"))["sources"]
else:
sources = []
sources.extend(vuln.to_dict() for vuln in clean_vulns)
with open(file_path, "w", encoding="utf-8") as json_file:
json.dump({"sources": sources}, json_file, indent=2, ensure_ascii=False)
DEBUG_DIR = _PROJECT_ROOT / "data" / "noise"
[docs]
class NoiseCollector:
"""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: Destination JSON file (e.g. ``data/noise/debug_noise_gdelt.json``).
"""
def __init__(self, output_path: Path) -> None:
self.output_path = output_path
self.records: list[dict] = []
[docs]
def add(
self,
*,
url: str,
title: str,
source: str,
reason: str,
body_preview: str = "",
stage: str = "",
) -> None:
"""Append one noise record.
Parameters:
url: The article URL that was rejected.
title: The article title.
source: Pipeline source name (e.g. ``"GDELT"``, ``"CyberScoop"``).
reason: Human-readable rejection reason.
body_preview: First 250 characters of the article body.
stage: Pipeline stage that rejected the article (e.g.
``"already_seen"``, ``"validation"``, ``"extraction"``).
"""
self.records.append(
{
"url": url,
"title": title,
"source": source,
"reason": reason,
"body_preview": body_preview[:250],
"stage": stage,
"timestamp": datetime.datetime.now().isoformat(timespec="seconds"),
}
)
[docs]
def flush(self) -> Path | None:
"""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.
"""
if not self.records:
return None
self.output_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.output_path, "w", encoding="utf-8") as f:
json.dump(
{"noise_records": self.records, "total": len(self.records)},
f,
ensure_ascii=False,
indent=2,
)
LOGGER.info(
"Wrote %d debug noise records to %s",
len(self.records),
self.output_path,
)
return self.output_path
_shutdown = threading.Event()
[docs]
def shutdown_requested() -> bool:
"""Check if a shutdown event has been triggered."""
return _shutdown.is_set()
[docs]
def install_handler() -> None:
"""Install a SIGINT handler that sets a global shutdown event."""
signal.signal(signal.SIGINT, lambda *_: _shutdown.set())
[docs]
def pause_if_shutdown(stats) -> bool:
"""
Check if a shutdown event has been triggered and pause the pipeline if so.
Args:
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.
"""
if _shutdown.is_set():
stats.paused = True
return True
return False
[docs]
def request_pause(stats) -> None:
"""
Request a pause in the pipeline by setting a global shutdown event.
Args:
stats: An object with a 'paused' attribute to indicate the paused state.
"""
stats.paused = True
_shutdown.set()
[docs]
def collect_as_completed(
futures: Iterable[Future[_T]],
on_result: Callable[[_T], None],
) -> None:
"""Collect future results, returning promptly once shutdown is requested.
Args:
futures: An iterable of Future objects to collect results from.
on_result: A callback function to process each result as it becomes available.
"""
pending = set(futures)
try:
# Wait for futures to complete, but check for shutdown requests periodically.
while pending and not _shutdown.is_set():
done, pending = wait(
pending,
timeout=_SHUTDOWN_POLL_INTERVAL,
return_when=FIRST_COMPLETED,
)
# Process results from completed futures.
for future in done:
on_result(future.result())
# If shutdown was requested, process any completed futures without blocking.
for future in pending:
if future.done() and not future.cancelled():
try:
on_result(future.result())
except Exception:
LOGGER.debug(
"Skipping result from interrupted worker", exc_info=True
)
except KeyboardInterrupt:
_shutdown.set()
[docs]
def shutdown_executor(executor) -> None:
"""Shut down an executor without waiting when shutdown was requested."""
executor.shutdown(wait=not _shutdown.is_set(), cancel_futures=_shutdown.is_set())
[docs]
def exit_if_shutdown(code: int = 0) -> int:
"""
Exit the process immediately after shutdown cleanup, without joining workers.
Args:
code: Exit code to return. Defaults to 0.
Returns:
The exit code, which can be used for logging or further processing.
"""
if _shutdown.is_set():
os._exit(code)
return code
CORVID