#!/usr/bin/env python """ Delete records from an Odoo database based on a model and domain filter. Usage: delete_records.py Example: delete_records.py mydb res.partner --domain "[('active', '=', False)]" --force """ import argparse import ast import json import multiprocessing as mp import os import odoorpc import color_log # Default configuration DEFAULT_HOST = "localhost" DEFAULT_PORT = 8069 DEFAULT_USERNAME = "admin" DEFAULT_PASSWORD = "admin" DEFAULT_DOMAIN = "[]" DEFAULT_PROCESS_SIZE = 30 # Logging levels OK, FAIL, INFO, NOTICE = 0, 1, 2, 3 def parse_arguments(): """Parse command-line arguments.""" parser = argparse.ArgumentParser(description="Delete records from an Odoo model.") parser.add_argument("db_name", help="Database name") parser.add_argument("base_model", help="Model to delete records from") parser.add_argument("--host", default=DEFAULT_HOST, help="Odoo server host") parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="Odoo server port") parser.add_argument("--username", default=DEFAULT_USERNAME, help="Odoo username") parser.add_argument("--password", default=DEFAULT_PASSWORD, help="Odoo password") parser.add_argument("--domain", default=DEFAULT_DOMAIN, help="Domain filter") parser.add_argument( "--process_size", type=int, default=DEFAULT_PROCESS_SIZE, help="Number of parallel processes", ) parser.add_argument( "--force", action="store_true", help="Force delete instead of archiving" ) parser.add_argument( "--soft", action="store_true", help="Archive instead of deleting" ) parser.add_argument( "--refresh-cache", action="store_true", help="Refresh related models cache" ) return parser.parse_args() def connect_to_odoo(host, port, db_name, username, password): """Establish connection to Odoo database.""" odoo = odoorpc.ODOO(host, port=port) color_log.Show(INFO, f"Available databases: {odoo.db.list()}") try: odoo.login(db_name, username, password) color_log.Show(OK, f"Connected to {host}:{port}, DB: {db_name}") return odoo except odoorpc.error.RPCError as e: color_log.Show(FAIL, f"Login failed: {e}") exit(1) except Exception as e: color_log.Show(FAIL, f"Connection error: {e}") exit(1) def parse_domain(domain_str): """Convert domain string to a list and validate it.""" try: domain = ast.literal_eval(domain_str) if not isinstance(domain, list) or not all( isinstance(t, (tuple, list)) and len(t) == 3 for t in domain ): raise ValueError("Domain must be a list of 3-tuples") return domain except (ValueError, SyntaxError) as e: color_log.Show(FAIL, f"Invalid domain format: {e}") exit(1) def get_related_fields(odoo, db_name, base_model, process_size, refresh_cache=False): """Fetch related fields for the base model, using cache if available.""" cache_file = f"cache/{db_name}/{base_model}.cache.json" if not refresh_cache and os.path.exists(cache_file): with open(cache_file, "r") as f: color_log.Show(INFO, f"Loaded related models from cache: {base_model}") return json.load(f) color_log.Show(INFO, f"Fetching related models for {base_model}...") related_models = {} skipped_models = set() def fetch_fields(model_name): if model_name in skipped_models: return {} try: fields = odoo.env[model_name].fields_get() related = {} for name, info in fields.items(): if ( info.get("type") in ["many2one", "many2many", "one2many"] and info.get("relation") == base_model ): related.setdefault(model_name, []).append(name) return related except Exception as e: color_log.Show(NOTICE, f"Skipping {model_name}: {e}") skipped_models.add(model_name) return {} model_ids = odoo.env["ir.model"].search([]) models = [m["model"] for m in odoo.env["ir.model"].read(model_ids, ["model"])] with mp.Pool(processes=process_size) as pool: results = pool.map(fetch_fields, models) for result in results: related_models.update(result) os.makedirs(os.path.dirname(cache_file), exist_ok=True) with open(cache_file, "w") as f: json.dump(related_models, f, indent=4) return related_models def unreference_record(odoo, record_id, model_name, related_models, process_id): """Remove references to a record from related models.""" for related_model, fields in related_models.items(): related_obj = odoo.env[related_model] for field in fields: try: related_ids = related_obj.search([(field, "=", record_id)]) if related_ids: related_obj.browse(related_ids).write({field: False}) color_log.Show( OK, f"{process_id}: Unreferenced {record_id} from {related_model} ({field})", ) except Exception as e: color_log.Show( NOTICE, f"{process_id}: Error unreferencing {record_id} in {related_model}: {e}", ) def delete_batch(odoo, batch, model_name, related_models, process_id, force, soft): """Delete or archive a batch of records.""" model = odoo.env[model_name] deleted, archived, skipped = 0, 0, 0 for record_id in batch: is_referenced = any( odoo.env[m].search_count([(f, "=", record_id)]) > 0 for m, fields in related_models.items() for f in fields ) try: if is_referenced and not force: model.browse(record_id).write({"active": False}) archived += 1 color_log.Show( INFO, f"{process_id}: Archived {model_name} ID {record_id}" ) elif not soft or force: model.browse(record_id).unlink() deleted += 1 color_log.Show( INFO, f"{process_id}: Deleted {model_name} ID {record_id}" ) except Exception as e: if force: try: unreference_record( odoo, record_id, model_name, related_models, process_id ) model.browse(record_id).unlink() deleted += 1 color_log.Show( INFO, f"{process_id}: Force deleted {model_name} ID {record_id}" ) except Exception as e: skipped += 1 color_log.Show(NOTICE, f"{process_id}: Skipped {record_id}: {e}") else: skipped += 1 color_log.Show(INFO, f"{process_id}: Skipped {record_id}: {e}") color_log.Show( OK, f"{process_id}: {model_name} - Deleted: {deleted}, Archived: {archived}, Skipped: {skipped}", ) def main(): """Orchestrate the deletion process.""" args = parse_arguments() odoo = connect_to_odoo(args.host, args.port, args.db_name, args.username, args.password) domain = parse_domain(args.domain) model = odoo.env[args.base_model] record_ids = model.search(domain) if not record_ids: color_log.Show( FAIL, f"No records found in {args.base_model} with domain {args.domain}" ) return related_models = get_related_fields( odoo, args.db_name, args.base_model, args.process_size, args.refresh_cache ) batches = [ record_ids[i : i + args.process_size] for i in range(0, len(record_ids), args.process_size) ] processes = [] for i, batch in enumerate(batches, 1): p = mp.Process( target=delete_batch, args=( odoo, batch, args.base_model, related_models, f"Process-{i}", args.force, args.soft, ), ) processes.append(p) p.start() for p in processes: p.join() color_log.Show(OK, "Deletion process completed.") if __name__ == "__main__": main()