win_monitor/backend/app/schemas.py

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 AutomationWorkflowImportRequest(BaseModel):
    workflow: AutomationWorkflowSaveRequest
    conflict_strategy: Literal["error", "replace"] = "error"


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"