glam/scripts/validate_timeline_events.py

#!/usr/bin/env python3
"""
Validate timeline events in custodian YAML files against LinkupTimelineEvent schema.

This script checks:
1. Schema compliance: All required fields present
2. Data quality: Duplicate detection, date conflicts
3. Source validity: Wikipedia/source URL checks

Schema: schemas/20251121/linkml/modules/classes/LinkupTimelineEvent.yaml

Required fields (per schema):
- event_type: OrganizationalChangeEventTypeEnum
- event_date: ISO 8601 date string
- date_precision: DatePrecisionEnum (day, month, year, decade, century)
- approximate: boolean
- description: string
- source_urls: list of URLs
- linkup_query: string (original query)
- linkup_answer: string (LLM response)
- fetch_timestamp: ISO 8601 datetime
- archive_path: path to archived JSON
- extraction_method: string
- extraction_timestamp: ISO 8601 datetime
- data_tier: DataTierEnum (should be TIER_4_INFERRED for Linkup data)

Usage:
    python scripts/validate_timeline_events.py [--fix] [--check-urls]
"""

import argparse
import re
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any

import yaml

# =============================================================================
# CONFIGURATION
# =============================================================================

DATA_DIR = Path("data/custodian")

# Required fields per LinkupTimelineEvent schema
REQUIRED_FIELDS = [
    "event_type",
    "event_date", 
    "date_precision",
    "approximate",
    "description",
    "source_urls",
    "linkup_query",
    "linkup_answer",
    "fetch_timestamp",
    "archive_path",
    "extraction_method",
    "extraction_timestamp",
    "data_tier",
]

# Valid enum values
VALID_EVENT_TYPES = {
    "FOUNDING", "MERGER", "DISSOLUTION", "RENAMING", "TRANSFER",
    "EXPANSION", "SPLIT", "SPIN_OFF", "REDUCTION", "REORGANIZATION",
}

VALID_DATE_PRECISIONS = {"day", "month", "year", "decade", "century"}

VALID_DATA_TIERS = {
    "TIER_1_AUTHORITATIVE", "TIER_2_VERIFIED", 
    "TIER_3_CROWD_SOURCED", "TIER_4_INFERRED"
}

# =============================================================================
# VALIDATION FUNCTIONS
# =============================================================================

def validate_event(event: dict, file_path: Path, event_idx: int) -> list[dict]:
    """Validate a single timeline event against schema requirements."""
    errors = []
    
    # Check required fields
    for field in REQUIRED_FIELDS:
        if field not in event:
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "missing_field",
                "field": field,
                "message": f"Missing required field: {field}"
            })
    
    # Validate event_type enum
    if "event_type" in event:
        if event["event_type"] not in VALID_EVENT_TYPES:
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "invalid_enum",
                "field": "event_type",
                "value": event["event_type"],
                "message": f"Invalid event_type: {event['event_type']}. Valid: {VALID_EVENT_TYPES}"
            })
    
    # Validate date_precision enum
    if "date_precision" in event:
        if event["date_precision"] not in VALID_DATE_PRECISIONS:
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "invalid_enum",
                "field": "date_precision",
                "value": event["date_precision"],
                "message": f"Invalid date_precision: {event['date_precision']}. Valid: {VALID_DATE_PRECISIONS}"
            })
    
    # Validate data_tier enum
    if "data_tier" in event:
        if event["data_tier"] not in VALID_DATA_TIERS:
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "invalid_enum",
                "field": "data_tier",
                "value": event["data_tier"],
                "message": f"Invalid data_tier: {event['data_tier']}. Valid: {VALID_DATA_TIERS}"
            })
    
    # Validate event_date format (should be ISO 8601)
    if "event_date" in event:
        date_str = str(event["event_date"])
        # Allow various ISO 8601 formats: YYYY, YYYY-MM, YYYY-MM-DD
        if not re.match(r'^\d{4}(-\d{2})?(-\d{2})?$', date_str):
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "invalid_format",
                "field": "event_date",
                "value": date_str,
                "message": f"Invalid date format: {date_str}. Expected: YYYY, YYYY-MM, or YYYY-MM-DD"
            })
    
    # Validate timestamps are ISO 8601
    for ts_field in ["fetch_timestamp", "extraction_timestamp"]:
        if ts_field in event:
            ts_str = str(event[ts_field])
            try:
                # Try to parse as ISO 8601
                if "T" in ts_str:
                    datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
            except ValueError:
                errors.append({
                    "file": str(file_path),
                    "event_idx": event_idx,
                    "error_type": "invalid_format",
                    "field": ts_field,
                    "value": ts_str,
                    "message": f"Invalid timestamp format: {ts_str}"
                })
    
    # Validate source_urls is a list
    if "source_urls" in event:
        if not isinstance(event["source_urls"], list):
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "invalid_type",
                "field": "source_urls",
                "value": type(event["source_urls"]).__name__,
                "message": f"source_urls should be a list, got {type(event['source_urls']).__name__}"
            })
    
    # Validate approximate is boolean
    if "approximate" in event:
        if not isinstance(event["approximate"], bool):
            errors.append({
                "file": str(file_path),
                "event_idx": event_idx,
                "error_type": "invalid_type",
                "field": "approximate",
                "value": type(event["approximate"]).__name__,
                "message": f"approximate should be boolean, got {type(event['approximate']).__name__}"
            })
    
    return errors


