mirror of
https://github.com/honmashironeko/ProxyCat.git
synced 2026-10-27 19:10:55 +00:00
407 lines
15 KiB
Python
407 lines
15 KiB
Python
"""
|
||
模块名称：modules.proxypool_service
|
||
功能描述：代理池在 ProxyCat 进程内的宿主：用专用线程承载池的 asyncio 事件循环，对外只暴露同步门面方法，
|
||
供 Flask 同步路由与 CLI 调用，另提供进程内事件循环使用的异步取用入口（取一个 / 取一批出口代理）。
|
||
职责边界：负责：事件循环线程的创建与销毁、Application 初始化与关闭、跨线程调用的桥接与超时/异常转换、配置推送；
|
||
不负责：HTTP 路由与请求解析（见 proxypool_api）、配置文件读写（见 ProxyCat 配置接口）、池业务逻辑（见 core.services）。
|
||
关键依赖：core.application（app / api.dependencies）、core.config.Config、core.logger、
|
||
modules.pool_config_ini、modules.logging_setup、modules.loop_noise_filter、modules.modules。
|
||
已知限制：
|
||
1. from modules.proxypool import POOL_DIR 仅靠导入副作用注入 sys.path，必须早于 core.* 导入且不可删除。
|
||
2. 在池事件循环线程内调用 call() 会死锁；call() 有线程守卫，池线程内调用直接抛 RuntimeError。
|
||
3. 调用超时必须大于池内部最长超时（aiohttp 30 秒）；DEFAULT_CALL_TIMEOUT 取 35 秒，超时后调用方放弃而池仍在跑。
|
||
4. stop() 默认 35 秒预算等于池内各段最坏耗时之和（插件停止 10 秒 + 配置落库 5 秒 + 写队列排空 15 秒 + 余量），改动任一段都要重算。
|
||
5. 上一次 stop() 超时未让线程退出时 start() 拒绝启动，避免两个 Application 并存并同写 proxies.db。
|
||
6. stop 超时后 _stopping 常置：is_running 为假而线程可能仍在写库，退出与恢复路径须按 has_live_thread 判断。
|
||
7. 语言提供者须在构造期注册（池未启动时面板仍按当前语言渲染池设置文案），不可因初始化阶段会重复注册而删除。
|
||
"""
|
||
|
||
import asyncio
|
||
import concurrent.futures
|
||
import logging
|
||
import os
|
||
import random
|
||
import threading
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Coroutine, Optional
|
||
|
||
from modules.proxypool import POOL_DIR # noqa: F401
|
||
from core.application.app import Application
|
||
from core.application.api.dependencies import (
|
||
set_container, set_dedup_cache, set_language_provider,
|
||
)
|
||
from core.config import Config
|
||
from core.logger import attach_pool_log_handlers
|
||
|
||
from modules import pool_config_ini
|
||
from modules.logging_setup import LogCategory, classify_record
|
||
from modules.loop_noise_filter import install_loop_noise_filter
|
||
from modules.modules import get_message
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
DEFAULT_CALL_TIMEOUT = 35.0
|
||
|
||
CONFIG_WATCH_INTERVAL = 3.0
|
||
|
||
|
||
class _PoolLogFilter(logging.Filter):
|
||
def filter(self, record: logging.LogRecord) -> bool:
|
||
return classify_record(record) == LogCategory.POOL
|
||
|
||
|
||
class PoolUnavailableError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class PoolTimeoutError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class ProxyPoolService:
|
||
def __init__(self, config: Config, config_ini_path: str | Path,
|
||
language_provider: Optional[Callable[[], str]] = None):
|
||
self._config = config
|
||
self._config_ini_path = Path(config_ini_path)
|
||
self._language_provider = language_provider
|
||
set_language_provider(language_provider)
|
||
|
||
self._thread: Optional[threading.Thread] = None
|
||
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
||
self._application: Optional[Application] = None
|
||
|
||
self._ready = threading.Event()
|
||
self._stopping = threading.Event()
|
||
self._startup_ok = False
|
||
self._lifecycle_lock = threading.RLock()
|
||
|
||
self._last_error: Optional[str] = None
|
||
self._last_mtime: float = 0.0
|
||
self._watch_task: Optional[asyncio.Task] = None
|
||
|
||
@property
|
||
def is_running(self) -> bool:
|
||
thread = self._thread
|
||
return bool(
|
||
thread is not None
|
||
and thread.is_alive()
|
||
and self._startup_ok
|
||
and self._ready.is_set()
|
||
and not self._stopping.is_set()
|
||
)
|
||
|
||
@property
|
||
def has_live_thread(self) -> bool:
|
||
thread = self._thread
|
||
return thread is not None and thread.is_alive()
|
||
|
||
@property
|
||
def last_error(self) -> Optional[str]:
|
||
return self._last_error
|
||
|
||
@property
|
||
def language(self) -> str:
|
||
if self._language_provider is None:
|
||
return 'cn'
|
||
return self._language_provider() or 'cn'
|
||
|
||
def start(self, timeout: float = 30.0) -> bool:
|
||
with self._lifecycle_lock:
|
||
if self.is_running:
|
||
return True
|
||
|
||
previous = self._thread
|
||
if previous is not None and previous.is_alive():
|
||
self._last_error = get_message('pool_previous_stop_incomplete', self.language)
|
||
logger.error(self._last_error)
|
||
return False
|
||
|
||
self._ready.clear()
|
||
self._stopping.clear()
|
||
self._startup_ok = False
|
||
self._last_error = None
|
||
self._remember_config_mtime()
|
||
|
||
self._thread = threading.Thread(
|
||
target=self._run_loop, name="proxypool-loop", daemon=True
|
||
)
|
||
self._thread.start()
|
||
|
||
if not self._ready.wait(timeout):
|
||
self._last_error = get_message('pool_start_timeout', self.language, f"{timeout:.0f}")
|
||
logger.error(self._last_error)
|
||
return False
|
||
|
||
return self.is_running
|
||
|
||
def stop(self, timeout: float = 35.0) -> bool:
|
||
with self._lifecycle_lock:
|
||
thread = self._thread
|
||
if thread is None or not thread.is_alive():
|
||
self._thread = None
|
||
return True
|
||
|
||
logger.info(get_message('pool_stopping', self.language))
|
||
self._stopping.set()
|
||
|
||
loop = self._loop
|
||
if loop is not None:
|
||
try:
|
||
loop.call_soon_threadsafe(loop.stop)
|
||
except RuntimeError:
|
||
pass
|
||
|
||
thread.join(timeout)
|
||
|
||
if thread.is_alive():
|
||
self._last_error = get_message('pool_stop_timeout', self.language, f"{timeout:.0f}")
|
||
logger.error(self._last_error)
|
||
return False
|
||
|
||
with self._lifecycle_lock:
|
||
self._thread = None
|
||
self._loop = None
|
||
self._stopping.clear()
|
||
|
||
logger.info(get_message('pool_stop_success', self.language))
|
||
return True
|
||
|
||
def restart(self, timeout: float = 30.0) -> bool:
|
||
with self._lifecycle_lock:
|
||
self.stop(timeout=timeout)
|
||
return self.start(timeout=timeout)
|
||
|
||
def _run_loop(self) -> None:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
self._loop = loop
|
||
install_loop_noise_filter(loop)
|
||
|
||
initialized = False
|
||
try:
|
||
loop.run_until_complete(self._initialize_application())
|
||
initialized = True
|
||
self._startup_ok = True
|
||
logger.info(get_message('pool_start_success', self.language))
|
||
except Exception as e:
|
||
self._startup_ok = False
|
||
self._last_error = f"{type(e).__name__}: {e}"
|
||
logger.error(get_message('pool_start_failed', self.language, e), exc_info=True)
|
||
finally:
|
||
self._ready.set()
|
||
|
||
if initialized:
|
||
try:
|
||
loop.run_forever()
|
||
except Exception as e:
|
||
self._last_error = f"{type(e).__name__}: {e}"
|
||
logger.error(get_message('pool_loop_crashed', self.language, e), exc_info=True)
|
||
|
||
self._shutdown_application(loop)
|
||
|
||
async def _initialize_application(self) -> None:
|
||
application = Application(self._config)
|
||
try:
|
||
await application.initialize()
|
||
except BaseException:
|
||
try:
|
||
await application.shutdown()
|
||
except Exception as e:
|
||
logger.warning("初始化失败后的收尾也失败了（不影响报错）: %s", e)
|
||
raise
|
||
|
||
self._application = application
|
||
set_container(application.get_container())
|
||
set_dedup_cache(application.dedup_cache)
|
||
set_language_provider(self._language_provider)
|
||
|
||
attach_pool_log_handlers(
|
||
self._config.logging,
|
||
logging.getLogger().level or logging.INFO,
|
||
record_filter=_PoolLogFilter(),
|
||
)
|
||
|
||
self._watch_task = asyncio.create_task(self._watch_config_file())
|
||
|
||
def _shutdown_application(self, loop: asyncio.AbstractEventLoop) -> None:
|
||
application = self._application
|
||
self._application = None
|
||
set_container(None)
|
||
set_dedup_cache(None)
|
||
|
||
if application is None:
|
||
self._close_loop(loop)
|
||
return
|
||
|
||
try:
|
||
loop.run_until_complete(application.shutdown())
|
||
except Exception as e:
|
||
logger.error("关闭代理池时出错: %s", e, exc_info=True)
|
||
|
||
self._close_loop(loop)
|
||
|
||
@staticmethod
|
||
def _close_loop(loop: asyncio.AbstractEventLoop) -> None:
|
||
try:
|
||
pending = [t for t in asyncio.all_tasks(loop) if not t.done()]
|
||
for task in pending:
|
||
task.cancel()
|
||
if pending:
|
||
loop.run_until_complete(
|
||
asyncio.gather(*pending, return_exceptions=True)
|
||
)
|
||
except Exception as e:
|
||
logger.debug("清理残留任务时出错（忽略）: %s", e)
|
||
finally:
|
||
try:
|
||
loop.close()
|
||
except Exception as e:
|
||
logger.debug("关闭事件循环时出错（忽略）: %s", e)
|
||
|
||
def call(self, coro: Coroutine, timeout: float = DEFAULT_CALL_TIMEOUT) -> Any:
|
||
if threading.current_thread() is self._thread:
|
||
coro.close()
|
||
raise RuntimeError("禁止在代理池事件循环线程内同步等待池调用（会死锁）")
|
||
|
||
if not self.is_running or self._loop is None:
|
||
coro.close()
|
||
raise PoolUnavailableError(
|
||
self._last_error or get_message('pool_not_running', self.language))
|
||
|
||
future = asyncio.run_coroutine_threadsafe(coro, self._loop)
|
||
try:
|
||
return future.result(timeout)
|
||
except concurrent.futures.TimeoutError:
|
||
future.cancel()
|
||
raise PoolTimeoutError(
|
||
get_message('pool_call_timeout', self.language, f"{timeout:.0f}")) from None
|
||
except concurrent.futures.CancelledError:
|
||
raise PoolUnavailableError(
|
||
get_message('pool_stopped', self.language)) from None
|
||
|
||
def submit(self, coro: Coroutine) -> concurrent.futures.Future:
|
||
if not self.is_running or self._loop is None:
|
||
coro.close()
|
||
raise PoolUnavailableError(
|
||
self._last_error or get_message('pool_not_running', self.language))
|
||
|
||
return asyncio.run_coroutine_threadsafe(coro, self._loop)
|
||
|
||
async def _run_route(self, coro, timeout: float):
|
||
future = self.submit(coro)
|
||
return await asyncio.wait_for(asyncio.wrap_future(future), timeout=timeout)
|
||
|
||
@staticmethod
|
||
def _text_content(result) -> str:
|
||
from core.application.api.http_types import RawResponse
|
||
|
||
return result.content if isinstance(result, RawResponse) else str(result)
|
||
|
||
async def fetch_random_proxy_url(self, timeout: float = 10.0) -> Optional[str]:
|
||
from core.application.api.routes import proxies as proxy_routes
|
||
|
||
if not self.is_running:
|
||
return None
|
||
|
||
result = await self._run_route(
|
||
proxy_routes.get_random_proxy(status='valid', format='text'), timeout
|
||
)
|
||
return self._text_content(result).strip() or None
|
||
|
||
async def fetch_proxy_entries(self, limit: int, timeout: float = 10.0) -> list:
|
||
from core.application.api.routes import proxies as proxy_routes
|
||
|
||
limit = int(limit)
|
||
if not self.is_running or limit <= 0:
|
||
return []
|
||
|
||
counted = await self._run_route(
|
||
proxy_routes.get_proxies_count(status='valid'), timeout
|
||
)
|
||
total = int((counted or {}).get('count', 0))
|
||
if total <= 0:
|
||
return []
|
||
|
||
pages = max(1, (total + limit - 1) // limit)
|
||
page = random.randint(1, pages)
|
||
result = await self._run_route(
|
||
proxy_routes.get_proxies(
|
||
status='valid', page=page, page_size=limit, format='json'
|
||
),
|
||
timeout,
|
||
)
|
||
return [entry for entry in (result or [])
|
||
if isinstance(entry, dict)][:limit]
|
||
|
||
async def fetch_proxy_urls(self, limit: int, timeout: float = 10.0) -> list:
|
||
entries = await self.fetch_proxy_entries(limit, timeout)
|
||
urls = [(entry.get('proxy_url') or '').strip() for entry in entries]
|
||
return [url for url in urls if url]
|
||
|
||
async def report_proxy_use(self, ip: str, port: int, ok: bool,
|
||
test_url: str = "", timeout: float = 5.0) -> None:
|
||
from core.application.api.routes import validation as validation_routes
|
||
|
||
if not self.is_running:
|
||
return
|
||
|
||
try:
|
||
future = self.submit(
|
||
validation_routes.report_proxy_use(ip, port, ok, test_url)
|
||
)
|
||
await asyncio.wait_for(asyncio.wrap_future(future), timeout=timeout)
|
||
except Exception as e:
|
||
logger.debug("使用期反馈上报失败（已忽略）: %s: %s", type(e).__name__, e)
|
||
|
||
def apply_config(self, config: Config, timeout: float = 15.0) -> None:
|
||
if not self.is_running or self._application is None:
|
||
raise PoolUnavailableError(
|
||
self._last_error or get_message('pool_not_running', self.language))
|
||
|
||
self.call(self._application.apply_config(config), timeout=timeout)
|
||
attach_pool_log_handlers(
|
||
config.logging,
|
||
logging.getLogger().level or logging.INFO,
|
||
record_filter=_PoolLogFilter(),
|
||
)
|
||
|
||
def reload_config_from_file(self, timeout: float = 15.0) -> None:
|
||
config = pool_config_ini.load_pool_config(self._config_ini_path)
|
||
self._config = config
|
||
self.apply_config(config, timeout=timeout)
|
||
|
||
def note_config_saved(self) -> None:
|
||
self._remember_config_mtime()
|
||
|
||
def _remember_config_mtime(self) -> None:
|
||
try:
|
||
self._last_mtime = os.path.getmtime(self._config_ini_path)
|
||
except OSError:
|
||
self._last_mtime = 0.0
|
||
|
||
async def _watch_config_file(self) -> None:
|
||
while True:
|
||
try:
|
||
await asyncio.sleep(CONFIG_WATCH_INTERVAL)
|
||
mtime = os.path.getmtime(self._config_ini_path)
|
||
if mtime <= self._last_mtime:
|
||
continue
|
||
|
||
config = pool_config_ini.reload_pool_config(self._config_ini_path)
|
||
if config is None:
|
||
logger.warning(get_message('pool_config_unreadable', self.language))
|
||
continue
|
||
|
||
logger.info(get_message('config_file_changed', self.language))
|
||
self._config = config
|
||
if self._application is not None:
|
||
await self._application.apply_config(config)
|
||
self._last_mtime = mtime
|
||
except asyncio.CancelledError:
|
||
break
|
||
except Exception as e:
|
||
logger.warning("配置文件兜底轮询出错（不影响运行）: %s", e)
|
||
|
||
|
||
__all__ = ["ProxyPoolService", "PoolUnavailableError", "PoolTimeoutError"]
|