n0mad1k
143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
"""WikiLeaks-specific crawl helpers for Mosaic collectors."""
|
|
import re
|
|
import logging
|
|
from urllib.parse import urljoin
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def extract_extra_links(html: str, base_url: str, source_name: str) -> list[str]:
|
|
"""Extract additional links using WikiLeaks-specific patterns.
|
|
|
|
Different WikiLeaks collections have different page structures.
|
|
This function handles the quirks of each.
|
|
"""
|
|
links = []
|
|
|
|
if source_name == 'vault7':
|
|
links.extend(_vault7_links(html, base_url))
|
|
elif source_name == 'vault8':
|
|
links.extend(_vault8_links(html, base_url))
|
|
elif source_name == 'cablegate':
|
|
links.extend(_cablegate_links(html, base_url))
|
|
elif source_name == 'spyfiles':
|
|
links.extend(_spyfiles_links(html, base_url))
|
|
elif source_name == 'emails':
|
|
links.extend(_email_links(html, base_url))
|
|
|
|
return links
|
|
|
|
|
|
def _vault7_links(html: str, base_url: str) -> list[str]:
|
|
"""Extract Vault 7 project page links.
|
|
|
|
Vault 7 has a main index with project names that link to sub-pages.
|
|
Each project page has document links.
|
|
"""
|
|
links = []
|
|
# Match project links like /vault7/#Dark Matter
|
|
for match in re.finditer(r'href=["\'](/vault7/[^"\']+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
# Match document/file links
|
|
for match in re.finditer(r'href=["\']([^"\']*(?:\.pdf|\.zip|\.tar\.gz)[^"\']*)["\']', html, re.IGNORECASE):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
return links
|
|
|
|
|
|
def _vault8_links(html: str, base_url: str) -> list[str]:
|
|
"""Extract Vault 8 source code repository links."""
|
|
links = []
|
|
for match in re.finditer(r'href=["\'](/vault8/[^"\']+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
return links
|
|
|
|
|
|
def _cablegate_links(html: str, base_url: str) -> list[str]:
|
|
"""Extract Cablegate cable links with pagination support.
|
|
|
|
Cablegate uses /plusd/cables/ with date-based navigation.
|
|
Individual cables are at /plusd/cables/CABLE_ID
|
|
"""
|
|
links = []
|
|
|
|
# Individual cable links
|
|
for match in re.finditer(r'href=["\'](/plusd/cables/[0-9A-Z]+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
# Pagination links
|
|
for match in re.finditer(r'href=["\'](/plusd/[^"\']*(?:page|offset|start)=[^"\']+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
# Date-based navigation
|
|
for match in re.finditer(r'href=["\'](/plusd/cables/\?[^"\']+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
return links
|
|
|
|
|
|
def _spyfiles_links(html: str, base_url: str) -> list[str]:
|
|
"""Extract Spy Files document links."""
|
|
links = []
|
|
for match in re.finditer(r'href=["\'](/spyfiles/[^"\']+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
return links
|
|
|
|
|
|
def _email_links(html: str, base_url: str) -> list[str]:
|
|
"""Extract email dump links with ID-based navigation."""
|
|
links = []
|
|
|
|
# Email ID links
|
|
for match in re.finditer(r'href=["\']([^"\']*emailid/\d+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
# Pagination
|
|
for match in re.finditer(r'href=["\']([^"\']*(?:page|offset)=\d+)["\']', html):
|
|
link = urljoin(base_url, match.group(1))
|
|
links.append(link)
|
|
|
|
return links
|
|
|
|
|
|
def parse_cable_date(cable_id: str) -> Optional[str]:
|
|
"""Extract date from cable ID format.
|
|
|
|
Cable IDs often start with a two-digit year and month:
|
|
e.g., 07PARIS1234 → 2007 (approximate)
|
|
"""
|
|
match = re.match(r'^(\d{2})[A-Z]', cable_id)
|
|
if match:
|
|
year = int(match.group(1))
|
|
# WikiLeaks cables span roughly 1966-2010
|
|
if year >= 66:
|
|
return f"19{year:02d}"
|
|
else:
|
|
return f"20{year:02d}"
|
|
return None
|
|
|
|
|
|
def detect_wikileaks_source(url: str) -> Optional[str]:
|
|
"""Detect which WikiLeaks collection a URL belongs to."""
|
|
if '/vault7/' in url:
|
|
return 'vault7'
|
|
elif '/vault8/' in url:
|
|
return 'vault8'
|
|
elif '/plusd/' in url or '/cablegate/' in url:
|
|
return 'cablegate'
|
|
elif '/spyfiles/' in url:
|
|
return 'spyfiles'
|
|
elif 'emailid/' in url or '/dnc-emails/' in url or '/podesta-emails/' in url:
|
|
return 'emails'
|
|
return None
|