78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from .config import settings
|
||
from .eip_client import client_singleton as eip
|
||
from .redis_store import store_singleton as kv
|
||
|
||
|
||
def _extract_device_ip_and_id(device: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
|
||
# 由于未知后端返回结构，尽可能容错地提取
|
||
ip = (
|
||
device.get("ip")
|
||
or device.get("public_ip")
|
||
or device.get("eip")
|
||
or device.get("addr")
|
||
or device.get("address")
|
||
)
|
||
edge_id = device.get("id") or device.get("edge") or device.get("mac") or device.get("device_id")
|
||
# 若 edge 是数组或对象，尝试取可打印的第一项
|
||
if isinstance(edge_id, list) and edge_id:
|
||
edge_id = edge_id[0]
|
||
if isinstance(edge_id, dict):
|
||
edge_id = edge_id.get("id") or edge_id.get("mac") or edge_id.get("edge")
|
||
return (str(ip) if ip else None, str(edge_id) if edge_id else None)
|
||
|
||
|
||
def select_unused_ip(devices: List[Dict[str, Any]]) -> Tuple[Optional[str], Optional[str]]:
|
||
for d in devices:
|
||
ip, edge_id = _extract_device_ip_and_id(d)
|
||
if not ip:
|
||
# 没有可识别的 IP，跳过
|
||
continue
|
||
if not kv.is_ip_used_today(ip):
|
||
return ip, edge_id
|
||
return None, None
|
||
|
||
|
||
def apply_gateway_route(edge_id: Optional[str], ip: str) -> Dict[str, Any]:
|
||
# 将选中的 edge 配置到网关规则中
|
||
rule = {
|
||
"table": 1,
|
||
"enable": True,
|
||
"edge": [edge_id] if edge_id else [],
|
||
"network": [],
|
||
"cityhash": settings.eip_default_cityhash or "",
|
||
}
|
||
config = {"id": 1, "rules": [rule]}
|
||
return eip.gateway_config_set(settings.eip_gateway_mac, config)
|
||
|
||
|
||
def rotate(cityhash: Optional[str] = None, num: Optional[int] = None) -> Dict[str, Any]:
|
||
geo = cityhash or settings.eip_default_cityhash
|
||
if not geo:
|
||
raise ValueError("cityhash 不能为空，请在请求中或环境变量中提供")
|
||
n = int(num or settings.eip_default_num or 10)
|
||
devices = eip.list_devices(geo=geo, offset=0, num=n)
|
||
ip, edge_id = select_unused_ip(devices)
|
||
if not ip:
|
||
return {"changed": False, "reason": "没有可用且今天未使用的 IP"}
|
||
|
||
_ = apply_gateway_route(edge_id=edge_id, ip=ip)
|
||
kv.add_used_ip_today(ip)
|
||
kv.set_current(ip=ip, edge_id=edge_id)
|
||
status = eip.gateway_status(settings.eip_gateway_mac)
|
||
return {"changed": True, "ip": ip, "edge": edge_id, "status": status}
|
||
|
||
|
||
def status() -> Dict[str, Any]:
|
||
cur = kv.get_current()
|
||
used_count = kv.get_used_count_today()
|
||
gw = {}
|
||
try:
|
||
gw = eip.gateway_status(settings.eip_gateway_mac)
|
||
except Exception:
|
||
pass
|
||
return {"current": cur, "used_today": used_count, "gateway": gw}
|
||
|
||
|