unload_clients_with_delete.py #6

#!/usr/bin/env python3
"""
Unload or delete Perforce clients that have been inactive for a configured
number of weeks, using the P4Python API.  Clients with files opened for
exclusive locks are deleted rather than unloaded.

"""

import argparse
import logging
import sys
import time
from typing import Iterable, List, Set

from P4 import P4, P4Exception  # type: ignore

import sdputils  # type: ignore


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Unload or delete inactive Perforce 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_clients_with_exclusive_locks(p4: P4) -> Set[str]:
    """Return a set of clients that have files opened with exclusive locks."""
    clients: Set[str] = set()
    try:
        opened = p4.run("opened", "-a")
    except P4Exception as exc:
        logging.error("Failed to list opened files: %s", exc)
        return clients
    for rec in opened:
        filetype = rec.get("type") or rec.get("Type")
        client = rec.get("client") or rec.get("Client")
        if client and filetype and "+l" in filetype:
            clients.add(client)
    return clients


def get_inactive_clients(p4: P4, weeks: int) -> List[str]:
    now = int(time.time())
    threshold = weeks * 7 * 24 * 60 * 60
    try:
        clients = p4.run("clients", "-a")
    except P4Exception as exc:
        logging.error("Failed to list clients: %s", exc)
        return []
    result: List[str] = []
    for rec in clients:
        name = rec.get("client") or rec.get("Client")
        if not name or "swarm-project" in name.lower():
            continue
        access_str = rec.get("Access") or rec.get("access")
        if not access_str:
            continue
        try:
            access = int(access_str)
        except (ValueError, TypeError):
            continue
        if now - access >= threshold:
            result.append(name)
    return result


def process_clients(p4: P4, inactive: List[str], exclusives: Set[str]) -> None:
    for client in inactive:
        try:
            if client in exclusives:
                # Delete clients with exclusive locks (unload will fail)
                p4.run("client", "-f", "-Fd", client)
                logging.info("Deleted client %s", client)
            else:
                p4.run("unload", "-f", "-L", "-z", "-c", client)
                logging.info("Unloaded client %s", client)
        except P4Exception as exc:
            logging.error("Failed to process 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("weeks"))
    inactive_clients = get_inactive_clients(p4, weeks)
    clients_with_excl = get_clients_with_exclusive_locks(p4)
    process_clients(p4, inactive_clients, clients_with_excl)
    return 0


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