From ff3114b51596b9a7c8e6756b05770f65924c27aa Mon Sep 17 00:00:00 2001 From: PeninsulaInd Date: Mon, 16 Feb 2026 18:04:23 -0600 Subject: [PATCH] Add .docx attachment uploads to ClickUp tasks after press release generation When a press release task completes, docx file paths are extracted from the tool output and uploaded as attachments to the ClickUp task. The completion comment now includes an attachment count note. Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 4 +++- cheddahbot/clickup.py | 27 +++++++++++++++++++++ cheddahbot/scheduler.py | 22 ++++++++++++++++- tests/test_clickup.py | 42 ++++++++++++++++++++++++++++++++- tests/test_scheduler_helpers.py | 36 ++++++++++++++++++++++++++++ 5 files changed, 128 insertions(+), 3 deletions(-) create mode 100644 tests/test_scheduler_helpers.py diff --git a/CLAUDE.md b/CLAUDE.md index 1bc9082..3f159ed 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -94,6 +94,7 @@ uv add --group test - **ClickUp field mapping**: `Work Category` field (not `Task Type`) identifies task types like "Press Release", "Link Building". The `Client` field (not `Company`) holds the client name. - **Notifications**: All scheduler events go through `NotificationBus.push()`, never directly to a UI - **Tests**: Use `respx` to mock httpx calls, `tmp_db` fixture for isolated SQLite instances +- **ClickUp attachments**: `ClickUpClient.upload_attachment()` uses module-level `httpx.post()` (not the shared client) for multipart uploads ## ClickUp Skill Mapping @@ -109,7 +110,7 @@ skill_map: company_name: "Client" # looks up "Client" custom field ``` -Task lifecycle: `to do` → discovered → approved/awaiting_approval → executing → completed/failed +Task lifecycle: `to do` → discovered → approved/awaiting_approval → executing → completed/failed (+ attachments uploaded) ## Testing @@ -122,6 +123,7 @@ Tests live in `tests/` and use pytest. All tests run offline with mocked APIs. - `test_email.py` — EmailClient SMTP send + attachments (mocked) - `test_docx_export.py` — Plain text → .docx formatting and file creation - `test_press_advantage.py` — Press Advantage API client, company parsing, link building, submit tool +- `test_scheduler_helpers.py` — `_extract_docx_paths` regex extraction from tool output Fixtures in `conftest.py`: `tmp_db` (fresh SQLite), `sample_clickup_task_data` (realistic API response). diff --git a/cheddahbot/clickup.py b/cheddahbot/clickup.py index 70cafe6..51e95b9 100644 --- a/cheddahbot/clickup.py +++ b/cheddahbot/clickup.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging from dataclasses import dataclass, field +from pathlib import Path from typing import Any import httpx @@ -170,6 +171,32 @@ class ClickUpClient: log.error("Failed to add comment to task %s: %s", task_id, e) return False + def upload_attachment(self, task_id: str, file_path: str | Path) -> bool: + """Upload a file attachment to a task. + + Uses module-level httpx.post() instead of self._client because the + shared client sets Content-Type: application/json which conflicts + with multipart/form-data uploads. + """ + fp = Path(file_path) + if not fp.exists(): + log.warning("Attachment file not found: %s", fp) + return False + try: + with open(fp, "rb") as f: + resp = httpx.post( + f"{BASE_URL}/task/{task_id}/attachment", + headers={"Authorization": self._token}, + files={"attachment": (fp.name, f, "application/octet-stream")}, + timeout=60.0, + ) + resp.raise_for_status() + log.info("Uploaded attachment %s to task %s", fp.name, task_id) + return True + except httpx.HTTPStatusError as e: + log.warning("Failed to upload attachment to task %s: %s", task_id, e) + return False + def get_custom_fields(self, list_id: str) -> list[dict]: """Get custom fields for a list.""" try: diff --git a/cheddahbot/scheduler.py b/cheddahbot/scheduler.py index fb286cc..94ed849 100644 --- a/cheddahbot/scheduler.py +++ b/cheddahbot/scheduler.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import logging +import re import threading from datetime import datetime, timezone from typing import TYPE_CHECKING @@ -20,6 +21,14 @@ log = logging.getLogger(__name__) HEARTBEAT_OK = "HEARTBEAT_OK" +# Matches **Docx:** `path/to/file.docx` patterns in tool output +_DOCX_PATH_RE = re.compile(r"\*\*Docx:\*\*\s*`([^`]+\.docx)`") + + +def _extract_docx_paths(result: str) -> list[str]: + """Extract .docx file paths from a tool result string.""" + return _DOCX_PATH_RE.findall(result) + class Scheduler: def __init__(self, config: Config, db: Database, agent: Agent, @@ -316,6 +325,16 @@ class Scheduler: f"Task description: {state.get('custom_fields', {})}" ) + # Extract and upload any docx deliverables + docx_paths = _extract_docx_paths(result) + state["deliverable_paths"] = docx_paths + uploaded_count = 0 + for path in docx_paths: + if client.upload_attachment(task_id, path): + uploaded_count += 1 + else: + log.warning("Failed to upload %s for task %s", path, task_id) + # Success state["state"] = "completed" state["completed_at"] = datetime.now(timezone.utc).isoformat() @@ -323,10 +342,11 @@ class Scheduler: # Update ClickUp client.update_task_status(task_id, self.config.clickup.review_status) + attach_note = f"\n📎 {uploaded_count} file(s) attached." if uploaded_count else "" comment = ( f"✅ CheddahBot completed this task.\n\n" f"Skill: {skill_name}\n" - f"Result:\n{result[:3000]}" + f"Result:\n{result[:3000]}{attach_note}" ) client.add_comment(task_id, comment) diff --git a/tests/test_clickup.py b/tests/test_clickup.py index ba74e97..0c37422 100644 --- a/tests/test_clickup.py +++ b/tests/test_clickup.py @@ -3,7 +3,6 @@ from __future__ import annotations import httpx -import pytest import respx from cheddahbot.clickup import BASE_URL, ClickUpClient, ClickUpTask @@ -256,6 +255,47 @@ class TestClickUpClient: assert fields[0]["name"] == "Task Type" client.close() + @respx.mock + def test_upload_attachment_success(self, tmp_path): + docx_file = tmp_path / "report.docx" + docx_file.write_bytes(b"fake docx content") + + respx.post(f"{BASE_URL}/task/t1/attachment").mock( + return_value=httpx.Response(200, json={}) + ) + + client = ClickUpClient(api_token="pk_test_123") + result = client.upload_attachment("t1", docx_file) + + assert result is True + request = respx.calls.last.request + assert b"report.docx" in request.content + # Multipart upload should NOT have application/json content-type + assert "multipart/form-data" in request.headers.get("content-type", "") + client.close() + + @respx.mock + def test_upload_attachment_http_failure(self, tmp_path): + docx_file = tmp_path / "report.docx" + docx_file.write_bytes(b"fake docx content") + + respx.post(f"{BASE_URL}/task/t1/attachment").mock( + return_value=httpx.Response(403, json={"err": "forbidden"}) + ) + + client = ClickUpClient(api_token="pk_test_123") + result = client.upload_attachment("t1", docx_file) + + assert result is False + client.close() + + def test_upload_attachment_file_not_found(self): + client = ClickUpClient(api_token="pk_test_123") + result = client.upload_attachment("t1", "/nonexistent/file.docx") + + assert result is False + client.close() + @respx.mock def test_auth_header_sent(self): respx.get(f"{BASE_URL}/list/list_1/task").mock( diff --git a/tests/test_scheduler_helpers.py b/tests/test_scheduler_helpers.py new file mode 100644 index 0000000..2ea6028 --- /dev/null +++ b/tests/test_scheduler_helpers.py @@ -0,0 +1,36 @@ +"""Tests for scheduler helper functions.""" + +from __future__ import annotations + +from cheddahbot.scheduler import _extract_docx_paths + + +class TestExtractDocxPaths: + def test_extracts_paths_from_realistic_output(self): + result = ( + "Press releases generated successfully!\n\n" + "**Docx:** `output/press_releases/acme-corp-launch.docx`\n" + "**Docx:** `output/press_releases/acme-corp-expansion.docx`\n" + "Files saved to output/press_releases/" + ) + paths = _extract_docx_paths(result) + + assert len(paths) == 2 + assert paths[0] == "output/press_releases/acme-corp-launch.docx" + assert paths[1] == "output/press_releases/acme-corp-expansion.docx" + + def test_returns_empty_list_when_no_paths(self): + result = "Task completed successfully. No files generated." + paths = _extract_docx_paths(result) + + assert paths == [] + + def test_only_matches_docx_extension(self): + result = ( + "**Docx:** `report.docx`\n" + "**PDF:** `report.pdf`\n" + "**Docx:** `summary.txt`\n" + ) + paths = _extract_docx_paths(result) + + assert paths == ["report.docx"]