ProxyCat/modules/modules.py

import asyncio, logging, random, httpx, re, os
from configparser import ConfigParser
from packaging import version
from colorama import Fore

MESSAGES = {
    'cn': {
        'getting_new_proxy': '正在获取新的代理IP',
        'new_proxy_is': '新的代理IP为: {}',
        'proxy_check_start': '开始检测代理地址...',
        'proxy_check_disabled': '代理检测已禁用',
        'valid_proxies': '有效代理地址: {}',
        'no_valid_proxies': '没有有效的代理地址',
        'proxy_check_failed': '{}代理 {} 检测失败: {}',
        'proxy_switch': '切换到新的代理: {}',
        'proxy_switch_detail': '已切换代理: {} -> {}',
        'proxy_consecutive_fails': '代理 {} 连续失败 {} 次，正在切换新代理',
        'proxy_invalid': '代理 {} 已失效，立即切换新代理',
        'connection_timeout': '连接超时',
        'proxy_invalid_switching': '代理地址失效，切换代理地址',
        'data_transfer_timeout': '数据传输超时，正在重试...',
        'connection_reset': '连接被重置',
        'transfer_cancelled': '传输被取消',
        'data_transfer_error': '数据传输错误: {}',
        'unsupported_protocol': '不支持的协议请求: {}',
        'client_error': '客户端处理出错: {}',
        'response_write_error': '响应写入错误: {}',
        'server_closing': '服务器正在关闭...',
        'program_interrupted': '程序被用户中断',
        'multiple_proxy_fail': '多次尝试获取有效代理失败，退出程序',
        'current_proxy': '当前代理',
        'next_switch': '下次切换',
        'seconds': '秒',
        'no_proxies_available': '没有可用的代理',
        'proxy_file_not_found': '代理文件不存在: {}',
        'auth_not_set': '未设置 (无需认证)',
        'public_account': '公众号',
        'blog': '博客',
        'proxy_mode': '代理轮换模式',
        'cycle': '循环',
        'load_balance': '负载均衡',
        'single_round': '单轮',
        'proxy_interval': '代理更换时间',
        'default_auth': '默认账号密码',
        'local_http': '本地监听地址 (HTTP)',
        'local_socks5': '本地监听地址 (SOCKS5)',
        'star_project': '开源项目求 Star',
        'client_request_error': '客户端请求错误: {}',
        'client_handle_error': '客户端处理错误: {}',
        'proxy_invalid_switch': '代理无效，切换代理',
        'request_fail_retry': '请求失败，重试剩余次数: {}',
        'request_error': '请求错误: {}',
        'user_interrupt': '用户中断程序',
        'new_version_found': '发现新版本！',
        'visit_quark': '请访问 https://pan.quark.cn/s/39b4b5674570 获取最新版本。',
        'visit_github': '请访问 https://github.com/honmashironeko/ProxyCat 获取最新版本。',
        'visit_baidu': '请访问 https://pan.baidu.com/s/1C9LVC9aiaQeYFSj_2mWH1w?pwd=13r5 获取最新版本。',
        'latest_version': '当前版本已是最新',
        'version_info_not_found': '无法在响应中找到版本信息',
        'update_check_error': '检查更新时发生错误: {}',
        'unauthorized_ip': '未授权的IP尝试访问: {}',
        'client_cancelled': '客户端连接已取消',
        'socks5_connection_error': 'SOCKS5连接错误: {}',
        'connect_timeout': '连接超时',
        'connection_reset': '连接被重置',
        'transfer_cancelled': '传输已取消',
        'client_request_error': '客户端请求处理错误: {}',
        'unsupported_protocol': '不支持的协议: {}',
        'proxy_invalid_switch': '代理无效，正在切换',
        'request_retry': '请求失败，重试中 (剩余{}次)',
        'request_error': '请求过程中出错: {}',
        'response_write_error': '写入响应时出错: {}',
        'consecutive_failures': '检测到连续代理失败: {}',
        'invalid_proxy': '当前代理无效: {}',
        'proxy_switched': '已从代理 {} 切换到 {}'
    },
    'en': {
        'getting_new_proxy': 'Getting new proxy IP',
        'new_proxy_is': 'New proxy IP is: {}',
        'proxy_check_start': 'Starting proxy check...',
        'proxy_check_disabled': 'Proxy check is disabled',
        'valid_proxies': 'Valid proxies: {}',
        'no_valid_proxies': 'No valid proxies found',
        'proxy_check_failed': '{} proxy {} check failed: {}',
        'proxy_switch': 'Switching to new proxy: {}',
        'proxy_switch_detail': 'Switched proxy: {} -> {}',
        'proxy_consecutive_fails': 'Proxy {} failed {} times consecutively, switching to new proxy',
        'proxy_invalid': 'Proxy {} is invalid, switching immediately',
        'connection_timeout': 'Connection timeout',
        'proxy_invalid_switching': 'Proxy invalid, switching to new proxy',
        'data_transfer_timeout': 'Data transfer timeout, retrying...',
        'connection_reset': 'Connection reset',
        'transfer_cancelled': 'Transfer cancelled',
        'data_transfer_error': 'Data transfer error: {}',
        'unsupported_protocol': 'Unsupported protocol request: {}',
        'client_error': 'Client handling error: {}',
        'response_write_error': 'Response write error: {}',
        'server_closing': 'Server is closing...',
        'program_interrupted': 'Program interrupted by user',
        'multiple_proxy_fail': 'Multiple attempts to get valid proxy failed, exiting',
        'current_proxy': 'Current Proxy',
        'next_switch': 'Next Switch',
        'seconds': 's',
        'no_proxies_available': 'No proxies available',
        'proxy_file_not_found': 'Proxy file not found: {}',
        'auth_not_set': 'Not set (No authentication required)',
        'public_account': 'WeChat Public Number',
        'blog': 'Blog',
        'proxy_mode': 'Proxy Rotation Mode',
        'cycle': 'Cycle',
        'load_balance': 'Load Balance',
        'single_round': 'Single Round',
        'proxy_interval': 'Proxy Change Interval',
        'default_auth': 'Default Username and Password',
        'local_http': 'Local Listening Address (HTTP)',
        'local_socks5': 'Local Listening Address (SOCKS5)',
        'star_project': 'Star the Project',
        'client_request_error': 'Client request error: {}',
        'client_handle_error': 'Client handling error: {}',
        'proxy_invalid_switch': 'Proxy invalid, switching proxy',
        'request_fail_retry': 'Request failed, retrying remaining times: {}',
        'request_error': 'Request error: {}',
        'user_interrupt': 'User interrupted the program',
        'new_version_found': 'New version found!',
        'visit_quark': 'Please visit https://pan.quark.cn/s/39b4b5674570 to get the latest version.',
        'visit_github': 'Please visit https://github.com/honmashironeko/ProxyCat to get the latest version.',
        'visit_baidu': 'Please visit https://pan.baidu.com/s/1C9LVC9aiaQeYFSj_2mWH1w?pwd=13r5 to get the latest version.',
        'latest_version': 'You are using the latest version',
        'version_info_not_found': 'Version information not found in the response',
        'update_check_error': 'Error occurred while checking for updates: {}',
        'unauthorized_ip': 'Unauthorized IP attempt: {}',
        'client_cancelled': 'Client connection cancelled',
        'socks5_connection_error': 'SOCKS5 connection error: {}',
        'connect_timeout': 'Connection timeout',
        'connection_reset': 'Connection reset',
        'transfer_cancelled': 'Transfer cancelled',
        'data_transfer_error': 'Data transfer error: {}',
        'client_request_error': 'Client request handling error: {}',
        'unsupported_protocol': 'Unsupported protocol: {}',
        'proxy_invalid_switch': 'Proxy invalid, switching',
        'request_retry': 'Request failed, retrying ({} left)',
        'request_error': 'Error during request: {}',
        'response_write_error': 'Error writing response: {}',
        'consecutive_failures': 'Consecutive proxy failures detected for {}',
        'invalid_proxy': 'Current proxy is invalid: {}',
        'proxy_switched': 'Switched from proxy {} to {}'
    }
}

