scripts/normalize.py

#!/usr/bin/env python3
"""
URL Normalization Script for GDELT Engine

This isolated Python script normalizes URLs for deduplication.
It can be called from Go via exec.Command or used standalone.

Usage:
    python normalize.py <url>
    echo "url1\nurl2" | python normalize.py --stdin
"""

import sys
import re
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode

# Tracking parameters to remove
TRACKING_PARAMS = {
    # UTM parameters
    'utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
    'utm_id', 'utm_cid',
    # Social media tracking
    'fbclid', 'gclid', 'dclid', 'msclkid', 'twclid', 'igshid',
    # Analytics
    'ref', 'source', 'campaign', 'mc_cid', 'mc_eid',
    '_ga', '_gl', '_hsenc', '_hsmi',
    # News specific
    'ncid', 'ocid', 'cmpid', 'ns_mchannel', 'ns_source', 'ns_campaign',
    # General
    'share', 'action', 'module', 'feed', 'rss',
}

# URL shortener domains that should not be normalized
SHORTENERS = {
    'bit.ly', 't.co', 'goo.gl', 'ow.ly', 'tinyurl.com', 'is.gd',
    'buff.ly', 'dlvr.it', 'j.mp', 'spr.ly', 'ht.ly',
}


def normalize_url(url: str) -> str:
    """
    Normalize a URL for deduplication.
    
    Steps:
    1. Parse URL
    2. Ensure scheme (default http)
    3. Lowercase host
    4. Remove www prefix
    5. Remove tracking parameters
    6. Remove fragment
    7. Clean path (remove trailing slash, normalize)
    """
    if not url:
        return ''
    
    # Ensure scheme
    if not url.startswith(('http://', 'https://')):
        url = 'http://' + url
    
    try:
        parsed = urlparse(url)
    except Exception:
        return ''
    
    # Skip shorteners - return as-is
    host = parsed.netloc.lower()
    if any(short in host for short in SHORTENERS):
        return url
    
    # Lowercase and clean host
    host = host.lower()
    if host.startswith('www.'):
        host = host[4:]
    
    # Remove tracking parameters
    query_params = parse_qs(parsed.query, keep_blank_values=True)
    clean_params = {
        k: v for k, v in query_params.items()
        if k.lower() not in TRACKING_PARAMS
    }
    clean_query = urlencode(clean_params, doseq=True) if clean_params else ''
    
    # Clean path
    path = parsed.path
    # Remove trailing slash (except for root)
    if path != '/' and path.endswith('/'):
        path = path.rstrip('/')
    # Normalize double slashes
    path = re.sub(r'/+', '/', path)
    
    # Reconstruct URL without fragment
    normalized = urlunparse((
        parsed.scheme,
        host,
        path,
        parsed.params,
        clean_query,
        ''  # No fragment
    ))
    
    return normalized


def extract_domain(url: str) -> str:
    """Extract the domain from a URL."""
    try:
        parsed = urlparse(url)
        host = parsed.netloc.lower()
        if host.startswith('www.'):
            host = host[4:]
        return host
    except Exception:
        return ''


def is_valid_news_url(url: str) -> bool:
    """Check if URL is likely a news article (not homepage, category, etc)."""
    if not url:
        return False
    
    try:
        parsed = urlparse(url)
        path = parsed.path
        
        # Must have a path beyond just /
        if not path or path == '/':
            return False
        
        # Common non-article patterns
        non_article_patterns = [
            r'^/category/',
            r'^/tag/',
            r'^/author/',
            r'^/page/',
            r'^/search',
            r'^/about',
            r'^/contact',
            r'^/privacy',
            r'^/terms',
            r'/feed/?$',
            r'/rss/?$',
        ]
        
        for pattern in non_article_patterns:
            if re.match(pattern, path, re.IGNORECASE):
                return False
        
        # Likely an article if has date-like pattern or enough path depth
        if re.search(r'/\d{4}/\d{2}/', path):  # Date in URL
            return True
        if path.count('/') >= 2:  # Multiple path segments
            return True
        if re.search(r'\.\w{3,4}$', path):  # Has file extension
            return True
        if len(path) > 20:  # Long path likely an article
            return True
            
        return True  # Default to true
        
    except Exception:
        return False


def main():
    """Main entry point."""
    if len(sys.argv) > 1:
        if sys.argv[1] == '--stdin':
            # Read URLs from stdin
            for line in sys.stdin:
                url = line.strip()
                if url:
                    normalized = normalize_url(url)
                    print(normalized)
        elif sys.argv[1] == '--validate':
            # Validate URLs from stdin
            for line in sys.stdin:
                url = line.strip()
                if url and is_valid_news_url(url):
                    print(normalize_url(url))
        else:
            # Single URL argument
            print(normalize_url(sys.argv[1]))
    else:
        print("Usage: normalize.py <url>", file=sys.stderr)
        print("       normalize.py --stdin", file=sys.stderr)
        print("       normalize.py --validate", file=sys.stderr)
        sys.exit(1)


if __name__ == '__main__':
    main()