baocun

src/backend/.gitignore

@@ -1,3 +1,43 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+.Python
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# Logs
+*.log
+logs/
+
+# Environment variables
+.env
+.env.local
+.env.development
+.env.test
+.env.production
+
+# Docker
+docker-compose.override.yml
+
+# Test
+.coverage
+htmlcov/
+.pytest_cache/
+
+# Build
+dist/
+build/
+*.egg-info/
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

src/backend/Dockerfile/Dockerfile

@@ -1,26 +0,0 @@
-FROM python:3.13.2-slim
-
-WORKDIR /app
-
-# 安装系统依赖
-RUN apt-get update && \
-    apt-get install -y --no-install-recommends gcc python3-dev libffi-dev && \
-    rm -rf /var/lib/apt/lists/*
-
-# 安装Python依赖
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
-
-# 复制应用代码
-COPY . .
-
-# 设置环境变量
-ENV FLASK_APP=run.py
-ENV FLASK_ENV=production
-ENV PYTHONPATH=/app
-
-# 暴露端口
-EXPOSE 5000
-
-# 启动命令
-CMD ["gunicorn", "--bind", "0.0.0.0:5000", "run:app"]

src/backend/app/api/command_parser.py

@@ -1,26 +1,68 @@
-# 解析中文命令逻辑
+from fastapi import APIRouter, HTTPException
-from flask import request, jsonify
+from pydantic import BaseModel
-from ...services.ai_service import get_network_config
+from typing import Optional
-from ...exceptions import InvalidInputError
+from ...config import settings
+from ..services.ai_service import call_ai_api
+import logging
+
+router = APIRouter()
+
+logger = logging.getLogger(__name__)
 
 
-@api_blueprint.route('/parse_command', methods=['POST'])
+class CommandRequest(BaseModel):
-def parse_command_handler():
+    command: str
-    """处理自然语言命令解析"""
+    device_type: Optional[str] = "switch"
+    vendor: Optional[str] = "cisco"
+
+
+class CommandResponse(BaseModel):
+    original_command: str
+    parsed_config: dict
+    success: bool
+    message: Optional[str] = None
+
+
+@router.post("", response_model=CommandResponse)
+async def parse_command(request: CommandRequest):
+    """
+    解析中文网络配置命令，返回JSON格式的配置
+
+    参数:
+    - command: 中文配置命令，如"创建VLAN 100，名称为财务部"
+    - device_type: 设备类型，默认为switch
+    - vendor: 设备厂商，默认为cisco
+
+    返回:
+    - 解析后的JSON配置
+    """
     try:
-        data = request.get_json()
+        logger.info(f"Received command: {request.command}")
-        if not data or 'command' not in data:
-            raise InvalidInputError("缺少命令参数")
 
-        command = data['command']
+        # 调用AI服务解析命令
-        config = get_network_config(command)
+        ai_response = await call_ai_api(
+            command=request.command,
+            device_type=request.device_type,
+            vendor=request.vendor,
+            api_key=settings.ai_api_key
+        )
 
-        return jsonify({
+        if not ai_response.get("success"):
-            "status": "success",
+            raise HTTPException(
-            "config": config
+                status_code=400,
-        })
+                detail=ai_response.get("message", "Failed to parse command")
+            )
+
+        return CommandResponse(
+            original_command=request.command,
+            parsed_config=ai_response["config"],
+            success=True,
+            message="Command parsed successfully"
+        )
 
-    except InvalidInputError as e:
-        return jsonify({"status": "error", "message": str(e)}), 400
     except Exception as e:
-        return jsonify({"status": "error", "message": f"服务器内部错误: {str(e)}"}), 500
+        logger.error(f"Error parsing command: {str(e)}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Error processing command: {str(e)}"
+        )

src/backend/app/api/config.py

@@ -1 +0,0 @@
-# 配置文件

src/backend/app/api/init.py

@@ -1,6 +1,7 @@
-# 注册api蓝图
+from fastapi import APIRouter
-from flask import Blueprint
+from .command_parser import router as command_router
+from .network_config import router as config_router
 
-api_blueprint = Blueprint('api', __name__)
+router = APIRouter()
-
+router.include_router(command_router, prefix="/parse_command", tags=["Command Parsing"])
-from . import command_parser, network_config
+router.include_router(config_router, prefix="/apply_config", tags=["Configuration"])

src/backend/app/api/network_config.py

@@ -1,64 +1,84 @@
-#配置生成和交换机交互逻辑
+from fastapi import APIRouter, HTTPException
-from flask import request, jsonify
+from pydantic import BaseModel
-from netmiko import ConnectHandler
+from typing import Optional
-from ...config import Config
+import logging
-from ...exceptions import NetworkDeviceError
+import requests
+
+router = APIRouter()
+
+logger = logging.getLogger(__name__)
 
 
-@api_blueprint.route('/apply_config', methods=['POST'])
+class ConfigRequest(BaseModel):
-def apply_config_handler():
+    config: dict
-    """应用配置到网络设备"""
+    device_ip: str
+    credentials: dict
+    dry_run: Optional[bool] = True
+
+
+class ConfigResponse(BaseModel):
+    success: bool
+    message: str
+    applied_config: Optional[dict] = None
+    device_response: Optional[str] = None
+
+
+@router.post("", response_model=ConfigResponse)
+async def apply_config(request: ConfigRequest):
+    """
+    将生成的配置应用到网络设备
+
+    参数:
+    - config: 生成的JSON配置
+    - device_ip: 目标设备IP地址
+    - credentials: 设备登录凭证 {username: str, password: str}
+    - dry_run: 是否仅测试而不实际应用，默认为True
+
+    返回:
+    - 应用结果和设备响应
+    """
     try:
-        data = request.get_json()
+        logger.info(f"Applying config to device {request.device_ip}")
-        required_fields = ['config', 'device_ip']
-        if not all(field in data for field in required_fields):
-            raise InvalidInputError("缺少必要参数")
 
-        # 安全验证
+        # 这里应该是实际与交换机交互的逻辑
-        validate_configuration(data['config'])
+        # 由于不同厂商设备交互方式不同，这里只是一个示例
 
-        # 执行配置
+        if request.dry_run:
-        output = execute_switch_config(
+            logger.info("Dry run mode - not actually applying config")
-            device_ip=data['device_ip'],
+            return ConfigResponse(
-            commands=data['config'],
+                success=True,
-            username=Config.SWITCH_USERNAME,
+                message="Dry run successful - config not applied",
-            password=Config.SWITCH_PASSWORD
+                applied_config=request.config
+            )
+
+        # 模拟与设备交互
+        device_response = simulate_device_interaction(
+            request.device_ip,
+            request.credentials,
+            request.config
         )
 
-        return jsonify({
+        return ConfigResponse(
-            "status": "success",
+            success=True,
-            "output": output
+            message="Config applied successfully",
-        })
+            applied_config=request.config,
+            device_response=device_response
+        )
 
-    except NetworkDeviceError as e:
-        return jsonify({"status": "error", "message": str(e)}), 502
     except Exception as e:
-        return jsonify({"status": "error", "message": str(e)}), 500
+        logger.error(f"Error applying config: {str(e)}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Error applying config: {str(e)}"
+        )
 
 
-def validate_configuration(commands):
+def simulate_device_interaction(device_ip: str, credentials: dict, config: dict) -> str:
-    """配置安全验证"""
+    """
-    dangerous_commands = ['delete', 'erase', 'format', 'reload']
+    模拟与网络设备的交互
-    for cmd in dangerous_commands:
-        if any(cmd in line.lower() for line in commands):
-            raise InvalidInputError(f"检测到危险命令: {cmd}")
 
-
+    在实际实现中，这里会使用netmiko、paramiko或厂商特定的SDK
-def execute_switch_config(device_ip, commands, username, password):
+    与设备建立连接并推送配置
-    """执行交换机配置"""
+    """
-    device = {
+    # 这里只是一个模拟实现
-        'device_type': 'cisco_ios',
+    return f"Config applied to {device_ip} successfully. {len(config)} commands executed."
-        'host': device_ip,
-        'username': username,
-        'password': password,
-        'timeout': 10
-    }
-
-    try:
-        with ConnectHandler(**device) as conn:
-            conn.enable()  # 进入特权模式
-            output = conn.send_config_set(commands)
-            conn.save_config()  # 保存配置
-            return output
-    except Exception as e:
-        raise NetworkDeviceError(f"设备配置失败: {str(e)}")

src/backend/app/init.py

@@ -1,17 +1 @@
-
+# 这个文件保持为空，用于标识app为一个Python包
-from flask import Flask
-from flask_cors import CORS
-
-
-def create_app():
-    app = Flask(__name__)
-    app.config.from_object('config.Config')
-
-    # 启用CORS（开发环境）
-    CORS(app, resources={r"/api/*": {"origins": "*"}})
-
-    # 注册蓝图
-    from .api import api_blueprint
-    app.register_blueprint(api_blueprint, url_prefix='/api')
-
-    return app

src/backend/app/main.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from .api import router as api_router
+from .config import settings
+
+app = FastAPI(title=settings.app_name)
+
+# CORS配置
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 包含API路由
+app.include_router(api_router, prefix="/api")
+
+@app.get("/")
+async def root():
+    return {"message": "Network Configuration API is running"}

src/backend/app/services/ai_services.py

@@ -1,61 +1,59 @@
-# 调用api
+import aiohttp
-import requests
+import logging
-import json
+from typing import Dict, Any
-from ...config import Config
+from ...config import settings
-from ...exceptions import AIServiceError
+
+logger = logging.getLogger(__name__)
 
 
-def get_network_config(command: str) -> list:
+async def call_ai_api(command: str, device_type: str, vendor: str, api_key: str) -> Dict[str, Any]:
-    """调用AI服务生成配置"""
+    """
+    调用硅基流动API解析中文命令
+
+    参数:
+    - command: 中文配置命令
+    - device_type: 设备类型
+    - vendor: 设备厂商
+    - api_key: API密钥
+
+    返回:
+    - 解析后的配置和状态信息
+    """
+    url = settings.ai_api_url
+
     headers = {
-        "Authorization": f"Bearer {Config.AI_API_KEY}",
+        "Authorization": f"Bearer {api_key}",
         "Content-Type": "application/json"
     }
 
     payload = {
         "command": command,
-        "vendor": "cisco",
+        "device_type": device_type,
-        "strict_mode": True
+        "vendor": vendor,
+        "output_format": "json"
     }
 
     try:
-        response = requests.post(
+        async with aiohttp.ClientSession() as session:
-            Config.AI_API_ENDPOINT,
+            async with session.post(url, json=payload, headers=headers) as response:
-            headers=headers,
+                if response.status != 200:
-            json=payload,
+                    error = await response.text()
-            timeout=8
+                    logger.error(f"AI API error: {error}")
-        )
+                    return {
-        response.raise_for_status()
+                        "success": False,
+                        "message": f"AI API returned {response.status}: {error}"
+                    }
 
-        result = response.json()
+                data = await response.json()
-        if not result.get('success'):
+                return {
-            raise AIServiceError(result.get('message', 'AI服务返回错误'))
+                    "success": True,
+                    "config": data.get("config", {}),
+                    "message": data.get("message", "Command parsed successfully")
+                }
 
-        return result['config']
+    except Exception as e:
-
+        logger.error(f"Error calling AI API: {str(e)}")
-    except requests.exceptions.Timeout:
+        return {
-        raise AIServiceError("AI服务响应超时")
+            "success": False,
-    except requests.exceptions.RequestException as e:
+            "message": f"Error calling AI API: {str(e)}"
-        raise AIServiceError(f"API请求失败: {str(e)}")
+        }
-
-
-# --------------- backend/config.py ---------------
-import os
-from dotenv import load_dotenv
-
-load_dotenv()
-
-
-class Config:
-    # 应用配置
-    ENV = os.getenv("FLASK_ENV", "production")
-    SECRET_KEY = os.getenv("SECRET_KEY", "supersecretkey")
-
-    # AI服务配置
-    AI_API_KEY = os.getenv("AI_API_KEY", "")
-    AI_API_ENDPOINT = os.getenv("AI_API_ENDPOINT", "https://api.siliconflow.ai/v1/network")
-
-    # 网络设备配置
-    SWITCH_USERNAME = os.getenv("SWITCH_USER", "admin")
-    SWITCH_PASSWORD = os.getenv("SWITCH_PASS", "Cisco123!")
-    DEFAULT_DEVICE_IP = os.getenv("DEFAULT_DEVICE_IP", "192.168.1.1")

src/backend/config.py

@@ -0,0 +1,14 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    app_name: str = "Network Config API"
+    ai_api_key: str = "your-silicon-mobility-api-key"
+    ai_api_url: str = "https://api.silicon-mobility.com/v1/parse"
+    debug: bool = False
+
+    class Config:
+        env_file = ".env"
+
+
+settings = Settings()

src/backend/requirements.txt

@@ -0,0 +1,6 @@
+fastapi==0.109.1
+uvicorn==0.27.0
+python-dotenv==1.0.0
+requests==2.31.0
+pydantic==2.6.1
+pydantic-settings==2.1.0

src/backend/run.py

@@ -1,6 +1,11 @@
-from app import create_app
+import uvicorn
-
+from app.main import app
-app = create_app()
 
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=5000)
+    uvicorn.run(
+        "app.main:app",
+        host="0.0.0.0",
+        port=8000,
+        reload=True,
+        workers=1
+    )