Spaces:
Sleeping
Sleeping
| import os | |
| import io | |
| import base64 | |
| import time | |
| import threading | |
| import shutil | |
| 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 | |
| context_instance = None | |
| playwright_instance = None | |
| # 页面管理（多标签页） | |
| pages = {} | |
| active_page_id = None | |
| page_counter = 0 | |
| # 线程锁 | |
| browser_lock = threading.Lock() | |
| # ========== 文件系统配置 ========== | |
| # 用户文件存储目录 | |
| USER_FILES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'user_files') | |
| os.makedirs(USER_FILES_DIR, exist_ok=True) | |
| # 确保初始有一些示例文件 | |
| def init_sample_files(): | |
| """初始化示例文件""" | |
| docs_dir = os.path.join(USER_FILES_DIR, '文档') | |
| os.makedirs(docs_dir, exist_ok=True) | |
| readme_file = os.path.join(docs_dir, '欢迎使用云电脑.txt') | |
| if not os.path.exists(readme_file): | |
| with open(readme_file, 'w', encoding='utf-8') as f: | |
| f.write("""欢迎使用云电脑！ | |
| 这是一个基于云端计算的云电脑系统。 | |
| 功能特性： | |
| - 云端浏览器：所有网页在服务器端渲染 | |
| - 文件管理：可以新建、编辑、删除文件 | |
| - 多标签页：支持打开多个浏览器标签 | |
| - 全屏模式：沉浸式体验 | |
| 提示： | |
| - 双击桌面图标打开应用 | |
| - 点击浏览器画面进行操作 | |
| - 按 ESC 退出全屏模式 | |
| 祝您使用愉快！ | |
| """) | |
| notes_dir = os.path.join(USER_FILES_DIR, '笔记') | |
| os.makedirs(notes_dir, exist_ok=True) | |
| init_sample_files() | |
| # ========== 浏览器配置 ========== | |
| 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') | |
| # 截图质量（越低越快，50% 是速度和质量的平衡点） | |
| SCREENSHOT_QUALITY = 50 | |
| def get_browser(): | |
| """获取或创建浏览器实例""" | |
| global browser_instance, context_instance, playwright_instance, pages, active_page_id, page_counter | |
| with browser_lock: | |
| 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', | |
| '--disable-background-networking', | |
| '--disable-default-apps', | |
| '--disable-extensions', | |
| '--disable-sync', | |
| '--disable-translate', | |
| '--hide-scrollbars', | |
| '--mute-audio', | |
| '--no-first-run', | |
| ] | |
| ) | |
| context_instance = 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_counter += 1 | |
| page_id = f'page_{page_counter}' | |
| page = context_instance.new_page() | |
| page.goto(DEFAULT_URL, timeout=30000, wait_until='domcontentloaded') | |
| pages[page_id] = page | |
| active_page_id = page_id | |
| return pages[active_page_id] | |
| def get_page(page_id=None): | |
| """获取指定页面或当前活动页面""" | |
| global active_page_id | |
| if page_id is None: | |
| page_id = active_page_id | |
| if page_id and page_id in pages: | |
| return pages[page_id] | |
| return get_browser() | |
| def take_screenshot(page_id=None): | |
| """截取当前页面截图 - 优化版（低延迟）""" | |
| page = get_page(page_id) | |
| try: | |
| screenshot = page.screenshot( | |
| type='jpeg', | |
| quality=SCREENSHOT_QUALITY, # 降低质量以减小体积 | |
| full_page=False, | |
| animations='disabled' # 禁用动画，加快截图 | |
| ) | |
| return base64.b64encode(screenshot).decode('utf-8') | |
| except Exception as e: | |
| print(f"Screenshot error: {e}") | |
| return None | |
| def create_new_page(url=None): | |
| """创建新标签页""" | |
| global page_counter, active_page_id, pages | |
| with browser_lock: | |
| get_browser() | |
| page_counter += 1 | |
| page_id = f'page_{page_counter}' | |
| page = context_instance.new_page() | |
| if url: | |
| if not url.startswith('http://') and not url.startswith('https://'): | |
| url = 'https://' + url | |
| page.goto(url, timeout=30000, wait_until='domcontentloaded') | |
| else: | |
| page.goto(DEFAULT_URL, timeout=30000, wait_until='domcontentloaded') | |
| pages[page_id] = page | |
| active_page_id = page_id | |
| return page_id, page | |
| def close_page(page_id): | |
| """关闭标签页""" | |
| global active_page_id, pages | |
| with browser_lock: | |
| if page_id in pages and len(pages) > 1: | |
| pages[page_id].close() | |
| del pages[page_id] | |
| if active_page_id == page_id: | |
| active_page_id = list(pages.keys())[0] | |
| return True | |
| return False | |
| def switch_page(page_id): | |
| """切换到指定页面""" | |
| global active_page_id | |
| with browser_lock: | |
| if page_id in pages: | |
| active_page_id = page_id | |
| return True | |
| return False | |
| def get_all_pages(): | |
| """获取所有页面信息""" | |
| result = [] | |
| for page_id, page in pages.items(): | |
| try: | |
| result.append({ | |
| 'id': page_id, | |
| 'title': page.title(), | |
| 'url': page.url, | |
| 'active': page_id == active_page_id | |
| }) | |
| except: | |
| result.append({ | |
| 'id': page_id, | |
| 'title': '加载中...', | |
| 'url': '', | |
| 'active': page_id == active_page_id | |
| }) | |
| return result | |
| # ========== 文件系统 API ========== | |
| def get_relative_path(path): | |
| """获取相对于用户目录的安全路径""" | |
| # 规范化路径，防止路径遍历攻击 | |
| if not path: | |
| path = '/' | |
| # 去掉开头的 / | |
| if path.startswith('/'): | |
| path = path[1:] | |
| # 拼接完整路径 | |
| full_path = os.path.join(USER_FILES_DIR, path) | |
| # 规范化路径 | |
| full_path = os.path.normpath(full_path) | |
| # 确保路径在用户目录内 | |
| if not full_path.startswith(os.path.normpath(USER_FILES_DIR)): | |
| return None, None | |
| # 计算相对路径 | |
| rel_path = os.path.relpath(full_path, USER_FILES_DIR) | |
| if rel_path == '.': | |
| rel_path = '' | |
| return full_path, rel_path | |
| def list_files(): | |
| """列出文件和文件夹""" | |
| path = request.args.get('path', '') | |
| full_path, rel_path = get_relative_path(path) | |
| if full_path is None: | |
| return jsonify({'success': False, 'error': '无效路径'}) | |
| if not os.path.exists(full_path): | |
| return jsonify({'success': False, 'error': '路径不存在'}) | |
| if not os.path.isdir(full_path): | |
| return jsonify({'success': False, 'error': '不是文件夹'}) | |
| items = [] | |
| try: | |
| for item in os.listdir(full_path): | |
| item_path = os.path.join(full_path, item) | |
| is_dir = os.path.isdir(item_path) | |
| size = os.path.getsize(item_path) if not is_dir else 0 | |
| # 获取修改时间 | |
| mtime = os.path.getmtime(item_path) | |
| items.append({ | |
| 'name': item, | |
| 'isDir': is_dir, | |
| 'size': size, | |
| 'modified': mtime, | |
| 'path': os.path.join(rel_path, item) if rel_path else item | |
| }) | |
| # 排序：文件夹在前，按名称排序 | |
| items.sort(key=lambda x: (not x['isDir'], x['name'].lower())) | |
| return jsonify({ | |
| 'success': True, | |
| 'path': rel_path, | |
| 'items': items | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def read_file(): | |
| """读取文件内容""" | |
| path = request.args.get('path', '') | |
| full_path, rel_path = get_relative_path(path) | |
| if full_path is None: | |
| return jsonify({'success': False, 'error': '无效路径'}) | |
| if not os.path.exists(full_path): | |
| return jsonify({'success': False, 'error': '文件不存在'}) | |
| if os.path.isdir(full_path): | |
| return jsonify({'success': False, 'error': '不能读取文件夹'}) | |
| try: | |
| # 检查文件大小，太大的文件不读取 | |
| size = os.path.getsize(full_path) | |
| if size > 10 * 1024 * 1024: # 10MB | |
| return jsonify({'success': False, 'error': '文件太大，无法读取'}) | |
| # 尝试以文本方式读取 | |
| try: | |
| with open(full_path, 'r', encoding='utf-8') as f: | |
| content = f.read() | |
| return jsonify({ | |
| 'success': True, | |
| 'path': rel_path, | |
| 'name': os.path.basename(full_path), | |
| 'content': content, | |
| 'size': size, | |
| 'isText': True | |
| }) | |
| except UnicodeDecodeError: | |
| # 二进制文件，返回 base64 | |
| with open(full_path, 'rb') as f: | |
| content = base64.b64encode(f.read()).decode('utf-8') | |
| return jsonify({ | |
| 'success': True, | |
| 'path': rel_path, | |
| 'name': os.path.basename(full_path), | |
| 'content': content, | |
| 'size': size, | |
| 'isText': False | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def write_file(): | |
| """写入文件内容""" | |
| data = request.get_json() | |
| path = data.get('path', '') | |
| content = data.get('content', '') | |
| full_path, rel_path = get_relative_path(path) | |
| if full_path is None: | |
| return jsonify({'success': False, 'error': '无效路径'}) | |
| try: | |
| # 确保父目录存在 | |
| parent_dir = os.path.dirname(full_path) | |
| os.makedirs(parent_dir, exist_ok=True) | |
| with open(full_path, 'w', encoding='utf-8') as f: | |
| f.write(content) | |
| return jsonify({ | |
| 'success': True, | |
| 'path': rel_path, | |
| 'size': len(content.encode('utf-8')) | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def create_file(): | |
| """新建文件或文件夹""" | |
| data = request.get_json() | |
| path = data.get('path', '') | |
| is_dir = data.get('isDir', False) | |
| full_path, rel_path = get_relative_path(path) | |
| if full_path is None: | |
| return jsonify({'success': False, 'error': '无效路径'}) | |
| if os.path.exists(full_path): | |
| return jsonify({'success': False, 'error': '已存在同名文件'}) | |
| try: | |
| if is_dir: | |
| os.makedirs(full_path, exist_ok=True) | |
| else: | |
| # 确保父目录存在 | |
| parent_dir = os.path.dirname(full_path) | |
| os.makedirs(parent_dir, exist_ok=True) | |
| # 创建空文件 | |
| with open(full_path, 'w', encoding='utf-8') as f: | |
| pass | |
| return jsonify({ | |
| 'success': True, | |
| 'path': rel_path, | |
| 'isDir': is_dir | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def delete_file(): | |
| """删除文件或文件夹""" | |
| data = request.get_json() | |
| path = data.get('path', '') | |
| full_path, rel_path = get_relative_path(path) | |
| if full_path is None: | |
| return jsonify({'success': False, 'error': '无效路径'}) | |
| if not os.path.exists(full_path): | |
| return jsonify({'success': False, 'error': '文件不存在'}) | |
| try: | |
| if os.path.isdir(full_path): | |
| shutil.rmtree(full_path) | |
| else: | |
| os.remove(full_path) | |
| return jsonify({'success': True}) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def rename_file(): | |
| """重命名文件""" | |
| data = request.get_json() | |
| path = data.get('path', '') | |
| new_name = data.get('newName', '') | |
| full_path, rel_path = get_relative_path(path) | |
| if full_path is None: | |
| return jsonify({'success': False, 'error': '无效路径'}) | |
| if not os.path.exists(full_path): | |
| return jsonify({'success': False, 'error': '文件不存在'}) | |
| if not new_name or '/' in new_name or '\\' in new_name: | |
| return jsonify({'success': False, 'error': '无效的文件名'}) | |
| try: | |
| parent_dir = os.path.dirname(full_path) | |
| new_path = os.path.join(parent_dir, new_name) | |
| if os.path.exists(new_path): | |
| return jsonify({'success': False, 'error': '已存在同名文件'}) | |
| os.rename(full_path, new_path) | |
| new_rel_path = os.path.join(os.path.dirname(rel_path), new_name) if os.path.dirname(rel_path) else new_name | |
| return jsonify({ | |
| 'success': True, | |
| 'oldPath': rel_path, | |
| 'newPath': new_rel_path | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| # ========== 浏览器 API ========== | |
| def index(): | |
| """主页 - 桌面环境""" | |
| return render_template('desktop.html', | |
| viewport_width=VIEWPORT_WIDTH, | |
| viewport_height=VIEWPORT_HEIGHT) | |
| def navigate(): | |
| """导航到指定网址""" | |
| data = request.get_json() | |
| url = data.get('url', '') | |
| page_id = data.get('pageId') | |
| if not url: | |
| return jsonify({'success': False, 'error': '请输入网址'}) | |
| if not url.startswith('http://') and not url.startswith('https://'): | |
| url = 'https://' + url | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.goto(url, timeout=30000, wait_until='domcontentloaded') | |
| time.sleep(0.3) # 减少等待时间 | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except PlaywrightTimeoutError: | |
| return jsonify({'success': False, 'error': '页面加载超时'}) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def screenshot(): | |
| """获取当前页面截图 - 快速版""" | |
| page_id = request.args.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| 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) | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.mouse.click(x, y) | |
| time.sleep(0.15) # 大幅减少等待时间 | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def type_text(): | |
| """处理键盘输入 - 快速响应""" | |
| data = request.get_json() | |
| text = data.get('text', '') | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.keyboard.type(text) | |
| time.sleep(0.1) # 减少等待时间 | |
| screenshot = take_screenshot(page_id) | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def keypress(): | |
| """处理特殊按键 - 快速响应""" | |
| data = request.get_json() | |
| key = data.get('key', '') | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| 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.1) # 减少等待时间 | |
| screenshot = take_screenshot(page_id) | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| 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) | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.mouse.wheel(delta_x, delta_y) | |
| time.sleep(0.08) # 大幅减少等待时间 | |
| screenshot = take_screenshot(page_id) | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def go_back(): | |
| """后退""" | |
| data = request.get_json() | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.go_back() | |
| time.sleep(0.3) | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def go_forward(): | |
| """前进""" | |
| data = request.get_json() | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.go_forward() | |
| time.sleep(0.3) | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def reload(): | |
| """刷新""" | |
| data = request.get_json() | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| page = get_page(page_id) | |
| page.reload() | |
| time.sleep(0.3) | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def new_tab(): | |
| """新建标签页""" | |
| data = request.get_json() or {} | |
| url = data.get('url') | |
| try: | |
| with browser_lock: | |
| page_id, page = create_new_page(url) | |
| time.sleep(0.2) | |
| screenshot = take_screenshot(page_id) | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'pageId': page_id, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def close_tab(): | |
| """关闭标签页""" | |
| data = request.get_json() | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| success = close_page(page_id) | |
| page = get_page() | |
| screenshot = take_screenshot() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': success, | |
| 'screenshot': screenshot, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def switch_tab(): | |
| """切换标签页""" | |
| data = request.get_json() | |
| page_id = data.get('pageId') | |
| try: | |
| with browser_lock: | |
| success = switch_page(page_id) | |
| page = get_page() | |
| screenshot = take_screenshot() | |
| current_url = page.url | |
| title = page.title() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': success, | |
| 'screenshot': screenshot, | |
| 'url': current_url, | |
| 'title': title, | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| except Exception as e: | |
| return jsonify({'success': False, 'error': str(e)}) | |
| def status(): | |
| """获取浏览器状态""" | |
| try: | |
| with browser_lock: | |
| page = get_page() | |
| pages_list = get_all_pages() | |
| return jsonify({ | |
| 'success': True, | |
| 'ready': True, | |
| 'url': page.url, | |
| 'title': page.title(), | |
| 'pages': pages_list, | |
| 'activePageId': active_page_id | |
| }) | |
| 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) | |