Wire Phase 3 test block pipeline into automated optimization flow
When a ClickUp task with a URL arrives at create_content, route it to the new optimization pipeline instead of the outline-gate Phase 1/Phase 2 flow. The pipeline runs 8 steps via the execution brain (scrape, deficit analysis, entity filtering, template writing, test block generation, readability rewrite, validation, surgical instructions doc) and uploads deliverables (test_block.html, optimization_instructions.md, validation_report.json) directly to ClickUp as attachments. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>fix/customer-field-migration
parent
236b64c11c
commit
813dd4cb01
|
|
@ -20,6 +20,7 @@ log = logging.getLogger(__name__)
|
|||
_ROOT_DIR = Path(__file__).resolve().parent.parent.parent
|
||||
_DATA_DIR = _ROOT_DIR / "data"
|
||||
_LOCAL_CONTENT_DIR = _DATA_DIR / "generated" / "content"
|
||||
_SCRIPTS_DIR = _ROOT_DIR / ".claude" / "skills" / "content-researcher" / "scripts"
|
||||
|
||||
EXEC_TOOLS = "Bash,Read,Edit,Write,Glob,Grep,WebSearch,WebFetch"
|
||||
|
||||
|
|
@ -351,6 +352,369 @@ def _build_phase2_prompt(
|
|||
return "\n".join(parts)
|
||||
|
||||
|
||||
def _build_optimization_prompt(
|
||||
url: str,
|
||||
keyword: str,
|
||||
cora_path: str,
|
||||
work_dir: str,
|
||||
scripts_dir: str,
|
||||
is_service_page: bool = False,
|
||||
capabilities_default: str = "",
|
||||
) -> str:
|
||||
"""Build the execution brain prompt for the Phase 3 optimization pipeline.
|
||||
|
||||
Produces 8 sequential steps that scrape the existing page, run deficit
|
||||
analysis, generate a test block, and create an optimization instructions
|
||||
document. All script commands use absolute paths so the CLI can execute
|
||||
them without any skill context.
|
||||
"""
|
||||
parts = [
|
||||
f"You are running an automated on-page optimization pipeline for "
|
||||
f"'{keyword}' on {url}.\n\n"
|
||||
f"Working directory: {work_dir}\n"
|
||||
f"Cora report: {cora_path}\n"
|
||||
f"Scripts directory: {scripts_dir}\n\n"
|
||||
f"Execute the following steps IN ORDER. Each step depends on the "
|
||||
f"previous step's output files. Do NOT skip steps.\n",
|
||||
|
||||
# Step 1 — Scrape existing page
|
||||
f"\n## Step 1 — Scrape Existing Page\n\n"
|
||||
f"Run the competitor scraper to fetch the current page content:\n\n"
|
||||
f"```bash\n"
|
||||
f'uv run --with requests,beautifulsoup4 python "{scripts_dir}/competitor_scraper.py" '
|
||||
f'"{url}" --output-dir "{work_dir}" --format text\n'
|
||||
f"```\n\n"
|
||||
f"This produces `existing_content.md` (or a text file named after the URL) "
|
||||
f"in the working directory. If the output file is not named `existing_content.md`, "
|
||||
f"rename it to `existing_content.md`.",
|
||||
|
||||
# Step 2 — Deficit analysis
|
||||
f"\n## Step 2 — Test Block Prep (Deficit Analysis)\n\n"
|
||||
f"Run the deficit analysis against the Cora report:\n\n"
|
||||
f"```bash\n"
|
||||
f'cd "{scripts_dir}" && uv run --with openpyxl python test_block_prep.py '
|
||||
f'"{work_dir}/existing_content.md" "{cora_path}" --format json '
|
||||
f'> "{work_dir}/prep_data.json"\n'
|
||||
f"```\n\n"
|
||||
f"This produces `prep_data.json` with word count deficits, missing entities, "
|
||||
f"density targets, and template generation instructions.",
|
||||
|
||||
# Step 3 — Filter entities (LLM step)
|
||||
f"\n## Step 3 — Filter Missing Entities for Topical Relevance\n\n"
|
||||
f'Read `{work_dir}/prep_data.json` and extract the `missing_entities` list. '
|
||||
f"Filter this list to keep ONLY entities that are topically relevant to "
|
||||
f"'{keyword}' and the page content. Remove generic/off-topic entities.\n\n"
|
||||
f"Write one entity per line to `{work_dir}/filtered_entities.txt`.\n\n"
|
||||
f"Be aggressive about filtering — only keep entities that a subject-matter "
|
||||
f"expert would expect to see on a page about '{keyword}'.",
|
||||
|
||||
# Step 4 — Write templates (LLM step)
|
||||
f"\n## Step 4 — Write Heading + Body Templates\n\n"
|
||||
f"Using the deficit data from `{work_dir}/prep_data.json` and the filtered "
|
||||
f"entities from `{work_dir}/filtered_entities.txt`, write:\n\n"
|
||||
f"1. H2 and H3 headings that incorporate target entities\n"
|
||||
f"2. Body sentence templates with `{{N}}` placeholder slots where entity "
|
||||
f"terms will be inserted programmatically\n\n"
|
||||
f"Format: Each template is a heading line followed by body sentences. "
|
||||
f"Each body sentence should have 1-3 `{{N}}` slots (numbered sequentially "
|
||||
f"starting from 1 within each sentence).\n\n"
|
||||
f"Write the output to `{work_dir}/templates.txt`.\n\n"
|
||||
f"Example format:\n"
|
||||
f"```\n"
|
||||
f"## Heading About {{1}} and {{2}}\n"
|
||||
f"Sentence with {{1}} integrated naturally. Another point about {{2}} "
|
||||
f"that provides value.\n"
|
||||
f"```",
|
||||
|
||||
# Step 5 — Generate test block (script)
|
||||
f"\n## Step 5 — Generate Test Block\n\n"
|
||||
f"Run the test block generator to fill template slots and produce the "
|
||||
f"HTML test block:\n\n"
|
||||
f"```bash\n"
|
||||
f'cd "{scripts_dir}" && uv run --with openpyxl python test_block_generator.py '
|
||||
f'"{work_dir}/templates.txt" "{work_dir}/prep_data.json" "{cora_path}" '
|
||||
f'--entities-file "{work_dir}/filtered_entities.txt" '
|
||||
f'--output-dir "{work_dir}"\n'
|
||||
f"```\n\n"
|
||||
f"This produces `test_block.md`, `test_block.html`, and `test_block_stats.json` "
|
||||
f"in the working directory.",
|
||||
|
||||
# Step 6 — Rewrite for readability (LLM step)
|
||||
f"\n## Step 6 — Rewrite Body Sentences for Readability\n\n"
|
||||
f"Read `{work_dir}/test_block.md`. Rewrite each body sentence to improve "
|
||||
f"readability and natural flow while preserving:\n"
|
||||
f"- ALL entity strings exactly as they appear (do not paraphrase entity terms)\n"
|
||||
f"- The overall heading structure\n"
|
||||
f"- The `<!-- HIDDEN TEST BLOCK -->` markers\n\n"
|
||||
f"Write the improved version back to `{work_dir}/test_block.md`.\n"
|
||||
f"Then regenerate the HTML version at `{work_dir}/test_block.html` with the "
|
||||
f"content wrapped in `<div style=\"display:none\">` tags.",
|
||||
|
||||
# Step 7 — Validate (script)
|
||||
f"\n## Step 7 — Validate Test Block\n\n"
|
||||
f"Run the before/after validation:\n\n"
|
||||
f"```bash\n"
|
||||
f'cd "{scripts_dir}" && uv run --with openpyxl python test_block_validate.py '
|
||||
f'"{work_dir}/existing_content.md" "{work_dir}/test_block.md" "{cora_path}" '
|
||||
f'--format json --output "{work_dir}/validation_report.json"\n'
|
||||
f"```\n\n"
|
||||
f"This produces `validation_report.json` with before/after metrics comparison.",
|
||||
|
||||
# Step 8 — Generate optimization instructions (LLM step)
|
||||
f"\n## Step 8 — Generate Optimization Instructions\n\n"
|
||||
f"Read the following files:\n"
|
||||
f"- `{work_dir}/existing_content.md` (current page)\n"
|
||||
f"- `{work_dir}/prep_data.json` (deficit analysis)\n"
|
||||
f"- `{work_dir}/validation_report.json` (before/after metrics)\n"
|
||||
f"- `{work_dir}/test_block.md` (generated test block)\n\n"
|
||||
f"Generate `{work_dir}/optimization_instructions.md` — a surgical playbook "
|
||||
f"for the human editor with these sections:\n\n"
|
||||
f"1. **Executive Summary** — one-paragraph overview of optimization opportunity\n"
|
||||
f"2. **Heading Changes** — specific H1/H2/H3 modifications with before→after\n"
|
||||
f"3. **Sections to Expand** — which sections need more content and what to add\n"
|
||||
f"4. **Entity Integration Points** — exact locations to weave in missing entities\n"
|
||||
f"5. **Meta Tag Updates** — title tag and meta description recommendations\n"
|
||||
f"6. **Content Gaps** — topics covered by competitors but missing from this page\n"
|
||||
f"7. **Priority Ranking** — rank all changes by expected SEO impact (high/medium/low)\n\n"
|
||||
f"Be specific and actionable. Reference exact headings and paragraphs from "
|
||||
f"the existing content. Do NOT rewrite the full page — this is a surgical guide.",
|
||||
]
|
||||
|
||||
if is_service_page:
|
||||
parts.append(
|
||||
f'\nNOTE: This is a **service page**. Company capabilities: '
|
||||
f'"{capabilities_default}"\n'
|
||||
f"Do NOT make specific claims about services, certifications, or "
|
||||
f"licenses not found on the existing page."
|
||||
)
|
||||
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Optimization pipeline (Phase 3 — test block + surgical instructions)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run_optimization(
|
||||
*,
|
||||
agent,
|
||||
config,
|
||||
ctx: dict | None,
|
||||
task_id: str,
|
||||
url: str,
|
||||
keyword: str,
|
||||
cora_path: str,
|
||||
is_service_page: bool = False,
|
||||
capabilities_default: str = "",
|
||||
) -> str:
|
||||
"""Run the Phase 3 optimization pipeline.
|
||||
|
||||
Requires a Cora report. Creates an isolated working directory, calls the
|
||||
execution brain with the 8-step optimization prompt, then finalizes by
|
||||
collecting deliverables and syncing ClickUp.
|
||||
"""
|
||||
if not cora_path:
|
||||
msg = (
|
||||
f"Error: No Cora report found for keyword '{keyword}'. "
|
||||
f"A Cora report is required for the optimization pipeline. "
|
||||
f"Please upload a Cora .xlsx report to the content Cora inbox."
|
||||
)
|
||||
log.error(msg)
|
||||
if task_id:
|
||||
_sync_clickup_fail(ctx, task_id, msg)
|
||||
return msg
|
||||
|
||||
slug = _slugify(keyword) or "unknown"
|
||||
work_dir = _LOCAL_CONTENT_DIR / slug / f"optimization-{task_id or 'manual'}"
|
||||
work_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
scripts_dir = str(_SCRIPTS_DIR)
|
||||
|
||||
# ClickUp: move to automation underway
|
||||
if task_id:
|
||||
_sync_clickup_start(ctx, task_id)
|
||||
|
||||
prompt = _build_optimization_prompt(
|
||||
url=url,
|
||||
keyword=keyword,
|
||||
cora_path=cora_path,
|
||||
work_dir=str(work_dir),
|
||||
scripts_dir=scripts_dir,
|
||||
is_service_page=is_service_page,
|
||||
capabilities_default=capabilities_default,
|
||||
)
|
||||
|
||||
log.info(
|
||||
"Optimization pipeline — running for '%s' (%s), work_dir=%s",
|
||||
keyword, url, work_dir,
|
||||
)
|
||||
try:
|
||||
exec_result = agent.execute_task(
|
||||
prompt,
|
||||
tools=EXEC_TOOLS,
|
||||
skip_permissions=True,
|
||||
)
|
||||
except Exception as e:
|
||||
error_msg = f"Optimization pipeline execution failed: {e}"
|
||||
log.error(error_msg)
|
||||
if task_id:
|
||||
_sync_clickup_fail(ctx, task_id, str(e))
|
||||
return f"Error: {error_msg}"
|
||||
|
||||
if exec_result.startswith("Error:"):
|
||||
if task_id:
|
||||
_sync_clickup_fail(ctx, task_id, exec_result)
|
||||
return exec_result
|
||||
|
||||
return _finalize_optimization(
|
||||
ctx=ctx,
|
||||
config=config,
|
||||
task_id=task_id,
|
||||
keyword=keyword,
|
||||
url=url,
|
||||
work_dir=work_dir,
|
||||
exec_result=exec_result,
|
||||
)
|
||||
|
||||
|
||||
def _finalize_optimization(
|
||||
*,
|
||||
ctx: dict | None,
|
||||
config,
|
||||
task_id: str,
|
||||
keyword: str,
|
||||
url: str,
|
||||
work_dir: Path,
|
||||
exec_result: str,
|
||||
) -> str:
|
||||
"""Collect deliverables from the working directory and sync ClickUp.
|
||||
|
||||
Required files: test_block.html, optimization_instructions.md.
|
||||
Optional: validation_report.json.
|
||||
"""
|
||||
required = ["test_block.html", "optimization_instructions.md"]
|
||||
missing = [f for f in required if not (work_dir / f).exists()]
|
||||
if missing:
|
||||
error_msg = (
|
||||
f"Optimization pipeline finished but required deliverables are "
|
||||
f"missing: {', '.join(missing)}. Working directory: {work_dir}"
|
||||
)
|
||||
log.error(error_msg)
|
||||
if task_id:
|
||||
_sync_clickup_fail(ctx, task_id, error_msg)
|
||||
return f"Error: {error_msg}"
|
||||
|
||||
# Collect all deliverable paths
|
||||
deliverable_names = [
|
||||
"test_block.html",
|
||||
"optimization_instructions.md",
|
||||
"validation_report.json",
|
||||
]
|
||||
found_files: dict[str, Path] = {}
|
||||
for name in deliverable_names:
|
||||
fpath = work_dir / name
|
||||
if fpath.exists():
|
||||
found_files[name] = fpath
|
||||
|
||||
# Copy deliverables to network path (if configured)
|
||||
slug = _slugify(keyword) or "unknown"
|
||||
if config and config.content.outline_dir:
|
||||
net_dir = Path(config.content.outline_dir) / slug
|
||||
try:
|
||||
net_dir.mkdir(parents=True, exist_ok=True)
|
||||
for name, fpath in found_files.items():
|
||||
dest = net_dir / name
|
||||
dest.write_bytes(fpath.read_bytes())
|
||||
log.info("Copied %s → %s", fpath, dest)
|
||||
except OSError as e:
|
||||
log.warning("Could not copy deliverables to network path %s: %s", net_dir, e)
|
||||
|
||||
# Sync ClickUp
|
||||
if task_id:
|
||||
_sync_clickup_optimization_complete(
|
||||
ctx=ctx,
|
||||
config=config,
|
||||
task_id=task_id,
|
||||
keyword=keyword,
|
||||
url=url,
|
||||
found_files=found_files,
|
||||
work_dir=work_dir,
|
||||
)
|
||||
|
||||
file_list = "\n".join(f"- `{p}`" for p in found_files.values())
|
||||
return (
|
||||
f"## Optimization Complete\n\n"
|
||||
f"**Keyword:** {keyword}\n"
|
||||
f"**URL:** {url}\n"
|
||||
f"**Deliverables:**\n{file_list}\n\n"
|
||||
f"---\n\n{exec_result}\n\n"
|
||||
f"## ClickUp Sync\nOptimization complete. Status: internal review."
|
||||
)
|
||||
|
||||
|
||||
def _sync_clickup_optimization_complete(
|
||||
*,
|
||||
ctx: dict | None,
|
||||
config,
|
||||
task_id: str,
|
||||
keyword: str,
|
||||
url: str,
|
||||
found_files: dict[str, Path],
|
||||
work_dir: Path,
|
||||
) -> None:
|
||||
"""Upload optimization deliverables to ClickUp and set status."""
|
||||
if not task_id or not ctx:
|
||||
return
|
||||
client = _get_clickup_client(ctx)
|
||||
if not client:
|
||||
return
|
||||
try:
|
||||
# Upload attachments
|
||||
for name, fpath in found_files.items():
|
||||
try:
|
||||
client.upload_attachment(task_id, fpath)
|
||||
log.info("Uploaded %s to ClickUp task %s", name, task_id)
|
||||
except Exception as e:
|
||||
log.warning("Failed to upload %s: %s", name, e)
|
||||
|
||||
# Build comment with validation summary
|
||||
comment_parts = [
|
||||
f"✅ Optimization pipeline complete for '{keyword}'.\n",
|
||||
f"**URL:** {url}\n",
|
||||
"**Deliverables attached:**",
|
||||
]
|
||||
for name in found_files:
|
||||
comment_parts.append(f"- {name}")
|
||||
|
||||
# Include validation summary if available
|
||||
val_path = work_dir / "validation_report.json"
|
||||
if val_path.exists():
|
||||
try:
|
||||
import json
|
||||
val_data = json.loads(val_path.read_text(encoding="utf-8"))
|
||||
summary = val_data.get("summary", "")
|
||||
if summary:
|
||||
comment_parts.append(f"\n**Validation Summary:**\n{summary}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
comment_parts.append(
|
||||
"\n**Next Steps:**\n"
|
||||
"1. Review `optimization_instructions.md` for surgical changes\n"
|
||||
"2. Deploy `test_block.html` hidden div to the page\n"
|
||||
"3. Monitor rankings for 2-4 weeks\n"
|
||||
"4. Apply surgical changes from the instructions doc"
|
||||
)
|
||||
|
||||
client.add_comment(task_id, "\n".join(comment_parts))
|
||||
client.update_task_status(task_id, config.clickup.review_status)
|
||||
except Exception as e:
|
||||
log.warning("Failed to sync optimization complete for %s: %s", task_id, e)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main tool
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -418,6 +782,22 @@ def create_content(
|
|||
|
||||
capabilities_default = config.content.company_capabilities_default if config else ""
|
||||
|
||||
# Optimization path: URL present → run Phase 3 test block pipeline
|
||||
# (skips the outline gate entirely)
|
||||
if url:
|
||||
return _run_optimization(
|
||||
agent=agent,
|
||||
config=config,
|
||||
ctx=ctx,
|
||||
task_id=task_id,
|
||||
url=url,
|
||||
keyword=keyword,
|
||||
cora_path=cora_path,
|
||||
is_service_page=is_service_page,
|
||||
capabilities_default=capabilities_default,
|
||||
)
|
||||
|
||||
# New content path: Phase 1 (outline) → human review → Phase 2 (write)
|
||||
if phase == 1:
|
||||
return _run_phase1(
|
||||
agent=agent,
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@ clickup:
|
|||
"On Page Optimization":
|
||||
tool: "create_content"
|
||||
auto_execute: false
|
||||
required_fields: [keyword, url]
|
||||
field_mapping:
|
||||
url: "IMSURL"
|
||||
keyword: "Keyword"
|
||||
|
|
|
|||
|
|
@ -2,16 +2,21 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from cheddahbot.config import Config, ContentConfig
|
||||
from cheddahbot.tools.content_creation import (
|
||||
_build_optimization_prompt,
|
||||
_build_phase1_prompt,
|
||||
_build_phase2_prompt,
|
||||
_finalize_optimization,
|
||||
_find_cora_report,
|
||||
_run_optimization,
|
||||
_save_content,
|
||||
_slugify,
|
||||
_sync_clickup_optimization_complete,
|
||||
continue_content,
|
||||
create_content,
|
||||
)
|
||||
|
|
@ -234,10 +239,9 @@ class TestCreateContentPhase1:
|
|||
def test_requires_context(self):
|
||||
assert create_content(keyword="kw", url="http://x", ctx=None).startswith("Error:")
|
||||
|
||||
def test_phase1_runs_without_prior_state(self, tmp_db, tmp_path):
|
||||
def test_phase1_runs_for_new_content(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
result = create_content(
|
||||
url="https://example.com/services",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -250,7 +254,6 @@ class TestCreateContentPhase1:
|
|||
def test_phase1_saves_outline_file(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
create_content(
|
||||
url="https://example.com",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -266,7 +269,6 @@ class TestCreateContentPhase1:
|
|||
mock_get_client.return_value = mock_client
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
create_content(
|
||||
url="https://example.com",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -280,7 +282,6 @@ class TestCreateContentPhase1:
|
|||
def test_phase1_includes_clickup_sync_marker(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
result = create_content(
|
||||
url="https://example.com",
|
||||
keyword="test keyword",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -329,7 +330,6 @@ class TestCreateContentPhase2:
|
|||
mock_get_client.return_value = self._make_phase2_client(outline_path)
|
||||
|
||||
result = create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -341,7 +341,6 @@ class TestCreateContentPhase2:
|
|||
mock_get_client.return_value = self._make_phase2_client(outline_path)
|
||||
|
||||
create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -355,7 +354,6 @@ class TestCreateContentPhase2:
|
|||
mock_get_client.return_value = self._make_phase2_client(outline_path)
|
||||
|
||||
create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -370,7 +368,6 @@ class TestCreateContentPhase2:
|
|||
mock_get_client.return_value = mock_client
|
||||
|
||||
create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -384,7 +381,6 @@ class TestCreateContentPhase2:
|
|||
mock_get_client.return_value = self._make_phase2_client(outline_path)
|
||||
|
||||
result = create_content(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -459,7 +455,6 @@ class TestErrorPropagation:
|
|||
"clickup_task_id": "task_err",
|
||||
}
|
||||
result = create_content(
|
||||
url="https://example.com",
|
||||
keyword="test",
|
||||
ctx=ctx,
|
||||
)
|
||||
|
|
@ -483,10 +478,473 @@ class TestErrorPropagation:
|
|||
"clickup_task_id": "task_err2",
|
||||
}
|
||||
result = create_content(
|
||||
url="https://example.com",
|
||||
keyword="test",
|
||||
ctx=ctx,
|
||||
)
|
||||
assert result.startswith("Error:")
|
||||
# Verify ClickUp was notified of the failure
|
||||
mock_client.update_task_status.assert_any_call("task_err2", "error")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _build_optimization_prompt
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildOptimizationPrompt:
|
||||
def test_contains_url_and_keyword(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
)
|
||||
assert "https://example.com/plumbing" in prompt
|
||||
assert "plumbing services" in prompt
|
||||
|
||||
def test_contains_cora_path(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com",
|
||||
keyword="kw",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
)
|
||||
assert "Z:/cora/report.xlsx" in prompt
|
||||
|
||||
def test_contains_all_script_commands(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com",
|
||||
keyword="kw",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
)
|
||||
assert "competitor_scraper.py" in prompt
|
||||
assert "test_block_prep.py" in prompt
|
||||
assert "test_block_generator.py" in prompt
|
||||
assert "test_block_validate.py" in prompt
|
||||
|
||||
def test_contains_step8_instructions(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com",
|
||||
keyword="kw",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
)
|
||||
assert "optimization_instructions.md" in prompt
|
||||
assert "Heading Changes" in prompt
|
||||
assert "Entity Integration Points" in prompt
|
||||
assert "Meta Tag Updates" in prompt
|
||||
assert "Priority Ranking" in prompt
|
||||
|
||||
def test_service_page_note(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com",
|
||||
keyword="kw",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
is_service_page=True,
|
||||
capabilities_default="Check website.",
|
||||
)
|
||||
assert "service page" in prompt
|
||||
assert "Check website." in prompt
|
||||
|
||||
def test_no_service_page_note_by_default(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com",
|
||||
keyword="kw",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
)
|
||||
assert "service page" not in prompt.lower().split("step")[0]
|
||||
|
||||
def test_all_eight_steps_present(self):
|
||||
prompt = _build_optimization_prompt(
|
||||
url="https://example.com",
|
||||
keyword="kw",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
work_dir="/tmp/work",
|
||||
scripts_dir="/scripts",
|
||||
)
|
||||
for step_num in range(1, 9):
|
||||
assert f"Step {step_num}" in prompt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _run_optimization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRunOptimization:
|
||||
def _make_ctx(self, tmp_db, tmp_path):
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
agent = MagicMock()
|
||||
agent.execute_task.return_value = "Optimization complete"
|
||||
return {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "opt_task_1",
|
||||
}
|
||||
|
||||
def test_fails_without_cora_report(self, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
result = _run_optimization(
|
||||
agent=ctx["agent"],
|
||||
config=ctx["config"],
|
||||
ctx=ctx,
|
||||
task_id="opt_task_1",
|
||||
url="https://example.com",
|
||||
keyword="plumbing services",
|
||||
cora_path="",
|
||||
)
|
||||
assert "Error:" in result
|
||||
assert "Cora report" in result
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._sync_clickup_fail")
|
||||
def test_syncs_clickup_on_missing_cora(self, mock_fail, tmp_db, tmp_path):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
_run_optimization(
|
||||
agent=ctx["agent"],
|
||||
config=ctx["config"],
|
||||
ctx=ctx,
|
||||
task_id="opt_task_1",
|
||||
url="https://example.com",
|
||||
keyword="plumbing services",
|
||||
cora_path="",
|
||||
)
|
||||
mock_fail.assert_called_once()
|
||||
assert mock_fail.call_args[0][1] == "opt_task_1"
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._finalize_optimization")
|
||||
@patch("cheddahbot.tools.content_creation._sync_clickup_start")
|
||||
def test_creates_work_dir_and_calls_execute(
|
||||
self, mock_start, mock_finalize, tmp_db, tmp_path
|
||||
):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
mock_finalize.return_value = "finalized"
|
||||
with patch(
|
||||
"cheddahbot.tools.content_creation._LOCAL_CONTENT_DIR",
|
||||
tmp_path / "content",
|
||||
):
|
||||
result = _run_optimization(
|
||||
agent=ctx["agent"],
|
||||
config=ctx["config"],
|
||||
ctx=ctx,
|
||||
task_id="opt_task_1",
|
||||
url="https://example.com/plumbing",
|
||||
keyword="plumbing services",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
)
|
||||
ctx["agent"].execute_task.assert_called_once()
|
||||
mock_start.assert_called_once_with(ctx, "opt_task_1")
|
||||
mock_finalize.assert_called_once()
|
||||
assert result == "finalized"
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._sync_clickup_fail")
|
||||
@patch("cheddahbot.tools.content_creation._sync_clickup_start")
|
||||
def test_syncs_clickup_on_execution_error(
|
||||
self, mock_start, mock_fail, tmp_db, tmp_path
|
||||
):
|
||||
ctx = self._make_ctx(tmp_db, tmp_path)
|
||||
ctx["agent"].execute_task.side_effect = RuntimeError("CLI crashed")
|
||||
with patch(
|
||||
"cheddahbot.tools.content_creation._LOCAL_CONTENT_DIR",
|
||||
tmp_path / "content",
|
||||
):
|
||||
result = _run_optimization(
|
||||
agent=ctx["agent"],
|
||||
config=ctx["config"],
|
||||
ctx=ctx,
|
||||
task_id="opt_task_1",
|
||||
url="https://example.com",
|
||||
keyword="plumbing services",
|
||||
cora_path="Z:/cora/report.xlsx",
|
||||
)
|
||||
assert "Error:" in result
|
||||
mock_fail.assert_called_once()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _finalize_optimization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestFinalizeOptimization:
|
||||
def _make_config(self, outline_dir: str = "") -> Config:
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=outline_dir)
|
||||
return cfg
|
||||
|
||||
def test_errors_on_missing_test_block(self, tmp_path):
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
# Only create instructions, not test_block.html
|
||||
(work_dir / "optimization_instructions.md").write_text("instructions")
|
||||
cfg = self._make_config()
|
||||
result = _finalize_optimization(
|
||||
ctx=None,
|
||||
config=cfg,
|
||||
task_id="",
|
||||
keyword="kw",
|
||||
url="https://example.com",
|
||||
work_dir=work_dir,
|
||||
exec_result="done",
|
||||
)
|
||||
assert "Error:" in result
|
||||
assert "test_block.html" in result
|
||||
|
||||
def test_errors_on_missing_instructions(self, tmp_path):
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
# Only create test_block, not instructions
|
||||
(work_dir / "test_block.html").write_text("<div>block</div>")
|
||||
cfg = self._make_config()
|
||||
result = _finalize_optimization(
|
||||
ctx=None,
|
||||
config=cfg,
|
||||
task_id="",
|
||||
keyword="kw",
|
||||
url="https://example.com",
|
||||
work_dir=work_dir,
|
||||
exec_result="done",
|
||||
)
|
||||
assert "Error:" in result
|
||||
assert "optimization_instructions.md" in result
|
||||
|
||||
def test_succeeds_with_required_files(self, tmp_path):
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
(work_dir / "test_block.html").write_text("<div>block</div>")
|
||||
(work_dir / "optimization_instructions.md").write_text("# Instructions")
|
||||
cfg = self._make_config()
|
||||
result = _finalize_optimization(
|
||||
ctx=None,
|
||||
config=cfg,
|
||||
task_id="",
|
||||
keyword="plumbing services",
|
||||
url="https://example.com",
|
||||
work_dir=work_dir,
|
||||
exec_result="all done",
|
||||
)
|
||||
assert "Optimization Complete" in result
|
||||
assert "plumbing services" in result
|
||||
assert "test_block.html" in result
|
||||
|
||||
def test_copies_to_network_path(self, tmp_path):
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
(work_dir / "test_block.html").write_text("<div>block</div>")
|
||||
(work_dir / "optimization_instructions.md").write_text("# Instructions")
|
||||
net_dir = tmp_path / "network"
|
||||
cfg = self._make_config(str(net_dir))
|
||||
_finalize_optimization(
|
||||
ctx=None,
|
||||
config=cfg,
|
||||
task_id="",
|
||||
keyword="plumbing services",
|
||||
url="https://example.com",
|
||||
work_dir=work_dir,
|
||||
exec_result="done",
|
||||
)
|
||||
assert (net_dir / "plumbing-services" / "test_block.html").exists()
|
||||
assert (net_dir / "plumbing-services" / "optimization_instructions.md").exists()
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._sync_clickup_optimization_complete")
|
||||
def test_syncs_clickup_when_task_id_present(self, mock_sync, tmp_path, tmp_db):
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
(work_dir / "test_block.html").write_text("<div>block</div>")
|
||||
(work_dir / "optimization_instructions.md").write_text("# Instructions")
|
||||
cfg = self._make_config()
|
||||
ctx = {"config": cfg, "db": tmp_db}
|
||||
_finalize_optimization(
|
||||
ctx=ctx,
|
||||
config=cfg,
|
||||
task_id="task_fin",
|
||||
keyword="kw",
|
||||
url="https://example.com",
|
||||
work_dir=work_dir,
|
||||
exec_result="done",
|
||||
)
|
||||
mock_sync.assert_called_once()
|
||||
call_kwargs = mock_sync.call_args.kwargs
|
||||
assert call_kwargs["task_id"] == "task_fin"
|
||||
assert "test_block.html" in call_kwargs["found_files"]
|
||||
assert "optimization_instructions.md" in call_kwargs["found_files"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _sync_clickup_optimization_complete
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSyncClickupOptimizationComplete:
|
||||
@patch("cheddahbot.tools.content_creation._get_clickup_client")
|
||||
def test_uploads_files_and_posts_comment(self, mock_get_client, tmp_path):
|
||||
mock_client = MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
tb_path = work_dir / "test_block.html"
|
||||
tb_path.write_text("<div>block</div>")
|
||||
inst_path = work_dir / "optimization_instructions.md"
|
||||
inst_path.write_text("# Instructions")
|
||||
val_path = work_dir / "validation_report.json"
|
||||
val_path.write_text(json.dumps({"summary": "All metrics improved."}))
|
||||
|
||||
cfg = Config()
|
||||
ctx = {"config": cfg}
|
||||
found_files = {
|
||||
"test_block.html": tb_path,
|
||||
"optimization_instructions.md": inst_path,
|
||||
"validation_report.json": val_path,
|
||||
}
|
||||
_sync_clickup_optimization_complete(
|
||||
ctx=ctx,
|
||||
config=cfg,
|
||||
task_id="task_sync",
|
||||
keyword="plumbing",
|
||||
url="https://example.com",
|
||||
found_files=found_files,
|
||||
work_dir=work_dir,
|
||||
)
|
||||
# 3 file uploads
|
||||
assert mock_client.upload_attachment.call_count == 3
|
||||
# Comment posted
|
||||
mock_client.add_comment.assert_called_once()
|
||||
comment = mock_client.add_comment.call_args[0][1]
|
||||
assert "plumbing" in comment
|
||||
assert "All metrics improved." in comment
|
||||
assert "Next Steps" in comment
|
||||
# Status set to internal review
|
||||
mock_client.update_task_status.assert_called_once_with(
|
||||
"task_sync", cfg.clickup.review_status
|
||||
)
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._get_clickup_client")
|
||||
def test_handles_no_validation_report(self, mock_get_client, tmp_path):
|
||||
mock_client = MagicMock()
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
tb_path = work_dir / "test_block.html"
|
||||
tb_path.write_text("<div>block</div>")
|
||||
inst_path = work_dir / "optimization_instructions.md"
|
||||
inst_path.write_text("# Instructions")
|
||||
|
||||
cfg = Config()
|
||||
ctx = {"config": cfg}
|
||||
found_files = {
|
||||
"test_block.html": tb_path,
|
||||
"optimization_instructions.md": inst_path,
|
||||
}
|
||||
_sync_clickup_optimization_complete(
|
||||
ctx=ctx,
|
||||
config=cfg,
|
||||
task_id="task_sync2",
|
||||
keyword="kw",
|
||||
url="https://example.com",
|
||||
found_files=found_files,
|
||||
work_dir=work_dir,
|
||||
)
|
||||
# 2 uploads (no validation_report.json)
|
||||
assert mock_client.upload_attachment.call_count == 2
|
||||
mock_client.add_comment.assert_called_once()
|
||||
|
||||
def test_noop_without_task_id(self, tmp_path):
|
||||
"""No ClickUp sync when task_id is empty."""
|
||||
work_dir = tmp_path / "work"
|
||||
work_dir.mkdir()
|
||||
cfg = Config()
|
||||
# Should not raise
|
||||
_sync_clickup_optimization_complete(
|
||||
ctx={"config": cfg},
|
||||
config=cfg,
|
||||
task_id="",
|
||||
keyword="kw",
|
||||
url="https://example.com",
|
||||
found_files={},
|
||||
work_dir=work_dir,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# create_content — Routing (URL → optimization vs new content → phases)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCreateContentRouting:
|
||||
@patch("cheddahbot.tools.content_creation._run_optimization")
|
||||
def test_url_present_routes_to_optimization(self, mock_opt, tmp_db, tmp_path):
|
||||
"""When URL is present, create_content should call _run_optimization."""
|
||||
mock_opt.return_value = "## Optimization Complete"
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
ctx = {
|
||||
"agent": MagicMock(),
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "routing_test",
|
||||
}
|
||||
result = create_content(
|
||||
keyword="plumbing services",
|
||||
url="https://example.com/plumbing",
|
||||
ctx=ctx,
|
||||
)
|
||||
mock_opt.assert_called_once()
|
||||
assert result == "## Optimization Complete"
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._run_optimization")
|
||||
def test_no_url_does_not_route_to_optimization(self, mock_opt, tmp_db, tmp_path):
|
||||
"""When URL is empty, create_content should NOT call _run_optimization."""
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
agent = MagicMock()
|
||||
agent.execute_task.return_value = "## Outline"
|
||||
ctx = {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "",
|
||||
}
|
||||
result = create_content(
|
||||
keyword="new keyword",
|
||||
url="",
|
||||
ctx=ctx,
|
||||
)
|
||||
mock_opt.assert_not_called()
|
||||
assert "Phase 1 Complete" in result
|
||||
|
||||
@patch("cheddahbot.tools.content_creation._run_optimization")
|
||||
def test_new_content_still_calls_phase1(self, mock_opt, tmp_db, tmp_path):
|
||||
"""Regression: new content (no URL) still goes through _run_phase1."""
|
||||
cfg = Config()
|
||||
cfg.content = ContentConfig(outline_dir=str(tmp_path / "outlines"))
|
||||
agent = MagicMock()
|
||||
agent.execute_task.return_value = "## Generated Outline\nContent..."
|
||||
ctx = {
|
||||
"agent": agent,
|
||||
"config": cfg,
|
||||
"db": tmp_db,
|
||||
"clickup_task_id": "",
|
||||
}
|
||||
create_content(
|
||||
keyword="new topic",
|
||||
url="",
|
||||
ctx=ctx,
|
||||
)
|
||||
mock_opt.assert_not_called()
|
||||
agent.execute_task.assert_called_once()
|
||||
# Verify it's the phase 1 prompt (new content path)
|
||||
call_args = agent.execute_task.call_args
|
||||
prompt = call_args.args[0] if call_args.args else call_args.kwargs.get("prompt", "")
|
||||
assert "new content creation project" in prompt
|
||||
|
|
|
|||
Loading…
Reference in New Issue