Spaces:
Paused
Paused
| 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 | |
| def index(): | |
| """主页""" | |
| return render_template('index.html', | |
| viewport_width=VIEWPORT_WIDTH, | |
| viewport_height=VIEWPORT_HEIGHT) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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)}) | |
| 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) | |