dekun
32c8f4b156
Web-based TCP/UDP port checker with firewall/GFW diagnosis, PM2 deployment config, and Ubuntu one-click install script. Co-authored-by: Cursor <cursoragent@cursor.com>
102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
"""CPCHECK - 云服务器端口检测工具 API"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from app.config import HOST, PORT
|
|
from app.detector import CheckResult, run_check
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
STATIC_DIR = BASE_DIR / "static"
|
|
|
|
app = FastAPI(
|
|
title="CPCHECK",
|
|
description="云服务器端口检测工具 - 支持 TCP/UDP 端口检测与问题诊断",
|
|
version="1.0.0",
|
|
)
|
|
|
|
if STATIC_DIR.exists():
|
|
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
|
|
|
|
|
class CheckRequest(BaseModel):
|
|
host: str = Field(..., min_length=1, max_length=255, description="域名或 IP 地址")
|
|
port: int = Field(..., ge=1, le=65535, description="端口号")
|
|
protocol: str = Field(default="tcp", description="协议类型: tcp 或 udp")
|
|
|
|
@field_validator("protocol")
|
|
@classmethod
|
|
def validate_protocol(cls, v: str) -> str:
|
|
v = v.lower().strip()
|
|
if v not in ("tcp", "udp"):
|
|
raise ValueError("协议必须是 tcp 或 udp")
|
|
return v
|
|
|
|
@field_validator("host")
|
|
@classmethod
|
|
def validate_host(cls, v: str) -> str:
|
|
v = v.strip()
|
|
if not v:
|
|
raise ValueError("主机地址不能为空")
|
|
return v
|
|
|
|
|
|
class CheckResponse(BaseModel):
|
|
success: bool
|
|
data: dict | None = None
|
|
error: str | None = None
|
|
|
|
|
|
def result_to_dict(result: CheckResult) -> dict:
|
|
return {
|
|
"host": result.host,
|
|
"port": result.port,
|
|
"protocol": result.protocol,
|
|
"resolved_ip": result.resolved_ip,
|
|
"port_status": result.port_status.value,
|
|
"diagnosis": result.diagnosis.value,
|
|
"diagnosis_message": result.diagnosis_message,
|
|
"host_reachable": result.host_reachable,
|
|
"dns_ok": result.dns_ok,
|
|
"latency_ms": result.latency_ms,
|
|
"details": result.details,
|
|
"elapsed_ms": result.elapsed_ms,
|
|
}
|
|
|
|
|
|
@app.get("/")
|
|
async def index():
|
|
index_file = STATIC_DIR / "index.html"
|
|
if index_file.exists():
|
|
return FileResponse(str(index_file))
|
|
return {"message": "CPCHECK API is running. Visit /docs for API documentation."}
|
|
|
|
|
|
@app.get("/api/health")
|
|
async def health():
|
|
return {"status": "ok", "service": "cpcheck"}
|
|
|
|
|
|
@app.post("/api/check", response_model=CheckResponse)
|
|
async def check_port(req: CheckRequest):
|
|
try:
|
|
result = run_check(req.host, req.port, req.protocol)
|
|
return CheckResponse(success=True, data=result_to_dict(result))
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
def start():
|
|
import uvicorn
|
|
|
|
uvicorn.run("app.main:app", host=HOST, port=PORT, reload=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
start()
|