scripts/mirror_dataroom.py — Data Room Toolkit

#!/usr/bin/env python3
# Data Room Toolkit 0.1.5
# By Zero Width Lab (zerowidthlab.com)
"""Mirror a local directory one way into a SharePoint document library."""
from __future__ import annotations
import argparse
import fnmatch
import hashlib
import json
import math
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import quote
import graph_client
__author__ = "Zero Width Lab"
GRAPH_ROOT = "https://graph.microsoft.com/v1.0"
DEFAULT_THRESHOLD = 4 * 1024 * 1024
DEFAULT_CHUNK_SIZE = 5 * 1024 * 1024
OFFICE_EXTENSIONS = {".docx", ".xlsx", ".pptx"}
SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
def positive_int(value: str) -> int:
  try:
    parsed = int(value)
  except ValueError as exc:
    raise argparse.ArgumentTypeError(
      "must be an integer greater than zero"
    ) from exc
  if parsed <= 0:
    raise argparse.ArgumentTypeError("must be an integer greater than zero")
  return parsed
def non_negative_int(value: str) -> int:
  try:
    parsed = int(value)
  except ValueError as exc:
    raise argparse.ArgumentTypeError("must be an integer at least zero") from exc
  if parsed < 0:
    raise argparse.ArgumentTypeError("must be an integer at least zero")
  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 at least zero"
    ) from exc
  if not math.isfinite(parsed) or parsed < 0:
    raise argparse.ArgumentTypeError("must be a finite number at least zero")
  return parsed
def sha256_hex(value: str) -> str:
  if len(value) != 64:
    raise argparse.ArgumentTypeError("must be a 64-character SHA-256 hex digest")
  try:
    int(value, 16)
  except ValueError as exc:
    raise argparse.ArgumentTypeError(
      "must be a 64-character SHA-256 hex digest"
    ) from exc
  return value.lower()
def say(text: Any = "", *, error: bool = False) -> None:
  """Print one human-readable line; untrusted text in it is escaped.
    Every text-mode line passes through graph_client.display_text, so a local
    path, remote name, eTag, id or error message containing a line break or
    other control character cannot start a new, forged output line.
    """
  print(graph_client.display_text(text), file=sys.stderr if error else sys.stdout)
def utc_now() -> str:
  return datetime.now(timezone.utc).isoformat()
def sha256_file(path: Path) -> str:
  digest = hashlib.sha256()
  with path.open("rb") as source:
    for block in iter(lambda: source.read(1024 * 1024), b""):
      digest.update(block)
  return digest.hexdigest()
def sha256_bytes(value: bytes) -> str:
  return hashlib.sha256(value).hexdigest()
def normalize_library_path(*parts: str | None) -> str:
  """Join safe relative library paths without allowing parent traversal."""
  return "/".join(
    graph_client.validate_remote_path(part, allow_empty=True)
    for part in parts if part is not None and part != ""
  )
def selected(
  relative_path: str, includes: Iterable[str], excludes: Iterable[str]
) -> bool:
  name = relative_path.rsplit("/", 1)[-1]
  include_patterns = list(includes)
  exclude_patterns = list(excludes)
  if include_patterns and not any(
    fnmatch.fnmatchcase(relative_path, pattern)
    or fnmatch.fnmatchcase(name, pattern)
    for pattern in include_patterns
  ):
    return False
  return not any(
    fnmatch.fnmatchcase(relative_path, pattern)
    or fnmatch.fnmatchcase(name, pattern)
    for pattern in exclude_patterns
  )
def inventory_files(
  source: Path, includes: Iterable[str], excludes: Iterable[str]
) -> list[tuple[Path, str]]:
  files: list[tuple[Path, str]] = []
  for path in sorted(source.rglob("*")):
    if path.is_symlink() or not path.is_file():
      continue
    relative = path.relative_to(source).as_posix()
    if selected(relative, includes, excludes):
      files.append((path, relative))
  return files
PLAN_FORMAT = "sharepoint-dataroom-plan-v4"
PLAN_KEYS = (
  "plan_format",
  "target",
  "options",
  "destination_root",
  "folders_to_create",
  "files",
)
LOCAL_ENTRY_KEYS = (
  "source_path",
  "size_bytes",
  "sha256",
  "destination_path",
)
ACTION_CREATE = "create"
ACTION_REPLACE = "replace"
ACTION_REFUSE = "refuse"
def plan_options(args: argparse.Namespace) -> dict[str, Any]:
  return {
    "verify": args.verify,
    "small_threshold_bytes": args.small_threshold,
    "chunk_size_bytes": args.chunk_size,
  }
def local_entries(
  files: Iterable[tuple[Path, str]], destination_root: str, small_threshold: int
) -> list[dict[str, Any]]:
  """Describe each selected local file: identity, size, hash and destination."""
  entries = []
  for path, relative in files:
    size = path.stat().st_size
    entries.append(
      {
        "source_path": relative,
        "size_bytes": size,
        "sha256": sha256_file(path),
        "destination_path": normalize_library_path(destination_root, relative),
        "upload_method": graph_client.upload_method(size, small_threshold),
      }
    )
  entries.sort(key=lambda entry: entry["source_path"])
  return entries
def build_plan(
  *,
  target: dict[str, Any],
  options: dict[str, Any],
  destination_root: str,
  entries: list[dict[str, Any]],
  folders_to_create: list[str] | None = None,
) -> dict[str, Any]:
  """Build the deterministic plan that a human approves.
    The plan binds the resolved target library (drive and site identity plus the
    destination path), the operational options, every destination folder that
    a live run would create (parents before children), and for every selected
    file its local identity (path, size, SHA-256) and the approved target
    action: "create" when nothing exists at the destination path, or "replace"
    together with the existing remote item's id and eTag.
    """
  return {
    "plan_format": PLAN_FORMAT,
    "target": target,
    "options": options,
    "destination_root": destination_root,
    "folders_to_create": sorted(folders_to_create or []),
    "files": entries,
  }
