from __future__ import annotations import re from collections.abc import Callable from html import unescape from html.parser import HTMLParser from urllib.parse import urljoin, urlparse import httpx from core.schemas import WebsiteIntentInspection, WebsitePageInspection FetchHtml = Callable[[str], str] KEY_LINK_TERMS = ( "pricing", "demo", "signup", "sign up", "get started", "about", "product", "features", "solutions", "customers", "docs", "contact", ) CTA_TERMS = ( "get started", "start", "sign up", "try", "book", "demo", "contact", "buy", "subscribe", "download", "learn more", ) class _PageParser(HTMLParser): def __init__(self, base_url: str) -> None: super().__init__() self.base_url = base_url self.title = "" self.headings: list[str] = [] self.links: list[tuple[str, str]] = [] self.button_texts: list[str] = [] self.body_chunks: list[str] = [] self._tag_stack: list[str] = [] self._current_href: str | None = None self._current_link_text: list[str] = [] self._current_button_text: list[str] = [] self._current_heading_text: list[str] = [] self._current_title_text: list[str] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: self._tag_stack.append(tag) attrs_by_name = {name: value for name, value in attrs} if tag == "a": self._current_href = attrs_by_name.get("href") self._current_link_text = [] elif tag == "button": self._current_button_text = [] elif tag in {"h1", "h2"}: self._current_heading_text = [] elif tag == "title": self._current_title_text = [] def handle_endtag(self, tag: str) -> None: if tag == "a" and self._current_href: label = _clean_text(" ".join(self._current_link_text)) if label: self.links.append((label, urljoin(self.base_url, self._current_href))) self._current_href = None elif tag == "button": label = _clean_text(" ".join(self._current_button_text)) if label: self.button_texts.append(label) elif tag in {"h1", "h2"}: heading = _clean_text(" ".join(self._current_heading_text)) if heading: self.headings.append(heading) elif tag == "title": self.title = _clean_text(" ".join(self._current_title_text)) if self._tag_stack: self._tag_stack.pop() def handle_data(self, data: str) -> None: if not data.strip(): return current_tag = self._tag_stack[-1] if self._tag_stack else "" if current_tag in {"script", "style", "noscript"}: return text = unescape(data) self.body_chunks.append(text) if self._current_href is not None: self._current_link_text.append(text) if "button" in self._tag_stack: self._current_button_text.append(text) if current_tag in {"h1", "h2"}: self._current_heading_text.append(text) if current_tag == "title": self._current_title_text.append(text) def inspect_website_intent( url: str, stated_outcome: str = "", fetch_html: FetchHtml | None = None, max_pages: int = 5, ) -> WebsiteIntentInspection: normalized_url = _normalize_url(url) fetch = fetch_html or _fetch_html pages = _crawl_key_pages(normalized_url, fetch, max_pages=max_pages) if not pages: raise ValueError(f"Could not inspect website URL: {url}") combined_text = _combined_page_text(pages).lower() purpose = _infer_purpose(combined_text) audience = _infer_audience(combined_text) action = _infer_primary_action(pages) confidence = _confidence_for(pages, purpose, audience, action) warnings = _mismatch_warnings(pages, combined_text, purpose, audience, action) drift_warnings = _intent_drift_warnings(stated_outcome, purpose, audience) return WebsiteIntentInspection( url=normalized_url, pages=pages, stated_outcome=stated_outcome.strip(), inferred_purpose=purpose, inferred_audience=audience, primary_action=action, confidence=confidence, mismatch_warnings=warnings, intent_drift_warnings=drift_warnings, ) def website_intent_description(intent: WebsiteIntentInspection) -> str: page_summaries = "; ".join( f"{page.title or page.url}: {', '.join(page.headings[:2])}" for page in intent.pages[:5] ) warnings = "; ".join(intent.mismatch_warnings) or "No obvious intent mismatch warnings." return ( f"Website URL: {intent.url}\n" f"Inferred purpose: {intent.inferred_purpose}\n" f"Inferred audience: {intent.inferred_audience}\n" f"Primary expected action: {intent.primary_action}\n" f"Inspected pages: {page_summaries}\n" f"Intent mismatch warnings: {warnings}" ) def inferred_flow_steps(intent: WebsiteIntentInspection) -> list[str]: return [ "Land on website", "Understand what the website is for", f"Decide whether to {intent.primary_action.lower()}", ] def _combined_page_text(pages: list[WebsitePageInspection]) -> str: return " ".join( text for page in pages for text in [page.title, *page.headings, *page.nav_labels, *page.ctas, page.text_excerpt] ) def _crawl_key_pages( url: str, fetch_html: FetchHtml, max_pages: int, ) -> list[WebsitePageInspection]: seen: set[str] = set() pages: list[WebsitePageInspection] = [] queue = [url] while queue and len(pages) < max_pages: current_url = queue.pop(0) if current_url in seen: continue seen.add(current_url) try: html = fetch_html(current_url) except Exception: continue parser = _parse_page(current_url, html) pages.append(_page_from_parser(current_url, parser)) if len(pages) == 1: queue.extend(_key_same_domain_links(url, parser.links, max_pages=max_pages - 1)) return pages def _parse_page(url: str, html: str) -> _PageParser: parser = _PageParser(url) parser.feed(html) return parser def _page_from_parser(url: str, parser: _PageParser) -> WebsitePageInspection: nav_labels = _unique([label for label, _href in parser.links])[:12] ctas = _unique( [ *parser.button_texts, *(label for label, _href in parser.links if _looks_like_cta(label)), ] )[:8] body = _clean_text(" ".join(parser.body_chunks)) return WebsitePageInspection( url=url, title=parser.title, headings=_unique(parser.headings)[:8], ctas=ctas, nav_labels=nav_labels, text_excerpt=body[:1200], ) def _fetch_html(url: str) -> str: response = httpx.get( url, follow_redirects=True, timeout=10, headers={"User-Agent": "SiegeIntentInspector/0.1"}, ) response.raise_for_status() content_type = response.headers.get("content-type", "") if "html" not in content_type.lower(): raise ValueError(f"URL did not return HTML: {url}") return response.text def _normalize_url(url: str) -> str: stripped = url.strip() if not stripped: raise ValueError("Website URL is empty") if not re.match(r"^https?://", stripped): stripped = f"https://{stripped}" return stripped def _key_same_domain_links( base_url: str, links: list[tuple[str, str]], max_pages: int, ) -> list[str]: base_host = urlparse(base_url).netloc.lower().removeprefix("www.") candidates: list[str] = [] for label, href in links: parsed = urlparse(href) host = parsed.netloc.lower().removeprefix("www.") if parsed.scheme not in {"http", "https"} or host != base_host: continue combined = f"{label} {parsed.path}".lower() if any(term in combined for term in KEY_LINK_TERMS): cleaned = parsed._replace(fragment="", query="").geturl() candidates.append(cleaned) return _unique(candidates)[:max_pages] def _infer_purpose(text: str) -> str: if any(term in text for term in ["learn", "student", "school", "teacher", "education"]): return "Help learners or educators improve learning outcomes" if any(term in text for term in ["ai", "automation", "workflow", "agent"]): return "Use AI or automation to improve a work process" if any(term in text for term in ["analytics", "dashboard", "report", "insight"]): return "Help users understand performance through analytics" if any(term in text for term in ["shop", "cart", "buy", "product"]): return "Sell products or services online" if any(term in text for term in ["book", "appointment", "schedule", "contact"]): return "Convert visitors into booked or contacted leads" return "Explain an offering and move visitors toward a next step" def _infer_audience(text: str) -> str: if "parent" in text: return "Parents and families" if any(term in text for term in ["student", "learner"]): return "Students and learners" if any(term in text for term in ["teacher", "school", "educator"]): return "Teachers and education teams" if any(term in text for term in ["developer", "api", "docs"]): return "Developers and technical teams" if any(term in text for term in ["team", "business", "enterprise", "company"]): return "Business teams" return "First-time website visitors" def _infer_primary_action(pages: list[WebsitePageInspection]) -> str: ctas = [cta.lower() for page in pages for cta in page.ctas] for phrase, action in [ ("sign up", "Sign up"), ("get started", "Get started"), ("start", "Start"), ("book", "Book a demo or appointment"), ("demo", "Request or watch a demo"), ("contact", "Contact the team"), ("download", "Download the product"), ("learn more", "Learn more"), ]: if any(phrase in cta for cta in ctas): return action return "Find the next step" def _confidence_for( pages: list[WebsitePageInspection], purpose: str, audience: str, action: str, ) -> str: signals = 0 if pages[0].title or pages[0].headings: signals += 1 if purpose != "Explain an offering and move visitors toward a next step": signals += 1 if audience != "First-time website visitors": signals += 1 if action != "Find the next step": signals += 1 if len(pages) > 1: signals += 1 return "high" if signals >= 4 else "medium" if signals >= 2 else "low" def _mismatch_warnings( pages: list[WebsitePageInspection], text: str, purpose: str, audience: str, action: str, ) -> list[str]: warnings: list[str] = [] homepage = pages[0] homepage_text = " ".join([homepage.title, *homepage.headings, homepage.text_excerpt]).lower() if not homepage.ctas: warnings.append("No obvious primary action was found on the homepage.") if action == "Find the next step": warnings.append("The website's intended next step is unclear from visible CTAs.") if "learning" in purpose.lower() and not any( term in text for term in ["subject", "practice", "lesson", "quiz", "tutor"] ): warnings.append( "The site appears education-oriented, but the learning experience is vague." ) if audience in {"Parents and families", "Students and learners"} and not any( term in text for term in ["price", "pricing", "free", "trial", "safety", "trust"] ): warnings.append( "The inferred audience likely needs pricing or trust signals, " "but they are hard to find." ) if len(homepage_text.split()) < 80: warnings.append("The homepage may not provide enough copy to explain the offering.") if not any(term in text for term in ["demo", "screenshot", "preview", "video", "how it works"]): warnings.append("The frontend does not clearly show the product or experience in use.") return warnings[:5] def _intent_drift_warnings( stated_outcome: str, inferred_purpose: str, inferred_audience: str, ) -> list[str]: if not stated_outcome.strip(): return [] stated_category = _intent_category(stated_outcome) inferred_category = _intent_category(f"{inferred_purpose} {inferred_audience}") if stated_category == "unknown" or inferred_category == "unknown": return [] if stated_category == inferred_category: return [] return [ ( "Stated outcome appears to target " f"{stated_category}, but the website reads as {inferred_category}." ) ] def _intent_category(text: str) -> str: lowered = text.lower() if any(term in lowered for term in ["learn", "student", "school", "teacher", "education"]): return "education" if any(term in lowered for term in ["accounting", "finance", "invoice", "bookkeeping"]): return "finance" if any(term in lowered for term in ["developer", "api", "docs", "sdk"]): return "developer tools" if any(term in lowered for term in ["sell", "shop", "cart", "buy", "ecommerce"]): return "commerce" if any(term in lowered for term in ["team", "business", "enterprise", "workflow"]): return "business software" return "unknown" def _looks_like_cta(label: str) -> bool: lowered = label.lower() return any(term in lowered for term in CTA_TERMS) def _clean_text(text: str) -> str: return re.sub(r"\s+", " ", unescape(text)).strip() def _unique(items: list[str]) -> list[str]: seen: set[str] = set() unique_items: list[str] = [] for item in items: cleaned = _clean_text(item) key = cleaned.lower() if cleaned and key not in seen: seen.add(key) unique_items.append(cleaned) return unique_items