Initial commit: Screen Leads app
Screen-capture lead tool: capture agent, Claude vision extractor with pluggable site recipes (LinkedIn), canonical funnel model, SQLite storage, FastAPI backend, dashboard, and setup/run scripts. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
"""Vision extraction via Claude.
|
||||
|
||||
Two calls:
|
||||
- extract_lead(): screenshots + a recipe's schema -> raw structured JSON.
|
||||
- classify_page(): one screenshot -> {site, page_type, address bar url}, used
|
||||
only when no native browser URL is available (visual site detection).
|
||||
|
||||
Both use structured outputs (output_config.format) so the response is
|
||||
schema-validated JSON, not free text.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
from functools import lru_cache
|
||||
from typing import Any
|
||||
|
||||
from config import ANTHROPIC_MODEL, MAX_IMAGES_PER_CALL
|
||||
from ai.recipes.base import SiteRecipe
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def _client():
|
||||
import anthropic
|
||||
|
||||
# Resolves ANTHROPIC_API_KEY (or an `ant auth login` profile) from the env.
|
||||
return anthropic.Anthropic()
|
||||
|
||||
|
||||
def _image_block(path: str) -> dict[str, Any]:
|
||||
with open(path, "rb") as f:
|
||||
data = base64.standard_b64encode(f.read()).decode("utf-8")
|
||||
return {
|
||||
"type": "image",
|
||||
"source": {"type": "base64", "media_type": "image/png", "data": data},
|
||||
}
|
||||
|
||||
|
||||
def _first_json(response) -> dict[str, Any]:
|
||||
for block in response.content:
|
||||
if block.type == "text":
|
||||
return json.loads(block.text)
|
||||
raise RuntimeError("model returned no text block")
|
||||
|
||||
|
||||
def extract_lead(recipe: SiteRecipe, screenshot_paths: list[str]) -> dict[str, Any]:
|
||||
paths = screenshot_paths[:MAX_IMAGES_PER_CALL]
|
||||
content: list[dict[str, Any]] = [_image_block(p) for p in paths]
|
||||
content.append({"type": "text", "text": recipe.extraction_prompt})
|
||||
|
||||
response = _client().messages.create(
|
||||
model=ANTHROPIC_MODEL,
|
||||
max_tokens=2048,
|
||||
messages=[{"role": "user", "content": content}],
|
||||
output_config={"format": {"type": "json_schema", "schema": recipe.extraction_schema}},
|
||||
)
|
||||
return _first_json(response)
|
||||
|
||||
|
||||
_CLASSIFY_SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"site": {"type": "string"}, # e.g. "linkedin", "other"
|
||||
"page_type": {"type": "string"}, # e.g. "profile", "search", "feed", "other"
|
||||
"address_bar_url": {"type": "string"},
|
||||
},
|
||||
"required": ["site", "page_type", "address_bar_url"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
_CLASSIFY_PROMPT = (
|
||||
"This is a screenshot of a web browser. Identify the site and page from the "
|
||||
"browser's address bar and the visible layout. Return: 'site' (lowercase, "
|
||||
"e.g. 'linkedin', or 'other'); 'page_type' ('profile' for a single person's "
|
||||
"page, otherwise 'search', 'feed', 'messaging', or 'other'); and "
|
||||
"'address_bar_url' (the URL text you can read, or \"\")."
|
||||
)
|
||||
|
||||
|
||||
def classify_page(screenshot_path: str) -> dict[str, Any]:
|
||||
response = _client().messages.create(
|
||||
model=ANTHROPIC_MODEL,
|
||||
max_tokens=512,
|
||||
messages=[{
|
||||
"role": "user",
|
||||
"content": [_image_block(screenshot_path), {"type": "text", "text": _CLASSIFY_PROMPT}],
|
||||
}],
|
||||
output_config={"format": {"type": "json_schema", "schema": _CLASSIFY_SCHEMA}},
|
||||
)
|
||||
return _first_json(response)
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Recipe registry. Register new site recipes here — nothing else changes."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from core.models import PageContext, PageMatch
|
||||
from .base import SiteRecipe
|
||||
from .linkedin import LinkedInRecipe
|
||||
|
||||
REGISTRY: list[SiteRecipe] = [
|
||||
LinkedInRecipe(),
|
||||
]
|
||||
|
||||
|
||||
def enabled_recipes() -> list[SiteRecipe]:
|
||||
return [r for r in REGISTRY if r.enabled]
|
||||
|
||||
|
||||
def match(ctx: PageContext) -> Optional[tuple[SiteRecipe, PageMatch]]:
|
||||
"""Return the first enabled recipe that recognises the page context."""
|
||||
for recipe in enabled_recipes():
|
||||
m = recipe.matches(ctx)
|
||||
if m is not None:
|
||||
return recipe, m
|
||||
return None
|
||||
@@ -0,0 +1,36 @@
|
||||
"""SiteRecipe interface. Each supported site implements one of these; adding a
|
||||
new site is a drop-in module registered in ai/recipes/__init__.py — the core
|
||||
capture loop never changes.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, Optional
|
||||
|
||||
from core.models import Lead, PageContext, PageMatch
|
||||
|
||||
|
||||
class SiteRecipe(ABC):
|
||||
site_id: str = ""
|
||||
display_name: str = ""
|
||||
enabled: bool = True
|
||||
# Which extracted page_type is a capturable target (e.g. a person's profile).
|
||||
target_page_type: str = "profile"
|
||||
|
||||
@abstractmethod
|
||||
def matches(self, ctx: PageContext) -> Optional[PageMatch]:
|
||||
"""Return a PageMatch if this recipe recognises the page, else None."""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def extraction_schema(self) -> dict[str, Any]:
|
||||
"""JSON schema (structured-output format) for the raw extraction."""
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def extraction_prompt(self) -> str:
|
||||
"""Instruction given to the vision model alongside the screenshots."""
|
||||
|
||||
@abstractmethod
|
||||
def to_canonical(self, raw: dict[str, Any]) -> Lead:
|
||||
"""Map this site's raw extraction to the canonical Lead schema."""
|
||||
@@ -0,0 +1,91 @@
|
||||
"""LinkedIn profile recipe — the first supported site.
|
||||
|
||||
Target page: a person's profile (linkedin.com/in/...), typically reached from
|
||||
search. Anything else on LinkedIn (feed, search results, messaging) is not a
|
||||
target; non-LinkedIn pages don't match at all.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any, Optional
|
||||
|
||||
from core.models import Lead, PageContext, PageMatch
|
||||
from .base import SiteRecipe
|
||||
|
||||
_PROFILE_RE = re.compile(r"linkedin\.com/in/[^/?#]+", re.I)
|
||||
_LINKEDIN_RE = re.compile(r"linkedin\.com", re.I)
|
||||
|
||||
|
||||
class LinkedInRecipe(SiteRecipe):
|
||||
site_id = "linkedin"
|
||||
display_name = "LinkedIn"
|
||||
enabled = True
|
||||
target_page_type = "profile"
|
||||
|
||||
def matches(self, ctx: PageContext) -> Optional[PageMatch]:
|
||||
url = ctx.best_url
|
||||
if url:
|
||||
if _PROFILE_RE.search(url):
|
||||
return PageMatch(self.site_id, "profile", is_target=True)
|
||||
if _LINKEDIN_RE.search(url):
|
||||
return PageMatch(self.site_id, "other", is_target=False)
|
||||
return None
|
||||
# No URL available: fall back to the AI's visual read.
|
||||
if (ctx.ai_site or "").lower() == "linkedin":
|
||||
page_type = (ctx.ai_page_type or "other").lower()
|
||||
return PageMatch(self.site_id, page_type, is_target=(page_type == "profile"))
|
||||
return None
|
||||
|
||||
@property
|
||||
def extraction_schema(self) -> dict[str, Any]:
|
||||
str_fields = [
|
||||
"full_name", "headline", "current_title", "current_company",
|
||||
"location", "about", "email", "phone", "website",
|
||||
]
|
||||
props: dict[str, Any] = {f: {"type": "string"} for f in str_fields}
|
||||
props["other_profiles"] = {"type": "array", "items": {"type": "string"}}
|
||||
props["confidence"] = {"type": "number"}
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": props,
|
||||
"required": str_fields + ["other_profiles", "confidence"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
@property
|
||||
def extraction_prompt(self) -> str:
|
||||
return (
|
||||
"These screenshots are a single LinkedIn profile page, captured top "
|
||||
"to bottom. Extract the person's details into the required JSON. "
|
||||
"Rules: use an empty string \"\" for any field not visible; do NOT "
|
||||
"guess or invent contact details. Email, phone and website usually "
|
||||
"only appear if the user opened the 'Contact info' panel — extract "
|
||||
"them only if they are actually shown. 'other_profiles' is any other "
|
||||
"social/web links visible (Twitter/X, GitHub, personal site). "
|
||||
"'confidence' is 0-1 for how legible the profile was."
|
||||
)
|
||||
|
||||
def to_canonical(self, raw: dict[str, Any]) -> Lead:
|
||||
socials: dict[str, str] = {}
|
||||
for link in raw.get("other_profiles") or []:
|
||||
low = link.lower()
|
||||
if "github.com" in low:
|
||||
socials["github"] = link
|
||||
elif "twitter.com" in low or "x.com" in low:
|
||||
socials["twitter"] = link
|
||||
else:
|
||||
socials.setdefault("other", link)
|
||||
return Lead(
|
||||
full_name=raw.get("full_name", "") or "",
|
||||
headline=raw.get("headline", "") or "",
|
||||
title=raw.get("current_title", "") or "",
|
||||
company=raw.get("current_company", "") or "",
|
||||
location=raw.get("location", "") or "",
|
||||
about=raw.get("about", "") or "",
|
||||
email=raw.get("email", "") or "",
|
||||
phone=raw.get("phone", "") or "",
|
||||
website=raw.get("website", "") or "",
|
||||
social_handles=socials,
|
||||
source_site=self.site_id,
|
||||
confidence=float(raw.get("confidence", 0) or 0),
|
||||
)
|
||||
Reference in New Issue
Block a user