from __future__ import annotations import asyncio import base64 import json import re from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from typing import Any, Optional from bs4 import BeautifulSoup from fastapi import FastAPI, HTTPException, Query from fastapi.responses import Response from pydantic import BaseModel, Field from playwright.sync_api import sync_playwright from app.db import execute_returning, fetch_all, fetch_one, init_pool from app import pa_live app = FastAPI(title="Foodlinkk Browser Agent", version="1.0.0") _executor = ThreadPoolExecutor(max_workers=4) _active_session: dict[str, Any] = {"id": None, "status": "idle", "screenshot_b64": None, "url": None} class BrowseRequest(BaseModel): url: str task: Optional[str] = Field(default=None, description="Optional task description for logging") site_id: Optional[int] = None wait_seconds: float = Field(default=3.0, ge=0, le=30) instruction: Optional[str] = Field(default=None, description="Klik/scroll instructies in NL") pa_label: Optional[str] = None pa_job_id: Optional[str] = None pa_index: Optional[int] = None pa_total: Optional[int] = None pa_query: Optional[str] = None pa_chat_id: Optional[int] = None pa_user_name: Optional[str] = None def _serialize(row: dict | None) -> dict | None: if not row: return None out = dict(row) for k, v in list(out.items()): if hasattr(v, "isoformat"): out[k] = v.isoformat() return out def _accept_cookies(page) -> str | None: for sel in ( 'button:has-text("Accepteren")', 'button:has-text("Alles accepteren")', 'button:has-text("Akkoord")', 'button:has-text("Accept")', "#onetrust-accept-btn-handler", '[data-testid="accept-all"]', ): try: btn = page.locator(sel).first if btn.is_visible(timeout=1200): btn.click(timeout=3000) page.wait_for_timeout(500) return f"Clicked: {sel}" except Exception: pass return None def _run_instructions(page, instruction: str) -> list[str]: """Parse Dutch/English instructions into Playwright actions.""" log: list[str] = [] if not instruction or not instruction.strip(): return log lower = instruction.lower() if any(w in lower for w in ("cookie", "accepteren", "privacy", "akkoord", "consent")): r = _accept_cookies(page) if r: log.append(r) for m in re.finditer(r"klik(?:ken)?(?:\s+op)?\s+['\"]?([^'\".\n]+)", instruction, re.I): label = m.group(1).strip() if len(label) < 2: continue try: page.get_by_role("button", name=re.compile(re.escape(label[:40]), re.I)).first.click(timeout=4000) log.append(f"Klik: button '{label[:40]}'") page.wait_for_timeout(800) continue except Exception: pass try: page.get_by_text(label[:60], exact=False).first.click(timeout=4000) log.append(f"Klik: tekst '{label[:40]}'") page.wait_for_timeout(800) except Exception as exc: log.append(f"Mislukt klik '{label[:30]}': {exc}") scroll_n = lower.count("scroll") + lower.count("naar beneden") + lower.count("verder") scroll_n = max(scroll_n, 1 if ("meer" in lower and "product" in lower) else 0) for i in range(min(scroll_n, 5)): page.evaluate("window.scrollBy(0, Math.min(window.innerHeight, 700))") page.wait_for_timeout(400) log.append(f"Scroll {i + 1}") if "filter" in lower or "categorie" in lower or "subcategorie" in lower: for word in re.findall(r"[a-zA-Z]{4,}", instruction): if word.lower() in ("filter", "categorie", "subcategorie", "klik", "scroll"): continue try: page.get_by_text(word, exact=False).first.click(timeout=2000) log.append(f"Filter/klik: {word}") page.wait_for_timeout(600) break except Exception: pass page.wait_for_timeout(1000) return log def _ocr_bytes(image_bytes: bytes) -> str: try: import io import pytesseract from PIL import Image img = Image.open(io.BytesIO(image_bytes)) return pytesseract.image_to_string(img, lang="nld+eng").strip() except Exception as exc: return f"OCR niet beschikbaar: {exc}" def _extract_links(html: str, base_url: str) -> list[dict[str, str]]: soup = BeautifulSoup(html, "html.parser") links: list[dict[str, str]] = [] seen: set[str] = set() for a in soup.find_all("a", href=True): href = a["href"].strip() text = (a.get_text() or "").strip()[:120] if not href or href.startswith("#") or href in seen: continue seen.add(href) links.append({"href": href, "text": text}) if len(links) >= 40: break return links def _browse_sync(url: str, task: Optional[str], site_id: Optional[int], wait_seconds: float, instruction: Optional[str] = None) -> dict: global _active_session session_row = execute_returning( """ INSERT INTO browser_sessions (url, task, status, site_id) VALUES (%s, %s, 'running', %s) RETURNING id """, (url, task or instruction, site_id), ) session_id = session_row["id"] _active_session = {"id": session_id, "status": "running", "screenshot_b64": None, "url": url} try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( viewport={"width": 1440, "height": 900}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 FoodlinkkBot/2.0", locale="nl-NL", ) page = context.new_page() page.goto(url, wait_until="domcontentloaded", timeout=90000) page.wait_for_timeout(int(wait_seconds * 1000)) try: page.wait_for_load_state("networkidle", timeout=15000) except Exception: pass steps_log: list[str] = [] cr = _accept_cookies(page) if cr: steps_log.append(cr) if instruction: steps_log.extend(_run_instructions(page, instruction)) elif task: steps_log.extend(_run_instructions(page, task)) shot = page.screenshot(type="jpeg", quality=72, full_page=False) b64 = base64.b64encode(shot).decode("ascii") _active_session["screenshot_b64"] = b64 title = page.title() final_url = page.url html = page.content() soup = BeautifulSoup(html, "html.parser") for tag in soup(["script", "style", "noscript"]): tag.decompose() text = re.sub(r"\s+", " ", soup.get_text(" ", strip=True))[:80000] links = _extract_links(html, final_url) row = execute_returning( """ UPDATE browser_sessions SET final_url=%s, title=%s, status='completed', content_text=%s, content_html=%s, screenshot_b64=%s, links=%s::jsonb, completed_at=NOW(), metadata=%s::jsonb WHERE id=%s RETURNING * """, ( final_url, title, text, html[:200000], b64, json.dumps(links), json.dumps({"task": task or instruction or "browse", "link_count": len(links), "steps": steps_log}), session_id, ), ) execute_returning( """ INSERT INTO crawled_pages (url, final_url, title, content, content_html, screenshot_b64, links, site_id, metadata, crawled_at) VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s,%s::jsonb,NOW()) ON CONFLICT (url) DO UPDATE SET final_url=EXCLUDED.final_url, title=EXCLUDED.title, content=EXCLUDED.content, content_html=EXCLUDED.content_html, screenshot_b64=EXCLUDED.screenshot_b64, links=EXCLUDED.links, site_id=EXCLUDED.site_id, metadata=EXCLUDED.metadata, crawled_at=NOW() RETURNING id """, ( url, final_url, title, text[:50000], html[:100000], b64, json.dumps(links), site_id, json.dumps({"session_id": session_id, "source": "browser-agent"}), ), ) browser.close() _active_session["status"] = "completed" out = _serialize(row) out["content_preview"] = text[:2000] out["screenshot_b64"] = b64 return out except Exception as exc: execute_returning( """ UPDATE browser_sessions SET status='failed', error_message=%s, completed_at=NOW() WHERE id=%s RETURNING id """, (str(exc)[:500], session_id), ) _active_session["status"] = "failed" raise @app.on_event("startup") def startup() -> None: init_pool() @app.get("/health") def health() -> dict[str, str]: return {"status": "ok", "service": "browser-agent"} @app.get("/live") def live_view() -> dict[str, Any]: return { "active": _active_session, "novnc_url": "http://10.4.7.18:6080/vnc.html?autoconnect=true&resize=scale&password=Foodlinkk2026&path=websockify", "gradio_url": "http://10.4.7.18:7788", } @app.get("/sessions") def list_sessions(limit: int = Query(default=20, ge=1, le=100)) -> dict[str, Any]: rows = fetch_all( """ SELECT id, url, final_url, title, task, status, created_at, completed_at, LEFT(content_text, 400) AS content_preview, (screenshot_b64 IS NOT NULL) AS has_screenshot, site_id FROM browser_sessions ORDER BY created_at DESC LIMIT %s """, (limit,), ) return {"sessions": [_serialize(r) for r in rows]} @app.get("/sessions/{session_id}") def get_session(session_id: int, include_screenshot: bool = Query(default=True)) -> dict[str, Any]: row = fetch_one("SELECT * FROM browser_sessions WHERE id = %s", (session_id,)) if not row: raise HTTPException(status_code=404, detail="Session not found") out = _serialize(row) if not include_screenshot: out.pop("screenshot_b64", None) out.pop("content_html", None) else: out["content_preview"] = (out.get("content_text") or "")[:3000] return {"session": out} @app.get("/sessions/{session_id}/screenshot.jpg") def session_screenshot_jpg(session_id: int) -> Response: row = fetch_one("SELECT screenshot_b64 FROM browser_sessions WHERE id = %s", (session_id,)) if not row or not row.get("screenshot_b64"): raise HTTPException(status_code=404, detail="No screenshot") data = base64.b64decode(row["screenshot_b64"]) return Response(content=data, media_type="image/jpeg") @app.get("/live/screenshot.jpg") def live_screenshot_jpg() -> Response: b64 = _active_session.get("screenshot_b64") if not b64: row = fetch_one( "SELECT screenshot_b64 FROM browser_sessions WHERE screenshot_b64 IS NOT NULL ORDER BY id DESC LIMIT 1" ) b64 = row.get("screenshot_b64") if row else None if not b64: raise HTTPException(status_code=404, detail="No active screenshot") return Response(content=base64.b64decode(b64), media_type="image/jpeg") @app.post("/browse") async def browse(req: BrowseRequest) -> dict[str, Any]: url = req.url.strip() if not url.startswith(("http://", "https://")): raise HTTPException(status_code=400, detail="URL must start with http:// or https://") loop = asyncio.get_event_loop() try: result = await loop.run_in_executor( _executor, _browse_sync, url, req.task, req.site_id, req.wait_seconds, req.instruction, ) except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc return {"ok": True, "session": result} class TaskRequest(BaseModel): url: str task: str = "open and summarize visible content" metadata: dict[str, Any] = Field(default_factory=dict) @app.post("/task") async def browser_task(req: TaskRequest) -> dict[str, Any]: """Compatibility endpoint for tools-api.""" browse_req = BrowseRequest(url=req.url, task=req.task, wait_seconds=4.0) return await browse(browse_req) class InstructRequest(BaseModel): url: str instruction: str wait_seconds: float = 5.0 @app.post("/browse/instruct") async def browse_instruct(req: InstructRequest) -> dict[str, Any]: browse_req = BrowseRequest(url=req.url, task=req.instruction, instruction=req.instruction, wait_seconds=req.wait_seconds) return await browse(browse_req) @app.get("/sessions/{session_id}/ocr") def session_ocr(session_id: int) -> dict[str, Any]: row = fetch_one("SELECT screenshot_b64, title, url FROM browser_sessions WHERE id = %s", (session_id,)) if not row or not row.get("screenshot_b64"): raise HTTPException(status_code=404, detail="Geen screenshot voor OCR") text = _ocr_bytes(base64.b64decode(row["screenshot_b64"])) return {"ok": True, "session_id": session_id, "ocr_text": text, "title": row.get("title"), "url": row.get("url")} @app.post("/ocr") def ocr_upload(body: dict[str, Any]) -> dict[str, Any]: b64 = body.get("image_b64") or "" if not b64: raise HTTPException(status_code=400, detail="image_b64 required") text = _ocr_bytes(base64.b64decode(b64)) return {"ok": True, "ocr_text": text} # --- VNC + full extract + photo analysis (added by patch) --- from app import extract as extract_mod def _browser_helpers() -> dict: return { "execute_returning": execute_returning, "serialize": _serialize, "accept_cookies": _accept_cookies, "run_instructions": _run_instructions, "extract_links": _extract_links, } class VncNavigateBody(BaseModel): url: str instruction: Optional[str] = None wait_seconds: float = 4.0 class ExtractFullBody(BaseModel): url: str instruction: Optional[str] = None wait_seconds: float = 4.0 scroll_pages: int = Field(default=6, ge=1, le=12) site_id: Optional[int] = None also_vnc: bool = False pa_label: Optional[str] = None pa_job_id: Optional[str] = None pa_index: Optional[int] = None pa_total: Optional[int] = None pa_query: Optional[str] = None pa_chat_id: Optional[int] = None pa_user_name: Optional[str] = None class PhotoAnalyzeBody(BaseModel): image_b64: str source: str = "upload" filename: Optional[str] = None storage_path: Optional[str] = None session_id: Optional[int] = None @app.post("/vnc/navigate") async def vnc_navigate(req: VncNavigateBody) -> dict[str, Any]: url = req.url.strip() if not url.startswith(("http://", "https://")): raise HTTPException(status_code=400, detail="URL must start with http:// or https://") loop = asyncio.get_event_loop() try: result = await loop.run_in_executor( _executor, extract_mod.vnc_navigate_sync, url, req.instruction, req.wait_seconds, _browser_helpers(), ) except Exception as exc: raise HTTPException(status_code=502, detail=f"VNC navigate failed: {exc}") from exc return result @app.post("/browse/extract-full") async def browse_extract_full(req: ExtractFullBody) -> dict[str, Any]: url = req.url.strip() if not url.startswith(("http://", "https://")): raise HTTPException(status_code=400, detail="URL must start with http:// or https://") if req.pa_label and req.pa_job_id: if req.pa_index == 1 or not pa_live.get_live().get("job_id"): pa_live.start_job( req.pa_job_id, req.pa_query or "", chat_id=req.pa_chat_id, user_name=req.pa_user_name or "", ) pa_live.update_slot( req.pa_label, status="loading", url=url, index=req.pa_index, total=req.pa_total, ) loop = asyncio.get_event_loop() try: result = await loop.run_in_executor( _executor, extract_mod.extract_full_sync, url, req.instruction, req.wait_seconds, req.scroll_pages, req.site_id, _browser_helpers(), ) if req.pa_label: pa_live.update_slot( req.pa_label, status="completed", url=result.get("final_url") or url, title=result.get("title"), screenshot_b64=result.get("screenshot_b64"), index=req.pa_index, total=req.pa_total, ) if req.also_vnc: try: await loop.run_in_executor( _executor, extract_mod.vnc_navigate_sync, url, req.instruction, req.wait_seconds, _browser_helpers(), ) except Exception: pass except Exception as exc: if req.pa_label: pa_live.update_slot(req.pa_label, status="failed", url=url, error=str(exc)[:300]) raise HTTPException(status_code=502, detail=str(exc)) from exc global _active_session _active_session = { "id": result.get("id"), "status": "completed", "screenshot_b64": result.get("screenshot_b64"), "url": url, } return {"ok": True, "session": result} @app.post("/photos/analyze") async def photos_analyze(req: PhotoAnalyzeBody) -> dict[str, Any]: if not req.image_b64.strip(): raise HTTPException(status_code=400, detail="image_b64 required") loop = asyncio.get_event_loop() try: row = await loop.run_in_executor( _executor, extract_mod.analyze_photo_sync, req.image_b64, req.source, req.filename, req.storage_path, req.session_id, execute_returning, ) except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc return {"ok": True, "photo": row} @app.get("/photos") def photos_list(limit: int = Query(default=30, ge=1, le=100)) -> dict[str, Any]: rows = fetch_all( """ SELECT id, source, filename, storage_path, LEFT(ocr_text, 400) AS ocr_preview, jsonb_array_length(COALESCE(detections, '[]'::jsonb)) AS box_count, jsonb_array_length(COALESCE(extracted_items, '[]'::jsonb)) AS item_count, session_id, created_at FROM photo_imports ORDER BY created_at DESC LIMIT %s """, (limit,), ) return {"photos": [_serialize(r) for r in rows]} @app.get("/photos/{photo_id}") def photos_get(photo_id: int) -> dict[str, Any]: row = fetch_one("SELECT * FROM photo_imports WHERE id = %s", (photo_id,)) if not row: raise HTTPException(status_code=404, detail="Photo not found") out = _serialize(row) out.pop("image_b64", None) return {"photo": out} @app.get("/photos/{photo_id}/image.jpg") def photos_image(photo_id: int) -> Response: row = fetch_one("SELECT image_b64 FROM photo_imports WHERE id = %s", (photo_id,)) if not row or not row.get("image_b64"): raise HTTPException(status_code=404, detail="No image") return Response(content=base64.b64decode(row["image_b64"]), media_type="image/jpeg") @app.get("/photos/{photo_id}/detections") def photos_detections(photo_id: int) -> dict[str, Any]: row = fetch_one( "SELECT id, detections, extracted_items, ocr_text FROM photo_imports WHERE id = %s", (photo_id,), ) if not row: raise HTTPException(status_code=404, detail="Photo not found") return { "ok": True, "photo_id": photo_id, "detections": row.get("detections") or [], "extracted_items": row.get("extracted_items") or [], "ocr_text": row.get("ocr_text") or "", } class PaJobStartBody(BaseModel): job_id: str query: str chat_id: Optional[int] = None user_name: Optional[str] = None sites: Optional[list[str]] = None @app.post("/pa/job/start") def pa_job_start(body: PaJobStartBody) -> dict[str, Any]: pa_live.start_job( body.job_id, body.query, chat_id=body.chat_id, user_name=body.user_name or "", sites=body.sites, ) return {"ok": True, "live": pa_live.get_live()} @app.post("/pa/job/comparing") def pa_job_comparing() -> dict[str, Any]: pa_live.set_comparing() return {"ok": True, "live": pa_live.get_live()} @app.post("/pa/job/done") def pa_job_done() -> dict[str, Any]: pa_live.finish_job() return {"ok": True, "live": pa_live.get_live()} @app.get("/pa/live") def pa_live_view() -> dict[str, Any]: return pa_live.get_live() @app.get("/pa/live/{label}/screenshot.jpg") def pa_slot_screenshot(label: str) -> Response: data = pa_live.get_slot_screenshot(label) if not data: raise HTTPException(status_code=404, detail="No screenshot for slot") return Response(content=data, media_type="image/jpeg") @app.get("/vnc/screenshot.jpg") async def vnc_screenshot_jpg() -> Response: loop = asyncio.get_event_loop() try: data = await loop.run_in_executor(_executor, extract_mod.vnc_screenshot_sync) except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc return Response(content=data, media_type="image/jpeg")