From 0becf1dd89dcc84f058ae569d894b943266c1dfc Mon Sep 17 00:00:00 2001 From: PeninsulaInd Date: Thu, 19 Feb 2026 18:38:51 -0600 Subject: [PATCH] Revert "Add link building agent with content generation pipeline" This reverts commit 8a98b37725ddee4c89e96e01ba941df7f8a40b9b. --- cheddahbot/tools/linkbuilding.py | 520 ------------------------------- config.yaml | 15 - skills/linkbuilding.md | 42 --- tests/test_linkbuilding.py | 466 --------------------------- 4 files changed, 1043 deletions(-) delete mode 100644 cheddahbot/tools/linkbuilding.py delete mode 100644 skills/linkbuilding.md delete mode 100644 tests/test_linkbuilding.py diff --git a/cheddahbot/tools/linkbuilding.py b/cheddahbot/tools/linkbuilding.py deleted file mode 100644 index aa7b852..0000000 --- a/cheddahbot/tools/linkbuilding.py +++ /dev/null @@ -1,520 +0,0 @@ -"""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 df5eb4d..3bb2eef 100644 --- a/config.yaml +++ b/config.yaml @@ -56,14 +56,6 @@ 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. @@ -93,10 +85,3 @@ 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 deleted file mode 100644 index 2b7a9d2..0000000 --- a/skills/linkbuilding.md +++ /dev/null @@ -1,42 +0,0 @@ ---- -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 deleted file mode 100644 index 17b3daf..0000000 --- a/tests/test_linkbuilding.py +++ /dev/null @@ -1,466 +0,0 @@ -"""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()