scripts/feature_extraction/html/extract_features.py

"""

Parallel HTML Feature Extraction Pipeline



Processes ~80k HTML files using multiprocessing for CPU-bound parsing.

Integrates quality filtering INTO the same parse pass (no double-parsing).

Includes checkpointing, progress tracking, and balanced output.



Usage:

    python scripts/feature_extraction/html/extract_features.py

    python scripts/feature_extraction/html/extract_features.py --no-filter

    python scripts/feature_extraction/html/extract_features.py --workers 8

"""
import argparse
import json
import logging
import sys
import time
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path

import pandas as pd
from tqdm import tqdm

# ---------------------------------------------------------------------------
# Resolve project root so imports work regardless of cwd
# ---------------------------------------------------------------------------
PROJECT_ROOT = Path(__file__).resolve().parents[3]  # src/
sys.path.insert(0, str(PROJECT_ROOT))

from scripts.feature_extraction.html.html_feature_extractor import HTMLFeatureExtractor

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S',
)
logger = logging.getLogger('extract_features')

# ---------------------------------------------------------------------------
# Quality filter constants
# ---------------------------------------------------------------------------
MIN_FILE_SIZE = 800           # bytes
MIN_TAGS = 8
MIN_WORDS = 30
ERROR_PATTERNS = [
    'page not found', '404 not found', '403 forbidden',
    'access denied', 'server error', 'not available',
    'domain for sale', 'website expired', 'coming soon',
    'under construction', 'parked domain', 'buy this domain',
    'domain has expired', 'this site can',
]


# ---------------------------------------------------------------------------
# Worker function – runs in a subprocess
# ---------------------------------------------------------------------------
def _process_file(args: tuple) -> dict | None:
    """

    Process a single HTML file: read → (optionally filter) → extract.



    Designed to run inside ProcessPoolExecutor – must be picklable

    (top-level function, not a method).



    Args:

        args: (file_path_str, label, apply_filter)



    Returns:

        Feature dict with 'filename' and 'label' added, or None on skip/error.

    """
    file_path_str, label, apply_filter = args

    try:
        path = Path(file_path_str)

        # --- Read file ---
        raw = path.read_text(encoding='utf-8', errors='ignore')

        # --- Quick pre-filter (before expensive parse) ---
        if apply_filter and len(raw) < MIN_FILE_SIZE:
            return None

        # --- Parse once with lxml ---
        from bs4 import BeautifulSoup
        try:
            soup = BeautifulSoup(raw, 'lxml')
        except Exception:
            soup = BeautifulSoup(raw, 'html.parser')

        # --- Quality filter (uses the already-parsed soup) ---
        if apply_filter:
            if not soup.find('body'):
                return None

            all_tags = soup.find_all()
            if len(all_tags) < MIN_TAGS:
                return None

            text = soup.get_text(separator=' ', strip=True).lower()
            words = text.split()
            if len(words) < MIN_WORDS:
                return None

            # Check error-page patterns (first 2000 chars only)
            text_head = text[:2000]
            for pat in ERROR_PATTERNS:
                if pat in text_head:
                    return None

            # Must have at least some content elements
            has_content = (
                len(soup.find_all('a')) > 0 or
                len(soup.find_all('form')) > 0 or
                len(soup.find_all('input')) > 0 or
                len(soup.find_all('img')) > 0 or
                len(soup.find_all('div')) > 3
            )
            if not has_content:
                return None

        # --- Extract features (re-parses internally with cache) ---
        extractor = HTMLFeatureExtractor()
        features = extractor.extract_features(raw)
        features['filename'] = path.name
        features['label'] = label
        return features

    except Exception:
        return None


