docs_builder.py

"""Build the Google Doc report and share it (anyone-with-link = Commenter).

The Docs API cannot set headings at create time, so we create a blank doc named
"{Keyword}-{GEO}" and then send one batchUpdate. Text is appended with a running
cursor; paragraph styles and link runs reference the just-inserted ranges.

Index math runs in UTF-16 code units — that is the unit the Docs API speaks.
Python's `len()` counts code points, so a single emoji (🎰, 🎲, 💰) is 1 code
point but 2 UTF-16 units; using `len()` for ranges silently shifts every
subsequent paragraph's start/endIndex when an emoji slips into competitor data.
`_utf16_len` keeps the bookkeeping aligned with what Google actually sees.
"""
from __future__ import annotations

import logging

from models import Competitor, GeneratedMeta, SiteStructure

log = logging.getLogger(__name__)


def _utf16_len(text: str) -> int:
    """Length of `text` measured in UTF-16 code units — what the Docs API uses
    to address ranges. Equals `len(text)` for the BMP, but counts non-BMP
    characters (emoji, many flags, certain CJK) as 2."""
    if not text:
        return 0
    return len(text.encode("utf-16-le")) // 2


class _Builder:
    """Accumulates Docs API requests while tracking the insertion index in
    UTF-16 code units (DOCS-001)."""

    def __init__(self):
        self.index = 1
        self.requests: list[dict] = []

    def para(self, text: str, style: str | None = None, link: str | None = None) -> None:
        self.requests.append(
            {"insertText": {"location": {"index": self.index}, "text": text + "\n"}}
        )
        start = self.index
        end = start + _utf16_len(text)  # newline (1 UTF-16 unit) sits at `end`
        if text and style:
            self.requests.append({
                "updateParagraphStyle": {
                    "range": {"startIndex": start, "endIndex": end + 1},
                    "paragraphStyle": {"namedStyleType": style},
                    "fields": "namedStyleType",
                }
            })
        if text and link:
            self.requests.append({
                "updateTextStyle": {
                    "range": {"startIndex": start, "endIndex": end},
                    "textStyle": {"link": {"url": link}},
                    "fields": "link",
                }
            })
        self.index = end + 1


def _trim(text: str, limit: int = 90) -> str:
    text = " ".join((text or "").split())
    return text if len(text) <= limit else text[: limit - 1] + "…"


def _structure_lines(s: SiteStructure) -> list[str]:
    lines: list[str] = []
    if s.headings:
        shown = " | ".join(f"H{lvl} {_trim(txt, 60)}" for lvl, txt in s.headings[:12])
        extra = f" (+{len(s.headings) - 12} more)" if len(s.headings) > 12 else ""
        lines.append(f"• Headings: {shown}{extra}")
    if s.nav:
        lines.append(f"• Navigation: {_trim(', '.join(s.nav[:15]), 300)}")
    if s.sections:
        lines.append(f"• Sections: {_trim(', '.join(s.sections[:15]), 300)}")
    if s.schema_types:
        lines.append(f"• Schema.org types: {', '.join(s.schema_types[:15])}")
    if s.sitemap_urls:
        sample = ", ".join(s.sitemap_urls[:5])
        lines.append(f"• Sitemap: {len(s.sitemap_urls)} URLs (e.g. {_trim(sample, 300)})")
    return lines or ["• (no structure extracted)"]


def _build_requests(keyword, geo, competitors, meta, failures) -> list[dict]:
    b = _Builder()
    b.para(f"Analysis for {keyword} - {geo}", style="HEADING_1")

    b.para("Competitor Reports", style="HEADING_2")
    for c in competitors:
        b.para(f"#{c.position} {c.domain}", style="HEADING_3", link=c.url)
        b.para(f"URL: {c.url}", link=c.url)
        b.para(f"H1: {c.h1}")
        b.para(f"Meta Title: {c.meta_title}")
        b.para(f"Meta Description: {c.meta_description}")
        b.para("Site structure:")
        for line in _structure_lines(c.structure):
            b.para(line)

    if failures:
        b.para("Failures (slots not filled within TOP-10)", style="HEADING_3")
        for label, reason in failures:
            b.para(f"• {label}: {reason}")

    b.para("Optimized SEO Content", style="HEADING_2")
    b.para(f"H1: {meta.h1}")
    b.para(f"Meta Title: {meta.meta_title}")
    b.para(f"Meta Description: {meta.meta_description}")

    b.para("")  # trailing blank so no styled paragraph is the document's last
    return b.requests


def _share_commenter(drive, doc_id: str) -> None:
    drive.permissions().create(
        fileId=doc_id,
        body={"type": "anyone", "role": "commenter"},
        fields="id",
    ).execute()


def create_skeleton(keyword, geo, docs, drive) -> tuple[str, str]:
    """Create the empty Doc + share it, return `(doc_id, url)`.

    Splitting the build into skeleton+populate is what lets us publish the
    Doc URL into the sheet's Result column the moment the row starts
    processing — while SERP / scrape / generation are still running and
    Status is still 'pending'. An observer can click the URL and watch the
    content arrive when `populate` runs at the end of the row.

    Sharing is best-effort here (same reasoning as in `build`): a sharing
    failure logs a warning but still returns the URL so the caller can put
    it in the sheet and the operator can re-share manually."""
    doc_id = docs.documents().create(body={"title": f"{keyword}-{geo}"}).execute()["documentId"]
    try:
        _share_commenter(drive, doc_id)
    except Exception as err:
        log.warning(
            "Doc %s skeleton created but Drive sharing failed (%s: %s); "
            "share manually from the link once the row completes",
            doc_id, type(err).__name__, err,
        )
    return doc_id, f"https://docs.google.com/document/d/{doc_id}/edit"


def populate(doc_id: str, keyword, geo, competitors, meta, failures, docs) -> None:
    """Fill a previously-created skeleton Doc with the full report.
    The skeleton body is empty, so the batchUpdate inserts everything
    starting at index 1 — same logic as the old monolithic `build`."""
    requests = _build_requests(keyword, geo, competitors, meta, failures)
    docs.documents().batchUpdate(documentId=doc_id, body={"requests": requests}).execute()


def build(keyword, geo, competitors, meta, failures, docs, drive) -> str:
    """Single-shot wrapper: skeleton + populate. Kept for callers (and tests)
    that don't need the early-publish split."""
    doc_id, url = create_skeleton(keyword, geo, docs, drive)
    populate(doc_id, keyword, geo, competitors, meta, failures, docs)
    return url