Sandbox 复用策略

class SmartSandboxManager:    """智能 Sandbox 复用管理器"""
    
    def __init__(self):
        self.sandboxes = {}  # key -> sandbox
        self.usage_count = {}  # key -> count
    
    def get_sandbox(self, user_id: str, session_id: str):        """获取或创建 Sandbox（支持复用）"""
        key = f"{user_id}:{session_id}"
        
        if key not in self.sandboxes:
            self.sandboxes[key] = Sandbox.create(
                template_type=TemplateType.BROWSER,
                template_name=os.getenv("BROWSER_TEMPLATE_NAME"),
                sandbox_idle_timeout_seconds=1800
            )
            self.usage_count[key] = 0
        
        self.usage_count[key] += 1
        return self.sandboxes[key]    
    def should_recreate(self, key: str, max_reuse=50):        """判断是否需要重建（防止状态累积）"""
        return self.usage_count.get(key, 0) >= max_reuse    
    def recreate_if_needed(self, key: str):        """按需重建 Sandbox"""
        if self.should_recreate(key):            if key in self.sandboxes:
                self.sandboxes[key].delete()                del self.sandboxes[key]

错误处理和重试机制

使用 tenacity 库实现智能重试：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_typeclass SandboxError(Exception):    """Sandbox 操作异常"""
    pass@retry(
    retry=retry_if_exception_type(SandboxError),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10))def execute_with_retry(sandbox, operation):    """带重试的操作执行"""
    try:        return operation(sandbox)    except ConnectionError:        raise SandboxError("连接失败")    except TimeoutError:        raise SandboxError("操作超时")    except Exception as e:        print(f"操作失败: {e}")        raise SandboxError(f"操作失败: {e}")# 使用示例def navigate_page(sandbox):    with sync_playwright() as p:
        browser = p.chromium.connect_over_cdp(sandbox.cdp_url)
        page = browser.contexts[0].pages[0]
        page.goto("https://example.com", timeout=30000)        return page.title()
result = execute_with_retry(sandbox, navigate_page)

安全性最佳实践

环境变量保护

import osfrom dotenv import load_dotenv
load_dotenv()# 验证必需的环境变量required_vars = ["DASHSCOPE_API_KEY", "AGENTRUN_ACCOUNT_ID"]
missing_vars = [var for var in required_vars if not os.getenv(var)]if missing_vars:    raise ValueError(f"缺少必需的环境变量: {', '.join(missing_vars)}")# 敏感信息不要硬编码API_KEY = os.getenv("DASHSCOPE_API_KEY")
ACCESS_KEY_ID = os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
ACCESS_KEY_SECRET = os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

URL 白名单

ALLOWED_DOMAINS = [    'example.com',    'aliyun.com',    'alibaba.com']def is_url_allowed(url: str) -> bool:    """检查 URL 是否在白名单中"""
    from urllib.parse import urlparse
    domain = urlparse(url).netloc    return any(allowed in domain for allowed in ALLOWED_DOMAINS)def safe_navigate(page, url: str):    """安全导航"""
    if not is_url_allowed(url):        raise ValueError(f"URL 不在白名单中: {url}")
    page.goto(url)

日志脱敏

import redef sanitize_log(log_text: str) -> str:    """日志脱敏"""
    # 脱敏 API Key
    log_text = re.sub(r'sk-[a-zA-Z0-9]{20,}', 'sk-***', log_text)    # 脱敏 Access Key
    log_text = re.sub(r'LTAI[a-zA-Z0-9]{12,}', 'LTAI***', log_text)    # 脱敏密码
    log_text ttzbw.org.cn= re.sub(r'password["\s:=]+[^"\s,}]+', 'password: ***', log_text, flags=re.IGNORECASE)    return log_text# 使用print(sanitize_log(f"使用 API Key: {API_KEY}"))

可观测性与监控

日志记录最佳实践

