559 lines
24 KiB
Python
559 lines
24 KiB
Python
from basecamp_client import BasecampClient
|
|
import json
|
|
import logging
|
|
|
|
# Set up logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger('basecamp_search')
|
|
|
|
class BasecampSearch:
|
|
"""
|
|
Utility for searching across Basecamp 3 projects and to-dos.
|
|
"""
|
|
|
|
def __init__(self, client=None, **kwargs):
|
|
"""Initialize with either an existing client or credentials."""
|
|
if client:
|
|
self.client = client
|
|
else:
|
|
self.client = BasecampClient(**kwargs)
|
|
|
|
def search_projects(self, query=None):
|
|
"""
|
|
Search all projects, optionally filtering by name.
|
|
|
|
Args:
|
|
query (str, optional): Text to search for in project names
|
|
|
|
Returns:
|
|
list: Filtered list of projects
|
|
"""
|
|
try:
|
|
projects = self.client.get_projects()
|
|
|
|
if query and projects:
|
|
query = query.lower()
|
|
projects = [
|
|
project for project in projects
|
|
if query in project.get('name', '').lower() or
|
|
query in (project.get('description') or '').lower()
|
|
]
|
|
|
|
return projects
|
|
except Exception as e:
|
|
logger.error(f"Error searching projects: {str(e)}")
|
|
return []
|
|
|
|
def get_all_todolists(self, project_id=None):
|
|
"""
|
|
Get all todolists, either for a specific project or across all projects.
|
|
|
|
Args:
|
|
project_id (int, optional): Specific project ID or None for all projects
|
|
|
|
Returns:
|
|
list: List of todolists with project info
|
|
"""
|
|
all_todolists = []
|
|
|
|
try:
|
|
if project_id:
|
|
# Get todolists for a specific project
|
|
project = self.client.get_project(project_id)
|
|
todolists = self.client.get_todolists(project_id)
|
|
|
|
for todolist in todolists:
|
|
todolist['project'] = {'id': project['id'], 'name': project['name']}
|
|
all_todolists.append(todolist)
|
|
else:
|
|
# Get todolists across all projects
|
|
projects = self.client.get_projects()
|
|
|
|
for project in projects:
|
|
project_id = project['id']
|
|
try:
|
|
todolists = self.client.get_todolists(project_id)
|
|
for todolist in todolists:
|
|
todolist['project'] = {'id': project['id'], 'name': project['name']}
|
|
all_todolists.append(todolist)
|
|
except Exception as e:
|
|
logger.error(f"Error getting todolists for project {project_id}: {str(e)}")
|
|
except Exception as e:
|
|
logger.error(f"Error getting all todolists: {str(e)}")
|
|
|
|
return all_todolists
|
|
|
|
def search_todolists(self, query=None, project_id=None):
|
|
"""
|
|
Search all todolists, optionally filtering by name and project.
|
|
|
|
Args:
|
|
query (str, optional): Text to search for in todolist names
|
|
project_id (int, optional): Specific project ID or None for all projects
|
|
|
|
Returns:
|
|
list: Filtered list of todolists
|
|
"""
|
|
todolists = self.get_all_todolists(project_id)
|
|
|
|
if query and todolists:
|
|
query = query.lower()
|
|
todolists = [
|
|
todolist for todolist in todolists
|
|
if query in todolist.get('name', '').lower() or
|
|
query in (todolist.get('description') or '').lower()
|
|
]
|
|
|
|
return todolists
|
|
|
|
def get_all_todos(self, project_id=None, todolist_id=None, include_completed=False):
|
|
"""
|
|
Get all todos, with various filtering options.
|
|
|
|
Args:
|
|
project_id (int, optional): Specific project ID or None for all projects
|
|
todolist_id (int, optional): Specific todolist ID or None for all todolists
|
|
include_completed (bool): Whether to include completed todos
|
|
|
|
Returns:
|
|
list: List of todos with project and todolist info
|
|
"""
|
|
all_todos = []
|
|
|
|
try:
|
|
# Case 1: Specific todolist (regardless of project)
|
|
if todolist_id:
|
|
try:
|
|
todolist = self.client.get_todolist(todolist_id)
|
|
todos = self.client.get_todos(todolist_id)
|
|
|
|
# In Basecamp 3, we need to add project info to the todolist
|
|
# Get project ID from the URL
|
|
project_links = [link for link in todolist.get('bucket', {}).get('links', [])
|
|
if link.get('type') == 'project']
|
|
if project_links:
|
|
project_url = project_links[0].get('href', '')
|
|
# Extract project ID from URL
|
|
parts = project_url.split('/')
|
|
if len(parts) > 0:
|
|
project_id = parts[-1]
|
|
try:
|
|
project = self.client.get_project(project_id)
|
|
project_name = project.get('name', 'Unknown Project')
|
|
except:
|
|
project_name = 'Unknown Project'
|
|
else:
|
|
project_name = 'Unknown Project'
|
|
else:
|
|
project_name = 'Unknown Project'
|
|
|
|
for todo in todos:
|
|
if not include_completed and todo.get('completed'):
|
|
continue
|
|
|
|
todo['project'] = {'id': project_id, 'name': project_name}
|
|
todo['todolist'] = {'id': todolist['id'], 'name': todolist['name']}
|
|
all_todos.append(todo)
|
|
except Exception as e:
|
|
logger.error(f"Error getting todos for todolist {todolist_id}: {str(e)}")
|
|
|
|
# Case 2: Specific project, all todolists
|
|
elif project_id:
|
|
project = self.client.get_project(project_id)
|
|
todolists = self.client.get_todolists(project_id)
|
|
|
|
for todolist in todolists:
|
|
try:
|
|
todos = self.client.get_todos(todolist['id'])
|
|
for todo in todos:
|
|
if not include_completed and todo.get('completed'):
|
|
continue
|
|
|
|
todo['project'] = {'id': project['id'], 'name': project['name']}
|
|
todo['todolist'] = {'id': todolist['id'], 'name': todolist['name']}
|
|
all_todos.append(todo)
|
|
except Exception as e:
|
|
logger.error(f"Error getting todos for todolist {todolist['id']}: {str(e)}")
|
|
|
|
# Case 3: All projects
|
|
else:
|
|
todolists = self.get_all_todolists()
|
|
|
|
for todolist in todolists:
|
|
try:
|
|
todos = self.client.get_todos(todolist['id'])
|
|
for todo in todos:
|
|
if not include_completed and todo.get('completed'):
|
|
continue
|
|
|
|
todo['project'] = todolist['project']
|
|
todo['todolist'] = {'id': todolist['id'], 'name': todolist['name']}
|
|
all_todos.append(todo)
|
|
except Exception as e:
|
|
logger.error(f"Error getting todos for todolist {todolist['id']}: {str(e)}")
|
|
except Exception as e:
|
|
logger.error(f"Error getting all todos: {str(e)}")
|
|
|
|
return all_todos
|
|
|
|
def search_todos(self, query=None, project_id=None, todolist_id=None, include_completed=False):
|
|
"""
|
|
Search all todos, with various filtering options.
|
|
|
|
Args:
|
|
query (str, optional): Text to search for in todo content
|
|
project_id (int, optional): Specific project ID or None for all projects
|
|
todolist_id (int, optional): Specific todolist ID or None for all todolists
|
|
include_completed (bool): Whether to include completed todos
|
|
|
|
Returns:
|
|
list: Filtered list of todos
|
|
"""
|
|
todos = self.get_all_todos(project_id, todolist_id, include_completed)
|
|
|
|
if query and todos:
|
|
query = query.lower()
|
|
# In Basecamp 3, the todo content is in the 'content' field
|
|
todos = [
|
|
t for t in todos
|
|
if query in t.get('content', '').lower() or
|
|
query in (t.get('description') or '').lower()
|
|
]
|
|
|
|
return todos
|
|
|
|
def search_messages(self, query=None, project_id=None):
|
|
"""
|
|
Search for messages across all projects or within a specific project.
|
|
|
|
Args:
|
|
query (str, optional): Search term to filter messages
|
|
project_id (int, optional): If provided, only search within this project
|
|
|
|
Returns:
|
|
list: Matching messages
|
|
"""
|
|
all_messages = []
|
|
|
|
try:
|
|
# Get projects to search in
|
|
if project_id:
|
|
projects = [self.client.get_project(project_id)]
|
|
else:
|
|
projects = self.client.get_projects()
|
|
|
|
for project in projects:
|
|
project_id = project['id']
|
|
logger.info(f"Searching messages in project {project_id} ({project.get('name', 'Unknown')})")
|
|
|
|
# Check for message boards in the dock
|
|
has_message_board = False
|
|
message_boards = []
|
|
|
|
for dock_item in project.get('dock', []):
|
|
if dock_item.get('name') == 'message_board' and dock_item.get('enabled', False):
|
|
has_message_board = True
|
|
message_boards.append(dock_item)
|
|
|
|
if not has_message_board:
|
|
logger.info(f"Project {project_id} ({project.get('name', 'Unknown')}) has no enabled message boards")
|
|
continue
|
|
|
|
# Get messages from each message board
|
|
for board in message_boards:
|
|
board_id = board.get('id')
|
|
try:
|
|
# First try getting the message board details
|
|
logger.info(f"Fetching message board {board_id} for project {project_id}")
|
|
board_endpoint = f"buckets/{project_id}/message_boards/{board_id}.json"
|
|
board_details = self.client.get(board_endpoint)
|
|
|
|
# Then get all messages in the board
|
|
logger.info(f"Fetching messages for board {board_id} in project {project_id}")
|
|
messages_endpoint = f"buckets/{project_id}/message_boards/{board_id}/messages.json"
|
|
messages = self.client.get(messages_endpoint)
|
|
|
|
logger.info(f"Found {len(messages)} messages in board {board_id}")
|
|
|
|
# Now get detailed content for each message
|
|
for message in messages:
|
|
try:
|
|
message_id = message.get('id')
|
|
# Get detailed message content
|
|
message_endpoint = f"buckets/{project_id}/messages/{message_id}.json"
|
|
detailed_message = self.client.get(message_endpoint)
|
|
|
|
# Add project info
|
|
detailed_message['project'] = {
|
|
'id': project_id,
|
|
'name': project.get('name', 'Unknown Project')
|
|
}
|
|
|
|
# Add to results
|
|
all_messages.append(detailed_message)
|
|
except Exception as e:
|
|
logger.error(f"Error getting detailed message {message.get('id', 'unknown')} in project {project_id}: {str(e)}")
|
|
# Still include basic message info
|
|
message['project'] = {
|
|
'id': project_id,
|
|
'name': project.get('name', 'Unknown Project')
|
|
}
|
|
all_messages.append(message)
|
|
except Exception as e:
|
|
logger.error(f"Error getting messages for board {board_id} in project {project_id}: {str(e)}")
|
|
|
|
# Try alternate approach: get messages directly for the project
|
|
try:
|
|
logger.info(f"Trying alternate approach for project {project_id}")
|
|
messages = self.client.get_messages(project_id)
|
|
|
|
logger.info(f"Found {len(messages)} messages in project {project_id} using direct method")
|
|
|
|
# Add project info to each message
|
|
for message in messages:
|
|
message['project'] = {
|
|
'id': project_id,
|
|
'name': project.get('name', 'Unknown Project')
|
|
}
|
|
all_messages.append(message)
|
|
except Exception as e2:
|
|
logger.error(f"Error getting messages directly for project {project_id}: {str(e2)}")
|
|
|
|
# Also check for message categories/topics
|
|
try:
|
|
# Try to get message categories
|
|
categories_endpoint = f"buckets/{project_id}/categories.json"
|
|
categories = self.client.get(categories_endpoint)
|
|
|
|
for category in categories:
|
|
category_id = category.get('id')
|
|
try:
|
|
# Get messages in this category
|
|
category_messages_endpoint = f"buckets/{project_id}/categories/{category_id}/messages.json"
|
|
category_messages = self.client.get(category_messages_endpoint)
|
|
|
|
# Add project and category info
|
|
for message in category_messages:
|
|
message['project'] = {
|
|
'id': project_id,
|
|
'name': project.get('name', 'Unknown Project')
|
|
}
|
|
message['category'] = {
|
|
'id': category_id,
|
|
'name': category.get('name', 'Unknown Category')
|
|
}
|
|
all_messages.append(message)
|
|
except Exception as e:
|
|
logger.error(f"Error getting messages for category {category_id} in project {project_id}: {str(e)}")
|
|
except Exception as e:
|
|
logger.info(f"No message categories found for project {project_id}: {str(e)}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error searching messages: {str(e)}")
|
|
|
|
# Filter by query if provided
|
|
if query and all_messages:
|
|
query = query.lower()
|
|
filtered_messages = []
|
|
|
|
for message in all_messages:
|
|
# Search in multiple fields
|
|
content_matched = False
|
|
|
|
# Check title/subject
|
|
if query in (message.get('subject', '') or '').lower():
|
|
content_matched = True
|
|
|
|
# Check content field
|
|
if not content_matched and query in (message.get('content', '') or '').lower():
|
|
content_matched = True
|
|
|
|
# Check content field with HTML
|
|
if not content_matched and 'content' in message:
|
|
content_html = message.get('content')
|
|
if content_html and query in content_html.lower():
|
|
content_matched = True
|
|
|
|
# Check raw content in various formats
|
|
if not content_matched:
|
|
# Try different content field formats
|
|
for field in ['raw_content', 'content_html', 'body', 'description', 'text']:
|
|
if field in message and message[field]:
|
|
if query in str(message[field]).lower():
|
|
content_matched = True
|
|
break
|
|
|
|
# Check title field
|
|
if not content_matched and 'title' in message and message['title']:
|
|
if query in message['title'].lower():
|
|
content_matched = True
|
|
|
|
# Check creator's name
|
|
if not content_matched and 'creator' in message and message['creator']:
|
|
creator = message['creator']
|
|
creator_name = f"{creator.get('name', '')} {creator.get('first_name', '')} {creator.get('last_name', '')}"
|
|
if query in creator_name.lower():
|
|
content_matched = True
|
|
|
|
# Include if content matched
|
|
if content_matched:
|
|
filtered_messages.append(message)
|
|
|
|
logger.info(f"Found {len(filtered_messages)} messages matching query '{query}' out of {len(all_messages)} total messages")
|
|
return filtered_messages
|
|
|
|
return all_messages
|
|
|
|
def search_schedule_entries(self, query=None, project_id=None):
|
|
"""
|
|
Search schedule entries across projects or in a specific project.
|
|
|
|
Args:
|
|
query (str, optional): Search term to filter schedule entries
|
|
project_id (int, optional): Specific project ID to search in
|
|
|
|
Returns:
|
|
list: Matching schedule entries
|
|
"""
|
|
try:
|
|
# Get the schedule entries (from all projects or a specific one)
|
|
if project_id:
|
|
entries = self.client.get_schedule_entries(project_id)
|
|
entries = entries.json() if hasattr(entries, 'json') else entries
|
|
else:
|
|
# Get all projects first
|
|
projects = self.client.get_projects()
|
|
|
|
# Then get schedule entries from each
|
|
entries = []
|
|
for project in projects:
|
|
project_entries = self.client.get_schedule_entries(project['id'])
|
|
project_entries = project_entries.json() if hasattr(project_entries, 'json') else project_entries
|
|
if project_entries:
|
|
for entry in project_entries:
|
|
entry['project'] = {
|
|
'id': project['id'],
|
|
'name': project['name']
|
|
}
|
|
entries.extend(project_entries)
|
|
|
|
# Filter by query if provided
|
|
if query and entries:
|
|
query = query.lower()
|
|
entries = [
|
|
entry for entry in entries
|
|
if query in entry.get('title', '').lower() or
|
|
query in (entry.get('description') or '').lower() or
|
|
(entry.get('creator') and query in entry['creator'].get('name', '').lower())
|
|
]
|
|
|
|
return entries
|
|
except Exception as e:
|
|
logger.error(f"Error searching schedule entries: {str(e)}")
|
|
return []
|
|
|
|
def search_comments(self, query=None, recording_id=None, bucket_id=None):
|
|
"""
|
|
Search for comments across resources or for a specific resource.
|
|
|
|
Args:
|
|
query (str, optional): Search term to filter comments
|
|
recording_id (int, optional): ID of the recording (todo, message, etc.) to search in
|
|
bucket_id (int, optional): Project/bucket ID
|
|
|
|
Returns:
|
|
list: Matching comments
|
|
"""
|
|
try:
|
|
# If both recording_id and bucket_id are provided, get comments for that specific recording
|
|
if recording_id and bucket_id:
|
|
comments = self.client.get_comments(recording_id, bucket_id)
|
|
# Otherwise we can't search across all comments as there's no endpoint for that
|
|
else:
|
|
logger.warning("Cannot search all comments across Basecamp - both recording_id and bucket_id are required")
|
|
return [{
|
|
"content": "To search comments, you need to specify both a recording ID (todo, message, etc.) and a bucket ID. Comments cannot be searched globally in Basecamp.",
|
|
"api_limitation": True,
|
|
"title": "Comment Search Limitation"
|
|
}]
|
|
|
|
# Filter by query if provided
|
|
if query and comments:
|
|
query = query.lower()
|
|
|
|
filtered_comments = []
|
|
for comment in comments:
|
|
# Check content
|
|
content_matched = False
|
|
content = comment.get('content', '')
|
|
if content and query in content.lower():
|
|
content_matched = True
|
|
|
|
# Check creator name
|
|
if not content_matched and comment.get('creator'):
|
|
creator_name = comment['creator'].get('name', '')
|
|
if creator_name and query in creator_name.lower():
|
|
content_matched = True
|
|
|
|
# If matched, add to results
|
|
if content_matched:
|
|
filtered_comments.append(comment)
|
|
|
|
return filtered_comments
|
|
|
|
return comments
|
|
except Exception as e:
|
|
logger.error(f"Error searching comments: {str(e)}")
|
|
return []
|
|
|
|
def search_campfire_lines(self, query=None, project_id=None, campfire_id=None):
|
|
"""
|
|
Search for lines in campfire chats.
|
|
|
|
Args:
|
|
query (str, optional): Search term to filter lines
|
|
project_id (int, optional): Project ID
|
|
campfire_id (int, optional): Campfire ID
|
|
|
|
Returns:
|
|
list: Matching chat lines
|
|
"""
|
|
try:
|
|
if not project_id or not campfire_id:
|
|
logger.warning("Cannot search campfire lines without project_id and campfire_id")
|
|
return [{
|
|
"content": "To search campfire lines, you need to specify both a project ID and a campfire ID.",
|
|
"api_limitation": True,
|
|
"title": "Campfire Search Limitation"
|
|
}]
|
|
|
|
lines = self.client.get_campfire_lines(project_id, campfire_id)
|
|
|
|
if query and lines:
|
|
query = query.lower()
|
|
|
|
filtered_lines = []
|
|
for line in lines:
|
|
# Check content
|
|
content_matched = False
|
|
content = line.get('content', '')
|
|
if content and query in content.lower():
|
|
content_matched = True
|
|
|
|
# Check creator name
|
|
if not content_matched and line.get('creator'):
|
|
creator_name = line['creator'].get('name', '')
|
|
if creator_name and query in creator_name.lower():
|
|
content_matched = True
|
|
|
|
# If matched, add to results
|
|
if content_matched:
|
|
filtered_lines.append(line)
|
|
|
|
return filtered_lines
|
|
|
|
return lines
|
|
except Exception as e:
|
|
logger.error(f"Error searching campfire lines: {str(e)}")
|
|
return []
|