#!/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 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 ", file=sys.stderr) print(" normalize.py --stdin", file=sys.stderr) print(" normalize.py --validate", file=sys.stderr) sys.exit(1) if __name__ == '__main__': main()