from __future__ import annotations from typing import Any, Literal from pydantic import BaseModel, Field, field_validator ConfirmStatus = Literal["PENDING", "TRUSTED", "SUSPICIOUS", "IGNORED", "NEED_MORE_INFO"] ItemType = Literal["service", "process"] AiProviderType = Literal["OPENAI", "OPENAI_COMPATIBLE", "GOOGLE_GEMINI"] MouseAutomationAction = Literal["move_to", "move_rel", "click", "double_click", "right_click", "drag_to", "scroll"] KeyboardAutomationAction = Literal["press", "hotkey", "write", "key_down", "key_up"] AutomationNodeType = str class StatusUpdate(BaseModel): confirm_status: ConfirmStatus user_note: str | None = None class BatchStatusUpdate(BaseModel): ids: list[int] = Field(default_factory=list) confirm_status: ConfirmStatus user_note: str | None = None class AiImportItem(BaseModel): type: ItemType name: str description: str | None = None judgement: Literal["TRUSTED", "SUSPICIOUS", "NEED_MORE_INFO"] risk_level: Literal["LOW", "MEDIUM", "HIGH"] reason: str | None = None suggestion: str | None = None tags: list[str] | None = None @field_validator("tags", mode="before") @classmethod def normalize_tags(cls, value: Any) -> list[str] | None: if value is None: return None if isinstance(value, str): return [value] if isinstance(value, list): names: list[str] = [] for item in value: if isinstance(item, str): names.append(item) elif isinstance(item, dict) and item.get("name"): names.append(str(item["name"])) return names raise ValueError("tags must be a string, list of strings, or list of objects with name") class AiImportRequest(BaseModel): items: list[AiImportItem] class PromptRequest(BaseModel): ids: list[int] | None = None scope: Literal["selected", "pending"] = "pending" class TagCreate(BaseModel): name: str = Field(min_length=1, max_length=80) description: str | None = None is_controllable: bool = True class TagUpdate(BaseModel): name: str = Field(min_length=1, max_length=80) description: str | None = None is_controllable: bool = True class TagAssignRequest(BaseModel): tag_ids: list[int] = Field(default_factory=list) class ProcessStartRequest(BaseModel): command: str = Field(min_length=1) cwd: str | None = None class AutomationPowerRequest(BaseModel): delay_seconds: int = Field(default=0, ge=0, le=86400) force: bool = False reason: str | None = Field(default=None, max_length=512) class AutomationProgramStartRequest(BaseModel): command: str = Field(min_length=1) cwd: str | None = None shell: bool = True class AutomationProgramStopRequest(BaseModel): pid: int | None = Field(default=None, ge=0) name: str | None = Field(default=None, min_length=1) timeout_seconds: float = Field(default=8, ge=0, le=60) kill_after_timeout: bool = True class AutomationScreenshotRequest(BaseModel): save_path: str | None = None include_base64: bool = True class AutomationMouseRequest(BaseModel): action: MouseAutomationAction x: int | None = None y: int | None = None duration: float = Field(default=0, ge=0, le=60) button: Literal["left", "middle", "right"] = "left" clicks: int = Field(default=1, ge=1, le=20) amount: int = 0 class AutomationKeyboardRequest(BaseModel): action: KeyboardAutomationAction key: str | None = None keys: list[str] | None = None text: str | None = None interval: float = Field(default=0, ge=0, le=10) class AutomationVisionAnalyzeRequest(BaseModel): provider_id: int | None = None model_id: int | None = None temperature: float = Field(default=0.1, ge=0, le=2) class AutomationScreenshotCaptureRequest(BaseModel): save: bool = True class AutomationElementLocateRequest(BaseModel): provider_id: int | None = None model_id: int | None = None temperature: float = Field(default=0.1, ge=0, le=2) class AutomationActionBase(BaseModel): screen_id: int | None = None provider_id: int | None = None model_id: int | None = None temperature: float = Field(default=0.1, ge=0, le=2) workflow_id: int | None = None node_id: int | None = None class AutomationMouseActionRequest(AutomationActionBase): x: int y: int mouse_action: Literal["click", "double_click", "right_click"] class AutomationKeyboardActionRequest(AutomationActionBase): keys: list[str] = Field(min_length=1) class AutomationTextInputRequest(AutomationActionBase): text: str class AutomationStartProgramRequest(AutomationActionBase): command: str = Field(min_length=1) cwd: str | None = None shell: bool = True class AutomationCloseProgramsRequest(BaseModel): pids: list[int] | None = None class AutomationWorkflowPosition(BaseModel): x: float = 80 y: float = 80 class AutomationWorkflowValueRef(BaseModel): source: Literal["literal", "variable", "node_output", "runtime"] value: Any = None name: str | None = None node_id: str | None = None output: str | None = None class AutomationWorkflowNode(BaseModel): id: str = Field(min_length=1, max_length=120) type: AutomationNodeType = Field(min_length=1, max_length=120) title: str | None = Field(default=None, max_length=160) position: AutomationWorkflowPosition = Field(default_factory=AutomationWorkflowPosition) params: dict[str, Any] = Field(default_factory=dict) inputs: dict[str, AutomationWorkflowValueRef | Any] = Field(default_factory=dict) class AutomationWorkflowSaveRequest(BaseModel): schema_version: Literal["workflow/v1"] = "workflow/v1" workflow_key: str | None = Field(default=None, min_length=1, max_length=80, pattern=r"^[a-zA-Z0-9][a-zA-Z0-9_-]*$") name: str = Field(min_length=1, max_length=160) description: str | None = None variables: dict[str, Any] = Field(default_factory=dict) settings: dict[str, Any] = Field(default_factory=dict) nodes: list[AutomationWorkflowNode] = Field(default_factory=list) edges: list[dict[str, Any]] = Field(default_factory=list) class AutomationWorkflowRunRequest(BaseModel): provider_id: int | None = None model_id: int | None = None temperature: float | None = Field(default=None, ge=0, le=2) variables: dict[str, Any] = Field(default_factory=dict) class AutomationWorkflowPlanRequest(BaseModel): requirement: str = Field(min_length=1, max_length=4000) provider_id: int | None = None model_id: int | None = None temperature: float | None = Field(default=None, ge=0, le=2) class AutomationWorkflowPlanContinueRequest(BaseModel): session_id: str = Field(min_length=1) user_message: str = Field(min_length=1, max_length=4000) provider_id: int | None = None model_id: int | None = None temperature: float | None = Field(default=None, ge=0, le=2) class SystemSettingsUpdate(BaseModel): default_ai_provider_id: int | None = None default_ai_model_id: int | None = None default_ai_temperature: float | None = Field(default=None, ge=0, le=2) automation_file_root: str | None = None automation_screen_path: str | None = None automation_error_path: str | None = None automation_runtime_path: str | None = None automation_auto_screenshot_enabled: bool | None = None automation_auto_screenshot_interval: int | None = Field(default=None, ge=1, le=3600) automation_remote_token: str | None = Field(default=None, max_length=256) class AiProviderCreate(BaseModel): name: str = Field(min_length=1, max_length=120) provider_type: AiProviderType base_url: str | None = None api_key: str | None = None enabled: bool = True class AiProviderUpdate(BaseModel): name: str = Field(min_length=1, max_length=120) provider_type: AiProviderType base_url: str | None = None api_key: str | None = None clear_api_key: bool = False enabled: bool = True class AiModelCreate(BaseModel): provider_id: int name: str = Field(min_length=1, max_length=160) display_name: str | None = None is_default: bool = False class AiModelUpdate(BaseModel): provider_id: int name: str = Field(min_length=1, max_length=160) display_name: str | None = None is_default: bool = False class AiChatRequest(BaseModel): provider_id: int model_id: int prompt: str = Field(min_length=1) temperature: float = Field(default=0.2, ge=0, le=2) class AiAnalyzeRequest(BaseModel): provider_id: int model_id: int temperature: float = Field(default=0.2, ge=0, le=2) ids: list[int] | None = None scope: Literal["selected", "pending"] = "pending"