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 <noreply@anthropic.com>
cora-start
PeninsulaInd 2026-02-16 18:04:23 -06:00
parent 388c800bce
commit ff3114b515
5 changed files with 128 additions and 3 deletions

View File

@ -94,6 +94,7 @@ uv add --group test <package>
- **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. - **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 - **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 - **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 ## ClickUp Skill Mapping
@ -109,7 +110,7 @@ skill_map:
company_name: "Client" # looks up "Client" custom field 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 ## 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_email.py` — EmailClient SMTP send + attachments (mocked)
- `test_docx_export.py` — Plain text → .docx formatting and file creation - `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_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). Fixtures in `conftest.py`: `tmp_db` (fresh SQLite), `sample_clickup_task_data` (realistic API response).

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import logging import logging
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path
from typing import Any from typing import Any
import httpx import httpx
@ -170,6 +171,32 @@ class ClickUpClient:
log.error("Failed to add comment to task %s: %s", task_id, e) log.error("Failed to add comment to task %s: %s", task_id, e)
return False 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]: def get_custom_fields(self, list_id: str) -> list[dict]:
"""Get custom fields for a list.""" """Get custom fields for a list."""
try: try:

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import json import json
import logging import logging
import re
import threading import threading
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
@ -20,6 +21,14 @@ log = logging.getLogger(__name__)
HEARTBEAT_OK = "HEARTBEAT_OK" 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: class Scheduler:
def __init__(self, config: Config, db: Database, agent: Agent, def __init__(self, config: Config, db: Database, agent: Agent,
@ -316,6 +325,16 @@ class Scheduler:
f"Task description: {state.get('custom_fields', {})}" 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 # Success
state["state"] = "completed" state["state"] = "completed"
state["completed_at"] = datetime.now(timezone.utc).isoformat() state["completed_at"] = datetime.now(timezone.utc).isoformat()
@ -323,10 +342,11 @@ class Scheduler:
# Update ClickUp # Update ClickUp
client.update_task_status(task_id, self.config.clickup.review_status) 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 = ( comment = (
f"✅ CheddahBot completed this task.\n\n" f"✅ CheddahBot completed this task.\n\n"
f"Skill: {skill_name}\n" f"Skill: {skill_name}\n"
f"Result:\n{result[:3000]}" f"Result:\n{result[:3000]}{attach_note}"
) )
client.add_comment(task_id, comment) client.add_comment(task_id, comment)

View File

@ -3,7 +3,6 @@
from __future__ import annotations from __future__ import annotations
import httpx import httpx
import pytest
import respx import respx
from cheddahbot.clickup import BASE_URL, ClickUpClient, ClickUpTask from cheddahbot.clickup import BASE_URL, ClickUpClient, ClickUpTask
@ -256,6 +255,47 @@ class TestClickUpClient:
assert fields[0]["name"] == "Task Type" assert fields[0]["name"] == "Task Type"
client.close() 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 @respx.mock
def test_auth_header_sent(self): def test_auth_header_sent(self):
respx.get(f"{BASE_URL}/list/list_1/task").mock( respx.get(f"{BASE_URL}/list/list_1/task").mock(

View File

@ -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"]