228 lines
8.5 KiB
Python
228 lines
8.5 KiB
Python
"""Site-editing tool set for BetterBot.
|
|
|
|
Provides list_files, read_file, and write_file tools that operate on
|
|
mounted project directories (Better Life SG website, Memoraiz frontend).
|
|
After writing, changes are committed and pushed via git.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import pathlib
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
from config import settings
|
|
from tool_registry import ToolSet, openai_tools_to_copilot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Project definitions ──────────────────────────────────────────
|
|
|
|
|
|
def _build_projects() -> dict[str, dict[str, Any]]:
|
|
"""Build the project map from environment-configured directories."""
|
|
site_dir = pathlib.Path(settings.SITE_DIR) if hasattr(settings, "SITE_DIR") else pathlib.Path("/site")
|
|
memoraiz_dir = pathlib.Path(settings.MEMORAIZ_DIR) if hasattr(settings, "MEMORAIZ_DIR") else pathlib.Path("/memoraiz")
|
|
|
|
projects: dict[str, dict[str, Any]] = {}
|
|
if site_dir.exists():
|
|
projects["betterlifesg"] = {
|
|
"dir": site_dir,
|
|
"label": "Better Life SG website",
|
|
"git_repo": site_dir.parent,
|
|
}
|
|
if memoraiz_dir.exists():
|
|
projects["memoraiz"] = {
|
|
"dir": memoraiz_dir,
|
|
"label": "Memoraiz app (React frontend)",
|
|
"git_repo": memoraiz_dir.parent,
|
|
}
|
|
return projects
|
|
|
|
|
|
PROJECTS = _build_projects()
|
|
PROJECT_NAMES = list(PROJECTS.keys()) or ["betterlifesg", "memoraiz"]
|
|
|
|
|
|
# ── Path safety ──────────────────────────────────────────────────
|
|
|
|
|
|
def _resolve(base: pathlib.Path, path: str) -> pathlib.Path:
|
|
"""Resolve a relative path inside a base dir, preventing path traversal."""
|
|
resolved = (base / path).resolve()
|
|
if not str(resolved).startswith(str(base.resolve())):
|
|
raise ValueError(f"Path traversal blocked: {path}")
|
|
return resolved
|
|
|
|
|
|
# ── Git push ─────────────────────────────────────────────────────
|
|
|
|
|
|
def _git_push(project_key: str, changed_file: str) -> str:
|
|
"""Commit and push changes in the project's git repo."""
|
|
proj = PROJECTS[project_key]
|
|
repo = proj["git_repo"]
|
|
if not (repo / ".git").exists():
|
|
return "(no git repo — skipped push)"
|
|
try:
|
|
subprocess.run(["git", "add", "-A"], cwd=repo, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "commit", "-m", f"betterbot: update {changed_file}"],
|
|
cwd=repo,
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
result = subprocess.run(
|
|
["git", "push", "origin", "HEAD"],
|
|
cwd=repo,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
logger.info("Git push for %s: %s", project_key, result.stderr.strip())
|
|
return f"Pushed {project_key} to git"
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error("Git/deploy error: %s\nstdout: %s\nstderr: %s", e, e.stdout, e.stderr)
|
|
return f"Push failed: {e.stderr or e.stdout or str(e)}"
|
|
|
|
|
|
# ── Tool implementations ─────────────────────────────────────────
|
|
|
|
|
|
def handle_tool_call(name: str, args: dict) -> str:
|
|
"""Execute a site-editing tool call and return the result as a string."""
|
|
project_key = args.get("project", "betterlifesg")
|
|
if project_key not in PROJECTS:
|
|
return f"Unknown project: {project_key}. Available: {', '.join(PROJECTS.keys())}"
|
|
base = PROJECTS[project_key]["dir"]
|
|
|
|
if name == "list_files":
|
|
subdir = args.get("subdirectory", "")
|
|
target = _resolve(base, subdir) if subdir else base
|
|
files = []
|
|
for p in sorted(target.rglob("*")):
|
|
if p.is_file() and not any(
|
|
part in (".git", "node_modules", "__pycache__") for part in p.parts
|
|
):
|
|
files.append(str(p.relative_to(base)))
|
|
return "\n".join(files[:200]) if files else "(no files found)"
|
|
|
|
if name == "read_file":
|
|
path = _resolve(base, args["path"])
|
|
if not path.exists():
|
|
return f"Error: {args['path']} does not exist."
|
|
if path.suffix in (".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico"):
|
|
return f"[Binary image file: {args['path']}, {path.stat().st_size} bytes]"
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
if name == "write_file":
|
|
path = _resolve(base, args["path"])
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(args["content"], encoding="utf-8")
|
|
push_result = _git_push(project_key, args["path"])
|
|
return f"OK — wrote {len(args['content'])} chars to {args['path']}. {push_result}"
|
|
|
|
return f"Unknown tool: {name}"
|
|
|
|
|
|
# ── OpenAI function-calling tool schemas ─────────────────────────
|
|
|
|
TOOLS = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "list_files",
|
|
"description": "List all files in a project directory.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"project": {
|
|
"type": "string",
|
|
"enum": PROJECT_NAMES,
|
|
"description": "Which project to list files for.",
|
|
},
|
|
"subdirectory": {
|
|
"type": "string",
|
|
"description": "Optional subdirectory to list, e.g. 'src/pages'. Defaults to root.",
|
|
"default": "",
|
|
},
|
|
},
|
|
"required": ["project"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "read_file",
|
|
"description": "Read the full contents of a project file.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"project": {
|
|
"type": "string",
|
|
"enum": PROJECT_NAMES,
|
|
"description": "Which project the file belongs to.",
|
|
},
|
|
"path": {
|
|
"type": "string",
|
|
"description": "Relative path inside the project directory.",
|
|
},
|
|
},
|
|
"required": ["project", "path"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "write_file",
|
|
"description": "Write (create or overwrite) a text file in a project directory. After writing, changes are committed and pushed to git automatically.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"project": {
|
|
"type": "string",
|
|
"enum": PROJECT_NAMES,
|
|
"description": "Which project the file belongs to.",
|
|
},
|
|
"path": {
|
|
"type": "string",
|
|
"description": "Relative path inside the project directory.",
|
|
},
|
|
"content": {
|
|
"type": "string",
|
|
"description": "The full file content to write.",
|
|
},
|
|
},
|
|
"required": ["project", "path", "content"],
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
SYSTEM_PROMPT = """\
|
|
You have access to site-editing tools for managing project files. \
|
|
When asked to change site content, use list_files to see what's available, \
|
|
read_file to understand the current state, then write_file to apply changes. \
|
|
Always write the COMPLETE file content — never partial. \
|
|
Changes are committed and pushed to git automatically after writing.\
|
|
"""
|
|
|
|
# ── ToolSet registration ─────────────────────────────────────────
|
|
|
|
|
|
def _factory(context: dict) -> list:
|
|
"""Build Copilot SDK tools from the OpenAI schemas."""
|
|
return openai_tools_to_copilot(TOOLS, handler=handle_tool_call)
|
|
|
|
|
|
site_editing_toolset = ToolSet(
|
|
name="site_editing",
|
|
description="Edit Better Life SG and Memoraiz project files",
|
|
capability="site_editing",
|
|
system_prompt_fragment=SYSTEM_PROMPT,
|
|
build_tools=_factory,
|
|
required_keys=[],
|
|
)
|