import hmac import time import uuid from typing import Any, Optional from fastapi import Depends, FastAPI, Header, HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from .config import SETTINGS from .models import ExecRequest, InteractRequest, LaunchRequest, SeeRequest, SeeZoomRequest, WindowActionRequest, WindowQuery from .services import ( apply_window_action, capture_region_image, capture_screen, draw_grid, encode_image, exec_action, exec_command as run_exec_command, get_displays, launch_app, list_windows, ) app = FastAPI(title="clickthrough", version="0.1.0") def _now_ms() -> int: return int(time.time() * 1000) def _request_id() -> str: return str(uuid.uuid4()) def _ok(data: Any, status_code: int = 200): return JSONResponse( status_code=status_code, content={ "ok": True, "request_id": _request_id(), "time_ms": _now_ms(), "data": data, "error": None, }, ) def _err(code: str, message: str, status_code: int, details: Any = None): return JSONResponse( status_code=status_code, content={ "ok": False, "request_id": _request_id(), "time_ms": _now_ms(), "data": None, "error": {"code": code, "message": message, "details": details}, }, ) @app.exception_handler(HTTPException) async def _http_exception_handler(_: Request, exc: HTTPException): detail = exc.detail if isinstance(detail, dict): message = str(detail.get("message", "request failed")) return _err("http_error", message, exc.status_code, detail) return _err("http_error", str(detail), exc.status_code) @app.exception_handler(RequestValidationError) async def _validation_exception_handler(_: Request, exc: RequestValidationError): return _err("validation_error", "request validation failed", 422, exc.errors()) @app.exception_handler(Exception) async def _unhandled_exception_handler(_: Request, exc: Exception): return _err("internal_error", "internal server error", 500, {"type": type(exc).__name__}) def _auth(x_clickthrough_token: Optional[str] = Header(default=None)): token = SETTINGS["token"] if token and x_clickthrough_token != token: raise HTTPException(status_code=401, detail="invalid token") @app.post("/see") def see(req: SeeRequest, _: None = Depends(_auth)): image, region, mon, displays, screen_selection = capture_region_image( req.screen, req.region_x, req.region_y, req.region_width, req.region_height, ) out_img = image meta = {"region": region, "screen": screen_selection, "display": mon, "displays": displays} if req.with_grid: out_img, grid_meta = draw_grid(image, region["x"], region["y"], req.grid_rows, req.grid_cols, req.include_labels) meta.update(grid_meta) return _ok( { "image": { "format": req.image_format, "base64": encode_image(out_img, req.image_format, req.jpeg_quality), "width": out_img.size[0], "height": out_img.size[1], }, "meta": meta, } ) @app.post("/see/zoom") def see_zoom(req: SeeZoomRequest, _: None = Depends(_auth)): base_img, mon, displays, screen_selection = capture_screen(req.screen) cx = req.center_x - mon["x"] cy = req.center_y - mon["y"] left = max(0, cx - (req.width // 2)) top = max(0, cy - (req.height // 2)) right = min(base_img.size[0], left + req.width) bottom = min(base_img.size[1], top + req.height) crop = base_img.crop((left, top, right, bottom)) region_x = mon["x"] + left region_y = mon["y"] + top meta = { "region": {"x": region_x, "y": region_y, "width": crop.size[0], "height": crop.size[1]}, "screen": screen_selection, "display": mon, "displays": displays, } out_img = crop if req.with_grid: out_img, grid_meta = draw_grid(crop, region_x, region_y, req.grid_rows, req.grid_cols, req.include_labels) meta.update(grid_meta) return _ok( { "image": { "format": req.image_format, "base64": encode_image(out_img, req.image_format, req.jpeg_quality), "width": out_img.size[0], "height": out_img.size[1], }, "meta": meta, } ) @app.post("/interact") def interact(req: InteractRequest, _: None = Depends(_auth)): return _ok(exec_action(req.action, req.screen)) @app.get("/health") def health(_: None = Depends(_auth)): return _ok( { "service": "clickthrough", "version": app.version, "dry_run": SETTINGS["dry_run"], "allowed_region": SETTINGS["allowed_region"], "exec": { "enabled": SETTINGS["exec_enabled"], "secret_configured": bool(SETTINGS["exec_secret"]), "default_shell": SETTINGS["exec_default_shell"], "default_timeout_s": SETTINGS["exec_default_timeout_s"], "max_timeout_s": SETTINGS["exec_max_timeout_s"], }, } ) @app.get("/displays") def displays(_: None = Depends(_auth)): return _ok({"displays": get_displays(), "default_screen": 0}) @app.post("/exec") def exec_command(req: ExecRequest, x_clickthrough_exec_secret: Optional[str] = Header(default=None), _: None = Depends(_auth)): expected = SETTINGS["exec_secret"] if not expected: raise HTTPException(status_code=403, detail="exec secret not configured") if not x_clickthrough_exec_secret or not hmac.compare_digest(x_clickthrough_exec_secret, expected): raise HTTPException(status_code=401, detail="invalid exec secret") return _ok(run_exec_command(req)) @app.get("/windows") def windows( title_contains: str | None = None, title_regex: str | None = None, process_name: str | None = None, hwnd: int | None = None, visible_only: bool = True, _: None = Depends(_auth), ): query = WindowQuery( title_contains=title_contains, title_regex=title_regex, process_name=process_name, hwnd=hwnd, visible_only=visible_only, ) matches = list_windows(query) return _ok({"windows": matches, "count": len(matches)}) @app.post("/windows/action") def window_action(req: WindowActionRequest, _: None = Depends(_auth)): return _ok(apply_window_action(req)) @app.post("/launch") def launch(req: LaunchRequest, _: None = Depends(_auth)): return _ok(launch_app(req)) if __name__ == "__main__": import uvicorn uvicorn.run("server.app:app", host=SETTINGS["host"], port=SETTINGS["port"], reload=False)