def local_projection(plan: dict[str, Any]) -> dict[str, Any]:
  """Return the part of a plan that can be checked without network access."""
  files = plan.get("files") if isinstance(plan.get("files"), list) else []
  return {
    "options": plan.get("options"),
    "destination_root": plan.get("destination_root"),
    "files": [
      {key: entry.get(key) for key in LOCAL_ENTRY_KEYS}
      if isinstance(entry, dict)
      else entry
      for entry in files
    ],
  }
def plan_digest(plan: dict[str, Any]) -> str:
  """Return a stable SHA-256 digest for a canonical plan structure."""
  canonical = json.dumps(
    plan,
    sort_keys=True,
    separators=(",", ":"),
    ensure_ascii=False,
  ).encode("utf-8")
  return sha256_bytes(canonical)
def plan_document(plan: dict[str, Any], digest: str) -> dict[str, Any]:
  return {**plan, "plan_digest": digest}
def write_plan(path: Path, plan: dict[str, Any], digest: str) -> None:
  try:
    graph_client.write_new_text(
      path, json.dumps(plan_document(plan, digest), indent=2) + "\n"
    )
  except (OSError, graph_client.GraphClientError) as exc:
    raise RuntimeError(f"could not write plan file: {exc}") from exc
def read_plan(path: Path) -> tuple[str, dict[str, Any] | None]:
  try:
    document = json.loads(path.read_text(encoding="utf-8"))
  except FileNotFoundError as exc:
    raise RuntimeError(f"plan file not found: {path}") from exc
  except (OSError, json.JSONDecodeError) as exc:
    raise RuntimeError(f"could not read plan file: {exc}") from exc
  if not isinstance(document, dict):
    raise RuntimeError("plan file must contain a JSON object")
  digest = document.get("plan_digest")
  if not isinstance(digest, str) or not digest:
    raise RuntimeError("plan file does not contain a plan_digest")
  if document.get("plan_format") != PLAN_FORMAT:
    raise RuntimeError(
      "plan file was not written by this version; rerun the dry run to "
      "create a new plan"
    )
  approved = {key: document.get(key) for key in PLAN_KEYS}
  if (
    not isinstance(approved["target"], dict)
    or not isinstance(approved["options"], dict)
    or not isinstance(approved["destination_root"], str)
    or not isinstance(approved["folders_to_create"], list)
    or not all(isinstance(path, str) for path in approved["folders_to_create"])
    or not isinstance(approved["files"], list)
  ):
    raise RuntimeError("plan file does not contain a valid plan")
  if plan_digest(approved) != digest:
    raise RuntimeError("plan file digest does not match its plan contents")
  return digest, approved
def _describe_entry(entry: dict[str, Any]) -> str:
  parts = [
    f"size={entry.get('size_bytes')}",
    f"sha256={entry.get('sha256')}",
    f"destination={entry.get('destination_path')}",
    f"method={entry.get('upload_method')}",
    f"action={entry.get('action')}",
  ]
  if entry.get("action") == ACTION_REPLACE:
    parts.append(f"remote_item={safe_identifier(entry.get('remote_item_id'))}")
    parts.append(f"remote_etag={entry.get('remote_etag')}")
  return " ".join(parts)
def plan_diff(
  approved: dict[str, Any] | None, current: dict[str, Any]
) -> list[str]:
  if approved is None:
    return [
      "Added files: unavailable without the approved plan file",
      "Removed files: unavailable without the approved plan file",
      "Changed files: unavailable without the approved plan file",
      "Use --plan-file to retain file-level diff details.",
    ]
  old_entries = {
    entry["source_path"]: entry
    for entry in approved.get("files", [])
    if isinstance(entry, dict) and isinstance(entry.get("source_path"), str)
  }
  new_entries = {
    entry["source_path"]: entry
    for entry in current.get("files", [])
    if isinstance(entry, dict) and isinstance(entry.get("source_path"), str)
  }
  lines: list[str] = []
  old_target = approved.get("target")
  new_target = current.get("target")
  if new_target is not None and old_target != new_target:
    old_target = old_target if isinstance(old_target, dict) else {}
    lines.append(
      "Target changed: "
      f"old drive={safe_identifier(old_target.get('drive_id'))} "
      f"site={safe_identifier(old_target.get('site_id'))} "
      f"path={old_target.get('destination_path')}; "
      f"new drive={safe_identifier(new_target.get('drive_id'))} "
      f"site={safe_identifier(new_target.get('site_id'))} "
      f"path={new_target.get('destination_path')}"
    )
  if approved.get("options") != current.get("options"):
    lines.append(
      f"Options changed: old={approved.get('options')}; "
      f"new={current.get('options')}"
    )
  added = sorted(set(new_entries) - set(old_entries))
  removed = sorted(set(old_entries) - set(new_entries))
  changed = sorted(
    path
    for path in set(old_entries) & set(new_entries)
    if old_entries[path] != new_entries[path]
  )
  old_folders = approved.get("folders_to_create")
  new_folders = current.get("folders_to_create")
  if new_folders is not None and old_folders != new_folders:
    lines.append(
      "Folders to create changed: "
      f"old={', '.join(old_folders or []) or 'none'}; "
      f"new={', '.join(new_folders) or 'none'}"
    )
  lines.append("Added files: " + (", ".join(added) if added else "none"))
  lines.append("Removed files: " + (", ".join(removed) if removed else "none"))
  if not changed:
    lines.append("Changed files: none")
  else:
    lines.append("Changed files:")
    for path in changed:
      lines.append(
        f"  {path}: old {_describe_entry(old_entries[path])}; "
        f"new {_describe_entry(new_entries[path])}"
      )
  if approved.get("destination_root") != current.get("destination_root"):
    lines.append(
      "Destination root changed: "
      f"old={approved.get('destination_root')}; "
      f"new={current.get('destination_root')}"
    )
  return lines
def report_plan_mismatch(
  approved_digest: str | None,
  approved_plan: dict[str, Any] | None,
  current_plan: dict[str, Any],
  current_digest: str | None,
) -> None:
  current = current_digest or "not computed (local plan differs before network access)"
  say(
    f"error: plan digest mismatch: approved {approved_digest}; "
    f"current {current}. Nothing was written.",
    error=True,
  )
  for line in plan_diff(approved_plan, current_plan):
    say(line, error=True)
