# -*- coding: UTF-8 -*- ''' @Project :domainScanDemo @File :register.py @IDE :PyCharm @Author :梦伴 @Date :2026/3/27 22:55 @explain : 域名注册状态检测工具 ''' import json import os import threading from collections import OrderedDict from datetime import datetime, timezone, timedelta # 日期时间处理，用于时区转换 from typing import Optional from loguru import logger # 日志记录 import requests import urllib3 _DIRECT_HTTP = requests.Session() _DIRECT_HTTP.trust_env = False _DIRECT_HTTP.mount("http://", requests.adapters.HTTPAdapter(pool_connections=256, pool_maxsize=512, max_retries=0)) _DIRECT_HTTP.mount("https://", requests.adapters.HTTPAdapter(pool_connections=256, pool_maxsize=512, max_retries=0)) _PROXY_MANAGERS = OrderedDict() _PROXY_MANAGER_LOCK = threading.Lock() def _proxy_manager_cache_limit() -> int: try: return max(1, int(os.getenv("DOMAINCHECK_REGISTER_PROXY_SESSION_CACHE_SIZE", "128") or 128)) except Exception: return 128 def _close_session_quietly(session) -> None: try: session.close() except Exception: pass def _get_http_manager(proxies: Optional[dict] = None): proxy_url = "" if proxies: proxy_url = str(proxies.get("https") or proxies.get("http") or "").strip() if not proxy_url: return _DIRECT_HTTP with _PROXY_MANAGER_LOCK: manager = _PROXY_MANAGERS.get(proxy_url) if manager is not None: _PROXY_MANAGERS.move_to_end(proxy_url) return manager manager = requests.Session() manager.trust_env = False manager.mount("http://", requests.adapters.HTTPAdapter(pool_connections=128, pool_maxsize=256, max_retries=0)) manager.mount("https://", requests.adapters.HTTPAdapter(pool_connections=128, pool_maxsize=256, max_retries=0)) _PROXY_MANAGERS[proxy_url] = manager while len(_PROXY_MANAGERS) > _proxy_manager_cache_limit(): _, stale_manager = _PROXY_MANAGERS.popitem(last=False) _close_session_quietly(stale_manager) return manager def _request_register(url: str, *, proxies: dict = None, timeout=None): manager = _get_http_manager(proxies) request_timeout = timeout if isinstance(timeout, urllib3.Timeout): request_timeout = (timeout.connect_timeout, timeout.read_timeout) return manager.get(url, timeout=request_timeout, proxies=proxies, allow_redirects=True) def _resolve_register_timeout(proxies: Optional[dict], budget_seconds: Optional[float] = None): if proxies: connect_timeout = float(os.getenv("DOMAINCHECK_REGISTER_PROXY_CONNECT_TIMEOUT", "1.0") or 1.0) read_timeout = float(os.getenv("DOMAINCHECK_REGISTER_PROXY_READ_TIMEOUT", "1.5") or 1.5) total_timeout = float(os.getenv("DOMAINCHECK_REGISTER_PROXY_TOTAL_TIMEOUT", "2.2") or 2.2) else: connect_timeout = float(os.getenv("DOMAINCHECK_REGISTER_DIRECT_CONNECT_TIMEOUT", "1.2") or 1.2) read_timeout = float(os.getenv("DOMAINCHECK_REGISTER_DIRECT_READ_TIMEOUT", "1.8") or 1.8) total_timeout = float(os.getenv("DOMAINCHECK_REGISTER_DIRECT_TOTAL_TIMEOUT", "2.6") or 2.6) connect_timeout = max(0.2, connect_timeout) read_timeout = max(0.3, read_timeout) total_timeout = max(max(connect_timeout, read_timeout), float(total_timeout or 0.0)) if budget_seconds not in (None, "", 0, "0"): remaining_budget = max(0.6, float(budget_seconds or 0.0)) total_timeout = min(total_timeout, remaining_budget) # requests 只原生支持 (connect, read)，这里把剩余预算重新切成更短的 # connect/read，避免单次 RDAP 请求把整个步骤预算一次性吃掉。 if total_timeout <= 1.0: connect_timeout = min(connect_timeout, 0.35) read_timeout = min(read_timeout, max(0.3, total_timeout - 0.2)) else: connect_timeout = min(connect_timeout, max(0.35, total_timeout * 0.35)) read_timeout = min(read_timeout, max(0.45, total_timeout - connect_timeout)) connect_timeout = max(0.2, min(connect_timeout, total_timeout)) read_timeout = max(0.3, min(read_timeout, total_timeout)) return urllib3.Timeout(connect=connect_timeout, read=read_timeout, total=total_timeout) def check_register(domain: str, postfix: str = 'com', proxies: dict = None, budget_seconds: Optional[float] = None): ''' 检测注册状态 :param domain: 待检测域名（不包含后缀） :param postfix: 域名后缀com、net（默认com） :param proxies: 代理（默认None） :return: 注册状态（2：可注册，3：已注册，-1：检测失败，过期时间（格式：YYYY-MM-DD HH:MM:SS）为空） ''' url = f'https://rdap.verisign.com/{postfix}/v1/domain/{domain}' # 构建RDAP查询URL # url = f'https://www.baidu.com' # 构建RDAP查询URL mode = "proxy" if proxies else "direct" timeout = _resolve_register_timeout(proxies, budget_seconds=budget_seconds) response = None try: response = _request_register(url, proxies=proxies, timeout=timeout) except Exception as e: logger.warning(f"注册状态检测请求异常: {domain}, 模式: {mode}, 错误: {e}") raise if int(response.status_code) == 200: # HTTP 200表示域名已注册 json_data = response.json() if response.content else {} # 解析JSON响应 for item in json_data['events']: # 遍历事件列表 if item['eventAction'] == 'expiration': # 如果是过期时间事件 utc_time = datetime.fromisoformat(item['eventDate'].replace('Z', '+00:00')) # 解析UTC时间 beijing_tz = timezone(timedelta(hours=8)) # 创建北京时区（UTC+8） beijing_time = utc_time.astimezone(beijing_tz) # 将UTC时间转换为北京时间 return 3, beijing_time.strftime("%Y-%m-%d %H:%M:%S") # 返回已注册状态和过期时间 return 3, '' if int(response.status_code) == 404: # HTTP 404表示域名不存在，可注册 return 2, '' # 返回可注册状态，过期时间为空 error_message = f"rdap unexpected status {response.status_code}" logger.warning(f"注册状态检测返回异常状态码: {domain}, 模式: {mode}, 状态码: {response.status_code}") raise RuntimeError(error_message) # def get_proxy(): # # url='http://api.ch12361.com/getProxy.php?group=A&count=1' # response = requests.get(url).json() # username=response['username'] # password=response['password'] # ip=response['ip'] # port=response['port'] # # proxy_url = f"http://{username}:{password}@{ip}:{port}" # logger.info(proxy_url) # return {'http':proxy_url,'https':proxy_url} # # t=datetime.now() # logger.info(check_register('75az.com',proxies=get_proxy())) # 测试查询nnsk.com域名的注册状态 # logger.info(datetime.now()-t)