email_pending_user_deletes.py #6

#!/usr/bin/env python3
"""
Warn Perforce users whose accounts have been inactive for a specified number
of weeks using the P4Python API.

This script connects to the Perforce server and retrieves all user accounts via
the P4Python `users` command.  For each user, it examines the last access
time and, if the account has been inactive longer than the configured
threshold, sends an email warning that the account will be deleted in one
week unless it is used.

"""

import argparse
import logging
import smtplib
import sys
import time
from email.message import EmailMessage
from typing import Iterable, Optional

from P4 import P4, P4Exception  # type: ignore

import sdputils  # type: ignore


def parse_args(argv: Iterable[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Send account inactivity warnings 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
    # Set user if available
    try:
        p4.user = utils.p4user  # type: ignore[attr-defined]
    except AttributeError:
        pass
    # Set password from file if exists
    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 send_email(smtp: smtplib.SMTP, sender: str, recipient: str, subject: str, body: str) -> None:
    msg = EmailMessage()
    msg["From"] = sender
    msg["To"] = recipient
    msg["Subject"] = subject
    msg.set_content(body)
    smtp.send_message(msg)


def warn_inactive_users(
    p4: P4,
    weeks: int,
    mailhost: str,
    administrator: str,
    complain_from: str,
    cc_admin: Optional[str],
    server: str,
) -> None:
    threshold_seconds = weeks * 7 * 24 * 60 * 60
    now = int(time.time())
    try:
        users = p4.run("users")
    except P4Exception as exc:
        logging.error("Failed to list users: %s", exc)
        return
    try:
        with smtplib.SMTP(mailhost) as smtp:
            for rec in users:
                username = rec.get("User") or rec.get("user")
                if not username:
                    continue
                # Skip protected accounts
                if username.lower() in sdputils.SKIP_USERS:
                    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_seconds:
                    continue
                email = rec.get("Email") or rec.get("email") or username
                subject = f"User {username} on {server} server is scheduled for deletion."
                body = (
                    f"Your Perforce account hasn't been used in the last {weeks} weeks. "
                    "It will be deleted in one week unless you log into Perforce before that time.\n\n"
                    "Thanks for your cooperation,\n"
                    "The Perforce Admin Team\n"
                )
                # Send to user and optionally CC admin
                recipients = [email]
                if cc_admin:
                    recipients.append(cc_admin)
                try:
                    for recip in recipients:
                        send_email(smtp, administrator, recip, subject, body)
                except Exception as exc:
                    logging.error("Failed to send email to %s: %s", recipients, exc)
    except Exception as exc:
        logging.error("Unable to connect to SMTP host %s: %s", mailhost, 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
    try:
        utils = sdputils.SDPUtils(args.instance)
        weeks = int(utils.get("userweeks"))
        administrator = utils.get("administrator")
        complain_from = utils.get("complain_from")
        cc_admin = utils.get("ccadmin") or None
        mailhost = utils.get("mailhost")
        warn_inactive_users(p4, weeks, mailhost, administrator, complain_from, cc_admin, utils.server)
    finally:
        if p4.connected():
            p4.disconnect()
    return 0


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