n0mad1k
229 lines
7.6 KiB
Python
229 lines
7.6 KiB
Python
"""Diplomatic cable parser for Mosaic — structured field extraction.
|
|
|
|
Cables have a specific format with fields like SUBJECT, ORIGIN,
|
|
CLASSIFICATION, TAGS, and a multi-section body.
|
|
"""
|
|
import re
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from .base import ParseResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Cable field patterns
|
|
CABLE_FIELDS = {
|
|
'VZCZCXYZ': 'transmission_id',
|
|
'DE ': 'from_code',
|
|
'ZNR': 'precedence',
|
|
'ZNY': 'precedence',
|
|
'FM ': 'from',
|
|
'TO ': 'to',
|
|
'INFO ': 'info',
|
|
'BT': 'break',
|
|
'UNCLAS': 'classification',
|
|
'CONFIDENTIAL': 'classification',
|
|
'SECRET': 'classification',
|
|
'SIPDIS': 'distribution',
|
|
'E.O.': 'executive_order',
|
|
'TAGS:': 'tags',
|
|
'SUBJECT:': 'subject',
|
|
'REF:': 'reference',
|
|
'REFS:': 'reference',
|
|
}
|
|
|
|
# Classification levels
|
|
CLASSIFICATIONS = ['UNCLASSIFIED', 'CONFIDENTIAL', 'SECRET', 'SECRET//NOFORN',
|
|
'CONFIDENTIAL//NOFORN', 'UNCLAS', 'UNCLAS SECTION']
|
|
|
|
# Section markers in cable bodies
|
|
SECTION_PATTERN = re.compile(r'^\d+\.\s+\((?:U|C|S|SBU)\)', re.MULTILINE)
|
|
|
|
# ANSI escape pattern
|
|
ANSI_ESCAPE = re.compile(r'\x1b\[[0-9;]*[a-zA-Z]|\x1b\].*?\x07')
|
|
|
|
|
|
class CableDocParser:
|
|
"""Parse diplomatic cables into structured fields."""
|
|
|
|
def parse(self, file_path: str) -> ParseResult:
|
|
"""Parse a diplomatic cable.
|
|
|
|
Extracts structured fields (SUBJECT, ORIGIN, CLASSIFICATION, TAGS, etc.)
|
|
as discrete fields BEFORE any chunking.
|
|
"""
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
|
|
content = f.read()
|
|
|
|
# Check if this is HTML (WikiLeaks format) or raw cable
|
|
if '<html' in content.lower()[:200] or '<!doctype' in content.lower()[:200]:
|
|
return self._parse_html_cable(content)
|
|
else:
|
|
return self._parse_raw_cable(content)
|
|
|
|
except Exception as e:
|
|
return ParseResult(error=f"Cable parse error: {str(e)}")
|
|
|
|
def _parse_raw_cable(self, content: str) -> ParseResult:
|
|
"""Parse a raw-format diplomatic cable."""
|
|
metadata = {}
|
|
|
|
lines = content.split('\n')
|
|
body_start = 0
|
|
|
|
# Extract header fields
|
|
for i, line in enumerate(lines):
|
|
line_stripped = line.strip()
|
|
if not line_stripped:
|
|
continue
|
|
|
|
# Classification
|
|
for cls in CLASSIFICATIONS:
|
|
if line_stripped.startswith(cls):
|
|
metadata['CLASSIFICATION'] = cls
|
|
break
|
|
|
|
# Named fields
|
|
if line_stripped.startswith('SUBJECT:'):
|
|
metadata['SUBJECT'] = line_stripped[8:].strip()
|
|
elif line_stripped.startswith('TAGS:'):
|
|
metadata['TAGS'] = line_stripped[5:].strip()
|
|
elif line_stripped.startswith('REF:') or line_stripped.startswith('REFS:'):
|
|
ref_start = 4 if line_stripped.startswith('REF:') else 5
|
|
metadata['REFERENCE'] = line_stripped[ref_start:].strip()
|
|
elif line_stripped.startswith('E.O.'):
|
|
metadata['EXECUTIVE_ORDER'] = line_stripped[4:].strip()
|
|
elif line_stripped.startswith('FM '):
|
|
metadata['ORIGIN'] = line_stripped[3:].strip()
|
|
elif line_stripped.startswith('TO '):
|
|
metadata.setdefault('TO', []).append(line_stripped[3:].strip())
|
|
|
|
# Detect body start (after numbered paragraph or after header block)
|
|
if SECTION_PATTERN.match(line_stripped):
|
|
body_start = i
|
|
break
|
|
if line_stripped == 'BT' and i > 5:
|
|
body_start = i + 1
|
|
break
|
|
|
|
if body_start == 0:
|
|
# Fallback: body starts after a double blank line
|
|
for i in range(len(lines) - 1):
|
|
if not lines[i].strip() and not lines[i+1].strip():
|
|
body_start = i + 2
|
|
break
|
|
|
|
body = '\n'.join(lines[body_start:]).strip()
|
|
|
|
# Extract sections from body
|
|
sections = self._extract_sections(body)
|
|
if sections:
|
|
metadata['sections'] = sections
|
|
|
|
# Build formatted text
|
|
text_parts = []
|
|
for key in ('SUBJECT', 'ORIGIN', 'CLASSIFICATION', 'TAGS', 'REFERENCE'):
|
|
if key in metadata:
|
|
val = metadata[key]
|
|
if isinstance(val, list):
|
|
val = ', '.join(val)
|
|
text_parts.append(f"{key}: {val}")
|
|
|
|
text_parts.append('')
|
|
text_parts.append(body)
|
|
|
|
text = '\n'.join(text_parts)
|
|
text = ANSI_ESCAPE.sub('', text)
|
|
|
|
return ParseResult(
|
|
success=True,
|
|
text=text,
|
|
metadata=metadata,
|
|
doc_type='cable',
|
|
title=metadata.get('SUBJECT', ''),
|
|
char_count=len(text),
|
|
)
|
|
|
|
def _parse_html_cable(self, content: str) -> ParseResult:
|
|
"""Parse a WikiLeaks HTML-format cable page."""
|
|
try:
|
|
from bs4 import BeautifulSoup
|
|
soup = BeautifulSoup(content, 'html.parser')
|
|
|
|
metadata = {}
|
|
|
|
# WikiLeaks cable pages have specific CSS classes/structures
|
|
# Try common patterns
|
|
title_el = soup.find('title')
|
|
if title_el:
|
|
metadata['SUBJECT'] = title_el.get_text(strip=True)
|
|
|
|
# Look for cable metadata in page
|
|
for label in ('origin', 'classification', 'tags', 'subject', 'date'):
|
|
el = soup.find(class_=re.compile(label, re.I))
|
|
if el:
|
|
metadata[label.upper()] = el.get_text(strip=True)
|
|
|
|
# Extract main content
|
|
body_el = (soup.find(id='cable-text') or
|
|
soup.find(class_='cable-text') or
|
|
soup.find('pre') or
|
|
soup.find(class_='content'))
|
|
|
|
if body_el:
|
|
body = body_el.get_text(separator='\n', strip=True)
|
|
else:
|
|
body = soup.get_text(separator='\n', strip=True)
|
|
|
|
# Build text
|
|
text_parts = []
|
|
for key in ('SUBJECT', 'ORIGIN', 'CLASSIFICATION', 'TAGS'):
|
|
if key in metadata:
|
|
text_parts.append(f"{key}: {metadata[key]}")
|
|
text_parts.append('')
|
|
text_parts.append(body)
|
|
|
|
text = '\n'.join(text_parts)
|
|
text = ANSI_ESCAPE.sub('', text)
|
|
|
|
return ParseResult(
|
|
success=True,
|
|
text=text,
|
|
metadata=metadata,
|
|
doc_type='cable',
|
|
title=metadata.get('SUBJECT', ''),
|
|
char_count=len(text),
|
|
)
|
|
|
|
except Exception as e:
|
|
return ParseResult(error=f"HTML cable parse error: {str(e)}")
|
|
|
|
def _extract_sections(self, body: str) -> list[dict]:
|
|
"""Extract numbered sections from cable body."""
|
|
sections = []
|
|
current_num = None
|
|
current_text = []
|
|
|
|
for line in body.split('\n'):
|
|
match = SECTION_PATTERN.match(line.strip())
|
|
if match:
|
|
if current_num is not None:
|
|
sections.append({
|
|
'number': current_num,
|
|
'text': '\n'.join(current_text).strip()
|
|
})
|
|
# Extract section number and classification
|
|
current_num = line.strip().split('.')[0]
|
|
current_text = [line]
|
|
elif current_num is not None:
|
|
current_text.append(line)
|
|
|
|
if current_num is not None:
|
|
sections.append({
|
|
'number': current_num,
|
|
'text': '\n'.join(current_text).strip()
|
|
})
|
|
|
|
return sections
|