deepseek-web/src/config.py

"""
DeepSeek Web Adapter 配置管理
"""
import json
import os
import logging
import tempfile
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Optional, List

logger = logging.getLogger(__name__)

CONFIG_DIR = Path.home() / ".deepseek_web_adapter"
CONFIG_FILE = CONFIG_DIR / "config.json"


@dataclass
class DeepSeekWebConfig:
    array_base_url: str = "http://47.92.218.141:8000"
    api_key: str = ""
    worker_id: str = ""
    worker_name: str = "DeepSeek-Web"
    worker_type: str = "ai"
    worker_location: str = ""
    admin_port: int = 9091
    log_level: str = "INFO"
    max_rounds: int = 30
    task_timeout: int = 600
    deepseek_url: str = "https://chat.deepseek.com"
    headless: bool = False
    user_data_dir: str = ""
    allowed_scopes: List[str] = field(default_factory=lambda: ["private", "chat"])
    reply_scope: str = "private"

    def __post_init__(self):
        if not self.user_data_dir:
            self.user_data_dir = str(Path.home() / ".deepseek_chat_profile")

    @property
    def profile(self) -> dict:
        return {
            "name": self.worker_name,
            "type": self.worker_type,
            "description": (
                "DeepSeek 网页版 AI Agent，通过 Playwright 控制浏览器与 DeepSeek 网页交互，"
                "支持本地命令执行、文件读写、目录浏览等操作。"
                "收到任务后会先回复确认，再逐步执行，过程中实时汇报进度。"
                "内置系统指令：[系统指令:取消] 终止任务，[系统指令:状态] 查看状态。"
            ),
            "capabilities": {
                "skills": [
                    "command_execution", "file_operations", "code_generation",
                    "text_analysis", "web_research", "data_processing"
                ],
                "domains": ["general", "software_development"],
                "tools": ["powershell", "python", "playwright"],
                "models": ["deepseek"],
                "languages": ["zh", "en"],
                "custom": {
                    "controlled_app": {
                        "name": "DeepSeek Web",
                        "vendor": "DeepSeek",
                        "description": "DeepSeek 网页版 AI 助手",
                    },
                    "control_method": {
                        "protocol": "Playwright",
                        "description": "通过 Playwright 控制 Chrome 浏览器与 DeepSeek 网页交互"
                    },
                    "message_types": self.allowed_scopes,
                    "max_concurrent_tasks": 1,
                    "system_commands": {
                        "取消": "终止当前任务",
                        "状态": "返回当前任务状态",
                    }
                }
            },
            "organization": "Array",
            "location": self.worker_location,
            "timezone": "Asia/Shanghai",
            "tags": ["AI", "DeepSeek", "Agent", "Web"],
            "max_concurrent_tasks": 1,
        }

    def save(self):
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        data = asdict(self)
        tmp_fd, tmp_path = tempfile.mkstemp(
            suffix=".tmp", prefix="cfg_", dir=str(CONFIG_DIR), text=True
        )
        try:
            with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            os.replace(tmp_path, str(CONFIG_FILE))
        except Exception:
            try:
                os.unlink(tmp_path)
            except OSError:
                pass
            raise
        logger.info(f"配置已保存到 {CONFIG_FILE}")

    @classmethod
    def load(cls) -> "DeepSeekWebConfig":
        if not CONFIG_FILE.exists():
            return cls()
        try:
            with open(CONFIG_FILE, "r", encoding="utf-8") as f:
                data = json.load(f)
            return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
        except Exception as e:
            logger.warning(f"加载配置失败: {e}，使用默认配置")
            return cls()

    @classmethod
    def from_env(cls) -> "DeepSeekWebConfig":
        cfg = cls()
        cfg.array_base_url = os.getenv("ARRAY_BASE_URL", cfg.array_base_url)
        cfg.api_key = os.getenv("ARRAY_API_KEY", cfg.api_key)
        cfg.worker_id = os.getenv("ARRAY_WORKER_ID", cfg.worker_id)
        cfg.admin_port = int(os.getenv("ADMIN_PORT", str(cfg.admin_port)))
        cfg.log_level = os.getenv("LOG_LEVEL", cfg.log_level)
        return cfg


config = DeepSeekWebConfig.load()