from __future__ import annotations import json import secrets import sqlite3 from typing import Any from fastapi import FastAPI, Header, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from . import ai_service, automation_service, settings_service, windows_automation from .control import ( CONFIRMED_CONTROL_STATUSES, restart_service, start_process, start_service, stop_process, stop_service, ) from .database import get_db, init_db from .scanner import now_iso, run_full_scan from .sensors import collect_sensors from .schemas import ( AiAnalyzeRequest, AiChatRequest, AiImportRequest, AiModelCreate, AiModelUpdate, AiProviderCreate, AiProviderUpdate, AutomationKeyboardRequest, AutomationKeyboardActionRequest, AutomationElementLocateRequest, AutomationMouseRequest, AutomationMouseActionRequest, AutomationPowerRequest, AutomationCloseProgramsRequest, AutomationScreenshotCaptureRequest, AutomationStartProgramRequest, AutomationProgramStartRequest, AutomationProgramStopRequest, AutomationScreenshotRequest, AutomationTextInputRequest, AutomationVisionAnalyzeRequest, AutomationWorkflowRunRequest, AutomationWorkflowSaveRequest, AutomationWorkflowPlanRequest, AutomationWorkflowPlanContinueRequest, BatchStatusUpdate, PromptRequest, StatusUpdate, SystemSettingsUpdate, TagAssignRequest, TagCreate, TagUpdate, ) from .smart import collect_all_smart, get_device_smart, scan_devices AI_PROMPT_TEMPLATE = """请作为资深的 Windows 系统安全专家，帮我分析下面这些 Windows 服务和进程是否可信，并严格按照 JSON 数组格式输出结果。 输出要求： 1. 必须且只能输出纯 JSON 数组，不要输出任何额外的解释、问候语，也不要使用 Markdown 代码块（如 ```json）包裹。 2. 每个对象必须包含以下 8 个字段：type、name、description、judgement、risk_level、reason、suggestion、tags。 3. type 只能是 "service" 或 "process"。 4. description 请简要说明该服务或进程的官方用途或常规功能（如果是未知/恶意程序，请描述其伪装意图或表现）。 5. judgement 只能是 "TRUSTED"、"SUSPICIOUS"、"NEED_MORE_INFO"。 6. risk_level 只能是 "LOW"、"MEDIUM"、"HIGH"。 7. 如果提供的信息不足以做出判断，请将 judgement 设为 "NEED_MORE_INFO"。 8. 待分析数据里的 tags 字段是当前已有标签上下文，不代表最终结论，但如果标签显示为“windows系统”或“本系统相关”，请在 reason 或 suggestion 中体现这一点。 9. 输出对象里的 tags 字段必须是字符串数组，填写你建议系统最终绑定到该对象上的标签名称。可以使用系统已有标签，也可以在确有必要时给出新的短标签名称；标签名称应简洁稳定，不要把长句放入标签。 JSON 格式示例： [ { "type": "service", "name": "WinDefend", "description": "Microsoft Defender 防病毒核心服务，负责保护系统免受恶意软件和间谍软件的威胁。", "judgement": "TRUSTED", "risk_level": "LOW", "reason": "这是 Microsoft 官方的安全组件，路径和名称符合系统原生服务的标准特征。", "suggestion": "可标记为可信，建议保持运行。", "tags": ["windows系统"] }, { "type": "process", "name": "unknown.exe", "description": "未知用途的执行文件，无明确的官方功能说明。", "judgement": "SUSPICIOUS", "risk_level": "HIGH", "reason": "进程位于用户 AppData 临时目录，启动命令行异常，且缺少有效的官方数字签名。", "suggestion": "建议立即隔离，检查文件的 SHA256 散列值及外部网络连接记录，不要直接运行或信任。", "tags": ["可疑程序"] } ] 下面是待分析数据： {pending_items_json} 系统中已有标签信息： {tags_json} """ app = FastAPI(title="Windows Monitor API", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def verify_automation_token(x_automation_token: str | None = Header(default=None)) -> None: configured = settings_service.automation_remote_token() if not configured: raise HTTPException(status_code=403, detail="Automation remote token is not configured") if not x_automation_token or not secrets.compare_digest(x_automation_token, configured): raise HTTPException(status_code=401, detail="Invalid automation token") @app.on_event("startup") def startup() -> None: init_db() def build_where( keyword: str | None, confirm_status: str | None, present: bool | None, fields: list[str], ) -> tuple[str, list[Any]]: clauses: list[str] = [] params: list[Any] = [] if keyword: like = f"%{keyword}%" clauses.append("(" + " OR ".join(f"{field} LIKE ?" for field in fields) + ")") params.extend([like] * len(fields)) if confirm_status: clauses.append("confirm_status = ?") params.append(confirm_status) if present is not None: clauses.append("is_present_now = ?") params.append(1 if present else 0) return ("WHERE " + " AND ".join(clauses)) if clauses else "", params def list_items( table: str, keyword: str | None, confirm_status: str | None, present: bool | None, page: int, page_size: int, fields: list[str], sort_by: str | None = None, sort_order: str | None = None, ) -> dict[str, Any]: where_sql, params = build_where(keyword, confirm_status, present, fields) order_sql = build_order_by(table, sort_by, sort_order) offset = (page - 1) * page_size with get_db() as conn: total = conn.execute(f"SELECT COUNT(*) AS total FROM {table} {where_sql}", params).fetchone()["total"] rows = conn.execute( f"SELECT * FROM {table} {where_sql} {order_sql} LIMIT ? OFFSET ?", [*params, page_size, offset], ).fetchall() rows = attach_item_metadata(conn, table_to_item_type(table), rows) return {"items": rows, "total": total, "page": page, "page_size": page_size} def build_order_by(table: str, sort_by: str | None, sort_order: str | None) -> str: allowed = { "windows_services": { "name", "display_name", "status", "start_type", "username", "is_present_now", "confirm_status", "first_seen_at", "last_seen_at", "updated_at", }, "windows_processes": { "name", "exe_path", "username", "status", "last_pid", "parent_pid", "is_present_now", "confirm_status", "create_time", "first_seen_at", "last_seen_at", "updated_at", }, } default_sql = "ORDER BY is_present_now DESC, last_seen_at DESC" if not sort_by or sort_by not in allowed.get(table, set()): return default_sql direction = "ASC" if sort_order == "asc" else "DESC" if sort_by == "is_present_now": return f"ORDER BY {sort_by} {direction}, last_seen_at DESC" return f"ORDER BY {sort_by} {direction}, is_present_now DESC, last_seen_at DESC" def get_item(table: str, item_id: int) -> dict[str, Any]: with get_db() as conn: item = conn.execute(f"SELECT * FROM {table} WHERE id = ?", (item_id,)).fetchone() if item: item = attach_item_metadata(conn, table_to_item_type(table), [item])[0] if not item: raise HTTPException(status_code=404, detail="Item not found") return item def table_to_item_type(table: str) -> str: if table == "windows_services": return "service" if table == "windows_processes": return "process" raise ValueError(f"Unsupported table: {table}") def bool_tag(row: dict[str, Any]) -> dict[str, Any]: item = dict(row) item["is_controllable"] = bool(item["is_controllable"]) item["is_builtin"] = bool(item["is_builtin"]) return item def all_tags(conn) -> list[dict[str, Any]]: return [ bool_tag(row) for row in conn.execute("SELECT * FROM tags ORDER BY is_builtin DESC, name ASC").fetchall() ] def tags_for_items(conn, item_type: str, item_ids: list[int]) -> dict[int, list[dict[str, Any]]]: if not item_ids: return {} placeholders = ",".join("?" for _ in item_ids) rows = conn.execute( f""" SELECT it.item_id, t.* FROM item_tags it JOIN tags t ON t.id = it.tag_id WHERE it.item_type = ? AND it.item_id IN ({placeholders}) ORDER BY t.name ASC """, [item_type, *item_ids], ).fetchall() result = {item_id: [] for item_id in item_ids} for row in rows: item_id = row["item_id"] tag = {key: value for key, value in row.items() if key != "item_id"} result.setdefault(item_id, []).append(bool_tag(tag)) return result def can_control_item(item_type: str, row: dict[str, Any], tags: list[dict[str, Any]]) -> bool: if row.get("confirm_status") not in CONFIRMED_CONTROL_STATUSES: return False if item_type == "process": protected_names = {"system idle process", "system", "registry"} if row.get("last_pid") in (0, 4) or (row.get("name") or "").lower() in protected_names: return False return all(tag.get("is_controllable", True) for tag in tags) def attach_item_metadata(conn, item_type: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: tag_map = tags_for_items(conn, item_type, [row["id"] for row in rows]) enriched = [] for row in rows: item = dict(row) item["is_present_now"] = bool(item.get("is_present_now")) item["tags"] = tag_map.get(row["id"], []) item["can_control"] = can_control_item(item_type, item, item["tags"]) enriched.append(item) return enriched def update_one(table: str, item_id: int, payload: StatusUpdate) -> dict[str, Any]: get_item(table, item_id) with get_db() as conn: conn.execute( f"UPDATE {table} SET confirm_status = ?, user_note = ?, updated_at = ? WHERE id = ?", (payload.confirm_status, payload.user_note, now_iso(), item_id), ) return get_item(table, item_id) def update_batch(table: str, payload: BatchStatusUpdate) -> dict[str, Any]: if not payload.ids: return {"updated": 0} placeholders = ",".join("?" for _ in payload.ids) with get_db() as conn: cursor = conn.execute( f""" UPDATE {table} SET confirm_status = ?, user_note = COALESCE(?, user_note), updated_at = ? WHERE id IN ({placeholders}) """, [payload.confirm_status, payload.user_note, now_iso(), *payload.ids], ) return {"updated": cursor.rowcount} def rows_for_prompt(table: str, item_type: str, payload: PromptRequest) -> list[dict[str, Any]]: with get_db() as conn: if payload.scope == "selected" and payload.ids: placeholders = ",".join("?" for _ in payload.ids) rows = conn.execute(f"SELECT * FROM {table} WHERE id IN ({placeholders})", payload.ids).fetchall() else: rows = conn.execute(f"SELECT * FROM {table} WHERE confirm_status = 'PENDING'").fetchall() rows = attach_item_metadata(conn, item_type, rows) return [normalize_prompt_row(item_type, row) for row in rows] def normalize_prompt_row(item_type: str, row: dict[str, Any]) -> dict[str, Any]: if item_type == "service": return { "type": "service", "id": row["id"], "name": row["name"], "display_name": row["display_name"], "status": row["status"], "start_type": row["start_type"], "username": row["username"], "binary_path": row["binary_path"], "description": row["description"], "is_present_now": bool(row["is_present_now"]), "tags": [ { "name": tag["name"], "description": tag["description"], "is_controllable": tag["is_controllable"], } for tag in row.get("tags", []) ], } return { "type": "process", "id": row["id"], "name": row["name"], "exe_path": row["exe_path"], "cmdline": row["cmdline"], "username": row["username"], "status": row["status"], "last_pid": row["last_pid"], "parent_pid": row["parent_pid"], "is_present_now": bool(row["is_present_now"]), "tags": [ { "name": tag["name"], "description": tag["description"], "is_controllable": tag["is_controllable"], } for tag in row.get("tags", []) ], } def markdown_table(rows: list[dict[str, Any]]) -> str: headers = ["type", "id", "name", "status", "tags", "path_or_command", "user", "present"] lines = ["| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"] for row in rows: path_or_command = row.get("binary_path") or row.get("exe_path") or row.get("cmdline") or "" values = [ str(row.get("type", "")), str(row.get("id", "")), str(row.get("name", "")), str(row.get("status", "")), ", ".join(tag.get("name", "") for tag in row.get("tags", [])).replace("|", "\\|"), str(path_or_command).replace("|", "\\|"), str(row.get("username", "")).replace("|", "\\|"), "yes" if row.get("is_present_now") else "no", ] lines.append("| " + " | ".join(values) + " |") return "\n".join(lines) def prompt_response(rows: list[dict[str, Any]]) -> dict[str, Any]: pending_json = json.dumps(rows, ensure_ascii=False, indent=2) table = markdown_table(rows) with get_db() as conn: tags_json = json.dumps(all_tags(conn), ensure_ascii=False, indent=2) prompt_text = AI_PROMPT_TEMPLATE.replace("{pending_items_json}", pending_json).replace("{tags_json}", tags_json) return {"prompt_text": prompt_text, "markdown_table": table, "items": rows} def set_item_tags(item_type: str, table: str, item_id: int, payload: TagAssignRequest) -> dict[str, Any]: get_item(table, item_id) unique_ids = sorted(set(payload.tag_ids)) now = now_iso() with get_db() as conn: if unique_ids: placeholders = ",".join("?" for _ in unique_ids) found = conn.execute(f"SELECT id FROM tags WHERE id IN ({placeholders})", unique_ids).fetchall() if len(found) != len(unique_ids): raise HTTPException(status_code=400, detail="One or more tag ids do not exist") conn.execute("DELETE FROM item_tags WHERE item_type = ? AND item_id = ?", (item_type, item_id)) for tag_id in unique_ids: conn.execute( "INSERT INTO item_tags (item_type, item_id, tag_id, created_at) VALUES (?, ?, ?, ?)", (item_type, item_id, tag_id, now), ) return get_item(table, item_id) def ensure_control_allowed(table: str, item_id: int) -> dict[str, Any]: item = get_item(table, item_id) if not item.get("can_control"): raise HTTPException(status_code=403, detail="This item is not controllable because it is unconfirmed or has a non-controllable tag") return item def normalize_import_tag_names(tag_names: list[str] | None) -> list[str]: if tag_names is None: return [] normalized = [] seen = set() for tag_name in tag_names: name = str(tag_name).strip()[:80] if not name or name in seen: continue seen.add(name) normalized.append(name) return normalized def ensure_tag_ids(conn, tag_names: list[str]) -> list[int]: tag_ids = [] now = now_iso() for tag_name in tag_names: row = conn.execute("SELECT id FROM tags WHERE name = ?", (tag_name,)).fetchone() if row: tag_ids.append(row["id"]) continue cursor = conn.execute( """ INSERT INTO tags (name, description, is_controllable, is_builtin, created_at, updated_at) VALUES (?, ?, 1, 0, ?, ?) """, (tag_name, "AI 自动新增标签", now, now), ) tag_ids.append(cursor.lastrowid) return tag_ids def replace_item_tags(conn, item_type: str, item_id: int, tag_ids: list[int]) -> None: now = now_iso() conn.execute("DELETE FROM item_tags WHERE item_type = ? AND item_id = ?", (item_type, item_id)) for tag_id in tag_ids: conn.execute( "INSERT INTO item_tags (item_type, item_id, tag_id, created_at) VALUES (?, ?, ?, ?)", (item_type, item_id, tag_id, now), ) def import_ai_results(table: str, item_type: str, payload: AiImportRequest) -> dict[str, Any]: updated = 0 with get_db() as conn: for item in payload.items: if item.type != item_type: continue matched_rows = conn.execute(f"SELECT id FROM {table} WHERE name = ?", (item.name,)).fetchall() tag_ids = ensure_tag_ids(conn, normalize_import_tag_names(item.tags)) if item.tags is not None else None for row in matched_rows: cursor = conn.execute( f""" UPDATE {table} SET confirm_status = ?, ai_description = ?, ai_reason = ?, ai_suggestion = ?, risk_level = ?, updated_at = ? WHERE id = ? """, ( item.judgement, item.description, item.reason, item.suggestion, item.risk_level, now_iso(), row["id"], ), ) if tag_ids is not None: replace_item_tags(conn, item_type, row["id"], tag_ids) updated += cursor.rowcount return {"updated": updated} def ai_update_preview(table: str, item_type: str, proposed_items: list[dict[str, Any]]) -> list[dict[str, Any]]: names = [item["name"] for item in proposed_items if item.get("type") == item_type and item.get("name")] if not names: return [] placeholders = ",".join("?" for _ in names) with get_db() as conn: rows = conn.execute( f""" SELECT id, name, confirm_status, ai_description, ai_reason, ai_suggestion, risk_level FROM {table} WHERE name IN ({placeholders}) """, names, ).fetchall() tag_map = tags_for_items(conn, item_type, [row["id"] for row in rows]) row_map = {} for row in rows: current = dict(row) current["tags"] = [tag["name"] for tag in tag_map.get(row["id"], [])] row_map[row["name"]] = current preview = [] for item in proposed_items: if item.get("type") != item_type: continue current = row_map.get(item.get("name")) preview.append( { "matched": current is not None, "current": current, "proposed": item, } ) return preview def analyze_items_with_ai(table: str, item_type: str, payload: AiAnalyzeRequest) -> dict[str, Any]: rows = rows_for_prompt(table, item_type, PromptRequest(scope=payload.scope, ids=payload.ids)) if not rows: raise HTTPException(status_code=400, detail="No items available for AI analysis") prompt_data = prompt_response(rows) result = ai_service.chat(payload.provider_id, payload.model_id, prompt_data["prompt_text"], payload.temperature) try: parsed_items = ai_service.parse_ai_items(result["content"]) except (json.JSONDecodeError, ValueError) as exc: raise HTTPException( status_code=502, detail=f"AI output is not valid import JSON: {exc}", ) from exc return { "items": parsed_items, "preview": ai_update_preview(table, item_type, parsed_items), "raw_output": result["content"], "provider": result["provider"], "model": result["model"], "prompt_text": prompt_data["prompt_text"], "markdown_table": prompt_data["markdown_table"], } @app.get("/api/dashboard") def dashboard() -> dict[str, Any]: with get_db() as conn: latest_scan = conn.execute("SELECT * FROM scan_records ORDER BY started_at DESC LIMIT 1").fetchone() service_total = conn.execute("SELECT COUNT(*) AS total FROM windows_services").fetchone()["total"] process_total = conn.execute("SELECT COUNT(*) AS total FROM windows_processes").fetchone()["total"] pending_services = conn.execute( "SELECT COUNT(*) AS total FROM windows_services WHERE confirm_status = 'PENDING'" ).fetchone()["total"] pending_processes = conn.execute( "SELECT COUNT(*) AS total FROM windows_processes WHERE confirm_status = 'PENDING'" ).fetchone()["total"] missing_services = conn.execute( "SELECT COUNT(*) AS total FROM windows_services WHERE is_present_now = 0" ).fetchone()["total"] missing_processes = conn.execute( "SELECT COUNT(*) AS total FROM windows_processes WHERE is_present_now = 0" ).fetchone()["total"] return { "latest_scan": latest_scan, "service_total": service_total, "process_total": process_total, "pending_services": pending_services, "pending_processes": pending_processes, "missing_services": missing_services, "missing_processes": missing_processes, } @app.get("/api/tags") def tags() -> dict[str, Any]: with get_db() as conn: rows = all_tags(conn) return {"items": rows} @app.post("/api/tags") def tag_create(payload: TagCreate) -> dict[str, Any]: now = now_iso() try: with get_db() as conn: cursor = conn.execute( """ INSERT INTO tags (name, description, is_controllable, is_builtin, created_at, updated_at) VALUES (?, ?, ?, 0, ?, ?) """, (payload.name.strip(), payload.description, 1 if payload.is_controllable else 0, now, now), ) tag_id = cursor.lastrowid row = conn.execute("SELECT * FROM tags WHERE id = ?", (tag_id,)).fetchone() except sqlite3.IntegrityError as exc: raise HTTPException(status_code=409, detail="Tag name already exists") from exc return bool_tag(row) @app.patch("/api/tags/{tag_id}") def tag_update(tag_id: int, payload: TagUpdate) -> dict[str, Any]: now = now_iso() try: with get_db() as conn: existing = conn.execute("SELECT * FROM tags WHERE id = ?", (tag_id,)).fetchone() if not existing: raise HTTPException(status_code=404, detail="Tag not found") conn.execute( """ UPDATE tags SET name = ?, description = ?, is_controllable = ?, updated_at = ? WHERE id = ? """, (payload.name.strip(), payload.description, 1 if payload.is_controllable else 0, now, tag_id), ) row = conn.execute("SELECT * FROM tags WHERE id = ?", (tag_id,)).fetchone() except sqlite3.IntegrityError as exc: raise HTTPException(status_code=409, detail="Tag name already exists") from exc return bool_tag(row) @app.delete("/api/tags/{tag_id}") def tag_delete(tag_id: int) -> dict[str, Any]: with get_db() as conn: row = conn.execute("SELECT * FROM tags WHERE id = ?", (tag_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="Tag not found") if row["is_builtin"]: raise HTTPException(status_code=400, detail="Built-in tags cannot be deleted") cursor = conn.execute("DELETE FROM tags WHERE id = ?", (tag_id,)) return {"deleted": cursor.rowcount} @app.get("/api/ai/providers") def ai_providers() -> dict[str, Any]: return {"items": ai_service.list_providers()} @app.post("/api/ai/providers") def ai_provider_create(payload: AiProviderCreate) -> dict[str, Any]: return ai_service.create_provider(payload) @app.patch("/api/ai/providers/{provider_id}") def ai_provider_update(provider_id: int, payload: AiProviderUpdate) -> dict[str, Any]: return ai_service.update_provider(provider_id, payload) @app.delete("/api/ai/providers/{provider_id}") def ai_provider_delete(provider_id: int) -> dict[str, Any]: return ai_service.delete_provider(provider_id) @app.get("/api/ai/models") def ai_models(provider_id: int | None = None) -> dict[str, Any]: return {"items": ai_service.list_models(provider_id)} @app.post("/api/ai/models") def ai_model_create(payload: AiModelCreate) -> dict[str, Any]: return ai_service.create_model(payload) @app.patch("/api/ai/models/{model_id}") def ai_model_update(model_id: int, payload: AiModelUpdate) -> dict[str, Any]: return ai_service.update_model(model_id, payload) @app.delete("/api/ai/models/{model_id}") def ai_model_delete(model_id: int) -> dict[str, Any]: return ai_service.delete_model(model_id) @app.post("/api/ai/test") def ai_test(payload: AiChatRequest) -> dict[str, Any]: return ai_service.chat(payload.provider_id, payload.model_id, payload.prompt, payload.temperature) @app.get("/api/settings") def system_settings() -> dict[str, Any]: return settings_service.list_settings() @app.put("/api/settings") def system_settings_update(payload: SystemSettingsUpdate) -> dict[str, Any]: return settings_service.update_settings(payload) @app.post("/api/automation/power/shutdown") def automation_shutdown(payload: AutomationPowerRequest) -> dict[str, Any]: return windows_automation.shutdown_windows(payload.delay_seconds, payload.force, payload.reason) @app.post("/api/automation/power/restart") def automation_restart(payload: AutomationPowerRequest) -> dict[str, Any]: return windows_automation.restart_windows(payload.delay_seconds, payload.force, payload.reason) @app.post("/api/automation/power/cancel") def automation_power_cancel() -> dict[str, Any]: return windows_automation.cancel_power_action() @app.post("/api/automation/programs/start") def automation_program_start(payload: AutomationProgramStartRequest) -> dict[str, Any]: return windows_automation.start_program(payload.command, payload.cwd, payload.shell) @app.post("/api/automation/programs/stop") def automation_program_stop(payload: AutomationProgramStopRequest) -> dict[str, Any]: return windows_automation.stop_program( pid=payload.pid, name=payload.name, timeout_seconds=payload.timeout_seconds, kill_after_timeout=payload.kill_after_timeout, ) @app.post("/api/automation/screenshot") def automation_screenshot(payload: AutomationScreenshotRequest) -> dict[str, Any]: return windows_automation.take_screenshot(payload.save_path, payload.include_base64) @app.post("/api/automation/mouse") def automation_mouse(payload: AutomationMouseRequest) -> dict[str, Any]: return windows_automation.mouse_action( action=payload.action, x=payload.x, y=payload.y, duration=payload.duration, button=payload.button, clicks=payload.clicks, amount=payload.amount, ) @app.post("/api/automation/keyboard") def automation_keyboard(payload: AutomationKeyboardRequest) -> dict[str, Any]: return windows_automation.keyboard_action( action=payload.action, key=payload.key, keys=payload.keys, text=payload.text, interval=payload.interval, ) @app.post("/api/automation/vision/analyze") def automation_vision_analyze(payload: AutomationVisionAnalyzeRequest) -> dict[str, Any]: return automation_service.analyze_screen(payload) @app.post("/api/automation/vision/screenshot") def automation_vision_screenshot(payload: AutomationScreenshotCaptureRequest) -> dict[str, Any]: return automation_service.capture_screenshot(payload) @app.post("/api/automation/screens/{screen_id}/elements/{element_id}/locate") def automation_element_locate(screen_id: int, element_id: int, payload: AutomationElementLocateRequest) -> dict[str, Any]: return automation_service.locate_element(screen_id, element_id, payload) @app.post("/api/automation/actions/mouse") def automation_action_mouse(payload: AutomationMouseActionRequest) -> dict[str, Any]: return automation_service.execute_mouse_action(payload) @app.post("/api/automation/actions/keyboard") def automation_action_keyboard(payload: AutomationKeyboardActionRequest) -> dict[str, Any]: return automation_service.execute_keyboard_action(payload) @app.post("/api/automation/actions/text-input") def automation_action_text_input(payload: AutomationTextInputRequest) -> dict[str, Any]: return automation_service.execute_text_input(payload) @app.post("/api/automation/actions/start-program") def automation_action_start_program(payload: AutomationStartProgramRequest) -> dict[str, Any]: return automation_service.execute_start_program(payload) @app.post("/api/automation/actions/close-opened-programs") def automation_action_close_opened_programs(payload: AutomationCloseProgramsRequest) -> dict[str, Any]: return automation_service.close_opened_programs(payload.pids) @app.get("/api/automation/workflows") def automation_workflows(page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200)) -> dict[str, Any]: return automation_service.list_workflows(page, page_size) @app.get("/api/automation/workflow-nodes") def automation_workflow_nodes() -> dict[str, Any]: return automation_service.list_workflow_node_definitions() @app.post("/api/automation/workflows/plan") def automation_workflow_plan(payload: AutomationWorkflowPlanRequest) -> dict[str, Any]: return automation_service.plan_workflow(payload) @app.post("/api/automation/workflows/plan/continue") def automation_workflow_plan_continue(payload: AutomationWorkflowPlanContinueRequest) -> dict[str, Any]: return automation_service.continue_workflow_plan(payload) @app.post("/api/automation/workflows") def automation_workflow_create(payload: AutomationWorkflowSaveRequest) -> dict[str, Any]: return automation_service.save_workflow(payload) @app.get("/api/automation/workflows/{workflow_id}") def automation_workflow_detail(workflow_id: int) -> dict[str, Any]: return automation_service.get_workflow(workflow_id) @app.get("/api/automation/workflows/by-key/{workflow_key}") def automation_workflow_detail_by_key(workflow_key: str) -> dict[str, Any]: return automation_service.get_workflow_by_key(workflow_key) @app.post("/api/automation/workflows/{workflow_id}/run") def automation_workflow_run(workflow_id: int, payload: AutomationWorkflowRunRequest) -> dict[str, Any]: return automation_service.run_workflow(workflow_id, payload) @app.post("/api/automation/workflows/by-key/{workflow_key}/run") def automation_workflow_run_by_key( workflow_key: str, payload: AutomationWorkflowRunRequest, x_automation_token: str | None = Header(default=None), ) -> dict[str, Any]: verify_automation_token(x_automation_token) return automation_service.run_workflow_by_key(workflow_key, payload) @app.put("/api/automation/workflows/{workflow_id}") def automation_workflow_update(workflow_id: int, payload: AutomationWorkflowSaveRequest) -> dict[str, Any]: return automation_service.update_workflow(workflow_id, payload) @app.delete("/api/automation/workflows/{workflow_id}") def automation_workflow_delete(workflow_id: int) -> dict[str, Any]: return automation_service.delete_workflow(workflow_id) @app.get("/api/automation/screens") def automation_screens(page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200)) -> dict[str, Any]: return automation_service.list_screens(page, page_size) @app.get("/api/automation/screens/{screen_id}") def automation_screen_detail(screen_id: int, include_image: bool = False) -> dict[str, Any]: return automation_service.get_screen(screen_id, include_image) @app.delete("/api/automation/screens/{screen_id}") def automation_screen_delete(screen_id: int) -> dict[str, Any]: return automation_service.delete_screen(screen_id) @app.get("/api/automation/errors") def automation_errors(page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200)) -> dict[str, Any]: return automation_service.list_errors(page, page_size) @app.get("/api/automation/errors/{error_id}") def automation_error_detail(error_id: int, include_images: bool = False) -> dict[str, Any]: return automation_service.get_error(error_id, include_images) @app.get("/api/sensors") def sensors() -> dict[str, Any]: return collect_sensors() @app.get("/api/smart/scan") def smart_scan() -> dict[str, Any]: return scan_devices() @app.get("/api/smart/devices") def smart_devices(include_jmb39x: bool = True, jmb39x_slots: int = Query(default=8, ge=0, le=16)) -> dict[str, Any]: return collect_all_smart(include_jmb39x=include_jmb39x, jmb39x_slots=jmb39x_slots) @app.get("/api/smart/device") def smart_device(device: str, device_type: str | None = None) -> dict[str, Any]: return get_device_smart(device, device_type) @app.post("/api/scans/run") def run_scan() -> dict[str, Any]: return run_full_scan() @app.get("/api/scans") def scan_history(page: int = 1, page_size: int = 20) -> dict[str, Any]: offset = (page - 1) * page_size with get_db() as conn: total = conn.execute("SELECT COUNT(*) AS total FROM scan_records").fetchone()["total"] rows = conn.execute( "SELECT * FROM scan_records ORDER BY started_at DESC LIMIT ? OFFSET ?", (page_size, offset), ).fetchall() return {"items": rows, "total": total, "page": page, "page_size": page_size} @app.get("/api/scans/{scan_id}") def scan_detail(scan_id: int) -> dict[str, Any]: with get_db() as conn: scan = conn.execute("SELECT * FROM scan_records WHERE id = ?", (scan_id,)).fetchone() if not scan: raise HTTPException(status_code=404, detail="Scan not found") return scan @app.get("/api/services") def services( keyword: str | None = None, confirm_status: str | None = None, present: bool | None = None, sort_by: str | None = None, sort_order: str | None = Query(default=None, pattern="^(asc|desc)$"), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200), ) -> dict[str, Any]: return list_items( "windows_services", keyword, confirm_status, present, page, page_size, ["name", "display_name", "binary_path", "description"], sort_by, sort_order, ) @app.patch("/api/services/batch") def service_batch_update(payload: BatchStatusUpdate) -> dict[str, Any]: return update_batch("windows_services", payload) @app.post("/api/services/import-ai") def service_import_ai(payload: AiImportRequest) -> dict[str, Any]: return import_ai_results("windows_services", "service", payload) @app.post("/api/services/analyze-ai") def service_analyze_ai(payload: AiAnalyzeRequest) -> dict[str, Any]: return analyze_items_with_ai("windows_services", "service", payload) @app.post("/api/services/ai-prompt") def service_ai_prompt(payload: PromptRequest) -> dict[str, Any]: return prompt_response(rows_for_prompt("windows_services", "service", payload)) @app.put("/api/services/{service_id}/tags") def service_tags_update(service_id: int, payload: TagAssignRequest) -> dict[str, Any]: return set_item_tags("service", "windows_services", service_id, payload) @app.post("/api/services/{service_id}/start") def service_start(service_id: int) -> dict[str, Any]: item = ensure_control_allowed("windows_services", service_id) if not item.get("is_present_now"): raise HTTPException(status_code=400, detail="This service was not present in the latest scan") result = start_service(item["name"]) with get_db() as conn: conn.execute( "UPDATE windows_services SET status = ?, updated_at = ? WHERE id = ?", (result.get("status") or "running", now_iso(), service_id), ) return result @app.post("/api/services/{service_id}/stop") def service_stop(service_id: int) -> dict[str, Any]: item = ensure_control_allowed("windows_services", service_id) if not item.get("is_present_now"): raise HTTPException(status_code=400, detail="This service was not present in the latest scan") result = stop_service(item["name"]) with get_db() as conn: conn.execute( "UPDATE windows_services SET status = ?, updated_at = ? WHERE id = ?", (result.get("status") or "stopped", now_iso(), service_id), ) return result @app.post("/api/services/{service_id}/restart") def service_restart(service_id: int) -> dict[str, Any]: item = ensure_control_allowed("windows_services", service_id) if not item.get("is_present_now"): raise HTTPException(status_code=400, detail="This service was not present in the latest scan") result = restart_service(item["name"]) with get_db() as conn: conn.execute( "UPDATE windows_services SET status = ?, updated_at = ? WHERE id = ?", (result.get("status") or "running", now_iso(), service_id), ) return result @app.get("/api/services/{service_id}") def service_detail(service_id: int) -> dict[str, Any]: return get_item("windows_services", service_id) @app.patch("/api/services/{service_id}") def service_update(service_id: int, payload: StatusUpdate) -> dict[str, Any]: return update_one("windows_services", service_id, payload) @app.get("/api/processes") def processes( keyword: str | None = None, confirm_status: str | None = None, present: bool | None = None, sort_by: str | None = None, sort_order: str | None = Query(default=None, pattern="^(asc|desc)$"), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200), ) -> dict[str, Any]: return list_items( "windows_processes", keyword, confirm_status, present, page, page_size, ["name", "exe_path", "cmdline", "username"], sort_by, sort_order, ) @app.patch("/api/processes/batch") def process_batch_update(payload: BatchStatusUpdate) -> dict[str, Any]: return update_batch("windows_processes", payload) @app.post("/api/processes/import-ai") def process_import_ai(payload: AiImportRequest) -> dict[str, Any]: return import_ai_results("windows_processes", "process", payload) @app.post("/api/processes/analyze-ai") def process_analyze_ai(payload: AiAnalyzeRequest) -> dict[str, Any]: return analyze_items_with_ai("windows_processes", "process", payload) @app.post("/api/processes/ai-prompt") def process_ai_prompt(payload: PromptRequest) -> dict[str, Any]: return prompt_response(rows_for_prompt("windows_processes", "process", payload)) @app.put("/api/processes/{process_id}/tags") def process_tags_update(process_id: int, payload: TagAssignRequest) -> dict[str, Any]: return set_item_tags("process", "windows_processes", process_id, payload) @app.post("/api/processes/{process_id}/start") def process_start(process_id: int) -> dict[str, Any]: item = ensure_control_allowed("windows_processes", process_id) result = start_process(item) with get_db() as conn: conn.execute( "UPDATE windows_processes SET last_pid = ?, is_present_now = 1, status = ?, updated_at = ? WHERE id = ?", (result.get("pid"), "running", now_iso(), process_id), ) return result @app.post("/api/processes/{process_id}/stop") def process_stop(process_id: int) -> dict[str, Any]: item = ensure_control_allowed("windows_processes", process_id) if not item.get("is_present_now"): raise HTTPException(status_code=400, detail="This process was not present in the latest scan") result = stop_process(item) with get_db() as conn: conn.execute( "UPDATE windows_processes SET is_present_now = 0, status = ?, updated_at = ? WHERE id = ?", ("stopped", now_iso(), process_id), ) return result @app.get("/api/processes/{process_id}") def process_detail(process_id: int) -> dict[str, Any]: return get_item("windows_processes", process_id) @app.patch("/api/processes/{process_id}") def process_update(process_id: int, payload: StatusUpdate) -> dict[str, Any]: return update_one("windows_processes", process_id, payload)