python+openai兼容api+html实现类似Clawdbot(MoltBot)openclaw的ai自主分析任务编写代码运行完成任务代码bfwbot-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python+openai兼容api+html实现类似Clawdbot(MoltBot)openclaw的ai自主分析任务编写代码运行完成任务代码bfwbot,升级了ai可自主打开浏览器搜索或填写信息完成任务，增加了单独的审核任务完成情况agent。

代码标签： python opena 兼容 api html 类似 Clawdbot MoltBot openc

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

import os
import sys
import json
import asyncio
import platform
import re
import datetime
import shutil
import time
import uuid
from typing import List, Dict, Any, Optional

# Web 框架依赖
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File, Request, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

# AI 依赖
from openai import OpenAI
# Playwright 依赖
from playwright.async_api import async_playwright, Page, BrowserContext

# ================= 配置区域 =================
# 主规划模型 (建议用推理能力强的)
PLANNER_CONFIG = {
    "api_key": "sk-",
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "model": "qwen-max"  # 逻辑规划模型
}



CODER_CONFIG = {
    "api_key": "sk-869c3bfe81f041f39ca447a04fb8149d",
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "model": "qwen3-coder-flash"  # 写代码模型
}


# 登录认证配置
AUTH_CONFIG = {
    "username": "admin",
    "password": "bfwbot2025", 
    "cookie_name": "bfwbot_session",
    "cookie_value": "authorized_token_xyz_v2"
}

# 目录配置
WORKSPACE_DIR = os.path.join(os.getcwd(), "workspace")
UPLOAD_DIR = os.path.join(os.getcwd(), "uploads")

os.makedirs(WORKSPACE_DIR, exist_ok=True)
os.makedirs(UPLOAD_DIR, exist_ok=True)

# ================= 数据存储 (内存数据库) =================
CHAT_SESSIONS: Dict[str, Dict[str, Any]] = {}

def save_log_to_session(session_id: str, log_entry: dict):
    if session_id not in CHAT_SESSIONS:
        CHAT_SESSIONS[session_id] = {
            "title": "新会话", 
            "timestamp": time.time(), 
            "logs": []
        }
    
    if log_entry.get("type") == "message" and log_entry.get("content"):
        current_title = CHAT_SESSIONS[session_id]["title"]
        if current_title in ["新会话", "无标题会话"]:
            title = log_entry.get("content", "")[:20]
            if not title and log_entry.get("attachments"):
                title = "文件处理会话"
            elif not title:
                title = "无标题会话"
            CHAT_SESSIONS[session_id]["title"] = title

    CHAT_SESSIONS[session_id]["logs"].append(log_entry)
    CHAT_SESSIONS[session_id]["timestamp"] = time.time()

# ================= 1. 环境与工具类 =================
class SystemEnvironment:
    @staticmethod
    def get_info() -> str:
        base_url = "http://localhost:8000"
        return (
            f"OS: {platform.system()} {platform.release()}\n"
            f"Web Root: {base_url}\n"
            f"【URL映射规则】(输出给用户时必须使用Web路径):\n"
            f" - 本地 {WORKSPACE_DIR} -> {base_url}/download/\n"
            f" - 本地 {UPLOAD_DIR}    -> {base_url}/uploads/\n"
        )

# ================= 2. 增强版自主浏览器智能体 =================
class BrowserAgent:
    def __init__(self, send_log_func):
        self.send_log = send_log_func
        self.playwright = None
        self.browser = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None
        self.is_active = False

    async def start(self):
        if not self.is_active:
            await self.send_log("browser", "正在启动浏览器引擎...")
            self.playwright = await async_playwright().start()
            self.browser = await self.playwright.chromium.launch(headless=False, args=["--start-maximized"]) # Headless=False 可视化调试，服务器部署改为True
            self.context = await self.browser.new_context(
                viewport={'width': 1280, 'height': 800},
                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"
            )
            self.page = await self.context.new_page()
            self.is_active = True

    async def stop(self):
        if self.is_active:
            if self.context: await self.context.close()
            if self.browser: await self.browser.close()
            if self.playwright: await self.playwright.stop()
            self.is_active = False
            self.page = None

    async def _capture_state(self) -> str:
        """获取当前页面截图和交互元素摘要"""
        if not self.page: return "浏览器未启动"
        
        # 截图
        filename = f"browser_{int(time.time())}.jpg"
        filepath = os.path.join(UPLOAD_DIR, filename)
        await self.page.screenshot(path=filepath, type="jpeg", quality=50)
        screenshot_url = f"/uploads/{filename}"
        
        await self.send_log("browser_snapshot", screenshot_url)

        # 提取关键交互元素 (简化版DOM)
        elements_script = """
            () => {
                const interactives = [];
                document.querySelectorAll('a, button, input, textarea, select').forEach((el, index) => {
                    if (el.offsetParent !== null) { // 可见元素
                        let text = el.innerText || el.placeholder || el.value || '';
                        text = text.replace(/\\s+/g, ' ').trim().substring(0, 30);
                        let label = el.tagName.toLowerCase();
                        if (el.id) label += `#${el.id}`;
                        interactives.push(`[${index}] <${label}> ${text}`);
                        el.setAttribute('data-ai-id', index); // 标记以便后续操作
                    }
 .........完整代码请登录后点击上方下载按钮下载查看