def safe_identifier(value: str | None) -> str:
  if not value:
    return ""
  return f"{value[:8]}..." if len(value) > 8 else value
def safe_error(
  error: BaseException,
  *,
  token: str | None = None,
  upload_url: str | None = None,
  identifiers: Iterable[str | None] = (),
) -> str:
  message = str(error)
  sensitive_values = [
    token,
    upload_url,
    os.environ.get("GRAPH_TENANT_ID"),
    os.environ.get("GRAPH_CLIENT_ID"),
    os.environ.get("GRAPH_CLIENT_SECRET"),
  ]
  for value in sensitive_values:
    if value:
      message = message.replace(value, "<redacted>")
  for identifier in identifiers:
    if identifier:
      for candidate in {identifier, quote(identifier, safe="")}:
        message = message.replace(candidate, safe_identifier(identifier))
  return message.replace("Authorization", "<redacted-header>")
def graph_item_content_url(drive_id: str, library_path: str) -> str:
  library_path = normalize_library_path(library_path)
  drive_segment = quote(drive_id, safe="")
  return (
    f"{GRAPH_ROOT}/drives/{drive_segment}/root:/"
    f"{graph_client.encode_path_segments(library_path)}:/content"
  )
def graph_item_url(drive_id: str, library_path: str) -> str:
  library_path = normalize_library_path(library_path)
  drive_segment = quote(drive_id, safe="")
  if not library_path:
    return f"{GRAPH_ROOT}/drives/{drive_segment}/root?$select=id,name,folder"
  return (
    f"{GRAPH_ROOT}/drives/{drive_segment}/root:/"
    f"{graph_client.encode_path_segments(library_path)}"
    "?$select=id,name,folder"
  )
def graph_item_state_url(drive_id: str, library_path: str) -> str:
  library_path = normalize_library_path(library_path)
  drive_segment = quote(drive_id, safe="")
  return (
    f"{GRAPH_ROOT}/drives/{drive_segment}/root:/"
    f"{graph_client.encode_path_segments(library_path)}"
    "?$select=id,eTag,folder,file"
  )
def graph_drive_url(drive_id: str) -> str:
  return f"{GRAPH_ROOT}/drives/{quote(drive_id, safe='')}?$select=id,sharePointIds"
def graph_children_url(drive_id: str, parent_path: str) -> str:
  parent_path = normalize_library_path(parent_path)
  drive_segment = quote(drive_id, safe="")
  if not parent_path:
    return f"{GRAPH_ROOT}/drives/{drive_segment}/root/children"
  return (
    f"{GRAPH_ROOT}/drives/{drive_segment}/root:/"
    f"{graph_client.encode_path_segments(parent_path)}:/children"
  )
def request_options(
  args: argparse.Namespace, token: str | None = None
) -> dict[str, Any]:
  options: dict[str, Any] = {
    "timeout": args.request_timeout,
    "max_retries": args.max_retries,
    "throttle": args.throttle,
  }
  if token is not None:
    options["token"] = token
  return options
def resolve_target(
  args: argparse.Namespace,
  token: str,
  config: dict[str, Any],
  destination_root: str,
) -> dict[str, Any]:
  """Resolve the target library with read-only Graph calls.
    The drive is resolved from ``drive_id`` or the site's default drive, then
    read back so that the plan records the drive id Graph reports and the
    SharePoint site id that owns it.
    """
  drive_id = graph_client.resolve_drive_id(
    token,
    drive_id=config.get("drive_id"),
    site_id=config.get("site_id"),
    timeout=args.request_timeout,
    max_retries=args.max_retries,
    throttle=args.throttle,
  )
  status, payload, _ = graph_client.request(
    "GET", graph_drive_url(drive_id), **request_options(args, token)
  )
  if status != 200 or not isinstance(payload, dict):
    raise RuntimeError(f"target drive lookup returned HTTP {status}")
  resolved_drive = payload.get("id")
  if not isinstance(resolved_drive, str) or not resolved_drive:
    raise RuntimeError("target drive lookup did not return a drive id")
  sharepoint_ids = payload.get("sharePointIds")
  site_id = (
    sharepoint_ids.get("siteId") if isinstance(sharepoint_ids, dict) else None
  )
  return {
    "drive_id": resolved_drive,
    "site_id": site_id if isinstance(site_id, str) and site_id else None,
    "destination_path": destination_root,
  }
def remote_action(
  args: argparse.Namespace, token: str, drive_id: str, destination: str
) -> dict[str, Any]:
  """Return the target action for one destination path (read-only GET)."""
  status, payload, _ = graph_client.request(
    "GET",
    graph_item_state_url(drive_id, destination),
    **request_options(args, token),
  )
  if status == 404:
    return {"action": ACTION_CREATE}
  if status == 200 and isinstance(payload, dict):
    if "folder" in payload:
      raise RuntimeError(
        f"destination path is an existing folder: {destination}"
      )
    item_id = payload.get("id")
    etag = payload.get("eTag")
    if not isinstance(item_id, str) or not item_id:
      raise RuntimeError(f"existing destination item has no id: {destination}")
    if not isinstance(etag, str) or not etag:
      raise RuntimeError(
        f"existing destination item has no eTag: {destination}"
      )
    return {
      "action": ACTION_REPLACE,
      "remote_item_id": item_id,
      "remote_etag": etag,
    }
  raise RuntimeError(
    f"destination check returned HTTP {status} for destination path {destination}"
  )
def required_folder(destination_root: str, source_path: str) -> str:
  """Return the destination folder that must exist for one source file."""
  return normalize_library_path(destination_root, source_path.rpartition("/")[0])
def folder_exists(
  args: argparse.Namespace, token: str, drive_id: str, folder_path: str
) -> bool:
  """Report whether a destination folder exists (read-only GET)."""
  status, payload, _ = graph_client.request(
    "GET",
    graph_item_url(drive_id, folder_path),
    **request_options(args, token),
  )
  if status == 200:
    if not isinstance(payload, dict) or "folder" not in payload:
      raise RuntimeError(f"destination path is not a folder: {folder_path}")
    return True
  if status == 404:
    return False
  raise RuntimeError(
    f"folder check returned HTTP {status} for destination path {folder_path}"
  )
