piedomains package¶

Submodules¶

piedomains.api module¶

Modern, intuitive API for piedomains domain classification.

This module provides a clean, class-based interface for domain content classification with support for text analysis, image analysis, and historical archive.org snapshots.

class piedomains.api.DomainClassifier(cache_dir=None)[source]¶

Bases: object

Main interface for domain content classification.

Supports multiple classification approaches: - Traditional ML: Text-based, image-based, and combined classification - Modern AI: LLM-based classification with multimodal support - Historical analysis via archive.org snapshots

Example (Traditional ML):

>>> classifier = DomainClassifier()
>>> run = classifier.classify(["google.com", "facebook.com"])
>>> for result in run["results"]:
...     print(f"{result['domain']}: {result['category']} ({result['confidence']:.3f})")
google.com: search (0.892)
facebook.com: socialnet (0.967)

# Historical analysis >>> run = classifier.classify([“google.com”], archive_date=”20200101”) >>> result = run[“results”][0] >>> print(f”Archive: {result[‘category’]} from {result[‘date_time_collected’]}”)

Example (LLM-based):

>>> classifier = DomainClassifier()
>>> classifier.configure_llm(
...     provider="openai",
...     model="gpt-4o",
...     api_key="sk-...",
...     categories=["news", "shopping", "social", "tech"]
... )
>>> results = classifier.classify_by_llm(["cnn.com"])
>>> print(f"LLM: {results[0]['category']} - {results[0]['reason']}")

Example (Separated workflow):

>>> collector = DataCollector()
>>> collection = collector.collect(["example.com"])
>>> text_results = classifier.classify_from_collection(collection, method="text")
>>> image_results = classifier.classify_from_collection(collection, method="images")
>>> # Same collected content, different classification approaches

JSON Output Schema:

The core classification methods return a dictionary containing results and a run report. Each row in results has a consistent structure:

Collection Data Schema (from collect_content):

{
    "collection_id": str,       # Unique identifier for collection
    "timestamp": str,           # ISO 8601 collection timestamp
    "config": {
        "cache_dir": str,       # Cache directory path
        "archive_date": str,    # Archive.org date (YYYYMMDD) or null
        "fetcher_type": str,    # "live" or "archive"
        "max_parallel": int     # Parallel fetch limit
    },
    "domains": [               # List of domain results
        {
            "url": str,         # Original input URL/domain
            "domain": str,      # Parsed domain name
            "text_path": str,   # HTML file, relative to cache_dir
            "image_path": str,  # Screenshot, relative to cache_dir
            "date_time_collected": str,  # ISO 8601 timestamp
            "fetch_success": bool,  # Whether collection succeeded
            "cached": bool,     # Whether data came from cache
            "error": str,       # Error if fetch_success is false
            "title": str,       # Page title (optional)
            "meta_description": str  # Meta description (optional)
        }
    ],
    "summary": {
        "total_domains": int,   # Total domains requested
        "successful": int,      # Successfully collected
        "failed": int           # Failed collections
    }
}

Classification Result Schema (from classify methods):

[
    {
        "url": str,             # Original input URL/domain
        "domain": str,          # Parsed domain name
        "text_path": str,       # Path to HTML file
        "image_path": str,      # Path to screenshot
        "date_time_collected": str,  # ISO 8601 timestamp
        "model_used": str,      # e.g. "text/shallalist_ml"
        "category": str,        # Highest-scoring category
        "confidence": float,    # Its probability (0.0-1.0)
        "categories": [         # Every category above the threshold,
            {                   # highest first; >=1 entry when classified,
                "category": str,      # [] when the row failed.
                "probability": float
            }
        ],
        "reason": str,          # LLM reasoning (null for ML models)
        "error": str,           # Error if classification failed
        "raw_predictions": dict,  # Full probability distribution

        # Combined classification specific fields:
        "text_category": str,   # Text-only prediction
        "text_confidence": float,   # Text confidence
        "image_category": str,  # Image-only prediction
        "image_confidence": float   # Image confidence
    }
]

Supported Categories:

The 44 categories in piedomains.constants.classes: adult, alcohol, automobile, dating, downloads, drugs, education, finance, fortunetelling, forum, gamble, government, cooking, games, gardening, pets, homestyle, hospitals, imagehosting, isp, jobsearch, library, military, movies, music, news, politics, radiotv, realestate, humor, restaurants, sports, travel, wellness, religion, science, searchengines, shopping, socialnet, urlshortener, weapons, webmail, parked, unavailable.

Two kinds of absence, for two different reasons.

Categories describing how a site is hosted or monetised (adv, tracker, spyware, redirector) are absent because a page does not state them. So are those asking about delivery mechanism or legality rather than subject: games split by whether they are played online, radio split by whether it is broadcast, downloads split by whether the licence permitted them.

parked and unavailable are present for the mirror-image reason. A domain that resolves to a for-sale page or a server autoindex has no site to classify, and saying so is both plainly readable from the text and the answer a caller can act on.

The categories are NOT mutually exclusive, and cannot be while they are one flat list. Four questions share the vocabulary – status, topic, risk, and what a site is – so shopping and automobile compete for one slot and a car dealership is honestly both. That is why categories exists: category is the argmax, categories is everything above multilabel_threshold (default 0.10).

On held-out data that raises the chance the correct label is reported from 79.7% to 86.6%, at 1.35 labels per domain – 65% still get exactly one. That is a recall figure. The evaluation gold is single-label, so it cannot say whether a second label is correct, only whether the right one is present. Judge extra labels on their probabilities, not on that number.

See training/taxonomy.py and docs/taxonomy.md.

Parameters:: cache_dir (str | None)

classify(domains, archive_date=None, use_cache=True, latest=False, use_screenshots=False)[source]¶

Classify domains from their page text.

Screenshots are opt-in, and the reason is measured. Combining the two models by calibrated late fusion was fitted on 1,704 held-out paired domains and scored on 1,742 more:

model	accuracy	macro-F1
text only	0.794	0.699
image only	0.429	0.306
fused (per-class)	0.798	0.700

+0.001 macro-F1 is noise at that sample size, and the fitted text weight is 0.973 — the optimiser puts almost nothing on the screenshot. So the default does not pay for loading a 350MB vision model and running it per domain.

Pass use_screenshots=True to fuse anyway. It is honest about its own value: without published fusion weights, or with the screenshot model unavailable, it returns the text answer rather than falling back to averaging.

Parameters:

