foodlinkk-command-center/browser-agent/app/extract.py

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