chemavx/researchowl
1
0
Fork 0
Code Issues Pull Requests 10 Actions Packages Projects Releases Wiki Activity
Files
3fed9c7aedf2e32de52189d6c622aa3cf700f3b4
researchowl/src/generator/generator.py
T
ChemaVXandClaude Fable 5 d31badeb5b
Build & Deploy ResearchOwl / build-and-push (push) Successful in 8s
Details
refactor(ghost): _admin_get/_admin_headers compartidos en GhostPublisher 
find_draft_by_title y fetch_published_menu duplicaban línea a línea el GET
admin de Ghost (token, URL, ClientSession, dict de headers, status check,
resp.json), y el dict Authorization/Accept-Version/Accept-Encoding estaba
copiado en 3 sitios — el incidente brotli ya demostró que el fix por copia
se deja sitios. Ahora los headers se construyen en un único punto
(_admin_headers, con SAFE_ACCEPT_ENCODING de config) y el GET en _admin_get;
publish_draft usa los mismos headers. aiohttp pasa a import de módulo en
generator.py (era 'import aiohttp as _aio' repetido dentro de dos métodos).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-05 16:00:31 +00:00

1208 lines
50 KiB
Python
Raw Blame History

"""
ResearchOwl Generators
Produces structured outputs from processed research using Claude or Ollama
"""
import base64
import hashlib
import hmac
import json
import re
import time
import aiohttp
import structlog
from src.config import settings, SAFE_ACCEPT_ENCODING
from src.llm import get_anthropic_client
from src.processor.processor import OllamaClient, ContentProcessor
from src.db.database import ResearchDB, OutputType
logger = structlog.get_logger()
PODCAST_SYSTEM = (
"Escribe SIEMPRE en español. "
"Eres un guionista de podcast. Escribe exactamente como un presentador HABLA — contracciones, "
"frases naturales, pausas, preguntas retóricas. "
"NUNCA repitas una frase o idea que ya escribiste. "
"Cada párrafo debe introducir información NUEVA. "
"Usa marcadores [PAUSA], [ÉNFASIS], [MÚSICA] con moderación."
)
BLOG_SYSTEM = (
"Escribe SIEMPRE en español. "
"Eres un periodista escribiendo un artículo de blog. Usa encabezados markdown claros. "
"NUNCA repitas el mismo dato o frase dos veces — si ya lo dijiste, avanza. "
"Cada sección debe añadir información nueva no cubierta en secciones anteriores."
)
BLOG_SYSTEM_EN = (
"You write ALWAYS in English. "
"You are a journalist writing a blog article. Use clear markdown headers. "
"NEVER repeat the same fact or phrase twice — if you said it, move on. "
"Each section must add new information not covered in other sections."
)
BLOG_PROMPT_EN = """\
Write a blog article about: "{topic}"
RULES — follow strictly:
- Each section under a heading must add NEW information not covered elsewhere
- Do NOT summarize previous sections at the start of each new section
- Do NOT repeat facts — if a fact appears once, do not mention it again
- Use concrete details, numbers, names — avoid vague generalities
- Target: 1000-1500 words
STRUCTURE:
# [Impactful headline]
[Hook paragraph — the most surprising fact]
## Background
[Context — what, when, who — only facts not covered elsewhere]
## Key Facts
[Most significant findings — each point must be distinct]
## Analysis / Significance
[What this means — without repeating the Key Facts section]
## Conclusion
[No more than 2 sentences summarizing, then a forward-looking statement]
RESEARCH MATERIAL:
{context}
Write the complete article in markdown:"""
REPORT_SYSTEM = (
"Escribe SIEMPRE en español. "
"Eres un analista de investigación. Escribe un informe estructurado y factual. "
"Sé conciso — NO rellenes con resúmenes redundantes. "
"NUNCA repitas un hallazgo ya listado. Cada hallazgo numerado debe ser distinto."
)
THREAD_SYSTEM = (
"Escribe SIEMPRE en español. "
"Escribes hilos de Twitter/X. Cada tweet debe tener menos de 280 caracteres. "
"NUNCA repitas información de un tweet anterior. "
"Cada tweet debe revelar algo NUEVO. Numéralos 1/N, 2/N..."
)
PROMPTS = {
OutputType.PODCAST: """\
Escribe un guion de podcast sobre: "{topic}"
REGLAS — sigue estrictamente:
- Escribe como PALABRA HABLADA: contracciones, ritmo natural, como si hablaras con un amigo
- NO uses encabezados formales como "SEGMENTO 1:" — fluye de forma natural
- Cada párrafo debe introducir un NUEVO hecho o ángulo — nunca repitas algo ya dicho
- Si te encuentras repitiendo, para y salta al siguiente punto nuevo
- Objetivo: 800-1200 palabras de contenido hablado real
ESTRUCTURA (usa transiciones naturales, no encabezados):
1. Gancho: abre con el hecho más sorprendente o dramático
2. Contexto: ¿cómo llegamos aquí?
3. Las evidencias o eventos clave (elige los 3 más interesantes)
4. La controversia o debate sobre el tema
5. ¿Qué significa esto / qué pasó después?
MATERIAL DE INVESTIGACIÓN:
{context}
Escribe el guion ahora (solo palabra hablada, sin acotaciones excepto [PAUSA] ocasional):""",
OutputType.BLOG: """\
Escribe un artículo de blog sobre: "{topic}"
REGLAS — sigue estrictamente:
- Cada sección bajo un encabezado debe añadir información NUEVA no cubierta en otro lugar
- NO resumas secciones anteriores al inicio de cada nueva sección
- NO repitas hechos — si un hecho aparece una vez, no lo menciones de nuevo
- Usa detalles concretos, números, nombres — evita generalidades vagas
- Objetivo: 1000-1500 palabras
ESTRUCTURA:
# [Titular impactante]
[Párrafo gancho — el hecho más sorprendente]
## Contexto
[Contexto — qué, cuándo, quién — solo hechos no cubiertos en otro lugar]
## Hechos Clave
[Los hallazgos más significativos — cada punto debe ser distinto]
## Análisis / Importancia
[Qué significa esto — sin repetir la sección de Hechos Clave]
## Conclusión
[Conclusión — no más de 2 oraciones resumiendo, luego una declaración prospectiva]
MATERIAL DE INVESTIGACIÓN:
{context}
Escribe el artículo completo en markdown:""",
OutputType.REPORT: """\
Escribe un informe de investigación sobre: "{topic}"
REGLAS — sigue estrictamente:
- Cada hallazgo numerado debe ser DISTINTO — sin contenido que se superponga
- El Resumen Ejecutivo NO debe repetir los hallazgos literalmente — solo los 2-3 puntos más críticos
- La calidad de las fuentes y contradicciones deben referenciar afirmaciones específicas, no declaraciones genéricas
- Sé preciso y conciso — sin relleno
ESTRUCTURA:
1. Resumen Ejecutivo (3-4 oraciones, solo puntos clave)
2. Hallazgos Clave (5-10 numerados, cada uno completamente distinto)
3. Análisis de Evidencia (lo que muestran las fuentes, con cualquier contradicción)
4. Cronología (si aplica — fechas/eventos específicos)
5. Conclusiones y Preguntas Abiertas
MATERIAL DE INVESTIGACIÓN:
{context}
Escribe el informe completo en markdown:""",
OutputType.THREAD: """\
Escribe un hilo de Twitter/X sobre: "{topic}"
REGLAS — sigue estrictamente:
- Cada tweet debe revelar UN nuevo hecho o idea — nunca repetir un tweet anterior
- Máximo 280 caracteres por tweet (cuenta cuidadosamente)
- Formato de numeración: 1/ 2/ 3/ ... N/
- El tweet gancho debe ser el hecho más sorprendente/provocador
- Avanza hacia una conclusión — no repitas el gancho al final
- 12-18 tweets en total
MATERIAL DE INVESTIGACIÓN:
{context}
Escribe el hilo (un tweet por línea, nada más):"""
}
OUTLINE_REPORT = """
IMPORTANTE: Los títulos y queries deben estar en español, independientemente del idioma del material de investigación.
Eres un editor de investigación. Dado el tema "{topic}" y el material
disponible, genera un outline detallado para un informe exhaustivo.
Devuelve SOLO una lista JSON de secciones, sin texto adicional, sin
markdown, sin explicaciones. Formato exacto:
[
{{"title": "Título de la sección", "query": "términos de búsqueda específicos para esta sección", "words": 800}},
...
]
Genera entre 6 y 10 secciones. Cada sección debe:
- Cubrir un ángulo distinto del tema
- Tener una query específica para recuperar chunks relevantes
- Indicar longitud objetivo en palabras (400-1200)
Material disponible (resumen):
{context_summary}
"""
OUTLINE_BLOG = """
IMPORTANTE: Los títulos y queries deben estar en español, independientemente del idioma del material de investigación.
Eres un editor de contenido. Dado el tema "{topic}" y el material
disponible, genera un outline para un artículo de blog exhaustivo.
Devuelve SOLO una lista JSON de secciones, sin texto adicional:
[
{{"title": "Título de sección", "query": "términos búsqueda", "words": 600}},
...
]
Genera entre 5 y 8 secciones. Primera sección = introducción gancho.
Última sección = conclusión con perspectiva original.
Material disponible (resumen):
{context_summary}
"""
OUTLINE_PODCAST = """
IMPORTANTE: Los títulos y queries deben estar en español, independientemente del idioma del material de investigación.
Eres un productor de podcast. Dado el tema "{topic}" y el material
disponible, genera un outline para un guion de podcast exhaustivo.
Devuelve SOLO una lista JSON de segmentos, sin texto adicional:
[
{{"title": "Nombre del segmento", "query": "términos búsqueda", "words": 700}},
...
]
Genera entre 5 y 7 segmentos. Flujo natural: gancho → contexto →
desarrollo → controversia → conclusión.
Material disponible (resumen):
{context_summary}
"""
# ─── Ghost CMS ────────────────────────────────────────────────────────────────
def _b64url(data: bytes | str) -> str:
if isinstance(data, str):
data = data.encode()
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def _extract_title(content: str) -> str:
"""Return first H1 heading from markdown, skipping the ResearchOwl header block."""
in_header = False
for line in content.splitlines():
stripped = line.strip()
if stripped == "---":
in_header = not in_header
continue
if in_header:
continue
if stripped.startswith("# ") and not stripped.startswith("## "):
return stripped[2:].strip()
return ""
def _strip_researchowl_header(content: str) -> str:
"""Remove the ---...--- metadata block that ResearchOwl prepends to outputs."""
lines = content.splitlines(keepends=True)
dashes_seen = 0
for i, line in enumerate(lines):
if line.strip() == "---":
dashes_seen += 1
if dashes_seen == 2:
return "".join(lines[i + 1:]).lstrip("\n")
return content
def _seo_checklist(slug: str) -> str:
"""Static pre-publish SEO reminder appended to the draft-published notice.
NOT a validator call: the fresh draft has no meta yet, so checking now would
be all-fail noise. Jose runs `seo-check <slug>` after filling these in."""
return (
"\n\n⚠️ Antes de publicar, añade: meta title, meta description (≤145), "
"custom excerpt, OG/Twitter, feature image + alt (campo Alt, NO Caption), "
"2-3 internal links (donde sea natural)\n"
f"Luego valida: seo-check {slug}"
)
# Language-aware default tag when no seo/tags are supplied. Fixes the historical
# hardcoded "investigacion" that mis-tagged EN posts; EN canonical is "investigation".
_DEFAULT_TAG = {"en": "investigation", "es": "investigacion"}
def _resolve_seo_mode(override: str | None = None) -> str:
"""Effective SEO autofill mode: 'off' | 'on' | 'dryrun'.
A per-call override ('dryrun'/'on', e.g. from `/generate blog en dry`) wins;
otherwise it derives from settings.seo_autofill."""
if override in ("on", "dryrun"):
return override
if settings.seo_autofill_dryrun:
return "dryrun"
if settings.seo_autofill_enabled:
return "on"
return "off"
def _seo_checklist_slim(slug: str) -> str:
"""Slimmed checklist for the autofill path: meta/OG/excerpt/tags/links are
already on the draft, so only the human-only bits remain."""
return (
"⚠️ Antes de publicar: añade *feature image + alt* (campo Alt, NO Caption), luego:\n"
f"`seo-check {slug}`"
)
def _bare_ghost_notice(ghost: "GhostPublisher", post: dict) -> str:
"""Today's exact draft-published notice (editor link + full checklist).
Used for the flag-off path and as the autofill failure fallback."""
return (
f"\n\n---\n"
f"📤 *Borrador publicado en Ghost*\n"
f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}"
f"{_seo_checklist(post.get('slug', ''))}"
)
def _seo_link_summary(inserted_pairs: list[dict], n_suggestions: int) -> str:
"""Report ACTUALLY-INSERTED links only ("phrase → /slug/"), with a separate
count of well-formed suggestions that were skipped (no verbatim body match).
Never surfaces raw rejected suggestions."""
n = len(inserted_pairs)
skipped = max(0, n_suggestions - n)
head = f"{n} insertados" + (f", {skipped} omitidos" if skipped else "")
if not inserted_pairs:
return head
body = "; ".join(f"{p['phrase']} → /{p['slug']}/" for p in inserted_pairs)
return f"{head} — {body}"
def _seo_live_message(ghost: "GhostPublisher", post: dict, seo: dict, inserted_pairs: list[dict]) -> str:
"""Separate Telegram message for the live autofill path (SEO written to draft)."""
slug = post.get("slug", "")
warns = seo.get("seo_warnings") or []
warn_line = ("\n⚠️ revisar: " + "; ".join(warns)) if warns else ""
n_sug = len(seo.get("internal_links", []))
return (
f"📤 *Borrador en Ghost — SEO autorrelleno*\n"
f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}\n\n"
f"🔖 *SEO escrito en el draft:*\n"
f"• meta title ({len(seo['meta_title'])}): {seo['meta_title']}\n"
f"• meta desc ({len(seo['meta_description'])}): {seo['meta_description']}\n"
f"• excerpt: {len(seo['custom_excerpt'])} car.\n"
f"• tags: {', '.join(seo['tags'])}\n"
f"• internal links: {_seo_link_summary(inserted_pairs, n_sug)}"
f"{warn_line}\n\n"
f"🖼️ *Imagen sugerida:*\n"
f"`imagefinder --query \"{seo['image_query']}\" --context \"{seo['image_context']}\" --lang {ghost.lang}`\n\n"
f"{_seo_checklist_slim(slug)}"
)
def _seo_dryrun_message(ghost: "GhostPublisher", post: dict, seo: dict, inserted_pairs: list[dict]) -> str:
"""Separate Telegram message for DRY-RUN: bare draft created, SEO only proposed."""
slug = post.get("slug", "")
warns = seo.get("seo_warnings") or []
warn_line = ("\n⚠️ revisar: " + "; ".join(warns)) if warns else ""
n_sug = len(seo.get("internal_links", []))
return (
f"🧪 *DRY-RUN SEO* — draft creado SIN escribir SEO (solo revisión)\n"
f"Editar: {ghost.url}/ghost/#/editor/post/{post['id']}\n\n"
f"🔖 *SEO propuesto (NO escrito en Ghost):*\n"
f"• meta title ({len(seo['meta_title'])}): {seo['meta_title']}\n"
f"• meta desc ({len(seo['meta_description'])}): {seo['meta_description']}\n"
f"• excerpt ({len(seo['custom_excerpt'])} car.): {seo['custom_excerpt']}\n"
f"• tags: {', '.join(seo['tags'])}\n"
f"• internal links que se insertarían: {_seo_link_summary(inserted_pairs, n_sug)}"
f"{warn_line}\n\n"
f"🖼️ *Imagen sugerida:*\n"
f"`imagefinder --query \"{seo['image_query']}\" --context \"{seo['image_context']}\" --lang {ghost.lang}`\n\n"
f"{_seo_checklist_slim(slug)}"
)
def _norm_title(t: str) -> str:
"""Normaliza un título para compararlo con lo que Ghost almacenó: colapsa
whitespace y trunca a 255 chars (límite de posts.title en Ghost — un H1
largo del LLM puede volver truncado/normalizado, y la igualdad exacta
fallaría justo en los casos que el guard debe cubrir)."""
return " ".join((t or "").split())[:255]
def _parse_ghost_ts(ts: str) -> float | None:
"""created_at de la Admin API (ISO 8601, p.ej. 2026-07-04T20:26:00.000+00:00
o sufijo Z) → epoch. None si no parsea."""
try:
from datetime import datetime
return datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp()
except Exception:
return None
class GhostPublisher:
def __init__(self, lang: str = "es"):
self.lang = lang
if lang == "en":
self.url = (settings.ghost_url_en or "").rstrip("/")
self.api_key = settings.ghost_api_key_en or ""
else:
self.url = (settings.ghost_url or "").rstrip("/")
self.api_key = settings.ghost_api_key or ""
def is_configured(self) -> bool:
return bool(self.url and self.api_key)
def _build_html(self, markdown_content: str) -> str:
"""Canonical markdown → Ghost body HTML conversion: strip the ResearchOwl
header, render markdown, drop the first <h1> (Ghost renders the title).
Shared so callers that need the linked HTML produce the SAME html that gets
posted (no drift between link-insertion and the mobiledoc body)."""
import markdown as _md
clean = _strip_researchowl_header(markdown_content)
html = _md.markdown(clean, extensions=["extra"])
return re.sub(r"<h1[^>]*>.*?</h1>", "", html, count=1, flags=re.DOTALL).lstrip()
def _make_token(self) -> str:
key_id, secret = self.api_key.split(":", 1)
now = int(time.time())
header = _b64url(json.dumps({"alg": "HS256", "typ": "JWT", "kid": key_id}))
payload = _b64url(json.dumps({"iat": now, "exp": now + 300, "aud": "/admin/"}))
signing = f"{header}.{payload}"
sig = _b64url(
hmac.new(bytes.fromhex(secret), signing.encode(), hashlib.sha256).digest()
)
return f"{signing}.{sig}"
def _admin_headers(self) -> dict:
"""Headers canónicos de la Admin API (token JWT fresco por llamada).
Único sitio donde se construyen: cualquier cambio (Accept-Version,
encoding) se hace aquí y cubre todas las llamadas a Ghost."""
return {
"Authorization": f"Ghost {self._make_token()}",
"Accept-Version": "v5.0",
# Nunca heredar el default de aiohttp: con un backend brotli
# instalado anunciaría br y el decode roto de aiohttp 3.14 rompe
# la lectura de respuestas (incidente 2026-07-04, KNOWN-ISSUES.md).
"Accept-Encoding": SAFE_ACCEPT_ENCODING,
}
async def _admin_get(self, query: str, timeout: float = 15) -> dict | None:
"""GET a {url}/ghost/api/admin/{query} con auth + headers canónicos.
None si non-200 (logueado); las excepciones de red/parseo suben tal
cual para que cada caller aplique su propia política best-effort."""
async with aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=timeout)) as sess:
async with sess.get(f"{self.url}/ghost/api/admin/{query}",
headers=self._admin_headers()) as resp:
if resp.status != 200:
body = await resp.text()
logger.warning("Ghost admin GET non-200", query=query[:80],
status=resp.status, body=body[:200])
return None
return await resp.json()
async def find_draft_by_title(self, title: str,
since: float | None = None) -> dict | None:
"""Busca entre los drafts recientes uno con título exacto.
Mitad de recuperación del guard anti-duplicados de publish_draft: si el
POST fue aceptado por Ghost pero la lectura de la respuesta falló
(p.ej. decode roto), el draft existe aunque el caller viera una
excepción. Con `since` (epoch) solo casan drafts creados en este
intento — un draft viejo del mismo título de una run anterior NO cuenta
como "ya publicado". La comparación usa _norm_title (whitespace
colapsado + truncado a 255) para sobrevivir la normalización de Ghost.
Best-effort: None si no hay match o si falla."""
data = await self._admin_get(
"posts/?filter=status:draft&order=created_at%20desc&limit=15"
"&fields=id,title,slug,created_at"
)
if data is None:
return None
wanted = _norm_title(title)
for p in data.get("posts", []):
if _norm_title(p.get("title") or "") != wanted:
continue
if since is not None:
created = _parse_ghost_ts(p.get("created_at") or "")
# 60s de margen por posible desfase de reloj Ghost/bot.
if created is None or created < since - 60:
continue
return p
return None
async def publish_draft(self, title: str, markdown_content: str,
tags: list[str] | None = None,
seo: dict | None = None,
body_html: str | None = None) -> dict:
"""Create a Ghost DRAFT. status is ALWAYS "draft" — never published.
Purely-additive SEO hooks (both default None → byte-identical to the
original behavior):
* body_html — pre-converted + internal-linked HTML from the caller; used
verbatim when given, else markdown_content is converted as before.
* seo — a generate_seo_fields dict; when given, its meta/OG/Twitter fields
are added to the post and its (allow-list) tags are used.
"""
# Caller-supplied linked HTML wins; otherwise convert markdown as before.
html = body_html if body_html is not None else self._build_html(markdown_content)
logger.info("Ghost publish_draft", html_length=len(html),
html_preview=html[:200], seo=bool(seo))
if not html.strip():
raise ValueError("Ghost: HTML vacío tras conversión markdown — contenido no enviado")
# Ghost 5.x (Lexical editor): el campo "html" es solo de lectura en la API.
# La forma fiable de enviar HTML arbitrario es via mobiledoc con HTML card,
# que Ghost acepta en todas las versiones de v5 y renderiza sin conversión.
mobiledoc = json.dumps({
"version": "0.3.1",
"atoms": [],
"cards": [["html", {"html": html}]],
"markups": [],
"sections": [[10, 0]],
})
# Language-aware default tag (fixes the hardcoded "investigacion" for EN).
tag_names = tags or [_DEFAULT_TAG.get(self.lang, "investigation")]
post_obj = {
"title": title,
"mobiledoc": mobiledoc,
"status": "draft", # NEVER "published" — draft only, always.
"tags": [{"name": t} for t in tag_names],
}
if seo:
post_obj.update({
"meta_title": seo["meta_title"],
"meta_description": seo["meta_description"],
"custom_excerpt": seo["custom_excerpt"],
"og_title": seo["og_title"],
"og_description": seo["og_description"],
"twitter_title": seo["twitter_title"],
"twitter_description": seo["twitter_description"],
})
body = {"posts": [post_obj]}
attempt_start = time.time()
async with aiohttp.ClientSession() as sess:
async with sess.post(
f"{self.url}/ghost/api/admin/posts/",
json=body,
headers=self._admin_headers(),
) as resp:
if resp.status not in (200, 201):
text = await resp.text()
raise ValueError(f"Ghost API {resp.status}: {text[:300]}")
# Guard anti-duplicados (incidente 2026-07-04): el 2xx confirma
# que Ghost YA creó el draft; si la lectura del body falla
# (p.ej. decode br roto), recuperamos el draft recién creado en
# vez de propagar — propagar hacía que el caller (o el usuario
# con /publish) reintentara y duplicara. El guard vive AQUÍ para
# cubrir a todos los callers, y solo se activa tras un POST
# aceptado: un fallo pre-POST jamás lo dispara.
try:
return await resp.json()
except Exception as read_err:
existing = None
try:
existing = await self.find_draft_by_title(
title, since=attempt_start)
except Exception as find_err:
logger.warning("publish_draft: draft-exists check failed",
error=str(find_err))
if existing:
logger.warning(
"publish_draft: POST aceptado pero lectura de la "
"respuesta falló; draft recuperado sin re-publicar",
post_id=existing["id"], error=str(read_err))
return {"posts": [existing]}
raise
# ─── Output generation ────────────────────────────────────────────────────────
class OutputGenerator:
def __init__(self, db: ResearchDB, ollama: OllamaClient, processor: ContentProcessor):
self.db = db
self.ollama = ollama
self.processor = processor
# Set during a blog generation when the autofill/dry-run path runs: a short
# markdown SEO summary the bot sends as a SEPARATE Telegram message (so it is
# never buried inside the long .md document). None on the flag-off path.
self.last_publish_notice: str | None = None
async def _publish_blog_to_ghost(self, lang: str, full_output: str, topic: str,
session_id: int, seo_override: str | None) -> str:
"""Publish a blog DRAFT to Ghost, gated by the SEO autofill mode.
Returns the ghost_notice to APPEND to the returned document (flag-off /
fallback bare-draft path only). For the autofill 'on'/'dryrun' paths it sets
self.last_publish_notice (a separate Telegram message) and returns "".
NEVER raises and NEVER blocks publishing: any autofill failure degrades to
today's exact bare-draft publish. status is always "draft".
"""
ghost = GhostPublisher(lang=lang)
if not ghost.is_configured():
return ""
title = _extract_title(full_output) or topic
mode = _resolve_seo_mode(seo_override)
if mode in ("on", "dryrun"):
try:
from src.seo.autofill import (
fetch_published_menu, generate_seo_fields, insert_internal_links,
)
article_md = _strip_researchowl_header(full_output)
menu = await fetch_published_menu(lang)
seo = await generate_seo_fields(
article_md, menu, lang, title=title,
db=self.db, session_id=session_id,
)
if seo is None:
raise RuntimeError("generate_seo_fields returned None")
# Build the SAME html that will be posted, then link it deterministically.
html = ghost._build_html(article_md)
linked_html, inserted_pairs = insert_internal_links(
html, seo["internal_links"], menu, lang)
if mode == "dryrun":
# Bare draft (today's write path) — do NOT write seo fields; only
# surface the proposal so Jose can inspect before trusting writes.
result = await ghost.publish_draft(title, full_output)
post = result["posts"][0]
self.last_publish_notice = _seo_dryrun_message(ghost, post, seo, inserted_pairs)
else:
result = await ghost.publish_draft(
title, full_output, tags=seo["tags"], seo=seo,
body_html=linked_html)
post = result["posts"][0]
self.last_publish_notice = _seo_live_message(ghost, post, seo, inserted_pairs)
logger.info("Auto-published blog to Ghost",
mode=mode, post_id=post["id"], links=len(inserted_pairs))
return ""
except Exception as e:
logger.warning("SEO autofill failed; falling back to bare draft",
error=str(e))
# Sin guard anti-duplicados aquí: vive DENTRO de publish_draft
# (solo se activa tras un POST aceptado). Un fallo pre-POST
# (Ollama caído, menú, links) cae aquí y DEBE publicar el
# contenido nuevo — un guard por título casaba con drafts
# viejos del mismo topic y lo descartaba en silencio.
# fall through to the bare-draft publish below
# OFF mode, or autofill failed → today's exact bare-draft publish.
try:
result = await ghost.publish_draft(title, full_output)
post = result["posts"][0]
logger.info("Auto-published blog to Ghost (bare)", post_id=post["id"])
return _bare_ghost_notice(ghost, post)
except Exception as e:
logger.warning("Auto-publish to Ghost failed", error=str(e))
return ""
async def generate(self, session_id: int, output_type: OutputType,
progress_callback=None, lang: str = "es",
seo_override: str | None = None) -> str:
"""Generate an output for a research session"""
self.last_publish_notice = None
if output_type in (OutputType.REPORT_EXTENDED,
OutputType.BLOG_EXTENDED,
OutputType.PODCAST_EXTENDED):
return await self.generate_extended(session_id, output_type, progress_callback,
lang=lang, seo_override=seo_override)
session = await self.db.get_session(session_id)
if not session:
raise ValueError(f"Session {session_id} not found")
topic = session["topic"]
logger.info("Generating output", type=output_type, topic=topic)
if progress_callback:
await progress_callback(f"🔍 Retrieving best research material for {output_type}...")
# RAG: get most relevant context for this output type
query = self._get_rag_query(output_type, topic)
context = await self.processor.rag_query(session_id, query, top_k=80)
if not context:
# Fallback: use raw top chunks
chunks = await self.db.get_top_chunks(session_id, limit=20)
context = "\n\n---\n\n".join(c["content"] for c in chunks)
if not context:
raise ValueError("No processed content available. Run /process first.")
backend = "Claude Haiku" if settings.anthropic_api_key else "Ollama"
if progress_callback:
await progress_callback(f"✍️ Generando {output_type} con {backend}... (2-5 min)")
# Build prompt
system = self._get_system(output_type)
prompt = PROMPTS[output_type].format(topic=topic, context=context)
if lang == "en" and output_type == OutputType.BLOG:
system = BLOG_SYSTEM_EN
prompt = BLOG_PROMPT_EN.format(topic=topic, context=context)
output = await self._generate(prompt, system, output_type, session_id)
# Add metadata header
stats = await self.db.get_session_stats(session_id)
header = self._build_header(topic, output_type, session, stats)
full_output = header + "\n\n" + output
# Save to DB
await self.db.save_output(session_id, output_type, full_output)
# Auto-publish to Ghost for blog outputs (autofill mode gated inside helper).
ghost_notice = ""
if output_type in (OutputType.BLOG, OutputType.BLOG_EXTENDED):
ghost_notice = await self._publish_blog_to_ghost(
lang, full_output, topic, session_id, seo_override)
logger.info("Output generated", type=output_type, length=len(full_output))
return full_output + ghost_notice
async def _generate(self, prompt: str, system: str, output_type: OutputType,
session_id: int | None = None) -> str:
if settings.anthropic_api_key:
return await self._generate_with_claude(prompt, system, output_type, session_id)
return await self._generate_with_ollama(prompt, system)
async def _generate_with_claude(self, prompt: str, system: str, output_type: OutputType,
session_id: int | None = None) -> str:
max_tokens = 4096 if output_type == OutputType.THREAD else 16000
try:
client = get_anthropic_client()
msg = await client.messages.create(
model=settings.claude_model,
max_tokens=max_tokens,
system=system,
messages=[{"role": "user", "content": prompt}],
)
if session_id is not None:
try:
await self.db.log_api_call(
session_id, "generation", settings.claude_model,
msg.usage.input_tokens, msg.usage.output_tokens
)
except Exception as log_err:
logger.warning("Failed to log API usage", error=str(log_err))
return msg.content[0].text.strip()
except Exception as e:
logger.warning("Claude generation failed, falling back to Ollama", error=str(e))
return await self._generate_with_ollama(prompt, system)
async def _generate_with_ollama(self, prompt: str, system: str) -> str:
return await self.ollama.generate(prompt, system=system, timeout=300, temperature=0.7)
def _get_rag_query(self, output_type: OutputType, topic: str) -> str:
queries = {
OutputType.PODCAST: f"{topic} story narrative facts interesting",
OutputType.BLOG: f"{topic} key facts evidence analysis",
OutputType.REPORT: f"{topic} evidence data official findings",
OutputType.THREAD: f"{topic} surprising facts shocking revelations",
}
return queries.get(output_type, topic)
def _get_system(self, output_type: OutputType) -> str:
systems = {
OutputType.PODCAST: PODCAST_SYSTEM,
OutputType.BLOG: BLOG_SYSTEM,
OutputType.REPORT: REPORT_SYSTEM,
OutputType.THREAD: THREAD_SYSTEM,
}
return systems.get(output_type, "You are a helpful research assistant.")
async def generate_extended(self, session_id: int, output_type: OutputType,
progress_callback=None, lang: str = "es",
seo_override: str | None = None) -> str:
"""
Generación por secciones para outputs exhaustivos.
1. Recupera muestra de contexto para el outline
2. Genera outline con Claude (lista de secciones)
3. Para cada sección: RAG específico → genera sección
4. Concatena y guarda
"""
session = await self.db.get_session(session_id)
if not session:
raise ValueError(f"Session {session_id} not found")
topic = session["topic"]
# Paso 1: contexto resumen para el outline (top 10 chunks)
top_chunks = await self.db.get_top_chunks(session_id, limit=10)
if not top_chunks:
raise ValueError("No processed content available. Run /process first.")
context_summary = "\n\n".join(
f"- {c.get('title', '')}: {c['content'][:300]}"
for c in top_chunks
)
if progress_callback:
await progress_callback("🗂️ Generando estructura del documento…")
# Paso 2: outline
base_type = output_type.value.replace("_extended", "")
outline_prompts = {
"report": OUTLINE_REPORT,
"blog": OUTLINE_BLOG,
"podcast": OUTLINE_PODCAST,
}
outline_prompt = outline_prompts[base_type].format(
topic=topic, context_summary=context_summary
)
outline_json = await self._generate_raw(outline_prompt, session_id)
try:
import json as _json
clean = outline_json.strip()
if clean.startswith("```"):
clean = "\n".join(clean.split("\n")[1:])
if clean.endswith("```"):
clean = "\n".join(clean.split("\n")[:-1])
sections = _json.loads(clean.strip())
except Exception as e:
logger.error("Failed to parse outline", error=str(e), raw=outline_json[:200])
raise ValueError(f"No se pudo generar el outline: {e}")
if progress_callback:
await progress_callback(
f"✍️ Generando {len(sections)} secciones… (esto tardará varios minutos)"
)
# Paso 3: generar cada sección
base_output_type = OutputType(base_type)
system = self._get_system(base_output_type)
if lang == "en" and output_type == OutputType.BLOG_EXTENDED:
system = BLOG_SYSTEM_EN
sections_text = []
for i, section in enumerate(sections, 1):
title = section.get("title", f"Sección {i}")
query = section.get("query", topic)
target_words = section.get("words", 600)
if progress_callback:
await progress_callback(
f"✍️ Sección {i}/{len(sections)}: {title[:40]}…"
)
section_context = await self.processor.rag_query(session_id, query, top_k=40)
if not section_context:
section_context = context_summary
lang_rule = "- Write in English\n" if lang == "en" else "- Escribe en español\n"
section_prompt = (
f"Escribe la sección '{title}' del {base_type} sobre: '{topic}'\n\n"
f"REGLAS:\n"
f"- NO incluyas ningún título o encabezado al inicio de tu respuesta — el título de la sección ya está incluido externamente\n"
f"- Esta es UNA sección de un documento más largo — no repitas introducción ni conclusión general\n"
f"- No incluyas encabezados del documento completo, solo el contenido de esta sección\n"
f"- Objetivo: aproximadamente {target_words} palabras\n"
f"- Usa SOLO información del material siguiente — no inventes datos\n"
f"{lang_rule}"
f"\nMATERIAL:\n{section_context}"
)
section_text = await self._generate(
section_prompt, system, base_output_type, session_id
)
sections_text.append(f"## {title}\n\n{section_text}")
# Paso 4: concatenar
full_content = "\n\n---\n\n".join(sections_text)
stats = await self.db.get_session_stats(session_id)
header = self._build_header(topic, output_type, session, stats)
full_output = header + "\n\n" + full_content
await self.db.save_output(session_id, output_type, full_output)
# Auto-publish to Ghost for extended blog outputs (autofill mode gated inside).
ghost_notice = ""
if output_type == OutputType.BLOG_EXTENDED:
ghost_notice = await self._publish_blog_to_ghost(
lang, full_output, topic, session_id, seo_override)
logger.info("Extended output generated", type=output_type,
sections=len(sections), length=len(full_output))
return full_output + ghost_notice
async def _generate_raw(self, prompt: str,
session_id: int | None = None) -> str:
if settings.anthropic_api_key:
try:
client = get_anthropic_client()
msg = await client.messages.create(
model=settings.claude_model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}],
)
if session_id is not None:
try:
await self.db.log_api_call(
session_id, "outline", settings.claude_model,
msg.usage.input_tokens, msg.usage.output_tokens
)
except Exception:
pass
return msg.content[0].text.strip()
except Exception as e:
logger.warning("Claude outline failed", error=str(e))
raise
raise ValueError("Claude API key required for extended generation")
def _build_header(self, topic: str, output_type: OutputType,
session: dict, stats: dict) -> str:
from datetime import datetime
dt = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
return f"""---
ResearchOwl | {output_type.upper()} OUTPUT
Topic: {topic}
Generated: {dt}
Sources: {stats.get('scraped', 0)} scraped | {stats.get('failed', 0)} failed
Iterations: {session.get('iterations', 0)}
Total words researched: {session.get('total_words', 0):,}
---
"""
def _remove_duplicate_headings(text: str) -> str:
lines = text.split('\n')
result = []
i = 0
while i < len(lines):
line = lines[i].rstrip()
if line.startswith('# ') and not line.startswith('## '):
h1_text = line[2:].strip().lower()
window = result[-5:] if len(result) >= 5 else result[:]
recent_h2s = {
p.strip()[3:].strip().lower()
for p in window
if p.strip().startswith('## ')
}
if h1_text in recent_h2s:
i += 1
continue
result.append(lines[i])
i += 1
return '\n'.join(result)
def generate_pdf(content: str, title: str = "ResearchOwl Output") -> bytes:
content = _remove_duplicate_headings(content)
try:
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import cm
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, HRFlowable
from reportlab.lib.enums import TA_LEFT
from reportlab.lib import colors
import io
import re
except ImportError:
raise ImportError("reportlab is required for PDF export — pip install reportlab")
buf = io.BytesIO()
doc = SimpleDocTemplate(
buf,
pagesize=A4,
rightMargin=2 * cm,
leftMargin=2 * cm,
topMargin=2.5 * cm,
bottomMargin=2 * cm,
title=title,
)
base = getSampleStyleSheet()
normal = ParagraphStyle("RO_Normal", parent=base["Normal"],
fontSize=10, leading=14, spaceAfter=4)
h1 = ParagraphStyle("RO_H1", parent=base["Heading1"],
fontSize=18, spaceBefore=12, spaceAfter=6,
textColor=colors.HexColor("#1a1a2e"))
h2 = ParagraphStyle("RO_H2", parent=base["Heading2"],
fontSize=14, spaceBefore=10, spaceAfter=4,
textColor=colors.HexColor("#16213e"))
h3 = ParagraphStyle("RO_H3", parent=base["Heading3"],
fontSize=12, spaceBefore=8, spaceAfter=4)
code_style = ParagraphStyle("RO_Code", parent=base["Code"],
fontSize=9, leading=12, fontName="Courier",
backColor=colors.HexColor("#f4f4f4"), spaceAfter=4)
bullet_style = ParagraphStyle("RO_Bullet", parent=normal,
leftIndent=20, bulletIndent=10, spaceAfter=2)
def md_to_para(text: str) -> str:
text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
text = re.sub(r'__(.+?)__', r'<b>\1</b>', text)
text = re.sub(r'\*(.+?)\*', r'<i>\1</i>', text)
text = re.sub(r'_(.+?)_', r'<i>\1</i>', text)
text = re.sub(r'`(.+?)`', r'<font name="Courier">\1</font>', text)
return text
story = []
lines = content.split("\n")
in_code = False
code_buf = []
for line in lines:
if line.startswith("```"):
if not in_code:
in_code = True
code_buf = []
else:
in_code = False
try:
story.append(Paragraph(
"<br/>".join(l.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
for l in code_buf),
code_style
))
except Exception:
pass
continue
if in_code:
code_buf.append(line)
continue
if re.match(r'^[-*_]{3,}$', line.strip()):
story.append(HRFlowable(width="100%", thickness=0.5,
color=colors.grey, spaceAfter=6))
continue
try:
if line.startswith("### "):
story.append(Paragraph(md_to_para(line[4:]), h3))
elif line.startswith("## "):
story.append(Paragraph(md_to_para(line[3:]), h2))
elif line.startswith("# "):
story.append(Paragraph(md_to_para(line[2:]), h1))
elif re.match(r'^[-*+] ', line):
story.append(Paragraph("• " + md_to_para(line[2:]), bullet_style))
elif re.match(r'^\d+\. ', line):
story.append(Paragraph(md_to_para(line), bullet_style))
elif line.strip() == "":
story.append(Spacer(1, 6))
else:
story.append(Paragraph(md_to_para(line), normal))
except Exception:
try:
story.append(Paragraph(line[:300], normal))
except Exception:
pass
doc.build(story)
return buf.getvalue()
async def generate_diff_summary(
topic: str,
new_urls: set,
old_urls: set,
new_chunks: list,
session_id: int,
db,
) -> str | None:
from src.config import settings
import structlog
diff_logger = structlog.get_logger()
added_urls = new_urls - old_urls
pct_new = len(added_urls) / max(len(new_urls), 1)
diff_logger.info("Diff analysis", topic=topic,
new_urls=len(new_urls), old_urls=len(old_urls),
added=len(added_urls), pct_new=round(pct_new, 2))
if pct_new < 0.20 and len(added_urls) < 5:
diff_logger.info("Diff: no significant new sources", topic=topic)
return None
if not new_chunks:
return None
context = "\n\n---\n\n".join(
f"[{c.get('source_type', 'web').upper()}] {c.get('title', '')}\n{c['content'][:400]}"
for c in new_chunks[:20]
)
if not settings.anthropic_api_key:
return (
f"📊 *Novedades detectadas sobre {topic}*\n\n"
f"• {len(added_urls)} fuentes nuevas encontradas\n"
f"• {len(new_chunks)} chunks de contenido procesados\n\n"
f"Usa /generate report para ver el análisis completo."
)
try:
client = get_anthropic_client()
prompt = (
f'Analiza el siguiente material de investigación sobre "{topic}" '
f'y genera un resumen BREVE (máximo 300 palabras) de las novedades '
f'más importantes encontradas. Escribe en español.\n\n'
f'Si el contenido es muy similar a investigaciones anteriores o no '
f'contiene información genuinamente nueva, responde SOLO con: '
f'"SIN_NOVEDADES"\n\n'
f'Material nuevo:\n{context}'
)
msg = await client.messages.create(
model=settings.claude_model,
max_tokens=500,
messages=[{"role": "user", "content": prompt}]
)
summary = msg.content[0].text.strip()
if summary == "SIN_NOVEDADES":
diff_logger.info("Diff: Claude found no new information", topic=topic)
return None
try:
await db.log_api_call(session_id, "diff", settings.claude_model,
msg.usage.input_tokens, msg.usage.output_tokens)
except Exception:
pass
return f"🔔 *Novedades — {topic}*\n\n{summary}\n\nUsa /generate para report completo."
except Exception as e:
diff_logger.warning("Diff summary generation failed", error=str(e))
return (
f"📊 *Actualización — {topic}*\n\n"
f"• {len(added_urls)} fuentes nuevas\n"
f"Usa /generate report para ver el análisis completo."
)
async def generate_comparison(
topic_a: str,
topic_b: str,
context_a: str,
context_b: str,
session_id_a: int,
db,
) -> str:
from src.config import settings
import structlog
cmp_logger = structlog.get_logger()
if not settings.anthropic_api_key:
raise ValueError("Claude API key required for comparison")
def _truncate(text: str, max_words: int = 3000) -> str:
words = text.split()
if len(words) > max_words:
return " ".join(words[:max_words]) + "\n\n[... contenido adicional truncado ...]"
return text
context_a = _truncate(context_a)
context_b = _truncate(context_b)
prompt = (
f'Eres un analista experto. Compara en profundidad estos dos temas:\n'
f'TEMA A: "{topic_a}"\n'
f'TEMA B: "{topic_b}"\n\n'
f'Escribe el análisis en español con esta estructura:\n\n'
f'## Resumen comparativo\n'
f'(2-3 párrafos con las diferencias y similitudes más importantes)\n\n'
f'## {topic_a}\n'
f'(Puntos clave únicos de este tema)\n\n'
f'## {topic_b}\n'
f'(Puntos clave únicos de este tema)\n\n'
f'## Similitudes\n'
f'(Qué tienen en común)\n\n'
f'## Diferencias clave\n'
f'(Tabla markdown o lista de las diferencias más relevantes)\n\n'
f'## Conclusión\n'
f'(Cuál es mejor/más relevante según el contexto, o qué conclusión se extrae)\n\n'
f'---\n'
f'MATERIAL DE INVESTIGACIÓN — {topic_a}:\n{context_a}\n\n'
f'---\n'
f'MATERIAL DE INVESTIGACIÓN — {topic_b}:\n{context_b}\n'
)
try:
client = get_anthropic_client()
msg = await client.messages.create(
model=settings.claude_model,
max_tokens=8192,
messages=[{"role": "user", "content": prompt}]
)
try:
await db.log_api_call(
session_id_a, "comparison", settings.claude_model,
msg.usage.input_tokens, msg.usage.output_tokens
)
except Exception:
pass
return msg.content[0].text.strip()
except Exception as e:
cmp_logger.error("Comparison generation failed", error=str(e))
raise
Reference in New Issue View Git Blame Copy Permalink