import loggingfrom datetime import datetime# 配置日志logging.basicConfig(
    level=logging.INFO,    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=ttkq.mo.cn[
        logging.FileHandler(f'sandbox_{datetime.now().strftime("%Y%m%d")}.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)class MonitoredSandboxManager:    """带监控的 Sandbox 管理器"""
    
    def create_sandbox(self, **kwargs):        """创建 Sandbox（带日志）"""
        start_time =ttkq-24.com time.time()
        logger.info(f"开始创建 Sandbox: {kwargs}")        
        try:
            sandbox = Sandbox.create(**kwargs)
            duration yixue5.com= time.time() - start_time
            logger.info(f"Sandbox 创建成功: {sandbox.sandbox_id}, 耗时: {duration:.2f}s")            return sandbox        except Exception as e:
            duration = time.time() - start_time
            logger.error(f"Sandbox 创建失败: {e}, 耗时: {duration:.2f}s")            raise
    
    def execute_task(self, sandbox, task_name: str, operation):        """执行任务（带日志）"""
        start_time = time.time()
        logger.info(f"开始执行任务: {task_name}, Sandbox: {sandbox.sandbox_id}")        
        try:
            result =ttkq-163.com operation(sandbox)
            duration = time.time() - start_time
            logger.info(f"任务执行成功: {task_name}, 耗时: {duration:.2f}s")            return result        except Exception as e:
            duration = time.time() - start_time
            logger.error(f"任务执行失败: {task_name}, 错误: {e}, 耗时: {duration:.2f}s")            raise

指标收集

from dataclasses import dataclassfrom typing import Dict, Listimport json@dataclassclass SandboxMetrics:    """Sandbox 指标"""
    sandbox_id: str
    create_time: float
    destroy_time: float ttkq-qq.cn= None
    total_requests: int = 0
    failed_requests:ttzbw.mo.cn int = 0
    total_duration: float = 0.0class MetricsCollector:    """指标收集器"""
    
    def __init__(self):
        self.metrics: Dict[str, SandboxMetrics] = {}    
    def record_creation(self, sandbox_id: str):        """记录创建"""
        self.metrics[sandbox_id] = SandboxMetrics(
            sandbox_id=sandbox_id,
            create_time=time.time()
        )    
    def record_request(self, sandbox_id: str, duration: float, success: bool):        """记录请求"""
        if sandbox_id in self.metrics:
            metric = self.metrics[sandbox_id]
            metric.total_requests += 1
            metric.total_duration += duration            if not success:
                metric.failed_requests += 1
    
    def record_destruction(self, sandbox_id: str):        """记录销毁"""
        if sandbox_id in self.metrics:
            self.metrics[sandbox_id].destroy_time = time.time()    
    def export_metrics(self, filepath: str):        """导出指标"""
        metrics_data =ttkqw.com [
            {                'sandbox_id': m.sandbox_id,                'create_time': m.create_time,                'destroy_time': m.destroy_time,                'total_requests': m.total_requests,                'failed_requests': m.failed_requests,                'success_rate': (m.total_requests - m.failed_requests) / m.total_requests if m.total_requests > 0 else 0,                'avg_duration': m.total_duration / m.total_requests if m.total_requests > 0 else 0,                'lifetime': m.destroy_time - m.create_time if m.destroy_time else time.time() - m.create_time
            }            for m in self.metrics.values()
        ]        
        with open(filepath, 'w') as f:
            json.dump(metrics_data, f, indent=2)# 使用collector = MetricsCollector()
collector.record_creation(sandbox.sandbox_id)# ... 执行任务 ...collector.export_metrics('metrics.json')

成本优化

按需创建与销毁

class CostOptimizedManager:    """成本优化的管理器"""
    
    def __init__(self, idle_threshold=300):
        self.idle_threshold = idle_threshold
        self.sandboxes zggren.com= {}
        self.last_used = {}    
    def get_sandbox(self, key: str):        """获取 Sandbox（懒加载）"""
        if key not in self.sandboxes:
            self.sandboxes[key] = Sandbox.create(
                template_type=TemplateType.BROWSER,
                template_name=os.getenv("BROWSER_TEMPLATE_NAME"),
                sandbox_idle_timeout_seconds=self.idle_threshold
            )
        
        self.last_used[key] = time.time()        return self.sandboxes[key]    
    def cleanup_idle(self): ttkqol.cn """清理闲置 Sandbox"""
        current_time  time.time()
        to_remove = []        
        for key, last_time in self.last_used.items():            if current_time - last_time > self.idle_threshold:
                to_remove.append(key)        
        for key in to_remove:            if key in self.sandboxes:
                self.sandboxes[key].delete()                del self.sandboxes[key]                del self.last_used[key]
                logger.info(f"清理闲置 Sandbox: {key}")

批量任务处理

async def batch_process_tasks(tasks: List[str], pool_size: int = 5):    """批量处理任务（复用 Sandbox）"""
    pool = SandboxPool(pool_size=pool_size)
    results = [ttzbapp.cn]    
    for task in tasks:
        sandbox = pool.acquire()        try:            # 处理任务
            result =huluzhibo.cn await process_task(sandbox, task)
            results.append(result)ttzbapp.com.cn finally:
            pool.release(sandbox)    
    return results

生产环境部署

环境配置

开发环境 (.env.dev)：

# 开发环境配置BROWSER_TEMPLATE_NAME=dev-browser-template
SANDBOX_IDLE_TIMEOUT=7200
POOL_SIZE=2
LOG_LEVEL=DEBUG

生产环境 (.env.prod)：

# 生产环境配置BROWSER_TEMPLATE_NAME=prod-browser-template
SANDBOX_IDLE_TIMEOUT=300
POOL_SIZE=10
LOG_LEVEL=INFO