mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
feat(infra): add MCP-only Drive GCCL compression shim
Adds 4-Infrastructure/shim/mcp_drive_gccl_compress.py: - downloads a Google Drive folder via @piotr-agier/google-drive-mcp - GCCL-delta-compresses contents into .tar.zst + JSON manifest - uploads archive + manifest back to Drive - optionally trashes the original folder - handles MCP listFolder pagination Updates 4-Infrastructure/AGENTS.md with the new shim anchor. No rclone or egress-heavy operations; all Drive I/O goes through MCP.
This commit is contained in:
parent
2b3d255a40
commit
0149a5c93d
2 changed files with 427 additions and 1 deletions
|
|
@ -313,7 +313,7 @@ python3 4-Infrastructure/storage/storage_agent.py --loop --interval 900
|
|||
|
||||
## Current Stack-Solidification Anchors
|
||||
|
||||
- `4-Infrastructure/shim/arxiv_oaipmh_harvest.py` — arXiv OAI-PMH harvester: fetches paper metadata (title, abstract, categories, authors) into arxiv DB on neon-64gb
|
||||
- `4-Infrastructure/shim/mcp_drive_gccl_compress.py` — MCP-only Google Drive folder compressor: downloads a Drive folder via `@piotr-agier/google-drive-mcp`, GCCL-delta-compresses contents into a `.tar.zst` archive with JSON manifest, uploads archive + manifest back to Drive, optionally trashes the original. No rclone.
|
||||
- `4-Infrastructure/shim/rrc_arxiv_kernel_refine.py` — RRC arXiv kernel refinement: title+abstract keyword search against arxiv_papers for unmatched equations
|
||||
- `4-Infrastructure/infra/lean_lsp_mcp_wrapper.py` — Python wrapper for `lean-lsp-mcp` to fix the schema of `lean_diagnostic_messages` for compatibility with strict validators like Moonshot/Kimi API.
|
||||
- `4-Infrastructure/shim/stack_solidification_audit.py`
|
||||
|
|
|
|||
426
4-Infrastructure/shim/mcp_drive_gccl_compress.py
Normal file
426
4-Infrastructure/shim/mcp_drive_gccl_compress.py
Normal file
|
|
@ -0,0 +1,426 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
mcp_drive_gccl_compress.py — Download a Google Drive folder via MCP, GCCL-compress it, upload archive.
|
||||
|
||||
Workflow:
|
||||
1. Recursively list all files in a Drive folder using the Google Drive MCP server.
|
||||
2. Download each file to a local temp clone.
|
||||
3. For each file, attempt GCCL delta compression using a previously-seen similar file as reference.
|
||||
4. Pack compressed results into a .tar.zst archive with a JSON manifest.
|
||||
5. Upload the archive back to Google Drive via MCP.
|
||||
6. Optionally move the original folder to trash via MCP.
|
||||
|
||||
Usage:
|
||||
python3 mcp_drive_gccl_compress.py <drive-folder-id> [options]
|
||||
|
||||
No rclone is used — only the Google Drive MCP server and gccl_waveprobe.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tarfile
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
# Insert Research Stack shim path for gccl_waveprobe
|
||||
REPO = Path("/home/allaun/Research Stack")
|
||||
SHIM_DIR = REPO / "4-Infrastructure/shim"
|
||||
sys.path.insert(0, str(SHIM_DIR))
|
||||
|
||||
try:
|
||||
import gccl_waveprobe as gw
|
||||
except ImportError as e:
|
||||
raise RuntimeError(f"Cannot import gccl_waveprobe from {SHIM_DIR}: {e}")
|
||||
|
||||
|
||||
DEFAULT_MCP_COMMAND = ["npx", "-y", "@piotr-agier/google-drive-mcp"]
|
||||
|
||||
|
||||
class DriveMcpClient:
|
||||
"""Very thin JSON-RPC stdio client for the Google Drive MCP server."""
|
||||
|
||||
def __init__(self, env: Optional[Dict[str, str]] = None):
|
||||
self._proc = subprocess.Popen(
|
||||
DEFAULT_MCP_COMMAND,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=False,
|
||||
env={**os.environ, **(env or {})},
|
||||
)
|
||||
self._next_id = 1
|
||||
self._initialize()
|
||||
|
||||
def _initialize(self) -> None:
|
||||
self._call(
|
||||
"initialize",
|
||||
{
|
||||
"protocolVersion": "2024-11-05",
|
||||
"capabilities": {},
|
||||
"clientInfo": {"name": "mcp-drive-gccl", "version": "1.0.0"},
|
||||
},
|
||||
)
|
||||
|
||||
def _send(self, msg: dict) -> None:
|
||||
payload = json.dumps(msg).encode("utf-8") + b"\n"
|
||||
self._proc.stdin.write(payload) # type: ignore[union-attr]
|
||||
self._proc.stdin.flush() # type: ignore[union-attr]
|
||||
|
||||
def _recv(self) -> dict:
|
||||
line = self._proc.stdout.readline() # type: ignore[union-attr]
|
||||
if not line:
|
||||
raise RuntimeError("MCP server closed stdout")
|
||||
return json.loads(line.decode("utf-8"))
|
||||
|
||||
def _call(self, method: str, params: dict) -> dict:
|
||||
msg = {"jsonrpc": "2.0", "id": self._next_id, "method": method, "params": params}
|
||||
self._next_id += 1
|
||||
self._send(msg)
|
||||
while True:
|
||||
resp = self._recv()
|
||||
if resp.get("id") == msg["id"]:
|
||||
if "error" in resp:
|
||||
raise RuntimeError(f"MCP error: {resp['error']}")
|
||||
return resp["result"]
|
||||
|
||||
def list_folder(
|
||||
self,
|
||||
folder_id: Optional[str] = None,
|
||||
page_size: int = 100,
|
||||
page_token: Optional[str] = None,
|
||||
) -> Tuple[List[Dict[str, Any]], Optional[str]]:
|
||||
params: Dict[str, Any] = {"pageSize": page_size}
|
||||
if folder_id:
|
||||
params["folderId"] = folder_id
|
||||
if page_token:
|
||||
params["pageToken"] = page_token
|
||||
result = self._call("tools/call", {"name": "listFolder", "arguments": params})
|
||||
text = self._extract_text(result)
|
||||
items = self._parse_folder_listing(text)
|
||||
next_token = self._parse_next_page_token(text)
|
||||
return items, next_token
|
||||
|
||||
def list_folder_all(self, folder_id: Optional[str] = None) -> List[Dict[str, Any]]:
|
||||
all_items: List[Dict[str, Any]] = []
|
||||
token: Optional[str] = None
|
||||
while True:
|
||||
items, token = self.list_folder(folder_id=folder_id, page_token=token)
|
||||
all_items.extend(items)
|
||||
if not token:
|
||||
break
|
||||
return all_items
|
||||
|
||||
def download_file(self, file_id: str, local_path: Path) -> None:
|
||||
result = self._call(
|
||||
"tools/call",
|
||||
{"name": "downloadFile", "arguments": {"fileId": file_id, "localPath": str(local_path), "overwrite": True}},
|
||||
)
|
||||
text = self._extract_text(result)
|
||||
if "Downloaded" not in text and "Saved" not in text:
|
||||
raise RuntimeError(f"Download failed: {text}")
|
||||
|
||||
def upload_file(self, local_path: Path, parent_folder_id: Optional[str] = None) -> str:
|
||||
args: Dict[str, Any] = {"localPath": str(local_path)}
|
||||
if parent_folder_id:
|
||||
args["parentFolderId"] = parent_folder_id
|
||||
result = self._call("tools/call", {"name": "uploadFile", "arguments": args})
|
||||
text = self._extract_text(result)
|
||||
if "Uploaded" not in text and "uploaded" not in text:
|
||||
raise RuntimeError(f"Upload failed: {text}")
|
||||
return text
|
||||
|
||||
def delete_item(self, item_id: str) -> None:
|
||||
self._call("tools/call", {"name": "deleteItem", "arguments": {"itemId": item_id}})
|
||||
|
||||
def close(self) -> None:
|
||||
if self._proc.poll() is None:
|
||||
self._proc.stdin.close() # type: ignore[union-attr]
|
||||
self._proc.wait(timeout=5)
|
||||
|
||||
@staticmethod
|
||||
def _parse_next_page_token(text: str) -> Optional[str]:
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if line.lower().startswith("nextpagetoken:") or line.lower().startswith("next page token:"):
|
||||
return line.split(":", 1)[1].strip()
|
||||
if "nextpagetoken" in line.lower() and ":" in line:
|
||||
return line.split(":", 1)[1].strip()
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _extract_text(result: dict) -> str:
|
||||
for item in result.get("content", []):
|
||||
if item.get("type") == "text":
|
||||
return item.get("text", "")
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def _parse_folder_listing(text: str) -> List[Dict[str, Any]]:
|
||||
items: List[Dict[str, Any]] = []
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("Contents"):
|
||||
continue
|
||||
# Format: 📄 name (mime) [id: ID, path: ...] [created: ..., modified: ...]
|
||||
# or 📁 name (ID: ID)
|
||||
if line.startswith("📄"):
|
||||
rest = line[2:].strip()
|
||||
# Format: "name (ID: FILE_ID, path: ...)" or "name (mime) (ID: ...)"
|
||||
if "(ID: " in rest:
|
||||
name = rest.split("(ID: ")[0].strip()
|
||||
# Remove trailing mime like "(text/plain)" from name
|
||||
if name.endswith(")") and "(" in name:
|
||||
name = name.rsplit("(", 1)[0].strip()
|
||||
fid = rest.split("(ID: ")[1].split(",", 1)[0].rstrip(")").strip()
|
||||
items.append({"type": "file", "name": name, "id": fid})
|
||||
elif line.startswith("📁"):
|
||||
# Format: name (ID: ID)
|
||||
rest = line[2:].strip()
|
||||
if "(ID: " in rest:
|
||||
name = rest.split("(ID: ")[0].strip()
|
||||
fid = rest.split("(ID: ")[1].rstrip(")").strip()
|
||||
items.append({"type": "folder", "name": name, "id": fid})
|
||||
return items
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
h = hashlib.sha256()
|
||||
with open(path, "rb") as f:
|
||||
while chunk := f.read(64 * 1024):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def gccl_compress_file(
|
||||
src_path: Path,
|
||||
reference_path: Optional[Path] = None,
|
||||
threshold_q16: int = 32768,
|
||||
) -> Tuple[bytes, Dict[str, Any]]:
|
||||
"""Compress a single file with GCCL. Optionally use reference for delta."""
|
||||
data = src_path.read_bytes()
|
||||
reference = reference_path.read_bytes() if reference_path else None
|
||||
|
||||
result = gw.gccl_delta_compress(
|
||||
data,
|
||||
reference=reference,
|
||||
gccl_threshold_q16=threshold_q16,
|
||||
wave_probe=gw.WaveProbe(id=1, sample_rate=48000, buffer_size=256),
|
||||
)
|
||||
|
||||
if result.gccl_decision == gw.Decision.ACCEPT and result.compressed_size < result.original_size:
|
||||
return result.compressed, {
|
||||
"mode": "delta" if reference else "raw-gccl",
|
||||
"reference": str(reference_path) if reference_path else None,
|
||||
"original_size": result.original_size,
|
||||
"compressed_size": result.compressed_size,
|
||||
"ratio": result.compression_ratio,
|
||||
"decision": str(result.gccl_decision),
|
||||
}
|
||||
else:
|
||||
# Compression rejected or expanded — store raw
|
||||
return data, {
|
||||
"mode": "raw",
|
||||
"reference": str(reference_path) if reference_path else None,
|
||||
"original_size": len(data),
|
||||
"compressed_size": len(data),
|
||||
"ratio": 1.0,
|
||||
"decision": str(result.gccl_decision) if result else "N/A",
|
||||
}
|
||||
|
||||
|
||||
def download_folder_recursive(
|
||||
client: DriveMcpClient,
|
||||
folder_id: str,
|
||||
local_root: Path,
|
||||
rename_map: Optional[Dict[str, str]] = None,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Recursively download a Drive folder. Returns manifest of downloaded files."""
|
||||
files: List[Dict[str, Any]] = []
|
||||
_download_folder_recursive(client, folder_id, local_root, files, rename_map or {})
|
||||
return files
|
||||
|
||||
|
||||
def _download_folder_recursive(
|
||||
client: DriveMcpClient,
|
||||
folder_id: str,
|
||||
local_dir: Path,
|
||||
files: List[Dict[str, Any]],
|
||||
rename_map: Dict[str, str],
|
||||
) -> None:
|
||||
local_dir.mkdir(parents=True, exist_ok=True)
|
||||
items = client.list_folder_all(folder_id)
|
||||
for item in items:
|
||||
safe_name = rename_map.get(item["id"], item["name"])
|
||||
local_path = local_dir / safe_name
|
||||
if item["type"] == "folder":
|
||||
_download_folder_recursive(client, item["id"], local_path, files, rename_map)
|
||||
else:
|
||||
print(f" downloading {safe_name} ...")
|
||||
client.download_file(item["id"], local_path)
|
||||
files.append({
|
||||
"drive_id": item["id"],
|
||||
"name": safe_name,
|
||||
"local_path": str(local_path),
|
||||
"sha256": sha256_file(local_path),
|
||||
})
|
||||
|
||||
|
||||
def build_reference_index(
|
||||
files: List[Dict[str, Any]],
|
||||
local_root: Path,
|
||||
) -> Dict[str, Optional[Path]]:
|
||||
"""For each file, pick a reference: same basename elsewhere, or largest previously seen file."""
|
||||
by_basename: Dict[str, Path] = {}
|
||||
refs: Dict[str, Optional[Path]] = {}
|
||||
for f in files:
|
||||
local_path = Path(f["local_path"])
|
||||
basename = local_path.name
|
||||
if basename in by_basename and by_basename[basename] != local_path:
|
||||
refs[f["local_path"]] = by_basename[basename]
|
||||
else:
|
||||
# Use largest previously seen file as reference
|
||||
refs[f["local_path"]] = max(by_basename.values(), key=lambda p: p.stat().st_size) if by_basename else None
|
||||
by_basename[basename] = local_path
|
||||
return refs
|
||||
|
||||
|
||||
def compress_folder(
|
||||
files: List[Dict[str, Any]],
|
||||
local_root: Path,
|
||||
archive_path: Path,
|
||||
threshold_q16: int = 32768,
|
||||
) -> Dict[str, Any]:
|
||||
"""GCCL-compress downloaded files and pack into a .tar.zst archive."""
|
||||
refs = build_reference_index(files, local_root)
|
||||
manifest: Dict[str, Any] = {
|
||||
"schema": "mcp_drive_gccl_archive_v1",
|
||||
"created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"base_dir": str(local_root),
|
||||
"files": [],
|
||||
}
|
||||
|
||||
temp_compressed = archive_path.parent / f"{archive_path.name}.tmp"
|
||||
with tarfile.open(temp_compressed, "w") as tar:
|
||||
for f in files:
|
||||
src = Path(f["local_path"])
|
||||
rel = src.relative_to(local_root)
|
||||
ref = refs.get(f["local_path"])
|
||||
|
||||
compressed_bytes, meta = gccl_compress_file(src, ref, threshold_q16)
|
||||
|
||||
arc_name = f"{rel}.gccl" if meta["mode"] != "raw" else str(rel)
|
||||
info = tarfile.TarInfo(name=arc_name)
|
||||
info.size = len(compressed_bytes)
|
||||
info.mtime = time.time()
|
||||
tar.addfile(info, io.BytesIO(compressed_bytes))
|
||||
|
||||
manifest["files"].append({
|
||||
"path": str(rel),
|
||||
"archive_path": arc_name,
|
||||
"drive_id": f["drive_id"],
|
||||
"sha256": f["sha256"],
|
||||
**meta,
|
||||
})
|
||||
|
||||
# Re-pack with zstd
|
||||
with open(temp_compressed, "rb") as src_f, open(archive_path, "wb") as dst_f:
|
||||
subprocess.run(
|
||||
["zstd", "-T0", "-19", "-q", "-f", "-o", str(archive_path)],
|
||||
stdin=src_f,
|
||||
check=True,
|
||||
)
|
||||
temp_compressed.unlink()
|
||||
|
||||
# Write manifest alongside
|
||||
manifest_path = archive_path.with_suffix(".manifest.json")
|
||||
manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
|
||||
|
||||
total_original = sum(m["original_size"] for m in manifest["files"])
|
||||
total_compressed = archive_path.stat().st_size + manifest_path.stat().st_size
|
||||
manifest["summary"] = {
|
||||
"file_count": len(files),
|
||||
"total_original_bytes": total_original,
|
||||
"archive_bytes": archive_path.stat().st_size,
|
||||
"manifest_bytes": manifest_path.stat().st_size,
|
||||
"total_output_bytes": total_compressed,
|
||||
"si_compression_ratio": total_original / total_compressed if total_compressed else 0,
|
||||
}
|
||||
manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
|
||||
return manifest
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("folder_id", help="Google Drive folder ID to compress")
|
||||
parser.add_argument("--work-dir", default="/tmp/mcp-drive-gccl", help="Local scratch directory")
|
||||
parser.add_argument("--archive-name", help="Output archive name (default: folder_id.tar.zst)")
|
||||
parser.add_argument("--upload-parent", help="Drive folder ID to upload archive into")
|
||||
parser.add_argument("--trash-original", action="store_true", help="Move original folder to trash after upload")
|
||||
parser.add_argument("--threshold-q16", type=int, default=32768, help="GCCL threshold in Q16.16 (default 32768 = 0.5)")
|
||||
parser.add_argument("--dry-run", action="store_true", help="Download and compress but do not upload or trash")
|
||||
args = parser.parse_args()
|
||||
|
||||
archive_name = args.archive_name or f"{args.folder_id}.tar.zst"
|
||||
work_dir = Path(args.work_dir)
|
||||
local_clone = work_dir / "clone" / args.folder_id
|
||||
archive_path = work_dir / "out" / archive_name
|
||||
manifest_path = archive_path.with_suffix(".manifest.json")
|
||||
|
||||
work_dir.mkdir(parents=True, exist_ok=True)
|
||||
local_clone.parent.mkdir(parents=True, exist_ok=True)
|
||||
archive_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
env = {"GOOGLE_DRIVE_OAUTH_CREDENTIALS": "/home/allaun/gcp-oauth.keys.json"}
|
||||
client = DriveMcpClient(env=env)
|
||||
|
||||
try:
|
||||
print(f"[*] Downloading folder {args.folder_id} to {local_clone}")
|
||||
files = download_folder_recursive(client, args.folder_id, local_clone)
|
||||
print(f"[+] Downloaded {len(files)} files")
|
||||
|
||||
if not files:
|
||||
print("[-] No files found; aborting")
|
||||
return 1
|
||||
|
||||
print(f"[*] GCCL-compressing into {archive_path}")
|
||||
manifest = compress_folder(files, local_clone, archive_path, args.threshold_q16)
|
||||
print(f"[+] Archive: {archive_path} ({archive_path.stat().st_size} bytes)")
|
||||
print(f"[+] Manifest: {manifest_path}")
|
||||
print(f"[+] SI compression ratio: {manifest['summary']['si_compression_ratio']:.2f}")
|
||||
|
||||
if not args.dry_run:
|
||||
print("[*] Uploading archive to Google Drive")
|
||||
archive_upload_text = client.upload_file(archive_path, args.upload_parent)
|
||||
print(f"[+] Archive upload result: {archive_upload_text}")
|
||||
|
||||
print("[*] Uploading manifest to Google Drive")
|
||||
manifest_upload_text = client.upload_file(manifest_path, args.upload_parent)
|
||||
print(f"[+] Manifest upload result: {manifest_upload_text}")
|
||||
|
||||
if args.trash_original:
|
||||
print(f"[*] Trashing original folder {args.folder_id}")
|
||||
client.delete_item(args.folder_id)
|
||||
print("[+] Original trashed")
|
||||
else:
|
||||
print("[DRY RUN] No upload or trash performed")
|
||||
|
||||
# Print manifest summary
|
||||
print(json.dumps(manifest["summary"], indent=2))
|
||||
return 0
|
||||
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Add table
Reference in a new issue