cdx_discovery.py

#!/usr/bin/env python3
"""
ChatGPT Share Discovery Tool - CDX and Common Crawl Scanner

Discovers archived ChatGPT share URLs from web archive databases.
Supports Internet Archive CDX API and Common Crawl indexes.
"""

import sys
import io
import argparse
import json
import requests
from typing import Set, List, Dict
from datetime import datetime
from pathlib import Path
import time

# Windows UTF-8 fix
if sys.platform == 'win32':
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')

class CDXDiscovery:
    """Discovers ChatGPT shares from CDX archives"""

    def __init__(self):
        self.discovered_guids: Set[str] = set()
        self.discovered_urls: List[Dict] = []
        self.stats = {
            "total_urls": 0,
            "unique_guids": 0,
            "source": "",
            "start_time": datetime.now().isoformat(),
            "end_time": None
        }

    def discover_from_cdx(self, limit: int = 10000, from_date: str = None, to_date: str = None) -> None:
        """
        Discover shares from Internet Archive CDX API

        Args:
            limit: Maximum number of results to fetch
            from_date: Start date (YYYYMMDD format)
            to_date: End date (YYYYMMDD format)
        """
        print(f"🔍 Querying Internet Archive CDX API...")

        # Build CDX query parameters
        params = {
            "url": "chatgpt.com/share/",
            "matchType": "prefix",
            "output": "json",
            "limit": limit,
            "filter": "statuscode:200",
            "collapse": "urlkey"  # Deduplicate by URL
        }

        if from_date:
            params["from"] = from_date
        if to_date:
            params["to"] = to_date

        try:
            response = requests.get(
                "https://web.archive.org/cdx/search/cdx",
                params=params,
                timeout=60
            )
            response.raise_for_status()

            lines = response.text.strip().split('\n')
            if not lines or lines[0].strip() == '':
                print("⚠️  No results found")
                return

            # Parse JSON response (first line is header)
            data = [json.loads(line) for line in lines if line.strip()]

            # Extract GUIDs and build URL list
            for row in data:
                if len(row) >= 3:
                    url = row[2]  # Original URL
                    timestamp = row[1]  # Capture timestamp

                    # Extract GUID from URL
                    if '/share/' in url:
                        guid = url.split('/share/')[-1].split('?')[0].split('/')[0]
                        if len(guid) == 36 and guid.count('-') == 4:
                            self.discovered_guids.add(guid)
                            self.discovered_urls.append({
                                "url": url,
                                "guid": guid,
                                "timestamp": timestamp,
                                "archive": "wayback"
                            })

            self.stats["total_urls"] = len(self.discovered_urls)
            self.stats["unique_guids"] = len(self.discovered_guids)
            self.stats["source"] = "cdx"

            print(f"✅ Found {len(self.discovered_urls)} archived URLs")
            print(f"✅ Extracted {len(self.discovered_guids)} unique GUIDs")

        except requests.RequestException as e:
            print(f"❌ Error querying CDX API: {e}")
            sys.exit(1)

    def discover_from_commoncrawl(self, index: str = "CC-MAIN-2024-10", limit: int = 10000) -> None:
        """
        Discover shares from Common Crawl index

        Args:
            index: Common Crawl index name (e.g., CC-MAIN-2024-10)
            limit: Maximum results to process
        """
        print(f"🔍 Querying Common Crawl index: {index}...")

        # Common Crawl CDX API endpoint
        url = f"https://index.commoncrawl.org/{index}-index"
        params = {
            "url": "chatgpt.com/share/*",
            "output": "json",
            "limit": limit
        }

        try:
            response = requests.get(url, params=params, timeout=60)
            response.raise_for_status()

            lines = response.text.strip().split('\n')
            if not lines or lines[0].strip() == '':
                print("⚠️  No results found in this index")
                return

            for line in lines:
                if line.strip():
                    data = json.loads(line)
                    url = data.get('url', '')
                    timestamp = data.get('timestamp', '')

                    if '/share/' in url:
                        guid = url.split('/share/')[-1].split('?')[0].split('/')[0]
                        if len(guid) == 36 and guid.count('-') == 4:
                            self.discovered_guids.add(guid)
                            self.discovered_urls.append({
                                "url": url,
                                "guid": guid,
                                "timestamp": timestamp,
                                "archive": "commoncrawl",
                                "index": index
                            })

            self.stats["total_urls"] = len(self.discovered_urls)
            self.stats["unique_guids"] = len(self.discovered_guids)
            self.stats["source"] = f"commoncrawl_{index}"

            print(f"✅ Found {len(self.discovered_urls)} archived URLs")
            print(f"✅ Extracted {len(self.discovered_guids)} unique GUIDs")

        except requests.RequestException as e:
            print(f"❌ Error querying Common Crawl: {e}")
            sys.exit(1)

    def save_results(self, output_file: str, format: str = "txt") -> None:
        """Save discovered GUIDs to file"""
        output_path = Path(output_file)

        if format == "txt":
            # Save GUIDs only (one per line)
            with output_path.open('w', encoding='utf-8') as f:
                for guid in sorted(self.discovered_guids):
                    f.write(f"{guid}\n")
            print(f"💾 Saved {len(self.discovered_guids)} GUIDs to {output_file}")

        elif format == "json":
            # Save full details
            data = {
                "stats": self.stats,
                "discovered_urls": self.discovered_urls,
                "unique_guids": sorted(list(self.discovered_guids))
            }
            with output_path.open('w', encoding='utf-8') as f:
                json.dump(data, f, indent=2)
            print(f"💾 Saved detailed results to {output_file}")

        elif format == "urls":
            # Save full URLs (for verification tool)
            with output_path.open('w', encoding='utf-8') as f:
                for item in self.discovered_urls:
                    f.write(f"{item['url']}\n")
            print(f"💾 Saved {len(self.discovered_urls)} URLs to {output_file}")