domains (list[str]) – List of domain names or URLs to classify e.g., [“google.com”, “https://facebook.com/page”]
archive_date (str | datetime | None) – For historical analysis. Format: “YYYYMMDD” or datetime object
use_cache (bool) – Whether to reuse cached content (default: True)
latest (bool) – Whether to download latest model versions (default: False)
use_screenshots (bool) – Fuse the screenshot model in (default: False). See above for what it is worth.

Returns:

{"results": [...], "report": {...}}. Each result carries url, domain, text_path, image_path, date_time_collected, model_used, category, confidence, raw_predictions, plus the status/stage/error_code/retryable outcome fields.

Return type:

dict

Example

>>> classifier = DomainClassifier()
>>> run = classifier.classify(["cnn.com", "bbc.com"])
>>> first = run["results"][0]
>>> print(f"{first['domain']}: {first['category']}")
cnn.com: news

classify_by_text(domains, archive_date=None, use_cache=True, latest=False)[source]¶

Classify domains using only text content analysis.

Faster than combined analysis, good for batch processing or when screenshots are not needed.

Parameters:

domains (list[str]) – List of domain names or URLs to classify
archive_date (str | datetime | None) – For historical analysis
use_cache (bool) – Whether to reuse cached content (default: True)
latest (bool) – Whether to download latest model versions (default: False)

Returns:

{"results": [...], "report": {...}}. Result rows contain:

url: Original URL/domain input
domain: Parsed domain name
text_path: Path to collected HTML file
image_path: Path to collected screenshot (may be None)
date_time_collected: When data was collected (ISO format)
model_used: “text/shallalist_ml”
category: Text classification prediction
confidence: Text confidence score (0-1)
reason: None (reasoning field for LLM models)
error: Error message if classification failed
raw_predictions: Full text probability distribution

Return type:

dict

Example

>>> classifier = DomainClassifier()
>>> run = classifier.classify_by_text(["wikipedia.org"])
>>> result = run["results"][0]
>>> print(f"{result['domain']}: {result['category']} ({result['confidence']:.3f})")
wikipedia.org: education (0.823)

classify_by_images(domains, archive_date=None, use_cache=True, latest=False)[source]¶

Classify domains using only homepage screenshot analysis.

Good for visual content classification, especially when text content is minimal or misleading.

Parameters:

domains (list[str]) – List of domain names or URLs to classify
archive_date (str | datetime | None) – For historical analysis
use_cache (bool) – Whether to reuse cached content (default: True)
latest (bool) – Whether to download latest model versions (default: False)

Returns:

{"results": [...], "report": {...}}. Result rows contain:

url: Original URL/domain input
domain: Parsed domain name
text_path: Path to collected HTML file (may be None)
image_path: Path to collected screenshot
date_time_collected: When data was collected (ISO format)
model_used: “image/shallalist_ml”
category: Image classification prediction
confidence: Image confidence score (0-1)
reason: None (reasoning field for LLM models)
error: Error message if classification failed
raw_predictions: Full image probability distribution

Return type:

dict

Example

>>> classifier = DomainClassifier()
>>> run = classifier.classify_by_images(["instagram.com"])
>>> result = run["results"][0]
>>> print(f"{result['domain']}: {result['category']} ({result['confidence']:.3f})")
instagram.com: socialnet (0.912)

configure_llm(provider, model, api_key=None, categories=None, **kwargs)[source]¶

Configure LLM for AI-powered domain classification.

Parameters:

provider (str) – LLM provider (‘openai’, ‘anthropic’, ‘google’, etc.)
model (str) – Model name (‘gpt-4o’, ‘claude-3-5-sonnet-20241022’, ‘gemini-1.5-pro’)
api_key (str | None) – API key for the provider (or set via environment variable)
categories (list[str] | None) – Custom classification categories
**kwargs – Additional LLMConfig parameters (temperature, max_tokens, etc.)

Return type:

None

Example

>>> classifier = DomainClassifier()
>>> classifier.configure_llm(
...     provider="openai",
...     model="gpt-4o",
...     api_key="sk-...",
...     categories=["news", "shopping", "social", "tech"]
... )

classify_by_llm(domains, custom_instructions=None, use_cache=True, mode='text')[source]¶

Classify domains using LLM analysis.

Parameters:

domains (list[str]) – List of domain names to classify
custom_instructions (str | None) – Optional custom classification instructions
use_cache (bool) – Whether to use cached content (default: True)
mode (str) – LLM mode - “text”, “image”, or “multimodal” (default: “text”)

Returns:

LLM classification results in JSON format with fields:

url: Original URL/domain input
domain: Parsed domain name
text_path: Path to collected HTML file
image_path: Path to collected screenshot (if applicable)
date_time_collected: When data was collected (ISO format)
model_used: “text/llm_{provider}_{model}” or similar
category: LLM classification prediction
confidence: LLM confidence score (0-1)
reason: LLM reasoning explanation
error: Error message if classification failed

Return type:

list[dict]

Example

>>> classifier = DomainClassifier()
>>> classifier.configure_llm("openai", "gpt-4o", api_key="sk-...")
>>> results = classifier.classify_by_llm(["cnn.com", "amazon.com"])
>>> print(f"{results[0]['domain']}: {results[0]['category']} - {results[0]['reason']}")
cnn.com: news - This domain contains current events and journalism content

Raises:

RuntimeError – If the operation cannot be completed in the current state.
ValueError – If an argument is invalid.

Parameters:

domains (list[str])
custom_instructions (str | None)
use_cache (bool)
mode (str)

Return type:

list[dict]

classify_by_llm_multimodal(domains, custom_instructions=None, use_cache=True)[source]¶

Classify domains using LLM multimodal analysis (text + screenshots).

Parameters:

domains (list[str]) – List of domain names to classify
custom_instructions (str | None) – Optional custom classification instructions
use_cache (bool) – Whether to use cached content (default: True)

Returns:

Multimodal LLM classification results in JSON format

Return type:

list[dict]

Example

>>> classifier = DomainClassifier()
>>> classifier.configure_llm("openai", "gpt-4o", api_key="sk-...")
>>> results = classifier.classify_by_llm_multimodal(["cnn.com"])
>>> print(f"{results[0]['domain']}: {results[0]['category']} - {results[0]['reason']}")
cnn.com: news - Based on text content and visual layout typical of news websites

get_llm_usage_stats()[source]¶

Get LLM usage statistics and cost tracking.

Returns:: Dictionary with usage stats or None if LLM not configured
Return type:: dict | None

Example

>>> classifier = DomainClassifier()
>>> classifier.configure_llm("openai", "gpt-4o")
>>> classifier.classify_by_llm(["example.com"])
>>> stats = classifier.get_llm_usage_stats()
>>> print(f"Cost: ${stats['estimated_cost_usd']:.4f}")

collect_content(domains, archive_date=None, collection_id=None, use_cache=True, batch_size=10)[source]¶

Collect website content for domains without performing inference.

Separates content collection from classification, enabling: - Content reuse across multiple models - Clear data lineage and inspection - Reproducible analysis workflows

Parameters:

domains (list[str]) – List of domain names or URLs to collect content for
archive_date (str | datetime | None) – For historical analysis
collection_id (str | None) – Identifier for this collection
use_cache (bool) – Whether to use cached content when available
batch_size (int) – Number of domains to process in parallel

Returns:

Collection metadata with file paths for downstream inference

Return type:

dict

Example

>>> classifier = DomainClassifier()
>>> collection = classifier.collect_content(["cnn.com", "bbc.com"])
>>> print(collection["domains"][0]["text_path"])
html/cnn.com.html

classify_from_collection(collection_data, method='combined', output_file=None, latest=False)[source]¶

Perform inference on previously collected content.

Parameters:

collection_data (dict) – Collection metadata from collect_content()
method (str) – Classification method - “text”, “images”, “combined”, or “llm”
output_file (str | None) – Path to save JSON results
latest (bool) – Whether to use latest model versions (default: False)

Returns:

Classification results in JSON format

Return type:

list[dict]

Example

>>> classifier = DomainClassifier()
>>> collection = classifier.collect_content(["cnn.com"])
>>> results = classifier.classify_from_collection(collection, method="text")
>>> print(results[0]["category"])
news

Raises:

RuntimeError – If the operation cannot be completed in the current state.
ValueError – If an argument is invalid.

Parameters:

collection_data (dict)
method (str)
output_file (str | None)
latest (bool)

Return type:

list[dict]

piedomains.api.classify_domains(domains, method='combined', archive_date=None, cache_dir=None)[source]¶

Quick domain classification function.

Parameters:

domains (list[str]) – List of domain names or URLs to classify
method (str) – Classification method - “combined”, “text”, or “images”
archive_date (str | datetime | None) – Optional historical date for archive.org analysis
cache_dir (str | None) – Optional cache directory override

Returns:

{"results": [...], "report": {...}}. Each result row carries status, stage and error_code; the report aggregates counts by reason and names the domains that produced nothing.

Return type:

dict

Example

>>> run = classify_domains(["cnn.com", "github.com"])
>>> for result in run["results"]:
...     print(f"{result['domain']}: {result['category']} ({result['confidence']:.3f})")
cnn.com: news (0.876)
github.com: computers (0.892)
>>> run["report"]["failed"]
0

piedomains.cli module¶

Command-line interface for piedomains.

piedomains.cli.build_parser()[source]¶

Construct the argument parser.

Returns:: The configured parser.
Return type:: argparse.ArgumentParser

piedomains.cli.training_scripts_dir()[source]¶

Locate the training scripts that ship with the package.

Every accuracy figure in the README is produced by these, and a number nobody can re-run is a number taken on faith. They are a subpackage, so they install with the library and run as python -m piedomains.training.<name> — but a path inside site-packages is not something anyone would guess, hence this.

Returns:: The directory holding train_text.py, evaluate.py and the rest.
Return type:: Path
Raises:: SystemExit – If the scripts are not found, rather than returning a path that does not exist and failing later somewhere less obvious.

piedomains.cli.main(argv=None)[source]¶

Run the command-line interface.

Parameters:: argv (list[str] | None) – Argument list to parse. Defaults to sys.argv[1:].
Returns:: Process exit code.
Return type:: int

piedomains.config module¶

Configuration management for piedomains.

class piedomains.config.Config(config_dict=None)[source]¶

Bases: object

Configuration class for piedomains settings.

Parameters:: config_dict (dict[str, Any] | None)

DEFAULT_CONFIG: ClassVar[dict[str, Any]] = {'address_cache_ttl': 300.0, 'allow_hosts': [], 'allowed_content_types': ['text/html', 'application/xhtml+xml', 'application/xml', 'text/xml', 'text/plain'], 'archive_backoff': 2, 'archive_fallback': True, 'archive_max_age_days': 365, 'archive_max_parallel': 2, 'archive_memento_rate': 4, 'archive_render_settle_ms': 1500, 'archive_retries': 3, 'archive_screenshot_timeout': 15000, 'archive_search_rate': 1, 'archive_window_days': 365, 'batch_size': 50, 'block_media': True, 'block_resources': ['media', 'video', 'font', 'websocket', 'manifest'], 'blocked_extensions': ['.exe', '.msi', '.scr', '.bat', '.cmd', '.com', '.pif', '.vbs', '.jar', '.app', '.dmg', '.pkg', '.deb', '.rpm', '.run', '.bin', '.elf', '.so', '.dll', '.dylib'], 'check_addresses': True, 'content_length_limits': {'application/pdf': 52428800, 'default': 10485760, 'text/html': 5242880}, 'content_safety_mode': 'moderate', 'crawl_delay': 1.0, 'dns_timeout': 5.0, 'enable_content_validation': True, 'extractor': 'trafilatura', 'filter_non_english': False, 'html_extension': '.html', 'http_timeout': 10, 'image_extension': '.png', 'image_size': (254, 254), 'log_format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s', 'log_level': 'INFO', 'max_concurrent_fetches': 8, 'max_content_length': 10485760, 'max_crawl_delay': 30.0, 'max_parallel': 4, 'max_retries': 3, 'min_tokens': 30, 'model_cache_dir': None, 'multilabel_threshold': 0.1, 'network_quiet_ms': 3000, 'obey_robots': True, 'parallel_workers': 4, 'playwright_headless': True, 'playwright_timeout': 30000, 'playwright_viewport': {'height': 1024, 'width': 1280}, 'proxy_server': '', 'retry_delay': 1, 'sandbox_mode_required': False, 'screenshot_scale': 1, 'settle_ms': 1500, 'strip_punctuation': False, 'suspicious_url_patterns': ['.*\\/[^\\/]*\\.(exe|msi|scr|bat|cmd|pif|vbs|jar)(\\?.*)?$', '.*\\.com\\/.*\\.(exe|msi|scr|bat|cmd|pif|vbs|jar)(\\?.*)?$', '.*\\/download\\/.*\\.(zip|rar|7z|tar\\.gz|tgz)(\\?.*)?$', '.*\\/attachment\\/.*', '.*[?&](download|attachment)=.*'], 'text_cleaning': 'minimal', 'user_agent': 'piedomains/0.14.0 (+https://github.com/themains/piedomains)', 'validate_domain_extensions': False, 'webdriver_timeout': 30, 'webdriver_window_size': '1280,1024'}¶

get(key, default=None)[source]¶

Get configuration value.

Parameters:

key (str) – Configuration key
default (Any) – Default value if key not found

Returns:

Configuration value, or default when the key is absent.

Return type:

Any

set(key, value)[source]¶

Set configuration value.

Parameters:

key (str) – Configuration key
value (Any) – Configuration value

update(config_dict)[source]¶

Update multiple configuration values.

Parameters:: config_dict (dict[str, Any]) – Configuration updates

to_dict()[source]¶

Get configuration as dictionary.

Returns:: Configuration dictionary
Return type:: dict[str, Any]

property http_timeout: int¶: HTTP request timeout in seconds.

property webdriver_timeout: int¶: WebDriver timeout in seconds.

property page_load_timeout: int¶: Page load timeout in seconds.

property max_retries: int¶: Maximum number of retries for failed operations.

property retry_delay: float¶: Delay between retries in seconds.

property screenshot_wait_time: int¶: Wait time after loading page before screenshot.

property webdriver_window_size: str¶: WebDriver window size.

property batch_size: int¶: Batch size for processing domains.

property parallel_workers: int¶: Number of parallel workers.

property user_agent: str¶: User agent string for HTTP requests.

property image_size: tuple¶: Image size for model input.

property enable_content_validation: bool¶: Whether content validation is enabled.

property content_safety_mode: str¶

strict, moderate, or permissive.

Type:: Content safety mode

property max_content_length: int¶: Maximum content length to download.

property sandbox_mode_required: bool¶: Whether sandbox mode is required for risky content.

property allowed_content_types: list¶: List of allowed MIME types.

property blocked_extensions: list¶: List of blocked file extensions.

property suspicious_url_patterns: list¶: List of regex patterns for suspicious URLs.

property content_length_limits: dict¶: Content length limits by content type.

piedomains.config.get_config()[source]¶

Get global configuration instance.

Returns:: Global configuration instance
Return type:: Config

piedomains.config.set_config(config)[source]¶

Set global configuration instance.

Parameters:: config (Config) – Configuration instance to set as global

piedomains.config.configure(**kwargs)[source]¶

Configure global settings.

Parameters:: **kwargs (Any) – Configuration key-value pairs

piedomains.constants module¶

Constants and classification categories for piedomains package.

This module defines the core classification categories used by piedomains for domain content classification, as well as filtering constants for text processing.

The categories are based on the Shallalist categorization system, a comprehensive classification scheme originally developed for web filtering and content analysis. These categories cover the major types of web content found across the internet.

Example

Accessing classification categories:

>>> from piedomains.constants import classes, most_common_words
>>> print(f"Available categories: {len(classes)}")
Available categories: 44
>>> print(f"Example categories: {classes[:3]}")
Example categories: ['adult', 'alcohol', 'automobile']
>>> print(f"Common words to filter: {most_common_words[:3]}")
Common words to filter: ['home', 'contact', 'us']

piedomains.constants.classes: list[str] = ['adult', 'alcohol', 'automobile', 'cooking', 'dating', 'downloads', 'drugs', 'education', 'finance', 'fortunetelling', 'forum', 'gamble', 'games', 'gardening', 'government', 'homestyle', 'hospitals', 'humor', 'imagehosting', 'isp', 'jobsearch', 'library', 'military', 'movies', 'music', 'news', 'pets', 'politics', 'radiotv', 'realestate', 'religion', 'restaurants', 'science', 'searchengines', 'shopping', 'socialnet', 'sports', 'travel', 'urlshortener', 'weapons', 'webmail', 'wellness', 'parked', 'unavailable']¶

Complete list of website classification categories.

This list contains 44 categories used for domain content classification. Derived from Shallalist but not identical to it: classes describing how a site is hosted or monetised are removed, so are those asking about delivery mechanism or legality, recreation and hobby are split, and the adult categories are merged. parked and unavailable are added, because a domain with no site behind it is a fact the caller can act on and the page states it plainly. See training/taxonomy.py.

The categories are used by both traditional ML models and LLM-based classification to provide consistent categorization across different classification methods.

Type:: List[str]

piedomains.constants.most_common_words: list[str] = ['home', 'contact', 'us', 'new', 'news', 'site', 'privacy', 'search', 'help', 'copyright', 'free', 'service', 'en', 'get', 'one', 'find', 'menu', 'account', 'next']¶

Common words to filter out during text preprocessing.

These words are extremely common across all website types and provide little discriminative value for classification. They are filtered out during text processing to focus on more meaningful content words.

This list includes: - Navigation elements (home, menu, next) - Generic marketing terms (free, new, get) - Common website sections (contact, help, privacy) - Linguistic articles and connectors (us, one, en)

Used by text preprocessing functions to clean content before model input.

Type:: List[str]

piedomains.constants.get_valid_categories()[source]¶

Get a copy of all valid classification categories.

Returns:: Complete list of valid category names for classification.
Return type:: List[str]

Example

>>> categories = get_valid_categories()
>>> if "news" in categories:
...     print("News category is available")
News category is available

piedomains.constants.is_valid_category(category)[source]¶

Check if a category name is valid for classification.

Parameters:: category (str) – Category name to validate.
Returns:: True if category is valid, False otherwise.
Return type:: bool

Example

>>> is_valid_category("news")
True
>>> is_valid_category("invalid_category")
False

piedomains.constants.get_category_count()[source]¶

Get the total number of available classification categories.

Returns:: Total number of classification categories.
Return type:: int

Example

>>> count = get_category_count()
>>> print(f"Total categories available: {count}")
Total categories available: 44

piedomains.content_processor module¶

Content processor for coordinating HTML and image extraction using Playwright.

Handles file I/O, caching, and coordination between different content types.

class piedomains.content_processor.ContentProcessor(cache_dir=None, archive_date=None)[source]¶

Bases: object

Coordinates content extraction for domains using Playwright fetcher.

Parameters:

cache_dir (str | None)
archive_date (str | None)

extract_all_content(domains, use_cache=True, parallel=True)[source]¶

Extract all content (HTML, text, screenshots) from domains.

Uses unified Playwright fetcher for everything.

Parameters:

domains (list[str]) – List of domain names or URLs
use_cache (bool) – Whether to use cached content
parallel (bool) – Whether to fetch in parallel

Returns:

Results keyed by domain name

Return type:

Dict[str, dict]

extract_html_content(domains, use_cache=True, *, force_fetch=False, allow_content_types=None, ignore_extensions=False)[source]¶

Extract HTML content for domains.

Maintains backwards compatibility with existing API.

Parameters:

domains (list[str]) – List of domain names or URLs
use_cache (bool) – Whether to use cached HTML files
force_fetch (bool) – Skip security validation (dangerous)
allow_content_types (list[str] | None) – Override allowed content types
ignore_extensions (bool) – Skip file extension validation

Returns:

(html_content_dict, errors_dict)

Return type:

Tuple[Dict[str, str], Dict[str, str]]

extract_text_content(domains, use_cache=True, *, force_fetch=False, allow_content_types=None, ignore_extensions=False)[source]¶

Extract and process text content from domains.

Maintains backwards compatibility with existing API.

Parameters:

domains (list[str]) – List of domain names or URLs
use_cache (bool) – Whether to use cached content
force_fetch (bool) – Skip security validation (dangerous)
allow_content_types (list[str] | None) – Override allowed content types
ignore_extensions (bool) – Skip file extension validation

Returns:

(processed_text_dict, errors_dict)

Return type:

Tuple[Dict[str, str], Dict[str, str]]

extract_image_content(domains, use_cache=True, *, force_fetch=False, ignore_extensions=False)[source]¶

Extract screenshot images for domains.

Maintains backwards compatibility with existing API.

Parameters:

domains (list[str]) – List of domain names or URLs
use_cache (bool) – Whether to use cached images
force_fetch (bool) – Skip security validation (dangerous)
ignore_extensions (bool) – Skip file extension validation

Returns:

(image_paths_dict, errors_dict)

Return type:

Tuple[Dict[str, str], Dict[str, str]]

prepare_image_tensors(image_paths)[source]¶

Convert images to numpy arrays for model input.

Parameters:: image_paths (dict[str, str]) – Domain name to image path mapping
Returns:: Domain name to image tensor mapping
Return type:: Dict[str, np.ndarray]

piedomains.content_validation module¶

Content validation utilities for security and safety checks.

This module provides comprehensive validation for URLs and content types to prevent security risks when processing unknown domains.

class piedomains.content_validation.ContentValidationResult(is_safe, content_type, content_length, error_message, warnings, sandbox_recommended)[source]¶

Bases: NamedTuple

Result of content validation check.

Parameters:

is_safe (bool)
content_type (str | None)
content_length (int | None)
error_message (str)
warnings (list[str])
sandbox_recommended (bool)

is_safe: bool¶: Alias for field number 0

content_type: str | None¶: Alias for field number 1

content_length: int | None¶: Alias for field number 2

error_message: str¶: Alias for field number 3

warnings: list[str]¶: Alias for field number 4

sandbox_recommended: bool¶: Alias for field number 5

class piedomains.content_validation.ContentValidator(config=None)[source]¶

Bases: object

Validates content safety and provides security recommendations.

validate_url(url, *, force_fetch=False, allow_content_types=None, ignore_extensions=False)[source]¶

Comprehensive URL and content validation.

Parameters:

url (str) – URL to validate
force_fetch (bool) – Skip validation and force content retrieval
allow_content_types (list[str] | None) – Override allowed content types
ignore_extensions (bool) – Skip file extension validation

Returns:

ContentValidationResult with validation details

Return type:

ContentValidationResult

validate_url_offline(url, *, ignore_extensions=False)[source]¶

Validate a URL without contacting the host.

Step 1 of validate_url() on its own. It exists so a caller can decide a URL is unfetchable – or ask robots.txt about it – before the preflight in step 2 issues a request. Sending that request first and then honouring robots means having already done the thing robots forbade.

Parameters:

url (str) – URL to validate.
ignore_extensions (bool) – Skip file-extension validation.

Returns:

Safe when nothing about the URL itself disqualifies it. Says nothing about what the host will serve.

Return type:

ContentValidationResult

get_sandbox_command(url, text_only=True)[source]¶

Generate sandbox execution command for a URL.

Parameters:

url (str)
text_only (bool)

Return type:

str

piedomains.context_managers module¶

Context managers for resource cleanup and management.

piedomains.context_managers.webdriver_context()[source]¶

DEPRECATED: Use PlaywrightFetcher context manager instead.

This function is maintained for backward compatibility.

piedomains.context_managers.playwright_context()[source]¶

Context manager for PlaywrightFetcher instances.

Yields:: PlaywrightFetcher – Playwright fetcher instance
Return type:: Generator[PlaywrightFetcher, None, None]

Ensures proper cleanup of Playwright resources.

Raises:: Exception – Propagated from the wrapped operation after cleanup.
Return type:: Generator[PlaywrightFetcher, None, None]

piedomains.context_managers.temporary_directory(suffix='', prefix='piedomains_')[source]¶

Context manager for temporary directories.

Parameters:

suffix (str) – Directory name suffix
prefix (str) – Directory name prefix

Yields:

str – Path to temporary directory

Return type:

Generator[str, None, None]

Ensures cleanup of temporary directories.

Raises:

Exception – Propagated from the wrapped operation after cleanup.

Parameters:

suffix (str)
prefix (str)

Return type:

Generator[str, None, None]

piedomains.context_managers.file_cleanup(*file_paths)[source]¶

Context manager for file cleanup.

Parameters:: *file_paths (str) – Paths to files that should be cleaned up
Return type:: Generator[None, None, None]

Ensures cleanup of specified files after context exits.

piedomains.context_managers.error_recovery(operation_name, fallback_value=None, reraise=False)[source]¶

Context manager for error recovery with logging.

Parameters:

operation_name (str) – Name of the operation for logging
fallback_value (Any) – Value to return on error (if not reraising)
reraise (bool) – Whether to reraise exceptions

Yields:

dict[str, Any] – Dictionary with ‘success’, ‘error’, ‘result’ keys

Raises:

Exception – Propagated from the wrapped operation after cleanup.

Return type:

Generator[dict[str, Any], None, None]

piedomains.context_managers.batch_progress_tracking(total_items, operation_name='Processing')[source]¶

Context manager for tracking batch processing progress.

Parameters:

total_items (int) – Total number of items to process
operation_name (str) – Name of the operation

Yields:

Callable[[int], None] – Function to update progress

Raises:

Exception – Propagated from the wrapped operation after cleanup.

Return type:

Generator[Callable[[int], None], None, None]

class piedomains.context_managers.ResourceManager[source]¶

Bases: object

Resource manager for tracking and cleaning up resources.

add_driver(driver)[source]¶: Add a WebDriver/fetcher instance for cleanup (deprecated).

add_fetcher(fetcher)[source]¶: Add a PlaywrightFetcher instance for cleanup.

add_temp_directory(path)[source]¶

Add a temporary directory for cleanup.

Parameters:: path (str)

add_temp_file(path)[source]¶

Add a temporary file for cleanup.

Parameters:: path (str)

cleanup_all()[source]¶: Clean up all tracked resources.

piedomains.data_collector module¶

Data collector for piedomains - separates data collection from inference.

This module provides clean separation between data fetching and classification, enabling reusability, transparency, and reproducibility.

class piedomains.data_collector.DataCollector(cache_dir='data', archive_date=None, max_parallel=None)[source]¶

Bases: object

Pure data collection for domain content analysis.

Separates data fetching from inference, enabling: - Data reuse across multiple models - Clear data lineage and inspection - Reproducible analysis workflows

Example

>>> collector = DataCollector(cache_dir="data")
>>> data = collector.collect(["cnn.com", "bbc.com"])
>>> # data is now JSON with file paths for downstream inference

Parameters:

cache_dir (str)
archive_date (str | None)
max_parallel (int | None)

collect(domains, collection_id=None, use_cache=True, save_metadata=True)[source]¶

Collect data for domains and return structured metadata.

Parameters:

domains (list[str]) – List of domain names or URLs to collect data for
collection_id (str | None) – Optional identifier for this collection (auto-generated if None)
use_cache (bool) – Whether to use cached data when available
save_metadata (bool) – Whether to save collection metadata to file

Returns:

Dictionary with collection metadata and file paths

Return type:

dict

Example

>>> collector = DataCollector()
>>> data = collector.collect(["cnn.com", "bbc.com"])
>>> print(data["domains"][0]["text_path"])
html/cnn.com.html

Raises:

ValueError – If an argument is invalid.

Parameters:

domains (list[str])
collection_id (str | None)
use_cache (bool)
save_metadata (bool)

Return type:

dict

collect_batch(domains, collection_id=None, use_cache=True, save_metadata=True, batch_size=10)[source]¶

Collect data for large batches of domains with optimized parallel processing.

Parameters:

domains (list[str]) – List of domain names or URLs
collection_id (str | None) – Optional identifier for this collection
use_cache (bool) – Whether to use cached data when available
save_metadata (bool) – Whether to save collection metadata to file
batch_size (int) – Number of domains to process in parallel

Returns:

Dictionary with collection metadata and file paths

Raises:

ValueError – If an argument is invalid.

Return type:

dict

load_collection(collection_id)[source]¶

Load previously saved collection metadata.

Parameters:: collection_id (str) – ID of the collection to load
Returns:: Collection metadata dictionary
Raises:: FileNotFoundError – If collection file doesn’t exist
Return type:: dict

list_collections()[source]¶

List all available collections.

Returns:: List of collection summaries
Return type:: list[dict]

piedomains.fetchers module¶

Playwright-based page fetcher for content extraction.

Supports live content fetching and archive.org historical snapshots. Unified pipeline for HTML, text extraction, and screenshots.

class piedomains.fetchers.FetchResult(url, success, html='', text='', screenshot_path='', title='', meta_description='', error='', error_code='', snapshot_timestamp='', source='live')[source]¶

Bases: object

Result from a single fetch operation.

Parameters:

url (str)
success (bool)
html (str)
text (str)
screenshot_path (str)
title (str)
meta_description (str)
error (str)
error_code (str)
snapshot_timestamp (str)
source (str)

url: str¶

success: bool¶

html: str = ''¶

text: str = ''¶

screenshot_path: str = ''¶

title: str = ''¶

meta_description: str = ''¶

error: str = ''¶

error_code: str = ''¶: Stable reason this fetch failed, from piedomains.outcomes.ErrorCode.

snapshot_timestamp: str = ''¶: For archive fetches, the capture actually used (YYYYMMDDHHMMSS). The requested date is not echoed here — this is what was really retrieved.

source: str = 'live'¶

live or archive. A live fetch that hit a bot wall and was recovered from archive.org reports archive, so a caller can tell which rows are not the site as it stands today.

Type:: Where the content came from

class piedomains.fetchers.BaseFetcher[source]¶

Bases: object

Base class for content fetchers with security validation.

async fetch_single(url, screenshot_path=None)[source]¶

Fetch HTML, text and (optionally) a screenshot for one URL.

Parameters:

url (str) – URL to fetch.
screenshot_path (str | None) – Where to write the screenshot, if wanted.

Returns:

The fetch outcome.

Return type:

FetchResult

Raises:

NotImplementedError – Always; subclasses provide the implementation.

async fetch_batch(urls, cache_dir='cache')[source]¶

Fetch several URLs concurrently.

Parameters:

urls (list[str]) – URLs to fetch.
cache_dir (str) – Directory screenshots are written under.

Returns:

One result per URL, in input order.

Return type:

list[FetchResult]

Raises:

NotImplementedError – Always; subclasses provide the implementation.

cleanup()[source]¶

Release any resources held by the fetcher.

Browsers are opened and closed inside each async with async_playwright() block, so there is nothing left to release; this exists so callers can invoke it unconditionally.

Return type:: None

fetch_both(url, output_path, **kwargs)[source]¶

Synchronously fetch HTML, text and a screenshot for one URL.

Defined on the base class so that both live and archive fetchers expose it — DataCollector calls this on whatever get_fetcher returned.

Parameters:

url (str) – URL or bare domain to fetch.
output_path (str) – Where to write the screenshot.
**kwargs – Accepted for call-site compatibility; unused.

Returns:

The fetch outcome.

Return type:

FetchResult

class piedomains.fetchers.PlaywrightFetcher(max_parallel=4)[source]¶

Bases: BaseFetcher

Unified Playwright fetcher for all content extraction.

Parameters:: max_parallel (int)

async fetch_single(url, screenshot_path=None)[source]¶

Fetch content from a single URL.

Parameters:

url (str)
screenshot_path (str | None)

Return type:

FetchResult

async fetch_batch(urls, cache_dir='cache')[source]¶

Fetch multiple URLs in parallel.

Parameters:

urls (list[str])
cache_dir (str)

Return type:

list[FetchResult]

fetch_html(url, **kwargs)[source]¶

Sync wrapper for HTML fetching.

Parameters:: url (str)
Return type:: tuple[bool, str, str]

fetch_content(url, **kwargs)[source]¶

Sync wrapper for content fetching (alias for fetch_single).

Parameters:: url (str)
Return type:: FetchResult

fetch_screenshot(url, output_path, **kwargs)[source]¶

Sync wrapper for screenshot.

Parameters:

url (str)
output_path (str)

Return type:

tuple[bool, str]

class piedomains.fetchers.ArchiveFetcher(target_date, max_parallel=None, max_age_days=None)[source]¶

Bases: BaseFetcher

Fetch historical snapshots from archive.org via the CDX + Memento APIs.

Snapshot discovery, closest-date matching, rate limiting and backoff are delegated to the wayback library rather than hand-rolled. Two playback modes are used deliberately:

text/HTML uses Mode.original (the id_ suffix) — the raw capture, with no injected Wayback JavaScript and no rewritten URLs, so no browser and no toolbar-stripping are needed.
screenshots use the if_ suffix, which suppresses the toolbar but keeps asset references pointing at archived copies, so the page still renders as it did. id_ would screenshot a page stripped of its CSS and images.

Only captures with HTTP status 200 are considered; a domain whose only captures are redirects or error pages fails loudly rather than having an archived 404 classified as content.

Parameters:

target_date (str | datetime)
max_parallel (int | None)
max_age_days (int | None)

iframe_url(record)[source]¶

Build the if_ playback URL used for screenshots.

Parameters:: record (CdxRecord) – The CDX record to render.
Returns:: A Wayback URL that renders without the toolbar but with archived assets intact.
Return type:: str

find_closest_record(url)[source]¶

Find the status-200 capture nearest the target date.

Parameters:: url (str) – URL or bare domain to look up.
Returns:: The nearest CdxRecord, or None if the domain has no usable capture in the search window.
Return type:: CdxRecord | None

async fetch_single(url, screenshot_path=None)[source]¶

Fetch one archived page’s HTML, text and optional screenshot.

Parameters:

url (str) – URL or bare domain to fetch.
screenshot_path (str | None) – Where to write the screenshot, if wanted.

Returns:

The fetch outcome, carrying the realized snapshot_timestamp.

Return type:

FetchResult

async fetch_batch(urls, cache_dir='cache')[source]¶

Fetch several archived pages, bounded by max_parallel.

Parameters:

urls (list[str]) – URLs or bare domains to fetch.
cache_dir (str) – Directory screenshots are written under.

Returns:

One result per URL, in input order.

Return type:

list[FetchResult]

fetch_html(url, **kwargs)[source]¶

Synchronously fetch archived HTML for one URL.

Parameters:

url (str) – URL or bare domain to fetch.
**kwargs – Accepted for call-site compatibility; unused.

Returns:

(success, html, error).

Return type:

tuple

piedomains.fetchers.get_fetcher(archive_date=None, max_parallel=4)[source]¶

Factory function to get appropriate fetcher.

Parameters:

archive_date (str | datetime | None) – If provided, returns ArchiveFetcher for this date. If None, returns PlaywrightFetcher for current content.
max_parallel (int) – Maximum number of parallel browser contexts

Returns:

Appropriate fetcher instance

Return type:

BaseFetcher

piedomains.http_client module¶

HTTP client with connection pooling and session management for improved performance.

class piedomains.http_client.PooledHTTPClient[source]¶

Bases: object

HTTP client with connection pooling and session reuse.

property session: Session¶: Get or create HTTP session with connection pooling.

get(url, timeout=None, **kwargs)[source]¶

Perform HTTP GET with retry logic and connection pooling.

Parameters:

url (str) – URL to fetch
timeout (float | None) – Request timeout (uses config default if None)
**kwargs – Additional arguments passed to requests.get

Returns:

HTTP response

Return type:

requests.Response

Raises:

OSError – If the socket-level operation fails after all retries.
requests.exceptions.RequestException – If the HTTP request fails after all retries.

close()[source]¶: Close the HTTP session.

piedomains.http_client.http_client()[source]¶

Context manager for getting a pooled HTTP client.

Yields:: PooledHTTPClient – HTTP client with connection pooling
Raises:: Exception – Propagated from the wrapped operation after cleanup.
Return type:: Generator[PooledHTTPClient, None, None]

piedomains.http_client.get_http_client()[source]¶

Get the global HTTP client instance.

Returns:: Global HTTP client with connection pooling
Return type:: PooledHTTPClient

piedomains.http_client.close_global_client()[source]¶: Close the global HTTP client.

piedomains.image module¶

Screenshot-based classification.

Replaces a ResNet50 that reported 52.9% at training time with a frozen backbone — only a linear head was ever fitted — and, in production, labelled Khan Academy and Yahoo as porn. Two independent bugs produced that: the backbone was never fine-tuned, and the serving path divided pixels by 255 before handing them to a graph that already baked in resnet50.preprocess_input, so every image arrived as a near-constant negative array.

Both are gone. The backbone is fully fine-tuned (training/train_image.py) and preprocessing is the model’s own AutoImageProcessor, so training and inference cannot drift apart.

A screenshot is a weak signal on its own, and that is the point. At 224px a page is unreadable — the model sees layout, colour and gross structure, not text. It exists to say something the text model cannot, which matters most on pages carrying almost no text. It is combined with text through calibrated late fusion, never used to overrule it blindly.

piedomains.image.DEFAULT_IMAGE_MODEL = 'gojiberries/piedomains-image'¶: Where the fine-tuned screenshot model lives. Overridable with PIEDOMAINS_IMAGE_MODEL (a Hub repo id or a local directory).

piedomains.image.resolve_image_model(latest=False)[source]¶

Decide which checkpoint to load.

A local directory is preferred when configured, so a freshly trained model can be evaluated before it is published.

Parameters:: latest (bool) – Re-resolve from the Hub even if a copy is cached.
Returns:: A local path or a Hugging Face Hub repo id.
Return type:: str

class piedomains.image.ImageClassifier(cache_dir=None, archive_date=None)[source]¶

Bases: object

Classify a website from a screenshot of its homepage.

Parameters:

cache_dir (str | None)
archive_date (str | None)

load_models(latest=False)[source]¶

Load the classifier, its preprocessor, labels and temperature.

Parameters:: latest (bool) – Re-resolve the model even if one is already loaded.
Raises:: RuntimeError – If the model cannot be loaded. Deliberately fatal — the predecessor substituted a model returning zeros, so every domain came back as a confident-looking wrong answer.
Return type:: None

predict_proba(image_path)[source]¶

Score one screenshot into a calibrated probability distribution.

Parameters:: image_path (str | Path) – Path to the screenshot.
Returns:: Class probabilities summing to 1, or None when the image cannot be read.
Return type:: dict[str, float] | None

classify(domains, latest=False)[source]¶

Classify domains from their cached screenshots.

Parameters:

domains (list[str]) – Domain names to classify.
latest (bool) – Whether to re-resolve the model.

Returns:

One result row per domain.

Return type:

list[dict]

classify_from_paths(data_paths, output_file=None, latest=False)[source]¶

Classify domains from collected screenshot paths.

Parameters:

data_paths (list[dict]) – Records carrying domain and image_path.
output_file (str | None) – Optional path to write JSON results to.
latest (bool) – Whether to re-resolve the model.

Returns:

One result row per record.

Return type:

list[dict]

classify_from_data(collection_data, output_file=None, latest=False)[source]¶

Classify every successfully collected domain in a collection envelope.

Parameters:

collection_data (dict) – The envelope from DataCollector.
output_file (str | None) – Optional path to write JSON results to.
latest (bool) – Whether to re-resolve the model.

Returns:

One result row per collected domain.

Return type:

list[dict]

piedomains.llm_classifier module¶

LLM-based domain classification using modern language models.

class piedomains.llm_classifier.LLMClassifier(config)[source]¶

Bases: object

LLM-based domain classifier using multiple AI providers.

This classifier leverages modern language models through the litellm library to classify domains based on text content and/or screenshots.

Variables:

config – LLM configuration settings
usage_stats – Dictionary tracking API usage and costs

Parameters:

config (LLMConfig)

get_usage_stats()[source]¶

Get current usage statistics.

Return type:: dict[str, Any]

reset_usage_stats()[source]¶

Reset usage statistics.

Return type:: None

classify_text(domains, content_dict)[source]¶

Classify domains using text content with LLM.

Parameters:

domains (list[str]) – List of domain names
content_dict (dict[str, str]) – Dict mapping domain to text content

Returns:

One dict per domain with keys: domain, category,: confidence, reasoning

Return type:

list[dict]

classify_multimodal(domains, content_dict, screenshot_dict)[source]¶

Classify domains using text content and screenshots with LLM.

Parameters:

domains (list[str]) – List of domain names
content_dict (dict[str, str]) – Dict mapping domain to text content
screenshot_dict (dict[str, str]) – Dict mapping domain to screenshot path

Returns:

One dict per domain with keys: domain, category,: confidence, reasoning

Return type:

list[dict]

classify_from_paths(data_paths, output_file=None, mode='text', cache_dir='cache')[source]¶

Classify domains using files from collected data paths.

Parameters:

data_paths (list[dict]) – List of dicts with domain data containing text_path/image_path, domain, etc.
output_file (str | None) – Optional path to save JSON results
mode (str) – Classification mode - “text”, “image”, or “multimodal”
cache_dir (str) – Directory the text_path/image_path entries are relative to

Returns:

List of classification result dictionaries (JSON format)

Return type:

list[dict]

Example

>>> classifier = LLMClassifier(config)
>>> data = [{"domain": "cnn.com", "text_path": "html/cnn.com.html", ...}]
>>> results = classifier.classify_from_paths(data, mode="text")
>>> print(results[0]["category"])
news

classify_from_data(collection_data, output_file=None, mode='text')[source]¶

Classify domains using collection metadata from DataCollector.

Parameters:

collection_data (dict) – Collection metadata dict from DataCollector.collect()
output_file (str | None) – Optional path to save JSON results
mode (str) – Classification mode - “text”, “image”, or “multimodal”

Returns:

List of classification result dictionaries (JSON format)

Return type:

list[dict]

Example

>>> from piedomains import DataCollector
>>> collector = DataCollector()
>>> data = collector.collect(["cnn.com"])
>>> classifier = LLMClassifier(config)
>>> results = classifier.classify_from_data(data, mode="multimodal")

Raises:

ValueError – If an argument is invalid.

Parameters:

collection_data (dict)
output_file (str | None)
mode (str)

Return type:

list[dict]

piedomains.piedomain module¶

Legacy prediction engine.

Only the static URL/domain validators are used in production; the extraction and inference helpers are superseded by fetchers, text and image.

class piedomains.piedomain.Piedomain[source]¶

Bases: object

Legacy engine retained for its input validators.

weights_loaded = False¶

img_width = 254¶

img_height = 254¶

static parse_url_to_domain(url)[source]¶

Extract domain name from a URL.

Parameters:: url (str) – Full URL or domain name
Returns:: Domain name extracted from URL
Return type:: str

static validate_url_or_domain(url_or_domain)[source]¶

Validate if input is a valid URL or domain name.

Parameters:: url_or_domain (str) – URL or domain name to validate
Returns:: True if valid URL or domain, False otherwise
Return type:: bool

static validate_domain_name(domain)[source]¶

Validate if a domain name is properly formatted.

Parameters:: domain (str) – Domain name to validate
Returns:: True if domain is valid, False otherwise
Return type:: bool

classmethod validate_domains(domains)[source]¶

Validate a list of domain names and separate valid from invalid.

Parameters:: domains (list) – List of domain names to validate
Returns:: (valid_domains, invalid_domains)
Return type:: tuple

classmethod validate_urls_or_domains(urls_or_domains)[source]¶

Validate a list of URLs or domains and separate valid from invalid.

Parameters:: urls_or_domains (list) – List of URLs or domain names to validate
Returns:: (valid_inputs, invalid_inputs, url_to_domain_map)
Return type:: tuple

classmethod text_from_html(text)[source]¶

Extract clean text content from HTML.

Parameters:: text (str) – Raw HTML content
Returns:: Cleaned text with unique lowercase words
Return type:: str

classmethod data_cleanup(s)[source]¶

Clean and normalize text data for model input.

Parameters:: s (str) – Raw text string
Returns:: Cleaned text with English words only, no stopwords or common terms
Return type:: str
Raises:: AttributeError – If the input is not of the expected type.

classmethod validate_input(input, path, type)[source]¶

Validate input parameters for prediction functions.

Parameters:

input (list) – List of URLs or domain names
path (str) – Path to HTML or image files
type (str) – Input type - ‘html’ or ‘image’

Returns:

True if operating in offline mode (using local files only)

Return type:

bool

Raises:

Exception – If neither URLs/domains nor valid path provided

piedomains.piedomains_logging module¶

Comprehensive logging configuration and utilities for piedomains.

This module provides centralized logging configuration with proper formatters, handlers, and log levels for the entire piedomains package. It supports both console and file logging with configurable log levels and formats.

Example

Basic usage:

>>> from piedomains.piedomains_logging import get_logger
>>> logger = get_logger()
>>> logger.info("Processing domain classification")

With custom configuration:

>>> from piedomains.piedomains_logging import configure_logging
>>> configure_logging(level="DEBUG", console_format="detailed")
>>> logger = get_logger()
>>> logger.debug("Detailed debug information")

class piedomains.piedomains_logging.JsonFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]¶

