app.py

import os
import io
import base64
import time
import threading
from flask import Flask, render_template, request, jsonify, send_file
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError

app = Flask(__name__)

# 浏览器实例管理
browser_instance = None
page_instance = None
playwright_instance = None

# 线程锁，确保 Playwright 操作的线程安全
browser_lock = threading.Lock()

# 配置
VIEWPORT_WIDTH = int(os.environ.get('VIEWPORT_WIDTH', 1280))
VIEWPORT_HEIGHT = int(os.environ.get('VIEWPORT_HEIGHT', 800))
DEFAULT_URL = os.environ.get('DEFAULT_URL', 'https://www.google.com')


def get_browser():
    """获取或创建浏览器实例"""
    global browser_instance, page_instance, playwright_instance
    
    if browser_instance is None:
        playwright_instance = sync_playwright().start()
        browser_instance = playwright_instance.chromium.launch(
            headless=True,
            args=[
                '--no-sandbox',
                '--disable-dev-shm-usage',
                '--disable-gpu',
                '--disable-extensions',
                '--disable-web-security',
                '--disable-features=IsolateOrigins,site-per-process',
            ]
        )
        context = browser_instance.new_context(
            viewport={'width': VIEWPORT_WIDTH, 'height': VIEWPORT_HEIGHT},
            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_instance = context.new_page()
        page_instance.goto(DEFAULT_URL, timeout=30000, wait_until='domcontentloaded')
    
    return page_instance


def take_screenshot():
    """截取当前页面截图"""
    page = get_browser()
    try:
        screenshot = page.screenshot(
            type='jpeg',
            quality=80,
            full_page=False
        )
        return base64.b64encode(screenshot).decode('utf-8')
    except Exception as e:
        print(f"Screenshot error: {e}")
        return None


@app.route('/')
def index():
    """主页"""
    return render_template('index.html', 
                         viewport_width=VIEWPORT_WIDTH,
                         viewport_height=VIEWPORT_HEIGHT)


@app.route('/api/navigate', methods=['POST'])
def navigate():
    """导航到指定网址"""
    data = request.get_json()
    url = data.get('url', '')
    
    if not url:
        return jsonify({'success': False, 'error': '请输入网址'})
    
    # 自动添加 https://
    if not url.startswith('http://') and not url.startswith('https://'):
        url = 'https://' + url
    
    try:
        page = get_browser()
        page.goto(url, timeout=30000, wait_until='domcontentloaded')
        time.sleep(0.5)  # 等待页面渲染
        
        screenshot = take_screenshot()
        current_url = page.url
        title = page.title()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot,
            'url': current_url,
            'title': title
        })
    except PlaywrightTimeoutError:
        return jsonify({'success': False, 'error': '页面加载超时'})
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/screenshot')
def screenshot():
    """获取当前页面截图"""
    try:
        page = get_browser()
        screenshot = take_screenshot()
        current_url = page.url
        title = page.title()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot,
            'url': current_url,
            'title': title
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/click', methods=['POST'])
def click():
    """处理点击事件"""
    data = request.get_json()
    x = data.get('x', 0)
    y = data.get('y', 0)
    
    try:
        page = get_browser()
        # 点击指定坐标
        page.mouse.click(x, y)
        time.sleep(0.3)  # 等待页面响应
        
        screenshot = take_screenshot()
        current_url = page.url
        title = page.title()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot,
            'url': current_url,
            'title': title
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/type', methods=['POST'])
def type_text():
    """处理键盘输入"""
    data = request.get_json()
    text = data.get('text', '')
    
    try:
        page = get_browser()
        page.keyboard.type(text)
        time.sleep(0.2)
        
        screenshot = take_screenshot()
        return jsonify({
            'success': True,
            'screenshot': screenshot
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/keypress', methods=['POST'])
def keypress():
    """处理特殊按键"""
    data = request.get_json()
    key = data.get('key', '')
    
    try:
        page = get_browser()
        
        # 处理特殊按键
        key_map = {
            'Enter': 'Enter',
            'Backspace': 'Backspace',
            'Tab': 'Tab',
            'Escape': 'Escape',
            'ArrowUp': 'ArrowUp',
            'ArrowDown': 'ArrowDown',
            'ArrowLeft': 'ArrowLeft',
            'ArrowRight': 'ArrowRight',
            ' ': 'Space',
        }
        
        if key in key_map:
            page.keyboard.press(key_map[key])
        elif len(key) == 1:
            page.keyboard.press(key)
        
        time.sleep(0.2)
        screenshot = take_screenshot()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/scroll', methods=['POST'])
def scroll():
    """处理滚动"""
    data = request.get_json()
    delta_x = data.get('deltaX', 0)
    delta_y = data.get('deltaY', 0)
    
    try:
        page = get_browser()
        page.mouse.wheel(delta_x, delta_y)
        time.sleep(0.2)
        
        screenshot = take_screenshot()
        return jsonify({
            'success': True,
            'screenshot': screenshot
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/go-back', methods=['POST'])
def go_back():
    """后退"""
    try:
        page = get_browser()
        page.go_back()
        time.sleep(0.5)
        
        screenshot = take_screenshot()
        current_url = page.url
        title = page.title()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot,
            'url': current_url,
            'title': title
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/go-forward', methods=['POST'])
def go_forward():
    """前进"""
    try:
        page = get_browser()
        page.go_forward()
        time.sleep(0.5)
        
        screenshot = take_screenshot()
        current_url = page.url
        title = page.title()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot,
            'url': current_url,
            'title': title
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/reload', methods=['POST'])
def reload():
    """刷新"""
    try:
        page = get_browser()
        page.reload()
        time.sleep(0.5)
        
        screenshot = take_screenshot()
        current_url = page.url
        title = page.title()
        
        return jsonify({
            'success': True,
            'screenshot': screenshot,
            'url': current_url,
            'title': title
        })
    except Exception as e:
        return jsonify({'success': False, 'error': str(e)})


@app.route('/api/status')
def status():
    """获取浏览器状态"""
    try:
        page = get_browser()
        return jsonify({
            'success': True,
            'ready': True,
            'url': page.url,
            'title': page.title()
        })
    except Exception as e:
        return jsonify({
            'success': True,
            'ready': False,
            'error': str(e)
        })


if __name__ == '__main__':
    port = int(os.environ.get('PORT', 7860))
    # 使用单线程模式，避免 Playwright 跨线程问题
    app.run(host='0.0.0.0', port=port, debug=False, threaded=False)