Add grounded blog-post generator pipeline

Docker Space bundling self-hosted SearXNG + a Gradio app that turns
topic/keywords/brief into an illustrated .docx blog post:
search-term LLM -> SearXNG -> OpenPageRank ranking -> content extraction
-> blog writing -> FLUX.1-schnell images -> Qwen2.5-VL captions -> docx.
All paid inference is billed to the user's HF token entered in the UI.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.gitattributes

@@ -33,3 +33,10 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+
+# Force LF for scripts/text so the Docker (Linux) build runs correctly
+*.sh text eol=lf
+*.py text eol=lf
+*.yml text eol=lf
+Dockerfile text eol=lf
+start.sh text eol=lf

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.pyc
+.cache/
+data/
+*.docx

Dockerfile

@@ -0,0 +1,48 @@
+# Blog Post Generator — HF Docker Space
+# Bundles a self-hosted SearXNG (internal, 127.0.0.1:8080) + a Gradio app (public, 7860).
+FROM searxng/searxng:latest
+
+USER root
+
+# --- system deps for trafilatura/lxml/Pillow (Alpine base) ---
+RUN apk add --no-cache \
+        bash \
+        libxml2 \
+        libxslt \
+        jpeg \
+        zlib \
+        build-base \
+        libxml2-dev \
+        libxslt-dev \
+        jpeg-dev \
+        zlib-dev
+
+# SearXNG installs into this virtualenv; reuse it for the app so we get one Python.
+ENV VENV=/usr/local/searxng/searx-pyenv
+ENV PATH="$VENV/bin:$PATH"
+ENV SEARXNG_SETTINGS_PATH=/etc/searxng/settings.yml
+# HF persistent storage mounts at /data; config.py falls back if absent.
+ENV DATA_DIR=/data
+ENV SEARXNG_URL=http://127.0.0.1:8080
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN "$VENV/bin/pip" install --no-cache-dir -r requirements.txt
+
+# App source
+COPY app.py start.sh ./
+COPY pipeline ./pipeline
+COPY searxng/settings.yml /etc/searxng/settings.yml
+
+# HF Spaces run the container as UID 1000 — make everything that gets written writable.
+RUN chmod +x /app/start.sh \
+    && mkdir -p /data /app/.cache \
+    && chgrp -R 0 /app /etc/searxng /usr/local/searxng /data \
+    && chmod -R g+rwX /app /etc/searxng /usr/local/searxng /data
+
+EXPOSE 7860
+
+USER 1000
+
+ENTRYPOINT ["/app/start.sh"]

README.md

@@ -4,8 +4,54 @@ emoji: ⚡
 colorFrom: yellow
 colorTo: blue
 sdk: docker
+app_port: 7860
 pinned: false