Bases: Formatter

Render log records as one JSON object per line.

Any keyword passed via extra= — run_id, domain, stage, error_code — is promoted to a top-level key so log lines can be filtered and correlated with the run report.

format(record)[source]¶

Serialize a record to a single-line JSON object.

Parameters:: record (LogRecord) – The record to render.
Returns:: A JSON object, newline-free.
Return type:: str

piedomains.piedomains_logging.bind_context(**fields)[source]¶

Bind fields onto every subsequent log record.

Used to thread run_id through a batch so that log lines can be joined against the run report.

Parameters:: **fields (object) – Key/value pairs to attach. A value of None unbinds.
Return type:: None

piedomains.piedomains_logging.clear_context()[source]¶

Remove all bound context fields.

Return type:: None

piedomains.piedomains_logging.get_logger(name=None)[source]¶

Get a logger instance for piedomains with proper configuration.

Parameters:: name (str | None) – Logger name. If None, uses ‘piedomains’ as the base logger. For module-specific loggers, pass __name__.
Returns:: Configured logger instance with appropriate handlers and formatting.
Return type:: logging.Logger

Example

>>> # Get the main piedomains logger
>>> logger = get_logger()
>>> logger.info("Main application log")

>>> # Get a module-specific logger
>>> logger = get_logger(__name__)
>>> logger.debug("Module-specific debug info")

