Find all workspaces in p4 that have any files in a depot path

p4_find_haves.py

#! python

import os
import click
import P4
from pathlib import Path
from typing import Set


@click.command()
@click.argument('search_path')
@click.option('--client')
def find(search_path, client=None):
    print(f"search path: {search_path}")
    print(f"client: {client}")

    p4 = P4.P4()
    p4.exception_level = p4.RAISE_ERROR
    p4.connect()
    print(p4)

    if not client:
        # find for all matching clients
        clients = get_clients_for_path(p4, search_path)
        print()
        for client in clients:
            find_client_haves(p4, client, search_path)
    else:
        # find for a single client
        client_data = p4.fetch_client(client)
        find_client_haves(p4, client_data, search_path)


def get_depot_path(full_depot_path):
    """
    Return //MyDepot from //MyDepot/path/to/file
    """
    return '//' + full_depot_path.lstrip('/').split('/')[0]


def get_clients_for_path(p4, search_path):
    print()
    print(f"finding clients with files synced in: {search_path}")

    # get info about all files
    all_files = get_all_p4_files(p4, search_path)
    print(f"    total existing files: {len(all_files)}")

    # get all clients
    clients = p4.run_clients()

    depot_path = get_depot_path(search_path)
    result = []
    for client in clients:
        client_data = p4.fetch_client(client['client'])
        for mapping in client_data.get('View', list()):
            if mapping.startswith(depot_path):
                result.append(client_data)
                break

    print(f"    matching clients: {len(result)}")
    return result


def get_depot_root_path(p4, depot_path):
    depot_where = p4.run_where(depot_path + '/...')[0]
    local_path = depot_where['path']
    if local_path.endswith('...'):
        local_path = local_path.rstrip('\.')
    return local_path


def remove_prefix(string, prefix):
    if string.startswith(prefix):
        return string[len(prefix):]
    return string


def normpath(path):
    return path.replace('\\', '/')


def differ_in_case(pathA, pathB):
    return pathA != pathB and pathA.lower() == pathB.lower()


def find_client_haves(p4, client, search_path):
    client_name = client['Client']
    client_host = client.get('Host', '')

    p4.host = client_host
    p4.client = client_name

    haves = p4.run_have(search_path)
    if len(haves) > 0:
        print(f"{client['Client']:<40} {len(haves)}")

    return haves


def get_all_p4_files(p4, search_path: str) -> Set[str]:
    def is_not_deleted(fstat):
        if 'action' in fstat and fstat['action'] == 'add':
            return True
        elif 'headAction' in fstat:
            return fstat['headAction'] not in ('delete', 'move/delete')

        LOG.info(f"Excluding p4 file: {fstat}")
        return False

    fstat_files = p4.run_fstat(f'{search_path}')
    files = [f for f in fstat_files if is_not_deleted(f)]
    return files


if __name__ == '__main__':
    find()