source/sphenixprod/dst_deleter.py

0001 #!/usr/bin/env python
0002 """
0003 dst_deleter.py — standalone DST file deletion helper.
0004
0005 Two subcommands:
0006
0007   generate  Query the FileCatalog and write a TSV work list (lfn<TAB>path).
0008             Inspect the list before proceeding.
0009
0010   execute   Read the work list in batches and, for each batch:
0011               1. Delete the corresponding rows from `files` and `datasets`.
0012               2. Unlink the physical files.
0013             DB deletion intentionally happens first: once a batch starts, the
0014             catalog should no longer advertise files that are being removed.
0015             Completed batches are removed from the work list so interrupted
0016             runs can be restarted.
0017
0018 Does NOT depend on any other module in this project.
0019 """
0020
0021 import argparse
0022 import glob
0023 import logging
0024 import os
0025 import random
0026 import shutil
0027 import subprocess
0028 import sys
0029 import time
0030 from pathlib import Path
0031
0032
0033 import psutil
0034 import pyodbc
0035
0036 _PROC = psutil.Process()
0037
0038 def _rss_mb() -> int:
0039     return _PROC.memory_info().rss // (1024 * 1024)
0040
0041 # ============================================================================
0042 # Inline logger  (mirrors simpleLogger.py)
0043 # ============================================================================
0044 CHATTY_LEVEL_NUM = 5
0045 logging.addLevelName(CHATTY_LEVEL_NUM, "CHATTY")
0046
0047 def _chatty(self, message, *args, **kws):
0048     if self.isEnabledFor(CHATTY_LEVEL_NUM):
0049         self._log(CHATTY_LEVEL_NUM, message, args, stacklevel=2, **kws)
0050 logging.Logger.chatty = _chatty
0051
0052
0053 class _Fmt(logging.Formatter):
0054     show_datetime = True
0055     grey     = "\x1b[38;20m"
0056     yellow   = "\x1b[33;20m"
0057     green    = "\x1b[32;20m"
0058     blue     = "\x1b[36;20m"
0059     red      = "\x1b[31;20m"
0060     bold_red = "\x1b[31;1m"
0061     reset    = "\x1b[0m"
0062     _datetime_fmt = "%(asctime)s [%(levelname)s] - %(message)s"
0063     _plain_fmt = "[%(levelname)s] - %(message)s"
0064
0065     def _base_format(self):
0066         return self._datetime_fmt if self.show_datetime else self._plain_fmt
0067
0068     def format(self, record):
0069         base_format = self._base_format()
0070         formats = {
0071             CHATTY_LEVEL_NUM: self.yellow + base_format + " (%(filename)s:%(lineno)d) " + self.reset,
0072             logging.DEBUG:    self.grey + base_format + " (%(filename)s:%(lineno)d) " + self.reset,
0073             logging.INFO:     self.green + base_format + self.reset,
0074             logging.WARNING:  self.blue + base_format + " (%(filename)s:%(lineno)d) " + self.reset,
0075             logging.ERROR:    self.red + base_format + " (%(filename)s:%(lineno)d) " + self.reset,
0076             logging.CRITICAL: self.bold_red + base_format + " (%(filename)s:%(lineno)d) " + self.reset,
0077         }
0078         formatter = logging.Formatter(formats.get(record.levelno, base_format))
0079         return formatter.format(record)
0080
0081
0082 def _set_log_timestamps_enabled(enabled: bool):
0083     _Fmt.show_datetime = enabled
0084
0085
0086 _log = logging.getLogger('dst_deleter')
0087 if not _log.hasHandlers():
0088     _ch = logging.StreamHandler()
0089     _ch.setFormatter(_Fmt())
0090     _log.addHandler(_ch)
0091
0092 CHATTY = _log.chatty
0093 DEBUG  = _log.debug
0094 INFO   = _log.info
0095 WARN   = _log.warning
0096 ERROR  = _log.error
0097
0098 # ============================================================================
0099 # DB connection strings  (mirrors sphenixdbutils.py)
0100 # ============================================================================
0101 if os.uname().sysname == 'Darwin':
0102     _FCW = 'DRIVER=PostgreSQL Unicode;SERVER=localhost;DATABASE=filecatalogdb;UID=eickolja'
0103     _FCR = 'DRIVER=PostgreSQL Unicode;SERVER=localhost;DATABASE=filecatalogdb;READONLY=True;UID=eickolja'
0104 else:
0105     _FCW = 'DSN=FileCatalog;UID=phnxrc'
0106     _FCR = 'DSN=FileCatalog;READONLY=True;UID=phnxrc'
0107
0108 # ============================================================================
0109 # DB query helper  (mirrors sphenixdbutils.dbQuery)
0110 # ============================================================================
0111 _RETRYABLE = {'40001', '53300', '57P03', '08006', '08001'}
0112
0113 def _db_query(cnxn_string: str, query: str, ntries: int = 5, dryrun: bool = False):
0114     CHATTY(f'[sql]\n{query}')
0115     if dryrun:
0116         INFO(f'[dryrun] would execute:\n{query}')
0117         return None
0118     for itry in range(ntries):
0119         try:
0120             INFO(f"DB connect starting | RSS {_rss_mb()} MB")
0121             conn = pyodbc.connect(cnxn_string)
0122             INFO(f"DB connect complete | RSS {_rss_mb()} MB")
0123             curs = conn.cursor()
0124             INFO(f"DB execute starting | RSS {_rss_mb()} MB")
0125             curs.execute(query)
0126             INFO(f"DB execute complete | RSS {_rss_mb()} MB")
0127             return curs  # pyodbc cursor holds a ref to conn; connection stays alive
0128         except pyodbc.Error as exc:
0129             state = exc.args[0]
0130             ERROR(f"Attempt {itry + 1}/{ntries}: {exc}")
0131             if state in _RETRYABLE:
0132                 delay = min(60, (2 ** itry) * (0.5 + random.random()))
0133                 WARN(f"Retrying in {delay:.1f}s …")
0134                 time.sleep(delay)
0135             else:
0136                 ERROR("Non-retryable DB error. Stop.")
0137                 sys.exit(41)
0138         except Exception as exc:
0139             ERROR(f"Unexpected error: {exc}")
0140             sys.exit(41)
0141     ERROR("Exhausted all DB attempts. Stop.")
0142     sys.exit(41)
0143
0144 def _close_cursor(curs) -> None:
0145     conn = getattr(curs, 'connection', None)
0146     curs.close()
0147     if conn is not None:
0148         conn.close()
0149
0150
0151 # ============================================================================
0152 # Run-number condition builder  (mirrors sphenixdbutils.list_to_condition)
0153 # ============================================================================
0154 def _run_condition(runs: list, table: str = '') -> str:
0155     col = f"{table}.runnumber" if table else "runnumber"
0156     runs = sorted(runs)
0157
0158     n = len(runs)
0159     if n == 0:
0160         ERROR("No run numbers supplied.")
0161         sys.exit(2)
0162     if n == 1:
0163         return f"{col} = {runs[0]}"
0164     if n == 2:
0165         return f"{col} >= {runs[0]} and {col} <= {runs[1]}"
0166     return f"{col} in ({','.join(str(r) for r in runs)})"
0167
0168 # ============================================================================
0169 # Subcommand: generate
0170 # ============================================================================
0171 def _sql_literal(val: str) -> str:
0172     return "'" + val.replace("'", "''") + "'"
0173
0174 def _sql_cond(col: str, val: str) -> str:
0175     op = 'like' if '%' in val else '='
0176     return f"{col} {op} {_sql_literal(val)}"
0177
0178
0179 def _warn_about_home_outfile(outfile: str) -> None:
0180     home = Path.home().resolve()
0181     outpath = Path(outfile).expanduser().resolve()
0182     if home == outpath or home in outpath.parents:
0183         WARN(f"Output file {str(outpath)!r} is under your home directory. Do not generate large deletion work lists there; use /tmp or a scratch area.")
0184     else:
0185         WARN("Do not generate large deletion work lists in your home directory; use /tmp or a scratch area.")
0186
0187
0188
0189 def cmd_generate(args):
0190     runs = _resolve_runs(args)
0191     INFO(f"Run selection: {len(runs)} run(s), first={runs[0]}, last={runs[-1]}")
0192
0193     if args.fetch_size <= 0:
0194         ERROR("--fetch-size must be positive.")
0195         sys.exit(2)
0196
0197     run_cond = _run_condition(runs, table='d')
0198     where_clause = f"""
0199 WHERE  {run_cond}
0200   AND  {_sql_cond('d.dataset', args.dataset)}
0201   AND  {_sql_cond('d.dsttype', args.dsttype)}
0202   AND  {_sql_cond('d.tag',     args.tag)}
0203 """
0204     estimate_query = f"""
0205 SELECT COUNT(*)
0206 FROM   files f
0207 LEFT JOIN datasets d ON f.lfn = d.filename
0208 {where_clause}
0209 ;
0210 """
0211     def page_query(last_lfn: str = None) -> str:
0212         page_where = where_clause
0213         if last_lfn is not None:
0214             page_where += f"  AND  f.lfn > {_sql_literal(last_lfn)}\n"
0215         return f"""
0216 SELECT f.lfn, f.full_file_path
0217 FROM   files f
0218 LEFT JOIN datasets d ON f.lfn = d.filename
0219 {page_where}
0220 ORDER BY f.lfn ASC
0221 LIMIT {args.fetch_size}
0222 ;
0223 """
0224     INFO(f"Querying FileCatalog: dataset={args.dataset!r} dsttype={args.dsttype!r} tag={args.tag!r}")
0225     _warn_about_home_outfile(args.outfile)
0226     INFO(f"Generate RSS at start: {_rss_mb()} MB")
0227
0228     if args.dryrun:
0229         INFO(f'[dryrun] would estimate line count with:\n{estimate_query}')
0230         INFO(f'[dryrun] would execute first page:\n{page_query()}')
0231         INFO(f'[dryrun] would write TSV to {args.outfile!r}')
0232         return
0233
0234     estimate_curs = _db_query(_FCR, estimate_query)
0235     estimated_entries = int(estimate_curs.fetchone()[0])
0236     _close_cursor(estimate_curs)
0237     estimated_lines = estimated_entries + 3
0238     INFO(f"Estimated output size: {estimated_entries:,} entries, about {estimated_lines:,} TSV lines including header. RSS {_rss_mb()} MB.")
0239
0240     count = 0
0241     last_lfn = None
0242     with open(args.outfile, 'w') as fh:
0243         fh.write(f"# dataset: {args.dataset}\n")
0244         fh.write(f"# dsttype: {args.dsttype}\n")
0245         fh.write(f"# tag:     {args.tag}\n")
0246         while True:
0247             INFO(f"Page query starting at {count:,} entries | RSS {_rss_mb()} MB")
0248             curs = _db_query(_FCR, page_query(last_lfn))
0249             INFO(f"Page query opened | RSS {_rss_mb()} MB")
0250             INFO(f"Fetch starting at {count:,} entries | RSS {_rss_mb()} MB")
0251             batch = curs.fetchmany(args.fetch_size)
0252             fetch_rss = _rss_mb()
0253             _close_cursor(curs)
0254             if not batch:
0255                 INFO(f"Fetch returned no rows | RSS {fetch_rss} MB")
0256                 break
0257             batch_count = len(batch)
0258             INFO(f"Fetch returned {batch_count:,} rows | RSS {fetch_rss} MB")
0259             for lfn, path in batch:
0260                 fh.write(f"{lfn}\t{path}\n")
0261             last_lfn = batch[-1][0]
0262             count += batch_count
0263             del batch
0264             INFO(f"  ... {count:,} entries written | RSS {_rss_mb()} MB")
0265
0266     INFO(f"Done. {count} entries written to {args.outfile!r}. Final RSS {_rss_mb()} MB.")
0267     INFO(f"Inspect the list, then run:  dst_deleter.py execute --infile {args.outfile}")
0268
0269 # ============================================================================
0270 # Subcommand: execute
0271 # ============================================================================
0272 def _copy_tail(fh, src: int, dst: int, chunk: int = 1024 * 1024) -> None:
0273     """
0274     Copy bytes [src, EOF) to [dst, ...) within an open r+b file handle, then truncate.
0275     src > dst always (we're removing a section from the front of the data).
0276     Copies in chunks so memory use is O(chunk), not O(file size).
0277     """
0278     read_pos, write_pos = src, dst
0279     while True:
0280         fh.seek(read_pos)
0281         data = fh.read(chunk)
0282         if not data:
0283             break
0284         fh.seek(write_pos)
0285         fh.write(data)
0286         read_pos  += len(data)
0287         write_pos += len(data)
0288     fh.truncate(write_pos)
0289
0290
0291 _MUNLINK = shutil.which('munlink')
0292
0293
0294 _DRYRUN_SHOW_MAX = 3
0295 # Conservative ARG_MAX budget: 2 MB total - ~512 KB for environment = ~1.5 MB for argv.
0296 # At ~175 chars per Lustre path that allows ~8000 paths, so 2000 gives comfortable headroom.
0297 _MUNLINK_CHUNK = 2000
0298
0299
0300 def _delete_files(paths: list, dryrun: bool, shown: list) -> int:
0301     """
0302     Unlink files. Returns number of files processed.
0303     shown is a one-element list [n] tracking how many dryrun paths have been
0304     printed so far across all batches; capped at _DRYRUN_SHOW_MAX.
0305
0306     Uses munlink(1) when available — it unlinks in bulk without per-file
0307     permission/attribute checks, which is significantly faster on Lustre.
0308     Falls back to Path.unlink() otherwise.
0309     """
0310     if _MUNLINK:
0311         return _delete_files_munlink(paths, dryrun, shown)
0312     return _delete_files_python(paths, dryrun, shown)
0313
0314
0315 def _delete_files_munlink(paths: list, dryrun: bool, shown: list) -> int:
0316     if dryrun:
0317         for p in paths:
0318             if shown[0] < _DRYRUN_SHOW_MAX:
0319                 DEBUG(f"[dryrun] would unlink {p}")
0320                 shown[0] += 1
0321         return len(paths)
0322     for i in range(0, len(paths), _MUNLINK_CHUNK):
0323         chunk = paths[i : i + _MUNLINK_CHUNK]
0324         result = subprocess.run([_MUNLINK] + chunk, capture_output=True, text=True)
0325         if result.returncode != 0:
0326             WARN(f"munlink returned {result.returncode}: {result.stderr.strip()}")
0327     return len(paths)
0328
0329
0330 def _delete_files_python(paths: list, dryrun: bool, shown: list) -> int:
0331     count = 0
0332     for p in paths:
0333         if dryrun:
0334             if shown[0] < _DRYRUN_SHOW_MAX:
0335                 DEBUG(f"[dryrun] would unlink {p}")
0336                 shown[0] += 1
0337             count += 1
0338             continue
0339         try:
0340             Path(p).unlink()
0341             count += 1
0342         except FileNotFoundError:
0343             WARN(f"Already gone: {p}")
0344         except OSError as exc:
0345             ERROR(f"Failed to unlink {p}: {exc}")
0346             sys.exit(1)
0347     return count
0348
0349
0350 def _delete_db_batch(lfns: list, dryrun: bool) -> None:
0351     """Delete one batch from `files` then `datasets` by lfn."""
0352     quoted = "','".join(lfns)
0353     in_clause = f"('{quoted}')"
0354
0355     files_sql    = f"DELETE FROM files    WHERE lfn      IN {in_clause}"
0356     datasets_sql = f"DELETE FROM datasets WHERE filename IN {in_clause}"
0357
0358     if dryrun:
0359         sample = "', '".join(lfns[:3])
0360         ellipsis = f", … ({len(lfns) - 3} more)" if len(lfns) > 3 else ""
0361         INFO(f"[dryrun] would DELETE FROM files/datasets WHERE lfn IN ('{sample}'{ellipsis})")
0362         return
0363
0364     curs = _db_query(_FCW, files_sql)
0365     if curs is not None:
0366         DEBUG(f"  files:    {curs.rowcount} rows deleted")
0367         curs.commit()
0368
0369     curs = _db_query(_FCW, datasets_sql)
0370     if curs is not None:
0371         DEBUG(f"  datasets: {curs.rowcount} rows deleted")
0372         curs.commit()
0373
0374
0375 _KNOWN_PREFIXES = [
0376     '/sphenix/lustre01/sphnxpro/production/',
0377     '/sphenix/data/data02/sphnxpro/production/',
0378     '/sphenix/data/data03/sphnxpro/production/',
0379 ]
0380
0381 def _cleanup_base_path(file_path: str, dsttype: str = None) -> str | None:
0382     """
0383     Derive the per-dsttype base directory from a known storage path.
0384     Path structure: {prefix}{dataset}/{physicsmode}/{tag}/{dsttype}/...
0385     Returns the base path up to and including dsttype, or None if unrecognised.
0386     If dsttype is given (from the TSV header), it overrides what is in the path
0387     and any SQL wildcard % is replaced with shell glob *.
0388     """
0389     prefix = next((p for p in _KNOWN_PREFIXES if file_path.startswith(p)), None)
0390     if prefix is None:
0391         return None
0392     parts = file_path[len(prefix):].split('/')
0393     if len(parts) < 4:
0394         return None
0395     dataset, physicsmode, tag, path_dsttype = parts[:4]
0396     effective_dsttype = dsttype.replace('%', '*') if dsttype else path_dsttype
0397     return f"{prefix}{dataset}/{physicsmode}/{tag}/{effective_dsttype}/"
0398
0399
0400 def cmd_execute(args):
0401     infile = args.infile
0402     if not Path(infile).exists():
0403         ERROR(f"Work list not found: {infile!r}")
0404         sys.exit(2)
0405
0406     if _MUNLINK:
0407         INFO(f"munlink found at {_MUNLINK}; will use it for bulk unlinking.")
0408     else:
0409         INFO("munlink not found; falling back to Python Path.unlink().")
0410
0411     total_files = total_batches = 0
0412     shown = [0]
0413     header = {}
0414     first_path = None
0415
0416     with open(infile, 'r+b') as fh:
0417         # Read header, record its end position and line count.
0418         header_lines = 0
0419         while True:
0420             pos = fh.tell()
0421             raw = fh.readline()
0422             if not raw:
0423                 break
0424             line = raw.decode().rstrip('\n')
0425             if line.startswith('#'):
0426                 header_lines += 1
0427                 if ':' in line:
0428                     key, _, val = line[1:].partition(':')
0429                     header[key.strip()] = val.strip()
0430             elif line:
0431                 parts = line.split('\t', 1)
0432                 first_line_path = parts[1] if len(parts) == 2 else None
0433                 fh.seek(pos)   # put the first data line back
0434                 break
0435         header_end = fh.tell()
0436
0437         wc = subprocess.run(['wc', '-l', infile], capture_output=True, text=True)
0438         grand_total = int(wc.stdout.split()[0]) - header_lines
0439         fh.seek(header_end)
0440
0441         if not args.dryrun:
0442             print(f"This will delete {grand_total} files, starting with:\n  {first_line_path}")
0443             answer = input("Are you sure? [y/N] ").strip().lower()
0444             if answer not in ('y', 'yes'):
0445                 INFO("Aborted.")
0446                 return
0447
0448         t_start = time.monotonic()
0449
0450         while True:
0451             # Read one batch of data lines.
0452             lfns, paths = [], []
0453             while len(lfns) < args.batch_size:
0454                 raw = fh.readline()
0455                 if not raw:
0456                     break
0457                 line = raw.decode().rstrip('\n')
0458                 if not line or line.startswith('#'):
0459                     continue
0460                 parts = line.split('\t', 1)
0461                 if len(parts) == 2:
0462                     if first_path is None:
0463                         first_path = parts[1]
0464                     lfns.append(parts[0])
0465                     paths.append(parts[1])
0466                 else:
0467                     WARN(f"Skipping malformed line: {line!r}")
0468
0469             if not lfns:
0470                 break
0471
0472             batch_end = fh.tell()
0473             total_batches += 1
0474             total_files += len(lfns)
0475             pct = 100.0 * total_files / grand_total if grand_total else 0.0
0476             elapsed = time.monotonic() - t_start
0477             remaining = grand_total - total_files
0478             eta_s = int(elapsed * remaining / total_files) if total_files else 0
0479             eta = f"{eta_s // 3600}h{eta_s % 3600 // 60}m{eta_s % 60}s"
0480             INFO(f"Batch {total_batches}: {total_files} | "
0481                  f"{grand_total} total ({pct:.1f}%) | ETA {eta} | RSS {_rss_mb()} MB")
0482
0483             # DB-first is intentional: avoid leaving catalog rows pointing at
0484             # files that this process has already started removing.
0485             _delete_db_batch(lfns, dryrun=args.dryrun)
0486             _delete_files(paths, dryrun=args.dryrun, shown=shown)
0487
0488             if not args.dryrun:
0489                 _copy_tail(fh, src=batch_end, dst=header_end)
0490                 fh.seek(header_end)
0491                 fh.flush()
0492
0493     if args.dryrun and total_files > _DRYRUN_SHOW_MAX:
0494         INFO(f"[dryrun] … and {total_files - _DRYRUN_SHOW_MAX} more files.")
0495
0496     INFO(f"Done. {total_batches} batch(es), {total_files} files processed.")
0497
0498     if first_path:
0499         base = _cleanup_base_path(first_path, dsttype=header.get('dsttype'))
0500         if base:
0501             escaped = base.replace('*', r'\*')
0502             INFO(f"To remove empty directories, run:\n  dst_deleter.py cleanup --path {escaped}")
0503
0504
0505 # ============================================================================
0506 # Subcommand: cleanup
0507 # ============================================================================
0508 def cmd_cleanup(args):
0509     bases = glob.glob(args.path)
0510     if not bases:
0511         ERROR(f"No paths matched: {args.path!r}")
0512         sys.exit(2)
0513
0514     removed = 0
0515     for base in sorted(bases):
0516         base = Path(base)
0517         INFO(f"Cleaning up {base}")
0518         for dirpath, _dirnames, _filenames in os.walk(base, topdown=False):
0519             p = Path(dirpath)
0520             if p == base:
0521                 continue
0522             try:
0523                 if args.dryrun:
0524                     if not any(p.iterdir()):
0525                         DEBUG(f"[dryrun] would rmdir {p}")
0526                         removed += 1
0527                 else:
0528                     p.rmdir()   # raises OSError if not empty
0529                     removed += 1
0530             except OSError:
0531                 pass
0532
0533     INFO(f"{'Would remove' if args.dryrun else 'Removed'} {removed} empty directories.")
0534
0535 # ============================================================================
0536 # Shared helpers
0537 # ============================================================================
0538 def _resolve_runs(args) -> list:
0539     if args.runs is not None:
0540         return args.runs
0541     p = Path(args.runlist)
0542     if not p.exists():
0543         ERROR(f"Run list file not found: {args.runlist}")
0544         sys.exit(2)
0545     with open(p) as fh:
0546         tokens = fh.read().split()
0547     runs = []
0548     for tok in tokens:
0549         try:
0550             runs.append(int(tok))
0551         except ValueError:
0552             WARN(f"Skipping non-integer token in runlist: {tok!r}")
0553     if not runs:
0554         ERROR("Run list file contained no valid run numbers.")
0555         sys.exit(2)
0556     return runs
0557
0558
0559 def _add_verbosity(parser):
0560     vgroup = parser.add_mutually_exclusive_group()
0561     vgroup.add_argument('-v', '--verbose', action='count', default=0,
0562                         help='Increase verbosity (-v INFO, -vv DEBUG, -vvv CHATTY).')
0563     vgroup.add_argument('-d', '--debug',  action='store_true', help='Alias for -vv (DEBUG).')
0564     vgroup.add_argument('--chatty',       action='store_true', help='Alias for -vvv (CHATTY).')
0565
0566
0567 def _set_loglevel(args):
0568     if args.chatty or args.verbose >= 3:
0569         _log.setLevel(CHATTY_LEVEL_NUM)
0570     elif args.debug or args.verbose == 2:
0571         _log.setLevel(logging.DEBUG)
0572     else:
0573         _log.setLevel(logging.INFO)
0574
0575 # ============================================================================
0576 # Argument parsing
0577 # ============================================================================
0578 def _parse_args():
0579     parser = argparse.ArgumentParser(
0580         description='Generate a DST file deletion list and execute batched file + DB cleanup.',
0581         formatter_class=argparse.RawDescriptionHelpFormatter,
0582         epilog="""
0583 Examples:
0584   # Step 1 — build the work list:
0585   dst_deleter.py generate \\
0586       --runs 82300 82400 \\
0587       --dataset run3oo --dsttype 'DST_TRIGGERED_%' --tag pro001_pcdb001_v001 \\
0588       --outfile /tmp/to_delete.tsv
0589
0590   # Inspect the list, then step 2 — delete files and DB entries in batches:
0591   dst_deleter.py execute --infile /tmp/to_delete.tsv
0592
0593   # Dry-run either step:
0594   dst_deleter.py generate ... --dryrun
0595   dst_deleter.py execute  --infile /tmp/to_delete.tsv --dryrun
0596
0597   # Step 3 — remove empty directories (path printed by execute):
0598   dst_deleter.py cleanup --path /sphenix/lustre01/sphnxpro/production/run3oo/physics/pro001_pcdb001_v001/DST_TRIGGERED_EVENT/
0599 """,
0600     )
0601
0602     sub = parser.add_subparsers(dest='command', required=True)
0603
0604     # -- generate -------------------------------------------------------------
0605     gen = sub.add_parser('generate', help='Query FileCatalog and write TSV work list.')
0606     gen.add_argument('--fetch-size', dest='fetch_size', type=int, default=50_000,
0607                      help='Rows fetched per paged DB query (default: 50000).')
0608
0609     rgroup = gen.add_mutually_exclusive_group(required=True)
0610     rgroup.add_argument('--runs', nargs='+', type=int, metavar='RUN',
0611                         help='One run, two for an inclusive range, or more for an explicit list.')
0612     rgroup.add_argument('--runlist', metavar='FILE',
0613                         help='Plain-text file with one run number per line.')
0614
0615     gen.add_argument('--dataset', required=True, help='Dataset name, e.g. run3oo')
0616     gen.add_argument('--dsttype', required=True, help="DST type, e.g. 'DST_TRIGGERED_%' (% triggers LIKE)")
0617     gen.add_argument('--tag',     required=True, help='Production tag, e.g. pro001_pcdb001_v001')
0618     gen.add_argument('-o', '--outfile', required=True, help='Output TSV file (lfn<TAB>full_file_path).')
0619     gen.add_argument('-n', '--dryrun', action='store_true', default=False,
0620                      help='Print SQL without querying or writing.')
0621     _add_verbosity(gen)
0622     gen.set_defaults(func=cmd_generate)
0623
0624     # ── execute ───────────────────────────────────────────────────────────────
0625     exe = sub.add_parser('execute', help='Delete files and DB entries batch by batch.')
0626
0627     exe.add_argument('-i', '--infile', required=True, help='TSV work list produced by generate.')
0628     exe.add_argument('--batch-size', dest='batch_size', type=int, default=10_000,
0629                      help='DB rows deleted and infile lines removed per batch (default: 10000). '
0630                           'munlink is called in internal chunks of 2000 regardless of this value.')
0631     exe.add_argument('-n', '--dryrun', action='store_true', default=False,
0632                      help='Print what would happen without deleting anything.')
0633     _add_verbosity(exe)
0634     exe.set_defaults(func=cmd_execute)
0635
0636     # ── cleanup ───────────────────────────────────────────────────────────────
0637     cln = sub.add_parser('cleanup', help='Remove empty directories under a base path.')
0638
0639     cln.add_argument('--path', required=True,
0640                      help='Base directory to clean up (printed by execute for lustre paths).')
0641     cln.add_argument('-n', '--dryrun', action='store_true', default=False,
0642                      help='Print directories that would be removed without removing them.')
0643     _add_verbosity(cln)
0644     cln.set_defaults(func=cmd_cleanup)
0645
0646     return parser.parse_args()
0647
0648
0649 def main():
0650     args = _parse_args()
0651     _set_log_timestamps_enabled(False)
0652     _set_loglevel(args)
0653
0654     args.func(args)
0655
0656
0657 if __name__ == '__main__':
0658     main()