import base64
import hashlib
import json
import pathlib
import os
import sys
ORG = "local"
SKIP_DIRS = {".git", "__pycache__", "target", "node_modules"}
def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def repo_dir(root: str, repo: str) -> str:
if "/" not in repo:
raise ValueError(f"bad repo id {repo!r}")
path = os.path.join(root, repo.split("/", 1)[1])
if not os.path.isdir(path):
raise ValueError(f"unknown repo {repo!r}")
return path
def walk_tree(root: str, repo: str) -> list[dict]:
base = repo_dir(root, repo)
entries = []
for dirpath, dirnames, filenames in os.walk(base):
dirnames[:] = sorted(d for d in dirnames if d not in SKIP_DIRS)
for name in sorted(dirnames):
full = os.path.join(dirpath, name)
rel = os.path.relpath(full, base)
entries.append({"path": rel, "type": "tree", "sha": sha256(rel.encode())})
for name in sorted(filenames):
full = os.path.join(dirpath, name)
rel = os.path.relpath(full, base)
with open(full, "rb") as f:
data = f.read()
entries.append(
{"path": rel, "type": "blob", "sha": sha256(data), "size": len(data)}
)
return entries
def blob_by_sha(root: str, repo: str, sha: str) -> bytes:
for entry in walk_tree(root, repo):
if entry["type"] == "blob" and entry["sha"] == sha:
with open(os.path.join(repo_dir(root, repo), entry["path"]), "rb") as f:
return f.read()
raise ValueError(f"no blob {sha} in {repo}")
def parse_query(q: str) -> tuple[str, str | None, str | None, str | None]:
repo = org = ext = None
terms = []
for token in q.split():
if token.startswith("repo:"):
repo = token[5:]
elif token.startswith("org:"):
org = token[4:]
elif token.startswith("extension:"):
ext = token[10:]
elif token.startswith("path:"):
terms.append(token[5:]) else:
terms.append(token)
return " ".join(terms), repo, org, ext
def search_code(root: str, q: str) -> list[dict]:
terms, repo_scope, _org, ext = parse_query(q)
needles = [t.lower() for t in terms.split() if t]
repos = [f"{ORG}/{repo_scope.split('/', 1)[1]}"] if repo_scope else [
f"{ORG}/{d}" for d in sorted(os.listdir(root))
if os.path.isdir(os.path.join(root, d)) and d not in SKIP_DIRS
]
items = []
for repo in repos:
if not os.path.isdir(os.path.join(root, repo.split("/", 1)[1])):
continue
for entry in walk_tree(root, repo):
if entry["type"] != "blob":
continue
if ext and not entry["path"].lower().endswith("." + ext.lstrip(".")):
continue
full = os.path.join(repo_dir(root, repo), entry["path"])
try:
text = open(full, encoding="utf-8", errors="replace").read()
except OSError:
continue
if text.startswith("\x00") or "\x00" in text[:8192]:
continue matched = [n for n in needles if n in text.lower()]
if needles and not matched:
continue
items.append(
{
"repo": repo,
"path": entry["path"],
"sha": entry["sha"],
"branch": "main",
"matches": matched,
}
)
if len(items) >= 25:
return items
return items
def handle(root: str, method: str, params: dict) -> dict:
if method == "initialize":
return {
"protocol": 1,
"name": "fs",
"capabilities": {"orgs": True, "code_search": True},
}
if method == "search/repos":
query = params.get("query", "").lower()
items = []
for d in sorted(os.listdir(root)):
full = os.path.join(root, d)
if os.path.isdir(full) and d not in SKIP_DIRS and query in d.lower():
items.append({"full_name": f"{ORG}/{d}"})
if not items:
items.append({"org": ORG})
return {"items": items[:20]}
if method == "org/repos":
return {
"repos": [
d
for d in sorted(os.listdir(root))
if os.path.isdir(os.path.join(root, d)) and d not in SKIP_DIRS
]
}
if method == "repo/tree":
repo = params["repo"]
return {
"entries": walk_tree(root, repo),
"truncated": False,
"branch": "main",
}
if method == "repo/clone_url":
return {"clone_url": repo_dir(root, params["repo"])}
if method == "repo/web_url":
base = pathlib.Path(repo_dir(root, params["repo"])).resolve().as_uri()
path = params.get("path", "")
line = params.get("line")
is_file = params.get("is_file", False)
url = f"{base}/{path}" if path else base
if is_file and line:
url += f"#L{line}"
return {"url": url}
if method == "org/url":
return {"url": pathlib.Path(root).resolve().as_uri()}
if method == "repo/blob":
data = blob_by_sha(root, params["repo"], params["sha"])
return {"bytes_b64": base64.b64encode(data).decode()}
if method == "search/code":
return {"items": search_code(root, params.get("q", ""))}
raise ValueError(f"unknown method {method!r}")
def main() -> None:
if len(sys.argv) != 2:
print(__doc__, file=sys.stderr)
sys.exit(2)
root = os.path.abspath(sys.argv[1])
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
req = json.loads(line)
except json.JSONDecodeError:
continue
try:
result = handle(root, req.get("method", ""), req.get("params") or {})
reply = {"jsonrpc": "2.0", "id": req.get("id"), "result": result}
except Exception as e: reply = {
"jsonrpc": "2.0",
"id": req.get("id"),
"error": {"code": 1, "message": str(e)},
}
sys.stdout.write(json.dumps(reply) + "\n")
sys.stdout.flush()
if __name__ == "__main__":
main()