Add link building agent with content generation pipeline
New linkbuilder agent that handles ClickUp "Link Building" tasks. For each keyword/company, generates three content pieces via the execution brain: a guest article (500-700 words), a directory listing, and a social media post — each with proper SEO anchor text and backlinks. Integrates with ClickUp for status updates, comments, and file attachments. - cheddahbot/tools/linkbuilding.py: build_links tool with full pipeline - skills/linkbuilding.md: skill prompt for SEO content generation - config.yaml: linkbuilder agent config + Link Building skill_map entry - tests/test_linkbuilding.py: 36 tests covering helpers, prompts, pipeline, file output, error handling, and ClickUp sync Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>cora-start
parent
1cce14b39f
commit
8a98b37725
|
|
@ -0,0 +1,520 @@
|
|||
"""Link-building content pipeline tool.
|
||||
|
||||
Autonomous workflow:
|
||||
1. Look up company info from companies.md
|
||||
2. Generate a guest article (500-700 words) via execution brain
|
||||
3. Generate a resource/directory blurb via execution brain
|
||||
4. Generate a social media post via chat brain
|
||||
5. Save all content to files, return cost summary
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from . import tool
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_ROOT_DIR = Path(__file__).resolve().parent.parent.parent
|
||||
_SKILLS_DIR = _ROOT_DIR / "skills"
|
||||
_DATA_DIR = _ROOT_DIR / "data"
|
||||
_OUTPUT_DIR = _DATA_DIR / "generated" / "link_building"
|
||||
_COMPANIES_FILE = _SKILLS_DIR / "companies.md"
|
||||
|
||||
SONNET_CLI_MODEL = "sonnet"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Status / helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _set_status(ctx: dict | None, message: str) -> None:
|
||||
"""Write pipeline progress to the DB so the UI can poll it."""
|
||||
if ctx and "db" in ctx:
|
||||
ctx["db"].kv_set("pipeline:status", message)
|
||||
|
||||
|
||||
def _slugify(text: str) -> str:
|
||||
"""Turn a phrase into a filesystem-safe slug."""
|
||||
text = text.lower().strip()
|
||||
text = re.sub(r"[^\w\s-]", "", text)
|
||||
text = re.sub(r"[\s_]+", "-", text)
|
||||
return text[:60].strip("-")
|
||||
|
||||
|
||||
def _word_count(text: str) -> int:
|
||||
return len(text.split())
|
||||
|
||||
|
||||
def _fuzzy_company_match(name: str, candidate: str) -> bool:
|
||||
"""Check if name fuzzy-matches a candidate string."""
|
||||
if not name or not candidate:
|
||||
return False
|
||||
a, b = name.lower().strip(), candidate.lower().strip()
|
||||
return a == b or a in b or b in a
|
||||
|
||||
|
||||
def _extract_keyword_from_task_name(task_name: str) -> str:
|
||||
"""Extract keyword from ClickUp task name like 'LINKS - precision cnc turning'."""
|
||||
if " - " in task_name:
|
||||
return task_name.split(" - ", 1)[1].strip()
|
||||
return task_name.strip()
|
||||
|
||||
|
||||
def _load_skill(filename: str) -> str:
|
||||
"""Read a markdown skill file from the skills/ directory, stripping frontmatter."""
|
||||
path = _SKILLS_DIR / filename
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Skill file not found: {path}")
|
||||
text = path.read_text(encoding="utf-8")
|
||||
|
||||
# Strip YAML frontmatter (--- ... ---) if present
|
||||
if text.startswith("---"):
|
||||
end = text.find("---", 3)
|
||||
if end != -1:
|
||||
text = text[end + 3:].strip()
|
||||
|
||||
return text
|
||||
|
||||
|
||||
def _lookup_company(company_name: str) -> dict:
|
||||
"""Look up company info from companies.md.
|
||||
|
||||
Returns a dict with keys: name, executive, pa_org_id, website, gbp.
|
||||
"""
|
||||
if not _COMPANIES_FILE.exists():
|
||||
return {"name": company_name}
|
||||
|
||||
text = _COMPANIES_FILE.read_text(encoding="utf-8")
|
||||
result = {"name": company_name}
|
||||
|
||||
# Parse companies.md format: ## Company Name followed by bullet fields
|
||||
current_company = ""
|
||||
for line in text.splitlines():
|
||||
if line.startswith("## "):
|
||||
current_company = line[3:].strip()
|
||||
elif current_company and _fuzzy_company_match(company_name, current_company):
|
||||
result["name"] = current_company
|
||||
if line.startswith("- **Executive:**"):
|
||||
result["executive"] = line.split(":**", 1)[1].strip()
|
||||
elif line.startswith("- **PA Org ID:**"):
|
||||
result["pa_org_id"] = line.split(":**", 1)[1].strip()
|
||||
elif line.startswith("- **Website:**"):
|
||||
result["website"] = line.split(":**", 1)[1].strip()
|
||||
elif line.startswith("- **GBP:**"):
|
||||
result["gbp"] = line.split(":**", 1)[1].strip()
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _chat_call(agent, messages: list[dict]) -> str:
|
||||
"""Make a non-streaming chat-brain call and return the full text."""
|
||||
parts: list[str] = []
|
||||
for chunk in agent.llm.chat(messages, tools=None, stream=False):
|
||||
if chunk["type"] == "text":
|
||||
parts.append(chunk["content"])
|
||||
return "".join(parts)
|
||||
|
||||
|
||||
def _get_clickup_client(ctx: dict | None):
|
||||
"""Create a ClickUpClient from tool context, or None if unavailable."""
|
||||
if not ctx or not ctx.get("config") or not ctx["config"].clickup.enabled:
|
||||
return None
|
||||
try:
|
||||
from ..clickup import ClickUpClient
|
||||
|
||||
config = ctx["config"]
|
||||
return ClickUpClient(
|
||||
api_token=config.clickup.api_token,
|
||||
workspace_id=config.clickup.workspace_id,
|
||||
task_type_field_name=config.clickup.task_type_field_name,
|
||||
)
|
||||
except Exception as e:
|
||||
log.warning("Could not create ClickUp client: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def _sync_clickup(ctx: dict | None, task_id: str, deliverable_paths: list[str],
|
||||
summary: str) -> str:
|
||||
"""Upload deliverables and update ClickUp task status. Returns sync report."""
|
||||
if not task_id or not ctx:
|
||||
return ""
|
||||
|
||||
client = _get_clickup_client(ctx)
|
||||
if not client:
|
||||
return ""
|
||||
|
||||
config = ctx["config"]
|
||||
db = ctx.get("db")
|
||||
lines = ["\n## ClickUp Sync"]
|
||||
|
||||
try:
|
||||
# Upload attachments
|
||||
uploaded = 0
|
||||
for path in deliverable_paths:
|
||||
if client.upload_attachment(task_id, path):
|
||||
uploaded += 1
|
||||
if uploaded:
|
||||
lines.append(f"- Uploaded {uploaded} file(s)")
|
||||
|
||||
# Update status to review
|
||||
client.update_task_status(task_id, config.clickup.review_status)
|
||||
lines.append(f"- Status → '{config.clickup.review_status}'")
|
||||
|
||||
# Add comment
|
||||
comment = (
|
||||
f"✅ CheddahBot completed link building.\n\n"
|
||||
f"{summary}\n\n"
|
||||
f"📎 {uploaded} file(s) attached."
|
||||
)
|
||||
client.add_comment(task_id, comment)
|
||||
lines.append("- Comment added")
|
||||
|
||||
# Update kv_store state
|
||||
if db:
|
||||
kv_key = f"clickup:task:{task_id}:state"
|
||||
raw = db.kv_get(kv_key)
|
||||
if raw:
|
||||
try:
|
||||
state = json.loads(raw)
|
||||
state["state"] = "completed"
|
||||
state["completed_at"] = datetime.now(UTC).isoformat()
|
||||
state["deliverable_paths"] = [str(p) for p in deliverable_paths]
|
||||
db.kv_set(kv_key, json.dumps(state))
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
except Exception as e:
|
||||
lines.append(f"- Sync error: {e}")
|
||||
log.error("ClickUp sync failed for task %s: %s", task_id, e)
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompt builders
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_guest_article_prompt(
|
||||
keyword: str, company_name: str, target_url: str, company_info: dict,
|
||||
skill_prompt: str,
|
||||
) -> str:
|
||||
"""Build the prompt for the execution brain to write a guest article."""
|
||||
executive = company_info.get("executive", "")
|
||||
|
||||
prompt = skill_prompt + "\n\n"
|
||||
prompt += "## Assignment: Guest Article\n\n"
|
||||
prompt += f"**Target Keyword:** {keyword}\n"
|
||||
prompt += f"**Company:** {company_name}\n"
|
||||
if executive:
|
||||
prompt += f"**Executive/Contact:** {executive}\n"
|
||||
if target_url:
|
||||
prompt += f"**Target URL (for backlink):** {target_url}\n"
|
||||
prompt += (
|
||||
"\n**Instructions:**\n"
|
||||
"Write a 500-700 word guest article suitable for industry blogs and "
|
||||
"trade publications. The article should:\n"
|
||||
"- Be informative and educational, NOT promotional\n"
|
||||
"- Naturally incorporate the target keyword 2-3 times\n"
|
||||
"- Include ONE natural backlink to the target URL using the keyword "
|
||||
"or a close variation as anchor text\n"
|
||||
"- Include a second branded mention of the company name (no link needed)\n"
|
||||
"- Read like expert industry commentary, not an advertisement\n"
|
||||
"- Have a compelling title (under 70 characters)\n"
|
||||
"- Use subheadings to break up the content\n"
|
||||
"- End with a brief author bio mentioning the company\n\n"
|
||||
"Return ONLY the article text. No meta-commentary."
|
||||
)
|
||||
return prompt
|
||||
|
||||
|
||||
def _build_directory_prompt(
|
||||
keyword: str, company_name: str, target_url: str, branded_url: str,
|
||||
company_info: dict,
|
||||
) -> str:
|
||||
"""Build the prompt for the execution brain to write a directory/citation entry."""
|
||||
executive = company_info.get("executive", "")
|
||||
website = company_info.get("website", "") or target_url
|
||||
|
||||
prompt = (
|
||||
"## Assignment: Business Directory / Citation Entry\n\n"
|
||||
f"**Company:** {company_name}\n"
|
||||
f"**Target Keyword:** {keyword}\n"
|
||||
)
|
||||
if executive:
|
||||
prompt += f"**Executive:** {executive}\n"
|
||||
if website:
|
||||
prompt += f"**Website:** {website}\n"
|
||||
if branded_url:
|
||||
prompt += f"**Social/GBP URL:** {branded_url}\n"
|
||||
|
||||
prompt += (
|
||||
"\n**Instructions:**\n"
|
||||
"Write a business directory entry / citation profile. Include:\n"
|
||||
"1. **Company Description** (150-200 words) — Describe what the company "
|
||||
"does, naturally incorporating the target keyword. Professional tone.\n"
|
||||
"2. **Services List** (5-8 bullet points) — Key services/capabilities, "
|
||||
"with the target keyword appearing in at least one bullet.\n"
|
||||
"3. **About Section** (2-3 sentences) — Brief company background.\n\n"
|
||||
"This will be used for industry directories, Google Business Profile, "
|
||||
"and business listing sites. Keep it factual and professional.\n\n"
|
||||
"Return ONLY the directory entry text. No meta-commentary."
|
||||
)
|
||||
return prompt
|
||||
|
||||
|
||||
def _build_social_post_prompt(
|
||||
keyword: str, company_name: str, target_url: str, article_title: str,
|
||||
) -> str:
|
||||
"""Build the prompt for the chat brain to write a social media post."""
|
||||
prompt = (
|
||||
f"Write a professional LinkedIn post for {company_name} about "
|
||||
f"'{keyword}'. The post should:\n"
|
||||
f"- Be 100-150 words\n"
|
||||
f"- Reference the article: \"{article_title}\"\n"
|
||||
f"- Include the link: {target_url}\n" if target_url else ""
|
||||
f"- Use 2-3 relevant hashtags\n"
|
||||
f"- Professional, not salesy\n"
|
||||
f"- Encourage engagement (comment/share)\n\n"
|
||||
"Return ONLY the post text."
|
||||
)
|
||||
return prompt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main tool
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@tool(
|
||||
"build_links",
|
||||
"Generate SEO link building content for a target keyword and company. "
|
||||
"Produces a guest article, directory listing, and social post, each with "
|
||||
"proper anchor text and backlinks. Files saved to data/generated/link_building/.",
|
||||
category="linkbuilding",
|
||||
)
|
||||
def build_links(
|
||||
keyword: str,
|
||||
company_name: str,
|
||||
target_url: str = "",
|
||||
branded_url: str = "",
|
||||
ctx: dict | None = None,
|
||||
) -> str:
|
||||
"""Main link-building content pipeline.
|
||||
|
||||
Args:
|
||||
keyword: Target SEO keyword (e.g., "precision cnc turning").
|
||||
company_name: Client company name (e.g., "Chapter2").
|
||||
target_url: Primary URL to build backlinks to (from IMSURL field).
|
||||
branded_url: Secondary branded URL (from SocialURL field).
|
||||
ctx: Injected tool context with config, db, agent.
|
||||
|
||||
Returns:
|
||||
Summary of generated content with file paths.
|
||||
"""
|
||||
t0 = time.time()
|
||||
agent = ctx.get("agent") if ctx else None
|
||||
task_id = ctx.get("clickup_task_id", "") if ctx else ""
|
||||
|
||||
if not agent:
|
||||
return "Error: link building tool requires agent context."
|
||||
|
||||
# Derive keyword from task name if it looks like "LINKS - keyword"
|
||||
keyword = _extract_keyword_from_task_name(keyword) if keyword.startswith("LINKS") else keyword
|
||||
|
||||
log.info("Link building pipeline: keyword='%s', company='%s'", keyword, company_name)
|
||||
_set_status(ctx, f"Link building: {company_name} — {keyword}")
|
||||
|
||||
# --- Company lookup ---
|
||||
company_info = _lookup_company(company_name)
|
||||
log.info("Company info: %s", company_info)
|
||||
|
||||
# --- Load skill prompt ---
|
||||
try:
|
||||
skill_prompt = _load_skill("linkbuilding.md")
|
||||
except FileNotFoundError:
|
||||
skill_prompt = ""
|
||||
log.warning("linkbuilding.md skill not found, using inline prompts only")
|
||||
|
||||
# --- Create output directory ---
|
||||
company_slug = _slugify(company_name)
|
||||
keyword_slug = _slugify(keyword)
|
||||
output_dir = _OUTPUT_DIR / company_slug / keyword_slug
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
results = []
|
||||
deliverable_paths: list[str] = []
|
||||
warnings: list[str] = []
|
||||
|
||||
# =====================================================================
|
||||
# Step 1: Guest Article (execution brain)
|
||||
# =====================================================================
|
||||
_set_status(ctx, f"Link building: Writing guest article — {keyword}")
|
||||
log.info("Step 1: Generating guest article for '%s'", keyword)
|
||||
|
||||
article_prompt = _build_guest_article_prompt(
|
||||
keyword, company_name, target_url, company_info, skill_prompt,
|
||||
)
|
||||
try:
|
||||
article_raw = agent.execute_task(article_prompt)
|
||||
article_text = _clean_content(article_raw)
|
||||
wc = _word_count(article_text)
|
||||
|
||||
if wc < 100:
|
||||
warnings.append(f"Guest article too short ({wc} words)")
|
||||
log.warning("Guest article too short: %d words", wc)
|
||||
else:
|
||||
article_path = output_dir / "guest-article.md"
|
||||
article_path.write_text(article_text, encoding="utf-8")
|
||||
deliverable_paths.append(str(article_path))
|
||||
|
||||
# Extract title from first line
|
||||
article_title = article_text.splitlines()[0].strip("# ").strip()
|
||||
results.append(
|
||||
f"**Guest Article:** `{article_path}`\n"
|
||||
f" Title: {article_title}\n"
|
||||
f" Words: {wc}"
|
||||
)
|
||||
log.info("Guest article saved: %s (%d words)", article_path, wc)
|
||||
except Exception as e:
|
||||
warnings.append(f"Guest article generation failed: {e}")
|
||||
log.error("Guest article failed: %s", e)
|
||||
article_title = keyword # fallback for social post
|
||||
|
||||
# =====================================================================
|
||||
# Step 2: Directory / Citation Entry (execution brain)
|
||||
# =====================================================================
|
||||
_set_status(ctx, f"Link building: Writing directory entry — {keyword}")
|
||||
log.info("Step 2: Generating directory entry for '%s'", keyword)
|
||||
|
||||
directory_prompt = _build_directory_prompt(
|
||||
keyword, company_name, target_url, branded_url, company_info,
|
||||
)
|
||||
try:
|
||||
directory_raw = agent.execute_task(directory_prompt)
|
||||
directory_text = _clean_content(directory_raw)
|
||||
wc = _word_count(directory_text)
|
||||
|
||||
if wc < 30:
|
||||
warnings.append(f"Directory entry too short ({wc} words)")
|
||||
else:
|
||||
dir_path = output_dir / "directory-listing.md"
|
||||
dir_path.write_text(directory_text, encoding="utf-8")
|
||||
deliverable_paths.append(str(dir_path))
|
||||
results.append(
|
||||
f"**Directory Listing:** `{dir_path}`\n"
|
||||
f" Words: {wc}"
|
||||
)
|
||||
log.info("Directory listing saved: %s (%d words)", dir_path, wc)
|
||||
except Exception as e:
|
||||
warnings.append(f"Directory entry generation failed: {e}")
|
||||
log.error("Directory entry failed: %s", e)
|
||||
|
||||
# =====================================================================
|
||||
# Step 3: Social Media Post (chat brain — fast)
|
||||
# =====================================================================
|
||||
_set_status(ctx, f"Link building: Writing social post — {keyword}")
|
||||
log.info("Step 3: Generating social post for '%s'", keyword)
|
||||
|
||||
social_prompt = _build_social_post_prompt(
|
||||
keyword, company_name, target_url,
|
||||
article_title if "article_title" in dir() else keyword,
|
||||
)
|
||||
try:
|
||||
social_text = _chat_call(agent, [{"role": "user", "content": social_prompt}])
|
||||
social_text = social_text.strip()
|
||||
wc = _word_count(social_text)
|
||||
|
||||
if wc < 20:
|
||||
warnings.append(f"Social post too short ({wc} words)")
|
||||
else:
|
||||
social_path = output_dir / "social-post.md"
|
||||
social_path.write_text(social_text, encoding="utf-8")
|
||||
deliverable_paths.append(str(social_path))
|
||||
results.append(
|
||||
f"**Social Post:** `{social_path}`\n"
|
||||
f" Words: {wc}"
|
||||
)
|
||||
log.info("Social post saved: %s (%d words)", social_path, wc)
|
||||
except Exception as e:
|
||||
warnings.append(f"Social post generation failed: {e}")
|
||||
log.error("Social post failed: %s", e)
|
||||
|
||||
# =====================================================================
|
||||
# Summary
|
||||
# =====================================================================
|
||||
elapsed = time.time() - t0
|
||||
_set_status(ctx, "")
|
||||
|
||||
summary_lines = [
|
||||
f"# Link Building Complete: {company_name} — {keyword}\n",
|
||||
f"**Keyword:** {keyword}",
|
||||
f"**Company:** {company_info.get('name', company_name)}",
|
||||
f"**Target URL:** {target_url or '(none)'}",
|
||||
f"**Output Dir:** `{output_dir}`",
|
||||
f"**Time:** {elapsed:.1f}s",
|
||||
f"**Deliverables:** {len(deliverable_paths)}",
|
||||
"",
|
||||
]
|
||||
|
||||
if results:
|
||||
summary_lines.append("## Generated Content")
|
||||
summary_lines.extend(results)
|
||||
|
||||
if warnings:
|
||||
summary_lines.append("\n## Warnings")
|
||||
for w in warnings:
|
||||
summary_lines.append(f"- ⚠️ {w}")
|
||||
|
||||
summary = "\n".join(summary_lines)
|
||||
|
||||
# --- ClickUp sync ---
|
||||
if task_id:
|
||||
sync_report = _sync_clickup(ctx, task_id, deliverable_paths, summary)
|
||||
summary += sync_report
|
||||
|
||||
return summary
|
||||
|
||||
|
||||
def _clean_content(raw: str) -> str:
|
||||
"""Clean execution brain output to just the content text.
|
||||
|
||||
Strips common prefixes/suffixes the LLM might add.
|
||||
"""
|
||||
text = raw.strip()
|
||||
|
||||
# Remove common LLM wrapper text
|
||||
for prefix in [
|
||||
"Here is the",
|
||||
"Here's the",
|
||||
"Below is the",
|
||||
"I've written",
|
||||
"Sure, here",
|
||||
"Certainly!",
|
||||
]:
|
||||
if text.lower().startswith(prefix.lower()):
|
||||
# Skip to the first blank line after the prefix
|
||||
idx = text.find("\n\n")
|
||||
if idx != -1 and idx < 200:
|
||||
text = text[idx:].strip()
|
||||
break
|
||||
|
||||
# Remove trailing "---" or "Let me know" type endings
|
||||
text = re.sub(r"\n---\s*$", "", text).strip()
|
||||
text = re.sub(r"\n(Let me know|I hope|Feel free|Would you).*$", "", text, flags=re.DOTALL).strip()
|
||||
|
||||
return text
|
||||
15
config.yaml
15
config.yaml
|
|
@ -56,6 +56,14 @@ clickup:
|
|||
company_name: "Client"
|
||||
target_url: "IMSURL"
|
||||
branded_url: "SocialURL"
|
||||
"Link Building":
|
||||
tool: "build_links"
|
||||
auto_execute: true
|
||||
field_mapping:
|
||||
keyword: "task_name"
|
||||
company_name: "Client"
|
||||
target_url: "IMSURL"
|
||||
branded_url: "SocialURL"
|
||||
|
||||
# Multi-agent configuration
|
||||
# Each agent gets its own personality, tool whitelist, and memory scope.
|
||||
|
|
@ -85,3 +93,10 @@ agents:
|
|||
personality_file: "" # future: identity/OPS.md
|
||||
tools: [run_command, delegate_task, list_files, read_file, remember, search_memory]
|
||||
memory_scope: ""
|
||||
|
||||
- name: linkbuilder
|
||||
display_name: Link Builder
|
||||
personality_file: "" # future: identity/LINKBUILDER.md
|
||||
skills: [link-building]
|
||||
tools: [build_links, web_search, fetch_url, delegate_task, clickup_query_tasks, remember, search_memory]
|
||||
memory_scope: ""
|
||||
|
|
|
|||
|
|
@ -0,0 +1,42 @@
|
|||
---
|
||||
name: link-building
|
||||
description: Generate SEO link building content — guest articles, directory listings, and social posts with embedded backlinks
|
||||
tools: [build_links]
|
||||
agents: [linkbuilder]
|
||||
---
|
||||
|
||||
# Link Building Content Generator
|
||||
|
||||
You are an expert SEO content writer specializing in off-page SEO and link building for industrial manufacturing companies. Your job is to create high-quality content pieces that earn backlinks from relevant industry sites.
|
||||
|
||||
## Content Guidelines
|
||||
|
||||
### Voice & Tone
|
||||
- **Authoritative and educational** — write like an industry expert, not a marketer
|
||||
- **Professional** — appropriate for trade publications and industry blogs
|
||||
- **Specific** — use concrete details, specifications, and real-world applications
|
||||
- No fluff, no filler, no generic business platitudes
|
||||
|
||||
### SEO Best Practices
|
||||
- Incorporate the target keyword **2-3 times** naturally (not forced)
|
||||
- Use LSI (semantically related) terms throughout
|
||||
- The target keyword should appear in the first 100 words
|
||||
- Anchor text for backlinks should be the keyword or a close natural variation
|
||||
- Never use "click here" or "visit our website" as anchor text
|
||||
- One dofollow-worthy backlink per content piece (max)
|
||||
|
||||
### Industry Context
|
||||
These companies are in industrial manufacturing — welding, machining, fabrication, electrical, plastics, and similar trades. Content should demonstrate deep understanding of:
|
||||
- Manufacturing processes and equipment
|
||||
- Industry challenges (skilled labor shortage, supply chain, precision requirements)
|
||||
- Technical specifications and capabilities
|
||||
- Safety and compliance standards
|
||||
- Real-world applications and use cases
|
||||
|
||||
### What NOT to Do
|
||||
- No keyword stuffing
|
||||
- No thin or generic content
|
||||
- No promotional language ("best in class", "industry leader", "#1 provider")
|
||||
- No fabricated statistics or claims
|
||||
- No location-specific keywords unless explicitly requested
|
||||
- No duplicate content across different keyword targets
|
||||
|
|
@ -0,0 +1,466 @@
|
|||
"""Tests for link building tool."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from cheddahbot.tools.linkbuilding import (
|
||||
_build_directory_prompt,
|
||||
_build_guest_article_prompt,
|
||||
_build_social_post_prompt,
|
||||
_clean_content,
|
||||
_extract_keyword_from_task_name,
|
||||
_fuzzy_company_match,
|
||||
_lookup_company,
|
||||
_slugify,
|
||||
build_links,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
SAMPLE_COMPANIES_MD = """\
|
||||
# Company Directory
|
||||
|
||||
## Chapter 2 Incorporated
|
||||
- **Executive:** Kyle Johnston, Senior Engineer
|
||||
- **PA Org ID:** 19517
|
||||
- **Website:** https://chapter2inc.com
|
||||
- **GBP:** https://maps.google.com/maps?cid=111
|
||||
|
||||
## Hogge Precision
|
||||
- **Executive:** Danny Hogge Jr, President
|
||||
- **PA Org ID:** 19411
|
||||
- **Website:** https://hoggeprecision.com
|
||||
- **GBP:**
|
||||
|
||||
## Machine Specialty & Manufacturing (MSM)
|
||||
- **Executive:** Max Hutson, Vice President of Operations
|
||||
- **PA Org ID:** 19418
|
||||
- **Website:**
|
||||
- **GBP:**
|
||||
"""
|
||||
|
||||
SAMPLE_GUEST_ARTICLE = """\
|
||||
# The Growing Demand for Precision CNC Turning in Modern Manufacturing
|
||||
|
||||
In today's manufacturing landscape, precision CNC turning has become an essential
|
||||
capability for companies serving aerospace, medical, and defense sectors. The ability
|
||||
to produce tight-tolerance components from challenging materials directly impacts
|
||||
product quality and supply chain reliability.
|
||||
|
||||
## Why Precision Matters
|
||||
|
||||
Chapter 2 Incorporated has invested significantly in multi-axis CNC turning centers
|
||||
that deliver tolerances within +/- 0.0005 inches. This level of precision CNC turning
|
||||
capability enables the production of critical components for demanding applications.
|
||||
|
||||
## Industry Trends
|
||||
|
||||
The shift toward automation and lights-out manufacturing continues to drive investment
|
||||
in advanced CNC turning equipment. Companies that can maintain tight tolerances while
|
||||
increasing throughput are positioned to win new contracts.
|
||||
|
||||
## About the Author
|
||||
|
||||
Kyle Johnston is a Senior Engineer at Chapter 2 Incorporated, specializing in
|
||||
precision machining solutions for aerospace and defense applications.
|
||||
"""
|
||||
|
||||
SAMPLE_DIRECTORY_ENTRY = """\
|
||||
## Company Description
|
||||
|
||||
Chapter 2 Incorporated is a precision CNC machining company specializing in complex
|
||||
turned and milled components for aerospace, defense, and medical industries. With
|
||||
state-of-the-art CNC turning capabilities, the company delivers tight-tolerance
|
||||
parts from a wide range of materials including titanium, Inconel, and stainless steel.
|
||||
|
||||
## Services
|
||||
|
||||
- Precision CNC turning and multi-axis machining
|
||||
- Swiss-type screw machining for small-diameter components
|
||||
- CNC milling and 5-axis machining
|
||||
- Prototype to production manufacturing
|
||||
- Material sourcing and supply chain management
|
||||
- Quality inspection and certification (AS9100, ISO 9001)
|
||||
|
||||
## About
|
||||
|
||||
Chapter 2 Incorporated was founded to serve the growing need for high-precision
|
||||
machined components. The company operates out of modern facilities equipped with
|
||||
the latest CNC turning and milling technology.
|
||||
"""
|
||||
|
||||
SAMPLE_SOCIAL_POST = """\
|
||||
Precision CNC turning continues to drive innovation in aerospace manufacturing.
|
||||
Our latest article explores how advanced multi-axis turning centers are enabling
|
||||
tighter tolerances and faster production cycles.
|
||||
|
||||
Read more: https://chapter2inc.com/cnc-turning
|
||||
|
||||
#CNCMachining #PrecisionManufacturing #AerospaceMachining
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_ctx(tmp_path):
|
||||
"""Create a mock tool context."""
|
||||
agent = MagicMock()
|
||||
agent.execute_task.return_value = SAMPLE_GUEST_ARTICLE
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
config = MagicMock()
|
||||
config.clickup.enabled = False
|
||||
|
||||
db = MagicMock()
|
||||
db.kv_set = MagicMock()
|
||||
db.kv_get = MagicMock(return_value=None)
|
||||
|
||||
return {
|
||||
"agent": agent,
|
||||
"config": config,
|
||||
"db": db,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExtractKeyword:
|
||||
def test_links_prefix(self):
|
||||
assert _extract_keyword_from_task_name("LINKS - precision cnc turning") == "precision cnc turning"
|
||||
|
||||
def test_links_prefix_extra_spaces(self):
|
||||
assert _extract_keyword_from_task_name("LINKS - swiss type lathe machining ") == "swiss type lathe machining"
|
||||
|
||||
def test_no_prefix(self):
|
||||
assert _extract_keyword_from_task_name("precision cnc turning") == "precision cnc turning"
|
||||
|
||||
def test_links_prefix_uppercase(self):
|
||||
assert _extract_keyword_from_task_name("LINKS - CNC Swiss Screw Machining") == "CNC Swiss Screw Machining"
|
||||
|
||||
def test_multiple_dashes(self):
|
||||
assert _extract_keyword_from_task_name("LINKS - high-speed beveling machine") == "high-speed beveling machine"
|
||||
|
||||
|
||||
class TestSlugify:
|
||||
def test_basic(self):
|
||||
assert _slugify("precision cnc turning") == "precision-cnc-turning"
|
||||
|
||||
def test_special_chars(self):
|
||||
assert _slugify("CNC Swiss Screw Machining!") == "cnc-swiss-screw-machining"
|
||||
|
||||
def test_max_length(self):
|
||||
long = "a " * 50
|
||||
assert len(_slugify(long)) <= 60
|
||||
|
||||
def test_hyphens(self):
|
||||
assert _slugify("high-speed beveling machine") == "high-speed-beveling-machine"
|
||||
|
||||
|
||||
class TestFuzzyCompanyMatch:
|
||||
def test_exact_match(self):
|
||||
assert _fuzzy_company_match("Chapter2", "Chapter2") is True
|
||||
|
||||
def test_case_insensitive(self):
|
||||
assert _fuzzy_company_match("chapter2", "Chapter2") is True
|
||||
|
||||
def test_substring_match(self):
|
||||
assert _fuzzy_company_match("Chapter 2", "Chapter 2 Incorporated") is True
|
||||
|
||||
def test_reverse_substring(self):
|
||||
assert _fuzzy_company_match("Chapter 2 Incorporated", "Chapter 2") is True
|
||||
|
||||
def test_no_match(self):
|
||||
assert _fuzzy_company_match("Chapter2", "Hogge Precision") is False
|
||||
|
||||
def test_empty(self):
|
||||
assert _fuzzy_company_match("", "Chapter2") is False
|
||||
assert _fuzzy_company_match("Chapter2", "") is False
|
||||
|
||||
|
||||
class TestCleanContent:
|
||||
def test_strips_preamble(self):
|
||||
raw = "Here is the guest article:\n\n# Title\nContent here."
|
||||
result = _clean_content(raw)
|
||||
assert result.startswith("# Title")
|
||||
|
||||
def test_strips_trailing_separator(self):
|
||||
raw = "Content here.\n---"
|
||||
result = _clean_content(raw)
|
||||
assert result == "Content here."
|
||||
|
||||
def test_strips_trailing_letmeknow(self):
|
||||
raw = "Content here.\nLet me know if you need any changes."
|
||||
result = _clean_content(raw)
|
||||
assert result == "Content here."
|
||||
|
||||
def test_passthrough_clean(self):
|
||||
raw = "# Title\n\nClean content."
|
||||
assert _clean_content(raw) == raw
|
||||
|
||||
|
||||
class TestLookupCompany:
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
def test_lookup_found(self, mock_file):
|
||||
mock_file.exists.return_value = True
|
||||
mock_file.read_text.return_value = SAMPLE_COMPANIES_MD
|
||||
|
||||
result = _lookup_company("Hogge Precision")
|
||||
assert result["name"] == "Hogge Precision"
|
||||
assert result["executive"] == "Danny Hogge Jr, President"
|
||||
assert result["pa_org_id"] == "19411"
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
def test_lookup_fuzzy(self, mock_file):
|
||||
mock_file.exists.return_value = True
|
||||
mock_file.read_text.return_value = SAMPLE_COMPANIES_MD
|
||||
|
||||
result = _lookup_company("Chapter 2")
|
||||
assert result["name"] == "Chapter 2 Incorporated"
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
def test_lookup_not_found(self, mock_file):
|
||||
mock_file.exists.return_value = True
|
||||
mock_file.read_text.return_value = SAMPLE_COMPANIES_MD
|
||||
|
||||
result = _lookup_company("Nonexistent Corp")
|
||||
assert result["name"] == "Nonexistent Corp"
|
||||
assert "executive" not in result
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
def test_lookup_no_file(self, mock_file):
|
||||
mock_file.exists.return_value = False
|
||||
|
||||
result = _lookup_company("Chapter2")
|
||||
assert result == {"name": "Chapter2"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompt builder tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPromptBuilders:
|
||||
def test_guest_article_prompt_includes_keyword(self):
|
||||
prompt = _build_guest_article_prompt(
|
||||
"precision cnc turning", "Chapter2", "https://chapter2.com", {}, ""
|
||||
)
|
||||
assert "precision cnc turning" in prompt
|
||||
assert "Chapter2" in prompt
|
||||
assert "500-700 word" in prompt
|
||||
|
||||
def test_guest_article_prompt_includes_url(self):
|
||||
prompt = _build_guest_article_prompt(
|
||||
"cnc machining", "Hogge", "https://hogge.com", {"executive": "Danny"}, ""
|
||||
)
|
||||
assert "https://hogge.com" in prompt
|
||||
assert "Danny" in prompt
|
||||
|
||||
def test_guest_article_prompt_includes_skill(self):
|
||||
prompt = _build_guest_article_prompt(
|
||||
"welding", "GullCo", "", {}, "Skill context here"
|
||||
)
|
||||
assert "Skill context here" in prompt
|
||||
|
||||
def test_directory_prompt_includes_fields(self):
|
||||
prompt = _build_directory_prompt(
|
||||
"cnc turning", "Chapter2", "https://ch2.com", "https://linkedin.com/ch2",
|
||||
{"executive": "Kyle Johnston"},
|
||||
)
|
||||
assert "cnc turning" in prompt
|
||||
assert "Chapter2" in prompt
|
||||
assert "Kyle Johnston" in prompt
|
||||
assert "150-200 words" in prompt
|
||||
|
||||
def test_social_post_prompt(self):
|
||||
prompt = _build_social_post_prompt(
|
||||
"cnc machining", "Hogge Precision", "https://hogge.com",
|
||||
"The Future of CNC Machining",
|
||||
)
|
||||
assert "LinkedIn" in prompt
|
||||
assert "Hogge Precision" in prompt
|
||||
assert "100-150 words" in prompt
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main tool integration tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildLinks:
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_generates_three_content_pieces(self, mock_output_dir, mock_companies, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
# Set up agent mocks for the three calls
|
||||
agent = mock_ctx["agent"]
|
||||
agent.execute_task.side_effect = [SAMPLE_GUEST_ARTICLE, SAMPLE_DIRECTORY_ENTRY]
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
result = build_links(
|
||||
keyword="precision cnc turning",
|
||||
company_name="Chapter2",
|
||||
target_url="https://chapter2inc.com",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
assert "Link Building Complete" in result
|
||||
assert "Guest Article" in result
|
||||
assert "Directory Listing" in result
|
||||
assert "Social Post" in result
|
||||
assert "3" in result # 3 deliverables
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_extracts_keyword_from_links_prefix(self, mock_output_dir, mock_companies, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
agent = mock_ctx["agent"]
|
||||
agent.execute_task.side_effect = [SAMPLE_GUEST_ARTICLE, SAMPLE_DIRECTORY_ENTRY]
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
result = build_links(
|
||||
keyword="LINKS - precision cnc turning",
|
||||
company_name="Chapter2",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
# The keyword should have been extracted, not passed as "LINKS - ..."
|
||||
assert "precision cnc turning" in result
|
||||
# The execute_task calls should use the extracted keyword
|
||||
call_args = agent.execute_task.call_args_list
|
||||
assert "precision cnc turning" in call_args[0][0][0]
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_saves_files_to_output_dir(self, mock_output_dir, mock_companies, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
agent = mock_ctx["agent"]
|
||||
agent.execute_task.side_effect = [SAMPLE_GUEST_ARTICLE, SAMPLE_DIRECTORY_ENTRY]
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
build_links(
|
||||
keyword="cnc turning",
|
||||
company_name="TestCo",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
# Check files were created
|
||||
company_dir = tmp_path / "testco" / "cnc-turning"
|
||||
assert (company_dir / "guest-article.md").exists()
|
||||
assert (company_dir / "directory-listing.md").exists()
|
||||
assert (company_dir / "social-post.md").exists()
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_handles_execution_failure_gracefully(self, mock_output_dir, mock_companies, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
agent = mock_ctx["agent"]
|
||||
agent.execute_task.side_effect = Exception("LLM timeout")
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
result = build_links(
|
||||
keyword="cnc turning",
|
||||
company_name="TestCo",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
# Should still complete with warnings instead of crashing
|
||||
assert "Warnings" in result
|
||||
assert "failed" in result.lower()
|
||||
|
||||
def test_no_agent_returns_error(self):
|
||||
result = build_links(keyword="test", company_name="Test", ctx={})
|
||||
assert "Error" in result
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_short_content_generates_warning(self, mock_output_dir, mock_companies, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
agent = mock_ctx["agent"]
|
||||
# Return very short content
|
||||
agent.execute_task.side_effect = ["Short.", "Also short."]
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": "Too brief."}])
|
||||
|
||||
result = build_links(
|
||||
keyword="test keyword",
|
||||
company_name="TestCo",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
assert "too short" in result.lower()
|
||||
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_sets_pipeline_status(self, mock_output_dir, mock_companies, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
agent = mock_ctx["agent"]
|
||||
agent.execute_task.side_effect = [SAMPLE_GUEST_ARTICLE, SAMPLE_DIRECTORY_ENTRY]
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
build_links(
|
||||
keyword="cnc turning",
|
||||
company_name="TestCo",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
# Should have set pipeline status multiple times
|
||||
db = mock_ctx["db"]
|
||||
status_calls = [c for c in db.kv_set.call_args_list if c[0][0] == "pipeline:status"]
|
||||
assert len(status_calls) >= 3 # At least once per step + clear
|
||||
|
||||
|
||||
class TestBuildLinksClickUpIntegration:
|
||||
@patch("cheddahbot.tools.linkbuilding._get_clickup_client")
|
||||
@patch("cheddahbot.tools.linkbuilding._COMPANIES_FILE")
|
||||
@patch("cheddahbot.tools.linkbuilding._OUTPUT_DIR")
|
||||
def test_clickup_sync_on_task_id(self, mock_output_dir, mock_companies,
|
||||
mock_get_client, tmp_path, mock_ctx):
|
||||
mock_output_dir.__truediv__ = lambda self, x: tmp_path / x
|
||||
mock_companies.exists.return_value = False
|
||||
|
||||
# Set up ClickUp mock
|
||||
cu_client = MagicMock()
|
||||
cu_client.upload_attachment.return_value = True
|
||||
cu_client.update_task_status.return_value = True
|
||||
cu_client.add_comment.return_value = True
|
||||
mock_get_client.return_value = cu_client
|
||||
|
||||
# Enable ClickUp in config
|
||||
mock_ctx["config"].clickup.enabled = True
|
||||
mock_ctx["config"].clickup.review_status = "internal review"
|
||||
mock_ctx["clickup_task_id"] = "task_abc123"
|
||||
|
||||
agent = mock_ctx["agent"]
|
||||
agent.execute_task.side_effect = [SAMPLE_GUEST_ARTICLE, SAMPLE_DIRECTORY_ENTRY]
|
||||
agent.llm.chat.return_value = iter([{"type": "text", "content": SAMPLE_SOCIAL_POST}])
|
||||
|
||||
result = build_links(
|
||||
keyword="cnc turning",
|
||||
company_name="TestCo",
|
||||
ctx=mock_ctx,
|
||||
)
|
||||
|
||||
assert "ClickUp Sync" in result
|
||||
cu_client.upload_attachment.assert_called()
|
||||
cu_client.update_task_status.assert_called_once_with("task_abc123", "internal review")
|
||||
cu_client.add_comment.assert_called_once()
|
||||
Loading…
Reference in New Issue