90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""代理IP白名单管理模块（兼容层）。
|
|
|
|
核心逻辑已迁移至 proxy_platforms/xiequ.py，本模块仅做向后兼容包装。
|
|
旧代码仍可 from .whitelist import WhitelistManager 正常使用。
|
|
"""
|
|
|
|
from typing import Optional
|
|
|
|
from loguru import logger
|
|
|
|
from .proxy_platforms.xiequ import XiequAdapter
|
|
from .proxy_platforms.base import (
|
|
BaseWhitelistAdapter,
|
|
_get_local_exit_ip,
|
|
get_exit_ip_via_proxy,
|
|
)
|
|
|
|
# 全局锁保留（基类内部使用，此处导出供旧代码引用）
|
|
from .proxy_platforms.base import _whitelist_sync_lock
|
|
|
|
|
|
class WhitelistManager:
|
|
"""携趣代理IP白名单管理器（兼容层，委托给 XiequAdapter）。"""
|
|
|
|
MEMO_PREFIX = XiequAdapter.MEMO_PREFIX
|
|
BASE_URL = XiequAdapter.BASE_URL
|
|
|
|
def __init__(self, uid: str, ukey: str):
|
|
self._adapter = XiequAdapter({"uid": uid, "ukey": ukey})
|
|
self.uid = uid
|
|
self.ukey = ukey
|
|
|
|
@property
|
|
def memo(self) -> str:
|
|
return self._adapter.memo
|
|
|
|
def _build_url(self, **params) -> str:
|
|
"""构建请求URL（兼容旧代码直接调用）。"""
|
|
return self._adapter._build_url(**params)
|
|
|
|
def get_whitelist_json(self) -> list[dict]:
|
|
"""
|
|
获取白名单列表（JSON格式，协固原始格式）。
|
|
|
|
Returns:
|
|
[{"IP": "x.x.x.x", "MEMO": "备注"}, ...]
|
|
"""
|
|
records = self._adapter.get_whitelist()
|
|
# 将统一格式映射回调固原始格式
|
|
return [{"IP": r["ip"], "MEMO": r["memo"]} for r in records]
|
|
|
|
def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]:
|
|
"""添加IP到白名单。"""
|
|
return self._adapter.add_ip(ip, retry=retry)
|
|
|
|
def delete_ip(self, ip: str) -> bool:
|
|
"""删除指定IP。"""
|
|
ok, _ = self._adapter.delete_ip(ip)
|
|
return ok
|
|
|
|
def get_memo_ip(self) -> Optional[str]:
|
|
"""获取当前备注对应的IP。"""
|
|
records = self._adapter.get_whitelist()
|
|
for record in records:
|
|
if record.get("memo") == self.memo:
|
|
return record.get("ip")
|
|
return None
|
|
|
|
def get_memo_records(self) -> list[dict]:
|
|
"""获取当前备注的所有记录（协固原始格式）。"""
|
|
records = self._adapter.get_whitelist()
|
|
return [
|
|
{"IP": r["ip"], "MEMO": r["memo"]}
|
|
for r in records
|
|
if r.get("memo") == self.memo
|
|
]
|
|
|
|
def sync_ip(self, current_ip: str, keep_recent: int = 3) -> tuple[bool, str]:
|
|
"""同步白名单IP。"""
|
|
return self._adapter.sync_ip(current_ip, keep_recent)
|
|
|
|
def test_connection(self) -> tuple[bool, str]:
|
|
"""测试白名单API连接。"""
|
|
return self._adapter.test_connection()
|
|
|
|
|
|
def get_local_exit_ip() -> Optional[str]:
|
|
"""获取本机当前公网出口IP（不走代理）。"""
|
|
return _get_local_exit_ip()
|