core/website.py

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