#!/usr/bin/env python3 """ Rivet Ideas Database API Helper Interact with the Notion "Rivet Ideas" database programmatically. """ import json import os import requests from datetime import datetime from typing import Optional, List, Dict, Any # Load configuration CONFIG_FILE = os.path.join(os.path.dirname(__file__), "notion_rivet_ideas_config.json") class RivetIdeasDB: """Client for interacting with the Rivet Ideas Notion database.""" def __init__(self): self.api_key = os.environ.get('NOTION_API_KEY') if not self.api_key: raise ValueError("NOTION_API_KEY environment variable is required") # Load config with open(CONFIG_FILE, 'r') as f: self.config = json.load(f) self.database_id = self.config['database_id'] self.base_url = "https://api.notion.com/v1" self.headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "Notion-Version": "2022-06-28" } def add_idea(self, title: str, description: str, rationale: str, priority: str = "Medium", effort: str = "Medium", categories: List[str] = None, status: str = "Proposed") -> Dict[str, Any]: """ Add a new idea to the database. Args: title: Name of the idea description: Full explanation rationale: Business case / why it matters priority: High, Medium, or Low effort: Small (<1 day), Medium (1-3 days), Large (>3 days) categories: List of categories from: Feature, Fix, Marketing, Operations, Technical, Business Model status: Proposed, Research, Investigate, Plan, Approved, or Rejected """ url = f"{self.base_url}/pages" properties = { "Title": {"title": [{"text": {"content": title}}]}, "Description": {"rich_text": [{"text": {"content": description}}]}, "Rationale": {"rich_text": [{"text": {"content": rationale}}]}, "Status": {"select": {"name": status}}, "Priority": {"select": {"name": priority}}, "Effort": {"select": {"name": effort}} } if categories: properties["Category"] = {"multi_select": [{"name": cat} for cat in categories]} payload = { "parent": {"database_id": self.database_id}, "properties": properties } response = requests.post(url, headers=self.headers, json=payload) if response.status_code == 200: result = response.json() print(f"✅ Idea added successfully!") print(f" ID: {result['id']}") print(f" URL: {result['url']}") return result else: print(f"❌ Error adding idea: {response.status_code}") print(response.text) return None def query_ideas(self, status: Optional[str] = None, priority: Optional[str] = None, category: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: """ Query ideas from the database with optional filters. Args: status: Filter by status priority: Filter by priority category: Filter by category limit: Maximum number of results """ url = f"{self.base_url}/databases/{self.database_id}/query" filters = [] if status: filters.append({"property": "Status", "select": {"equals": status}}) if priority: filters.append({"property": "Priority", "select": {"equals": priority}}) if category: filters.append({"property": "Category", "multi_select": {"contains": category}}) payload = {"page_size": limit} if len(filters) == 1: payload["filter"] = filters[0] elif len(filters) > 1: payload["filter"] = {"and": filters} response = requests.post(url, headers=self.headers, json=payload) if response.status_code == 200: results = response.json().get("results", []) print(f"✅ Found {len(results)} ideas") return results else: print(f"❌ Error querying ideas: {response.status_code}") print(response.text) return [] def update_idea_status(self, page_id: str, status: str) -> bool: """Update the status of an existing idea.""" url = f"{self.base_url}/pages/{page_id}" payload = { "properties": { "Status": {"select": {"name": status}} } } response = requests.patch(url, headers=self.headers, json=payload) if response.status_code == 200: print(f"✅ Status updated to '{status}'") return True else: print(f"❌ Error updating status: {response.status_code}") return False def get_approved_ideas(self) -> List[Dict[str, Any]]: """Get all approved ideas (ready for implementation).""" return self.query_ideas(status="Approved") def get_high_priority_ideas(self) -> List[Dict[str, Any]]: """Get all high priority ideas.""" return self.query_ideas(priority="High") def list_all_ideas(self) -> None: """Print a formatted list of all ideas.""" ideas = self.query_ideas() print(f"\n{'='*60}") print(f"RIVET IDEAS DATABASE ({len(ideas)} total)") print(f"{'='*60}\n") for idea in ideas: props = idea['properties'] title = props['Title']['title'][0]['text']['content'] if props['Title']['title'] else "Untitled" status = props['Status']['select']['name'] if props['Status']['select'] else "Unknown" priority = props['Priority']['select']['name'] if props['Priority']['select'] else "Unknown" effort = props['Effort']['select']['name'] if props['Effort']['select'] else "Unknown" print(f"📌 {title}") print(f" Status: {status} | Priority: {priority} | Effort: {effort}") print(f" ID: {idea['id']}") print() def main(): """CLI interface for the Rivet Ideas database.""" import argparse parser = argparse.ArgumentParser(description='Rivet Ideas Database CLI') subparsers = parser.add_subparsers(dest='command', help='Commands') # Add idea command add_parser = subparsers.add_parser('add', help='Add a new idea') add_parser.add_argument('--title', required=True, help='Idea title') add_parser.add_argument('--description', required=True, help='Idea description') add_parser.add_argument('--rationale', required=True, help='Business rationale') add_parser.add_argument('--priority', default='Medium', choices=['High', 'Medium', 'Low']) add_parser.add_argument('--effort', default='Medium', choices=['Small', 'Medium', 'Large']) add_parser.add_argument('--categories', nargs='+', help='Categories (space-separated)') add_parser.add_argument('--status', default='Proposed', choices=['Proposed', 'Research', 'Investigate', 'Plan', 'Approved', 'Rejected']) # List command subparsers.add_parser('list', help='List all ideas') # Query command query_parser = subparsers.add_parser('query', help='Query ideas with filters') query_parser.add_argument('--status', help='Filter by status') query_parser.add_argument('--priority', help='Filter by priority') query_parser.add_argument('--category', help='Filter by category') # Approved command subparsers.add_parser('approved', help='List approved ideas') # Update command update_parser = subparsers.add_parser('update', help='Update idea status') update_parser.add_argument('--id', required=True, help='Page ID of the idea') update_parser.add_argument('--status', required=True, choices=['Proposed', 'Research', 'Investigate', 'Plan', 'Approved', 'Rejected']) args = parser.parse_args() if not args.command: parser.print_help() return try: db = RivetIdeasDB() except ValueError as e: print(f"❌ {e}") print("\nTo use this tool, set your Notion API key:") print("export NOTION_API_KEY='your-integration-token'") return if args.command == 'add': db.add_idea( title=args.title, description=args.description, rationale=args.rationale, priority=args.priority, effort=args.effort, categories=args.categories, status=args.status ) elif args.command == 'list': db.list_all_ideas() elif args.command == 'query': ideas = db.query_ideas(status=args.status, priority=args.priority, category=args.category) for idea in ideas: props = idea['properties'] title = props['Title']['title'][0]['text']['content'] if props['Title']['title'] else "Untitled" print(f"- {title} (ID: {idea['id']})") elif args.command == 'approved': ideas = db.get_approved_ideas() print(f"\n📋 APPROVED IDEAS ({len(ideas)}):\n") for idea in ideas: props = idea['properties'] title = props['Title']['title'][0]['text']['content'] if props['Title']['title'] else "Untitled" print(f"✅ {title}") print(f" URL: {idea['url']}\n") elif args.command == 'update': db.update_idea_status(args.id, args.status) if __name__ == "__main__": main()