-short_description: Writes Blog POsts
+short_description: Illustrated blog-post generator, exported as .docx
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Blog Post Generator
+
+Turn a **topic + keywords + brief** into a fully-written, illustrated blog post exported
+as a `.docx` file. The pipeline is *grounded*: instead of writing blind, it runs real web
+search, ranks the sources by authority, and writes from the most authoritative pages.
+
+## Pipeline
+
+1. **Search-term reasoning** — an LLM turns your topic/keywords/brief into optimized queries.
+2. **Web search** — a self-hosted [SearXNG](https://github.com/searxng/searxng) instance
+   (running inside this Space) returns the top ~25 result domains.
+3. **Authority ranking** — [OpenPageRank](https://www.domcop.com/openpagerank/documentation)
+   scores those domains; the top 5 pages are selected.
+4. **Content extraction** — main article text is extracted from those pages.
+5. **Writing** — a strong LLM writes an SEO-aware post from that material, weaving your
+   keywords in and marking where images belong.
+6. **Illustration** — image prompts are written by an LLM and rendered with
+   **FLUX.1 [schnell]**.
+7. **Captioning** — a vision model captions each generated image.
+8. **Export** — everything is assembled into a downloadable `.docx`.
+
+## Billing
+
+All *paid* AI calls (reasoning, writing, image generation, captioning) run through
+**Hugging Face Inference Providers using the HF token you enter in the UI**, so they are
+billed to **you**. Nothing is stored server-side.
+
+Get a token at <https://huggingface.co/settings/tokens/new?ownUserPermissions=inference.serverless.write&tokenType=fineGrained>
+(needs *"Make calls to Inference Providers"* permission).
+
+## Configuration (Space owner)
+
+- **Secret `OPR_API_KEY`** — a free OpenPageRank API key
+  (<https://www.domcop.com/openpagerank/auth/signup>). Required for authority ranking.
+- **Persistent storage** *(optional)* — if enabled, `/data` is used to cache search
+  results, scores, images and docx between runs. Without it the app still works; the cache
+  just lives in a local directory and is not persisted.
+
+## Models (fixed defaults)
+
+| Step | Model |
+| --- | --- |
+| Search terms & image prompts | `openai/gpt-oss-120b` |
+| Blog writing | `deepseek-ai/DeepSeek-V3-0324` (fallback `Qwen/Qwen2.5-72B-Instruct`) |
+| Image generation | `black-forest-labs/FLUX.1-schnell` |
+| Image captioning | `Qwen/Qwen2.5-VL-72B-Instruct` |

app.py

@@ -0,0 +1,113 @@
+"""Gradio UI for the Blog Post Generator HF Space.
+
+Inputs: topic, primary/secondary keyword, brief, and the user's HF token (password).
+All paid AI calls are billed to that token. Outputs a live status log, a Markdown
+preview, an image gallery, and a downloadable .docx.
+"""
+from __future__ import annotations
+
+import re
+import traceback
+
+import gradio as gr
+
+from pipeline import config, orchestrator
+
+INTRO = """
+# 📝 Blog Post Generator
+Turn a topic + keywords + brief into a fully-written, **illustrated** blog post exported as `.docx`.
+
+The pipeline: **search-term reasoning → self-hosted SearXNG web search → OpenPageRank
+authority ranking → content extraction → LLM writing → FLUX.1-schnell images → vision
+captioning → DOCX export.**
+
+> All paid AI calls run through **Hugging Face Inference Providers using the token you
+> enter below**, so they are billed to **you**. The token is used only for this run and
+> is not stored.
+"""
+
+
+def _preview(markdown: str, images: list) -> str:
+    """Replace [IMAGE:] markers with their caption text for a readable Markdown preview."""
+    imgs = [im for im in images if im.get("path")]
+    it = iter(imgs)
+
+    def repl(_m):
+        im = next(it, None)
+        cap = (im or {}).get("caption") or (im or {}).get("scene") or "image"
+        return f"\n> 🖼️ *{cap}*\n"
+
+    return re.sub(r"\[IMAGE:\s*.+?\]", repl, markdown, flags=re.DOTALL)
+
+
+def generate(topic, primary, secondary, brief, hf_token, progress=gr.Progress()):
+    log_lines = []
+    gallery, docx_file, preview = [], None, ""
+
+    def status():
+        return "\n".join(f"- {ln}" for ln in log_lines)
+
+    try:
+        for frac, message, result in orchestrator.run(
+            hf_token, topic, primary, secondary, brief
+        ):
+            progress(frac, desc=message)
+            log_lines.append(message)
+
+            if result.get("images"):
+                gallery = [
+                    (im["path"], im.get("caption") or im.get("scene", ""))
+                    for im in result["images"]
+                    if im.get("path")
+                ]
+            if result.get("markdown"):
+                preview = _preview(result["markdown"], result.get("images", []))
+            if result.get("docx_path"):
+                docx_file = result["docx_path"]
+
+            yield status(), preview, gallery, docx_file
+    except Exception as e:  # noqa: BLE001 - show the user a clean error
+        log_lines.append(f"❌ **Error:** {e}")
+        traceback.print_exc()
+        yield status(), preview, gallery, docx_file
+
+
+def build_ui() -> gr.Blocks:
+    with gr.Blocks(title="Blog Post Generator", theme=gr.themes.Soft()) as demo:
+        gr.Markdown(INTRO)
+        with gr.Row():
+            with gr.Column(scale=1):
+                topic = gr.Textbox(label="Topic", placeholder="e.g. Home composting for beginners")
+                primary = gr.Textbox(label="Primary keyword", placeholder="e.g. home composting")
+                secondary = gr.Textbox(label="Secondary keyword", placeholder="e.g. kitchen scraps")
+                brief = gr.Textbox(
+                    label="Brief",
+                    lines=5,
+                    placeholder="Audience, angle, tone, must-cover points…",
+                )
+                hf_token = gr.Textbox(
+                    label="Hugging Face token (billed to you)",
+                    type="password",
+                    placeholder="hf_...",
+                )
+                run_btn = gr.Button("Generate blog post", variant="primary")
+            with gr.Column(scale=2):
+                status_box = gr.Markdown(label="Progress")
+                docx_out = gr.File(label="Download .docx")
+                gallery = gr.Gallery(label="Generated images", columns=2, height=320)
+                preview = gr.Markdown(label="Preview")
+
+        run_btn.click(
+            fn=generate,
+            inputs=[topic, primary, secondary, brief, hf_token],
+            outputs=[status_box, preview, gallery, docx_out],
+        )
+        gr.Markdown(
+            f"Models: writer `{config.MODEL_WRITER}` · reasoning `{config.MODEL_REASONING}` · "
+            f"images `{config.MODEL_IMAGE}` · vision `{config.MODEL_VISION}`."
+        )
+    return demo
+
+
+if __name__ == "__main__":
+    build_ui().queue().launch(server_name="0.0.0.0", server_port=7860)

pipeline/__init__.py

@@ -0,0 +1 @@
+"""Blog Post Generator pipeline package."""

pipeline/cache.py

@@ -0,0 +1,43 @@
+"""Tiny keyed cache under DATA_DIR for JSON-serializable pipeline artifacts.
+
+Used to avoid re-running search / ranking / extraction for identical inputs. Image
+and docx binaries are written directly under OUT_DIR by their own modules; this
+handles the lightweight JSON steps.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any, Optional
+
+from . import config
+
+
+def run_key(*parts: str) -> str:
+    """Stable short hash identifying a run from its inputs."""
+    joined = " ".join(p or "" for p in parts)
+    h = hashlib.sha256(joined.encode("utf-8"))
+    return h.hexdigest()[:16]
+
+
+def _path(key: str, name: str):
+    return config.CACHE_DIR / f"{key}.{name}.json"
+
+
+def get(key: str, name: str) -> Optional[Any]:
+    p = _path(key, name)
+    if p.exists():
+        try:
+            return json.loads(p.read_text(encoding="utf-8"))
+        except Exception:
+            return None
+    return None
+
+
+def put(key: str, name: str, value: Any) -> None:
+    p = _path(key, name)
+    try:
+        p.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
+    except Exception:
+        # Cache is best-effort; never fail the pipeline over it.
+        pass

pipeline/captions.py

@@ -0,0 +1,52 @@
+"""Step 7: caption each generated image with a vision-language model."""
+from __future__ import annotations
+
+import base64
+import mimetypes
+from pathlib import Path
+from typing import List
+
+from huggingface_hub import InferenceClient
+
+from . import config
+
+_CAPTION_PROMPT = (
+    "Write a concise, engaging one-sentence caption for this blog illustration. "
+    "Describe what is shown; do not start with 'This image' or 'A picture of'. "
+    "Return only the caption."
+)
+
+
+def _data_uri(path: Path) -> str:
+    mime = mimetypes.guess_type(str(path))[0] or "image/png"
+    b64 = base64.b64encode(path.read_bytes()).decode("utf-8")
+    return f"data:{mime};base64,{b64}"
+
+
+def caption_images(client: InferenceClient, images: List[dict]) -> List[dict]:
+    """Add a 'caption' key to each image dict that has a valid 'path'."""
+    for item in images:
+        path = item.get("path")
+        if not path or not Path(path).exists():
+            item["caption"] = ""
+            continue
+        try:
+            resp = client.chat.completions.create(
+                model=config.MODEL_VISION,
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {"type": "text", "text": _CAPTION_PROMPT},
+                            {"type": "image_url", "image_url": {"url": _data_uri(Path(path))}},
+                        ],
+                    }
+                ],
+                max_tokens=80,
+                temperature=0.5,
+            )
+            item["caption"] = (resp.choices[0].message.content or "").strip().strip('"')
+        except Exception:
+            # fall back to the scene description if captioning fails
+            item["caption"] = item.get("scene", "")
+    return images

pipeline/config.py

@@ -0,0 +1,54 @@
+"""Central configuration: model ids, storage paths, and pipeline constants.
+
+Everything tunable lives here so models/limits can be swapped in one place.
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+# --- Models (all served via HF Inference Providers, billed to the user's token) ---
+# Fast, capable general model used for reasoning-style tasks (search terms, image prompts).
+MODEL_REASONING = os.environ.get("MODEL_REASONING", "openai/gpt-oss-120b")
+# Strong long-form writer for the actual blog post, with a fallback if unavailable.
+MODEL_WRITER = os.environ.get("MODEL_WRITER", "deepseek-ai/DeepSeek-V3-0324")
+MODEL_WRITER_FALLBACK = os.environ.get("MODEL_WRITER_FALLBACK", "Qwen/Qwen2.5-72B-Instruct")
+# Text-to-image model (as requested).
+MODEL_IMAGE = os.environ.get("MODEL_IMAGE", "black-forest-labs/FLUX.1-schnell")
+# Vision-language model for captioning generated images.
+MODEL_VISION = os.environ.get("MODEL_VISION", "Qwen/Qwen2.5-VL-72B-Instruct")
+
+# --- External services ---
+SEARXNG_URL = os.environ.get("SEARXNG_URL", "http://127.0.0.1:8080")
+OPR_API_KEY = os.environ.get("OPR_API_KEY", "")
+OPR_ENDPOINT = "https://openpagerank.com/api/v1.0/getPageRank"
+
+# --- Pipeline constants ---
+N_SEARCH_TERMS = int(os.environ.get("N_SEARCH_TERMS", "5"))   # queries generated by the LLM
+TOP_N = int(os.environ.get("TOP_N", "25"))                    # results ranked by OpenPageRank
+TOP_K = int(os.environ.get("TOP_K", "5"))                     # top pages used as source material
+N_IMAGES = int(os.environ.get("N_IMAGES", "3"))              # illustrations per post
+SOURCE_CHAR_CAP = int(os.environ.get("SOURCE_CHAR_CAP", "4000"))  # chars kept per source page
+HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "20"))     # seconds for outbound HTTP
+
+
+def _resolve_data_dir() -> Path:
+    """Prefer HF persistent storage (/data); fall back to a local dir if not writable."""
+    candidate = Path(os.environ.get("DATA_DIR", "/data"))
+    try:
+        candidate.mkdir(parents=True, exist_ok=True)
+        probe = candidate / ".write_test"
+        probe.write_text("ok", encoding="utf-8")
+        probe.unlink()
+        return candidate
+    except Exception:
+        fallback = Path(__file__).resolve().parent.parent / ".cache"
+        fallback.mkdir(parents=True, exist_ok=True)
+        return fallback
+
+
+DATA_DIR = _resolve_data_dir()
+CACHE_DIR = DATA_DIR / "cache"
+OUT_DIR = DATA_DIR / "out"
+CACHE_DIR.mkdir(parents=True, exist_ok=True)
+OUT_DIR.mkdir(parents=True, exist_ok=True)

pipeline/docx_builder.py

@@ -0,0 +1,116 @@
+"""Step 8: assemble the Markdown post + images + captions into a .docx file."""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import List
+
+from docx import Document
+from docx.enum.text import WD_ALIGN_PARAGRAPH
+from docx.shared import Inches, Pt, RGBColor
+
+_IMAGE_MARKER = re.compile(r"\[IMAGE:\s*.+?\]", re.DOTALL)
+_INLINE = re.compile(r"(\*\*.+?\*\*|\*.+?\*|`.+?`)")
+_MAX_IMG_WIDTH = Inches(6.0)
+
+
+def build_docx(markdown: str, images: List[dict], out_path: Path) -> Path:
+    """Render `markdown` to a Word document, inserting `images` at [IMAGE:] markers in order."""
+    doc = Document()
+    img_iter = iter([im for im in images if im.get("path")])
+
+    for block in _split_blocks(markdown):
+        if _IMAGE_MARKER.fullmatch(block.strip()):
+            _insert_next_image(doc, img_iter)
+            continue
+        # a block may still contain an inline marker mixed with text
+        if _IMAGE_MARKER.search(block):
+            for piece in _IMAGE_MARKER.split(block):
+                if piece.strip():
+                    _render_line(doc, piece.strip())
+            _insert_next_image(doc, img_iter)
+            continue
+        _render_line(doc, block)
+
+    # any leftover images that never got placed → append at the end
+    for im in img_iter:
+        _add_picture(doc, im)
+
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    doc.save(out_path)
+    return out_path
+
+
+def _split_blocks(md: str) -> List[str]:
+    """Split into logical blocks (headings/paragraphs/markers), preserving order."""
+    blocks: List[str] = []
+    for raw in md.split("\n"):
+        line = raw.rstrip()
+        if line.strip() == "":
+            continue
+        blocks.append(line)
+    return blocks
+
+
+def _render_line(doc: Document, line: str) -> None:
+    stripped = line.strip()
+    if stripped.startswith("#"):
+        level = len(stripped) - len(stripped.lstrip("#"))
+        text = stripped[level:].strip()
+        doc.add_heading(text, level=min(max(level, 1), 4))
+    elif re.match(r"^[-*]\s+", stripped):
+        text = re.sub(r"^[-*]\s+", "", stripped)
+        p = doc.add_paragraph(style="List Bullet")
+        _add_runs(p, text)
+    elif re.match(r"^\d+\.\s+", stripped):
+        text = re.sub(r"^\d+\.\s+", "", stripped)
+        p = doc.add_paragraph(style="List Number")
+        _add_runs(p, text)
+    elif stripped.startswith(">"):
+        p = doc.add_paragraph(style="Intense Quote")
+        _add_runs(p, stripped.lstrip("> ").strip())
+    else:
+        p = doc.add_paragraph()
+        _add_runs(p, stripped)
+
+
+def _add_runs(paragraph, text: str) -> None:
+    """Add text to a paragraph, honoring **bold**, *italic*, and `code`."""
+    for token in _INLINE.split(text):
+        if not token:
+            continue
+        if token.startswith("**") and token.endswith("**"):
+            paragraph.add_run(token[2:-2]).bold = True
+        elif token.startswith("*") and token.endswith("*"):
+            paragraph.add_run(token[1:-1]).italic = True
+        elif token.startswith("`") and token.endswith("`"):
+            run = paragraph.add_run(token[1:-1])
+            run.font.name = "Consolas"
+        else:
+            paragraph.add_run(token)
+
+
+def _insert_next_image(doc: Document, img_iter) -> None:
+    im = next(img_iter, None)
+    if im is not None:
+        _add_picture(doc, im)
+
+
+def _add_picture(doc: Document, im: dict) -> None:
+    path = im.get("path")
+    if not path or not Path(path).exists():
+        return
+    try:
+        p = doc.add_paragraph()
+        p.alignment = WD_ALIGN_PARAGRAPH.CENTER
+        p.add_run().add_picture(str(path), width=_MAX_IMG_WIDTH)
+    except Exception:
+        return
+    caption = (im.get("caption") or "").strip()
+    if caption:
+        cap = doc.add_paragraph()
+        cap.alignment = WD_ALIGN_PARAGRAPH.CENTER
+        run = cap.add_run(caption)
+        run.italic = True
+        run.font.size = Pt(9)
+        run.font.color.rgb = RGBColor(0x66, 0x66, 0x66)

pipeline/extract.py

@@ -0,0 +1,47 @@
+"""Step 4: fetch the top-ranked pages and extract clean article text."""
+from __future__ import annotations
+
+from typing import List
+
+import trafilatura
+
+from . import config
+
+
+def extract_sources(results: List[dict]) -> List[dict]:
+    """For each result, download + extract main text (capped). Skips pages that fail.
+
+    Adds 'text' to each result dict and returns only those with usable content.
+    """
+    sources: List[dict] = []
+    for r in results:
+        url = r.get("url")
+        if not url:
+            continue
+        text = _extract_one(url)
+        if not text:
+            # fall back to the search snippet so the source still contributes something
+            text = (r.get("snippet") or "").strip()
+        if not text:
+            continue
+        r = dict(r)
+        r["text"] = text[: config.SOURCE_CHAR_CAP]
+        sources.append(r)
+    return sources
+
+
+def _extract_one(url: str) -> str:
+    try:
+        downloaded = trafilatura.fetch_url(url)
+        if not downloaded:
+            return ""
+        text = trafilatura.extract(
+            downloaded,
+            include_comments=False,
+            include_tables=False,
+            no_fallback=False,
+            favor_precision=True,
+        )
+        return (text or "").strip()
+    except Exception:
+        return ""

pipeline/images.py

@@ -0,0 +1,56 @@
+"""Step 6: turn each [IMAGE: ...] marker into a FLUX prompt and render it."""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List
+
+from huggingface_hub import InferenceClient
+
+from . import config, llm
+
+_PROMPT_SYSTEM = (
+    "You are a prompt engineer for the FLUX text-to-image model. Given a short scene "
+    "description for a blog illustration and the article topic, write ONE vivid, concrete "
+    "image prompt (single line, <60 words). Describe subject, setting, composition, "
+    "lighting and style. Prefer clean, editorial, photographic or tasteful illustrative "
+    "styles suitable for a professional blog. No text/words in the image. Return only the prompt."
+)
+
+
+def _flux_prompt(client: InferenceClient, topic: str, scene: str) -> str:
+    try:
+        p = llm.chat(
+            client,
+            config.MODEL_REASONING,
+            _PROMPT_SYSTEM,
+            f"Article topic: {topic}\nScene: {scene}\nWrite the FLUX prompt.",
+            max_tokens=150,
+            temperature=0.8,
+        )
+        p = p.strip().strip('"')
+        return p or scene
+    except Exception:
+        return f"{scene}, editorial photography, clean composition, natural lighting"
+
+
+def generate_images(
+    client: InferenceClient,
+    topic: str,
+    scenes: List[str],
+    run_dir: Path,
+) -> List[dict]:
+    """Render one image per scene. Returns [{scene, prompt, path|None, error?}]."""
+    run_dir.mkdir(parents=True, exist_ok=True)
+    out: List[dict] = []
+    for i, scene in enumerate(scenes):
+        prompt = _flux_prompt(client, topic, scene)
+        item = {"scene": scene, "prompt": prompt, "path": None}
+        try:
+            image = client.text_to_image(prompt=prompt, model=config.MODEL_IMAGE)
+            path = run_dir / f"image_{i + 1}.png"
+            image.save(path)
+            item["path"] = str(path)
+        except Exception as e:  # noqa: BLE001 - one bad image shouldn't kill the run
+            item["error"] = str(e)
+        out.append(item)
+    return out

pipeline/llm.py

@@ -0,0 +1,53 @@
+"""Shared helpers for talking to HF Inference Providers with the user's token.
+
+Every call here is billed to whoever owns `hf_token`. A single InferenceClient is
+built per request in app.py and threaded through the pipeline.
+"""
+from __future__ import annotations
+
+from typing import List, Optional
+
+from huggingface_hub import InferenceClient
+
+
+def make_client(hf_token: str) -> InferenceClient:
+    """Build an InferenceClient bound to the user's token (auto provider selection)."""
+    token = (hf_token or "").strip()
+    if not token:
+        raise ValueError("A Hugging Face token is required (paid calls are billed to it).")
+    return InferenceClient(token=token)
+
+
+def chat(
+    client: InferenceClient,
+    model: str,
+    system: str,
+    user: str,
+    *,
+    max_tokens: int = 1024,
+    temperature: float = 0.7,
+    fallback_model: Optional[str] = None,
+) -> str:
+    """Run a chat completion and return the assistant text.
+
+    Falls back to `fallback_model` once if the primary model errors (e.g. no provider).
+    """
+    messages = [
+        {"role": "system", "content": system},
+        {"role": "user", "content": user},
+    ]
+    models: List[str] = [model] + ([fallback_model] if fallback_model else [])
+    last_err: Optional[Exception] = None
+    for m in models:
+        try:
+            resp = client.chat.completions.create(
+                model=m,
+                messages=messages,
+                max_tokens=max_tokens,
+                temperature=temperature,
+            )
+            return (resp.choices[0].message.content or "").strip()
+        except Exception as e:  # noqa: BLE001 - surface a clean error after trying fallback
+            last_err = e
+            continue
+    raise RuntimeError(f"LLM call failed for {models}: {last_err}")

pipeline/openpagerank.py

@@ -0,0 +1,76 @@
+"""Step 3: rank the candidate domains by authority using OpenPageRank.
+
+API: GET https://openpagerank.com/api/v1.0/getPageRank
+Auth: header  API-OPR: <key>
+Params: repeated domains[]=<domain>  (up to 100 per call)
+Response: response[] with page_rank_decimal / page_rank_integer / rank per domain.
+Docs: https://www.domcop.com/openpagerank/documentation
+"""
+from __future__ import annotations
+
+from typing import Dict, List
+
+import requests
+
+from . import config
+
+
+def _chunks(items: List[str], size: int = 100):
+    for i in range(0, len(items), size):
+        yield items[i : i + size]
+
+
+def fetch_pageranks(domains: List[str]) -> Dict[str, float]:
+    """Return {domain: page_rank_decimal}. Missing/unknown domains map to 0.0.
+
+    Raises RuntimeError if the API key is missing so the caller can surface it.
+    """
+    if not config.OPR_API_KEY:
+        raise RuntimeError(
+            "OPR_API_KEY is not set. The Space owner must add an OpenPageRank API key "
+            "(free at domcop.com) as a Space secret."
+        )
+    scores: Dict[str, float] = {d: 0.0 for d in domains}
+    headers = {"API-OPR": config.OPR_API_KEY}
+    for chunk in _chunks(domains, 100):
+        params = [("domains[]", d) for d in chunk]
+        try:
+            r = requests.get(
+                config.OPR_ENDPOINT,
+                params=params,
+                headers=headers,
+                timeout=config.HTTP_TIMEOUT,
+            )
+            r.raise_for_status()
+            payload = r.json()
+        except Exception as e:  # noqa: BLE001
+            raise RuntimeError(f"OpenPageRank request failed: {e}") from e
+
+        for item in payload.get("response", []) or []:
+            dom = (item.get("domain") or "").lower()
+            if dom not in scores:
+                continue
+            try:
+                scores[dom] = float(item.get("page_rank_decimal") or 0.0)
+            except (TypeError, ValueError):
+                scores[dom] = 0.0
+    return scores
+
+
+def rank_results(results: List[dict], top_k: int = config.TOP_K) -> List[dict]:
+    """Attach OpenPageRank scores to SearXNG results and return the top-K by authority.
+
+    Falls back to the incoming (SearXNG) order if the API is unavailable.
+    """
+    domains = [r["domain"] for r in results]
+    try:
+        scores = fetch_pageranks(domains)
+        for r in results:
+            r["page_rank"] = scores.get(r["domain"], 0.0)
+        ranked = sorted(results, key=lambda x: x.get("page_rank", 0.0), reverse=True)
+    except RuntimeError:
+        # Preserve SearXNG order; mark scores as unknown.
+        for r in results:
+            r.setdefault("page_rank", None)
+        ranked = results
+    return ranked[:top_k]

pipeline/orchestrator.py

@@ -0,0 +1,100 @@
+"""End-to-end orchestration of the blog-post pipeline.
+
+Kept UI-agnostic: `run()` is a generator that yields (progress_fraction, message,
+partial_result) tuples so any front-end can drive a progress bar. The final yield
+carries the complete result dict.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Iterator, Tuple
+
+from . import (
+    cache,
+    captions,
+    config,
+    extract,
+    images,
+    llm,
+    openpagerank,
+    search_terms,
+    searxng_client,
+    writer,
+)
+
+
+def run(
+    hf_token: str,
+    topic: str,
+    primary_keyword: str,
+    secondary_keyword: str,
+    brief: str,
+) -> Iterator[Tuple[float, str, dict]]:
+    if not topic.strip():
+        raise ValueError("Please enter a topic.")
+    client = llm.make_client(hf_token)  # raises if token missing → billed to user
+
+    key = cache.run_key(topic, primary_keyword, secondary_keyword, brief)
+    run_dir = config.OUT_DIR / key
+    run_dir.mkdir(parents=True, exist_ok=True)
+    result: dict = {"key": key}
+
+    # 1) search terms
+    yield 0.05, "Generating search terms…", result
+    terms = cache.get(key, "terms") or search_terms.generate_search_terms(
+        client, topic, primary_keyword, secondary_keyword, brief
+    )
+    cache.put(key, "terms", terms)
+    result["terms"] = terms
+
+    # 2) SearXNG search
+    yield 0.15, f"Searching the web for {len(terms)} queries…", result
+    candidates = cache.get(key, "candidates") or searxng_client.search(terms, config.TOP_N)
+    if not candidates:
+        raise RuntimeError(
+            "No search results from SearXNG. Is the SearXNG service running "
+            "(check the Space logs)?"
+        )
+    cache.put(key, "candidates", candidates)
+    result["candidates"] = candidates
+
+    # 3) OpenPageRank authority ranking
+    yield 0.30, f"Ranking {len(candidates)} domains by OpenPageRank…", result
+    top = openpagerank.rank_results(candidates, config.TOP_K)
+    cache.put(key, "top", top)
+    result["top"] = top
+
+    # 4) extract source material
+    yield 0.40, f"Extracting content from top {len(top)} pages…", result
+    sources = extract.extract_sources(top)
+    if not sources:
+        raise RuntimeError("Could not extract readable content from the top pages.")
+    result["sources"] = [{k: v for k, v in s.items() if k != "text"} for s in sources]
+
+    # 5) write the post
+    yield 0.55, "Writing the blog post…", result
+    markdown = writer.write_post(
+        client, topic, primary_keyword, secondary_keyword, brief, sources
+    )
+    (run_dir / "post.md").write_text(markdown, encoding="utf-8")
+    result["markdown"] = markdown
+
+    # 6) generate images
+    scenes = writer.parse_image_markers(markdown)
+    yield 0.70, f"Generating {len(scenes)} images with FLUX.1-schnell…", result
+    imgs = images.generate_images(client, topic, scenes, run_dir)
+    result["images"] = imgs
+
+    # 7) caption images
+    yield 0.85, "Captioning images…", result
+    imgs = captions.caption_images(client, imgs)
+    result["images"] = imgs
+
+    # 8) build docx
+    yield 0.95, "Building the .docx file…", result
+    from . import docx_builder
+
+    docx_path = docx_builder.build_docx(markdown, imgs, run_dir / "blog.docx")
+    result["docx_path"] = str(docx_path)
+
+    yield 1.0, "Done.", result

pipeline/search_terms.py

@@ -0,0 +1,70 @@
+"""Step 1: LLM turns the topic/keywords/brief into optimized web-search queries."""
+from __future__ import annotations
+
+import json
+import re
+from typing import List
+
+from huggingface_hub import InferenceClient
+
+from . import config, llm
+
+_SYSTEM = (
+    "You are an SEO research assistant. Given a blog topic, its target keywords, and a "
+    "brief, produce a small set of high-signal web search queries that will surface the "
+    "most authoritative, information-rich sources to write the post from. Vary angle and "
+    "specificity. Return ONLY a JSON array of query strings, nothing else."
+)
+
+
+def _fallback_terms(topic: str, primary: str, secondary: str) -> List[str]:
+    base = [t for t in [topic, primary, secondary] if t]
+    extra = [f"{topic} {primary}".strip(), f"{topic} guide", f"{topic} best practices"]
+    seen, out = set(), []
+    for t in base + extra:
+        t = t.strip()
+        if t and t.lower() not in seen:
+            seen.add(t.lower())
+            out.append(t)
+    return out[: config.N_SEARCH_TERMS]
+
+
+def generate_search_terms(
+    client: InferenceClient,
+    topic: str,
+    primary_keyword: str,
+    secondary_keyword: str,
+    brief: str,
+) -> List[str]:
+    user = (
+        f"Topic: {topic}\n"
+        f"Primary keyword: {primary_keyword}\n"
+        f"Secondary keyword: {secondary_keyword}\n"
+        f"Brief: {brief}\n\n"
+        f"Return {config.N_SEARCH_TERMS} search queries as a JSON array."
+    )
+    try:
+        raw = llm.chat(
+            client, config.MODEL_REASONING, _SYSTEM, user,
+            max_tokens=400, temperature=0.4,
+        )
+        terms = _parse_terms(raw)
+        if terms:
+            return terms[: config.N_SEARCH_TERMS]
+    except Exception:
+        pass
+    return _fallback_terms(topic, primary_keyword, secondary_keyword)
+
+
+def _parse_terms(raw: str) -> List[str]:
+    """Extract a list of strings from a (possibly fenced) LLM response."""
+    match = re.search(r"\[.*\]", raw, re.DOTALL)
+    if match:
+        try:
+            data = json.loads(match.group(0))
+            return [str(t).strip() for t in data if str(t).strip()]
+        except Exception:
+            pass
+    # line-based fallback
+    lines = [re.sub(r'^[\s\-\*\d\.\)"]+', "", ln).strip().strip('"') for ln in raw.splitlines()]
+    return [ln for ln in lines if ln]

pipeline/searxng_client.py

@@ -0,0 +1,74 @@
+"""Step 2: query the in-container SearXNG JSON API and aggregate results by domain."""
+from __future__ import annotations
+
+from typing import Dict, List
+from urllib.parse import urlparse
+
+import requests
+import tldextract
+
+from . import config
+
+
+def registrable_domain(url: str) -> str:
+    """Return the registrable domain (e.g. 'blog.example.co.uk' -> 'example.co.uk')."""
+    ext = tldextract.extract(url)
+    if ext.domain and ext.suffix:
+        return f"{ext.domain}.{ext.suffix}".lower()
+    return (urlparse(url).netloc or url).lower()
+
+
+def _search_one(term: str) -> List[dict]:
+    params = {"q": term, "format": "json", "safesearch": "1", "language": "en"}
+    try:
+        r = requests.get(
+            f"{config.SEARXNG_URL}/search",
+            params=params,
+            timeout=config.HTTP_TIMEOUT,
+            headers={"User-Agent": "BlogPostGenerator/1.0"},
+        )
+        r.raise_for_status()
+        return r.json().get("results", []) or []
+    except Exception:
+        return []
+
+
+def search(terms: List[str], top_n: int = config.TOP_N) -> List[dict]:
+    """Run all queries, aggregate scores per domain, and return the top-N domains.
+
+    Each returned item: {domain, url, title, snippet, score}. One representative URL
+    (the highest-scoring) is kept per registrable domain so OpenPageRank ranks domains.
+    """
+    by_domain: Dict[str, dict] = {}
+    for term in terms:
+        for res in _search_one(term):
+            url = res.get("url")
+            if not url or not url.startswith("http"):
+                continue
+            dom = registrable_domain(url)
+            if not dom:
+                continue
+            score = float(res.get("score") or 0.0)
+            entry = by_domain.get(dom)
+            if entry is None:
+                by_domain[dom] = {
+                    "domain": dom,
+                    "url": url,
+                    "title": res.get("title") or "",
+                    "snippet": res.get("content") or "",
+                    "score": score,
+                    "_best": score,  # highest single-result score seen for this domain
+                }
+            else:
+                entry["score"] += score
+                # keep the single best-scoring page as the domain's representative URL
+                if score > entry["_best"]:
+                    entry["_best"] = score
+                    entry["url"] = url
+                    entry["title"] = res.get("title") or entry["title"]
+                    entry["snippet"] = res.get("content") or entry["snippet"]
+
+    ranked = sorted(by_domain.values(), key=lambda x: x["score"], reverse=True)
+    for e in ranked:
+        e.pop("_best", None)
+    return ranked[:top_n]

pipeline/writer.py

@@ -0,0 +1,105 @@
+"""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
+
+_SYSTEM = (
+    "You are an expert blog writer and SEO 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"
+    "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"- 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"
+    "Return Markdown only — no preamble, no code fences."
+)
+
+
+def write_post(
+    client: InferenceClient,
+    topic: str,
+    primary_keyword: str,
+    secondary_keyword: str,
+    brief: str,
+    sources: List[dict],
+) -> str:
+    source_block = _format_sources(sources)
+    user = (
+        f"Topic: {topic}\n"
+        f"Primary keyword: {primary_keyword}\n"
+        f"Secondary keyword: {secondary_keyword}\n"
+        f"Brief: {brief}\n\n"
+        f"SOURCE MATERIAL (ranked by domain authority):\n{source_block}\n\n"
+        "Write the full blog post in Markdown now."
+    )
+    md = llm.chat(
+        client,
+        config.MODEL_WRITER,
+        _SYSTEM,
+        user,
+        max_tokens=3500,
+        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)]

