import csv import glob import json import os import subprocess import sys import time from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) ROOT_DIR = os.path.dirname(SCRIPTS_DIR) CONFIG = { 'create_csv': os.path.join(ROOT_DIR, 'CREATE.csv'), 'export_folder': os.path.join(ROOT_DIR, 'export'), 'report_folder': os.path.join(ROOT_DIR, 'report'), 'location': 'us-central1', 'concurrency': 3, } def resolve_gcloud(): candidates = [ os.path.join(os.environ.get("LOCALAPPDATA", ""), "Google", "Cloud SDK", "google-cloud-sdk", "bin", "gcloud.cmd"), os.path.join(os.environ.get("ProgramFiles", ""), "Google", "Cloud SDK", "google-cloud-sdk", "bin", "gcloud.cmd"), os.path.join(os.environ.get("ProgramFiles(x86)", ""), "Google", "Cloud SDK", "google-cloud-sdk", "bin", "gcloud.cmd"), ] for p in candidates: if os.path.isfile(p): return p return "gcloud" def run_cmd(cmd, timeout=300): try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout ) return result.returncode == 0, result.stdout.strip(), result.stderr.strip() except subprocess.TimeoutExpired: return False, "", "Command timed out" except FileNotFoundError: return False, "", "gcloud CLI not found" except Exception as e: return False, "", str(e) def read_create_csv(filepath): mappings = [] with open(filepath, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: mappings.append(row) return mappings def count_html_files(folder_path): return len(glob.glob(os.path.join(folder_path, '*.html'))) def create_bucket(gcloud_cmd, bucket_name, location): print(f" [CREATE] gs://{bucket_name}") cmd = [ gcloud_cmd, "storage", "buckets", "create", f"gs://{bucket_name}", f"--location={location}", "--uniform-bucket-level-access" ] ok, out, err = run_cmd(cmd) if not ok: if "AlreadyExists" in err or "409" in err: print(f" [EXISTS] gs://{bucket_name}") return True print(f" [FAIL] Create: {err}") return False return True def set_public_access(gcloud_cmd, bucket_name): print(f" [PUBLIC] gs://{bucket_name}") cmd = [ gcloud_cmd, "storage", "buckets", "add-iam-policy-binding", f"gs://{bucket_name}", "--member=allUsers", "--role=roles/storage.objectViewer" ] ok, out, err = run_cmd(cmd) if not ok: print(f" [FAIL] Public access: {err}") return False return True def upload_content(gcloud_cmd, bucket_name, foldering, export_folder): folder_path = os.path.join(export_folder, foldering) print(f" [UPLOAD] {foldering} -> gs://{bucket_name}/") cmd = [ gcloud_cmd, "storage", "cp", "-r", f"{folder_path}/*", f"gs://{bucket_name}/" ] ok, out, err = run_cmd(cmd, timeout=600) if not ok: print(f" [FAIL] Upload: {err}") return False return True def process_mapping(gcloud_cmd, mapping, location, export_folder): start = time.time() bucket = mapping['bucket'] foldering = mapping['foldering'] name = mapping['name'] source = mapping.get('source', '') result = { 'bucket': bucket, 'foldering': foldering, 'name': name, 'source': source, 'files': 0, 'success': False, 'bucket_created': False, 'public_set': False, 'uploaded': False, 'error': '', 'duration': 0, } folder_path = os.path.join(export_folder, foldering) if not os.path.isdir(folder_path): result['error'] = f'Folder not found: {foldering}' result['duration'] = f"{time.time() - start:.1f}" print(f" [SKIP] {foldering} - folder not found, skipping bucket creation") return result result['files'] = count_html_files(folder_path) try: if not create_bucket(gcloud_cmd, bucket, location): result['error'] = 'Bucket creation failed' result['duration'] = f"{time.time() - start:.1f}" return result result['bucket_created'] = True if not set_public_access(gcloud_cmd, bucket): result['error'] = 'Public access failed' result['duration'] = f"{time.time() - start:.1f}" return result result['public_set'] = True if not upload_content(gcloud_cmd, bucket, foldering, export_folder): result['error'] = 'Upload failed' result['duration'] = f"{time.time() - start:.1f}" return result result['uploaded'] = True result['success'] = True except Exception as e: result['error'] = str(e) result['duration'] = f"{time.time() - start:.1f}" status = "SUCCESS" if result['success'] else "FAILED" print(f" [{status}] {foldering} -> gs://{bucket}/ ({result['duration']}s)") return result def generate_summary_json(results, report_folder): if not os.path.exists(report_folder): os.makedirs(report_folder) summary = [] for r in results: if 'Folder not found' in r['error']: status = 'SKIP' elif r['success']: status = 'SUCCESS' else: status = 'FAILED' summary.append({ 'bucket': r['bucket'], 'foldering': r['foldering'], 'name': r['name'], 'source': r.get('source', ''), 'files': r['files'], 'url': f"https://storage.googleapis.com/{r['bucket']}/" if r['success'] else '', 'create': 'YES' if r['bucket_created'] else 'NO', 'public': 'YES' if r['public_set'] else 'NO', 'upload': 'YES' if r['uploaded'] else 'NO', 'status': status, 'error': r['error'], 'duration': r['duration'], }) json_file = os.path.join(report_folder, 'deploy-summary.json') with open(json_file, 'w', encoding='utf-8') as f: json.dump(summary, f, indent=2) return json_file def main(): import argparse parser = argparse.ArgumentParser(description='GCS Bulk Deployment') parser.add_argument('--create-csv', default=CONFIG['create_csv'], help='Path to CREATE.csv') parser.add_argument('--location', default=CONFIG['location'], help='Bucket location') parser.add_argument('--concurrency', type=int, default=CONFIG['concurrency'], help='Max parallel operations') parser.add_argument('--export-folder', default=CONFIG['export_folder'], help='Export folder path') args = parser.parse_args() gcloud_cmd = resolve_gcloud() print("=" * 50) print(" GCS Bulk Deployment (Python)") print("=" * 50) print() if not os.path.exists(args.create_csv): print(f"Error: {args.create_csv} not found.") sys.exit(1) if not os.path.isdir(args.export_folder): print(f"Error: Export folder '{args.export_folder}' not found.") sys.exit(1) mappings = read_create_csv(args.create_csv) print(f"Found {len(mappings)} bucket mappings in {args.create_csv}") print(f"Concurrency: {args.concurrency}") print(f"Location: {args.location}") print() results = [] with ThreadPoolExecutor(max_workers=args.concurrency) as executor: futures = { executor.submit( process_mapping, gcloud_cmd, m, args.location, args.export_folder ): m for m in mappings } for future in as_completed(futures): result = future.result() results.append(result) results.sort(key=lambda x: x['foldering']) success_count = sum(1 for r in results if r['success']) skip_count = sum(1 for r in results if 'Folder not found' in r['error']) fail_count = len(results) - success_count - skip_count print() print("=" * 50) print(" DEPLOYMENT SUMMARY") print("=" * 50) print(f"Total: {len(results)}") print(f"Success: {success_count}") print(f"Skipped: {skip_count}") print(f"Failed: {fail_count}") print("=" * 50) generate_summary_json(results, CONFIG['report_folder']) if __name__ == '__main__': main()