def missing_folders(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  folders: Iterable[str],
  destination_root: str = "",
) -> list[str]:
  """Inspect only the selected destination and folders beneath it."""
  root = normalize_library_path(destination_root)
  if not folder_exists(args, token, drive_id, root):
    raise RuntimeError("selected destination folder must already exist: " + root)
  known: dict[str, bool] = {root: True}
  for folder in sorted(set(folders)):
    folder = normalize_library_path(folder)
    if folder != root and root and not folder.startswith(root + "/"):
      raise UnapprovedFolder("folder is outside selected destination: " + folder)
    current = folder
    while current not in known:
      known[current] = folder_exists(args, token, drive_id, current)
      if known[current]:
        break
      current = current.rpartition("/")[0]
  return sorted(path for path, exists in known.items() if not exists)
def compute_plan(
  args: argparse.Namespace,
  token: str,
  config: dict[str, Any],
  source: Path,
  destination_root: str,
) -> tuple[dict[str, Any], list[tuple[Path, str]]]:
  """Inventory the source and resolve target, folder and per-file actions.
    Every request made here is a read-only GET.
    """
  files = inventory_files(source, args.include, args.exclude)
  entries = local_entries(files, destination_root, args.small_threshold)
  target = resolve_target(args, token, config, destination_root)
  folders = missing_folders(
    args,
    token,
    target["drive_id"],
    (required_folder(destination_root, entry["source_path"]) for entry in entries),
    destination_root,
  )
  for entry in entries:
    entry.update(
      remote_action(args, token, target["drive_id"], entry["destination_path"])
    )
    if entry["action"] == ACTION_REPLACE:
      entry["upload_method"] = "chunked"
      if entry["size_bytes"] == 0:
        entry["action"] = ACTION_REFUSE
        entry["planning_error"] = (
          "zero-byte replacement refused: guarded upload sessions cannot "
          "carry an empty byte range; no replacement will be attempted"
        )
  plan = build_plan(
    target=target,
    options=plan_options(args),
    destination_root=destination_root,
    entries=entries,
    folders_to_create=folders,
  )
  return plan, files
class UnapprovedFolder(RuntimeError):
  """A live run tried to create a folder that the approved plan does not list."""
def create_planned_folders(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  folder_path: str,
  approved_folders: set[str],
  created_folders: set[str],
  created: list[str] | None = None,
  destination_root: str = "",
) -> list[str]:
  """Create the approved missing folders on the path to one destination folder.
    Only folders listed in the approved plan are created, parents first, each
    at most once per run. Folders not listed were confirmed to exist during
    live revalidation and are not touched. Each confirmed creation is appended
    to ``created`` at once, so it stays recorded if a later request fails.
    """
  root = normalize_library_path(destination_root)
  for path in approved_folders:
    if path == root or (root and not path.startswith(root + "/")):
      raise UnapprovedFolder("approved folder is outside selected destination: " + path)
    normalize_library_path(path)
  normalized = normalize_library_path(folder_path)
  if normalized != root and root and not normalized.startswith(root + "/"):
    raise UnapprovedFolder("folder is outside selected destination: " + normalized)
  segments = normalized.split("/") if normalized else []
  created = [] if created is None else created
  for depth in range(1, len(segments) + 1):
    path = "/".join(segments[:depth])
    if path not in approved_folders or path in created_folders:
      continue
    create_folder(args, token, drive_id, path, approved_folders)
    created_folders.add(path)
    created.append(path)
  return created
def create_folder(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  folder_path: str,
  approved_folders: set[str],
) -> None:
  """Create one folder that the approved plan lists (fails if it exists)."""
  if folder_path not in approved_folders:
    raise UnapprovedFolder(
      f"refusing to create a folder the approved plan does not list: {folder_path}"
    )
  parent, _, name = folder_path.rpartition("/")
  create_status, create_payload, _ = graph_client.request(
    "POST",
    graph_children_url(drive_id, parent),
    json_body={
      "name": name,
      "folder": {},
      "@microsoft.graph.conflictBehavior": "fail",
    },
    write=True,
    **request_options(args, token),
  )
  if create_status in {200, 201}:
    if not isinstance(create_payload, dict) or "folder" not in create_payload:
      raise RuntimeError(
        f"folder creation returned a non-folder for destination path {folder_path}"
      )
    return
  raise RuntimeError(
    f"folder creation returned HTTP {create_status} for destination path "
    f"{folder_path}; the approved plan listed it as missing"
  )
class RemoteFileChanged(RuntimeError):
  """The destination changed before commit; cancellation is recorded."""
  def __init__(self, message: str, session_cancel: dict[str, Any]):
    super().__init__(message)
    self.session_cancel = session_cancel
class LocalFileChanged(RuntimeError):
  """The local bytes no longer match the approved plan."""
  def __init__(self, message: str, session_cancel: dict[str, Any] | None = None):
    super().__init__(message)
    self.session_cancel = session_cancel
def check_local_integrity(local_path: Path, planned: dict[str, Any]) -> None:
  """Refuse unless the local file still has the approved size and SHA-256."""
  try:
    size = local_path.stat().st_size
    digest = sha256_file(local_path)
  except OSError as exc:
    raise LocalFileChanged(
      f"local file could not be read after approval: {planned['source_path']}"
    ) from exc
  if size != planned["size_bytes"] or digest != planned["sha256"]:
    raise LocalFileChanged(
      f"local file changed after approval: {planned['source_path']}"
    )
def cancel_upload_session(
  args: argparse.Namespace, token: str, upload_url: str
) -> dict[str, Any]:
  """Best-effort DELETE of a Graph-issued upload session; never raises."""
  try:
    status, _, _ = graph_client.request(
      "DELETE",
      upload_url,
      graph_issued_url=True,
      write=True,
      **request_options(args),
    )
  except Exception as exc:
    return {
      "status": "error",
      "detail": safe_error(exc, token=token, upload_url=upload_url),
    }
  return {
    "status": "cancelled" if status in {200, 204} else "not-confirmed",
    "http_status": status,
  }