requirements.txt

@@ -0,0 +1,8 @@
+gradio==4.44.1
+huggingface_hub>=0.28.0
+requests>=2.32.0
+trafilatura>=1.12.0
+python-docx>=1.1.2
+Pillow>=10.4.0
+PyYAML>=6.0.2
+tldextract>=5.1.0

searxng/settings.yml

@@ -0,0 +1,34 @@
+# Minimal SearXNG override — inherits all engine defaults, changes only what we need.
+# See https://docs.searxng.org/admin/settings/index.html
+use_default_settings: true
+
+general:
+  instance_name: "Blog Post Generator Search"
+  # keep it internal; no public UI needed
+  enable_metrics: false
+
+server:
+  # replaced at container start if left as the placeholder
+  secret_key: "ultrasecretkey_change_me"
+  bind_address: "127.0.0.1"
+  port: 8080
+  # single trusted caller (our own app on localhost) — no bot limiter
+  limiter: false
+  public_instance: false
+  image_proxy: false
+
+search:
+  # JSON is required for our client; keep html for manual debugging
+  formats:
+    - html
+    - json
+  # be resilient: don't fail the whole query if one engine errors
+  autocomplete: ""
+  default_lang: "en"
+
+ui:
+  static_use_hash: true
+
+# Redis is not available in this single-container setup; SearXNG runs without it.
+redis:
+  url: false

