""" ExportGuard — BigQuery One-Click Setup. Run this AFTER setting up a Google Cloud project and service account. Usage: set GOOGLE_APPLICATION_CREDENTIALS=C:\path\to\key.json set GOOGLE_CLOUD_PROJECT=your-project-id python backend/setup_bigquery.py --dataset=exportguard This script: 1. Creates the BigQuery dataset (if not exists) 2. Creates raw tables (shipments, country_risk, buyer_history) 3. Loads CSV data from data/raw/ into the tables 4. Creates the 3 analytics views (buyer_risk_summary, country_risk_trend, shipment_outcomes_by_hs_code) """ import argparse import os import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent DATA_RAW = PROJECT_ROOT / "data" / "raw" sys.path.insert(0, str(PROJECT_ROOT)) # ── Check credentials ─────────────────────────────────────────────────── def check_credentials(): cred_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "") project = os.environ.get("GOOGLE_CLOUD_PROJECT", "") if not cred_path or not Path(cred_path).exists(): print("=" * 60) print("ERROR: GOOGLE_APPLICATION_CREDENTIALS not set or file not found.") print("") print("Steps to fix:") print(" 1. Go to https://console.cloud.google.com/") print(" 2. Create a new project (or select existing)") print(" 3. Go to APIs & Services > Library") print(" 4. Enable 'BigQuery API' and 'Cloud Storage API'") print(" 5. Go to IAM & Admin > Service Accounts") print(" 6. Create a new service account → click 'Create and Continue'") print(" 7. Assign roles:") print(" - BigQuery Data Editor") print(" - BigQuery Job User") print(" - Storage Object Admin") print(" 8. Click 'Done', then click on the new service account") print(" 9. Go to 'Keys' tab → 'Add Key' → 'Create New Key' → JSON") print(" 10. Download the JSON file to your computer") print("") print("Then run:") print(f' set GOOGLE_APPLICATION_CREDENTIALS=C:\\path\\to\\downloaded-key.json') print(f' set GOOGLE_CLOUD_PROJECT=your-project-id') print(f" python backend/setup_bigquery.py --dataset=exportguard") print("=" * 60) return False if not project: print("ERROR: GOOGLE_CLOUD_PROJECT environment variable not set.") print(f'Example: set GOOGLE_CLOUD_PROJECT=my-project-12345') return False return True # ── Main setup ────────────────────────────────────────────────────────── def setup_bigquery(dataset_id: str): from google.cloud import bigquery client = bigquery.Client() project = client.project print(f"Using project: {project}") print(f"Dataset: {dataset_id}") # 1. Create dataset dataset_ref = bigquery.DatasetReference(project, dataset_id) try: client.get_dataset(dataset_ref) print(f"[OK] Dataset {dataset_id} already exists") except Exception: dataset = bigquery.Dataset(dataset_ref) dataset.location = "US" client.create_dataset(dataset) print(f"[OK] Created dataset {dataset_id}") # 2. Table schemas tables = { "shipments": [ bigquery.SchemaField("shipment_id", "STRING"), bigquery.SchemaField("exporter_id", "STRING"), bigquery.SchemaField("buyer_id", "STRING"), bigquery.SchemaField("buyer_country", "STRING"), bigquery.SchemaField("hs_code", "INT64"), bigquery.SchemaField("product_category", "STRING"), bigquery.SchemaField("shipment_date", "DATE"), bigquery.SchemaField("invoice_value_usd", "FLOAT64"), bigquery.SchemaField("payment_terms", "STRING"), bigquery.SchemaField("payment_delay_days", "INT64"), bigquery.SchemaField("was_disputed", "BOOL"), bigquery.SchemaField("was_paid_in_full", "BOOL"), ], "country_risk": [ bigquery.SchemaField("country", "STRING"), bigquery.SchemaField("month", "DATE"), bigquery.SchemaField("political_stability_score", "FLOAT64"), bigquery.SchemaField("currency_volatility_index", "FLOAT64"), bigquery.SchemaField("trade_sanctions_flag", "INT64"), ], "buyer_history": [ bigquery.SchemaField("buyer_id", "STRING"), bigquery.SchemaField("total_orders", "INT64"), bigquery.SchemaField("total_value_usd", "FLOAT64"), bigquery.SchemaField("avg_payment_delay_days", "FLOAT64"), bigquery.SchemaField("dispute_rate", "FLOAT64"), bigquery.SchemaField("paid_in_full_rate", "FLOAT64"), bigquery.SchemaField("avg_invoice_value", "FLOAT64"), bigquery.SchemaField("first_order_date", "DATE"), bigquery.SchemaField("last_order_date", "DATE"), bigquery.SchemaField("primary_country", "STRING"), bigquery.SchemaField("primary_category", "STRING"), bigquery.SchemaField("value_trend", "STRING"), ], } # 3. Create tables + load CSV data for table_name, schema in tables.items(): csv_path = DATA_RAW / f"{table_name}.csv" if not csv_path.exists(): print(f"[SKIP] {csv_path} not found — skipping {table_name}") continue table_id = f"{project}.{dataset_id}.{table_name}" # Delete existing table if present try: client.delete_table(table_id) except Exception: pass table = bigquery.Table(table_id, schema=schema) client.create_table(table) print(f"[OK] Created table {table_id}") # Load CSV job_config = bigquery.LoadJobConfig( source_format=bigquery.SourceFormat.CSV, skip_leading_rows=1, schema=schema, write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE, autodetect=False, ) with open(csv_path, "rb") as f: load_job = client.load_table_from_file(f, table_id, job_config=job_config) load_job.result() table = client.get_table(table_id) print(f"[OK] Loaded {table.num_rows:,} rows into {table_name}") # 4. Create views views_sql = { "buyer_risk_summary": f""" CREATE OR REPLACE VIEW `{project}.{dataset_id}.buyer_risk_summary` AS WITH buyer_stats AS ( SELECT buyer_id, buyer_country, COUNT(*) AS total_shipments, SUM(invoice_value_usd) AS total_value, AVG(payment_delay_days) AS avg_delay_days, SAFE_DIVIDE(SUM(CAST(was_disputed AS INT64)), COUNT(*)) AS dispute_rate, SAFE_DIVIDE(SUM(CAST(was_paid_in_full AS INT64)), COUNT(*)) AS paid_in_full_rate, MAX(shipment_date) AS last_shipment, DATE_DIFF(CURRENT_DATE(), MAX(shipment_date), DAY) AS days_since_last_order, ROUND( (AVG(payment_delay_days) / 180) * 30 + SAFE_DIVIDE(SUM(CAST(was_disputed AS INT64)), COUNT(*)) * 40 + (1 - SAFE_DIVIDE(SUM(CAST(was_paid_in_full AS INT64)), COUNT(*))) * 30 , 1) AS risk_score FROM `{project}.{dataset_id}.shipments` GROUP BY buyer_id, buyer_country ) SELECT *, CASE WHEN risk_score < 30 THEN 'Low Risk' WHEN risk_score < 60 THEN 'Medium Risk' ELSE 'High Risk' END AS risk_category FROM buyer_stats """, "country_risk_trend": f""" CREATE OR REPLACE VIEW `{project}.{dataset_id}.country_risk_trend` AS SELECT country, month, political_stability_score, currency_volatility_index, trade_sanctions_flag, ROUND( (1 - political_stability_score) * 50 + currency_volatility_index * 40 + trade_sanctions_flag * 10 , 1) AS composite_risk_index FROM `{project}.{dataset_id}.country_risk` ORDER BY country, month """, "shipment_outcomes_by_hs_code": f""" CREATE OR REPLACE VIEW `{project}.{dataset_id}.shipment_outcomes_by_hs_code` AS SELECT hs_code, product_category, COUNT(*) AS total_shipments, SUM(invoice_value_usd) AS total_value_usd, ROUND(AVG(invoice_value_usd), 0) AS avg_invoice_value, ROUND(AVG(payment_delay_days), 1) AS avg_delay_days, SAFE_DIVIDE(SUM(CAST(was_disputed AS INT64)), COUNT(*)) AS dispute_rate, SAFE_DIVIDE(SUM(CAST(was_paid_in_full AS INT64)), COUNT(*)) AS paid_in_full_rate, COUNTIF(payment_terms = 'advance') AS advance_count, COUNTIF(payment_terms = 'lc') AS lc_count, COUNTIF(payment_terms LIKE 'credit_%') AS credit_count FROM `{project}.{dataset_id}.shipments` GROUP BY hs_code, product_category ORDER BY total_value_usd DESC """, } for view_name, sql in views_sql.items(): try: job = client.query(sql) job.result() print(f"[OK] Created view {view_name}") except Exception as e: print(f"[ERROR] Creating view {view_name}: {e}") print(f"\n{'=' * 60}") print(f"SETUP COMPLETE!") print(f"{'=' * 60}") print(f"") print(f"BigQuery tables and views are ready at: {project}.{dataset_id}") print(f"") print(f"Views created:") for v in views_sql: print(f" - {v}") print(f"") print(f"Next: Open Looker Studio → https://lookerstudio.google.com/") print(f" 1. Click 'Create' → 'Report'") print(f" 2. Click 'Add Data' → 'BigQuery'") print(f" 3. Select your project → {dataset_id} → choose a view") print(f" 4. For each view, create the corresponding chart:") print(f"") print(f" CHART 1: Buyer Risk Leaderboard") print(f" View: buyer_risk_summary") print(f" Chart: Bar chart") print(f" Dimension: buyer_id") print(f" Metric: risk_score (sorted DESC, limit 20)") print(f"") print(f" CHART 2: Country Risk Over Time") print(f" View: country_risk_trend") print(f" Chart: Time series (line chart)") print(f" Dimension: month") print(f" Metric: composite_risk_index (AVG)") print(f" Breakdown: country") print(f"") print(f" CHART 3: HS-code Exposure Breakdown") print(f" View: shipment_outcomes_by_hs_code") print(f" Chart: Stacked bar chart") print(f" Dimension: product_category") print(f" Metrics: total_value_usd") def main(): parser = argparse.ArgumentParser(description="ExportGuard BigQuery Setup") parser.add_argument("--dataset", default="exportguard", help="BigQuery dataset name") args = parser.parse_args() if not check_credentials(): sys.exit(1) setup_bigquery(args.dataset) if __name__ == "__main__": main()