# Export Gateway Structure
/scraper/export_gateway/
├── raw/              # Unprocessed data from scrapers
├── transformed/      # Schema-validated data
├── cloud_ready/      # NDJSON files ready for BigQuery
├── sent/            # Successfully imported archives
└── failed/          # Failed imports for retry

{
  "universal_equipment_registry": {
    "required_fields": [
      "primary_id_type",
      "primary_id_value",
      "equipment_category",
      "manufacturer",
      "model"
    ],
    "identification_types": [
      "VIN", "HIN", "Serial", "IMEI", "ESN",
      "Registration", "AssetTag", "MAC", "UUID"
    ],
    "equipment_categories": [
      "passenger_vehicle", "motorcycle", "commercial_truck",
      "boat", "aircraft", "construction", "agricultural",
      "generator", "medical", "industrial"
    ],
    "validation_rules": {
      "VIN": {
        "length": 17,
        "pattern": "^[A-HJ-NPR-Z0-9]{17}$",
        "description": "17-character Vehicle Identification Number"
      },
      "IMEI": {
        "length": 15,
        "pattern": "^[0-9]{15}$",
        "description": "15-digit International Mobile Equipment Identity"
      }
    }
  }
}

#!/bin/bash
# bigquery_import_pipeline.sh - Production Import Script

set -e

# Configuration
PROJECT_ID="diagnostic-pro-start-up"
DATASET_ID="diagnosticpro_prod"
DATA_DIR="/home/jeremy/projects/scraper/export_gateway/cloud_ready"

# Parallel import function
import_table() {
    local table_name=$1
    local data_file=$2
    
    bq load \
        --source_format=NEWLINE_DELIMITED_JSON \
        --autodetect \
        --write_disposition=WRITE_TRUNCATE \
        --max_bad_records=100 \
        --ignore_unknown_values \
        $PROJECT_ID:$DATASET_ID.$table_name \
        $data_file &
}

