diff --git a/cheddahbot/tools/linkbuilding.py b/cheddahbot/tools/linkbuilding.py new file mode 100644 index 0000000..aa7b852 --- /dev/null +++ b/cheddahbot/tools/linkbuilding.py @@ -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 diff --git a/config.yaml b/config.yaml index 3bb2eef..df5eb4d 100644 --- a/config.yaml +++ b/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: "" diff --git a/skills/linkbuilding.md b/skills/linkbuilding.md new file mode 100644 index 0000000..2b7a9d2 --- /dev/null +++ b/skills/linkbuilding.md @@ -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 diff --git a/tests/test_linkbuilding.py b/tests/test_linkbuilding.py new file mode 100644 index 0000000..17b3daf --- /dev/null +++ b/tests/test_linkbuilding.py @@ -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()