电报下载多源镜像站健康检查与自动故障切换机制

# 示例：health_check.py 核心逻辑框架
import requests
import hashlib
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

class MirrorHealthChecker:
    def __init__(self, mirror_list):
        """
        mirror_list: 列表，每个元素是镜像站的字典，包含 `url`, `name`, `weight`, `expected_hash` 等信息。
        """
        self.mirrors = mirror_list
        self.timeout = 10  # 全局超时时间
        self.chunk_size_for_speed_test = 1024 * 512  # 速度测试时下载的数据块大小（512KB）

    def check_single_mirror(self, mirror):
        """检查单个镜像站，返回健康状态字典。"""
        result = {
            'name': mirror['name'],
            'url': mirror['url'],
            'is_healthy': False,
            'latency': None,
            'status_code': None,
            'download_speed_mbps': None,
            'hash_match': False,
            'last_check': time.time()
        }

        try:
            # 1. 检查基本HTTP可用性与TTFB
            start_time = time.time()
            # 使用HEAD方法获取文件头，避免下载整个文件
            head_resp = requests.head(mirror['url'], timeout=self.timeout, allow_redirects=True)
            result['status_code'] = head_resp.status_code
            result['latency'] = (time.time() - start_time) * 1000  # 毫秒

            if head_resp.status_code != 200:
                return result  # 状态码异常，直接返回不健康

            # 2. 检查文件大小（通过Content-Length）
            content_length = head_resp.headers.get('Content-Length')
            if not content_length:
                # 有些服务器可能不返回Content-Length，需要备用方案
                pass
            # 这里可以加入预期文件大小的校验

            # 3. （可选）进行小范围下载测试速度与哈希
            # 仅下载文件开头部分用于速度测试和哈希校验（假设文件开头部分具有代表性）
            start_speed_test = time.time()
            range_header = {'Range': f'bytes=0-{self.chunk_size_for_speed_test-1}'}
            dl_resp = requests.get(mirror['url'], headers=range_header, timeout=self.timeout, stream=True)
            dl_resp.raise_for_status()

            downloaded = 0
            hasher = hashlib.sha256()
            for chunk in dl_resp.iter_content(chunk_size=8192):
                downloaded += len(chunk)
                hasher.update(chunk)
                if downloaded >= self.chunk_size_for_speed_test:
                    break

            download_duration = time.time() - start_speed_test
            if download_duration > 0:
                result['download_speed_mbps'] = (downloaded / 1024 / 1024) / download_duration  # MB/s

            # 计算部分哈希（注意：这只是文件头部的哈希，用于快速校验，非完整文件哈希）
            partial_hash = hasher.hexdigest()
            # 与预存的“文件头部预期哈希”进行对比（这个值需要预先通过分析官方文件得出）
            if mirror.get('expected_partial_hash'):
                result['hash_match'] = (partial_hash == mirror['expected_partial_hash'])

            # 4. 综合判定健康状态
            # 这是一个简单的逻辑示例，实际应根据业务需求调整阈值和条件
            if (result['status_code'] == 200 and
                result['latency'] < 500 and
                (result['download_speed_mbps'] is None or result['download_speed_mbps'] > 0.5) and
                result['hash_match']):
                result['is_healthy'] = True

        except (requests.exceptions.Timeout,
                requests.exceptions.ConnectionError,
                requests.exceptions.HTTPError) as e:
            # 记录错误日志
            print(f"检查镜像 {mirror['name']} 时出错: {e}")
            result['error'] = str(e)

        return result

    def run_checks(self):
        """并发检查所有镜像站。"""
        health_results = {}
        with ThreadPoolExecutor(max_workers=5) as executor:
            future_to_mirror = {executor.submit(self.check_single_mirror, mirror): mirror for mirror in self.mirrors}
            for future in as_completed(future_to_mirror):
                mirror = future_to_mirror[future]
                try:
                    result = future.result()
                    health_results[mirror['name']] = result
                except Exception as e:
                    print(f"执行检查 {mirror['name']} 时发生未预期错误: {e}")
        return health_results

# 配置示例
if __name__ == '__main__':
    my_mirrors = [
        {
            'name': 'Mirror_CN_East',
            'url': 'https://mirror-cn-east.dinbao-cn.com/telegram/tsetup-latest-x64.exe',
            'expected_partial_hash': 'abc123...'  # 需预先计算
        },
        {
            'name': 'Mirror_US_West',
            'url': 'https://mirror-us-west.dinbao-cn.com/telegram/tsetup-latest-x64.exe',
            'expected_partial_hash': 'def456...'
        },
        # ... 更多镜像
    ]
    checker = MirrorHealthChecker(my_mirrors)
    current_health = checker.run_checks()
    print(current_health)

# 示例：scheduler.py 核心逻辑
class DownloadScheduler:
    def __init__(self, health_checker):
        self.health_checker = health_checker
        # 预设的镜像站元数据，包括权重和地域标签
        self.mirror_metadata = {...}

    def get_best_mirror(self, user_ip=None):
        """根据当前健康状况和用户IP，返回最佳镜像站URL。"""
        health_status = self.health_checker.get_latest_health() # 获取最新健康状态

        healthy_mirrors = []
        for name, status in health_status.items():
            if status['is_healthy']:
                mirror_info = self.mirror_metadata[name].copy()
                mirror_info.update(status) # 合并动态状态
                healthy_mirrors.append(mirror_info)

        if not healthy_mirrors:
            return None  # 或无镜像站可用，返回错误页面

        # 调度算法：此处以“加权最少连接（模拟）+地域亲和”为例
        # 1. 地域过滤：如果有用户IP且能解析出地域，优先选择同地域/同国家的镜像
        if user_ip:
            user_region = self._geoip_lookup(user_ip)
            regional_mirrors = [m for m in healthy_mirrors if m['region'] == user_region]
            if regional_mirrors:
                candidate_pool = regional_mirrors
            else:
                candidate_pool = healthy_mirrors
        else:
            candidate_pool = healthy_mirrors

        # 2. 加权选择：这里简化为按权重随机选择，模拟加权轮询。
        # 更复杂的实现可以维护每个镜像的当前连接数。
        total_weight = sum(m['weight'] for m in candidate_pool)
        r = random.uniform(0, total_weight)
        upto = 0
        for mirror in candidate_pool:
            upto += mirror['weight']
            if upto >= r:
                return mirror['download_url'] # 返回最终给用户的重定向地址

        # 保底逻辑
        return candidate_pool[0]['download_url']