verify_shares.py

#!/usr/bin/env python3
"""
ChatGPT Share Verification Tool

Verifies accessibility of discovered ChatGPT share URLs.
Checks HTTP status and content to categorize shares as:
- accessible: Valid share with content
- deleted: Returns 200 but shows "Can't load" message
- not_found: Returns 404
- error: Other errors (timeouts, network issues, etc.)
"""

import sys
import io
import argparse
import json
import requests
from typing import Dict, List
from datetime import datetime
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

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

class ShareVerifier:
    """Verifies accessibility of ChatGPT shares"""

    def __init__(self, concurrent: int = 5, delay: float = 0.5):
        self.concurrent = concurrent
        self.delay = delay
        self.results = {
            "accessible": [],
            "deleted": [],
            "not_found": [],
            "error": []
        }
        self.stats = {
            "total_checked": 0,
            "accessible": 0,
            "deleted": 0,
            "not_found": 0,
            "error": 0,
            "start_time": datetime.now().isoformat(),
            "end_time": None
        }

    def verify_single_url(self, url: str) -> Dict:
        """Verify a single URL"""
        result = {
            "url": url,
            "guid": self._extract_guid(url),
            "status": None,
            "category": None,
            "checked_at": datetime.now().isoformat()
        }

        try:
            # Use GET to check content, not just status
            response = requests.get(
                url,
                allow_redirects=False,
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0 (Research Bot)'}
            )

            result["status"] = response.status_code

            if response.status_code == 200:
                # Check if content shows deletion message
                is_deleted = (
                    "Can't load shared conversation" in response.text or
                    "Can't load shared conversation" in response.text
                )

                if is_deleted:
                    result["category"] = "deleted"
                    self.results["deleted"].append(result)
                    self.stats["deleted"] += 1
                else:
                    result["category"] = "accessible"
                    self.results["accessible"].append(result)
                    self.stats["accessible"] += 1

            elif response.status_code == 404:
                result["category"] = "not_found"
                self.results["not_found"].append(result)
                self.stats["not_found"] += 1

            else:
                result["category"] = "error"
                result["error"] = f"Unexpected status code: {response.status_code}"
                self.results["error"].append(result)
                self.stats["error"] += 1

        except requests.Timeout:
            result["category"] = "error"
            result["error"] = "Timeout"
            self.results["error"].append(result)
            self.stats["error"] += 1

        except requests.RequestException as e:
            result["category"] = "error"
            result["error"] = str(e)
            self.results["error"].append(result)
            self.stats["error"] += 1

        self.stats["total_checked"] += 1

        # Rate limiting delay
        time.sleep(self.delay)

        return result

    def _extract_guid(self, url: str) -> str:
        """Extract GUID from ChatGPT share URL"""
        if '/share/' in url:
            guid = url.split('/share/')[-1].split('?')[0].split('/')[0]
            if len(guid) == 36 and guid.count('-') == 4:
                return guid
        return None

    def verify_urls(self, urls: List[str]) -> None:
        """Verify multiple URLs concurrently"""
        total = len(urls)
        print(f"🔍 Verifying {total} URLs with {self.concurrent} concurrent connections...")

        with ThreadPoolExecutor(max_workers=self.concurrent) as executor:
            future_to_url = {executor.submit(self.verify_single_url, url): url for url in urls}

            for i, future in enumerate(as_completed(future_to_url), 1):
                result = future.result()

                # Progress output
                if i % 10 == 0 or i == total:
                    accessible = self.stats["accessible"]
                    deleted = self.stats["deleted"]
                    not_found = self.stats["not_found"]
                    errors = self.stats["error"]

                    print(f"Progress: {i}/{total} | "
                          f"✅ {accessible} accessible | "
                          f"🗑️  {deleted} deleted | "
                          f"❌ {not_found} not found | "
                          f"⚠️  {errors} errors", end='\r')

        print()  # New line after progress
        print(f"\n📊 Verification Complete:")
        print(f"  ✅ Accessible: {self.stats['accessible']}")
        print(f"  🗑️  Deleted: {self.stats['deleted']}")
        print(f"  ❌ Not Found: {self.stats['not_found']}")
        print(f"  ⚠️  Errors: {self.stats['error']}")

    def save_results(self, output_file: str) -> None:
        """Save verification results to JSON"""
        self.stats["end_time"] = datetime.now().isoformat()

        output_data = {
            "stats": self.stats,
            "results": self.results
        }

        output_path = Path(output_file)
        with output_path.open('w', encoding='utf-8') as f:
            json.dump(output_data, f, indent=2)

        print(f"💾 Results saved to {output_file}")

    def export_accessible_guids(self, output_file: str) -> None:
        """Export only accessible GUIDs to text file"""
        accessible_guids = [r["guid"] for r in self.results["accessible"] if r["guid"]]

        if accessible_guids:
            output_path = Path(output_file)
            with output_path.open('w', encoding='utf-8') as f:
                for guid in sorted(accessible_guids):
                    f.write(f"{guid}\n")

            print(f"💾 Exported {len(accessible_guids)} accessible GUIDs to {output_file}")
        else:
            print("⚠️  No accessible shares found to export")

def load_urls_from_file(file_path: str) -> List[str]:
    """Load URLs from text file"""
    urls = []
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if line and line.startswith('http'):
                urls.append(line)
            elif line and '-' in line and len(line) == 36:
                # GUID format - convert to URL
                urls.append(f"https://chatgpt.com/share/{line}")

    return urls

def main():
    parser = argparse.ArgumentParser(
        description="Verify accessibility of ChatGPT share URLs",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Verify URLs from file
  python verify_shares.py --input cdx_shares.txt --output results.json

  # Verify with custom concurrency and delay
  python verify_shares.py --input shares.txt --output results.json --concurrent 10 --delay 0.3

  # Export only accessible GUIDs
  python verify_shares.py --input shares.txt --output results.json --export-accessible accessible.txt
        """
    )

    parser.add_argument(
        '--input',
        required=True,
        help='Input file containing URLs or GUIDs (one per line)'
    )

    parser.add_argument(
        '--output',
        default='verification_results.json',
        help='Output JSON file for results (default: verification_results.json)'
    )

    parser.add_argument(
        '--export-accessible',
        help='Export accessible GUIDs to separate text file'
    )

    parser.add_argument(
        '--concurrent',
        type=int,
        default=5,
        help='Number of concurrent connections (default: 5)'
    )

    parser.add_argument(
        '--delay',
        type=float,
        default=0.5,
        help='Delay between requests in seconds (default: 0.5)'
    )

    args = parser.parse_args()

    # Load URLs
    print(f"📂 Loading URLs from {args.input}...")
    urls = load_urls_from_file(args.input)
    print(f"📋 Loaded {len(urls)} URLs")

    if not urls:
        print("❌ No URLs found in input file")
        sys.exit(1)

    # Verify
    verifier = ShareVerifier(concurrent=args.concurrent, delay=args.delay)
    verifier.verify_urls(urls)

    # Save results
    verifier.save_results(args.output)

    # Export accessible GUIDs if requested
    if args.export_accessible:
        verifier.export_accessible_guids(args.export_accessible)

if __name__ == '__main__':
    main()