pipeline/writer.py

"""Step 5: write the blog post from the extracted source material.

Output is Markdown with:
  # H1 title, ## H2 sections, normal paragraphs, and
  [IMAGE: <scene description>] markers where illustrations belong.
"""
from __future__ import annotations

import re
from typing import List

from huggingface_hub import InferenceClient

from . import config, llm
from .aeo_guidelines import AEO_GUIDELINES
from .factual_accuracy import FACTUAL_ACCURACY_GUIDELINES

# Content-goal directives shape the post's stance and tone. One is selected in the UI.
GOAL_GUIDANCE = {
    "Informational": (
        "Content goal: INFORMATIONAL. Prioritize clear, objective, comprehensive "
        "explanation. Teach the reader, define terms, stay neutral and factual, and favor "
        "accuracy and completeness over opinion or salesmanship."
    ),
    "Persuasive": (
        "Content goal: PERSUASIVE. Build a convincing case for a clear position or action. "
        "Lead with benefits, address likely objections and counterpoints, support claims "
        "with concrete evidence and examples, and close with a motivating call to action — "
        "without overstating or making unsupported claims."
    ),
    "Authoritative": (
        "Content goal: AUTHORITATIVE. Write as the definitive reference on the topic. Be "
        "precise and thorough, back every key claim with specific evidence, data or primary "
        "sources, cover edge cases and limitations, and use a confident expert tone that "
        "earns citation and trust."
    ),
    "Thought Leadership": (
        "Content goal: THOUGHT LEADERSHIP. Offer an original, forward-looking perspective. "
        "Frame the topic within broader industry trends, share a distinctive point of view "
        "or informed prediction, challenge conventional assumptions where warranted, and "
        "support opinions with clear reasoning and evidence."
    ),
}
DEFAULT_GOAL = "Informational"


def _goal_directive(content_goal: str) -> str:
    return GOAL_GUIDANCE.get(content_goal, GOAL_GUIDANCE[DEFAULT_GOAL])


def _build_system(target_wordcount: int, content_goal: str) -> str:
    return (
        "You are an expert blog writer and SEO/AEO editor. Using ONLY the supplied source "
        "material as factual grounding, write an original, engaging, well-structured blog "
        "post. Do not copy sentences verbatim from the sources; synthesize in your own voice.\n\n"
        f"{_goal_directive(content_goal)}\n\n"
        "Structural requirements:\n"
        "- Start with a single '# ' H1 title that includes the primary keyword.\n"
        "- Use the primary keyword naturally in the first 100 words and in at least one '## ' heading.\n"
        "- Weave the secondary keyword in naturally 1-3 times.\n"
        "- Use '## ' subheadings to organize sections; write substantive paragraphs.\n"
        "- Honor the brief's angle, audience, and tone.\n"
        f"- Target length: about {target_wordcount} words (stay within roughly ±10%).\n"
        f"- Insert exactly {config.N_IMAGES} image markers of the form "
        "'[IMAGE: a vivid visual scene to illustrate this section]' at natural points "
        "(never two in a row, not in the title).\n"
        "- End with a short conclusion.\n\n"
        "AI-citation (AEO) requirements:\n"
        f"{AEO_GUIDELINES}\n"
        f"{FACTUAL_ACCURACY_GUIDELINES}\n"
        "Return Markdown only — no preamble, no code fences. Use Markdown pipe tables for "
        "any tables."
    )


def write_post(
    client: InferenceClient,
    topic: str,
    primary_keyword: str,
    secondary_keyword: str,
    brief: str,
    sources: List[dict],
    target_wordcount: int = config.DEFAULT_WORD_COUNT,
    content_goal: str = DEFAULT_GOAL,
) -> str:
    source_block = _format_sources(sources)
    user = (
        f"Topic: {topic}\n"
        f"Primary keyword: {primary_keyword}\n"
        f"Secondary keyword: {secondary_keyword}\n"
        f"Content goal: {content_goal}\n"
        f"Target word count: {target_wordcount}\n"
        f"Brief: {brief}\n\n"
        f"SOURCE MATERIAL (ranked by domain authority):\n{source_block}\n\n"
        f"Write the full ~{target_wordcount}-word blog post in Markdown now, matching the "
        f"'{content_goal}' content goal and following the AI-citation guidelines (direct "
        "answer up top, question-style headings, a comparison/feature table, an FAQ, cited "
        "sources, stated limitations)."
    )
    # Allow enough output tokens for the requested length (~1.6 tokens/word + overhead).
    max_tokens = max(1500, min(int(target_wordcount * 2) + 600, 8000))
    md = llm.chat(
        client,
        config.MODEL_WRITER,
        _build_system(target_wordcount, content_goal),
        user,
        max_tokens=max_tokens,
        temperature=0.7,
        fallback_model=config.MODEL_WRITER_FALLBACK,
    )
    md = _strip_fences(md)
    return _ensure_image_markers(md)


def _format_sources(sources: List[dict]) -> str:
    parts = []
    for i, s in enumerate(sources, 1):
        pr = s.get("page_rank")
        pr_str = f"PR={pr}" if pr is not None else "PR=n/a"
        parts.append(
            f"[Source {i}] {s.get('title', '')} ({s.get('domain', '')}, {pr_str})\n"
            f"URL: {s.get('url', '')}\n"
            f"{s.get('text', '')}\n"
        )
    return "\n".join(parts)


def _strip_fences(md: str) -> str:
    md = md.strip()
    if md.startswith("```"):
        md = re.sub(r"^```[a-zA-Z]*\n", "", md)
        md = re.sub(r"\n```$", "", md)
    return md.strip()


def _ensure_image_markers(md: str) -> str:
    """Guarantee at least one image marker so the illustration step has work to do."""
    if re.search(r"\[IMAGE:", md):
        return md
    # inject one marker after the first paragraph following the H1
    lines = md.splitlines()
    out, injected = [], False
    for idx, line in enumerate(lines):
        out.append(line)
        if not injected and idx > 0 and line.strip() == "" and lines[idx - 1].strip() and not lines[idx - 1].startswith("#"):
            out.append("[IMAGE: a compelling hero image illustrating the topic]")
            out.append("")
            injected = True
    if not injected:
        out.append("\n[IMAGE: a compelling hero image illustrating the topic]")
    return "\n".join(out)


def parse_image_markers(md: str) -> List[str]:
    """Return the scene descriptions from all [IMAGE: ...] markers, in order."""
    return [m.strip() for m in re.findall(r"\[IMAGE:\s*(.+?)\]", md, re.DOTALL)]