Ubuntu
428 lines
12 KiB
Python
428 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
OpenCLI Integration Adapter
|
|
将OpenCLI能力封装为Python接口
|
|
"""
|
|
|
|
import subprocess
|
|
import json
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Any
|
|
|
|
|
|
class OpenCLIAdapter:
|
|
"""OpenCLI适配器"""
|
|
|
|
def __init__(self, clis_dir: Optional[str] = None):
|
|
"""
|
|
初始化OpenCLI适配器
|
|
|
|
Args:
|
|
clis_dir: 适配器存储目录，默认 ~/.opencli/clis
|
|
"""
|
|
self.clis_dir = clis_dir or os.path.expanduser("~/.opencli/clis")
|
|
self._check_opencli()
|
|
|
|
def _check_opencli(self) -> bool:
|
|
"""检查OpenCLI是否已安装"""
|
|
if not shutil.which("opencli"):
|
|
raise RuntimeError(
|
|
"OpenCLI未安装。请先运行: npm install -g @jackwener/opencli"
|
|
)
|
|
return True
|
|
|
|
def _run_command(self, cmd: List[str], capture_output: bool = True) -> Dict[str, Any]:
|
|
"""
|
|
运行OpenCLI命令
|
|
|
|
Args:
|
|
cmd: 命令列表
|
|
capture_output: 是否捕获输出
|
|
|
|
Returns:
|
|
{"success": bool, "stdout": str, "stderr": str, "returncode": int}
|
|
"""
|
|
try:
|
|
result = subprocess.run(
|
|
["opencli"] + cmd,
|
|
capture_output=capture_output,
|
|
text=True,
|
|
timeout=120
|
|
)
|
|
|
|
return {
|
|
"success": result.returncode == 0,
|
|
"stdout": result.stdout,
|
|
"stderr": result.stderr,
|
|
"returncode": result.returncode
|
|
}
|
|
except subprocess.TimeoutExpired:
|
|
return {
|
|
"success": False,
|
|
"stdout": "",
|
|
"stderr": "Command timeout",
|
|
"returncode": -1
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"success": False,
|
|
"stdout": "",
|
|
"stderr": str(e),
|
|
"returncode": -1
|
|
}
|
|
|
|
# ==================== 核心功能 ====================
|
|
|
|
def setup(self) -> Dict[str, Any]:
|
|
"""
|
|
初始化OpenCLI环境
|
|
检查Browser Bridge连通性
|
|
"""
|
|
return self._run_command(["setup"])
|
|
|
|
def doctor(self) -> Dict[str, Any]:
|
|
"""
|
|
诊断OpenCLI状态
|
|
检查daemon、扩展和浏览器连接
|
|
"""
|
|
return self._run_command(["doctor"])
|
|
|
|
def list(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
列出所有可用的CLI适配器
|
|
|
|
Returns:
|
|
适配器列表
|
|
"""
|
|
result = self._run_command(["list", "--json"])
|
|
|
|
if result["success"]:
|
|
try:
|
|
return json.loads(result["stdout"])
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return []
|
|
|
|
def explore(self, url: str, output_dir: Optional[str] = None) -> Dict[str, Any]:
|
|
"""
|
|
探索网站API
|
|
AI自动发现网站接口
|
|
|
|
Args:
|
|
url: 目标网站URL
|
|
output_dir: 输出目录
|
|
|
|
Returns:
|
|
探索结果
|
|
"""
|
|
cmd = ["explore", url]
|
|
if output_dir:
|
|
cmd.extend(["--output", output_dir])
|
|
|
|
return self._run_command(cmd)
|
|
|
|
def synthesize(self, name: str, url: Optional[str] = None) -> Dict[str, Any]:
|
|
"""
|
|
生成适配器
|
|
基于探索结果生成CLI适配器
|
|
|
|
Args:
|
|
name: 适配器名称
|
|
url: 网站URL（如已explore可省略）
|
|
|
|
Returns:
|
|
生成结果
|
|
"""
|
|
cmd = ["synthesize", name]
|
|
if url:
|
|
cmd.extend(["--url", url])
|
|
|
|
return self._run_command(cmd)
|
|
|
|
def register(self, name: str, adapter_path: Optional[str] = None) -> Dict[str, Any]:
|
|
"""
|
|
注册适配器
|
|
|
|
Args:
|
|
name: 适配器名称
|
|
adapter_path: 适配器文件路径
|
|
|
|
Returns:
|
|
注册结果
|
|
"""
|
|
cmd = ["register", name]
|
|
if adapter_path:
|
|
cmd.append(adapter_path)
|
|
|
|
return self._run_command(cmd)
|
|
|
|
def execute(self, cli_name: str, command: str, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
执行CLI命令
|
|
|
|
Args:
|
|
cli_name: CLI名称
|
|
command: 命令名称
|
|
**kwargs: 命令参数
|
|
|
|
Returns:
|
|
执行结果
|
|
"""
|
|
cmd = [cli_name, command]
|
|
|
|
# 添加参数
|
|
for key, value in kwargs.items():
|
|
if isinstance(value, bool) and value:
|
|
cmd.append(f"--{key}")
|
|
else:
|
|
cmd.extend([f"--{key}", str(value)])
|
|
|
|
return self._run_command(cmd)
|
|
|
|
# ==================== 便捷方法 ====================
|
|
|
|
def search(self, cli_name: str, query: str, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
便捷搜索方法
|
|
|
|
Args:
|
|
cli_name: CLI名称
|
|
query: 搜索关键词
|
|
**kwargs: 其他参数
|
|
|
|
Returns:
|
|
搜索结果
|
|
"""
|
|
return self.execute(cli_name, "search", query=query, **kwargs)
|
|
|
|
def get(self, cli_name: str, id: str, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
便捷获取方法
|
|
|
|
Args:
|
|
cli_name: CLI名称
|
|
id: 资源ID
|
|
**kwargs: 其他参数
|
|
|
|
Returns:
|
|
获取结果
|
|
"""
|
|
return self.execute(cli_name, "get", id=id, **kwargs)
|
|
|
|
# ==================== Electron应用控制 ====================
|
|
|
|
def control_electron(self, app_name: str, action: str, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
控制Electron应用
|
|
|
|
Args:
|
|
app_name: 应用名称
|
|
action: 动作名称
|
|
**kwargs: 动作参数
|
|
|
|
Returns:
|
|
执行结果
|
|
"""
|
|
return self.execute(f"electron:{app_name}", action, **kwargs)
|
|
|
|
# ==================== 适配器管理 ====================
|
|
|
|
def create_yaml_adapter(self, name: str, config: Dict[str, Any]) -> str:
|
|
"""
|
|
创建YAML声明式适配器
|
|
|
|
Args:
|
|
name: 适配器名称
|
|
config: 适配器配置
|
|
|
|
Returns:
|
|
适配器文件路径
|
|
"""
|
|
import yaml
|
|
|
|
adapter_path = os.path.join(self.clis_dir, f"{name}.yaml")
|
|
os.makedirs(self.clis_dir, exist_ok=True)
|
|
|
|
with open(adapter_path, 'w', encoding='utf-8') as f:
|
|
yaml.dump(config, f, allow_unicode=True)
|
|
|
|
return adapter_path
|
|
|
|
def get_adapter_info(self, name: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
获取适配器信息
|
|
|
|
Args:
|
|
name: 适配器名称
|
|
|
|
Returns:
|
|
适配器信息
|
|
"""
|
|
# 检查YAML适配器
|
|
yaml_path = os.path.join(self.clis_dir, f"{name}.yaml")
|
|
if os.path.exists(yaml_path):
|
|
import yaml
|
|
with open(yaml_path, 'r', encoding='utf-8') as f:
|
|
return yaml.safe_load(f)
|
|
|
|
# 检查TypeScript适配器
|
|
ts_path = os.path.join(self.clis_dir, f"{name}.ts")
|
|
if os.path.exists(ts_path):
|
|
return {"type": "typescript", "path": ts_path}
|
|
|
|
return None
|
|
|
|
def remove_adapter(self, name: str) -> bool:
|
|
"""
|
|
删除适配器
|
|
|
|
Args:
|
|
name: 适配器名称
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
removed = False
|
|
|
|
yaml_path = os.path.join(self.clis_dir, f"{name}.yaml")
|
|
if os.path.exists(yaml_path):
|
|
os.remove(yaml_path)
|
|
removed = True
|
|
|
|
ts_path = os.path.join(self.clis_dir, f"{name}.ts")
|
|
if os.path.exists(ts_path):
|
|
os.remove(ts_path)
|
|
removed = True
|
|
|
|
return removed
|
|
|
|
# ==================== 状态检查 ====================
|
|
|
|
def is_ready(self) -> Dict[str, Any]:
|
|
"""
|
|
检查OpenCLI是否就绪
|
|
|
|
Returns:
|
|
状态信息
|
|
"""
|
|
checks = {
|
|
"opencli_installed": shutil.which("opencli") is not None,
|
|
"node_version": None,
|
|
"chrome_running": False,
|
|
"clis_dir_exists": os.path.exists(self.clis_dir),
|
|
}
|
|
|
|
# 检查Node版本
|
|
try:
|
|
result = subprocess.run(
|
|
["node", "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
checks["node_version"] = result.stdout.strip()
|
|
except:
|
|
pass
|
|
|
|
# 检查Chrome是否运行（简单检查）
|
|
try:
|
|
result = subprocess.run(
|
|
["pgrep", "-x", "chrome"],
|
|
capture_output=True,
|
|
timeout=2
|
|
)
|
|
checks["chrome_running"] = result.returncode == 0
|
|
except:
|
|
pass
|
|
|
|
checks["ready"] = (
|
|
checks["opencli_installed"] and
|
|
checks["node_version"] and
|
|
checks["clis_dir_exists"]
|
|
)
|
|
|
|
return checks
|
|
|
|
|
|
# ==================== 命令行接口 ====================
|
|
|
|
def main():
|
|
"""命令行入口"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='OpenCLI Adapter')
|
|
subparsers = parser.add_subparsers(dest='command', help='可用命令')
|
|
|
|
# setup
|
|
subparsers.add_parser('setup', help='初始化OpenCLI')
|
|
|
|
# doctor
|
|
subparsers.add_parser('doctor', help='诊断OpenCLI状态')
|
|
|
|
# list
|
|
subparsers.add_parser('list', help='列出可用适配器')
|
|
|
|
# explore
|
|
explore_parser = subparsers.add_parser('explore', help='探索网站API')
|
|
explore_parser.add_argument('url', help='目标网站URL')
|
|
|
|
# synthesize
|
|
synth_parser = subparsers.add_parser('synthesize', help='生成适配器')
|
|
synth_parser.add_argument('name', help='适配器名称')
|
|
synth_parser.add_argument('--url', help='网站URL')
|
|
|
|
# exec
|
|
exec_parser = subparsers.add_parser('exec', help='执行CLI命令')
|
|
exec_parser.add_argument('cli', help='CLI名称')
|
|
exec_parser.add_argument('action', help='动作名称')
|
|
exec_parser.add_argument('--args', nargs='*', help='额外参数')
|
|
|
|
# status
|
|
subparsers.add_parser('status', help='检查状态')
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
return
|
|
|
|
adapter = OpenCLIAdapter()
|
|
|
|
if args.command == 'setup':
|
|
result = adapter.setup()
|
|
print(result["stdout"] if result["success"] else result["stderr"])
|
|
|
|
elif args.command == 'doctor':
|
|
result = adapter.doctor()
|
|
print(result["stdout"] if result["success"] else result["stderr"])
|
|
|
|
elif args.command == 'list':
|
|
adapters = adapter.list()
|
|
print(json.dumps(adapters, indent=2, ensure_ascii=False))
|
|
|
|
elif args.command == 'explore':
|
|
print(f"正在探索: {args.url}")
|
|
result = adapter.explore(args.url)
|
|
print(result["stdout"] if result["success"] else result["stderr"])
|
|
|
|
elif args.command == 'synthesize':
|
|
print(f"正在生成适配器: {args.name}")
|
|
result = adapter.synthesize(args.name, args.url)
|
|
print(result["stdout"] if result["success"] else result["stderr"])
|
|
|
|
elif args.command == 'exec':
|
|
print(f"执行: {args.cli} {args.action}")
|
|
result = adapter.execute(args.cli, args.action)
|
|
print(result["stdout"] if result["success"] else result["stderr"])
|
|
|
|
elif args.command == 'status':
|
|
status = adapter.is_ready()
|
|
print(json.dumps(status, indent=2, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|