def get_message(key, lang='cn', *args):
    try:
        return MESSAGES[lang][key].format(*args) if args else MESSAGES[lang][key]
    except KeyError:
        return MESSAGES['cn'][key] if key in MESSAGES['cn'] else key

def print_banner(config):
    language = config.get('language', 'cn').lower()
    has_auth = config.get('username') and config.get('password')
    auth_info = f"{config.get('username')}:{config.get('password')}" if has_auth else get_message('auth_not_set', language)
    
    http_addr = f"http://{auth_info}@127.0.0.1:{config.get('port')}" if has_auth else f"http://127.0.0.1:{config.get('port')}"
    socks5_addr = f"socks5://{auth_info}@127.0.0.1:{config.get('port')}" if has_auth else f"socks5://127.0.0.1:{config.get('port')}"
    
    banner_info = [
        (get_message('public_account', language), '樱花庄的本间白猫'),
        (get_message('blog', language), 'https://y.shironekosan.cn'),
        (get_message('proxy_mode', language), get_message('cycle', language) if config.get('mode') == 'cycle' else get_message('load_balance', language) if config.get('mode') == 'load_balance' else get_message('single_round', language)),
        (get_message('proxy_interval', language), f"{config.get('interval')}{get_message('seconds', language)}"),
        (get_message('default_auth', language), auth_info),
        (get_message('local_http', language), http_addr),
        (get_message('local_socks5', language), socks5_addr),
        (get_message('star_project', language), 'https://github.com/honmashironeko/ProxyCat'),
    ]
    print(f"{Fore.MAGENTA}{'=' * 55}")
    for key, value in banner_info:
        print(f"{Fore.YELLOW}{key}: {Fore.GREEN}{value}")
    print(f"{Fore.MAGENTA}{'=' * 55}\n")

