zsglpt/screenshot_worker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
独立截图进程 - 使用已保存的Cookies直接截图
"""

import sys
import json
import os
import time
import traceback

def take_screenshot(config):
    """执行截图任务"""
    from playwright.sync_api import sync_playwright

    username = config['username']
    browse_type = config.get('browse_type', '应读')
    screenshot_path = config['screenshot_path']
    cookies_file = config.get('cookies_file', '')

    # 安全修复：验证截图路径在允许的目录内，防止路径遍历攻击
    ALLOWED_SCREENSHOT_DIRS = [
        '/root/zsglpt/screenshots',
        '/root/zsglpt/static/screenshots',
        '/tmp/zsglpt_screenshots'
    ]

    def is_safe_screenshot_path(path):
        """验证截图路径是否安全"""
        abs_path = os.path.abspath(path)
        return any(abs_path.startswith(os.path.abspath(allowed_dir))
                   for allowed_dir in ALLOWED_SCREENSHOT_DIRS)

    if not is_safe_screenshot_path(screenshot_path):
        return {
            'success': False,
            'message': '非法截图路径',
            'screenshot_path': ''
        }

    result = {
        'success': False,
        'message': '',
        'screenshot_path': screenshot_path
    }

    playwright = None
    browser = None
    context = None

    try:
        print(f"[截图进程] 启动浏览器...", flush=True)
        playwright = sync_playwright().start()

        browser = playwright.chromium.launch(
            headless=True,
            args=[
                '--no-sandbox',
                '--disable-setuid-sandbox',
                '--disable-dev-shm-usage',
                '--disable-gpu'
            ]
        )

        # 创建 context
        context = browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
        )

        page = context.new_page()
        page.set_default_timeout(30000)

        # 加载已保存的 Cookies
        if cookies_file and os.path.exists(cookies_file):
            print(f"[截图进程] 加载Cookies: {cookies_file}", flush=True)
            with open(cookies_file, 'r', encoding='utf-8') as f:
                cookies_data = json.load(f)
                cookies = cookies_data.get('cookies', [])
                if cookies:
                    context.add_cookies(cookies)
                    print(f"[截图进程] 已加载 {len(cookies)} 个Cookie", flush=True)
        else:
            print(f"[截图进程] 警告: Cookies文件不存在", flush=True)

        # 根据浏览类型导航到对应页面
        if browse_type == "应读":
            url = "https://zsgl.gat.zj.gov.cn/web/learn/readList"
        elif browse_type == "应学":
            url = "https://zsgl.gat.zj.gov.cn/web/learn/learnList"
        elif browse_type == "应考":
            url = "https://zsgl.gat.zj.gov.cn/web/exam"
        else:
            url = "https://zsgl.gat.zj.gov.cn/web/learn/readList"

        print(f"[截图进程] 导航到: {url}", flush=True)
        page.goto(url, wait_until='networkidle', timeout=30000)
        
        # 等待页面加载
        time.sleep(3)

        # 检查是否被重定向到登录页
        if '/login' in page.url.lower() or '/web/' == page.url.rstrip('/').split('/')[-1]:
            print(f"[截图进程] 登录已过期，需要重新登录", flush=True)
            result['message'] = '登录已过期'
            return result

        # 确保截图目录存在
        os.makedirs(os.path.dirname(screenshot_path), exist_ok=True)

        # 截图
        print(f"[截图进程] 截图保存到: {screenshot_path}", flush=True)
        page.screenshot(path=screenshot_path, full_page=False, type='jpeg', quality=85)

        # 验证截图文件
        if os.path.exists(screenshot_path) and os.path.getsize(screenshot_path) > 1000:
            result['success'] = True
            result['message'] = '截图成功'
            print(f"[截图进程] 截图成功!", flush=True)
        else:
            result['message'] = '截图文件异常'

    except Exception as e:
        result['message'] = f'截图出错: {str(e)}'
        print(f"[截图进程] 错误: {traceback.format_exc()}", flush=True)

    finally:
        try:
            if context:
                context.close()
            if browser:
                browser.close()
            if playwright:
                playwright.stop()
        except:
            pass

    return result


def main():
    if len(sys.argv) < 2:
        print("用法: python screenshot_worker.py <config_json_file>")
        sys.exit(1)

    config_file = sys.argv[1]

    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            config = json.load(f)
    except Exception as e:
        print(json.dumps({'success': False, 'message': f'读取配置失败: {e}'}))
        sys.exit(1)

    result = take_screenshot(config)

    # 输出 JSON 结果
    print("===RESULT===", flush=True)
    print(json.dumps(result, ensure_ascii=False), flush=True)

    # 清理配置文件
    try:
        os.remove(config_file)
    except:
        pass

    sys.exit(0 if result['success'] else 1)


if __name__ == '__main__':
    main()