# ---------------------------------------------------------------------------
# Directory processor
# ---------------------------------------------------------------------------
def extract_from_directory(

    html_dir: Path,

    label: int,

    apply_filter: bool = True,

    max_workers: int = 6,

    limit: int | None = None,

) -> list[dict]:
    """

    Extract features from all .html files in a directory using multiprocessing.



    Args:

        html_dir: Directory with .html files

        label: 0 = legitimate, 1 = phishing

        apply_filter: Apply quality filter

        max_workers: Number of parallel workers

        limit: Max files to return (None = all)



    Returns:

        List of feature dictionaries

    """
    html_files = sorted(html_dir.glob('*.html'))
    total = len(html_files)
    label_name = 'Phishing' if label == 1 else 'Legitimate'

    logger.info(f"\n{'='*60}")
    logger.info(f"Processing {label_name}: {total:,} files")
    logger.info(f"  Directory: {html_dir}")
    logger.info(f"  Quality filter: {'ON' if apply_filter else 'OFF'}")
    logger.info(f"  Workers: {max_workers}")
    logger.info(f"{'='*60}")

    # Build task list
    tasks = [(str(f), label, apply_filter) for f in html_files]

    results = []
    n_filtered = 0
    t0 = time.perf_counter()

    with ProcessPoolExecutor(max_workers=max_workers) as pool:
        futures = {pool.submit(_process_file, t): t for t in tasks}

        with tqdm(total=total, desc=f'{label_name}', unit='file') as pbar:
            for future in as_completed(futures):
                pbar.update(1)
                result = future.result()
                if result is None:
                    n_filtered += 1
                else:
                    results.append(result)
                    if limit and len(results) >= limit:
                        # Cancel remaining futures
                        for f in futures:
                            f.cancel()
                        break

    elapsed = time.perf_counter() - t0
    speed = total / elapsed if elapsed > 0 else 0

    logger.info(f"  Extracted: {len(results):,} quality samples")
    logger.info(f"  Filtered out: {n_filtered:,} ({n_filtered/max(total,1)*100:.1f}%)")
    logger.info(f"  Time: {elapsed:.1f}s ({speed:.0f} files/sec)")

    return results


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main():
    parser = argparse.ArgumentParser(
        description='Extract HTML features for phishing detection (parallel)')
    parser.add_argument('--phishing-dir', type=str, nargs='+',
                        default=['data/html/phishing', 'data/html/phishing_v1'],
                        help='Directories with phishing HTML files')
    parser.add_argument('--legit-dir', type=str, nargs='+',
                        default=['data/html/legitimate', 'data/html/legitimate_v1'],
                        help='Directories with legitimate HTML files')
    parser.add_argument('--output', type=str, default='data/features/html_features.csv',
                        help='Output CSV path')
    parser.add_argument('--workers', type=int, default=6,
                        help='Number of parallel workers (default: 6)')
    parser.add_argument('--no-filter', action='store_true',
                        help='Disable quality filtering')
    parser.add_argument('--limit', type=int, default=None,
                        help='Limit samples per class (for testing)')
    parser.add_argument('--no-balance', action='store_true',
                        help='Do not balance classes')
    args = parser.parse_args()

    apply_filter = not args.no_filter

    # Resolve paths relative to project root
    phishing_dirs = [(PROJECT_ROOT / d).resolve() for d in args.phishing_dir]
    legit_dirs = [(PROJECT_ROOT / d).resolve() for d in args.legit_dir]
    output_path = (PROJECT_ROOT / args.output).resolve()

    logger.info("=" * 70)
    logger.info("HTML FEATURE EXTRACTION PIPELINE")
    logger.info("=" * 70)
    for d in phishing_dirs:
        logger.info(f"  Phishing dir:  {d}")
    for d in legit_dirs:
        logger.info(f"  Legitimate dir: {d}")
    logger.info(f"  Output:        {output_path}")
    logger.info(f"  Workers:       {args.workers}")
    logger.info(f"  Quality filter: {'ON' if apply_filter else 'OFF'}")

    # Validate directories
    for d in phishing_dirs:
        if not d.exists():
            logger.warning(f"Phishing directory not found (skipping): {d}")
    for d in legit_dirs:
        if not d.exists():
            logger.warning(f"Legitimate directory not found (skipping): {d}")

    # ---- Extract features ----
    t_start = time.perf_counter()

    phishing_features = []
    for d in phishing_dirs:
        if d.exists():
            phishing_features.extend(extract_from_directory(
                d, label=1, apply_filter=apply_filter,
                max_workers=args.workers, limit=args.limit))

    legit_features = []
    for d in legit_dirs:
        if d.exists():
            legit_features.extend(extract_from_directory(
                d, label=0, apply_filter=apply_filter,
                max_workers=args.workers, limit=args.limit))

    # ---- Balance ----
    if not args.no_balance:
        min_count = min(len(phishing_features), len(legit_features))
        logger.info(f"\nBalancing to {min_count:,} per class")
        # Shuffle before truncating to get random sample
        import random
        random.seed(42)
        random.shuffle(phishing_features)
        random.shuffle(legit_features)
        phishing_features = phishing_features[:min_count]
        legit_features = legit_features[:min_count]

    # ---- Build DataFrame ----
    all_features = phishing_features + legit_features
    if not all_features:
        logger.error("No features extracted!")
        sys.exit(1)

    df = pd.DataFrame(all_features)

    # Reorder columns: metadata first, then sorted features
    meta_cols = ['filename', 'label']
    feature_cols = sorted([c for c in df.columns if c not in meta_cols])
    df = df[meta_cols + feature_cols]

    # Shuffle rows
    df = df.sample(frac=1, random_state=42).reset_index(drop=True)

    # ---- Save ----
    output_path.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(output_path, index=False)

    elapsed = time.perf_counter() - t_start

    # ---- Summary ----
    logger.info("\n" + "=" * 70)
    logger.info("EXTRACTION COMPLETE")
    logger.info("=" * 70)
    logger.info(f"  Total samples: {len(df):,}")
    logger.info(f"    Phishing:    {(df['label']==1).sum():,}")
    logger.info(f"    Legitimate:  {(df['label']==0).sum():,}")
    logger.info(f"  Features:      {len(feature_cols)}")
    logger.info(f"  Total time:    {elapsed:.1f}s")
    logger.info(f"  Output:        {output_path}")
    logger.info("=" * 70)

    # Quick stats
    numeric = df[feature_cols].describe().T[['mean', 'std', 'min', 'max']]
    logger.info(f"\nFeature statistics (sample):")
    logger.info(numeric.head(15).to_string())


if __name__ == '__main__':
    main()