def write_headers(entry: dict[str, Any]) -> dict[str, str]:
  """Return headers that make Graph enforce the approved replace action."""
  if entry.get("action") == ACTION_REPLACE:
    return {"If-Match": str(entry["remote_etag"])}
  return {}
def conflict_behavior(entry: dict[str, Any]) -> str:
  """Create must fail on an existing item; replace is guarded by If-Match."""
  return "replace" if entry.get("action") == ACTION_REPLACE else "fail"
def upload_small(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  local_path: Path,
  entry: dict[str, Any],
) -> tuple[int, Any | None]:
  if entry.get("action") == ACTION_REPLACE:
    return upload_large(args, token, drive_id, local_path, entry)
  data = local_path.read_bytes()
  if len(data) != entry["size_bytes"] or sha256_bytes(data) != entry["sha256"]:
    raise LocalFileChanged(
      f"local file changed after approval: {entry['source_path']}"
    )
  url = graph_item_content_url(drive_id, entry["destination_path"])
  url += f"?@microsoft.graph.conflictBehavior={conflict_behavior(entry)}"
  status, payload, _ = graph_client.request(
    "PUT",
    url,
    headers={
      "Content-Type": graph_client.mime_type_for(local_path),
      **write_headers(entry),
    },
    data=data,
    write=True,
    **request_options(args, token),
  )
  return status, payload
def upload_large(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  local_path: Path,
  entry: dict[str, Any],
) -> tuple[int, Any | None]:
  graph_client.validate_upload_chunk(args.chunk_size)
  destination = normalize_library_path(entry["destination_path"])
  total = entry["size_bytes"]
  if total == 0:
    raise RuntimeError("zero-byte upload sessions are unsupported; no file was written")
  if local_path.stat().st_size != total:
    raise LocalFileChanged(
      f"local file changed after approval: {entry['source_path']}"
    )
  drive_segment = quote(drive_id, safe="")
  session_url = (
    f"{GRAPH_ROOT}/drives/{drive_segment}/root:/"
    f"{graph_client.encode_path_segments(destination)}"
    ":/createUploadSession"
  )
  status, payload, _ = graph_client.request(
    "POST",
    session_url,
    headers=write_headers(entry),
    json_body={
      "item": {"@microsoft.graph.conflictBehavior": conflict_behavior(entry)}
    },
    write=True,
    **request_options(args, token),
  )
  if status not in {200, 201} or not isinstance(payload, dict):
    return status, payload
  upload_url = payload.get("uploadUrl")
  if not isinstance(upload_url, str) or not upload_url:
    raise RuntimeError("upload session response did not include an upload URL")
  final_status = status
  final_payload: Any | None = payload
  running_digest = hashlib.sha256()
  with local_path.open("rb") as source:
    offset = 0
    while offset < total:
      chunk = source.read(args.chunk_size)
      if not chunk:
        break
      end = offset + len(chunk) - 1
      running_digest.update(chunk)
      if end + 1 > total or (
        end + 1 == total and running_digest.hexdigest() != entry["sha256"]
      ):
        raise LocalFileChanged(
          f"local file changed after approval: {entry['source_path']}",
          session_cancel=cancel_upload_session(args, token, upload_url),
        )
      if end + 1 == total and entry.get("action") == ACTION_REPLACE:
        try:
          check_status, state, _ = graph_client.request(
            "GET", graph_item_state_url(drive_id, entry["destination_path"]),
            **request_options(args, token),
          )
          if (check_status != 200 or not isinstance(state, dict)
              or state.get("eTag") != entry["remote_etag"]
              or state.get("id") != entry["remote_item_id"]):
            raise RuntimeError("destination item changed before final byte range")
        except Exception as exc:
          raise RemoteFileChanged(
            safe_error(exc, token=token, identifiers=(drive_id,)),
            cancel_upload_session(args, token, upload_url),
          ) from None
      try:
        final_status, final_payload, _ = graph_client.request(
          "PUT",
          upload_url,
          headers={
            "Content-Length": str(len(chunk)),
            "Content-Range": f"bytes {offset}-{end}/{total}",
          },
          data=chunk,
          graph_issued_url=True,
          write=True,
          **{**request_options(args), "max_retries": (
            0 if end + 1 == total and entry.get("action") == ACTION_REPLACE
            else args.max_retries
          )},
        )
      except Exception as exc:
        unknown = isinstance(exc, graph_client.WriteOutcomeUnknown)
        raise (graph_client.WriteOutcomeUnknown if unknown else RuntimeError)(
          safe_error(exc, token=token, upload_url=upload_url)
        ) from None
      if final_status not in {200, 201, 202}:
        return final_status, final_payload
      offset = end + 1
  if offset != total:
    raise LocalFileChanged(
      f"local file changed after approval: {entry['source_path']}",
      session_cancel=cancel_upload_session(args, token, upload_url),
    )
  return final_status, final_payload
def write_refusal_detail(status: int, entry: dict[str, Any]) -> str:
  if status == 409 and entry.get("action") == ACTION_CREATE:
    return (
      "upload returned HTTP 409: an item now exists at the destination, "
      "but the approved action was create; nothing was replaced"
    )
  if status == 412 and entry.get("action") == ACTION_REPLACE:
    return (
      "upload returned HTTP 412: the destination item changed after "
      "approval (eTag no longer matches); nothing was replaced"
    )
  return f"upload returned HTTP {status}"
