scripts/dataroom_search.py — Data Room Toolkit

#!/usr/bin/env python3
# Data Room Toolkit 0.1.5
# By Zero Width Lab (zerowidthlab.com)
"""Inventory and optionally rank files in a SharePoint document library."""

from __future__ import annotations

import argparse
import json
import math
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterator
from urllib.parse import quote

from graph_client import (
    ConfigError,
    CredentialError,
    GraphClientError,
    acquire_token,
    collection_page,
    check_collection_url,
    write_new_text,
    LocalOutputError,
    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 parse_since(value: str) -> datetime:
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            "must be an ISO 8601 date or date-time"
        ) from exc
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def parse_keywords(values: list[str] | None) -> list[str]:
    keywords: list[str] = []
    for value in values or []:
        for keyword in value.split(","):
            clean = keyword.strip()
            if clean and clean.casefold() not in {item.casefold() for item in keywords}:
                keywords.append(clean)
    return keywords


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


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


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_collection(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"Graph request failed with HTTP {status}")
        items, link = collection_page(payload, url)
        for item in items:
            yield item
        next_url = link


def initial_item_url(drive_id: str, root_path: str) -> str:
    root_path = validate_remote_path(root_path, allow_empty=True)
    drive = quote(drive_id, safe="")
    selection = "?$select=id,name"
    if root_path:
        return (
            f"{GRAPH_ROOT}/drives/{drive}/root:/"
            f"{encode_path_segments(root_path)}:{selection}"
        )
    return f"{GRAPH_ROOT}/drives/{drive}/root{selection}"