piedomains.piedomains_logging.configure_logging(level='INFO', console_format='default', file_path=None, file_level='DEBUG', force_reconfigure=False)[source]¶

Configure logging for the piedomains package with comprehensive options.

Parameters:

level (str | int) – Console logging level. Can be string (‘DEBUG’, ‘INFO’, etc.) or logging constant (logging.INFO, etc.). Defaults to ‘INFO’.
console_format (str) – Console log format style. Options: - ‘default’: Standard format with timestamp and level - ‘detailed’: Detailed format with module/function/line info - ‘simple’: Simple format with just level and message Defaults to ‘default’.
file_path (str | None) – Path to log file. If provided, enables file logging. Directory will be created if it doesn’t exist.
file_level (str | int) – File logging level (if file_path provided). Defaults to ‘DEBUG’ for comprehensive file logs.
force_reconfigure (bool) – If True, reconfigure even if already configured. Defaults to False.

Return type:

None

Example

>>> # Basic console logging
>>> configure_logging(level="DEBUG")

>>> # Console + file logging with detailed format
>>> configure_logging(
...     level="INFO",
...     console_format="detailed",
...     file_path="/var/log/piedomains/app.log",
...     file_level="DEBUG"
... )

Raises:

ValueError – If an argument is invalid.

Parameters:

