astrbot_plugin_qq_group_daily_analysis/src/utils/pdf_utils.py

"""
PDF工具模块
负责PDF相关的安装和管理功能
"""

import sys
import asyncio
from astrbot.api import logger


class PDFInstaller:
    """PDF功能安装器"""

    @staticmethod
    async def install_pyppeteer(config_manager):
        """安装pyppeteer依赖"""
        try:
            logger.info("开始安装 pyppeteer...")

            # 使用asyncio安装pyppeteer和兼容的websockets版本
            logger.info("安装 pyppeteer==1.0.2 和兼容的依赖...")
            process = await asyncio.create_subprocess_exec(
                sys.executable, "-m", "pip", "install",
                "pyppeteer==1.0.2", "websockets==10.4",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )

            stdout, stderr = await process.communicate()

            if process.returncode == 0:
                logger.info("pyppeteer 安装成功")
                logger.info(f"安装输出: {stdout.decode()}")

                # 重新加载pyppeteer模块
                success = config_manager.reload_pyppeteer()
                if success:
                    return "✅ pyppeteer 安装成功！PDF 功能现已可用。"
                else:
                    return "⚠️ pyppeteer 安装完成，但重新加载失败。请重启 AstrBot 以使用 PDF 功能。"
            else:
                error_msg = stderr.decode()
                logger.error(f"pyppeteer 安装失败: {error_msg}")
                return f"❌ pyppeteer 安装失败: {error_msg}"

        except Exception as e:
            logger.error(f"安装 pyppeteer 时出错: {e}")
            return f"❌ 安装过程中出错: {str(e)}"

    @staticmethod
    async def install_system_deps():
        """通过 pyppeteer 自动安装 Chromium"""
        try:
            logger.info("正在通过 pyppeteer 自动安装 Chromium...")
            
            # 直接通过 pyppeteer 下载 Chromium
            success = await PDFInstaller._download_chromium_via_pyppeteer()
            
            if success:
                return """✅ Chromium 自动安装成功！

系统依赖已自动配置完成。
现在可以使用 PDF 功能了。"""
            else:
                return """⚠️ 通过 pyppeteer 自动安装 Chromium 失败

请尝试以下方法：
1. 确保网络连接正常
2. 检查是否有防火墙或代理限制
3. 手动运行：path/to/your/actual/sys/executable/python -c "import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())"
4. 或者手动安装 Chrome/Chromium 浏览器

安装完成后，重启 AstrBot"""

        except Exception as e:
            logger.error(f"通过 pyppeteer 安装 Chromium 时出错: {e}")
            return f"❌ 通过 pyppeteer 安装 Chromium 时出错: {str(e)}"

    @staticmethod
    async def _download_chromium_via_pyppeteer():
        """通过 pyppeteer 自动下载 Chromium"""
        try:
            logger.info("通过 pyppeteer 自动下载 Chromium...")
            
            # 导入 pyppeteer 并尝试下载
            try:
                import pyppeteer
                from pyppeteer import launch
                from pyppeteer.errors import BrowserError
                
                # 尝试直接下载 Chromium 而不启动浏览器
                logger.info("尝试直接下载 Chromium...")
                try:
                    # 使用 pyppeteer 的内部下载方法
                    from pyppeteer.connection import Connection
                    from pyppeteer.browser import Browser
                    from pyppeteer.launcher import Launcher
                    
                    # 创建 Launcher 实例但不启动浏览器
                    launcher = Launcher(
                        headless=True,
                        args=['--no-sandbox', '--disable-setuid-sandbox']
                    )
                    
                    # 只下载 Chromium
                    await launcher._get_chromium_revision()
                    await launcher._download_chromium()
                    
                    logger.info("Chromium 下载完成")
                    return True
                    
                except Exception as download_error:
                    logger.warning(f"直接下载 Chromium 失败，尝试启动浏览器: {download_error}")
                
                # 根据操作系统设置不同的参数
                import platform
                system = platform.system().lower()
                
                if system == "linux":
                    # Linux 环境下需要更多参数来避免权限问题
                    browser_args = [
                        '--no-sandbox',
                        '--disable-setuid-sandbox',
                        '--disable-dev-shm-usage',
                        '--disable-accelerated-2d-canvas',
                        '--no-first-run',
                        '--no-zygote',
                        '--disable-gpu',
                        '--disable-background-timer-throttling',
                        '--disable-backgrounding-occluded-windows',
                        '--disable-renderer-backgrounding',
                        '--disable-features=TranslateUI',
                        '--disable-ipc-flooding-protection'
                    ]
                else:
                    # Windows/macOS 环境下的标准参数
                    browser_args = [
                        '--no-sandbox',
                        '--disable-setuid-sandbox',
                        '--disable-gpu'
                    ]
                
                # 尝试启动浏览器，这会触发自动下载
                logger.info("启动 pyppeteer 浏览器以触发 Chromium 自动下载...")
                browser = await launch(
                    headless=True,
                    args=browser_args,
                    ignoreHTTPSErrors=True,
                    dumpio=True  # 输出浏览器日志用于调试
                )
                
                # 获取 Chromium 路径
                chromium_path = pyppeteer.executablePath()
                logger.info(f"Chromium 自动下载完成，路径: {chromium_path}")
                
                await browser.close()
                return True
                
            except BrowserError as e:
                logger.error(f"浏览器错误: {e}", exc_info=True)
                
                # 备用方法：使用命令行触发下载
                try:
                    logger.info("尝试使用命令行触发 Chromium 自动下载...")
                    
                    # 根据操作系统设置不同的命令
                    import platform
                    system = platform.system().lower()
                    
                    if system == "linux":
                        cmd = [
                            sys.executable, "-c",
                            """
import pyppeteer
import asyncio
import platform

async def download_chrome():
    try:
        browser = await pyppeteer.launch(
            headless=True,
            args=[
                '--no-sandbox',
                '--disable-setuid-sandbox',
                '--disable-dev-shm-usage',
                '--disable-gpu',
                '--single-process'
            ]
        )
        await browser.close()
        print("Chromium 下载成功")
    except Exception as e:
        print(f"下载失败: {e}")
        raise

asyncio.run(download_chrome())
                            """
                        ]
                    else:
                        cmd = [
                            sys.executable, "-c",
                            "import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())"
                        ]
                    
                    process = await asyncio.create_subprocess_exec(
                        *cmd,
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE
                    )
                    
                    stdout, stderr = await process.communicate()
                    
                    if process.returncode == 0:
                        logger.info("成功通过命令行触发 Chromium 自动下载")
                        return True
                    else:
                        logger.error(f"命令行触发自动下载失败: {stderr.decode()}")
                        
                        # 最后的备用方案：手动下载
                        logger.info("尝试手动下载 Chromium...")
                        return await PDFInstaller._manual_download_chromium()
                        
                except Exception as e2:
                    logger.error(f"命令行触发自动下载也失败: {e2}")
                    return await PDFInstaller._manual_download_chromium()
                    
        except Exception as e:
            logger.error(f"通过 pyppeteer 自动下载 Chromium 时出错: {e}", exc_info=True)
            return False

    @staticmethod
    async def _manual_download_chromium():
        """手动下载 Chromium 的备用方案"""
        try:
            logger.info("尝试手动下载 Chromium...")
            
            # 尝试使用 wget 或 curl 下载 Chromium
            import platform
            system = platform.system().lower()
            
            if system == "linux":
                # 检查是否有 wget 或 curl
                import shutil
                download_cmd = None
                
                if shutil.which("wget"):
                    download_cmd = [
                        "wget", "-q", "-O", "/tmp/chromium.tar.gz",
                        "https://commondatastorage.googleapis.com/chromium-browser-snapshots/Linux_x64/1000003/chrome-linux.zip"
                    ]
                elif shutil.which("curl"):
                    download_cmd = [
                        "curl", "-s", "-o", "/tmp/chromium.tar.gz",
                        "https://commondatastorage.googleapis.com/chromium-browser-snapshots/Linux_x64/1000003/chrome-linux.zip"
                    ]
                
                if download_cmd:
                    logger.info("使用系统下载工具下载 Chromium...")
                    process = await asyncio.create_subprocess_exec(
                        *download_cmd,
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE
                    )
                    
                    stdout, stderr = await process.communicate()
                    
                    if process.returncode == 0:
                        logger.info("Chromium 下载成功，请手动解压到合适位置")
                        return True
                    else:
                        logger.error(f"下载失败: {stderr.decode()}")
            
            logger.warning("无法自动下载 Chromium，请手动安装 Chrome/Chromium")
            return False
            
        except Exception as e:
            logger.error(f"手动下载 Chromium 失败: {e}")
            return False

    @staticmethod
    def get_pdf_status(config_manager) -> str:
        """获取PDF功能状态"""
        if config_manager.pyppeteer_available:
            version = config_manager.pyppeteer_version or "未知版本"
            return f"✅ PDF 功能可用 (pyppeteer {version})"
        else:
            return "❌ PDF 功能不可用 - 需要安装 pyppeteer"