94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from http.client import HTTPConnection, HTTPException
|
|
from pathlib import Path
|
|
import traceback
|
|
from contextlib import closing
|
|
from typing import Optional
|
|
|
|
WEBSERVER_HOST = "localhost"
|
|
WEBSERVER_ENDPOINT = "/api/provenance/call"
|
|
PORT_FILE_SUFFIX = "-provenance-port.txt"
|
|
|
|
class ProvenanceHookError(RuntimeError):
|
|
pass
|
|
|
|
def http_request(method, host, port, location, *, body: Optional[bytes] = None, headers={}, timeout=None) -> bytes:
|
|
with closing(HTTPConnection(host, port, timeout=timeout)) as connection:
|
|
connection.request(method, location, body=body, headers=headers)
|
|
|
|
def get_server_port():
|
|
claude_root = os.getenv("CLAUDE_PROJECT_DIR")
|
|
path_hash = hashlib.md5(claude_root.encode('utf-8')).hexdigest()
|
|
port_file = Path(tempfile.gettempdir()) / (path_hash + PORT_FILE_SUFFIX)
|
|
|
|
return int(port_file.read_text("utf-8").strip())
|
|
|
|
|
|
def send_diff_to_webserver(file_path, timestamp_ms):
|
|
try:
|
|
port = get_server_port()
|
|
except FileNotFoundError as e:
|
|
raise ProvenanceHookError(
|
|
f"Could not determine API port: {e.filename} does not exist") from e
|
|
except Exception as e:
|
|
raise ProvenanceHookError("Could not determine API port") from e
|
|
|
|
url = f"http://{WEBSERVER_HOST}:{port}{WEBSERVER_ENDPOINT}"
|
|
|
|
try:
|
|
payload = {"file_path": file_path, "timestamp": timestamp_ms}
|
|
return http_request(
|
|
"POST",
|
|
WEBSERVER_HOST,
|
|
port=port,
|
|
location=WEBSERVER_ENDPOINT,
|
|
body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
|
|
headers={'Content-Type': 'application/json'},
|
|
timeout=0.5
|
|
)
|
|
|
|
except (HTTPException, OSError, ConnectionError) as e:
|
|
raise ProvenanceHookError(
|
|
f"Network error while sending diff to {url}") from e
|
|
except Exception as e:
|
|
raise ProvenanceHookError(
|
|
f"Unknown error while sending diff to {url}") from e
|
|
|
|
|
|
def extract_file_path(tool_name, tool_input):
|
|
if tool_name in ["Write", "Edit", "MultiEdit"]:
|
|
return tool_input.get('file_path', 'unknown')
|
|
if tool_name == "NotebookEdit":
|
|
return tool_input.get('notebook_path', 'unknown')
|
|
return 'unknown'
|
|
|
|
|
|
def excepthook(type, value, traceback_):
|
|
traceback.print_exception(type, value, traceback_, file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def main():
|
|
data = json.load(sys.stdin)
|
|
tool_name = data.get('tool_name', 'unknown')
|
|
|
|
modification_tools = [
|
|
"Write", "Edit", "MultiEdit", "NotebookEdit"
|
|
]
|
|
|
|
if tool_name in modification_tools:
|
|
tool_input = data.get('tool_input', {})
|
|
file_path = extract_file_path(tool_name, tool_input)
|
|
if file_path:
|
|
timestamp_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
|
send_diff_to_webserver(file_path, timestamp_ms)
|
|
|
|
if __name__ == "__main__":
|
|
sys.excepthook = excepthook
|
|
sys.exit(main())
|