level (str | int)
console_format (str)
file_path (str | None)
file_level (str | int)
force_reconfigure (bool)

Return type:

None

piedomains.piedomains_logging.set_level(level)[source]¶

Change the logging level for all existing piedomains loggers.

Parameters:: level (str | int) – New logging level. Can be string (‘DEBUG’, ‘INFO’, etc.) or logging constant (logging.INFO, etc.).
Return type:: None

Example

>>> set_level("DEBUG")  # Enable debug logging
>>> set_level(logging.WARNING)  # Only warnings and errors

piedomains.piedomains_logging.get_effective_level()[source]¶

Get the current effective logging level for console output.

Returns:: Current logging level name (e.g., ‘INFO’, ‘DEBUG’, ‘WARNING’).
Return type:: str

Example

>>> current_level = get_effective_level()
>>> print(f"Current log level: {current_level}")
Current log level: INFO

piedomains.piedomains_logging.disable_logging()[source]¶

Disable all piedomains logging output.

This is useful for testing or when running in quiet mode. Use configure_logging() to re-enable logging.

Example

>>> disable_logging()  # Silence all piedomains logs
>>> # ... run operations silently ...
>>> configure_logging()  # Re-enable logging

Return type:: None

piedomains.piedomains_logging.is_debug_enabled()[source]¶

