cli.py

"""Command-line inspector: print the saved TSMIS connections in ArcGIS Pro projects.

Projects are only read. Nothing is saved: no lists, settings, exports or updates.
"""
import argparse
import importlib.util
import os
import subprocess
import sys
from pathlib import Path

sys.dont_write_bytecode = True  # Leave the extracted script folder exactly as shipped.

from runtime import default_folder, find_arcgis_python, worker_environment  # noqa: E402
from version import APP_NAME, VERSION  # noqa: E402
from worker import run  # noqa: E402

PROJECT_COLUMNS = (("project", "Project"), ("environments", "Environment"), ("versions", "Saved version"),
                   ("services", "Service name"), ("status", "Status"))
LAYER_COLUMNS = (("project", "Project"), ("map", "Map"), ("layer", "Layer / table"), ("environment", "Environment"),
                 ("version", "Saved version"), ("service", "Service name"), ("status", "Status"))
PREPARED = "--in-arcgis-python"


def parse_args(argv):
    parser = argparse.ArgumentParser(prog="cli.py", description="List the saved TSMIS connections in ArcGIS Pro "
                                     "projects. Projects are only read; nothing is saved.")
    parser.add_argument("folder", nargs="?", default=default_folder(),
                        help="folder containing .aprx projects (default: %(default)s)")
    parser.add_argument("--no-recursive", dest="recursive", action="store_false",
                        help="inspect only this folder, not its subfolders")
    parser.add_argument("--match", default="tsmis", help="identifier to find in service URLs, workspaces and "
                        "datasets (default: tsmis, which also matches tsnr)")
    parser.add_argument("--layers", action="store_true", help="also list every matching layer and table")
    parser.add_argument("--version", action="version", version=APP_NAME + " CLI " + VERSION)
    parser.add_argument(PREPARED, dest="prepared", action="store_true", help=argparse.SUPPRESS)
    args = parser.parse_args(argv)
    # cmd passes a quoted path ending in a backslash as C:\Projects" and folder names cannot contain quotes.
    args.folder = args.folder.strip().strip('"') or parser.get_default("folder")
    args.match = args.match.strip()
    if not args.match:
        parser.error("--match needs an identifier, usually tsmis")
    return args


def cell(value):
    """A list takes one line per item; anything else is a single line."""
    if isinstance(value, list):
        return [str(item) for item in value] or [""]
    return ["" if value is None else str(value)]


def table(rows, columns):
    """Fixed-width columns, laid out like the desktop table: one line per list item."""
    lines = [[[title] for _, title in columns]] + [[cell(row.get(key)) for key, _ in columns] for row in rows]
    widths = [max(len(item) for line in lines for item in line[i]) for i in range(len(columns))]
    lines.insert(1, [["-" * width] for width in widths])
    rendered = []
    for line in lines:
        for index in range(max(len(items) for items in line)):
            rendered.append("  ".join((items[index] if index < len(items) else "").ljust(width)
                                      for items, width in zip(line, widths)).rstrip())
    return "\n".join(rendered)


def report(root, projects, layers, out):
    if not projects:
        return

    def name(project):
        return os.path.relpath(project["path"], root)

    print(table([dict(project, project=name(project)) for project in projects], PROJECT_COLUMNS), file=out)
    if layers:
        rows = [dict(row, project=name(project)) for project in projects for row in project["rows"] if row.get("is_tsmis")]
        print("", file=out)
        print(table(rows, LAYER_COLUMNS) if rows else "No matching layers or tables.", file=out)
    issues = [(name(project), issue) for project in projects for issue in project["errors"]]
    issues += [(name(project) + " > " + row["layer"], row["error"])
               for project in projects for row in project["rows"] if row.get("error")]
    if issues:
        print("\nRead issues:", file=out)
        for where, issue in issues:
            print("  " + where + ": " + issue, file=out)


def resolve_folder(folder, err):
    """The absolute folder path, or None after explaining that it does not exist."""
    root = os.path.abspath(folder)
    if os.path.isdir(root):
        return root
    print("Folder not found: " + root + ". Pass the folder that holds your ArcGIS Pro projects.", file=err)
    return None


def inspect_folder(folder, recursive=True, match="tsmis", layers=False, out=None, err=None, arcpy_module=None):
    """Read every project under the folder and print the results. Returns the exit code."""
    out, err = out or sys.stdout, err or sys.stderr
    root = resolve_folder(folder, err)
    if root is None:
        return 2
    projects, warnings, state = [], [], {"total": 0, "error": ""}

    def emit(event):
        kind = event["type"]
        if kind == "discovered":
            state["total"] = event["total"]
            warnings.extend(event["warnings"])
            print("Found " + str(event["total"]) + " project file(s) under " + root, file=err)
        elif kind == "progress":
            counter = "[" + str(event["current"]) + "/" + str(state["total"]) + "] " if "current" in event else ""
            print(counter + event["message"], file=err)
        elif kind == "runtime":
            print("ArcGIS Pro " + event["arcgis_version"], file=err)
        elif kind == "project":
            projects.append(event["project"])
        elif kind == "fatal":
            state["error"] = event["message"]

    code = 0
    try:
        run({"root": root, "recursive": recursive, "match": match}, emit, arcpy_module)
    except KeyboardInterrupt:
        state["error"] = "Stopped before every project was read."
        code = 130
    report(root, projects, layers, out)
    for warning in warnings:
        print("Warning: " + warning, file=err)
    if state["error"]:
        if projects:
            print("Only " + str(len(projects)) + " of " + str(state["total"]) + " projects were read.", file=err)
        print(state["error"], file=err)
        return code or 1
    if not state["total"]:
        print("No .aprx projects found under " + root, file=out)
    return 0


def arcpy_available():
    try:
        return importlib.util.find_spec("arcpy") is not None
    except (ImportError, ValueError):
        return False


def relaunch(argv):
    """ArcPy needs ArcGIS Pro's interpreter and environment. Run the scan in a
    child interpreter prepared exactly like the desktop app's worker."""
    python = sys.executable if arcpy_available() else find_arcgis_python()
    if not python:
        print("ArcGIS Pro was not found on this PC. Install ArcGIS Pro, or run cli.py with its python.exe.", file=sys.stderr)
        return 1
    command = [python, "-B", "-u", str(Path(__file__).resolve()), PREPARED, *argv]
    process = subprocess.Popen(command, env=worker_environment(python))
    try:
        return process.wait()
    except KeyboardInterrupt:
        return process.wait()  # The child received the same Ctrl+C and prints what it has read.


def main(argv=None):
    argv = sys.argv[1:] if argv is None else list(argv)
    args = parse_args(argv)
    try:
        if args.prepared:
            return inspect_folder(args.folder, args.recursive, args.match, args.layers)
        # A wrong path is reported here, before ArcGIS Pro's Python is started.
        return relaunch(argv) if resolve_folder(args.folder, sys.stderr) else 2
    except KeyboardInterrupt:
        print("Stopped.", file=sys.stderr)
        return 130


if __name__ == "__main__":
    raise SystemExit(main())