delete_unload_clients.py #7

#!/usr/bin/env python3
"""
Delete unloaded Perforce clients that have been inactive for a configured
number of weeks using the P4Python API.

This script identifies client workspaces stored in the unload depot whose
last access time exceeds a configured threshold.  It then forcibly deletes
those clients using `p4 client -fd -Fd`.  Clients are identified via the
`p4 clients -U` command, which lists unloaded client metadata.
"""

import argparse
import logging
import sys
import time
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="Delete inactive unloaded clients via 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_clients(p4: P4, weeks: int) -> List[str]:
    now = int(time.time())
    threshold = weeks * 7 * 24 * 60 * 60
    try:
        clients = p4.run("clients", "-U")
    except P4Exception as exc:
        logging.error("Failed to list unloaded clients: %s", exc)
        return []
    result: List[str] = []
    for rec in clients:
        name = rec.get("client") or rec.get("Client")
        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:
            result.append(name)
    return result


def delete_clients(p4: P4, clients: List[str]) -> None:
    for client in clients:
        try:
            # -fd (force delete) and -Fd (delete unloaded client)
            p4.run("client", "-fd", "-Fd", client)
            logging.info("Deleted unloaded client %s", client)
        except P4Exception as exc:
            logging.error("Failed to delete client %s: %s", client, exc)


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("deleteweeks"))
    clients = get_inactive_unloaded_clients(p4, weeks)
    delete_clients(p4, clients)
    return 0


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