242 lines
8.0 KiB
Python
242 lines
8.0 KiB
Python
"""Run the press-release pipeline for up to N ClickUp tasks.
|
|
|
|
Usage:
|
|
uv run python scripts/run_pr_pipeline.py # discover + execute up to 3
|
|
uv run python scripts/run_pr_pipeline.py --dry-run # discover only, don't execute
|
|
uv run python scripts/run_pr_pipeline.py --max 1 # execute only 1 task
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
log = logging.getLogger("pr_pipeline")
|
|
|
|
# ── Bootstrap CheddahBot (config, db, agent, tools) ─────────────────────
|
|
|
|
from cheddahbot.config import load_config
|
|
from cheddahbot.db import Database
|
|
from cheddahbot.llm import LLMAdapter
|
|
from cheddahbot.agent import Agent
|
|
from cheddahbot.clickup import ClickUpClient
|
|
|
|
|
|
def bootstrap():
|
|
"""Set up config, db, agent, and tool registry — same as __main__.py."""
|
|
config = load_config()
|
|
db = Database(config.db_path)
|
|
llm = LLMAdapter(
|
|
default_model=config.chat_model,
|
|
openrouter_key=config.openrouter_api_key,
|
|
ollama_url=config.ollama_url,
|
|
lmstudio_url=config.lmstudio_url,
|
|
)
|
|
|
|
agent_cfg = config.agents[0] if config.agents else None
|
|
agent = Agent(config, db, llm, agent_config=agent_cfg)
|
|
|
|
# Memory
|
|
try:
|
|
from cheddahbot.memory import MemorySystem
|
|
scope = agent_cfg.memory_scope if agent_cfg else ""
|
|
memory = MemorySystem(config, db, scope=scope)
|
|
agent.set_memory(memory)
|
|
except Exception as e:
|
|
log.warning("Memory not available: %s", e)
|
|
|
|
# Tools
|
|
from cheddahbot.tools import ToolRegistry
|
|
tools = ToolRegistry(config, db, agent)
|
|
agent.set_tools(tools)
|
|
|
|
# Skills
|
|
try:
|
|
from cheddahbot.skills import SkillRegistry
|
|
skills = SkillRegistry(config.skills_dir)
|
|
agent.set_skills_registry(skills)
|
|
except Exception as e:
|
|
log.warning("Skills not available: %s", e)
|
|
|
|
return config, db, agent, tools
|
|
|
|
|
|
def discover_pr_tasks(config):
|
|
"""Poll ClickUp for Press Release tasks — same logic as scheduler._poll_clickup()."""
|
|
client = ClickUpClient(
|
|
api_token=config.clickup.api_token,
|
|
workspace_id=config.clickup.workspace_id,
|
|
task_type_field_name=config.clickup.task_type_field_name,
|
|
)
|
|
space_id = config.clickup.space_id
|
|
skill_map = config.clickup.skill_map
|
|
|
|
if not space_id:
|
|
log.error("No space_id configured")
|
|
return [], client
|
|
|
|
# Discover field filter (Work Category UUID + options)
|
|
list_ids = client.get_list_ids_from_space(space_id)
|
|
if not list_ids:
|
|
log.error("No lists found in space %s", space_id)
|
|
return [], client
|
|
|
|
first_list = next(iter(list_ids))
|
|
field_filter = client.discover_field_filter(
|
|
first_list, config.clickup.task_type_field_name
|
|
)
|
|
|
|
# Build custom fields filter for API query
|
|
custom_fields_filter = None
|
|
if field_filter and field_filter.get("options"):
|
|
import json
|
|
field_id = field_filter["field_id"]
|
|
options = field_filter["options"]
|
|
# Only Press Release
|
|
pr_opt_id = options.get("Press Release")
|
|
if pr_opt_id:
|
|
custom_fields_filter = json.dumps(
|
|
[{"field_id": field_id, "operator": "ANY", "value": [pr_opt_id]}]
|
|
)
|
|
log.info("Filtering for Press Release option ID: %s", pr_opt_id)
|
|
else:
|
|
log.warning("'Press Release' not found in Work Category options: %s", list(options.keys()))
|
|
return [], client
|
|
|
|
# Due date window (3 weeks)
|
|
now_ms = int(datetime.now(UTC).timestamp() * 1000)
|
|
due_date_lt = now_ms + (3 * 7 * 24 * 60 * 60 * 1000)
|
|
|
|
tasks = client.get_tasks_from_space(
|
|
space_id,
|
|
statuses=config.clickup.poll_statuses,
|
|
due_date_lt=due_date_lt,
|
|
custom_fields=custom_fields_filter,
|
|
)
|
|
|
|
# Client-side filter: must be Press Release + have due date in window
|
|
pr_tasks = []
|
|
for task in tasks:
|
|
if task.task_type != "Press Release":
|
|
continue
|
|
if not task.due_date:
|
|
continue
|
|
try:
|
|
if int(task.due_date) > due_date_lt:
|
|
continue
|
|
except (ValueError, TypeError):
|
|
continue
|
|
pr_tasks.append(task)
|
|
|
|
return pr_tasks, client
|
|
|
|
|
|
def execute_task(agent, tools, config, client, task):
|
|
"""Execute a single PR task — same logic as scheduler._execute_task()."""
|
|
skill_map = config.clickup.skill_map
|
|
mapping = skill_map.get("Press Release", {})
|
|
tool_name = mapping.get("tool", "write_press_releases")
|
|
|
|
task_id = task.id
|
|
|
|
# Build tool args from field mapping
|
|
field_mapping = mapping.get("field_mapping", {})
|
|
args = {}
|
|
for tool_param, source in field_mapping.items():
|
|
if source == "task_name":
|
|
args[tool_param] = task.name
|
|
elif source == "task_description":
|
|
args[tool_param] = task.custom_fields.get("description", "")
|
|
else:
|
|
args[tool_param] = task.custom_fields.get(source, "")
|
|
|
|
args["clickup_task_id"] = task_id
|
|
|
|
log.info("=" * 70)
|
|
log.info("EXECUTING: %s", task.name)
|
|
log.info(" Task ID: %s", task_id)
|
|
log.info(" Tool: %s", tool_name)
|
|
log.info(" Args: %s", {k: v for k, v in args.items() if k != "clickup_task_id"})
|
|
log.info("=" * 70)
|
|
|
|
# Move to "automation underway"
|
|
client.update_task_status(task_id, config.clickup.automation_status)
|
|
|
|
try:
|
|
result = tools.execute(tool_name, args)
|
|
|
|
if result.startswith("Skipped:") or result.startswith("Error:"):
|
|
log.error("Task skipped/errored: %s", result[:500])
|
|
client.add_comment(
|
|
task_id,
|
|
f"⚠️ CheddahBot could not execute this task.\n\n{result[:2000]}",
|
|
)
|
|
client.update_task_status(task_id, config.clickup.error_status)
|
|
return False
|
|
|
|
log.info("Task completed successfully!")
|
|
log.info("Result preview:\n%s", result[:1000])
|
|
return True
|
|
|
|
except Exception as e:
|
|
log.error("Task failed with exception: %s", e, exc_info=True)
|
|
client.add_comment(
|
|
task_id,
|
|
f"❌ CheddahBot failed to complete this task.\n\nError: {str(e)[:2000]}",
|
|
)
|
|
client.update_task_status(task_id, config.clickup.error_status)
|
|
return False
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Run PR pipeline from ClickUp")
|
|
parser.add_argument("--dry-run", action="store_true", help="Discover only, don't execute")
|
|
parser.add_argument("--max", type=int, default=3, help="Max tasks to execute (default: 3)")
|
|
args = parser.parse_args()
|
|
|
|
log.info("Bootstrapping CheddahBot...")
|
|
config, db, agent, tools = bootstrap()
|
|
|
|
log.info("Polling ClickUp for Press Release tasks...")
|
|
pr_tasks, client = discover_pr_tasks(config)
|
|
|
|
if not pr_tasks:
|
|
log.info("No Press Release tasks found in statuses %s", config.clickup.poll_statuses)
|
|
return
|
|
|
|
log.info("Found %d Press Release task(s):", len(pr_tasks))
|
|
for i, task in enumerate(pr_tasks):
|
|
status_str = f"status={task.status}" if hasattr(task, "status") else ""
|
|
log.info(" %d. %s (id=%s) %s", i + 1, task.name, task.id, status_str)
|
|
log.info(" Custom fields: %s", task.custom_fields)
|
|
|
|
if args.dry_run:
|
|
log.info("Dry run — not executing. Use without --dry-run to execute.")
|
|
return
|
|
|
|
# Execute up to --max tasks
|
|
to_run = pr_tasks[: args.max]
|
|
log.info("Will execute %d task(s) (max=%d)", len(to_run), args.max)
|
|
|
|
results = []
|
|
for i, task in enumerate(to_run):
|
|
log.info("\n>>> Task %d/%d <<<", i + 1, len(to_run))
|
|
success = execute_task(agent, tools, config, client, task)
|
|
results.append((task.name, success))
|
|
|
|
log.info("\n" + "=" * 70)
|
|
log.info("RESULTS SUMMARY")
|
|
log.info("=" * 70)
|
|
for name, success in results:
|
|
status = "OK" if success else "FAILED"
|
|
log.info(" [%s] %s", status, name)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|