251 lines
8.4 KiB
Python
Executable File
251 lines
8.4 KiB
Python
Executable File
#!/usr/bin/env python
|
|
"""
|
|
Delete records from an Odoo database based on a model and domain filter.
|
|
|
|
Usage:
|
|
delete_records.py <db_name> <base_model>
|
|
|
|
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()
|