def verification_result(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  local_path: Path,
  destination: str,
  remote_metadata: Any | None,
  local_sha256: str | None,
) -> dict[str, Any]:
  if args.verify == "none":
    return {"mode": "none", "status": "skipped"}
  local_size = local_path.stat().st_size
  office_file = local_path.suffix.lower() in OFFICE_EXTENSIONS
  if args.verify == "size":
    remote_size = (
      remote_metadata.get("size") if isinstance(remote_metadata, dict) else None
    )
    passed = remote_size == local_size
    if passed:
      state = "passed"
    elif office_file:
      state = "informational"
    else:
      state = "failed"
    result = {
      "mode": "size",
      "status": state,
      "local_size_bytes": local_size,
      "remote_size_bytes": remote_size,
    }
    if state == "informational":
      result["detail"] = "Office files may change size when saved by SharePoint"
    return result
  status, content, _ = graph_client.request(
    "GET",
    graph_item_content_url(drive_id, destination),
    expect_json=False,
    graph_issued_url=True,
    **request_options(args, token),
  )
  if status != 200 or not isinstance(content, bytes):
    return {
      "mode": "sha256",
      "status": "failed",
      "detail": f"content verification returned HTTP {status}",
    }
  remote_sha256 = sha256_bytes(content)
  local_digest = local_sha256 or sha256_file(local_path)
  if remote_sha256 == local_digest:
    state = "passed"
  elif office_file:
    state = "informational"
  else:
    state = "failed"
  result = {
    "mode": "sha256",
    "status": state,
    "local_sha256": local_digest,
    "remote_sha256": remote_sha256,
  }
  if state == "informational":
    result["detail"] = "Office files may change when saved by SharePoint"
  return result
def upload_one(
  args: argparse.Namespace,
  token: str,
  drive_id: str,
  local_path: Path,
  planned: dict[str, Any],
  destination_root: str,
  approved_folders: set[str],
  created_folders: set[str],
) -> dict[str, Any]:
  relative_path = planned["source_path"]
  destination = planned["destination_path"]
  method = planned["upload_method"]
  entry: dict[str, Any] = {
    "source_path": relative_path,
    "destination_path": destination,
    "size_bytes": planned["size_bytes"],
    "upload_method": method,
    "action": planned["action"],
    "status": "pending",
  }
  if planned.get("planning_error"):
    entry.update(status="failed", error=planned["planning_error"], created_folders=[])
    return entry
  if planned["action"] == ACTION_REPLACE:
    entry["remote_item_id_prefix"] = safe_identifier(planned.get("remote_item_id"))
  entry["created_folders"] = []
  try:
    graph_client.validate_upload_chunk(args.chunk_size)
    check_local_integrity(local_path, planned)
    create_planned_folders(
      args,
      token,
      drive_id,
      required_folder(destination_root, relative_path),
      approved_folders,
      created_folders,
      entry["created_folders"],
      destination_root,
    )
    if method == "direct":
      status, payload = upload_small(args, token, drive_id, local_path, planned)
    else:
      status, payload = upload_large(args, token, drive_id, local_path, planned)
    entry["http_status"] = status
    if status not in {200, 201}:
      entry["status"] = "failed"
      entry["error"] = write_refusal_detail(status, planned)
      return entry
    verification = verification_result(
      args,
      token,
      drive_id,
      local_path,
      destination,
      payload,
      planned["sha256"],
    )
    entry["verification"] = verification
    entry["status"] = "failed" if verification["status"] == "failed" else "uploaded"
    if isinstance(payload, dict):
      entry["item_id_prefix"] = safe_identifier(payload.get("id"))
      entry["remote_modified"] = payload.get("lastModifiedDateTime")
    entry["uploaded_at_utc"] = utc_now()
  except Exception as exc:
    entry["status"] = "failed"
    entry["error"] = safe_error(exc, token=token, identifiers=(drive_id,))
    if isinstance(exc, graph_client.WriteOutcomeUnknown):
      entry["outcome"] = "unknown"
    session_cancel = getattr(exc, "session_cancel", None)
    if session_cancel is not None:
      entry["upload_session_cancel"] = session_cancel
  return entry
def dry_run_entry(planned: dict[str, Any], verify: str) -> dict[str, Any]:
  entry = {
    "source_path": planned["source_path"],
    "destination_path": planned["destination_path"],
    "size_bytes": planned["size_bytes"],
    "upload_method": planned["upload_method"],
    "action": planned["action"],
    "status": "dry-run",
    "verification": {"mode": verify, "status": "not-run"},
  }
  if planned.get("planning_error"):
    entry.update(status="failed", error=planned["planning_error"])
  if planned["action"] == ACTION_REPLACE:
    entry["remote_item_id_prefix"] = safe_identifier(planned.get("remote_item_id"))
  return entry
def make_audit(
  *,
  started: str,
  mode: str,
  operator: str,
  destination_root: str,
  target: dict[str, Any],
  entries: list[dict[str, Any]],
  digest: str,
  hosts_contacted: list[str],
  plan: dict[str, Any],
) -> dict[str, Any]:
  planned_files = plan.get("files", [])
  return {
    "started_at_utc": started,
    "completed_at_utc": utc_now(),
    "mode": mode,
    "operator": operator,
    "drive_id_prefix": safe_identifier(target.get("drive_id")),
    "site_id_prefix": safe_identifier(target.get("site_id")),
    "destination_path": destination_root,
    "options": plan.get("options"),
    "plan_digest": digest,
    "planned_folder_creates": list(plan.get("folders_to_create", [])),
    "planned_creates": sum(
      1 for entry in planned_files if entry.get("action") == ACTION_CREATE
    ),
    "planned_replacements": [
      entry["destination_path"]
      for entry in planned_files
      if entry.get("action") == ACTION_REPLACE
    ],
    "hosts_contacted": hosts_contacted,
    "uploads": entries,
    "failures": [entry for entry in entries if entry["status"] == "failed"],
  }
def write_audit(path: Path, audit: dict[str, Any]) -> None:
  try:
    graph_client.write_new_text(path, json.dumps(audit, indent=2) + "\n")
  except (OSError, graph_client.GraphClientError) as exc:
    raise RuntimeError(f"could not write audit log: {exc}") from exc
