基础api服务可进行,可扫描网络交换机，需用设备进一步调试

src/backend/.gitignore

@@ -1,7 +1,10 @@
+# Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]
 *$py.class
 *.so
+
+# Python packaging
 .Python
 build/
 develop-eggs/
@@ -23,6 +26,8 @@ MANIFEST
 # IDE
 .idea/
 .vscode/
+*.swp
+*.swo
 
 # Environment
 .env
@@ -35,7 +40,10 @@ venv.bak/
 
 # Logs
 *.log
+*.log.*
+logs/
 
-# 忽略.env文件
-.env
-*.secret
+# Secrets
+*.secret
+*.key
+*.pem

src/backend/Dockerfile

@@ -2,16 +2,21 @@ FROM python:3.13-slim
 
 WORKDIR /app
 
+# 1. 先复制依赖文件并安装
 COPY ./requirements.txt /app/requirements.txt
-
 RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt
 
+# 2. 复制项目代码（排除 .env 和缓存文件）
 COPY . /app
 
-ENV PYTHONPATH=/app
-ENV PORT=8000
-ENV HOST=0.0.0.0
-.env
-*.secret
+# 3. 环境变量配置
+ENV PYTHONPATH=/app \
+    PORT=8000 \
+    HOST=0.0.0.0
 
-CMD ["uvicorn", "run:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
+# 4. 安全设置
+RUN find /app -name "*.pyc" -delete && \
+    find /app -name "__pycache__" -exec rm -rf {} +
+
+# 5. 启动命令（修正路径）
+CMD ["uvicorn", "src.backend.app:app", "--host", "0.0.0.0", "--port", "8000"]

src/backend/app/__init__.py

@@ -4,7 +4,7 @@ from src.backend.app.utils.logger import setup_logging
 from src.backend.config import settings
 
 app = FastAPI()
-app.include_router(router)
+app.include_router(router,prefix="/api")
 
 def create_app() -> FastAPI:
     # 设置日志

src/backend/app/api/endpoints.py

@@ -2,16 +2,36 @@ from fastapi import APIRouter, Depends, HTTPException
 from typing import Any
 from pydantic import BaseModel
 
-from src.backend.app.services.ai_services import AIService
-from src.backend.app.api.network_config import SwitchConfigurator
-from src.backend.config import settings
+from ...app.services.ai_services import AIService
+from ...app.api.network_config import SwitchConfigurator
+from ...config import settings
+from ..services.network_scanner import NetworkScanner
 
 router = APIRouter(prefix="/api", tags=["API"])
+scanner = NetworkScanner()
 
 @router.get("/test")
 async def test_endpoint():
     return {"message": "Hello World"}
 
+@router.get("/scan_network", summary="扫描网络中的交换机")
+async def scan_network(subnet: str = "192.168.1.0/24"):
+    try:
+        devices = scanner.scan_subnet(subnet)
+        return {
+            "success": True,
+            "devices": devices,
+            "count": len(devices)
+        }
+    except Exception as e:
+        raise HTTPException(500, f"扫描失败: {str(e)}")
+
+@router.get("/list_devices", summary="列出已发现的交换机")
+async def list_devices():
+    return {
+        "devices": scanner.load_cached_devices()
+    }
+
 class CommandRequest(BaseModel):
     command: str
 

src/backend/app/services/network_scanner.py

@@ -0,0 +1,49 @@
+import nmap
+import json
+from pathlib import Path
+from typing import List, Dict
+from ..utils.logger import logger
+
+
+class NetworkScanner:
+    def __init__(self, cache_path: str = "switch_devices.json"):
+        self.cache_path = Path(cache_path)
+        self.nm = nmap.PortScanner()
+
+    def scan_subnet(self, subnet: str = "192.168.1.0/24") -> List[Dict]:
+        """扫描指定子网的交换机设备"""
+        logger.info(f"Scanning subnet: {subnet}")
+
+        # 扫描开放22(SSH)或23(Telnet)端口的设备
+        self.nm.scan(
+            hosts=subnet,
+            arguments="-p 22,23 --open -T4"
+        )
+
+        devices = []
+        for host in self.nm.all_hosts():
+            if self.nm[host].state() == "up":
+                device = {
+                    "ip": host,
+                    "ports": list(self.nm[host]["tcp"].keys()),
+                    "mac": self.nm[host].get("addresses", {}).get("mac", "unknown")
+                }
+                devices.append(device)
+                logger.debug(f"Found device: {device}")
+
+        self._save_to_cache(devices)
+        return devices
+
+    def _save_to_cache(self, devices: List[Dict]):
+        """保存扫描结果到本地文件"""
+        with open(self.cache_path, "w") as f:
+            json.dump(devices, f, indent=2)
+        logger.info(f"Saved {len(devices)} devices to cache")
+
+    def load_cached_devices(self) -> List[Dict]:
+        """从缓存加载设备列表"""
+        if not self.cache_path.exists():
+            return []
+
+        with open(self.cache_path) as f:
+            return json.load(f)

src/backend/requirements.txt

@@ -4,4 +4,5 @@ python-dotenv==1.0.0
 requests==2.28.2
 paramiko==3.1.0
 pydantic==1.10.7
-loguru==0.7.0
+loguru==0.7.0
+python-nmap==0.7.1