def main():
    parser = argparse.ArgumentParser(
        description="Discover archived ChatGPT shares from web archive databases",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Discover from CDX (Wayback Machine)
  python cdx_discovery.py --source cdx --limit 10000 --output cdx_shares.txt

  # Discover from Common Crawl
  python cdx_discovery.py --source commoncrawl --index CC-MAIN-2024-10 --output cc_shares.json --format json

  # Date-filtered CDX query
  python cdx_discovery.py --source cdx --from-date 20240101 --to-date 20241231 --output 2024_shares.txt
        """
    )

    parser.add_argument(
        '--source',
        choices=['cdx', 'commoncrawl'],
        required=True,
        help='Archive source to query'
    )

    parser.add_argument(
        '--limit',
        type=int,
        default=10000,
        help='Maximum number of results (default: 10000)'
    )

    parser.add_argument(
        '--output',
        default='discovered_shares.txt',
        help='Output file path (default: discovered_shares.txt)'
    )

    parser.add_argument(
        '--format',
        choices=['txt', 'json', 'urls'],
        default='txt',
        help='Output format: txt (GUIDs only), json (full details), urls (full URLs)'
    )

    # CDX-specific options
    parser.add_argument(
        '--from-date',
        help='Start date for CDX query (YYYYMMDD format)'
    )

    parser.add_argument(
        '--to-date',
        help='End date for CDX query (YYYYMMDD format)'
    )

    # Common Crawl-specific options
    parser.add_argument(
        '--index',
        default='CC-MAIN-2024-10',
        help='Common Crawl index name (default: CC-MAIN-2024-10)'
    )

    args = parser.parse_args()

    # Run discovery
    discovery = CDXDiscovery()

    if args.source == 'cdx':
        discovery.discover_from_cdx(
            limit=args.limit,
            from_date=args.from_date,
            to_date=args.to_date
        )
    elif args.source == 'commoncrawl':
        discovery.discover_from_commoncrawl(
            index=args.index,
            limit=args.limit
        )

    # Save results
    if discovery.discovered_guids:
        discovery.stats["end_time"] = datetime.now().isoformat()
        discovery.save_results(args.output, format=args.format)
    else:
        print("❌ No GUIDs discovered")
        sys.exit(1)

if __name__ == '__main__':
    main()