Jerry/AI-powered-switches
1
1
Fork 0
mirror of https://github.com/Jerryplusy/AI-powered-switches.git synced 2025-10-14 09:49:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
Jerry 2025-08-30 15:47:09 +08:00
parent 29dd4ec839
commit 842e562b91
2 changed files with 37 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
src/backend/app/api/endpoints.py
View File

@ -1,18 +1,15 @@
# File: D:\Python work\AI-powered-switches\src\backend\app\api\endpoints.py
import socket import socket
from datetime import datetime, timedelta from datetime import datetime, timedelta
from fastapi import (APIRouter, HTTPException, Response, WebSocket, WebSocketDisconnect) from fastapi import (APIRouter, HTTPException, Response, WebSocket, WebSocketDisconnect)
from typing import List from typing import List
from pydantic import BaseModel from pydantic import BaseModel
import asyncio import asyncio
from fastapi.responses import HTMLResponse, JSONResponse from fastapi.responses import HTMLResponse
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
import io import io
import base64 import base64
import psutil import psutil
import ipaddress import ipaddress
import json
from ..services.switch_traffic_monitor import get_switch_monitor from ..services.switch_traffic_monitor import get_switch_monitor
from ..utils import logger from ..utils import logger
@ -23,9 +20,6 @@ from ..services.network_scanner import NetworkScanner
from ...app.services.traffic_monitor import traffic_monitor from ...app.services.traffic_monitor import traffic_monitor
from ...app.models.traffic_models import TrafficRecord, SwitchTrafficRecord from ...app.models.traffic_models import TrafficRecord, SwitchTrafficRecord
from src.backend.app.api.database import SessionLocal from src.backend.app.api.database import SessionLocal
from ..services.network_visualizer import NetworkVisualizer
from ..services.config_validator import ConfigValidator
from ..services.report_generator import ReportGenerator
router = APIRouter(prefix="", tags=["API"]) router = APIRouter(prefix="", tags=["API"])
scanner = NetworkScanner() scanner = NetworkScanner()
@ -451,69 +445,3 @@ async def get_network_adapters():
return {"networks": networks} return {"networks": networks}
except Exception as e: except Exception as e:
return {"error": f"获取网络适配器信息失败: {str(e)}"} return {"error": f"获取网络适配器信息失败: {str(e)}"}
visualizer = NetworkVisualizer()
report_gen = ReportGenerator()
@router.get("/topology/visualize", response_class=HTMLResponse)
async def visualize_topology():
"""获取网络拓扑可视化图"""
try:
devices = await list_devices()
await asyncio.to_thread(visualizer.update_topology, devices["devices"])
image_data = await asyncio.to_thread(visualizer.generate_topology_image)
return f"""
<html>
<head><title>Network Topology</title></head>
<body>
<h1>Network Topology</h1>
<img src="data:image/png;base64,{image_data}" alt="Network Topology">
</body>
</html>
"""
except Exception as e:
raise HTTPException(500, detail=str(e))
@router.post("/config/validate")
async def validate_config(config: dict):
"""验证配置有效性"""
is_valid, errors = await asyncio.to_thread(ConfigValidator.validate_full_config, config)
return {
"valid": is_valid,
"errors": errors,
"has_security_risks": len(
await asyncio.to_thread(ConfigValidator.check_security_risks, config.get("commands", []))) > 0
}
@router.get("/reports/traffic/{ip}")
async def get_traffic_report(ip: str, days: int = 1):
"""获取流量分析报告"""
try:
report = await asyncio.to_thread(report_gen.generate_traffic_report, ip, days)
return JSONResponse(content=report)
except Exception as e:
raise HTTPException(500, detail=str(e))
@router.get("/reports/traffic")
async def get_local_traffic_report(days: int = 1):
"""获取本地网络流量报告"""
try:
report = await asyncio.to_thread(report_gen.generate_traffic_report, days=days)
return JSONResponse(content=report)
except Exception as e:
raise HTTPException(500, detail=str(e))
@router.get("/topology/traffic_heatmap")
async def get_traffic_heatmap(minutes: int = 10):
"""获取流量热力图数据"""
try:
heatmap = await asyncio.to_thread(visualizer.get_traffic_heatmap, minutes)
return {"heatmap": heatmap}
except Exception as e:
raise HTTPException(500, detail=str(e))

47
src/backend/config.py
View File

@ -1,24 +1,49 @@
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings
from dotenv import load_dotenv from dotenv import load_dotenv
import os import os
import sys
load_dotenv() ENV_FILE = ".env"
if not os.path.exists(ENV_FILE):
default_env_content = """
APP_NAME=AI Network Configurator
DEBUG=True
API_PREFIX=/api
SILICONFLOW_API_KEY=your-api-key-here
SILICONFLOW_API_URL=https://api.siliconflow.cn/v1
SWITCH_USERNAME=admin
SWITCH_PASSWORD=admin
SWITCH_TIMEOUT=10
ENSP_DEFAULT_IP=172.17.99.201
ENSP_DEFAULT_PORT=2000
"""
with open(ENV_FILE, "w", encoding="utf-8") as f:
f.write(default_env_content)
print(f"已生成默认配置文件 {ENV_FILE} ，请修改后重新运行程序。")
sys.exit(1)
load_dotenv(ENV_FILE)
class Settings(BaseSettings): class Settings(BaseSettings):
APP_NAME: str = "AI Network Configurator" APP_NAME: str
DEBUG: bool = True DEBUG: bool
API_PREFIX: str = "/api" API_PREFIX: str
SILICONFLOW_API_KEY: str = os.getenv("SILICONFLOW_API_KEY", "sk-oftmyihyxitocscgjdicafzgezprwqpzzgkzsvoxrakkagmd") SILICONFLOW_API_KEY: str
SILICONFLOW_API_URL: str = os.getenv("SILICONFLOW_API_URL", "https://api.siliconflow.cn/v1") SILICONFLOW_API_URL: str
SWITCH_USERNAME: str = os.getenv("SWITCH_USERNAME", "admin") SWITCH_USERNAME: str
SWITCH_PASSWORD: str = os.getenv("SWITCH_PASSWORD", "admin") SWITCH_PASSWORD: str
SWITCH_TIMEOUT: int = os.getenv("SWITCH_TIMEOUT", 10) SWITCH_TIMEOUT: int
ENSP_DEFAULT_IP: str = "172.17.99.201" ENSP_DEFAULT_IP: str
ENSP_DEFAULT_PORT: int = 2000 ENSP_DEFAULT_PORT: int
class Config: class Config:
env_file = ".env" env_file = ".env"