hosts_updater.py

# -*- coding: utf-8 -*-
"""
GitHub Hosts 更新核心逻辑模块
"""

import logging
import re
import requests
from pathlib import Path
from typing import Optional, Tuple

from config import (
    HOSTS_FILE_PATH, REMOTE_HOSTS_URL, GITHUB_START_MARKER, GITHUB_END_MARKER,
    REQUEST_TIMEOUT, REQUEST_RETRY_COUNT
)

logger = logging.getLogger(__name__)


def fetch_remote_hosts() -> Optional[str]:
    """
    从远程URL获取最新hosts内容
    
    Returns:
        str: 远程hosts内容，失败返回None
    """
    try:
        for attempt in range(REQUEST_RETRY_COUNT):
            try:
                logger.info(f"正在获取远程hosts内容 (尝试 {attempt + 1}/{REQUEST_RETRY_COUNT})")
                response = requests.get(REMOTE_HOSTS_URL, timeout=REQUEST_TIMEOUT)
                response.raise_for_status()
                
                content = response.text
                logger.info(f"成功获取远程hosts内容，长度: {len(content)} 字符")
                return content
                
            except requests.exceptions.RequestException as e:
                logger.warning(f"第 {attempt + 1} 次请求失败: {e}")
                if attempt == REQUEST_RETRY_COUNT - 1:
                    raise
                    
    except Exception as e:
        logger.error(f"获取远程hosts内容失败: {e}")
        return None