def build_parser() -> argparse.ArgumentParser:
  parser = argparse.ArgumentParser(
    description="Mirror a local folder one way into a SharePoint library.",
    epilog=SIGNATURE,
  )
  parser.add_argument(
    "--source", required=True, type=Path, help="local directory to mirror"
  )
  parser.add_argument(
    "--dest-path", required=True, help="destination folder in the library"
  )
  parser.add_argument("--drive-id")
  parser.add_argument("--site-id")
  parser.add_argument("--root-path")
  parser.add_argument("--config", type=Path)
  parser.add_argument("--small-threshold", type=positive_int)
  parser.add_argument("--chunk-size", type=positive_int)
  parser.add_argument(
    "--verify", choices=("sha256", "size", "none"), default="sha256"
  )
  parser.add_argument("--throttle", type=non_negative_float)
  parser.add_argument("--max-retries", type=non_negative_int)
  parser.add_argument("--request-timeout", type=non_negative_float)
  mode = parser.add_mutually_exclusive_group()
  mode.add_argument(
    "--dry-run",
    action="store_true",
    help=(
      "resolve the target library, the missing folders to create and each "
      "file's create or replace action with read-only requests and show the "
      "complete plan without writing to SharePoint (default)"
    ),
  )
  mode.add_argument(
    "--live",
    action="store_true",
    help=(
      "apply an approved plan; refuses before any write if the target, "
      "a file, the folders to create or any create/replace action changed"
    ),
  )
  parser.add_argument(
    "--yes", action="store_true", help="skip the interactive live confirmation"
  )
  parser.add_argument(
    "--audit-log",
    type=Path,
    help="create a new private JSON audit file; existing parent required, no symlinks",
  )
  parser.add_argument(
    "--plan-file",
    type=Path,
    help=(
      "create a new private dry-run plan (existing parent, no symlinks) "
      "or read an approved live plan"
    ),
  )
  parser.add_argument(
    "--plan-digest",
    type=sha256_hex,
    help="approved dry-run plan digest required for live uploads",
  )
  parser.add_argument(
    "--operator",
    default=os.environ.get("SHAREPOINT_DATAROOM_OPERATOR", "unspecified"),
  )
  parser.add_argument("--include", action="append", default=[], metavar="GLOB")
  parser.add_argument("--exclude", action="append", default=[], metavar="GLOB")
  parser.add_argument(
    "--json", action="store_true", help="emit the audit record as JSON"
  )
  return parser
def plan_lines(plan: dict[str, Any]) -> list[str]:
  """Return the complete human-readable plan.
    It lists every folder a live run would create and every file action,
    including every replacement. Callers print each line through say().
    """
  target = plan.get("target") or {}
  options = plan.get("options") or {}
  files = plan.get("files") or []
  folders = plan.get("folders_to_create") or []
  lines = [
    f"Target drive: {safe_identifier(target.get('drive_id'))}; "
    f"site: {safe_identifier(target.get('site_id')) or 'not reported'}; "
    f"destination path: {plan.get('destination_root') or '(library root)'}",
    f"Options: verify={options.get('verify')}; "
    f"direct-upload threshold={options.get('small_threshold_bytes')} bytes; "
    f"chunk size={options.get('chunk_size_bytes')} bytes",
  ]
  lines.extend(f"  create folder (missing): {folder}" for folder in folders)
  for entry in files:
    if entry.get("planning_error"):
      detail = "REFUSE: " + entry["planning_error"]
    elif entry.get("action") == ACTION_REPLACE:
      detail = (
        f"REPLACE existing item {safe_identifier(entry.get('remote_item_id'))} "
        f"eTag {entry.get('remote_etag')}"
      )
    else:
      detail = "create (no item at this path)"
    lines.append(
      f"  {detail}: {entry.get('source_path')} -> "
      f"{entry.get('destination_path')} ({entry.get('upload_method')}, "
      f"{entry.get('size_bytes')} bytes, sha256 {entry.get('sha256')})"
    )
  replacements = [
    entry.get("destination_path")
    for entry in files
    if entry.get("action") == ACTION_REPLACE
  ]
  creates = sum(entry.get("action") == ACTION_CREATE for entry in files)
  lines.append(
    f"Planned folder creates: {len(folders)}; planned creates: {creates}; "
    f"planned replacements: {len(replacements)}"
  )
  if replacements:
    lines.append(
      "Existing destination files that a live run would replace:"
    )
    lines.extend(f"  {path}" for path in replacements)
  else:
    lines.append("Existing destination files that a live run would replace: none")
  return lines
def failure_detail(entry: dict[str, Any]) -> str:
  """Return the per-file failure text printed to stderr in text mode."""
  parts = []
  if entry.get("error"):
    parts.append(str(entry["error"]))
  verification = entry.get("verification")
  if isinstance(verification, dict) and verification.get("status") == "failed":
    parts.append(
      "verification failed ("
      f"{verification.get('mode')}): "
      f"{verification.get('detail') or 'remote content does not match local'}"
    )
  cancel = entry.get("upload_session_cancel")
  if isinstance(cancel, dict):
    parts.append(f"upload session cancel: {cancel.get('status')}")
  return "; ".join(parts) or "failed"
def emit(audit: dict[str, Any], as_json: bool, plan: dict[str, Any]) -> None:
  if as_json:
    print(json.dumps(audit, indent=2))
    return
  say(f"Mode: {audit['mode']}")
  if audit["mode"] == "live" and any(item.get("planning_error") for item in plan["files"]):
    say("No file was uploaded: the plan contains refused actions")
  say(f"Destination: {audit['destination_path']}")
  say(f"Plan digest: {audit['plan_digest']}")
  if audit["mode"] == "dry-run":
    for line in plan_lines(plan):
      say(line)
  else:
    for entry in audit["uploads"]:
      say(
        f"{entry['status']} ({entry['action']}): {entry['source_path']} -> "
        f"{entry['destination_path']} ({entry['upload_method']})"
      )
      if entry["status"] == "failed":
        say(f"error: {entry['source_path']}: {failure_detail(entry)}", error=True)
  say(f"Files: {len(audit['uploads'])}; failures: {len(audit['failures'])}")