def walk_files(
    client: GraphSession,
    drive_id: str,
    root_path: str,
    max_depth: int | None,
    throttle: float,
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
    status, root, _ = request_json(client, "GET", initial_item_url(drive_id, root_path))
    if status != 200 or not root.get("id"):
        raise RuntimeError(f"cannot read the selected library root, HTTP {status}")

    drive = quote(drive_id, safe="")
    stack: list[tuple[str, int]] = [(root_path, 0)]
    files: list[dict[str, Any]] = []
    skipped: list[dict[str, str]] = []
    while stack:
        parent_path, depth = stack.pop()
        route = f"root:/{encode_path_segments(parent_path)}:" if parent_path else "root"
        item_url = (
            f"{GRAPH_ROOT}/drives/{drive}/{route}/children"
            "?$top=200&$select=id,name,size,folder,file,lastModifiedDateTime,"
            "lastModifiedBy,parentReference"
        )
        try:
            for item in iter_collection(client, item_url):
                try:
                    name = validate_remote_path(item.get("name"))
                except ConfigError as exc:
                    raise ConfigError(f"item {ascii(item.get('name'))}: {exc}") from exc
                if "/" in name:
                    raise ConfigError("remote item name must be one path segment")
                path = combine_remote_path(parent_path, name)
                if "folder" in item:
                    if max_depth is None or depth < max_depth:
                        child_id = item.get("id")
                        if child_id:
                            stack.append((path, depth + 1))
                elif "file" in item:
                    files.append(
                        {
                            "name": name,
                            "path": path,
                            "size": int(item.get("size") or 0),
                            "modified": item.get("lastModifiedDateTime"),
                            "modifier": item.get("lastModifiedBy"),
                        }
                    )
        except RuntimeError as exc:
            skipped.append({"path": parent_path, "error": str(exc)})
            say(
                f"warning: skipping folder '{parent_path or '(root)'}': {exc}",
                error=True,
            )
            continue
    return files, skipped


def keyword_rank(path: str, keywords: list[str]) -> tuple[int, list[str]]:
    folded = path.casefold()
    matched = [keyword for keyword in keywords if keyword.casefold() in folded]
    score = sum(folded.count(keyword.casefold()) for keyword in matched)
    return score, matched


def content_search(
    client: GraphSession,
    drive_id: str,
    root_path: str,
    keywords: list[str],
    redact_users: bool,
    skipped: list[dict[str, str]] | None = None,
    outside_root: list[str] | None = None,
) -> list[dict[str, Any]]:
    root_path = validate_remote_path(root_path, allow_empty=True)
    results: list[dict[str, Any]] = []
    seen: set[str] = set()
    drive = quote(drive_id, safe="")
    search_base = (
        f"root:/{encode_path_segments(root_path)}:/search"
        if root_path
        else "root/search"
    )
    for keyword in keywords:
        encoded_keyword = quote(keyword, safe="")
        url = (
            f"{GRAPH_ROOT}/drives/{drive}/{search_base}(q='{encoded_keyword}')"
            "?$top=100&$select=id,name,size,lastModifiedDateTime,lastModifiedBy,"
            "parentReference"
        )
        for item in iter_collection(client, url):
            key = str(item.get("id") or item.get("webUrl") or item.get("name"))
            if key in seen:
                continue
            seen.add(key)
            try:
                details = item
                parent = (details.get("parentReference") or {}).get("path")
                if not parent:
                    item_id = item.get("id")
                    if not isinstance(item_id, str) or not item_id:
                        raise ConfigError("missing item id for path lookup")
                    validate_remote_path(item_id)
                    status, details, _ = request_json(client, "GET",
                        f"{GRAPH_ROOT}/drives/{drive}/items/{quote(item_id, safe='')}"
                        "?$select=parentReference,name")
                    if status != 200:
                        raise ConfigError(f"path lookup failed with HTTP {status}")
                    parent = (details.get("parentReference") or {}).get("path")
                if not isinstance(parent, str) or "root:" not in parent:
                    raise ConfigError("parent path cannot be resolved")
                parent = parent.split("root:", 1)[1]
                if parent.startswith("/"):
                    parent = parent[1:]
                parent = validate_remote_path(parent, allow_empty=True)
                name = validate_remote_path(details.get("name"))
                if "/" in name:
                    raise ConfigError("remote item name must be one path segment")
                path = combine_remote_path(parent, name)
                if root_path and not path.startswith(root_path + "/"):
                    if outside_root is not None:
                        outside_root.append(path)
                    continue
            except (RuntimeError, ValueError, TypeError, AttributeError) as exc:
                reason = {"item": ascii(item.get("id") or item.get("name")), "error": str(exc)}
                if skipped is not None:
                    skipped.append(reason)
                say(f"warning: skipping content-search hit {reason['item']}: {exc}", error=True)
                continue
            results.append(
                {
                    "name": item.get("name"),
                    "path": path,
                    "size": int(item.get("size") or 0),
                    "modified": item.get("lastModifiedDateTime"),
                    "modifier": modifier_name(item, redact_users),
                    "matched_keyword": keyword,
                }
            )
    return results


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Inventory files in 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(
        "--keywords",
        action="append",
        help="Repeat the flag or provide comma-separated terms.",
    )
    parser.add_argument("--content-search", action="store_true",
                        help="Print content matches in text or JSON output.")
    recent = parser.add_mutually_exclusive_group()
    recent.add_argument("--recent-days", type=non_negative_int)
    recent.add_argument("--since", type=parse_since)
    parser.add_argument("--max-depth", type=non_negative_int)
    parser.add_argument("--out", type=Path)
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--redact-users", action="store_true")
    parser.add_argument(
        "--allow-partial",
        action="store_true",
        help=(
            "exit 0 even when folders were skipped; the output still reports "
            "incomplete: true and the skipped folders (default: exit 2)"
        ),
    )
    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:
    parser = build_parser()
    args = parser.parse_args(argv)
    keywords = parse_keywords(args.keywords)
    if args.content_search and any("'" in keyword for keyword in keywords):
        parser.error("--keywords cannot contain a single quote with --content-search")
    if args.content_search and not keywords:
        parser.error("--content-search requires at least one --keywords value")

    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))
        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,
        )
        root_path = config["root_path"]

        files, skipped_folders = walk_files(
            client, str(drive_id), root_path, args.max_depth, throttle
        )
        since = args.since
        if since is None:
            recent_days = 120 if args.recent_days is None else args.recent_days
            since = datetime.now(timezone.utc) - timedelta(days=recent_days)

        inventory: list[dict[str, Any]] = []
        for file_item in files:
            modified_raw = file_item.get("modified")
            try:
                modified = parse_since(str(modified_raw))
            except argparse.ArgumentTypeError:
                modified = None
            if modified is not None and modified < since:
                continue
            score, matches = keyword_rank(str(file_item["path"]), keywords)
            inventory.append(
                {
                    "name": file_item["name"],
                    "path": file_item["path"],
                    "size": file_item["size"],
                    "modified": modified_raw,
                    "modifier": modifier_name(
                        {"lastModifiedBy": file_item.get("modifier")},
                        args.redact_users,
                    ),
                    **({"rank": score, "keyword_matches": matches} if keywords else {}),
                }
            )
        if keywords:
            inventory.sort(key=lambda item: (-int(item["rank"]), str(item["path"])))
        else:
            inventory.sort(key=lambda item: str(item["path"]))

        skipped_hits: list[dict[str, str]] = []
        outside_root: list[str] = []
        report = {
            "drive": str(drive_id)[:8] + "...",
            "root_path": root_path,
            "file_count": len(inventory),
            "files": inventory,
            "content_search": (
                content_search(
                    client,
                    str(drive_id),
                    root_path,
                    keywords,
                    args.redact_users,
                    skipped_hits,
                    outside_root,
                )
                if args.content_search
                else []
            ),
        }
        if skipped_folders:
            report["incomplete"] = True
            report["skipped_folders"] = skipped_folders
        report["outside_root"] = len(outside_root)
        report["skipped_content_hits"] = skipped_hits
        report["skipped_content_hit_count"] = len(skipped_hits)
        if skipped_hits:
            report["incomplete"] = True
        rendered = json.dumps(report, indent=2, ensure_ascii=False)
        if args.out:
            write_new_text(args.out, rendered + "\n")
        if args.json:
            print(rendered)
        else:
            for item in inventory:
                rank = f" rank={item['rank']}" if "rank" in item else ""
                modifier = item.get("modifier") or "unknown"
                say(
                    f"{item['modified']} {item['size']} {item['path']} "
                    f"modifier={modifier}{rank}"
                )
            for hit in report["content_search"]:
                say(f"Content match: {hit['path']} keyword={hit['matched_keyword']}")
            if outside_root:
                say(f"Content hits outside root: {len(outside_root)}")
            if skipped_hits:
                say(f"Content hits skipped: {len(skipped_hits)} (see stderr).")
            say(f"Files: {len(inventory)}")
            if skipped_folders:
                say(
                    "Incomplete inventory: "
                    f"{len(skipped_folders)} folder(s) skipped "
                    "(see stderr for details)."
                )
        return 0 if not (skipped_folders or skipped_hits) or args.allow_partial else 2
    except (CredentialError, ConfigError) as exc:
        say(f"error: {exc}", error=True)
        return 1
    except LocalOutputError as exc:
        say(f"error: local output failed: {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