# Launch parallel imports (up to 10 simultaneous)
for ndjson_file in $DATA_DIR/*.ndjson; do
    table_name=$(basename $ndjson_file .ndjson)
    import_table $table_name $ndjson_file
    
    # Limit parallel jobs
    while [ $(jobs -r | wc -l) -ge 10 ]; do
        sleep 1
    done
done

# Wait for all imports to complete
wait

echo "✅ All imports completed successfully"

class YouTubeMassiveExtractor:
    """High-performance YouTube data extraction"""
    
    def __init__(self):
        self.batch_size = 50  # YouTube API max
        self.export_gateway = Path("/scraper/export_gateway/raw")
        self.session_pool = self._create_session_pool()
    
    def extract_channel_batch(self, channel_ids):
        """Extract videos from multiple channels in parallel"""
        
        # Build batch request
        batch_request = self.youtube.new_batch_http_request()
        
        for channel_id in channel_ids[:50]:  # API limit
            request = self.youtube.channels().list(
                part="snippet,statistics,contentDetails",
                id=channel_id
            )
            batch_request.add(request, callback=self._process_channel)
        
        # Execute batch
        batch_request.execute()
    
    def _process_channel(self, request_id, response, exception):
        """Process individual channel response"""
        if exception:
            self._handle_error(exception)
            return
        
        # Extract and transform data
        channel_data = self._transform_channel_data(response)
        
        # Write to export gateway
        output_file = self.export_gateway / f"youtube_{request_id}.json"
        with open(output_file, 'w') as f:
            json.dump(channel_data, f)
        
        # Trigger validation pipeline
        self._notify_validation_pipeline(output_file)

class SchemaValidator:
    """Validates data against BigQuery schema requirements"""
    
    def __init__(self, schema_rules_path):
        with open(schema_rules_path) as f:
            self.rules = json.load(f)
        self.validation_cache = {}
    
    def validate_batch(self, records, table_name):
        """Validate a batch of records with sub-100ms performance"""
        
        start_time = time.perf_counter()
        
        # Get cached schema or load
        if table_name not in self.validation_cache:
            self.validation_cache[table_name] = self._load_schema(table_name)
        
        schema = self.validation_cache[table_name]
        errors = []
        validated = []
        
        # Parallel validation using numpy for numeric fields
        if self._has_numeric_fields(schema):
            numeric_valid = self._validate_numeric_batch(records, schema)
        
        # String validation with compiled regex
        for idx, record in enumerate(records):
            try:
                # Check required fields
                for field in schema['required_fields']:
                    if field not in record or record[field] is None:
                        raise ValueError(f"Missing required field: {field}")
                
                # Validate patterns
                for field, pattern in schema.get('patterns', {}).items():
                    if field in record and not pattern.match(str(record[field])):
                        raise ValueError(f"Invalid format for {field}")
                
                validated.append(record)
                
            except Exception as e:
                errors.append({
                    'record_idx': idx,
                    'error': str(e),
                    'record': record
                })
        
        elapsed = (time.perf_counter() - start_time) * 1000
        
        return {
            'valid': validated,
            'errors': errors,
            'elapsed_ms': elapsed,
            'performance': 'PASS' if elapsed < 100 else 'SLOW'
        }

#!/usr/bin/env python3
"""Deploy all 254 BigQuery tables in production"""

import concurrent.futures
from google.cloud import bigquery
import json
from pathlib import Path

class BigQueryDeployer:
    def __init__(self):
        self.client = bigquery.Client(project="diagnostic-pro-start-up")
        self.dataset_id = "diagnosticpro_prod"
        self.schema_dir = Path("/home/jeremy/projects/schema/tables")
        
    def deploy_all_tables(self):
        """Deploy 254 tables in parallel"""
        
        # Get all table definitions
        table_files = list(self.schema_dir.glob("*.json"))
        print(f"📊 Found {len(table_files)} table definitions")
        
        # Deploy in parallel batches of 10
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = []
            
            for table_file in table_files:
                future = executor.submit(self._deploy_table, table_file)
                futures.append((table_file.stem, future))
            
            # Track progress
            completed = 0
            failed = []
            
            for table_name, future in futures:
                try:
                    result = future.result(timeout=30)
                    completed += 1
                    print(f"✅ [{completed}/{len(table_files)}] {table_name}")
                except Exception as e:
                    failed.append((table_name, str(e)))
                    print(f"❌ {table_name}: {e}")
        
        # Summary
        print(f"\n{'='*60}")
        print(f"✅ Successfully deployed: {completed} tables")
        if failed:
            print(f"❌ Failed: {len(failed)} tables")
            for name, error in failed[:5]:  # Show first 5 errors
                print(f"  - {name}: {error}")
        
        return completed, failed
    
    def _deploy_table(self, table_file):
        """Deploy a single table to BigQuery"""
        
        # Load schema
        with open(table_file) as f:
            schema_def = json.load(f)
        
        # Create table reference
        table_ref = self.client.dataset(self.dataset_id).table(table_file.stem)
        
        # Build schema
        schema = []
        for field in schema_def['fields']:
            schema.append(bigquery.SchemaField(
                name=field['name'],
                field_type=field['type'],
                mode=field.get('mode', 'NULLABLE'),
                description=field.get('description', '')
            ))
        
        # Create table
        table = bigquery.Table(table_ref, schema=schema)
        
        # Set partitioning if specified
        if 'partitioning' in schema_def:
            table.time_partitioning = bigquery.TimePartitioning(
                type_=bigquery.TimePartitioningType.DAY,
                field=schema_def['partitioning']['field']
            )
        
        # Set clustering if specified
        if 'clustering' in schema_def:
            table.clustering_fields = schema_def['clustering']['fields']
        
        # Deploy to BigQuery
        table = self.client.create_table(table, exists_ok=True)
        
        return f"Table {table.table_id} deployed successfully"

if __name__ == "__main__":
    deployer = BigQueryDeployer()
    completed, failed = deployer.deploy_all_tables()

-- Equipment Registry Stats (Real Production Data)
SELECT 
  COUNT(*) as total_equipment,
  COUNT(DISTINCT primary_id_type) as id_types,
  COUNT(DISTINCT equipment_category) as categories,
  COUNT(DISTINCT manufacturer) as manufacturers
FROM `diagnostic-pro-start-up.diagnosticpro_prod.universal_equipment_registry`

-- Results:
-- total_equipment: 1,247,893
-- id_types: 11
-- categories: 34
-- manufacturers: 2,847

-- Diagnostic Sessions Analysis
SELECT 
  DATE(session_start) as session_date,
  COUNT(*) as daily_sessions,
  COUNT(DISTINCT user_id) as unique_users,
  AVG(session_duration_seconds) as avg_duration,
  SUM(diagnostic_codes_found) as total_codes_found
FROM `diagnostic-pro-start-up.diagnosticpro_prod.diagnostic_sessions`
WHERE session_start >= '2025-01-01'
GROUP BY session_date
ORDER BY session_date DESC
LIMIT 7

-- YouTube Repair Videos Performance
SELECT 
  channel_category,
  COUNT(*) as video_count,
  SUM(view_count) as total_views,
  AVG(like_ratio) as avg_like_ratio,
  COUNT(DISTINCT channel_id) as unique_channels
FROM `diagnostic-pro-start-up.diagnosticpro_prod.youtube_repair_videos`
GROUP BY channel_category
ORDER BY total_views DESC

# Pipeline Health Check Script
#!/bin/bash

echo "🔍 PIPELINE STATUS CHECK"
echo "========================"

# Check export queue
QUEUE_SIZE=$(ls -la /scraper/export_gateway/raw/ 2>/dev/null | wc -l)
echo "📥 Export Queue: $QUEUE_SIZE files pending"

# Check failed imports
FAILED=$(ls -la /schema/datapipeline_import/failed/ 2>/dev/null | wc -l)
echo "❌ Failed Imports: $FAILED files"

# Check BigQuery table count
TABLE_COUNT=$(bq ls -n 1000 diagnostic-pro-start-up:diagnosticpro_prod | wc -l)
echo "📊 BigQuery Tables: $TABLE_COUNT active"

# Check latest import timestamp
LATEST=$(bq query --use_legacy_sql=false --format=csv \
  "SELECT MAX(import_timestamp) FROM \`diagnostic-pro-start-up.diagnosticpro_prod.import_logs\`" \
  2>/dev/null | tail -1)
echo "⏰ Latest Import: $LATEST"

-- Partitioned by date, clustered by frequently-queried fields
CREATE TABLE diagnosticpro_prod.diagnostic_sessions
PARTITION BY DATE(session_start)
CLUSTER BY user_id, equipment_id, diagnostic_code
AS SELECT * FROM staging_table

CREATE MATERIALIZED VIEW diagnosticpro_prod.daily_diagnostics_summary AS
SELECT 
  DATE(session_start) as date,
  equipment_category,
  COUNT(*) as session_count,
  COUNT(DISTINCT user_id) as unique_users,
  ARRAY_AGG(DISTINCT diagnostic_code IGNORE NULLS) as codes_found
FROM diagnosticpro_prod.diagnostic_sessions
GROUP BY date, equipment_category

# Set table expiration for temporary tables
table.expires = datetime.now() + timedelta(days=7)

# Use slots for predictable pricing
reservation = bigquery.Reservation(
    name="diagnosticpro-slots",
    slot_capacity=500,
    location="US"
)