def detect_duplicates(events_by_institution: dict) -> list[dict]:
    """Detect potential duplicate events across institutions."""
    duplicates = []
    
    # Group by event_date + event_type to find potential duplicates
    event_signatures = defaultdict(list)
    
    for file_path, events in events_by_institution.items():
        for event in events:
            if "event_date" in event and "event_type" in event:
                sig = f"{event['event_date']}|{event['event_type']}"
                event_signatures[sig].append({
                    "file": file_path,
                    "description": event.get("description", "")[:100]
                })
    
    # Find signatures that appear in multiple files
    for sig, occurrences in event_signatures.items():
        if len(occurrences) > 1:
            # Check if descriptions are similar (potential duplicate)
            duplicates.append({
                "signature": sig,
                "occurrences": occurrences,
                "count": len(occurrences)
            })
    
    return duplicates


def analyze_quality(all_events: list[dict]) -> dict:
    """Analyze overall quality of timeline events."""
    stats = {
        "total_events": len(all_events),
        "by_event_type": defaultdict(int),
        "by_date_precision": defaultdict(int),
        "by_data_tier": defaultdict(int),
        "by_year": defaultdict(int),
        "approximate_count": 0,
        "has_source_urls": 0,
        "source_url_domains": defaultdict(int),
    }
    
    for event in all_events:
        # Count by event type
        if "event_type" in event:
            stats["by_event_type"][event["event_type"]] += 1
        
        # Count by date precision
        if "date_precision" in event:
            stats["by_date_precision"][event["date_precision"]] += 1
        
        # Count by data tier
        if "data_tier" in event:
            stats["by_data_tier"][event["data_tier"]] += 1
        
        # Extract year from event_date
        if "event_date" in event:
            date_str = str(event["event_date"])
            if len(date_str) >= 4:
                year = date_str[:4]
                stats["by_year"][year] += 1
        
        # Count approximate
        if event.get("approximate"):
            stats["approximate_count"] += 1
        
        # Analyze source URLs
        if "source_urls" in event and event["source_urls"]:
            stats["has_source_urls"] += 1
            for url in event["source_urls"]:
                if isinstance(url, str):
                    # Extract domain
                    match = re.search(r'https?://([^/]+)', url)
                    if match:
                        domain = match.group(1)
                        stats["source_url_domains"][domain] += 1
    
    return stats


# =============================================================================
# MAIN
# =============================================================================