def read_local_hosts() -> Optional[str]:
    """
    读取本地hosts文件内容
    
    Returns:
        str: 本地hosts内容，失败返回None
    """
    try:
        hosts_path = Path(HOSTS_FILE_PATH)
        if not hosts_path.exists():
            logger.warning(f"Hosts文件不存在: {HOSTS_FILE_PATH}")
            return None
            
        with open(hosts_path, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
            logger.info(f"成功读取本地hosts文件，长度: {len(content)} 字符")
            return content
            
    except Exception as e:
        logger.error(f"读取本地hosts文件失败: {e}")
        return None


def extract_github_section(hosts_content: str) -> str:
    """
    从hosts内容中提取GitHub520部分
    
    Args:
        hosts_content: hosts文件内容
        
    Returns:
        str: GitHub520部分内容
    """
    if not hosts_content:
        return ""
    
    # 使用正则表达式匹配GitHub520部分
    pattern = rf"{re.escape(GITHUB_START_MARKER)}.*?{re.escape(GITHUB_END_MARKER)}"
    match = re.search(pattern, hosts_content, re.DOTALL)
    
    if match:
        github_section = match.group(0)
        logger.info(f"成功提取GitHub520部分，长度: {len(github_section)} 字符")
        return github_section
    else:
        logger.info("未找到GitHub520部分")
        return ""


def normalize_hosts_content(content: str) -> str:
    """
    标准化hosts内容（去除多余空行、统一换行符）
    
    Args:
        content: hosts内容
        
    Returns:
        str: 标准化后的内容
    """
    if not content:
        return ""
    
    # 统一换行符为\n
    content = content.replace('\r\n', '\n').replace('\r', '\n')
    
    # 去除首尾空白
    content = content.strip()
    
    # 去除多余的空行（保留单个空行）
    lines = content.split('\n')
    normalized_lines = []
    prev_empty = False
    
    for line in lines:
        is_empty = not line.strip()
        if not is_empty or not prev_empty:
            normalized_lines.append(line)
        prev_empty = is_empty
    
    return '\n'.join(normalized_lines)


def compare_hosts(local_github_section: str, remote_content: str) -> bool:
    """
    比对本地和远程的GitHub hosts部分是否有差异
    
    Args:
        local_github_section: 本地GitHub520部分
        remote_content: 远程hosts内容
        
    Returns:
        bool: True表示有差异，需要更新
    """
    # 标准化内容
    local_normalized = normalize_hosts_content(local_github_section)
    remote_normalized = normalize_hosts_content(remote_content)
    
    # 比对内容
    has_changes = local_normalized != remote_normalized
    
    if has_changes:
        logger.info("检测到hosts内容变化，需要更新")
        logger.debug(f"本地内容长度: {len(local_normalized)}")
        logger.debug(f"远程内容长度: {len(remote_normalized)}")
    else:
        logger.info("hosts内容无变化，无需更新")
    
    return has_changes


def update_hosts_file(remote_content: str) -> bool:
    """
    更新本地hosts文件
    
    Args:
        remote_content: 远程hosts内容
        
    Returns:
        bool: 更新是否成功
    """
    try:
        hosts_path = Path(HOSTS_FILE_PATH)
        
        # 检查是否有管理员权限
        try:
            import ctypes
            if not ctypes.windll.shell32.IsUserAnAdmin():
                logger.error("需要管理员权限才能修改hosts文件")
                return False
        except:
            logger.warning("无法检查管理员权限，继续尝试更新")
        
        # 读取当前hosts文件
        current_content = read_local_hosts()
        if current_content is None:
            logger.error("无法读取当前hosts文件")
            return False
        
        # 备份原文件
        backup_path = hosts_path.with_suffix('.backup')
        try:
            with open(backup_path, 'w', encoding='utf-8') as f:
                f.write(current_content)
            logger.info(f"已创建hosts文件备份: {backup_path}")
        except Exception as e:
            logger.warning(f"创建备份文件失败: {e}")
        
        # 删除旧的GitHub520部分
        pattern = rf"{re.escape(GITHUB_START_MARKER)}.*?{re.escape(GITHUB_END_MARKER)}"
        updated_content = re.sub(pattern, '', current_content, flags=re.DOTALL)
        
        # 去除多余空行
        updated_content = re.sub(r'\n\s*\n\s*\n', '\n\n', updated_content)
        
        # 添加新的hosts内容
        if updated_content.strip() and not updated_content.endswith('\n'):
            updated_content += '\n'
        
        updated_content += '\n' + remote_content + '\n'
        
        # 写入新内容
        with open(hosts_path, 'w', encoding='utf-8') as f:
            f.write(updated_content)
        
        logger.info("Hosts文件更新成功")
        return True
        
    except PermissionError as e:
        logger.error(f"权限不足，无法修改hosts文件: {e}")
        logger.info("请以管理员身份运行程序")
        return False
    except Exception as e:
        logger.error(f"更新hosts文件失败: {e}")
        
        # 尝试恢复备份
        backup_path = Path(HOSTS_FILE_PATH).with_suffix('.backup')
        if backup_path.exists():
            try:
                with open(backup_path, 'r', encoding='utf-8') as f:
                    backup_content = f.read()
                with open(HOSTS_FILE_PATH, 'w', encoding='utf-8') as f:
                    f.write(backup_content)
                logger.info("已从备份恢复hosts文件")
            except Exception as restore_error:
                logger.error(f"恢复备份失败: {restore_error}")
        
        return False


def check_and_update() -> Tuple[bool, str]:
    """
    检查并更新hosts文件的主函数
    
    Returns:
        Tuple[bool, str]: (是否成功, 结果消息)
    """
    logger.info("开始检查GitHub hosts更新")
    
    try:
        # 获取远程hosts内容
        remote_content = fetch_remote_hosts()
        if remote_content is None:
            return False, "获取远程hosts内容失败"
        
        # 读取本地hosts文件
        local_content = read_local_hosts()
        if local_content is None:
            return False, "读取本地hosts文件失败"
        
        # 提取本地GitHub520部分
        local_github_section = extract_github_section(local_content)
        
        # 比对内容是否有变化
        if not compare_hosts(local_github_section, remote_content):
            return True, "hosts内容无变化，无需更新"
        
        # 更新hosts文件
        if update_hosts_file(remote_content):
            return True, "hosts文件更新成功"
        else:
            return False, "hosts文件更新失败"
            
    except Exception as e:
        error_msg = f"检查更新过程中发生错误: {e}"
        logger.error(error_msg)
        return False, error_msg