from __future__ import annotations import base64 import json import os import re from typing import Any from playwright.sync_api import sync_playwright VNC_CDP_URL = os.getenv("VNC_CDP_URL", "http://10.4.7.18:9222") WEIGHT_RE = re.compile( r"(\d+[.,]?\d*\s*(?:g|gr|gram|kg|ml|cl|l|liter|st|stuks?|x\d+|pack|zak|doos))", re.I, ) PRICE_RE = re.compile(r"(€\s?\d+[.,]\d{2}|EUR\s?\d+[.,]\d{2})", re.I) def extract_items_from_text(text: str, source: str = "ocr") -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] seen: set[str] = set() for line in re.split(r"[\n\r]+", text): line = re.sub(r"\s+", " ", line).strip() if len(line) < 8: continue weight = WEIGHT_RE.search(line) price = PRICE_RE.search(line) if not weight and not price: continue key = line[:80].lower() if key in seen: continue seen.add(key) items.append({ "raw": line[:300], "weight": weight.group(1) if weight else None, "price": price.group(1) if price else None, "source": source, }) return items[:200] def ocr_with_boxes(image_bytes: bytes) -> tuple[str, list[dict[str, Any]]]: try: import io import pytesseract from PIL import Image img = Image.open(io.BytesIO(image_bytes)) text = pytesseract.image_to_string(img, lang="nld+eng").strip() data = pytesseract.image_to_data(img, lang="nld+eng", output_type=pytesseract.Output.DICT) boxes: list[dict[str, Any]] = [] for i, word in enumerate(data.get("text", [])): word = (word or "").strip() conf_raw = data["conf"][i] conf = int(conf_raw) if str(conf_raw).isdigit() else -1 if not word or conf < 40: continue boxes.append({ "text": word, "confidence": conf, "x": data["left"][i], "y": data["top"][i], "w": data["width"][i], "h": data["height"][i], }) return text, boxes[:500] except Exception as exc: return f"OCR niet beschikbaar: {exc}", [] def vnc_screenshot_sync() -> bytes: with sync_playwright() as p: browser = p.chromium.connect_over_cdp(VNC_CDP_URL) context = browser.contexts[0] if browser.contexts else browser.new_context() page = context.pages[0] if context.pages else context.new_page() return page.screenshot(type="jpeg", quality=72, full_page=False) def vnc_navigate_sync(url: str, instruction: str | None, wait_seconds: float, helpers: dict) -> dict[str, Any]: accept_cookies = helpers["accept_cookies"] run_instructions = helpers["run_instructions"] log: list[str] = [] with sync_playwright() as p: browser = p.chromium.connect_over_cdp(VNC_CDP_URL) context = browser.contexts[0] if browser.contexts else browser.new_context() page = context.pages[0] if context.pages else context.new_page() page.goto(url, wait_until="domcontentloaded", timeout=90000) page.wait_for_timeout(int(wait_seconds * 1000)) cr = accept_cookies(page) if cr: log.append(cr) if instruction: log.extend(run_instructions(page, instruction)) shot = page.screenshot(type="jpeg", quality=72, full_page=False) title = page.title() final_url = page.url return { "ok": True, "url": url, "final_url": final_url, "title": title, "steps": log, "screenshot_b64": base64.b64encode(shot).decode("ascii"), } def extract_full_sync( url: str, instruction: str | None, wait_seconds: float, scroll_pages: int, site_id: int | None, helpers: dict, ) -> dict: from bs4 import BeautifulSoup execute_returning = helpers["execute_returning"] serialize = helpers["serialize"] accept_cookies = helpers["accept_cookies"] run_instructions = helpers["run_instructions"] extract_links = helpers["extract_links"] session_row = execute_returning( "INSERT INTO browser_sessions (url, task, status, site_id) VALUES (%s, %s, 'running', %s) RETURNING id", (url, instruction or "full-extract", site_id), ) session_id = session_row["id"] all_items: list[dict] = [] all_ocr: list[str] = [] steps_log: list[str] = [] 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 FoodlinkkBot/2.0 FullExtract", locale="nl-NL", ) page = context.new_page() page.goto(url, wait_until="domcontentloaded", timeout=90000) page.wait_for_timeout(int(wait_seconds * 1000)) cr = accept_cookies(page) if cr: steps_log.append(cr) if instruction: steps_log.extend(run_instructions(page, instruction)) page.evaluate("window.scrollTo(0, 0)") page.wait_for_timeout(400) for i in range(max(1, min(scroll_pages, 12))): shot = page.screenshot(type="jpeg", quality=70, full_page=False) text, _ = ocr_with_boxes(shot) all_ocr.append(text) all_items.extend(extract_items_from_text(text, f"ocr-scroll-{i + 1}")) page.evaluate("window.scrollBy(0, Math.min(window.innerHeight * 0.85, 800))") page.wait_for_timeout(500) steps_log.append(f"Scroll+OCR {i + 1}") full_shot = page.screenshot(type="jpeg", quality=65, full_page=True) full_b64 = base64.b64encode(full_shot).decode("ascii") full_text, full_boxes = ocr_with_boxes(full_shot) all_items.extend(extract_items_from_text(full_text, "ocr-fullpage")) 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)) all_items.extend(extract_items_from_text(text, "html-text")) links = extract_links(html, final_url) seen: set[str] = set() deduped: list[dict] = [] for it in all_items: k = (it.get("raw") or "")[:100] if k in seen: continue seen.add(k) deduped.append(it) meta = { "task": "full-extract", "steps": steps_log, "scroll_pages": scroll_pages, "extracted_items": deduped[:300], "ocr_combined": "\n\n".join(all_ocr)[:50000], "detection_boxes": full_boxes[:200], } 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[:120000], html[:250000], full_b64, json.dumps(links), json.dumps(meta), session_id, ), ) browser.close() out = serialize(row) out["extracted_items"] = deduped[:300] out["detection_boxes"] = meta["detection_boxes"] out["content_preview"] = text[:2000] 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), ) raise def analyze_photo_sync( image_b64: str, source: str, filename: str | None, storage_path: str | None, session_id: int | None, execute_returning, ) -> dict: image_bytes = base64.b64decode(image_b64) text, boxes = ocr_with_boxes(image_bytes) items = extract_items_from_text(text, "photo-ocr") row = execute_returning( """ INSERT INTO photo_imports (source, filename, storage_path, image_b64, ocr_text, detections, extracted_items, session_id, metadata) VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s, %s::jsonb) RETURNING * """, ( source, filename, storage_path, image_b64, text, json.dumps(boxes), json.dumps(items), session_id, json.dumps({"box_count": len(boxes), "item_count": len(items)}), ), ) out = dict(row) for k, v in list(out.items()): if hasattr(v, "isoformat"): out[k] = v.isoformat() out.pop("image_b64", None) return out