Check if DEBUG level logging is currently enabled.

Returns:: True if DEBUG logging is enabled, False otherwise.
Return type:: bool

Example

>>> if is_debug_enabled():
...     logger.debug("This will only run if debug is enabled")

piedomains.text_processor module¶

Text processing utilities for domain content analysis.

Handles HTML content extraction, text cleaning, and preprocessing.

class piedomains.text_processor.TextProcessor[source]¶

Bases: object

Handles text extraction and cleaning from HTML content.

static extract_text_from_html(html_content)[source]¶

Extract clean, visible text from HTML content.

Parameters:: html_content (str) – Raw HTML content
Returns:: Cleaned visible text content
Return type:: str

static clean_and_normalize_text(text)[source]¶

Normalise page text for the model.

What this used to do, and why it was wrong. Until v0.12.0 this deduplicated tokens twice, sorted them alphabetically, and dropped every non-ASCII character. The model therefore received an alphabetised set of words – the stored training text is literally "accueil adresse alfonso aller anciens animation ans archives...". Across 14 real cached pages that discarded 73% of all words, and asahi.com kept 2.7% of its own.

Three things were lost, and each matters more than it sounds:

Term frequency. 200 mentions of “sports” and one sportsbook advert in the footer weighed exactly the same. That is the mechanism behind deadspin.com being classified gamble at 0.98 confidence, and 13% of predictions on Tranco-top-100k domains landing in blockable categories.
Word order, to an alphabetical sort – which nullifies the entire reason to use a contextual encoder.
Every non-Latin script, which made a multilingual model multilingual in name only.

