from __future__ import annotations import argparse import json import sys from .bq_load import run_load from .config import load_config from .extract import ExtractRunError, run_extract from .netsuite_revenue_admission import AdmissionError from .netsuite_revenue_config import load_revenue_sync_config from .netsuite_revenue_nexuscrm import NexusCRMRevenueContractError from .netsuite_revenue_sync import RevenueSyncGateError, run_revenue_sync from .outbound import OutboundRunError, run_outbound from .outbound_config import load_outbound_config from .salesforce_compat import ( render_salesforce_order_compatibility_sql, render_salesforce_order_cutover_sql, render_salesforce_order_materialize_sql, ) from .source_inspect import inspect_exact_tidb_tables, inspect_tables, table_selection_contract def main() -> None: parser = argparse.ArgumentParser(description="PingCAP GCP-native extract/load pipelines") subparsers = parser.add_subparsers(dest="command", required=True) inspect_parser = subparsers.add_parser("inspect", help="Inspect selected source tables") inspect_parser.add_argument("--source", required=True, help="Path to a source YAML") inspect_parser.add_argument("--tables", help="Comma-separated exact table names") inspect_parser.add_argument( "--compact", action="store_true", help="Print one-line JSON suitable for bounded runtime log receipts", ) extract_parser = subparsers.add_parser("extract", help="Extract a source run into GCS and write a manifest") extract_parser.add_argument("--source", required=True, help="Path to a source YAML") extract_parser.add_argument("--run-date") extract_parser.add_argument("--run-id") extract_parser.add_argument("--tables", help="Comma-separated table names") load_parser = subparsers.add_parser("load", help="Load manifests from GCS into BigQuery") load_parser.add_argument("--source", required=True, help="Path to a source YAML") load_parser.add_argument("--manifest", help="Manifest URI or local path") load_parser.add_argument("--latest", action="store_true", help="Load the newest unloaded manifest") load_parser.add_argument("--all-unloaded", action="store_true", help="Load every unloaded manifest in order") load_parser.add_argument("--tables", help="Comma-separated exact tables expected in the manifest") run_parser = subparsers.add_parser("run-pipeline", help="Extract then load the same run") run_parser.add_argument("--source", required=True, help="Path to a source YAML") run_parser.add_argument("--run-date") run_parser.add_argument("--run-id") run_parser.add_argument("--tables", help="Comma-separated table names") compat_parser = subparsers.add_parser( "salesforce-compat-sql", help="Print NexusCRM-backed Salesforce compatibility SQL", ) compat_parser.add_argument("--object", choices=["order"], required=True) compat_parser.add_argument("--project", default="gcp-digital-bi") materialize_parser = subparsers.add_parser( "salesforce-compat-materialize-sql", help="Print SQL that materializes a NexusCRM-backed Salesforce compatibility table", ) materialize_parser.add_argument("--object", choices=["order"], required=True) materialize_parser.add_argument("--project", default="gcp-digital-bi") materialize_parser.add_argument("--dataset", default="salesforce") cutover_parser = subparsers.add_parser( "salesforce-compat-cutover-sql", help="Print controlled backup/materialize/cutover SQL for a Salesforce compatibility table", ) cutover_parser.add_argument("--object", choices=["order"], required=True) cutover_parser.add_argument("--project", default="gcp-digital-bi") cutover_parser.add_argument("--dataset", default="salesforce") cutover_parser.add_argument("--backup-suffix", required=True) outbound_parser = subparsers.add_parser("outbound", help="Write a configured BigQuery dataset into NexusCRM") outbound_parser.add_argument("--job", required=True, help="Path to an outbound job YAML") outbound_mode = outbound_parser.add_mutually_exclusive_group() outbound_mode.add_argument( "--dry-run", action="store_true", help="Read and validate source rows without authenticating to or writing NexusCRM", ) outbound_mode.add_argument( "--validate-only", action="store_true", help="Validate source rows and the live NexusCRM object contract without writing records", ) outbound_parser.add_argument("--run-id") outbound_parser.add_argument("--limit", type=int, help="Limit source rows for a manual staging probe") revenue_parser = subparsers.add_parser( "netsuite-revenue-to-nexuscrm", help="Freeze, admit, and API-upsert NetSuite revenue candidates into NexusCRM", ) revenue_parser.add_argument( "--job", default="configs/jobs/netsuite_revenue_to_nexuscrm.yaml", help="Path to the NetSuite revenue NexusCRM API job YAML", ) revenue_parser.add_argument("--posting-period-id", required=True) revenue_parser.add_argument("--run-id") revenue_parser.add_argument( "--fixture-workbook", help="Local XLSX fixture; creates a candidate only and cannot commit", ) revenue_parser.add_argument( "--live-source", action="store_true", help="Fetch the TBA RESTlet (requires separate authorization)", ) revenue_parser.add_argument( "--replay-candidate-dir", help="Reuse a frozen local candidate; never fetches source pages", ) revenue_parser.add_argument( "--fixture-account-matches", help="Local JSON object of Account ID to exactly-one/zero/multiple CRM IDs for admission-only tests", ) revenue_parser.add_argument( "--validate-only", action="store_true", help="Run authorized read-only NexusCRM Describe/query/Account-admission checks without writes", ) revenue_parser.add_argument( "--commit", action="store_true", help="Enable NexusCRM Composite Records API writes after all named gates pass", ) revenue_parser.add_argument("--expected-project") revenue_parser.add_argument("--expected-environment") revenue_parser.add_argument("--expected-target-base-url") revenue_parser.add_argument("--expected-api-version") revenue_parser.add_argument("--expected-describe-sha256") revenue_parser.add_argument("--expected-manifest-sha256") revenue_parser.add_argument("--candidate-run-id") args = parser.parse_args() if args.command == "salesforce-compat-sql": if args.object == "order": print(render_salesforce_order_compatibility_sql(project=args.project), end="") return if args.command == "salesforce-compat-materialize-sql": if args.object == "order": print(render_salesforce_order_materialize_sql(project=args.project, dataset=args.dataset), end="") return if args.command == "salesforce-compat-cutover-sql": if args.object == "order": print( render_salesforce_order_cutover_sql( project=args.project, dataset=args.dataset, backup_suffix=args.backup_suffix, ), end="", ) return if args.command == "outbound": config = load_outbound_config(args.job) try: result = run_outbound( config, dry_run=args.dry_run, validate_only=args.validate_only, run_id=args.run_id, limit=args.limit, ) except OutboundRunError as error: print(json.dumps(error.summary, indent=2), file=sys.stderr) raise SystemExit(1) from error print(json.dumps(result, indent=2)) return if args.command == "netsuite-revenue-to-nexuscrm": config = load_revenue_sync_config(args.job) fixture_account_matches = None if args.fixture_account_matches: fixture_account_matches = json.loads(args.fixture_account_matches) if not isinstance(fixture_account_matches, dict): raise ValueError("--fixture-account-matches must be a JSON object") try: result = run_revenue_sync( config, run_id=args.run_id, posting_period_id=args.posting_period_id, commit=args.commit, validate_only=args.validate_only, expected_project=args.expected_project, expected_environment=args.expected_environment, expected_target_base_url=args.expected_target_base_url, expected_api_version=args.expected_api_version, expected_describe_sha256=args.expected_describe_sha256, expected_manifest_sha256=args.expected_manifest_sha256, candidate_run_id=args.candidate_run_id, fixture_workbook=args.fixture_workbook, live_source=args.live_source, replay_candidate_dir=args.replay_candidate_dir, fixture_account_matches=fixture_account_matches, ) except (RevenueSyncGateError, AdmissionError, NexusCRMRevenueContractError) as error: print(json.dumps({"status": "blocked", "reason": str(error)}), file=sys.stderr) raise SystemExit(1) from error print(json.dumps(result, indent=2, sort_keys=True)) return config = load_config(args.source) if args.command == "inspect": selected = parse_tables(args.tables) if config.source_type == "tidbcloud" and selected is not None: tables = inspect_exact_tidb_tables(config, selected) elif selected is not None: tables = inspect_tables(config, selected=selected) else: tables = inspect_tables(config) payload = { "source": config.name, "source_database": config.source_database, "table_count": len(tables), "tables": [ { "name": table.name, "rows_estimate": table.rows_estimate, "primary_key": table.primary_key, "watermark_columns": table.watermark_columns, } for table in tables ], } if config.source_type == "tidbcloud": payload["selection"] = table_selection_contract( config, [table.name for table in tables], requested=selected, ) if args.compact: print(json.dumps(payload, separators=(",", ":"))) else: print(json.dumps(payload, indent=2)) elif args.command == "extract": tables = parse_tables(args.tables) try: result = run_extract(config, run_date=args.run_date, run_id=args.run_id, tables=tables) except ExtractRunError as error: print(json.dumps(error.summary, indent=2, sort_keys=True), file=sys.stderr) raise SystemExit(1) from error print(json.dumps(result, indent=2)) elif args.command == "load": tables = parse_tables(args.tables) print( json.dumps( run_load( config, manifest_uri=args.manifest, latest=args.latest, all_unloaded=args.all_unloaded, tables=tables, ), indent=2, ) ) elif args.command == "run-pipeline": tables = parse_tables(args.tables) try: extract = run_extract(config, run_date=args.run_date, run_id=args.run_id, tables=tables) except ExtractRunError as error: print(json.dumps(error.summary, indent=2, sort_keys=True), file=sys.stderr) raise SystemExit(1) from error loads = run_load(config, manifest_uri=extract["manifest_uri"], tables=tables) print(json.dumps({"extract": extract, "loads": loads}, indent=2)) def parse_tables(value: str | None) -> list[str] | None: if value is None: return None tables = [item.strip() for item in value.split(",") if item.strip()] if not tables: raise ValueError("--tables must contain at least one exact table name") return tables if __name__ == "__main__": main()