def main(argv: list[str] | None = None) -> int:
  graph_client.reset_contacted_hosts()
  parser = build_parser()
  args = parser.parse_args(argv)
  if args.live and not args.plan_digest and not args.plan_file:
    parser.error("--live requires --plan-digest or --plan-file")
  source = args.source.expanduser()
  if source.is_symlink() or not source.is_dir():
    parser.error("--source must name an existing directory")
  started = utc_now()
  try:
    config = graph_client.load_config(
      args.config,
      overrides={
        "drive_id": args.drive_id,
        "site_id": args.site_id,
        "root_path": args.root_path,
        "small_file_threshold_bytes": args.small_threshold,
        "upload_chunk_bytes": args.chunk_size,
        "throttle_seconds": args.throttle,
        "max_retries": args.max_retries,
        "request_timeout_seconds": args.request_timeout,
      },
      require_target=True,
    )
    args.small_threshold = int(
      config.get("small_file_threshold_bytes", DEFAULT_THRESHOLD)
    )
    args.chunk_size = int(config.get("upload_chunk_bytes", DEFAULT_CHUNK_SIZE))
    args.throttle = float(config.get("throttle_seconds", 0.5))
    args.max_retries = int(config.get("max_retries", 4))
    args.request_timeout = float(config.get("request_timeout_seconds", 60))
    if args.small_threshold <= 0 or args.chunk_size <= 0:
      raise ValueError(
        "upload thresholds and chunk size must be greater than zero"
      )
    if args.max_retries < 0 or args.throttle < 0 or args.request_timeout <= 0:
      raise ValueError("request settings are outside their valid range")
    destination_root = normalize_library_path(
      config.get("root_path"), args.dest_path
    )
    local_plan = {
      "options": plan_options(args),
      "destination_root": destination_root,
      "files": local_entries(
        inventory_files(source, args.include, args.exclude),
        destination_root,
        args.small_threshold,
      ),
    }
  except (
    OSError,
    ValueError,
    json.JSONDecodeError,
    graph_client.GraphClientError,
  ) as exc:
    say(f"error: {safe_error(exc)}", error=True)
    return 1
  mode = "live" if args.live else "dry-run"
  operator = args.operator or "unspecified"
  if mode == "live" and not args.yes and not sys.stdin.isatty():
    say(
      "error: --live requires --yes when standard input is not interactive",
      error=True,
    )
    return 1
  approved_plan: dict[str, Any] | None = None
  approved_digest = args.plan_digest
  if args.plan_file and mode == "live" and approved_digest is None:
    try:
      file_digest, approved_plan = read_plan(args.plan_file)
    except RuntimeError as exc:
      say(f"error: {exc}", error=True)
      return 1
    approved_digest = file_digest
  if approved_plan is not None and local_projection(approved_plan) != local_projection(
    local_plan
  ):
    report_plan_mismatch(
      approved_digest, local_projection(approved_plan), local_projection(local_plan), None
    )
    return 1
  identifiers = (config.get("drive_id"), config.get("site_id"))
  def resolve_plan() -> tuple[dict[str, Any], list[tuple[Path, str]], str]:
    plan, plan_files = compute_plan(args, token, config, source, destination_root)
    return plan, plan_files, plan_digest(plan)
  try:
    token = graph_client.acquire_token(
      timeout=args.request_timeout,
      max_retries=args.max_retries,
      throttle=args.throttle,
    )
    current_plan, files, current_digest = resolve_plan()
  except (graph_client.GraphClientError, RuntimeError, OSError, ValueError) as exc:
    say(f"error: {safe_error(exc, identifiers=identifiers)}", error=True)
    return 1
  if mode == "live" and approved_digest != current_digest:
    report_plan_mismatch(
      approved_digest, approved_plan, current_plan, current_digest
    )
    return 1
  planning_refusals = any(entry.get("planning_error") for entry in current_plan["files"])
  if mode == "live" and not args.yes and not planning_refusals:
    say(f"Plan digest: {current_digest}")
    for line in plan_lines(current_plan):
      say(line)
    answer = input(
      f"Apply this plan ({len(files)} files) to the target library? [yes/NO]: "
    )
    if answer.strip().lower() not in {"y", "yes"}:
      say("Upload cancelled.", error=True)
      return 1
    try:
      current_plan, files, current_digest = resolve_plan()
    except (
      graph_client.GraphClientError,
      RuntimeError,
      OSError,
      ValueError,
    ) as exc:
      say(
        f"error: {safe_error(exc, identifiers=identifiers)}", error=True
      )
      return 1
    if approved_digest != current_digest:
      report_plan_mismatch(
        approved_digest, approved_plan, current_plan, current_digest
      )
      return 1
  target = current_plan["target"]
  entries = []
  audit_output = None
  audit_failed = False
  try:
    if args.audit_log:
      audit_output = graph_client.open_new_text(args.audit_log)
  except graph_client.GraphClientError as exc:
    say(f"error: could not create audit log: {exc}", error=True)
    return 1
  try:
    if mode == "dry-run" or planning_refusals:
      if args.plan_file and mode == "dry-run":
        write_plan(args.plan_file, current_plan, current_digest)
      entries = [dry_run_entry(planned, args.verify) for planned in current_plan["files"]]
    else:
      local_paths = {relative: path for path, relative in files}
      approved_folders = set(current_plan["folders_to_create"])
      created_folders: set[str] = set()
      for planned in current_plan["files"]:
        try:
          entries.append(upload_one(args, token, target["drive_id"],
            local_paths[planned["source_path"]], planned, destination_root,
            approved_folders, created_folders))
        except BaseException as exc:
          entry = dry_run_entry(planned, args.verify)
          entry.update(status="failed", error=safe_error(exc, identifiers=identifiers) or type(exc).__name__)
          entries.append(entry)
          raise
  except (RuntimeError, OSError, ValueError) as exc:
    say(f"error: {safe_error(exc, identifiers=identifiers)}", error=True)
    return 1
  finally:
    try:
      audit = make_audit(started=started, mode=mode, operator=operator,
        destination_root=destination_root, target=target, entries=entries,
        digest=current_digest, hosts_contacted=graph_client.contacted_hosts(), plan=current_plan)
      if audit_output is not None:
        audit_output.write(json.dumps(audit, indent=2) + "\n")
        audit_output.flush()
    except OSError as exc:
      audit_failed = True
      say(f"error: could not write audit log: {exc}", error=True)
    finally:
      if audit_output is not None:
        try:
          audit_output.close()
        except OSError as exc:
          audit_failed = True
          say(f"error: could not close audit log: {exc}", error=True)
  if audit_failed:
    return 1
  emit(audit, args.json, current_plan)
  return 2 if audit["failures"] else 0
if __name__ == "__main__":
  raise SystemExit(main())

License: MIT (read it).

Back to the listing