from typing import Optional from fastapi import APIRouter, Body, Query, Form,Request from pydantic import BaseModel from ..rotation_service import rotate as rotate_impl, status as status_impl ,citie_list as cities_impl,get_port router = APIRouter() class RotateRequest(BaseModel): id: Optional[int] = None citys: Optional[str] = None @router.post("/rotate") def rotate( req: Optional[RotateRequest] = Body(...), ): # 优先级:Query > Form > JSON client_id = req.id cities = str(req.citys).split(',') # effective_cityhash = req.cityhash # effective_num = req.num result = rotate_impl(client_id=client_id,cities =cities) # 如果轮换成功,更新线路状态 if result.get("changed", False): from ..line_status_manager import status_manager from datetime import datetime status_manager.update_line_status( client_id, current_ip=result.get("ip"), last_rotate_time=datetime.now().isoformat(), status="active", edge_device=result.get("edge"), geo_location=result.get("geo") ) status_manager.increment_rotate_count(client_id) return result @router.get("/status") def get_status(): return status_impl() @router.get("/cities") def get_cities(): return cities_impl() @router.get("/lines") def get_lines(): """获取所有线路信息""" from ..line_status_manager import status_manager # 使用新的方法获取所有线路状态 lines_data = status_manager.get_all_lines_with_defaults() # 格式化返回数据 lines = [] for line_data in lines_data: lines.append({ "id": line_data["id"], "port": get_port(line_data["id"]), "ip": line_data.get("current_ip"), "status": line_data.get("status", "active"), "last_rotate_time": line_data.get("last_rotate_time"), "edge_device": line_data.get("edge_device"), "geo_location": line_data.get("geo_location"), "rotate_count": line_data.get("rotate_count", 0) }) return {"data": lines}