scripts/file_history.py — Data Room Toolkit

#!/usr/bin/env python3
# Data Room Toolkit 0.1.5
# By Zero Width Lab (zerowidthlab.com)
"""List the version history for one or more files in a SharePoint library."""

from __future__ import annotations

import argparse
import json
import math
import sys
from typing import Any, Iterator
from urllib.parse import quote

from graph_client import (
    ConfigError,
    CredentialError,
    GraphClientError,
    acquire_token,
    collection_page,
    check_collection_url,
    rooted_remote_path,
    validate_remote_path,
    display_text,
    encode_path_segments,
    load_config,
    request,
    resolve_drive_id,
)

__author__ = "Zero Width Lab"


def say(text: object = "", *, error: bool = False) -> None:
    """Print one human-readable line with untrusted text escaped by display_text."""
    print(display_text(text), file=sys.stderr if error else sys.stdout)


GRAPH_ROOT = "https://graph.microsoft.com/v1.0"
REDACTED_USER = "[redacted]"


class GraphSession:
    def __init__(self, token: str, timeout: float, max_retries: int, throttle: float):
        self.token = token
        self.timeout = timeout
        self.max_retries = max_retries
        self.throttle = throttle

    def request(
        self, method: str, url: str
    ) -> tuple[int, Any, dict[str, str]]:
        return request(
            method,
            url,
            token=self.token,
            timeout=self.timeout,
            max_retries=self.max_retries,
            throttle=self.throttle,
            expect_json=True,
        )


def non_negative_int(value: str) -> int:
    try:
        parsed = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be an integer >= 0") from exc
    if parsed < 0:
        raise argparse.ArgumentTypeError("must be an integer >= 0")
    return parsed


def non_negative_float(value: str) -> float:
    try:
        parsed = float(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be a finite number >= 0") from exc
    if not math.isfinite(parsed) or parsed < 0:
        raise argparse.ArgumentTypeError("must be a finite number >= 0")
    return parsed


def combine_remote_path(*parts: str | None) -> str:
    return "/".join(validate_remote_path(part) for part in parts if part)


def modifier_name(item: dict[str, Any], redact: bool) -> str | None:
    actor = item.get("lastModifiedBy") or item.get("modifiedBy") or {}
    name = (actor.get("user") or {}).get("displayName")
    if not name:
        return None
    return REDACTED_USER if redact else str(name)


def request_json(
    client: GraphSession, method: str, url: str
) -> tuple[int, dict[str, Any], dict[str, str]]:
    status, payload, headers = client.request(method, url)
    return status, payload if isinstance(payload, dict) else {}, headers


def iter_versions(client: GraphSession, url: str) -> Iterator[dict[str, Any]]:
    next_url: str | None = url
    visited: set[str] = set()
    while next_url:
        check_collection_url(next_url, visited)
        status, payload, _ = client.request("GET", next_url)
        if not 200 <= status < 300:
            raise RuntimeError(f"version history request failed with HTTP {status}")
        items, link = collection_page(payload, url)
        for item in items:
            yield item
        next_url = link


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="List file version history from a SharePoint document library.",
        epilog="By Zero Width Lab (zerowidthlab.com)",
    )
    parser.add_argument("--drive-id")
    parser.add_argument("--site-id")
    parser.add_argument("--root-path")
    parser.add_argument("--config")
    parser.add_argument("--path", action="append", required=True)
    parser.add_argument("--redact-users", action="store_true")
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--max-retries", type=non_negative_int)
    parser.add_argument("--throttle", type=non_negative_float)
    return parser


def merged_setting(
    cli_value: Any, config: dict[str, Any], key: str, default: Any = None
) -> Any:
    return cli_value if cli_value is not None else config.get(key, default)


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    failures = 0
    results: list[dict[str, Any]] = []
    try:
        overrides = {
            key: value
            for key, value in {
                "drive_id": args.drive_id,
                "site_id": args.site_id,
                "root_path": args.root_path,
                "max_retries": args.max_retries,
                "throttle_seconds": args.throttle,
            }.items()
            if value is not None
        }
        config = load_config(args.config, overrides=overrides, require_target=True)
        max_retries = int(config.get("max_retries", 4))
        throttle = float(config.get("throttle_seconds", 0.0))
        timeout = float(config.get("request_timeout_seconds", 60.0))
        root_path = config["root_path"]
        paths = [validate_remote_path(path) for path in args.path]
        token = acquire_token(
            timeout=timeout, max_retries=max_retries, throttle=throttle
        )
        client = GraphSession(token, timeout, max_retries, throttle)
        drive_id = resolve_drive_id(
            token,
            drive_id=config.get("drive_id"),
            site_id=config.get("site_id"),
            timeout=timeout,
            max_retries=max_retries,
            throttle=throttle,
        )
        drive = quote(str(drive_id), safe="")

        for requested_path in paths:
            full_path = rooted_remote_path(root_path, requested_path)
            encoded = encode_path_segments(full_path)
            metadata_url = (
                f"{GRAPH_ROOT}/drives/{drive}/root:/{encoded}:"
                "?$select=id,name,size,lastModifiedDateTime,lastModifiedBy"
            )
            status, metadata, _ = request_json(client, "GET", metadata_url)
            if status == 404:
                failures += 1
                message = "file not found"
                results.append({"path": requested_path, "error": message, "status": 404})
                say(f"error: {requested_path}: {message}", error=True)
                continue
            if not 200 <= status < 300 or not metadata.get("id"):
                failures += 1
                message = f"metadata request failed with HTTP {status}"
                results.append(
                    {"path": requested_path, "error": message, "status": status}
                )
                say(f"error: {requested_path}: {message}", error=True)
                continue

            versions_url = (
                f"{GRAPH_ROOT}/drives/{drive}/root:/{encoded}:/versions?$top=200"
            )
            try:
                versions = [
                    {
                        "id": version.get("id"),
                        "modified": version.get("lastModifiedDateTime"),
                        "size": int(version.get("size") or 0),
                        "modifier": modifier_name(version, args.redact_users),
                    }
                    for version in iter_versions(client, versions_url)
                ]
            except RuntimeError as exc:
                failures += 1
                results.append({"path": requested_path, "error": str(exc)})
                say(f"error: {requested_path}: {exc}", error=True)
                continue
            versions.sort(key=lambda item: str(item.get("modified") or ""))
            results.append(
                {
                    "path": requested_path,
                    "current": {
                        "name": metadata.get("name"),
                        "size": int(metadata.get("size") or 0),
                        "modified": metadata.get("lastModifiedDateTime"),
                        "modifier": modifier_name(metadata, args.redact_users),
                    },
                    "versions": versions,
                }
            )
        if args.json:
            print(json.dumps({"results": results}, indent=2, ensure_ascii=False))
        else:
            for result in results:
                say(f"{result['path']}")
                if result.get("error"):
                    say(f"  error: {result['error']}")
                    continue
                for version in result["versions"]:
                    modifier = version.get("modifier") or "unknown"
                    say(
                        f"  {version['id']} {version['modified']} "
                        f"{version['size']} {modifier}"
                    )
        return 1 if failures else 0
    except (CredentialError, ConfigError) as exc:
        say(f"error: {exc}", error=True)
        return 1
    except GraphClientError as exc:
        status = f" with HTTP {exc.status}" if exc.status is not None else ""
        say(f"error: Microsoft Graph request failed{status}: {exc}", error=True)
        return 1
    except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
        say(f"error: {exc}", error=True)
        return 1


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

License: MIT (read it).

Back to the listing