None of it was unreasonable for the model this pipeline was built for: a GlobalAveragePooling1D bag-of-embeddings is order-invariant by construction. The preprocessing simply was not revisited when the model became mmBERT.

So this now extracts, collapses whitespace and lowercases. Nothing else. A transformer wants sentences.

Set text_cleaning="legacy" in the config to restore the old behaviour, which is required to reproduce v0.11.0 and earlier.

Parameters:: text (str) – Raw text to clean
Returns:: Normalised text, order and frequency intact.
Return type:: str
Raises:: AttributeError – If the input is not of the expected type.

classmethod process_html_to_text(html_content)[source]¶

Complete pipeline: extract text from HTML and clean it.

Parameters:: html_content (str) – Raw HTML content
Returns:: Clean, processed text ready for model input
Return type:: str

static extract_with_trafilatura(html_content)[source]¶

Extract main content using trafilatura.

trafilatura powers FineWeb and RefinedWeb and tops the WCXB benchmark on articles, and it is the default here — see config.py for the measurement. An earlier note in this file said the opposite, that it was worse at 16 of 33 pages under the token floor against the legacy cleaner’s 14. That comparison was invalid: it counted trafilatura’s raw words against the legacy path’s cleaned tokens, and the legacy cleaner deduplicated, so the two sides were not the same quantity.

What it does discard is navigation and tile chrome. That costs site-level signal on a homepage and removes a great deal of noise: on deadspin.com it cuts gambling tokens from 260 (7.1% of the page) to 7 (1.1%), which only started to matter once the cleaner stopped collapsing repeated words to one.

Parameters:: html_content (str) – Raw HTML content.
Returns:: Extracted main content, or "" when nothing was found.
Return type:: str

classmethod extract_text(html_content)[source]¶

Extract page text using the configured extractor.

Defaults to the legacy visible-text walk, because the shipped TensorFlow model was trained on its output – switching wholesale would break train/serve parity. Set extractor = "trafilatura" in config (or train a model on its output) to use the better one.

Parameters:: html_content (str) – Raw HTML content.
Returns:: Extracted text, falling back to the legacy walk if the configured extractor returns nothing.
Return type:: str

piedomains.text module¶

Text-based domain classification using HTML content analysis.

piedomains.text.DEFAULT_TEXT_MODEL = 'gojiberries/piedomains-text'¶: Where the fine-tuned classifier lives. Overridable with PIEDOMAINS_TEXT_MODEL (a Hub repo id or a local directory), which is what training/train_text.py output is pointed at before it is published.

piedomains.text.resolve_text_model(latest=False)[source]¶

Decide which checkpoint to load.

A local directory is preferred when one is configured, so a freshly trained model can be evaluated before it is published anywhere.