logo1 = r"""
      |\      _,,,---,,_  by 本间白猫
ZZZzz /,`.-'`'    -.  ;-;;,_
     |,4-  ) )-,_. ,\ (  `'-'
    '---''(_/--'  `-'\_)  ProxyCat 
"""
logo2 = r"""
             *     ,MMM8&&&.            *
                  MMMM88&&&&&    .
                 MMMM88&&&&&&&
     *           MMM88&&&&&&&&
                 MMM88&&&&&&&&
                 'MMM88&&&&&&'
                   'MMM8&&&'      *    
            /\/|_    __/\\
           /    -\  /-   ~\  .              '
           \    =_YT_ =   /
           /==*(`    `\ ~ \         ProxyCat
          /     \     /    `\      by 本间白猫
          |     |     ) ~   (
         /       \   /     ~ \\
         \       /   \~     ~/
  _/\_/\_/\__  _/_/\_/\__~__/_/\_/\_/\_/\_/\_
  |  |  |  | ) ) |  |  | ((  |  |  |  |  |  |
  |  |  |  |( (  |  |  |  \\ |  |  |  |  |  |
  |  |  |  | )_) |  |  |  |))|  |  |  |  |  | 
  |  |  |  |  |  |  |  |  (/ |  |  |  |  |  |
  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
"""
logo3 = r"""
        /\_/\         _
       /``   \       / )
       |n n   |__   ( (
      =(Y =.'`   `\  \ \\
       {`"`        \  ) ) 
       {       /    |/ /
        \\   ,(     / /
ProxyCat) ) /-'\  ,_.' by 本间白猫
       (,(,/ ((,,/
"""
logo4 = r"""
                   .-o=o-.
               ,  /=o=o=o=\ .--.
              _|\|=o=O=o=O=|    \\
          __.'  a`\=o=o=o=(`\   /
          '.   a 4/`|.-""'`\ \ ;'`)   .---.
            \   .'  /   .--'  |_.'   / .-._)
 by 本间白猫 `)  _.'   /     /`-.__.' /
  ProxyCat  `'-.____;     /'-.___.-'
                       `\"""`
"""

logos_list = [logo1, logo2, logo3, logo4]
def logos():
    selected_logo = random.choice(logos_list)
    print(selected_logo)

