Initial Context Extractor: extension + Playwright/Camoufox package
CI / Lint + tests (push) Has been cancelled

Ship a shared markdown/prompt core used by a Brave/Chrome MV3 extension and a Python automation API, with pytest coverage, CI, and packaging smoke.
This commit is contained in:
2026-07-15 14:52:22 -04:00
commit c91cd221d1
30 changed files with 2611 additions and 0 deletions
+12
View File
@@ -0,0 +1,12 @@
# Automation package
Install from this directory:
```bash
python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
playwright install chromium
pytest
```
Project docs and consumer Cursor prompt live in the repo root (`../README.md`, `../CURSOR_PROMPT.md`).
+4
View File
@@ -0,0 +1,4 @@
from .session import AsyncExtractorSession, ExtractorSession
__all__ = ["AsyncExtractorSession", "ExtractorSession"]
__version__ = "1.3.0"
+74
View File
@@ -0,0 +1,74 @@
"""Headless CLI: fetch a URL, capture console/network/errors, print an AI-ready prompt.
Usage:
context-extractor https://example.com
context-extractor https://example.com --selector "#main" --engine camoufox
context-extractor https://example.com --wait 2000 --out prompt.md
"""
from __future__ import annotations
import argparse
import sys
from .session import ExtractorSession
def main() -> int:
parser = argparse.ArgumentParser(prog="context-extractor", description=__doc__)
parser.add_argument("url", help="URL to load")
parser.add_argument("--selector", default=None, help="CSS selector to extract (default: body)")
parser.add_argument("--wait", type=int, default=0, help="Extra milliseconds to wait after load")
parser.add_argument(
"--max-chars", type=int, default=None,
help="Cap page content in the prompt to this many chars (default: 20000). "
"SPA pages can extract hundreds of thousands of chars; this keeps output LLM-sized.",
)
parser.add_argument(
"--engine", choices=["chromium", "firefox", "webkit", "camoufox"], default="chromium",
help="Which browser engine to drive (default: chromium)",
)
parser.add_argument("--headed", action="store_true", help="Show the browser window")
parser.add_argument("--out", default=None, help="Write the prompt to a file instead of stdout")
args = parser.parse_args()
if args.engine == "camoufox":
try:
from camoufox import Camoufox
except ImportError:
print(
"camoufox is not installed. Install with: pip install 'context-extractor[camoufox]'",
file=sys.stderr,
)
return 1
with Camoufox(headless=not args.headed) as browser:
page = browser.new_page()
prompt = _run(page, args)
else:
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = getattr(p, args.engine).launch(headless=not args.headed)
page = browser.new_page()
prompt = _run(page, args)
browser.close()
if args.out:
with open(args.out, "w", encoding="utf-8") as f:
f.write(prompt)
print(f"Wrote {len(prompt):,} chars to {args.out}", file=sys.stderr)
else:
print(prompt)
return 0
def _run(page, args) -> str:
session = ExtractorSession(page)
page.goto(args.url, wait_until="domcontentloaded")
if args.wait:
page.wait_for_timeout(args.wait)
return session.build_ai_prompt(args.selector, max_chars=args.max_chars)
if __name__ == "__main__":
raise SystemExit(main())
+1
View File
@@ -0,0 +1 @@
../../../extension/core/dom.js
+1
View File
@@ -0,0 +1 @@
../../../extension/core/prompt.js
+243
View File
@@ -0,0 +1,243 @@
"""Console/network/error capture + markdown extraction for a Playwright page.
Works with:
- Plain Playwright (any browser: chromium, firefox, webkit)
- Camoufox (daijro/camoufox), sync or async — Camoufox hands you a normal
Playwright Page/BrowserContext, so everything here works unmodified.
Design notes (see project README for the full rationale):
- Console/network/page-error capture uses Playwright's *native* event hooks
(page.on("console"/"request"/"requestfinished"/"requestfailed"/"pageerror")).
This deliberately avoids injecting a JS patcher into the page (the trick
the browser extension has to use), because:
1. It's more robust — no reliance on page.add_init_script(), which is
known to be unreliable under Camoufox's isolated-world execution
model (see https://github.com/daijro/camoufox/issues/48).
2. It captures more than the extension does (any resource type, not
just fetch/XHR), and can't be blocked by a page's CSP.
- Markdown extraction and CSS-selector helpers *do* need to run inside the
page (DOM traversal). Those live in extension/core/*.js and are read from
disk here, then run via page.evaluate(). This is safe under Camoufox's
default *isolated* world because that code only reads the DOM / mutates a
detached clone — it never writes to the live page — so no
`main_world_eval` workaround is required.
"""
from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Any, Optional
_JS_DIR = Path(__file__).parent / "js"
MAX_ENTRIES = 200
def _read_js(name: str) -> str:
path = _JS_DIR / name
if not path.exists():
raise FileNotFoundError(
f"Missing shared JS file: {path}. This package expects to run from "
"a checkout of the context-extractor repo where automation/context_extractor/js/*.js "
"are symlinks into extension/core/."
)
return path.read_text(encoding="utf-8")
def _now_ms() -> int:
return int(time.time() * 1000)
class _Store:
"""Bounded ring-buffer store matching the extension's shape exactly, so
build_ai_prompt() output is identical between the extension and here."""
def __init__(self) -> None:
self.console: list[dict[str, Any]] = []
self.errors: list[dict[str, Any]] = []
self.network: list[dict[str, Any]] = []
def push(self, which: str, entry: dict[str, Any]) -> None:
arr = getattr(self, which)
arr.append(entry)
while len(arr) > MAX_ENTRIES:
arr.pop(0)
def clear(self, which: Optional[str] = None) -> None:
if which:
getattr(self, which).clear()
else:
self.console.clear()
self.errors.clear()
self.network.clear()
def as_dict(self) -> dict[str, Any]:
return {"console": self.console, "errors": self.errors, "network": self.network}
_CONSOLE_LEVEL_MAP = {"warning": "warn"}
def _build_extract_script(selector: Optional[str]) -> str:
dom_js = _read_js("dom.js")
sel_json = json.dumps(selector or "")
return f"""
(() => {{
{dom_js}
let el = null;
const sel = {sel_json};
if (sel) {{ try {{ el = document.querySelector(sel); }} catch (_) {{ el = null; }} }}
if (!el) el = document.body;
return extractMarkdown(el);
}})()
"""
def _build_prompt_script(
meta: dict[str, Any], markdown: str, store: dict[str, Any], max_chars: Optional[int] = None
) -> str:
prompt_js = _read_js("prompt.js")
max_chars_js = json.dumps(max_chars) if max_chars is not None else "undefined"
return f"""
(() => {{
{prompt_js}
return buildAIPrompt({json.dumps(meta)}, {json.dumps(markdown)}, {json.dumps(store)}, {max_chars_js});
}})()
"""
class _CaptureMixin:
"""Shared event-handling logic. Playwright event callbacks are plain
synchronous functions in both the sync and async APIs (Playwright invokes
them itself; you never await them), so this is safe to share as-is."""
def _init_capture_state(self) -> None:
self.store = _Store()
self._request_starts: dict[int, float] = {}
def _on_console(self, msg) -> None:
level = _CONSOLE_LEVEL_MAP.get(msg.type, msg.type)
self.store.push("console", {"level": level, "ts": _now_ms(), "msg": msg.text})
def _on_pageerror(self, error) -> None:
message = getattr(error, "message", None) or str(error)
stack = getattr(error, "stack", "") or ""
self.store.push("errors", {
"type": "error", "ts": _now_ms(), "msg": message,
"source": "", "line": 0, "col": 0, "stack": stack,
})
def _on_request(self, request) -> None:
self._request_starts[id(request)] = time.monotonic()
def _on_requestfinished(self, request) -> None:
start = self._request_starts.pop(id(request), None)
duration = round((time.monotonic() - start) * 1000) if start else 0
status = 0
try:
response = request.response()
if response:
status = response.status
except Exception:
pass
self.store.push("network", {
"type": request.resource_type, "method": request.method, "url": request.url,
"status": status, "ts": _now_ms(), "duration": duration, "error": None,
})
def _on_requestfailed(self, request) -> None:
start = self._request_starts.pop(id(request), None)
duration = round((time.monotonic() - start) * 1000) if start else 0
failure = getattr(request, "failure", None)
error_text = failure.get("errorText") if isinstance(failure, dict) else str(failure or "Network error")
self.store.push("network", {
"type": request.resource_type, "method": request.method, "url": request.url,
"status": 0, "ts": _now_ms(), "duration": duration, "error": error_text or "Network error",
})
def get_store(self) -> dict[str, Any]:
return self.store.as_dict()
def clear(self, which: Optional[str] = None) -> None:
self.store.clear(which)
class ExtractorSession(_CaptureMixin):
"""Sync API. Use with `playwright.sync_api` or sync `camoufox.Camoufox`.
Example:
from playwright.sync_api import sync_playwright
from context_extractor import ExtractorSession
with sync_playwright() as p:
page = p.chromium.launch().new_page()
session = ExtractorSession(page)
page.goto("https://example.com")
print(session.build_ai_prompt())
"""
def __init__(self, page) -> None:
self.page = page
self._init_capture_state()
page.on("console", self._on_console)
page.on("pageerror", self._on_pageerror)
page.on("request", self._on_request)
page.on("requestfinished", self._on_requestfinished)
page.on("requestfailed", self._on_requestfailed)
def extract_markdown(self, selector: Optional[str] = None) -> dict[str, Any]:
markdown = self.page.evaluate(_build_extract_script(selector))
return {
"url": self.page.url,
"title": self.page.title(),
"ts": _now_ms(),
"selector": selector or "body",
"markdown": markdown or "",
}
def build_ai_prompt(self, selector: Optional[str] = None, max_chars: Optional[int] = None) -> str:
extracted = self.extract_markdown(selector)
meta = {k: extracted[k] for k in ("url", "title", "ts", "selector")}
script = _build_prompt_script(meta, extracted["markdown"], self.get_store(), max_chars)
return self.page.evaluate(script)
class AsyncExtractorSession(_CaptureMixin):
"""Async API. Use with `playwright.async_api` or `camoufox.AsyncCamoufox`.
Example:
from camoufox.async_api import AsyncCamoufox
from context_extractor import AsyncExtractorSession
async with AsyncCamoufox(headless=True) as browser:
page = await browser.new_page()
session = AsyncExtractorSession(page)
await page.goto("https://example.com")
print(await session.build_ai_prompt())
"""
def __init__(self, page) -> None:
self.page = page
self._init_capture_state()
page.on("console", self._on_console)
page.on("pageerror", self._on_pageerror)
page.on("request", self._on_request)
page.on("requestfinished", self._on_requestfinished)
page.on("requestfailed", self._on_requestfailed)
async def extract_markdown(self, selector: Optional[str] = None) -> dict[str, Any]:
markdown = await self.page.evaluate(_build_extract_script(selector))
return {
"url": self.page.url,
"title": await self.page.title(),
"ts": _now_ms(),
"selector": selector or "body",
"markdown": markdown or "",
}
async def build_ai_prompt(self, selector: Optional[str] = None, max_chars: Optional[int] = None) -> str:
extracted = await self.extract_markdown(selector)
meta = {k: extracted[k] for k in ("url", "title", "ts", "selector")}
script = _build_prompt_script(meta, extracted["markdown"], self.get_store(), max_chars)
return await self.page.evaluate(script)
+40
View File
@@ -0,0 +1,40 @@
"""Minimal example: capture context from a page using Camoufox (anti-detect Firefox).
Camoufox hands you a normal Playwright Page, so ExtractorSession works
unmodified. No `main_world_eval` or init-script workarounds are needed here:
capture uses Playwright's native page.on(...) hooks (not JS injection), and
markdown extraction only *reads* the DOM, which works fine in Camoufox's
default isolated world.
Run:
pip install -e ./automation[camoufox]
python automation/examples/scrape_camoufox.py https://example.com
"""
import sys
from camoufox import Camoufox
from context_extractor import ExtractorSession
def main() -> None:
url = sys.argv[1] if len(sys.argv) > 1 else "https://example.com"
with Camoufox(headless=True, geoip=True) as browser:
page = browser.new_page()
session = ExtractorSession(page)
page.goto(url, wait_until="domcontentloaded")
page.wait_for_timeout(1000) # let SPA fetches settle
print(session.build_ai_prompt())
# Or grab pieces individually:
# extracted = session.extract_markdown("#main-content")
# store = session.get_store() # {"console": [...], "errors": [...], "network": [...]}
if __name__ == "__main__":
main()
@@ -0,0 +1,31 @@
"""Async variant, using AsyncCamoufox + AsyncExtractorSession.
Run:
pip install -e ./automation[camoufox]
python automation/examples/scrape_camoufox_async.py https://example.com
"""
import asyncio
import sys
from camoufox.async_api import AsyncCamoufox
from context_extractor import AsyncExtractorSession
async def main() -> None:
url = sys.argv[1] if len(sys.argv) > 1 else "https://example.com"
async with AsyncCamoufox(headless=True, geoip=True) as browser:
page = await browser.new_page()
session = AsyncExtractorSession(page)
await page.goto(url, wait_until="domcontentloaded")
await page.wait_for_timeout(1000)
print(await session.build_ai_prompt())
if __name__ == "__main__":
asyncio.run(main())
+35
View File
@@ -0,0 +1,35 @@
"""Minimal example: capture context from a page using plain Playwright + Chromium.
Run:
pip install -e ./automation
playwright install chromium
python automation/examples/scrape_chromium.py https://example.com
"""
import sys
from playwright.sync_api import sync_playwright
from context_extractor import ExtractorSession
def main() -> None:
url = sys.argv[1] if len(sys.argv) > 1 else "https://example.com"
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
# Attach capture *before* navigating so console/network from the very
# first load are seen.
session = ExtractorSession(page)
page.goto(url, wait_until="networkidle")
print(session.build_ai_prompt()) # selector=None -> whole <body>
browser.close()
if __name__ == "__main__":
main()
+34
View File
@@ -0,0 +1,34 @@
[project]
name = "context-extractor"
version = "1.3.0"
description = "Capture console/network/errors and extract AI-ready markdown from a Playwright or Camoufox page."
readme = "README.md"
requires-python = ">=3.9"
license = { text = "MIT" }
authors = [{ name = "Context Extractor contributors" }]
keywords = ["playwright", "camoufox", "scraping", "llm", "markdown", "browser-extension"]
dependencies = [
"playwright>=1.44",
]
[project.optional-dependencies]
camoufox = ["camoufox[geoip]>=0.4"]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23"]
[project.scripts]
context-extractor = "context_extractor.cli:main"
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
include = ["context_extractor*"]
[tool.setuptools.package-data]
context_extractor = ["js/*.js"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
filterwarnings = ["ignore::DeprecationWarning"]
+28
View File
@@ -0,0 +1,28 @@
from __future__ import annotations
from pathlib import Path
import pytest
from playwright.sync_api import sync_playwright
FIXTURE = Path(__file__).parent / "fixtures" / "sample.html"
@pytest.fixture(scope="session")
def playwright_browser():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
yield browser
browser.close()
@pytest.fixture
def page(playwright_browser):
page = playwright_browser.new_page()
yield page
page.close()
@pytest.fixture
def fixture_url() -> str:
return FIXTURE.resolve().as_uri()
+33
View File
@@ -0,0 +1,33 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<title>Context Extractor Fixture</title>
</head>
<body>
<header>
<h1>Fixture Page</h1>
<nav>Skip this noise</nav>
</header>
<main id="main-content">
<h2>Section</h2>
<p>Hello <strong>world</strong>, visit <a href="/docs">docs</a>.</p>
<ul>
<li>Alpha</li>
<li>Beta</li>
</ul>
<div style="display:none">hidden-inline-style-blob {"secretConfig": "should-not-appear"}</div>
<div class="css-hidden">hidden-via-stylesheet-blob {"lixTracking": "should-not-appear"}</div>
<div hidden>hidden-attribute-blob should-not-appear</div>
<div aria-hidden="true">aria-hidden-blob should-not-appear</div>
</main>
<style>.css-hidden { display: none; }</style>
<script>
console.log("fixture boot");
console.warn("fixture warn");
console.error("fixture error log");
setTimeout(() => { throw new Error("fixture boom"); }, 20);
fetch("/missing-endpoint").catch(() => {});
</script>
</body>
</html>
+71
View File
@@ -0,0 +1,71 @@
"""Unit-style tests for the shared prompt.js formatter (via page.evaluate)."""
from __future__ import annotations
from pathlib import Path
from context_extractor.session import _build_prompt_script, _read_js
CORE = Path(__file__).resolve().parents[2] / "extension" / "core"
def test_shared_js_files_exist_and_match_package():
assert (CORE / "dom.js").is_file()
assert (CORE / "prompt.js").is_file()
# package-side copies/symlinks must resolve to the same source
assert "function extractMarkdown" in _read_js("dom.js")
assert "function buildAIPrompt" in _read_js("prompt.js")
def test_build_ai_prompt_script_formats_errors_and_failures(page):
page.goto("about:blank")
meta = {
"url": "https://example.test/page",
"title": "T",
"ts": 1_700_000_000_000,
"selector": "body",
}
store = {
"console": [{"level": "warn", "ts": 1, "msg": "careful"}],
"errors": [{
"type": "error", "ts": 2, "msg": "boom",
"source": "a.js", "line": 3, "col": 4, "stack": "Error: boom\n at x",
}],
"network": [{
"type": "fetch", "method": "GET", "url": "https://example.test/api",
"status": 500, "ts": 3, "duration": 12, "error": None,
}],
}
script = _build_prompt_script(meta, "# Hello", store)
out = page.evaluate(script)
assert "# Page Context" in out
assert "https://example.test/page" in out
assert "## Page Content" in out
assert "# Hello" in out
assert "## JavaScript Errors" in out
assert "boom" in out
assert "## Console Errors & Warnings" in out
assert "[WARN] careful" in out
assert "## Failed Requests" in out
assert "GET 500 https://example.test/api" in out
def test_build_ai_prompt_truncates_huge_markdown_by_default(page):
"""Regression test: a JS-heavy SPA (e.g. LinkedIn) can extract 700k+ chars
of body content. The prompt must not blindly include all of it."""
page.goto("about:blank")
meta = {"url": "https://example.test/huge", "title": "T", "ts": 1, "selector": "body"}
huge_markdown = "x" * 50_000
script = _build_prompt_script(meta, huge_markdown, {"console": [], "errors": [], "network": []})
out = page.evaluate(script)
assert "truncated to 20,000 of 50,000 chars" in out
assert len(out) < 21_000 # header/footer overhead + capped body, nowhere near 50k
def test_build_ai_prompt_respects_custom_max_chars(page):
page.goto("about:blank")
meta = {"url": "https://example.test/huge", "title": "T", "ts": 1, "selector": "body"}
markdown = "y" * 1000
script = _build_prompt_script(meta, markdown, {"console": [], "errors": [], "network": []}, max_chars=200)
out = page.evaluate(script)
assert "truncated to 200 of 1,000 chars" in out
+84
View File
@@ -0,0 +1,84 @@
from __future__ import annotations
from context_extractor import AsyncExtractorSession, ExtractorSession, __version__
def test_version_semverish():
assert __version__.count(".") >= 1
def test_capture_console_errors_and_network(page, fixture_url):
session = ExtractorSession(page)
page.goto(fixture_url, wait_until="domcontentloaded")
page.wait_for_timeout(150)
store = session.get_store()
levels = {e["level"] for e in store["console"]}
assert "warn" in levels
assert "error" in levels or any("error" in (e.get("msg") or "").lower() for e in store["console"])
assert any(e["msg"] == "fixture boom" or "fixture boom" in (e.get("msg") or "") for e in store["errors"])
assert store["network"], "expected at least the document request"
def test_extract_markdown_body_and_selector(page, fixture_url):
session = ExtractorSession(page)
page.goto(fixture_url, wait_until="domcontentloaded")
full = session.extract_markdown()
assert full["title"] == "Context Extractor Fixture"
assert "Fixture Page" in full["markdown"]
assert "**world**" in full["markdown"]
assert "[docs](/docs)" in full["markdown"]
assert "- Alpha" in full["markdown"]
scoped = session.extract_markdown("#main-content")
assert scoped["selector"] == "#main-content"
assert "Section" in scoped["markdown"]
assert "Skip this noise" not in scoped["markdown"]
def test_hidden_nodes_are_excluded_from_markdown(page, fixture_url):
"""Regression test: SPAs (e.g. LinkedIn) stash hydration/experiment JSON in
hidden DOM nodes, not just <script> tags. Any hidden node must be excluded
regardless of *how* it's hidden (inline style, stylesheet class, [hidden],
aria-hidden)."""
session = ExtractorSession(page)
page.goto(fixture_url, wait_until="domcontentloaded")
markdown = session.extract_markdown("#main-content")["markdown"]
assert "should-not-appear" not in markdown
assert "secretConfig" not in markdown
assert "lixTracking" not in markdown
# Sanity: visible content in the same container still comes through.
assert "Alpha" in markdown
assert "Beta" in markdown
def test_build_ai_prompt_has_sections(page, fixture_url):
session = ExtractorSession(page)
page.goto(fixture_url, wait_until="domcontentloaded")
page.wait_for_timeout(150)
prompt = session.build_ai_prompt("#main-content")
assert prompt.startswith("# Page Context")
assert "## Page Content" in prompt
assert "Hello **world**" in prompt
assert "JavaScript Errors" in prompt or "Console Errors" in prompt
assert "_Extracted by Context Extractor_" in prompt
def test_clear_store(page, fixture_url):
session = ExtractorSession(page)
page.goto(fixture_url, wait_until="domcontentloaded")
page.wait_for_timeout(100)
assert session.get_store()["console"]
session.clear("console")
assert session.get_store()["console"] == []
session.clear()
store = session.get_store()
assert store == {"console": [], "errors": [], "network": []}
def test_async_session_exported():
assert AsyncExtractorSession is not None