Janady13/AI_Brain / scripts/bulk_update.py

1"""Utilities for orchestrating high-volume marketing process refreshes.""" 2from __future__ import annotations 3 4import argparse 5import sys 6import time 7from dataclasses import dataclass 8from typing import Any, Dict 9 10from services.seeder_api.app import MarketingQuery, marketing_processes 11 12 13@dataclass(slots=True) 14class BulkUpdatePlan: 15 """Summary metrics describing a bulk update execution.""" 16 17 iterations: int 18 batch_size: int 19 executed_iterations: int 20 dry_run: bool 21 sample_runtime_seconds: float 22 estimated_total_runtime_seconds: float 23 24 def to_dict(self) -> Dict[str, Any]: 25 return { 26 "iterations": self.iterations, 27 "batch_size": self.batch_size, 28 "executed_iterations": self.executed_iterations, 29 "dry_run": self.dry_run, 30 "sample_runtime_seconds": self.sample_runtime_seconds, 31 "estimated_total_runtime_seconds": self.estimated_total_runtime_seconds, 32 } 33 34 35DEFAULT_PROMPT = ( 36 "Scale omni-channel revenue loops for autonomous marketing agents with global payments coverage" 37) 38 39 40def execute_bulk_updates( 41 *, 42 iterations: int = 1_000_000, 43 batch_size: int = 1_000, 44 prompt: str = DEFAULT_PROMPT, 45 limit: int = 2, 46 dry_run: bool = True, 47) -> BulkUpdatePlan: 48 """Drive repeated marketing process orchestrations. 49 50 Args: 51 iterations: Total number of update cycles requested. 52 batch_size: Frequency for progress logging and throttling. 53 prompt: Prompt forwarded into the marketing processes orchestration. 54 limit: How many campaign recommendations to request per iteration. 55 dry_run: When true, only run a small sample to compute projections. 56 57 Returns: 58 A :class:`BulkUpdatePlan` summarising the execution profile. 59 """ 60 61 if iterations <= 0: 62 raise ValueError("iterations must be positive") 63 if batch_size <= 0: 64 raise ValueError("batch_size must be positive") 65 if limit <= 0: 66 raise ValueError("limit must be positive") 67 68 sample_iterations = min(5, iterations) 69 start = time.perf_counter() 70 executed = 0 71 for _ in range(sample_iterations): 72 marketing_processes(MarketingQuery(prompt=prompt, limit=limit)) 73 executed += 1 74 sample_runtime = time.perf_counter() - start 75 per_iteration = sample_runtime / executed if executed else 0.0 76 estimated_total = per_iteration * iterations 77 78 if dry_run: 79 return BulkUpdatePlan( 80 iterations=iterations, 81 batch_size=batch_size, 82 executed_iterations=executed, 83 dry_run=True, 84 sample_runtime_seconds=sample_runtime, 85 estimated_total_runtime_seconds=estimated_total, 86 ) 87 88 processed = executed 89 while processed < iterations: 90 marketing_processes(MarketingQuery(prompt=prompt, limit=limit)) 91 processed += 1 92 if processed % batch_size == 0 or processed == iterations: 93 remaining = iterations - processed 94 eta = max(0.0, estimated_total - per_iteration * processed) 95 print( 96 f"Processed {processed:,}/{iterations:,} updates | remaining {remaining:,} | " 97 f"est. remaining {eta:0.1f}s", 98 file=sys.stderr, 99 ) 100 101 total_runtime = time.perf_counter() - start 102 return BulkUpdatePlan( 103 iterations=iterations, 104 batch_size=batch_size, 105 executed_iterations=processed, 106 dry_run=False, 107 sample_runtime_seconds=sample_runtime, 108 estimated_total_runtime_seconds=total_runtime, 109 ) 110 111 112def _positive_int(value: str) -> int: 113 try: 114 parsed = int(value) 115 except ValueError as exc: # pragma: no cover - argparse already ensures message 116 raise argparse.ArgumentTypeError(str(exc)) from exc 117 if parsed <= 0: 118 raise argparse.ArgumentTypeError("value must be positive") 119 return parsed 120 121 122def main(argv: list[str] | None = None) -> int: 123 parser = argparse.ArgumentParser( 124 description=( 125 "Simulate or execute large-scale marketing process updates using the " 126 "FastAPI orchestration stack." 127 ) 128 ) 129 parser.add_argument( 130 "--iterations", 131 type=_positive_int, 132 default=1_000_000, 133 help="Total update cycles to run (default: 1,000,000).", 134 ) 135 parser.add_argument( 136 "--batch-size", 137 type=_positive_int, 138 default=1_000, 139 help="How often to log progress when executing updates (default: 1,000).", 140 ) 141 parser.add_argument( 142 "--limit", 143 type=_positive_int, 144 default=2, 145 help="Campaign recommendation limit per iteration (default: 2).", 146 ) 147 parser.add_argument( 148 "--prompt", 149 default=DEFAULT_PROMPT, 150 help="Prompt describing the marketing objective (default: preconfigured omni-channel brief).", 151 ) 152 parser.add_argument( 153 "--execute", 154 action="store_true", 155 help="Execute the full run instead of returning a dry-run projection.", 156 ) 157 158 args = parser.parse_args(argv) 159 plan = execute_bulk_updates( 160 iterations=args.iterations, 161 batch_size=args.batch_size, 162 prompt=args.prompt, 163 limit=args.limit, 164 dry_run=not args.execute, 165 ) 166 payload = json_dumps(plan.to_dict()) 167 print(payload) 168 return 0 169 170 171def json_dumps(data: Dict[str, Any]) -> str: 172 # Local helper to avoid importing json at module import time. 173 import json 174 175 return json.dumps(data, indent=2, sort_keys=True) 176 177 178if __name__ == "__main__": # pragma: no cover - CLI entry point 179 raise SystemExit(main()) 180