import argparse import stripe from stripe.error import InvalidRequestError, CardError, APIError import logging from dataclasses import dataclass, asdict import os from dotenv import load_dotenv import json import asyncio logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) load_dotenv(dotenv_path=".env") STRIPE_API_KEY = os.getenv("STRIPE_WRITE_API_KEY") logging.info(f"Using Stripe WRITE API key: {STRIPE_API_KEY}") stripe.api_key = STRIPE_API_KEY # Stripe rate limit settings MAX_REQUESTS_PER_SECOND = 50 REQUEST_INTERVAL = 1 / MAX_REQUESTS_PER_SECOND def read_charge_ids(filepath): """Read charge IDs from a file, one ID per line.""" with open(filepath, "r") as f: return [line.strip() for line in f if line.strip()] @dataclass class RefundResult: charge_id: str success: bool refund_id: str | None = None error_message: str | None = None amount: int | None = None async def refund_charge(charge_id, dry_run=False) -> RefundResult: """Refund a single Stripe charge with reason 'fraudulent'.""" try: if dry_run: # In dry run mode, just simulate a successful refund logging.info(f"DRY RUN: Would refund charge {charge_id} as fraudulent") return RefundResult(charge_id=charge_id, success=True, refund_id="dry_run") # Add a small delay to respect rate limits await asyncio.sleep(REQUEST_INTERVAL) # First, get the charge details to confirm it's valid and not already refunded charge = await asyncio.to_thread(stripe.Charge.retrieve, charge_id) if charge.refunded: logging.warning(f"Charge {charge_id} is already refunded") return RefundResult( charge_id=charge_id, success=False, error_message="Charge already refunded", ) # Perform the refund refund = await asyncio.to_thread( stripe.Refund.create, charge=charge_id, reason="fraudulent" ) logging.info( f"Successfully refunded charge {charge_id} (refund ID: {refund.id})" ) return RefundResult( charge_id=charge_id, success=True, refund_id=refund.id, amount=charge.amount ) except InvalidRequestError as e: logging.error(f"Invalid request error refunding charge {charge_id}: {str(e)}") return RefundResult( charge_id=charge_id, success=False, error_message=str(e), ) except CardError as e: logging.error(f"Card error refunding charge {charge_id}: {str(e)}") return RefundResult( charge_id=charge_id, success=False, error_message=str(e), ) except APIError as e: logging.error(f"API error refunding charge {charge_id}: {str(e)}") return RefundResult( charge_id=charge_id, success=False, error_message=str(e), ) except Exception as e: logging.error(f"Unexpected error refunding charge {charge_id}: {str(e)}") return RefundResult( charge_id=charge_id, success=False, error_message=f"Unexpected error: {str(e)}", ) async def process_charge_batch(batch: list[str], dry_run=False) -> list[RefundResult]: """Process a batch of charge IDs concurrently.""" tasks = [refund_charge(charge_id, dry_run) for charge_id in batch] return await asyncio.gather(*tasks) def process_all_charge_batches( charge_ids: list[str], batch_size: int = 10, dry_run=False ): """Process all charge IDs in batches.""" results = [] successful_refunds = 0 failed_refunds = 0 total_refunded_amount = 0 for i in range(0, len(charge_ids), batch_size): batch = charge_ids[i : i + batch_size] batch_results = asyncio.run(process_charge_batch(batch, dry_run)) results.extend(batch_results) # Update statistics for result in batch_results: if result.success: successful_refunds += 1 if result.amount: total_refunded_amount += result.amount else: failed_refunds += 1 logging.info(f"Processed batch of {len(batch)} charges") logging.info(f"Processed {len(results)} / {len(charge_ids)} charges") logging.info( f"Successful refunds: {successful_refunds}, Failed refunds: {failed_refunds}" ) if not dry_run: logging.info( f"Total amount refunded so far: ${total_refunded_amount / 100:.2f}" ) return results def op_refund_charges(args): """Operation to refund charges in bulk.""" charge_ids = read_charge_ids(args.charge_ids_file) if args.sample and args.sample > 0: sample_size = min(args.sample, len(charge_ids)) charge_ids = charge_ids[:sample_size] logging.info(f"Using sample of {sample_size} charge IDs") logging.info(f"Processing {len(charge_ids)} charge IDs") logging.info(f"Dry run mode: {args.dry_run}") results = process_all_charge_batches( charge_ids, batch_size=args.batch_size, dry_run=args.dry_run ) # Summarize results successful_refunds = sum(1 for r in results if r.success) failed_refunds = sum(1 for r in results if not r.success) total_refunded_amount = sum(r.amount or 0 for r in results if r.success) logging.info("=== Refund Summary ===") logging.info(f"Total charges processed: {len(results)}") logging.info(f"Successful refunds: {successful_refunds}") logging.info(f"Failed refunds: {failed_refunds}") if not args.dry_run: logging.info(f"Total amount refunded: ${total_refunded_amount / 100:.2f}") # Write detailed results to output file if args.output_file: with open(args.output_file, "w") as f: json.dump([asdict(r) for r in results], f, indent=2) logging.info(f"Detailed results written to {args.output_file}") # Write failed charges to a separate file for retry if needed if failed_refunds > 0 and args.output_file: failed_file = f"{os.path.splitext(args.output_file)[0]}_failed.txt" with open(failed_file, "w") as f: for result in results: if not result.success: f.write(f"{result.charge_id}\n") logging.info(f"Failed charge IDs written to {failed_file}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Refund Stripe charges in bulk") parser.add_argument( "--charge-ids-file", required=True, help="Path to file containing charge IDs, one per line", ) parser.add_argument( "--output-file", required=False, help="Path to output file for detailed results", ) parser.add_argument( "--batch-size", type=int, default=10, help="Number of charges to process in each batch", ) parser.add_argument( "--dry-run", action="store_true", help="Run in dry-run mode (no actual refunds processed)", ) parser.add_argument( "--sample", type=int, default=0, help="Only process a sample of N charge IDs", ) args = parser.parse_args() op_refund_charges(args)