"""Free public "store support audit" tool — logic + rendering. For a crawled Shopify store we score how many of the 12 universal high-intent buyer questions its OWN public site answers (reusing the honest gap report), grade it, and persist it (AuditResult, keyed by domain) so /audit/ is a PERMANENT, shareable, crawlable page — the build-as-marketing / programmatic-SEO wedge. Everything here is honest by construction: real buyer questions, HIT/MISS from the store's own indexed pages, conservative MISS threshold, no invented figures. """ # ruff: noqa: E501 - this module embeds HTML/CSS page templates from __future__ import annotations import asyncio import html import io import json as _json import logging import re as _re from datetime import UTC, datetime import httpx from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.demo_gap import GAP_QUESTIONS from app.models import AuditResult from app.rag import extract from app.shopify.policies import fetch_shop_policies log = logging.getLogger(__name__) PUBLIC_BASE = "https://atendyo.com" BACKEND_BASE = "https://victor34593993-flexigo-support-bot.hf.space" def grade_for(answered: int, total: int) -> str: """A clear letter grade from the answered ratio (honest, not punitive).""" if total <= 0: return "F" pct = answered / total if pct >= 0.9: return "A" if pct >= 0.75: return "B" if pct >= 0.6: return "C" if pct >= 0.4: return "D" return "F" def _norm_domain(raw: str) -> str: d = (raw or "").strip().lower() d = d.split("//")[-1].split("/")[0].split("?")[0].removeprefix("www.") # #17: a domain is later interpolated into HTML attributes + inline JS. Restrict # to real hostname characters so a store-controlled value can never carry # quotes/<>/angle brackets that would break out into an XSS/attribute injection. import re d = re.sub(r"[^a-z0-9.-]", "", d) return d[:255] async def save_audit( db: AsyncSession, domain: str, report: dict, lang: str = "es" ) -> AuditResult: """Upsert the audit result for a domain (re-audits update the same row).""" domain = _norm_domain(domain) answered = int(report.get("answered", 0)) total = int(report.get("total", 12)) row = ( await db.execute(select(AuditResult).where(AuditResult.domain == domain)) ).scalar_one_or_none() now = datetime.now(UTC) if row is None: row = AuditResult(domain=domain, created_at=now) db.add(row) row.brand = (report.get("brand") or domain)[:255] row.answered = answered row.total = total row.grade = grade_for(answered, total) row.questions = report.get("questions", []) row.lang = "en" if lang == "en" else "es" row.updated_at = now await db.flush() return row async def get_audit(db: AsyncSession, domain: str) -> AuditResult | None: return ( await db.execute( select(AuditResult).where(AuditResult.domain == _norm_domain(domain)) ) ).scalar_one_or_none() async def get_or_audit(db: AsyncSession, domain: str, lang: str = "es") -> AuditResult: """Cached audit if we have one, else audit it now (used by the compare view).""" row = await get_audit(db, domain) if row is None: row = await audit_store(db, domain, lang) return row # Common Shopify info pages worth reading (beyond /policies/*). Best-effort. _INFO_PATHS = [ "", "pages/faq", "pages/faqs", "pages/about", "pages/about-us", "pages/contact", "pages/size-guide", "pages/sizing", "pages/shipping", "pages/shipping-returns", "pages/returns", "pages/help", "pages/frequently-asked-questions", ] async def _gather_store_content(domain: str, *, cap: int = 14000) -> str: """Reliable, server-rendered content for the 12 questions: the store's REAL /policies/* pages (refund/shipping/privacy/terms) + a few common info pages. Independent of the flaky JS crawl, so a JS-heavy storefront is still scored from its real policy text. Best-effort: failures are skipped.""" domain = _norm_domain(domain) base = "https://" + domain parts: list[str] = [] try: for pol in await fetch_shop_policies(domain): parts.append(f"[{pol['title']}]\n{pol['text']}") except Exception: # noqa: BLE001 log.warning("audit: policy fetch failed for %s", domain, exc_info=True) async def _one(client: httpx.AsyncClient, path: str) -> str: url = f"{base}/{path}".rstrip("/") try: txt = (await extract.extract_url(url, client=client) or "").strip() return f"[{path or 'home'}]\n{txt[:3000]}" if len(txt) >= 40 else "" except Exception: # noqa: BLE001 return "" try: async with httpx.AsyncClient( timeout=extract.CRAWL_TIMEOUT, follow_redirects=True, event_hooks={"request": [extract._ssrf_guard_hook]}, ) as client: results = await asyncio.gather(*[_one(client, p) for p in _INFO_PATHS]) parts.extend(r for r in results if r) except Exception: # noqa: BLE001 log.warning("audit: page fetch failed for %s", domain, exc_info=True) # Strip boilerplate: a line that repeats across 3+ of the fetched pages is the # nav/menu/footer, not real policy/info content — dropping it lets the judge # see the actual answers instead of menu noise. Keep the [label] markers. from collections import Counter lines = "\n\n".join(parts).split("\n") freq = Counter(ln.strip() for ln in lines if len(ln.strip()) > 3) cleaned = [ln for ln in lines if ln.startswith("[") or freq.get(ln.strip(), 0) < 3] return "\n".join(cleaned)[:cap] def _parse_verdicts(raw: str) -> list[bool]: """Parse the LLM's verdicts robustly: strip markdown fences, try a bare JSON array, else any [...] block, else extract the run of true/false tokens.""" s = (raw or "").strip().strip("`") s = _re.sub(r"^json", "", s, flags=_re.I).strip() candidates = [s] block = _re.search(r"\[[\s\S]*?\]", s) if block: candidates.append(block.group(0)) for cand in candidates: try: arr = _json.loads(cand) if isinstance(arr, list): return [bool(x) for x in arr][:12] except (ValueError, TypeError): pass # last resort: read the ordered true/false words the model emitted toks = _re.findall(r"\b(true|false|sí|si|no|yes)\b", s, flags=_re.I) if toks: return [t.lower() in ("true", "sí", "si", "yes") for t in toks][:12] return [] async def _judge_raw(content: str, lang: str) -> tuple[list[bool], str]: """Returns (verdicts, raw_llm_text) — raw exposed for debugging.""" from app.config import get_settings from app.llm.router import build_router_from_settings qs = GAP_QUESTIONS.get("en" if lang == "en" else "es", GAP_QUESTIONS["es"]) numbered = "\n".join(f"{i + 1}. {q}" for i, q in enumerate(qs)) sys = ( "You audit an online store's PUBLIC content against 12 questions shoppers ask " "before buying. For EACH question, decide if the content gives a usable answer " "(a brief mention that actually answers counts; pure navigation/menu text does " "not). Reply with ONLY a JSON array of exactly 12 values true or false, in order " "(true = answered). No prose, no keys, just the array." ) user = ( f"STORE CONTENT:\n{content or '(no content found)'}\n\n" f"QUESTIONS:\n{numbered}\n\nJSON array of 12 true/false:" ) router = build_router_from_settings(get_settings()) result = await router.chat( [{"role": "system", "content": sys}, {"role": "user", "content": user}], # gpt-oss-120b is a REASONING model: it spends tokens "thinking" before the # answer, so a small max_tokens leaves the content EMPTY. Give it room. tools=[], temperature=0.0, max_tokens=2000, ) raw = (result.content or "").strip() verdicts = _parse_verdicts(raw) # #7: do NOT pad an empty/unparseable judge to a fake 0/12. gpt-oss-120b is a # reasoning model that can spend all tokens "thinking" and return empty/prose; # padding would publish a permanent, SEO-indexed, DEFAMATORY "0/12 grade F" for # a store that may be excellent. Require at least half the verdicts parsed, # else signal failure so audit_store -> 422 and no page is persisted. if len(verdicts) < 6: raise ValueError("judge_unparseable") verdicts = verdicts[:12] + [False] * (12 - len(verdicts)) return verdicts, raw async def _judge_questions(content: str, lang: str) -> list[bool]: verdicts, _ = await _judge_raw(content, lang) return verdicts async def audit_store(db: AsyncSession, domain: str, lang: str = "es") -> AuditResult: """Fetch the store's REAL public content + LLM-judge the 12 buyer questions, then persist. Independent of the demo crawl (reliable + no lead-email noise). Raises ValueError('no_content') when the store can't be read — we never show a fake 0/12 just because a probe failed.""" domain = _norm_domain(domain) content = await _gather_store_content(domain) if len(content) < 80: raise ValueError("no_content") verdicts = await _judge_questions(content, lang) qs = GAP_QUESTIONS.get("en" if lang == "en" else "es", GAP_QUESTIONS["es"]) questions = [ {"q": q, "status": "hit" if v else "miss"} for q, v in zip(qs, verdicts, strict=False) ] report = { "brand": domain, "total": len(qs), "answered": sum(1 for v in verdicts if v), "questions": questions, } return await save_audit(db, domain, report, lang=lang) # ── rendering ─────────────────────────────────────────────────────────────── _T = { "es": { "title": "{brand}: responde {a}/{t} preguntas de compra | Auditoría Atendyo", "desc": "{brand} responde {a} de {t} preguntas clave que tus clientes hacen antes de comprar. Mira cuáles fallan (gratis).", "h1": "{brand} responde {a}/{t} preguntas de compra", "sub": "Estas son las preguntas que tus clientes hacen ANTES de comprar. Cada una sin respuesta clara en tu web = una venta que se escapa o un ticket de soporte.", "answers": "Tu web responde", "gaps_h": "No responde con claridad:", "ok_h": "Sí responde:", "cta_h": "El asistente de IA de Atendyo responde TODAS estas, 24/7, con la información de tu propia tienda.", "cta_btn": "Probar gratis con mi tienda", "share": "Comparte tu resultado", "copy": "Copiar enlace", "audit_another": "Auditar otra tienda", "powered": "Auditoría gratis de Atendyo · atendyo.com", "compare_h": "Compáralo con un competidor", "compare_btn": "Comparar", "badge_h": "Muestra tu nota en tu web (insignia)", }, "en": { "title": "{brand}: answers {a}/{t} shopper questions | Atendyo Audit", "desc": "{brand} answers {a} of {t} key questions your customers ask before buying. See which ones fail (free).", "h1": "{brand} answers {a}/{t} shopper questions", "sub": "These are the questions your customers ask BEFORE buying. Each one your site doesn't clearly answer = a lost sale or a support ticket.", "answers": "Your site answers", "gaps_h": "Not clearly answered:", "ok_h": "Answered:", "cta_h": "Atendyo's AI assistant answers ALL of these, 24/7, from your own store's info.", "cta_btn": "Try free on my store", "share": "Share your result", "copy": "Copy link", "audit_another": "Audit another store", "powered": "Free audit by Atendyo · atendyo.com", "compare_h": "Compare it with a competitor", "compare_btn": "Compare", "badge_h": "Show your score on your site (badge)", }, } _GRADE_COLOR = {"A": "#16a34a", "B": "#65a30d", "C": "#ca8a04", "D": "#ea580c", "F": "#dc2626"} def _question_text(q: dict, lang: str) -> str: return str(q.get("q") or "") def render_audit_page(row: AuditResult, *, base: str = PUBLIC_BASE) -> str: lang = "en" if row.lang == "en" else "es" t = _T[lang] brand = html.escape(row.brand or row.domain) a, tot = row.answered, row.total grade = row.grade or grade_for(a, tot) color = _GRADE_COLOR.get(grade, "#dc2626") qs = row.questions or [] gaps = [q for q in qs if q.get("status") == "miss"] oks = [q for q in qs if q.get("status") == "hit"] page_url = f"{base}/audit/{row.domain}" og = f"{base}/audit/{row.domain}/og.png" def _li(items): return "".join(f"
  • {html.escape(_question_text(q, lang))}
  • " for q in items) title = t["title"].format(brand=brand, a=a, t=tot) desc = t["desc"].format(brand=brand, a=a, t=tot) try_url = f"{base}/demo" # "try the bot that answers these" -> the instant demo badge_embed = html.escape( f'Atendyo' ) return f""" {title}
    {grade}

    {t['h1'].format(brand=brand, a=a, t=tot)}

    {t['answers']} {a}/{tot}

    {t['sub']}

    {f'

    {t["gaps_h"]}

      {_li(gaps)}
    ' if gaps else ''} {f'

    {t["ok_h"]}

      {_li(oks)}
    ' if oks else ''}

    {t['cta_h']}

    {t['cta_btn']}
    """ _LANDING = { "es": { "title": "¿Qué preguntas de compra NO responde tu tienda Shopify? Compruébalo gratis (2026) | Atendyo", "desc": "Pega la URL de tu tienda Shopify y en 60 segundos te decimos cuántas de las 12 preguntas que tus clientes hacen antes de comprar responde tu web. Gratis, sin registro.", "h1": "¿Qué preguntas de tus clientes NO responde tu tienda?", "sub": "Pega la URL de tu tienda Shopify. En 60 segundos analizamos tu propia web y te decimos cuántas de las 12 preguntas clave de compra responde — y cuáles se te escapan. Gratis, sin registro.", "ph": "tutienda.com", "btn": "Auditar mi tienda gratis", "working": "Leyendo toda tu tienda y analizando las respuestas… esto puede tardar 1-2 minutos, no cierres la página.", "err": "No pudimos analizar esa tienda. Revisa la URL e inténtalo otra vez.", }, "en": { "title": "Which shopper questions does your Shopify store NOT answer? Check free (2026) | Atendyo", "desc": "Paste your Shopify store URL and in 60 seconds we tell you how many of the 12 questions customers ask before buying your site answers. Free, no signup.", "h1": "Which of your customers' questions does your store NOT answer?", "sub": "Paste your Shopify store URL. In 60 seconds we analyze your own site and tell you how many of the 12 key buying questions it answers — and which ones slip through. Free, no signup.", "ph": "yourstore.com", "btn": "Audit my store free", "working": "Reading your whole store and analyzing the answers… this can take 1-2 minutes, don't close the page.", "err": "We couldn't analyze that store. Check the URL and try again.", }, } def render_audit_landing(lang: str = "es", *, base: str = PUBLIC_BASE) -> str: lang = "en" if lang == "en" else "es" t = _LANDING[lang] return f""" {t['title']}

    {t['h1']}

    {t['sub']}

    """ def badge_svg(row: AuditResult) -> str: """A small shields-style SVG badge a store can embed ('Atendyo · X/12'). Each embed is a free backlink + distribution to that store's visitors.""" grade = row.grade or grade_for(row.answered, row.total) color = _GRADE_COLOR.get(grade, "#dc2626") val = f"{row.answered}/{row.total}" lw, vw = 64, 44 return ( f'' '' '' f'' f'' f'' '' f'Atendyo' f'{val}' '' ) def render_compare_page(a: AuditResult, b: AuditResult, *, base: str = PUBLIC_BASE) -> str: """Head-to-head: two stores side by side, per-question ✓/✗ + a winner. Rides 'X vs Y' search intent and is more shareable than a single score.""" lang = "en" if (a.lang == "en" or b.lang == "en") else "es" qs = GAP_QUESTIONS.get("en" if lang == "en" else "es", GAP_QUESTIONS["es"]) qa = {q.get("q"): q.get("status") for q in (a.questions or [])} qb = {q.get("q"): q.get("status") for q in (b.questions or [])} na, nb = html.escape(a.brand or a.domain), html.escape(b.brand or b.domain) yes, no = "✓", "✗" rows = "".join( f'{html.escape(q)}' f'{yes if qa.get(q) == "hit" else no}' f'{yes if qb.get(q) == "hit" else no}' for q in qs ) if a.answered > b.answered: winner = (f"Gana {na}" if lang == "es" else f"{na} wins") + f" ({a.answered} vs {b.answered})" elif b.answered > a.answered: winner = (f"Gana {nb}" if lang == "es" else f"{nb} wins") + f" ({b.answered} vs {a.answered})" else: winner = ("Empate" if lang == "es" else "Tie") + f" ({a.answered}/{a.total})" t = _T[lang] page_url = f"{base}/audit/{a.domain}/vs/{b.domain}" title = (f"{na} vs {nb}: ¿quién responde mejor a sus clientes? | Atendyo" if lang == "es" else f"{na} vs {nb}: who answers shoppers better? | Atendyo") head = ("Preguntas de compra que responde cada tienda" if lang == "es" else "Buyer questions each store answers") return f""" {title}
    {na}
    {nb}
    {a.answered}/{a.total}
    {b.answered}/{b.total}
    {winner}
    {rows}

    {t['cta_h']}

    {t['cta_btn']}
    """ def og_image_bytes(row: AuditResult) -> bytes: """A 1200x630 share card with the score. Bulletproof: any failure returns a minimal valid PNG instead of raising (the page must never break).""" try: from PIL import Image, ImageDraw, ImageFont grade = row.grade or grade_for(row.answered, row.total) color = _GRADE_COLOR.get(grade, "#dc2626") img = Image.new("RGB", (1200, 630), "#0f1830") d = ImageDraw.Draw(img) def font(sz: int): try: return ImageFont.truetype("DejaVuSans-Bold.ttf", sz) except Exception: # noqa: BLE001 try: return ImageFont.load_default(sz) except TypeError: return ImageFont.load_default() d.text((70, 70), "Atendyo · Auditoría de tienda", font=font(34), fill="#8aa0c8") brand = (row.brand or row.domain)[:34] d.text((70, 180), brand, font=font(64), fill="#ffffff") d.text((70, 300), f"responde {row.answered}/{row.total}", font=font(58), fill="#ffffff") d.text((70, 372), "preguntas de compra", font=font(40), fill="#cbd5e1") # grade chip + centered letter (anchor="mm" centers cleanly; fallback if # the loaded font doesn't support anchors) d.rounded_rectangle((900, 180, 1130, 410), radius=28, fill=color) gf = font(150) try: d.text((1015, 295), grade, font=gf, fill="#ffffff", anchor="mm") except (ValueError, TypeError): d.text((975, 225), grade, font=gf, fill="#ffffff") d.text((70, 520), "atendyo.com/audit · pruébalo gratis con tu tienda", font=font(30), fill="#8aa0c8") buf = io.BytesIO() img.save(buf, format="PNG") return buf.getvalue() except Exception: # noqa: BLE001 - never break the page over an image # 1x1 transparent PNG fallback import base64 return base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mbut" "FAAAAAElFTkSuQmCC" )