DEFAULT_CONFIG = {
    'port': '1080',
    'mode': 'cycle',
    'interval': '300',
    'username': 'neko',
    'password': '123456',
    'use_getip': 'False',
    'proxy_file': 'ip.txt',
    'check_proxies': 'True',
    'whitelist_file': '',
    'blacklist_file': '',
    'ip_auth_priority': 'whitelist',
    'language': 'cn'
}

def load_config(config_file='config/config.ini'):
    config = ConfigParser()
    config.read(config_file, encoding='utf-8')
    
    settings = {}
    if config.has_section('SETTINGS'):
        settings.update(dict(config.items('SETTINGS')))

        for key in ['proxy_file', 'whitelist_file', 'blacklist_file']:
            if key in settings and settings[key]:
                config_dir = os.path.dirname(config_file)
                settings[key] = os.path.join(config_dir, settings[key])
            
    return {**DEFAULT_CONFIG, **settings}

def load_ip_list(file_path):
    if not file_path or not os.path.exists(file_path):
        return set()
    
    with open(file_path, 'r') as f:
        return {line.strip() for line in f if line.strip()}

async def check_proxy(proxy):
    proxy_type = proxy.split('://')[0]
    check_funcs = {
        'http': check_http_proxy,
        'https': check_https_proxy,
        'socks5': check_socks_proxy
    }
    
    if proxy_type not in check_funcs:
        return False
    
    try:
        return await check_funcs[proxy_type](proxy)
    except Exception as e:
        logging.error(f"{proxy_type.upper()}代理 {proxy} 检测失败: {e}")
        return False

async def check_http_proxy(proxy):
    async with httpx.AsyncClient(proxies={'http://': proxy}, timeout=10) as client:
        response = await client.get('http://www.baidu.com')
        return response.status_code == 200

async def check_https_proxy(proxy):
    async with httpx.AsyncClient(proxies={'https://': proxy}, timeout=10) as client:
        response = await client.get('https://www.baidu.com')
        return response.status_code == 200

async def check_socks_proxy(proxy):
    proxy_type, proxy_addr = proxy.split('://')
    proxy_host, proxy_port = proxy_addr.split(':')
    proxy_port = int(proxy_port)
    try:
        reader, writer = await asyncio.wait_for(asyncio.open_connection(proxy_host, proxy_port), timeout=5)
        writer.write(b'\x05\x01\x00')
        await writer.drain()
        response = await asyncio.wait_for(reader.readexactly(2), timeout=5)
        writer.close()
        await writer.wait_closed()
        return response == b'\x05\x00'
    except Exception:
        return False

async def check_proxies(proxies):
    valid_proxies = []
    for proxy in proxies:
        if await check_proxy(proxy):
            valid_proxies.append(proxy)
    return valid_proxies

async def check_for_updates(language='cn'):
    try:
        async with httpx.AsyncClient() as client:
            response = await asyncio.wait_for(client.get("https://y.shironekosan.cn/1.html"), timeout=10)
            response.raise_for_status()
            content = response.text
            match = re.search(r'<p>(ProxyCat-V\d+\.\d+\.\d+)</p>', content)
            if match:
                latest_version = match.group(1)
                CURRENT_VERSION = "ProxyCat-V1.9.2"
                if version.parse(latest_version.split('-V')[1]) > version.parse(CURRENT_VERSION.split('-V')[1]):
                    print(f"{Fore.YELLOW}{get_message('new_version_found', language)} 当前版本: {CURRENT_VERSION}, 最新版本: {latest_version}")
                    print(f"{Fore.YELLOW}{get_message('visit_quark', language)}")
                    print(f"{Fore.YELLOW}{get_message('visit_github', language)}")
                    print(f"{Fore.YELLOW}{get_message('visit_baidu', language)}")
                else:
                    print(f"{Fore.GREEN}{get_message('latest_version', language)} ({CURRENT_VERSION})")
            else:
                print(f"{Fore.RED}{get_message('version_info_not_found', language)}")
    except Exception as e:
        print(f"{Fore.RED}{get_message('update_check_error', language, e)}")