unloadaccessdates.py #2

#!/usr/bin/env python3
"""
Generate lists of clients and labels in the unload depot that have not been
accessed for a configured number of weeks, using the P4Python API.

This script is typically called by deletion utilities (e.g. to build the
input lists for deleting unloaded clients).  It can also run stand‑alone,
producing ``unload_clients.txt`` and ``unload_labels.txt`` in the current
directory.  Each file contains one name per line for clients or labels
whose last access is older than the threshold defined in ``maintenance.cfg``.

Key improvements over the original version:

* Uses Python 3 exclusively and removes deprecated compatibility imports.
* Uses the P4Python API (`P4`) to list unloaded clients and labels instead
  of invoking the ``p4`` binary via `os.system`.  The P4 API returns
  structured dictionaries, which we inspect directly.
* Employs context managers (`with` statements) when writing output files,
  ensuring proper resource cleanup【957881026265990†L335-L352】.
* Utilizes the SDP utilities to discover server configuration and login
  credentials.
"""

import argparse
import logging
import sys
import time
from pathlib import Path
from typing import Iterable, List

from P4 import P4, P4Exception  # type: ignore

import sdputils  # type: ignore


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Generate unload client/label lists using P4Python",
    )
    parser.add_argument(
        "instance",
        nargs="?",
        default="1",
        help="SDP instance identifier (default: '1')",
    )
    return parser.parse_args(list(argv))


def connect_p4(instance: str) -> P4:
    utils = sdputils.SDPUtils(instance)
    p4 = P4()
    p4.port = utils.server
    try:
        p4.user = utils.p4user  # type: ignore[attr-defined]
    except AttributeError:
        pass
    passwd_path = f"/p4/common/config/.p4passwd.p4_{instance}.admin"
    try:
        with open(passwd_path) as pf:
            p4.password = pf.read().strip()
    except FileNotFoundError:
        pass
    p4.connect()
    if p4.password:
        p4.run_login()
    return p4


def get_inactive_unloaded(p4: P4, spec: str, weeks: int) -> List[str]:
    """Return a list of unloaded specs (clients or labels) inactive longer than ``weeks`` weeks."""
    now = int(time.time())
    threshold = weeks * 7 * 24 * 60 * 60
    names: List[str] = []
    try:
        # -U lists unloaded specs; '-a' is not used because unloaded lists are separate
        recs = p4.run(spec, "-U")
    except P4Exception as exc:
        logging.error("Failed to list unloaded %s: %s", spec, exc)
        return names
    for rec in recs:
        name = None
        access_str = None
        if spec == "clients":
            name = rec.get("client") or rec.get("Client")
        elif spec == "labels":
            name = rec.get("label") or rec.get("Label")
        access_str = rec.get("Access") or rec.get("access")
        if not name or not access_str:
            continue
        try:
            access = int(access_str)
        except (ValueError, TypeError):
            continue
        if now - access >= threshold:
            names.append(name)
    return names


def write_list(path: Path, names: List[str]) -> None:
    with path.open("w") as fh:
        for name in sorted(names):
            fh.write(f"{name}\n")


def main(argv: Iterable[str] | None = None) -> int:
    args = parse_args(argv or sys.argv[1:])
    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
    try:
        p4 = connect_p4(args.instance)
    except P4Exception:
        return 1
    utils = sdputils.SDPUtils(args.instance)
    weeks = int(utils.get("weeks"))
    # Generate lists for clients and labels
    clients = get_inactive_unloaded(p4, "clients", weeks)
    labels = get_inactive_unloaded(p4, "labels", weeks)
    write_list(Path("unload_clients.txt"), clients)
    write_list(Path("unload_labels.txt"), labels)
    logging.info("Wrote %d clients to unload_clients.txt and %d labels to unload_labels.txt", len(clients), len(labels))
    return 0


if __name__ == "__main__":
    sys.exit(main())