start.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+# Launch SearXNG (internal) then the Gradio app (public). SearXNG runs in the
+# background bound to 127.0.0.1:8080; the app talks to it over localhost.
+set -euo pipefail
+
+SEARXNG_HOST="127.0.0.1"
+SEARXNG_PORT="8080"
+
+# A settings.yml secret_key must be set. If the shipped one is still the
+# placeholder, generate one at runtime (settings dir is made writable in the image).
+SETTINGS="${SEARXNG_SETTINGS_PATH:-/etc/searxng/settings.yml}"
+if grep -q 'ultrasecretkey_change_me' "$SETTINGS" 2>/dev/null; then
+    KEY="$(python -c 'import secrets; print(secrets.token_hex(32))')"
+    # portable in-place edit (BusyBox sed supports -i)
+    sed -i "s/ultrasecretkey_change_me/${KEY}/" "$SETTINGS" || true
+fi
+
+echo "[start] launching SearXNG on ${SEARXNG_HOST}:${SEARXNG_PORT} ..."
+python -m searx.webapp &
+SEARXNG_PID=$!
+
+# Wait for SearXNG to answer before starting the app (max ~30s).
+for i in $(seq 1 30); do
+    if python - <<'PY' 2>/dev/null; then
+import socket, sys
+s = socket.socket()
+s.settimeout(1)
+try:
+    s.connect(("127.0.0.1", 8080))
+    sys.exit(0)
+except Exception:
+    sys.exit(1)
+finally:
+    s.close()
+PY
+        echo "[start] SearXNG is up."
+        break
+    fi
+    if ! kill -0 "$SEARXNG_PID" 2>/dev/null; then
+        echo "[start] SearXNG process exited early; continuing (search will be degraded)."
+        break
+    fi
+    echo "[start] waiting for SearXNG ($i/30) ..."
+    sleep 1
+done
+
+echo "[start] launching Gradio app on 0.0.0.0:7860 ..."
+exec python /app/app.py