#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import gzip import hashlib import json import mimetypes import re import shutil import subprocess import time import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any from file_history_contract import BLOB_VERIFICATION_CONTRACT DEFAULT_LOGIN_DOMAIN = "login.salesforce.com" DEFAULT_API_VERSION = "v63.0" DEFAULT_METADATA_MANIFEST_NAME = "manifest_content_version.json" DEFAULT_METADATA_FILE = "ContentVersion.csv.gz" DEFAULT_CONTENT_DOCUMENT_FILE = "ContentDocument.csv.gz" DEFAULT_CONTENT_DOCUMENT_LINK_FILE = "ContentDocumentLink.csv.gz" DEFAULT_CHUNK_SIZE = 100 DEFAULT_BLOB_PREFIX = "salesforce/content/blobs" SAFE_SF_ID_RE = re.compile(r"^[A-Za-z0-9]{15,18}$") def utc_now() -> datetime: return datetime.now(timezone.utc) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def ensure_dir(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) def load_kv_config(path: Path) -> dict[str, str]: data: dict[str, str] = {} for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or ":" not in line: continue key, value = line.split(":", 1) data[key.strip()] = value.strip() return data def load_json(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"{path} must be a JSON object") return payload class SalesforceClient: def __init__(self, config: dict[str, str], api_version: str) -> None: self.login_domain = config.get("tokenDomain") or DEFAULT_LOGIN_DOMAIN self.client_id = config["clientId"] self.client_secret = config["clientSecret"] self.username = config["username"] self.password = config["password"] self.api_version = api_version self.instance_url = "" self.access_token = "" def login(self) -> None: token_url = f"https://{self.login_domain}/services/oauth2/token" form = urllib.parse.urlencode( { "grant_type": "password", "client_id": self.client_id, "client_secret": self.client_secret, "username": self.username, "password": self.password, } ).encode("utf-8") req = urllib.request.Request(token_url, data=form, method="POST") with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) self.instance_url = data["instance_url"] self.access_token = data["access_token"] def _headers(self) -> dict[str, str]: if not self.access_token: self.login() return {"Authorization": f"Bearer {self.access_token}"} def _api(self, path: str) -> str: if not self.instance_url: self.login() return f"{self.instance_url}/services/data/{self.api_version}{path}" def _request_json(self, url: str) -> dict[str, Any]: req = urllib.request.Request(url, headers=self._headers()) with urllib.request.urlopen(req, timeout=180) as resp: return json.loads(resp.read().decode("utf-8")) def describe(self, object_name: str) -> dict[str, Any]: return self._request_json(self._api(f"/sobjects/{object_name}/describe")) def query_pages(self, soql: str) -> list[dict[str, Any]]: url = self._api("/query") + "?q=" + urllib.parse.quote(soql, safe="") records: list[dict[str, Any]] = [] while url: payload = self._request_json(url) for record in payload.get("records", []): if isinstance(record, dict): record.pop("attributes", None) records.append(record) next_url = payload.get("nextRecordsUrl") url = f"{self.instance_url}{next_url}" if next_url else "" return records def download_version_data(self, version_id: str) -> bytes: url = self._api(f"/sobjects/ContentVersion/{version_id}/VersionData") req = urllib.request.Request(url, headers=self._headers()) with urllib.request.urlopen(req, timeout=300) as resp: return resp.read() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Export ContentVersion metadata and deduplicated VersionData blobs from an existing daily snapshot." ) parser.add_argument("--config", required=True, help="Path to readonly Salesforce config file.") parser.add_argument("--snapshot-dir", required=True, help="Existing daily snapshot directory.") parser.add_argument("--content-document-file", default=DEFAULT_CONTENT_DOCUMENT_FILE, help="ContentDocument source file name.") parser.add_argument( "--content-document-link-file", default=DEFAULT_CONTENT_DOCUMENT_LINK_FILE, help="ContentDocumentLink source file name whose ContentDocumentId values are unioned into the export scope.", ) parser.add_argument("--metadata-file", default=DEFAULT_METADATA_FILE, help="ContentVersion metadata file name.") parser.add_argument("--metadata-manifest-name", default=DEFAULT_METADATA_MANIFEST_NAME, help="ContentVersion metadata manifest file name.") parser.add_argument("--base-manifest-name", default="manifest.json", help="Base snapshot manifest file name.") parser.add_argument("--api-version", default=DEFAULT_API_VERSION, help="Salesforce API version.") parser.add_argument("--chunk-size", type=int, default=DEFAULT_CHUNK_SIZE, help="SOQL IN-clause chunk size.") parser.add_argument("--blob-prefix", default=DEFAULT_BLOB_PREFIX, help="Blob storage prefix under bucket.") parser.add_argument("--s3-bucket", help="Optional S3 bucket override.") parser.add_argument("--s3-prefix", help="Optional S3 prefix override for metadata snapshot upload.") parser.add_argument("--skip-s3-upload", action="store_true", help="Skip S3 upload.") parser.add_argument( "--skip-blob-upload", action="store_true", help="Deprecated fail-closed flag; ContentVersion metadata-only manifests are forbidden.", ) parser.add_argument( "--blob-throttle-seconds", type=float, default=0.0, help="Sleep this many seconds after each blob download+upload to rate-limit the run. Default 0 (no throttle).", ) return parser.parse_args() def valid_sf_id(value: str) -> bool: return bool(SAFE_SF_ID_RE.match(value)) def read_salesforce_ids(path: Path, *, field_name: str) -> list[str]: with gzip.open(path, "rt", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) ids = { str(row.get(field_name, "")).strip() for row in reader if valid_sf_id(str(row.get(field_name, "")).strip()) } return sorted(ids) def read_content_document_ids(path: Path) -> list[str]: return read_salesforce_ids(path, field_name="Id") def read_linked_content_document_ids(path: Path) -> list[str]: return read_salesforce_ids(path, field_name="ContentDocumentId") def collect_content_document_ids( *, content_document_path: Path, content_document_link_path: Path | None, ) -> dict[str, Any]: content_document_ids = read_content_document_ids(content_document_path) linked_document_ids: list[str] = [] if content_document_link_path and content_document_link_path.exists(): linked_document_ids = read_linked_content_document_ids(content_document_link_path) return { "ids": sorted(set(content_document_ids) | set(linked_document_ids)), "content_document_id_count": len(content_document_ids), "linked_content_document_id_count": len(linked_document_ids), } def require_metadata_scope_complete( document_ids: list[str], metadata_rows: list[dict[str, Any]], ) -> int: """Require at least one visible ContentVersion for every scoped document.""" expected = set(document_ids) observed = { str(row.get("ContentDocumentId") or "").strip() for row in metadata_rows if str(row.get("ContentDocumentId") or "").strip() } missing = sorted(expected - observed) if missing: raise RuntimeError( f"ContentVersion export is missing metadata for {len(missing)} of " f"{len(expected)} ContentDocument IDs; source principal file " "visibility may have changed" ) return len(observed & expected) def require_blob_export_complete( *, bucket: str, expected: int, uploaded: int, skipped: int, failed: int, verified: int, ) -> None: if not bucket.strip(): raise RuntimeError( "ContentVersion blob export has no S3 bucket authority; refusing " "to publish a metadata-only file-history manifest" ) available = uploaded + skipped attempted = available + failed if ( failed > 0 or attempted != expected or available != expected or verified != expected ): raise RuntimeError( "ContentVersion blob export is incomplete: " f"expected={expected} uploaded={uploaded} skipped={skipped} " f"failed={failed} verified={verified}; refusing to publish an incomplete " "file-history manifest" ) def require_load_eligible_blob_export(*, skip_blob_upload: bool) -> None: if skip_blob_upload: raise RuntimeError( "ContentVersion metadata-only manifests are forbidden; omit the " "ContentVersion export instead of using --skip-blob-upload" ) def pick_metadata_fields(describe: dict[str, Any]) -> list[str]: required = [ "Id", "ContentDocumentId", "Title", "FileExtension", "FileType", "ContentSize", "Checksum", "VersionNumber", "IsLatest", "PublishStatus", "PathOnClient", "ContentModifiedDate", "SystemModstamp", "CreatedDate", "LastModifiedDate", ] available: set[str] = set() for field in describe.get("fields", []): name = field.get("name") if not name or field.get("deprecatedAndHidden"): continue if str(name) == "VersionData": continue available.add(str(name)) fields = [name for name in required if name in available] if not fields: raise ValueError("no usable ContentVersion metadata fields discovered") return fields def sql_quote(value: str) -> str: return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'" def chunked(values: list[str], size: int) -> list[list[str]]: return [values[idx : idx + size] for idx in range(0, len(values), size)] def write_rows_gzip(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> None: with gzip.open(path, "wt", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({name: row.get(name, "") for name in fieldnames}) def upload_bytes_to_s3( data: bytes, s3_uri: str, content_type: str, sha256: str, ) -> None: aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for S3 upload but was not found in PATH") proc = subprocess.run( [ aws_path, "s3", "cp", "-", s3_uri, "--content-type", content_type, "--metadata", f"sha256={sha256}", ], input=data, check=True, ) assert proc.returncode == 0 def upload_file_to_s3(local_path: Path, s3_uri: str) -> None: aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for S3 upload but was not found in PATH") subprocess.run([aws_path, "s3", "cp", str(local_path), s3_uri], check=True) def s3_exists(s3_uri: str) -> bool: aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for S3 checks but was not found in PATH") result = subprocess.run([aws_path, "s3", "ls", s3_uri], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return result.returncode == 0 def head_s3_blob(bucket: str, key: str) -> dict[str, Any] | None: """Read one exact S3 key; only a not-found response is treated as absent.""" aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for S3 checks but was not found in PATH") proc = subprocess.run( [ aws_path, "s3api", "head-object", "--bucket", bucket, "--key", key, "--output", "json", ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) if proc.returncode != 0: error = proc.stderr.strip() if any(marker in error for marker in ("404", "Not Found", "NoSuchKey")): return None raise RuntimeError(f"S3 head-object failed for s3://{bucket}/{key}: {error}") payload = json.loads(proc.stdout or "{}") if not isinstance(payload, dict): raise RuntimeError(f"S3 head-object returned invalid JSON for s3://{bucket}/{key}") return payload def download_s3_blob(bucket: str, key: str) -> bytes: aws_path = shutil.which("aws") if not aws_path: raise RuntimeError("aws CLI is required for S3 readback but was not found in PATH") proc = subprocess.run( [aws_path, "s3", "cp", f"s3://{bucket}/{key}", "-", "--only-show-errors"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if proc.returncode != 0: error = proc.stderr.decode("utf-8", errors="replace").strip() raise RuntimeError(f"S3 readback failed for s3://{bucket}/{key}: {error}") return proc.stdout def required_content_size(row: dict[str, Any]) -> int: raw = str(row.get("ContentSize") or "").strip() try: size = int(raw) except ValueError as exc: raise RuntimeError(f"invalid ContentSize={raw!r}") from exc if size < 0: raise RuntimeError(f"invalid ContentSize={raw!r}") return size def salesforce_md5(row: dict[str, Any]) -> str | None: checksum = str(row.get("Checksum") or "").strip().lower() return checksum if re.fullmatch(r"[0-9a-f]{32}", checksum) else None def verify_blob_bytes( row: dict[str, Any], data: bytes, *, source: str, ) -> str: expected_size = required_content_size(row) if len(data) != expected_size: raise RuntimeError( f"{source} byte length mismatch: expected={expected_size} actual={len(data)}" ) checksum = salesforce_md5(row) if checksum and hashlib.md5(data).hexdigest() != checksum: # noqa: S324 -- source integrity check raise RuntimeError(f"{source} checksum mismatch") return hashlib.sha256(data).hexdigest() def verify_existing_s3_blob( *, client: SalesforceClient, row: dict[str, Any], bucket: str, key: str, head: dict[str, Any], ) -> str: expected_size = required_content_size(row) if optional_content_length(head) != expected_size: raise RuntimeError( f"S3 ContentLength mismatch: expected={expected_size} " f"actual={head.get('ContentLength')!r}" ) s3_data = download_s3_blob(bucket, key) s3_sha256 = verify_blob_bytes(row, s3_data, source="S3 readback") metadata = head.get("Metadata") if isinstance(head.get("Metadata"), dict) else {} stored_sha256 = str(metadata.get("sha256") or "").strip().lower() if stored_sha256 and stored_sha256 != s3_sha256: raise RuntimeError("S3 sha256 metadata mismatch") if not salesforce_md5(row): version_id = str(row.get("Id") or "").strip() source_data = client.download_version_data(version_id) source_sha256 = verify_blob_bytes(row, source_data, source="Salesforce source") if source_sha256 != s3_sha256: raise RuntimeError("S3 readback differs from Salesforce source") return s3_sha256 def optional_content_length(head: dict[str, Any]) -> int | None: value = head.get("ContentLength") try: return int(value) if value is not None else None except (TypeError, ValueError): return None def verify_uploaded_s3_blob( *, bucket: str, key: str, row: dict[str, Any], source_sha256: str, ) -> None: head = head_s3_blob(bucket, key) if head is None: raise RuntimeError(f"uploaded S3 object is missing at exact key s3://{bucket}/{key}") data = download_s3_blob(bucket, key) readback_sha256 = verify_blob_bytes(row, data, source="S3 upload readback") metadata = head.get("Metadata") if isinstance(head.get("Metadata"), dict) else {} stored_sha256 = str(metadata.get("sha256") or "").strip().lower() if optional_content_length(head) != len(data): raise RuntimeError("uploaded S3 ContentLength does not match readback bytes") if stored_sha256 != source_sha256 or readback_sha256 != source_sha256: raise RuntimeError("uploaded S3 object failed sha256 readback verification") def resolve_metadata_s3_target(snapshot_dir: Path, args: argparse.Namespace, base_manifest: dict[str, Any]) -> str | None: if args.skip_s3_upload: return None if args.s3_bucket: prefix = (args.s3_prefix or "salesforce/full").strip("/") parts = snapshot_dir.parts[-2:] return f"s3://{args.s3_bucket}/{prefix}/{parts[0]}/{parts[1]}/" s3_uri = base_manifest.get("s3_uri") return str(s3_uri) if s3_uri else None def guess_content_type(path_on_client: str, file_type: str) -> str: guessed, _ = mimetypes.guess_type(path_on_client or "") if guessed: return guessed normalized = (file_type or "").lower() if normalized == "pdf": return "application/pdf" if normalized in {"png", "jpg", "jpeg", "gif", "bmp", "svg"}: return f"image/{'jpeg' if normalized == 'jpg' else normalized}" return "application/octet-stream" def main() -> int: args = parse_args() require_load_eligible_blob_export(skip_blob_upload=args.skip_blob_upload) snapshot_dir = Path(args.snapshot_dir).expanduser().resolve() ensure_dir(snapshot_dir) readonly_config = load_kv_config(Path(args.config).expanduser().resolve()) client = SalesforceClient(readonly_config, api_version=args.api_version) client.login() base_manifest_path = snapshot_dir / args.base_manifest_name base_manifest = load_json(base_manifest_path) if base_manifest_path.exists() else {} content_document_path = snapshot_dir / args.content_document_file if not content_document_path.exists(): raise FileNotFoundError(f"ContentDocument source file not found: {content_document_path}") content_document_link_path = snapshot_dir / args.content_document_link_file doc_scope = collect_content_document_ids( content_document_path=content_document_path, content_document_link_path=content_document_link_path, ) doc_ids = doc_scope["ids"] if not doc_ids: raise ValueError( f"no ContentDocument IDs found in {content_document_path} " f"or {content_document_link_path}" ) describe = client.describe("ContentVersion") fields = pick_metadata_fields(describe) metadata_rows: list[dict[str, Any]] = [] total_chunks = 0 for idx, chunk in enumerate(chunked(doc_ids, max(1, args.chunk_size)), start=1): total_chunks += 1 quoted_ids = ", ".join(sql_quote(item) for item in chunk) soql = ( f"SELECT {', '.join(fields)} FROM ContentVersion " f"WHERE ContentDocumentId IN ({quoted_ids})" ) rows = client.query_pages(soql) metadata_rows.extend(rows) print(f"[ok] metadata_chunk={idx} doc_ids={len(chunk)} rows={len(rows)}") time.sleep(0.05) metadata_rows.sort(key=lambda row: (str(row.get("ContentDocumentId", "")), str(row.get("Id", "")))) metadata_document_count = require_metadata_scope_complete( doc_ids, metadata_rows, ) metadata_path = snapshot_dir / args.metadata_file write_rows_gzip(metadata_path, fields, metadata_rows) metadata_s3_dir = resolve_metadata_s3_target(snapshot_dir, args, base_manifest) blob_uploaded = 0 blob_skipped = 0 blob_failed = 0 blob_verified = 0 blob_bucket = str(args.s3_bucket or "").strip() if not blob_bucket and metadata_s3_dir: blob_bucket = metadata_s3_dir.split("/")[2] if blob_bucket: blob_prefix = args.blob_prefix.strip("/") for row in metadata_rows: version_id = str(row.get("Id", "")).strip() if not valid_sf_id(version_id): blob_failed += 1 print(f"[warn] blob_failed version_id={version_id!r} error=invalid Salesforce ID") continue blob_key = f"{blob_prefix}/{version_id}" blob_uri = f"s3://{blob_bucket}/{blob_key}" try: existing_head = head_s3_blob(blob_bucket, blob_key) if existing_head is not None: verify_existing_s3_blob( client=client, row=row, bucket=blob_bucket, key=blob_key, head=existing_head, ) blob_skipped += 1 blob_verified += 1 continue data = client.download_version_data(version_id) source_sha256 = verify_blob_bytes( row, data, source="Salesforce source", ) content_type = guess_content_type( str(row.get("PathOnClient", "")), str(row.get("FileType", "")), ) upload_bytes_to_s3(data, blob_uri, content_type, source_sha256) verify_uploaded_s3_blob( bucket=blob_bucket, key=blob_key, row=row, source_sha256=source_sha256, ) blob_uploaded += 1 blob_verified += 1 if (blob_uploaded + blob_skipped + blob_failed) % 50 == 0: print( f"[ok] blob_progress uploaded={blob_uploaded} skipped={blob_skipped} " f"failed={blob_failed} verified={blob_verified}" ) if args.blob_throttle_seconds > 0: time.sleep(args.blob_throttle_seconds) except Exception as exc: blob_failed += 1 print(f"[warn] blob_failed version_id={version_id} error={exc}") require_blob_export_complete( bucket=blob_bucket, expected=len(metadata_rows), uploaded=blob_uploaded, skipped=blob_skipped, failed=blob_failed, verified=blob_verified, ) if metadata_s3_dir and not args.skip_s3_upload: metadata_uri = metadata_s3_dir.rstrip("/") + "/" + metadata_path.name upload_file_to_s3(metadata_path, metadata_uri) print(f"[ok] uploaded={metadata_uri}") else: metadata_uri = None manifest = { "manifest_version": 1, "export_type": "content_version_layered", "snapshot_time": utc_now().isoformat(), "snapshot_dir": str(snapshot_dir), "source_org": client.instance_url, "api_version": args.api_version, "source_file": str(content_document_path.name), "content_document_link_file": str(content_document_link_path.name) if content_document_link_path.exists() else None, "content_document_id_count": doc_scope["content_document_id_count"], "linked_content_document_id_count": doc_scope["linked_content_document_id_count"], "doc_id_count": len(doc_ids), "chunk_size": args.chunk_size, "chunk_count": total_chunks, "record_count": len(metadata_rows), "metadata_document_count": metadata_document_count, "metadata_file": metadata_path.name, "metadata_sha256": sha256_file(metadata_path), "metadata_s3_uri": metadata_uri, "blob_prefix": args.blob_prefix.strip("/"), "blob_s3_bucket": blob_bucket or None, "blob_expected": len(metadata_rows), "blob_available": blob_uploaded + blob_skipped, "blob_uploaded": blob_uploaded, "blob_skipped": blob_skipped, "blob_failed": blob_failed, "blob_verified": blob_verified, "blob_verification_contract": BLOB_VERIFICATION_CONTRACT, "fields": fields, "base_manifest": base_manifest_path.name if base_manifest_path.exists() else None, } manifest_path = snapshot_dir / args.metadata_manifest_name manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8") if metadata_s3_dir and not args.skip_s3_upload: manifest_uri = metadata_s3_dir.rstrip("/") + "/" + manifest_path.name upload_file_to_s3(manifest_path, manifest_uri) print(f"[ok] uploaded={manifest_uri}") print(f"[ok] metadata_file={metadata_path}") print(f"[ok] metadata_manifest={manifest_path}") return 0 if __name__ == "__main__": raise SystemExit(main())