353 lines
14 KiB
Python
353 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
GitLink CLI Demo Server — 子任务一 交互式展示
|
||
启动后访问 http://127.0.0.1:8765
|
||
"""
|
||
|
||
import http.server
|
||
import json
|
||
import subprocess
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import shutil
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
PORT = 8765
|
||
WORKING_DIR = r"D:\code\SE\Evolution_and_Maintenance_of_SE\Mission2\gitlink-cli"
|
||
HTML_DIR = Path(__file__).parent
|
||
|
||
# ── 定位 Claude Code CLI ─────────────────────────
|
||
def _find_claude() -> str | None:
|
||
"""查找 claude 可执行文件，优先返回 .exe 以避开 .cmd 的编码问题"""
|
||
# 1) 直接找 claude.exe（npm 全局安装路径）
|
||
candidates = [
|
||
# npm global on Windows
|
||
Path(os.environ.get("APPDATA", "")) / r"npm\node_modules\@anthropic-ai\claude-code\bin\claude.exe",
|
||
# 直接 which（可能返回 .cmd）
|
||
shutil.which("claude"),
|
||
shutil.which("claude.exe"),
|
||
]
|
||
for p in candidates:
|
||
if p and Path(str(p)).is_file():
|
||
return str(p)
|
||
|
||
# 2) 尝试 shutil.which 返回的 .cmd 对应的 .exe
|
||
cmd = shutil.which("claude.cmd")
|
||
if cmd:
|
||
exe = Path(cmd).with_suffix(".exe")
|
||
if exe.is_file():
|
||
return str(exe)
|
||
|
||
return None
|
||
|
||
CLAUDE_EXE = _find_claude()
|
||
|
||
# 全局：当前正在运行的进程（用于取消）
|
||
_proc_lock = threading.Lock()
|
||
_current_proc = None
|
||
|
||
|
||
class DemoHandler(http.server.BaseHTTPRequestHandler):
|
||
"""HTTP 请求处理器：静态文件 + API 端点"""
|
||
|
||
# ── 日志 ────────────────────────────────────────────
|
||
def log_message(self, fmt, *args):
|
||
print(f"[{self.log_date_time_string()}] {args[0]}")
|
||
|
||
# ── GET ─────────────────────────────────────────────
|
||
def do_GET(self):
|
||
path = self.path.split("?")[0]
|
||
if path in ("/", "/index.html"):
|
||
self._serve_file("index.html", "text/html; charset=utf-8")
|
||
else:
|
||
self.send_error(404)
|
||
|
||
# ── POST ────────────────────────────────────────────
|
||
def do_POST(self):
|
||
if self.path == "/api/exec":
|
||
self._handle_exec()
|
||
elif self.path == "/api/claude":
|
||
self._handle_claude()
|
||
elif self.path == "/api/cancel":
|
||
self._handle_cancel()
|
||
else:
|
||
self.send_error(404)
|
||
|
||
# ── OPTIONS (CORS preflight) ────────────────────────
|
||
def do_OPTIONS(self):
|
||
self._send_cors_headers()
|
||
self.send_response(204)
|
||
self.end_headers()
|
||
|
||
# ── 取消当前命令 ────────────────────────────────────
|
||
def _handle_cancel(self):
|
||
global _current_proc
|
||
with _proc_lock:
|
||
proc = _current_proc
|
||
if proc is None or proc.poll() is not None:
|
||
self._send_json({"ok": True, "message": "没有正在运行的命令"})
|
||
return
|
||
try:
|
||
proc.kill()
|
||
self._send_json({"ok": True, "message": "已发送终止信号"})
|
||
except Exception as e:
|
||
self._send_json({"ok": False, "error": str(e)})
|
||
|
||
# ── Claude Code 执行 ────────────────────────────────
|
||
def _handle_claude(self):
|
||
global _current_proc
|
||
|
||
content_length = int(self.headers.get("Content-Length", 0))
|
||
body = self.rfile.read(content_length)
|
||
try:
|
||
data = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
self._send_json({"ok": False, "error": "Invalid JSON body"})
|
||
return
|
||
|
||
prompt = data.get("prompt", "").strip()
|
||
if not prompt:
|
||
self._send_json({"ok": False, "error": "Empty prompt", "session_id": data.get("session_id", "")})
|
||
return
|
||
|
||
timeout = min(data.get("timeout", 300), 600) # max 10 min for Claude
|
||
|
||
if not CLAUDE_EXE:
|
||
self._send_json({
|
||
"ok": False, "stdout": "",
|
||
"stderr": "❌ 找不到 Claude Code CLI（claude.exe）。请确认已通过 npm 安装：npm install -g @anthropic-ai/claude-code",
|
||
"exit_code": -1,
|
||
"session_id": "",
|
||
})
|
||
return
|
||
|
||
# ── 会话管理：支持多轮对话 ──
|
||
session_id = data.get("session_id", "").strip()
|
||
new_session = data.get("new_session", False)
|
||
|
||
if new_session or not session_id:
|
||
# 新会话：生成 UUID
|
||
session_id = str(uuid.uuid4())
|
||
args = [CLAUDE_EXE, "-p", prompt, "--session-id", session_id, "--permission-mode", "bypassPermissions"]
|
||
else:
|
||
# 继续已有会话
|
||
args = [CLAUDE_EXE, "-p", prompt, "--resume", session_id, "--permission-mode", "bypassPermissions"]
|
||
|
||
proc = None
|
||
try:
|
||
proc = subprocess.Popen(
|
||
args,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
cwd=WORKING_DIR,
|
||
env={**os.environ, "NO_COLOR": "1"},
|
||
)
|
||
|
||
# 注册为当前进程（允许取消）
|
||
with _proc_lock:
|
||
_current_proc = proc
|
||
|
||
try:
|
||
stdout, stderr = proc.communicate(timeout=timeout)
|
||
exit_code = proc.returncode
|
||
killed = False
|
||
except subprocess.TimeoutExpired:
|
||
proc.kill()
|
||
stdout, stderr = proc.communicate()
|
||
exit_code = -1
|
||
killed = True
|
||
|
||
if killed:
|
||
self._send_json({
|
||
"ok": False,
|
||
"stdout": stdout or "",
|
||
"stderr": f"❌ Claude Code 执行超时（超过 {timeout} 秒）已被终止",
|
||
"exit_code": -1,
|
||
"session_id": session_id,
|
||
})
|
||
else:
|
||
self._send_json({
|
||
"ok": exit_code == 0,
|
||
"stdout": stdout,
|
||
"stderr": stderr,
|
||
"exit_code": exit_code,
|
||
"session_id": session_id,
|
||
})
|
||
|
||
except FileNotFoundError:
|
||
self._send_json({
|
||
"ok": False, "stdout": "",
|
||
"stderr": "❌ 找不到 Claude Code CLI（claude.exe）。请确认已通过 npm 安装：npm install -g @anthropic-ai/claude-code",
|
||
"exit_code": -1,
|
||
"session_id": session_id,
|
||
})
|
||
except Exception as e:
|
||
self._send_json({
|
||
"ok": False, "stdout": "",
|
||
"stderr": f"❌ {e}", "exit_code": -1,
|
||
"session_id": session_id,
|
||
})
|
||
finally:
|
||
with _proc_lock:
|
||
if _current_proc is proc:
|
||
_current_proc = None
|
||
|
||
# ── 命令执行 ────────────────────────────────────────
|
||
def _handle_exec(self):
|
||
global _current_proc
|
||
|
||
content_length = int(self.headers.get("Content-Length", 0))
|
||
body = self.rfile.read(content_length)
|
||
try:
|
||
data = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
self._send_json({"ok": False, "error": "Invalid JSON body"})
|
||
return
|
||
|
||
command = data.get("command", "").strip()
|
||
if not command:
|
||
self._send_json({"ok": False, "error": "Empty command"})
|
||
return
|
||
|
||
timeout = min(data.get("timeout", 120), 300) # max 5 min
|
||
|
||
tmp = None
|
||
proc = None
|
||
try:
|
||
tmp = tempfile.NamedTemporaryFile(
|
||
mode="w", suffix=".ps1", delete=False, encoding="utf-8-sig"
|
||
)
|
||
tmp.write(command)
|
||
tmp.close()
|
||
|
||
proc = subprocess.Popen(
|
||
[
|
||
"powershell",
|
||
"-ExecutionPolicy", "Bypass",
|
||
"-NoProfile",
|
||
"-File", tmp.name,
|
||
],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
cwd=WORKING_DIR,
|
||
env={**os.environ, "NO_COLOR": "1"},
|
||
)
|
||
|
||
# 注册为当前进程（允许取消）
|
||
with _proc_lock:
|
||
_current_proc = proc
|
||
|
||
try:
|
||
stdout, stderr = proc.communicate(timeout=timeout)
|
||
exit_code = proc.returncode
|
||
killed = False
|
||
except subprocess.TimeoutExpired:
|
||
proc.kill()
|
||
stdout, stderr = proc.communicate()
|
||
exit_code = -1
|
||
killed = True
|
||
|
||
if killed:
|
||
self._send_json({
|
||
"ok": False,
|
||
"stdout": stdout or "",
|
||
"stderr": f"❌ 命令执行超时（超过 {timeout} 秒）已被终止",
|
||
"exit_code": -1,
|
||
})
|
||
else:
|
||
self._send_json({
|
||
"ok": exit_code == 0,
|
||
"stdout": stdout,
|
||
"stderr": stderr,
|
||
"exit_code": exit_code,
|
||
})
|
||
|
||
except FileNotFoundError:
|
||
self._send_json({
|
||
"ok": False, "stdout": "",
|
||
"stderr": "❌ 找不到 PowerShell，请确认系统已安装 PowerShell",
|
||
"exit_code": -1,
|
||
})
|
||
except Exception as e:
|
||
self._send_json({
|
||
"ok": False, "stdout": "",
|
||
"stderr": f"❌ {e}", "exit_code": -1,
|
||
})
|
||
finally:
|
||
with _proc_lock:
|
||
if _current_proc is proc:
|
||
_current_proc = None
|
||
if tmp:
|
||
try:
|
||
os.unlink(tmp.name)
|
||
except OSError:
|
||
pass
|
||
|
||
# ── 辅助方法 ────────────────────────────────────────
|
||
def _serve_file(self, filename: str, content_type: str):
|
||
filepath = HTML_DIR / filename
|
||
try:
|
||
with open(filepath, "rb") as f:
|
||
content = f.read()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(content)))
|
||
self.send_header("Cache-Control", "no-cache")
|
||
self._send_cors_headers()
|
||
self.end_headers()
|
||
self.wfile.write(content)
|
||
except FileNotFoundError:
|
||
self.send_error(404)
|
||
|
||
def _send_json(self, data: dict):
|
||
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self._send_cors_headers()
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def _send_cors_headers(self):
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||
|
||
|
||
def main():
|
||
if not os.path.isdir(WORKING_DIR):
|
||
print(f"⚠️ 警告：工作目录不存在 — {WORKING_DIR}")
|
||
print(" 请修改 server.py 中的 WORKING_DIR 变量")
|
||
sys.exit(1)
|
||
|
||
server = http.server.ThreadingHTTPServer(("127.0.0.1", PORT), DemoHandler)
|
||
|
||
claude_status = f"✅ {CLAUDE_EXE}" if CLAUDE_EXE else "❌ 未找到 claude.exe（Skills 功能不可用）"
|
||
print(f"""
|
||
╔════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
|
||
║ GitLink CLI 作品展示 · 交互式平台 ║
|
||
╠════════════════════════════════════════════════════════════════════════════════════════════════════════════╣
|
||
║ 打开浏览器访问: http://127.0.0.1:{PORT} ║
|
||
║ 工作目录: {WORKING_DIR} ║
|
||
║ Claude Code: {claude_status:<55} ║
|
||
║ 按 Ctrl+C 停止服务器 ║
|
||
╚════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
|
||
""")
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
print("\n👋 服务器已停止。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|