exportguard-backend / backend /setup_bigquery.py
ExportGuard Deploy
Deploy ExportGuard
1c0bd9a
Raw
History Blame Contribute Delete
11.5 kB
"""
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()