Parameters:: latest (bool) – Bypass any local cache and re-resolve from the Hub.
Returns:: A local path or a Hugging Face Hub repo id.
Return type:: str

class piedomains.text.TextClassifier(cache_dir=None, archive_date=None)[source]¶

Bases: object

Text-based domain content classifier.

Parameters:

cache_dir (str | None)
archive_date (str | None)

load_models(latest=False)[source]¶

Load the classifier, its tokenizer, labels and temperature.

The model is a HuggingFace sequence-classification checkpoint written by training/train_text.py: safetensors weights, a tokenizer, a labels.json giving class order, and a calibration.json holding the fitted temperature.

Parameters:: latest (bool) – Re-resolve the model even if one is already loaded.
Raises:: RuntimeError – If the model cannot be loaded. Deliberately fatal – this used to substitute a model returning all zeros, so every domain came back as adv at confidence 0.0, indistinguishable from a real prediction.
Return type:: None

classify(domains, latest=False)[source]¶

Classify domains using their cached HTML content.

Parameters:

domains (list[str]) – List of domain names to classify
latest (bool) – Whether to use latest model version

Returns:

List of classification result dictionaries

Return type:

list[dict]

Example

>>> classifier = TextClassifier()
>>> results = classifier.classify(["cnn.com", "bbc.com"])
>>> print(results[0]["category"])
news

classify_from_paths(data_paths, output_file=None, latest=False)[source]¶

Classify domains using HTML files from collected data paths.

Parameters:

data_paths (list[dict]) – List of dicts with domain data containing text_path, domain, etc.
output_file (str | None) – Optional path to save JSON results
latest (bool) – Whether to use latest model version

Returns:

List of classification result dictionaries (JSON format)

Return type:

list[dict]

Example

>>> classifier = TextClassifier()
>>> data = [{"domain": "cnn.com", "text_path": "html/cnn.com.html", ...}]
>>> results = classifier.classify_from_paths(data)
>>> print(results[0]["category"])
news

classify_from_data(collection_data, output_file=None, latest=False)[source]¶

Classify domains using collection metadata from DataCollector.

Parameters:

collection_data (dict) – Collection metadata dict from DataCollector.collect()
output_file (str | None) – Optional path to save JSON results
latest (bool) – Whether to use latest model version

Returns:

List of classification result dictionaries (JSON format)

Return type:

list[dict]

Example

>>> from piedomains import DataCollector
>>> collector = DataCollector()
>>> data = collector.collect(["cnn.com"])
>>> classifier = TextClassifier()
>>> results = classifier.classify_from_data(data)

piedomains.utils module¶

Utility functions for file operations, downloads, and security.

This module provides essential utility functions for the piedomains package, including secure file downloads, tar archive extraction with path traversal protection, and configuration management.

The utilities focus on security-first implementation, particularly for handling downloaded archives and preventing common security vulnerabilities like path traversal attacks.

piedomains.utils.REPO_BASE_URL = 'https://dataverse.harvard.edu/api/access/datafile/7081895'¶

Base URL for model data repository.

Can be overridden via PIEDOMAINS_MODEL_URL environment variable. Defaults to Harvard Dataverse hosting the piedomains model files.

Type:: str

piedomains.utils.download_file(url, target, file_name, timeout=30)[source]¶

Download and extract a compressed model file from a remote repository.

This function downloads a tar.gz file from the specified URL, saves it to the target directory, extracts it using secure extraction methods, and cleans up the downloaded archive file.

Parameters:

url (str) – URL of the remote file to download. Should point to a valid tar.gz archive containing model data.
target (str) – Local directory path where the file should be downloaded and extracted. Directory will be created if it doesn’t exist.
file_name (str) – Name to use for the downloaded file. Should include appropriate extension (e.g., “model.tar.gz”).
timeout (int) – HTTP request timeout in seconds. Defaults to 30 seconds for large model files.

Returns:

True if download and extraction completed successfully,: False if any error occurred during the process.

Return type:

bool

Example

>>> success = download_file(
...     url="https://example.com/model.tar.gz",
...     target="/path/to/models",
...     file_name="text_model.tar.gz"
... )
>>> if success:
...     print("Model downloaded and extracted successfully")

Security:

Uses safe_extract() to prevent path traversal attacks
Validates archive contents before extraction
Automatically removes downloaded archive after extraction
Logs all errors for security monitoring

Note

The downloaded tar.gz file is automatically deleted after extraction to save disk space. Only the extracted contents remain in the target directory.

piedomains.utils.is_within_directory(directory, target)[source]¶

Check if a target path is within a specified directory (security check).

This function validates that a file path is contained within a directory to prevent path traversal attacks when extracting archives. It resolves all symbolic links and relative path components before comparison.

Parameters:

directory (str) – The base directory path that should contain the target.
target (str) – The target file/directory path to validate.

Returns:

True if target is within directory, False if it would escape: the directory boundary (indicating a potential path traversal attack).

Return type:

bool

Example

>>> # Safe path
>>> is_within_directory("/safe/dir", "/safe/dir/file.txt")
True

>>> # Path traversal attempt
>>> is_within_directory("/safe/dir", "/safe/dir/../../../etc/passwd")
False

>>> # Another traversal attempt
>>> is_within_directory("/safe/dir", "/safe/dir/subdir/../../../etc/passwd")
False

Security:: This function is critical for preventing path traversal attacks (also known as directory traversal or dot-dot-slash attacks) where malicious archives attempt to extract files outside the intended directory.

Note

This function uses os.path.abspath() to resolve all relative path components and symbolic links before performing the security check.

piedomains.utils.safe_extract(tar, path='.', members=None, *, numeric_owner=False)[source]¶

Securely extract a tar archive with path traversal protection.

This function provides a secure wrapper around tarfile.extractall() that validates all archive members to prevent path traversal attacks. It checks each member’s path before extraction to ensure it stays within the target directory.

Parameters:

tar (TarFile) – Open tar file object to extract from.
path (str) – Directory path where archive should be extracted. Defaults to current directory (“.”).
members (list | None) – Specific members to extract. If None, extracts all members. Defaults to None.
numeric_owner (bool) – If True, preserve numeric user/group IDs. If False, use current user. Defaults to False.

Return type:

None

Example

>>> import tarfile
>>> with tarfile.open("model.tar.gz", "r:gz") as tar:
...     safe_extract(tar, "/safe/extraction/dir")

Security:

Validates every archive member before extraction
Prevents path traversal attacks (e.g., “../../../etc/passwd”)
Logs security violations for monitoring
Raises exceptions rather than silently failing

Note

This function should always be used instead of tarfile.extractall() when handling archives from untrusted sources, which includes downloaded model files.

Raises:

SecurityError – If a security validation fails.

Parameters:

tar (TarFile)
path (str)
members (list | None)
numeric_owner (bool)

Return type:

None

exception piedomains.utils.SecurityError[source]¶

Bases: Exception

Exception raised for security violations during file operations.

This exception is raised when security checks fail, particularly during archive extraction when path traversal attempts are detected.

Example

>>> try:
...     safe_extract(malicious_tar, "/safe/dir")
... except SecurityError as e:
...     logger.error(f"Security violation: {e}")

piedomains.utils.get_file_hash(file_path, algorithm='sha256')[source]¶

Calculate cryptographic hash of a file for integrity verification.

Parameters:

file_path (str) – Path to the file to hash.
algorithm (str) – Hash algorithm to use (‘md5’, ‘sha1’, ‘sha256’, ‘sha512’). Defaults to ‘sha256’ for security.

Returns:

Hexadecimal hash digest of the file.

Return type:

str

Example

>>> hash_value = get_file_hash("model.tar.gz", "sha256")
>>> print(f"File hash: {hash_value}")

Raises:

FileNotFoundError – If the referenced file does not exist.
OSError – If a filesystem operation fails.
ValueError – If an argument is invalid.

Parameters:

file_path (str)
algorithm (str)

Return type:

str

Module contents¶

Piedomains: Domain content classification library.

This module provides lazy imports to avoid dependency issues when optional dependencies (like playwright) are not installed.

piedomains package¶

Subpackages¶

Submodules¶

piedomains.api module¶

piedomains.cli module¶

piedomains.config module¶

piedomains.constants module¶

piedomains.content_processor module¶

piedomains.content_validation module¶

piedomains.context_managers module¶

piedomains.data_collector module¶

piedomains.fetchers module¶

piedomains.http_client module¶

piedomains.image module¶

piedomains.llm_classifier module¶

piedomains.piedomain module¶

piedomains.piedomains_logging module¶

piedomains.text_processor module¶

piedomains.text module¶

piedomains.utils module¶

Module contents¶