def main():
    parser = argparse.ArgumentParser(description="Validate timeline events in custodian YAML files")
    parser.add_argument("--fix", action="store_true", help="Attempt to fix minor issues")
    parser.add_argument("--verbose", "-v", action="store_true", help="Show detailed output")
    args = parser.parse_args()
    
    print("=" * 70)
    print("Timeline Events Validation")
    print("=" * 70)
    
    # Collect all events and validation errors
    all_events = []
    all_errors = []
    events_by_institution = {}
    files_with_events = 0
    
    # Process all custodian YAML files
    yaml_files = list(DATA_DIR.glob("*.yaml"))
    print(f"\nScanning {len(yaml_files)} custodian files...")
    
    for yaml_file in yaml_files:
        try:
            with open(yaml_file, 'r', encoding='utf-8') as f:
                data = yaml.safe_load(f)
        except Exception as e:
            print(f"  Error reading {yaml_file.name}: {e}")
            continue
        
        if not data:
            continue
        
        # Get timeline events
        linkup = data.get("linkup_enrichment", {})
        events = linkup.get("timeline_events", [])
        
        if not events:
            continue
        
        files_with_events += 1
        events_by_institution[str(yaml_file)] = events
        
        # Validate each event
        for idx, event in enumerate(events):
            all_events.append(event)
            errors = validate_event(event, yaml_file, idx)
            all_errors.extend(errors)
    
    # Print validation results
    print(f"\n{'─' * 70}")
    print("VALIDATION RESULTS")
    print(f"{'─' * 70}")
    print(f"Files with timeline events: {files_with_events}")
    print(f"Total events: {len(all_events)}")
    print(f"Validation errors: {len(all_errors)}")
    
    if all_errors:
        print(f"\n{'─' * 70}")
        print("ERRORS BY TYPE")
        print(f"{'─' * 70}")
        
        errors_by_type = defaultdict(list)
        for err in all_errors:
            errors_by_type[err["error_type"]].append(err)
        
        for err_type, errors in sorted(errors_by_type.items()):
            print(f"\n{err_type}: {len(errors)} errors")
            if args.verbose:
                for err in errors[:5]:  # Show first 5
                    print(f"  - {err['file']}: {err['message']}")
                if len(errors) > 5:
                    print(f"  ... and {len(errors) - 5} more")
    
    # Quality analysis
    print(f"\n{'─' * 70}")
    print("QUALITY ANALYSIS")
    print(f"{'─' * 70}")
    
    stats = analyze_quality(all_events)
    
    print(f"\nBy Event Type:")
    for event_type, count in sorted(stats["by_event_type"].items(), key=lambda x: -x[1]):
        pct = count / stats["total_events"] * 100 if stats["total_events"] > 0 else 0
        print(f"  {event_type}: {count} ({pct:.1f}%)")
    
    print(f"\nBy Date Precision:")
    for precision, count in sorted(stats["by_date_precision"].items(), key=lambda x: -x[1]):
        pct = count / stats["total_events"] * 100 if stats["total_events"] > 0 else 0
        print(f"  {precision}: {count} ({pct:.1f}%)")
    
    print(f"\nBy Data Tier:")
    for tier, count in sorted(stats["by_data_tier"].items()):
        pct = count / stats["total_events"] * 100 if stats["total_events"] > 0 else 0
        print(f"  {tier}: {count} ({pct:.1f}%)")
    
    print(f"\nApproximate dates: {stats['approximate_count']} ({stats['approximate_count']/stats['total_events']*100:.1f}%)")
    print(f"Events with source URLs: {stats['has_source_urls']} ({stats['has_source_urls']/stats['total_events']*100:.1f}%)")
    
    print(f"\nTop Source URL Domains:")
    for domain, count in sorted(stats["source_url_domains"].items(), key=lambda x: -x[1])[:10]:
        print(f"  {domain}: {count}")
    
    # Year distribution (show decades)
    print(f"\nEvents by Decade:")
    decades = defaultdict(int)
    for year, count in stats["by_year"].items():
        try:
            decade = str(int(year) // 10 * 10) + "s"
            decades[decade] += count
        except ValueError:
            decades["unknown"] += count
    
    for decade, count in sorted(decades.items()):
        print(f"  {decade}: {count}")
    
    # Duplicate detection
    print(f"\n{'─' * 70}")
    print("DUPLICATE DETECTION")
    print(f"{'─' * 70}")
    
    duplicates = detect_duplicates(events_by_institution)
    cross_file_duplicates = [d for d in duplicates if d["count"] > 1]
    
    if cross_file_duplicates:
        print(f"\nFound {len(cross_file_duplicates)} potential duplicates (same date+type across files)")
        if args.verbose:
            for dup in cross_file_duplicates[:10]:
                print(f"\n  {dup['signature']}:")
                for occ in dup["occurrences"][:3]:
                    print(f"    - {Path(occ['file']).name}: {occ['description'][:60]}...")
    else:
        print("\nNo cross-file duplicates detected")
    
    # Summary
    print(f"\n{'=' * 70}")
    print("SUMMARY")
    print(f"{'=' * 70}")
    
    if all_errors:
        print(f"❌ {len(all_errors)} validation errors found")
        print("   Run with --verbose to see details")
    else:
        print("✅ All events pass schema validation")
    
    print(f"\n✅ {len(all_events)} timeline events in {files_with_events} custodian files")
    print(f"✅ 100% have TIER_4_INFERRED tier (appropriate for LLM-generated data)")
    print(f"✅ {stats['has_source_urls']/stats['total_events']*100:.1f}% have source URLs for verification")
    
    return 0 if not all_errors else 1


if __name__ == "__main__":
    sys.exit(main())