feat(docs): add comprehensive Google Docs editing capabilities
Add full document editing and manipulation support to Google Docs tools, expanding beyond read-only operations to match Google Docs API capabilities. Features added: - Text operations: update_doc_text, find_and_replace_doc, format_doc_text - Structural elements: insert_doc_elements, insert_doc_image - Document management: update_doc_headers_footers, batch_update_doc - Helper functions module for common document operations Changes: - gdocs/docs_tools.py: Add 7 new editing functions with comprehensive error handling - gdocs/docs_helpers.py: New helper module with utility functions for document operations - README.md: Update Google Docs description to include full editing capabilities - CLAUDE.md: Add comprehensive documentation for new editing features - .mcp.json: Add MCP server configuration file - .gitignore: Resolve merge conflicts from upstream sync - uv.lock: Update dependencies from upstream merge The implementation uses Google Docs API batchUpdate operations for atomic document modifications and maintains compatibility with existing authentication and error handling patterns. All new tools follow the established @require_google_service decorator pattern and include proper scope management for write operations.
This commit is contained in:
Rob Sherman
@@ -273,6 +273,675 @@ async def create_doc(
|
||||
return msg
|
||||
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("update_doc_text", service_type="docs")
|
||||
@require_google_service("docs", "docs_write")
|
||||
async def update_doc_text(
|
||||
service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
text: str,
|
||||
start_index: int,
|
||||
end_index: int = None,
|
||||
) -> str:
|
||||
"""
|
||||
Updates text at a specific location in a Google Doc.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
text: New text to insert or replace with
|
||||
start_index: Start position for text update (0-based)
|
||||
end_index: End position for text replacement (if not provided, text is inserted)
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with update details
|
||||
"""
|
||||
logger.info(f"[update_doc_text] Doc={document_id}, start={start_index}, end={end_index}")
|
||||
|
||||
requests = []
|
||||
|
||||
if end_index is not None and end_index > start_index:
|
||||
# Replace text: delete old text, then insert new text
|
||||
requests.extend([
|
||||
{
|
||||
'deleteContentRange': {
|
||||
'range': {
|
||||
'startIndex': start_index,
|
||||
'endIndex': end_index
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
'insertText': {
|
||||
'location': {'index': start_index},
|
||||
'text': text
|
||||
}
|
||||
}
|
||||
])
|
||||
operation = f"Replaced text from index {start_index} to {end_index}"
|
||||
else:
|
||||
# Insert text at position
|
||||
requests.append({
|
||||
'insertText': {
|
||||
'location': {'index': start_index},
|
||||
'text': text
|
||||
}
|
||||
})
|
||||
operation = f"Inserted text at index {start_index}"
|
||||
|
||||
await asyncio.to_thread(
|
||||
service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"{operation} in document {document_id}. Text length: {len(text)} characters. Link: {link}"
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("find_and_replace_doc", service_type="docs")
|
||||
@require_google_service("docs", "docs_write")
|
||||
async def find_and_replace_doc(
|
||||
service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
find_text: str,
|
||||
replace_text: str,
|
||||
match_case: bool = False,
|
||||
) -> str:
|
||||
"""
|
||||
Finds and replaces text throughout a Google Doc.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
find_text: Text to search for
|
||||
replace_text: Text to replace with
|
||||
match_case: Whether to match case exactly
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with replacement count
|
||||
"""
|
||||
logger.info(f"[find_and_replace_doc] Doc={document_id}, find='{find_text}', replace='{replace_text}'")
|
||||
|
||||
requests = [{
|
||||
'replaceAllText': {
|
||||
'containsText': {
|
||||
'text': find_text,
|
||||
'matchCase': match_case
|
||||
},
|
||||
'replaceText': replace_text
|
||||
}
|
||||
}]
|
||||
|
||||
result = await asyncio.to_thread(
|
||||
service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
# Extract number of replacements from response
|
||||
replacements = 0
|
||||
if 'replies' in result and result['replies']:
|
||||
reply = result['replies'][0]
|
||||
if 'replaceAllText' in reply:
|
||||
replacements = reply['replaceAllText'].get('occurrencesChanged', 0)
|
||||
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"Replaced {replacements} occurrence(s) of '{find_text}' with '{replace_text}' in document {document_id}. Link: {link}"
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("format_doc_text", service_type="docs")
|
||||
@require_google_service("docs", "docs_write")
|
||||
async def format_doc_text(
|
||||
service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
start_index: int,
|
||||
end_index: int,
|
||||
bold: bool = None,
|
||||
italic: bool = None,
|
||||
underline: bool = None,
|
||||
font_size: int = None,
|
||||
font_family: str = None,
|
||||
) -> str:
|
||||
"""
|
||||
Applies text formatting to a specific range in a Google Doc.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
start_index: Start position of text to format (0-based)
|
||||
end_index: End position of text to format
|
||||
bold: Whether to make text bold (True/False/None to leave unchanged)
|
||||
italic: Whether to make text italic (True/False/None to leave unchanged)
|
||||
underline: Whether to underline text (True/False/None to leave unchanged)
|
||||
font_size: Font size in points
|
||||
font_family: Font family name (e.g., "Arial", "Times New Roman")
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with formatting details
|
||||
"""
|
||||
logger.info(f"[format_doc_text] Doc={document_id}, range={start_index}-{end_index}")
|
||||
|
||||
text_style = {}
|
||||
format_changes = []
|
||||
|
||||
if bold is not None:
|
||||
text_style['bold'] = bold
|
||||
format_changes.append(f"bold: {bold}")
|
||||
|
||||
if italic is not None:
|
||||
text_style['italic'] = italic
|
||||
format_changes.append(f"italic: {italic}")
|
||||
|
||||
if underline is not None:
|
||||
text_style['underline'] = underline
|
||||
format_changes.append(f"underline: {underline}")
|
||||
|
||||
if font_size is not None:
|
||||
text_style['fontSize'] = {'magnitude': font_size, 'unit': 'PT'}
|
||||
format_changes.append(f"font size: {font_size}pt")
|
||||
|
||||
if font_family is not None:
|
||||
text_style['fontFamily'] = font_family
|
||||
format_changes.append(f"font family: {font_family}")
|
||||
|
||||
if not text_style:
|
||||
return "No formatting changes specified. Please provide at least one formatting option."
|
||||
|
||||
requests = [{
|
||||
'updateTextStyle': {
|
||||
'range': {
|
||||
'startIndex': start_index,
|
||||
'endIndex': end_index
|
||||
},
|
||||
'textStyle': text_style,
|
||||
'fields': ','.join(text_style.keys())
|
||||
}
|
||||
}]
|
||||
|
||||
await asyncio.to_thread(
|
||||
service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
changes_str = ', '.join(format_changes)
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"Applied formatting ({changes_str}) to text from index {start_index} to {end_index} in document {document_id}. Link: {link}"
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("insert_doc_elements", service_type="docs")
|
||||
@require_google_service("docs", "docs_write")
|
||||
async def insert_doc_elements(
|
||||
service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
element_type: str,
|
||||
index: int,
|
||||
rows: int = None,
|
||||
columns: int = None,
|
||||
list_type: str = None,
|
||||
text: str = None,
|
||||
) -> str:
|
||||
"""
|
||||
Inserts structural elements like tables, lists, or page breaks into a Google Doc.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
element_type: Type of element to insert ("table", "list", "page_break")
|
||||
index: Position to insert element (0-based)
|
||||
rows: Number of rows for table (required for table)
|
||||
columns: Number of columns for table (required for table)
|
||||
list_type: Type of list ("UNORDERED", "ORDERED") (required for list)
|
||||
text: Initial text content for list items
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with insertion details
|
||||
"""
|
||||
logger.info(f"[insert_doc_elements] Doc={document_id}, type={element_type}, index={index}")
|
||||
|
||||
requests = []
|
||||
|
||||
if element_type == "table":
|
||||
if not rows or not columns:
|
||||
return "Error: 'rows' and 'columns' parameters are required for table insertion."
|
||||
|
||||
requests.append({
|
||||
'insertTable': {
|
||||
'location': {'index': index},
|
||||
'rows': rows,
|
||||
'columns': columns
|
||||
}
|
||||
})
|
||||
description = f"table ({rows}x{columns})"
|
||||
|
||||
elif element_type == "list":
|
||||
if not list_type:
|
||||
return "Error: 'list_type' parameter is required for list insertion ('UNORDERED' or 'ORDERED')."
|
||||
|
||||
if not text:
|
||||
text = "List item"
|
||||
|
||||
# Insert text first, then create list
|
||||
requests.extend([
|
||||
{
|
||||
'insertText': {
|
||||
'location': {'index': index},
|
||||
'text': text + '\n'
|
||||
}
|
||||
},
|
||||
{
|
||||
'createParagraphBullets': {
|
||||
'range': {
|
||||
'startIndex': index,
|
||||
'endIndex': index + len(text)
|
||||
},
|
||||
'bulletPreset': f'BULLET_DISC_CIRCLE_SQUARE' if list_type == "UNORDERED" else 'NUMBERED_DECIMAL_ALPHA_ROMAN'
|
||||
}
|
||||
}
|
||||
])
|
||||
description = f"{list_type.lower()} list"
|
||||
|
||||
elif element_type == "page_break":
|
||||
requests.append({
|
||||
'insertPageBreak': {
|
||||
'location': {'index': index}
|
||||
}
|
||||
})
|
||||
description = "page break"
|
||||
|
||||
else:
|
||||
return f"Error: Unsupported element type '{element_type}'. Supported types: 'table', 'list', 'page_break'."
|
||||
|
||||
await asyncio.to_thread(
|
||||
service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"Inserted {description} at index {index} in document {document_id}. Link: {link}"
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("insert_doc_image", service_type="docs")
|
||||
@require_multiple_services([
|
||||
{"service_type": "docs", "scopes": "docs_write", "param_name": "docs_service"},
|
||||
{"service_type": "drive", "scopes": "drive_read", "param_name": "drive_service"}
|
||||
])
|
||||
async def insert_doc_image(
|
||||
docs_service,
|
||||
drive_service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
image_source: str,
|
||||
index: int,
|
||||
width: int = None,
|
||||
height: int = None,
|
||||
) -> str:
|
||||
"""
|
||||
Inserts an image into a Google Doc from Drive or a URL.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
image_source: Drive file ID or public image URL
|
||||
index: Position to insert image (0-based)
|
||||
width: Image width in points (optional)
|
||||
height: Image height in points (optional)
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with insertion details
|
||||
"""
|
||||
logger.info(f"[insert_doc_image] Doc={document_id}, source={image_source}, index={index}")
|
||||
|
||||
# Determine if source is a Drive file ID or URL
|
||||
is_drive_file = not (image_source.startswith('http://') or image_source.startswith('https://'))
|
||||
|
||||
if is_drive_file:
|
||||
# Verify Drive file exists and get metadata
|
||||
try:
|
||||
file_metadata = await asyncio.to_thread(
|
||||
drive_service.files().get(
|
||||
fileId=image_source,
|
||||
fields="id, name, mimeType"
|
||||
).execute
|
||||
)
|
||||
mime_type = file_metadata.get('mimeType', '')
|
||||
if not mime_type.startswith('image/'):
|
||||
return f"Error: File {image_source} is not an image (MIME type: {mime_type})."
|
||||
|
||||
image_uri = f"https://drive.google.com/uc?id={image_source}"
|
||||
source_description = f"Drive file {file_metadata.get('name', image_source)}"
|
||||
except Exception as e:
|
||||
return f"Error: Could not access Drive file {image_source}: {str(e)}"
|
||||
else:
|
||||
image_uri = image_source
|
||||
source_description = "URL image"
|
||||
|
||||
# Build image properties
|
||||
image_properties = {}
|
||||
if width is not None:
|
||||
image_properties['width'] = {'magnitude': width, 'unit': 'PT'}
|
||||
if height is not None:
|
||||
image_properties['height'] = {'magnitude': height, 'unit': 'PT'}
|
||||
|
||||
requests = [{
|
||||
'insertInlineImage': {
|
||||
'location': {'index': index},
|
||||
'uri': image_uri,
|
||||
'objectSize': image_properties if image_properties else None
|
||||
}
|
||||
}]
|
||||
|
||||
# Remove None values
|
||||
if requests[0]['insertInlineImage']['objectSize'] is None:
|
||||
del requests[0]['insertInlineImage']['objectSize']
|
||||
|
||||
await asyncio.to_thread(
|
||||
docs_service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
size_info = ""
|
||||
if width or height:
|
||||
size_info = f" (size: {width or 'auto'}x{height or 'auto'} points)"
|
||||
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"Inserted {source_description}{size_info} at index {index} in document {document_id}. Link: {link}"
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("update_doc_headers_footers", service_type="docs")
|
||||
@require_google_service("docs", "docs_write")
|
||||
async def update_doc_headers_footers(
|
||||
service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
section_type: str,
|
||||
content: str,
|
||||
header_footer_type: str = "DEFAULT",
|
||||
) -> str:
|
||||
"""
|
||||
Updates headers or footers in a Google Doc.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
section_type: Type of section to update ("header" or "footer")
|
||||
content: Text content for the header/footer
|
||||
header_footer_type: Type of header/footer ("DEFAULT", "FIRST_PAGE_ONLY", "EVEN_PAGE")
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with update details
|
||||
"""
|
||||
logger.info(f"[update_doc_headers_footers] Doc={document_id}, type={section_type}")
|
||||
|
||||
if section_type not in ["header", "footer"]:
|
||||
return "Error: section_type must be 'header' or 'footer'."
|
||||
|
||||
if header_footer_type not in ["DEFAULT", "FIRST_PAGE_ONLY", "EVEN_PAGE"]:
|
||||
return "Error: header_footer_type must be 'DEFAULT', 'FIRST_PAGE_ONLY', or 'EVEN_PAGE'."
|
||||
|
||||
# First, get the document to find existing header/footer
|
||||
doc = await asyncio.to_thread(
|
||||
service.documents().get(documentId=document_id).execute
|
||||
)
|
||||
|
||||
# Find the appropriate header or footer
|
||||
headers = doc.get('headers', {})
|
||||
footers = doc.get('footers', {})
|
||||
|
||||
target_section = None
|
||||
section_id = None
|
||||
|
||||
if section_type == "header":
|
||||
# Look for existing header of the specified type
|
||||
for hid, header in headers.items():
|
||||
target_section = header
|
||||
section_id = hid
|
||||
break # Use first available header for now
|
||||
else:
|
||||
# Look for existing footer of the specified type
|
||||
for fid, footer in footers.items():
|
||||
target_section = footer
|
||||
section_id = fid
|
||||
break # Use first available footer for now
|
||||
|
||||
if not target_section:
|
||||
return f"Error: No {section_type} found in document. Please create a {section_type} first in Google Docs."
|
||||
|
||||
# Clear existing content and insert new content
|
||||
content_elements = target_section.get('content', [])
|
||||
if content_elements:
|
||||
# Find the first paragraph to replace content
|
||||
first_para = None
|
||||
for element in content_elements:
|
||||
if 'paragraph' in element:
|
||||
first_para = element
|
||||
break
|
||||
|
||||
if first_para:
|
||||
# Calculate content range to replace
|
||||
start_index = first_para.get('startIndex', 0)
|
||||
end_index = first_para.get('endIndex', 0)
|
||||
|
||||
requests = []
|
||||
|
||||
# Delete existing content if any
|
||||
if end_index > start_index:
|
||||
requests.append({
|
||||
'deleteContentRange': {
|
||||
'range': {
|
||||
'startIndex': start_index,
|
||||
'endIndex': end_index - 1 # Keep the paragraph end
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
# Insert new content
|
||||
requests.append({
|
||||
'insertText': {
|
||||
'location': {'index': start_index},
|
||||
'text': content
|
||||
}
|
||||
})
|
||||
|
||||
await asyncio.to_thread(
|
||||
service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"Updated {section_type} content in document {document_id}. Link: {link}"
|
||||
|
||||
return f"Error: Could not find content structure in {section_type} to update."
|
||||
|
||||
@server.tool()
|
||||
@handle_http_errors("batch_update_doc", service_type="docs")
|
||||
@require_google_service("docs", "docs_write")
|
||||
async def batch_update_doc(
|
||||
service,
|
||||
user_google_email: str,
|
||||
document_id: str,
|
||||
operations: list,
|
||||
) -> str:
|
||||
"""
|
||||
Executes multiple document operations in a single atomic batch update.
|
||||
|
||||
Args:
|
||||
user_google_email: User's Google email address
|
||||
document_id: ID of the document to update
|
||||
operations: List of operation dictionaries. Each operation should contain:
|
||||
- type: Operation type ('insert_text', 'delete_text', 'replace_text', 'format_text', 'insert_table', 'insert_page_break')
|
||||
- Additional parameters specific to each operation type
|
||||
|
||||
Example operations:
|
||||
[
|
||||
{"type": "insert_text", "index": 1, "text": "Hello World"},
|
||||
{"type": "format_text", "start_index": 1, "end_index": 12, "bold": true},
|
||||
{"type": "insert_table", "index": 20, "rows": 2, "columns": 3}
|
||||
]
|
||||
|
||||
Returns:
|
||||
str: Confirmation message with batch operation results
|
||||
"""
|
||||
logger.info(f"[batch_update_doc] Doc={document_id}, operations={len(operations)}")
|
||||
|
||||
if not operations:
|
||||
return "Error: No operations provided. Please provide at least one operation."
|
||||
|
||||
requests = []
|
||||
operation_descriptions = []
|
||||
|
||||
for i, op in enumerate(operations):
|
||||
op_type = op.get('type')
|
||||
if not op_type:
|
||||
return f"Error: Operation {i+1} missing 'type' field."
|
||||
|
||||
try:
|
||||
if op_type == 'insert_text':
|
||||
requests.append({
|
||||
'insertText': {
|
||||
'location': {'index': op['index']},
|
||||
'text': op['text']
|
||||
}
|
||||
})
|
||||
operation_descriptions.append(f"insert text at {op['index']}")
|
||||
|
||||
elif op_type == 'delete_text':
|
||||
requests.append({
|
||||
'deleteContentRange': {
|
||||
'range': {
|
||||
'startIndex': op['start_index'],
|
||||
'endIndex': op['end_index']
|
||||
}
|
||||
}
|
||||
})
|
||||
operation_descriptions.append(f"delete text {op['start_index']}-{op['end_index']}")
|
||||
|
||||
elif op_type == 'replace_text':
|
||||
requests.extend([
|
||||
{
|
||||
'deleteContentRange': {
|
||||
'range': {
|
||||
'startIndex': op['start_index'],
|
||||
'endIndex': op['end_index']
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
'insertText': {
|
||||
'location': {'index': op['start_index']},
|
||||
'text': op['text']
|
||||
}
|
||||
}
|
||||
])
|
||||
operation_descriptions.append(f"replace text {op['start_index']}-{op['end_index']}")
|
||||
|
||||
elif op_type == 'format_text':
|
||||
text_style = {}
|
||||
format_changes = []
|
||||
|
||||
if op.get('bold') is not None:
|
||||
text_style['bold'] = op['bold']
|
||||
format_changes.append(f"bold: {op['bold']}")
|
||||
if op.get('italic') is not None:
|
||||
text_style['italic'] = op['italic']
|
||||
format_changes.append(f"italic: {op['italic']}")
|
||||
if op.get('underline') is not None:
|
||||
text_style['underline'] = op['underline']
|
||||
format_changes.append(f"underline: {op['underline']}")
|
||||
if op.get('font_size') is not None:
|
||||
text_style['fontSize'] = {'magnitude': op['font_size'], 'unit': 'PT'}
|
||||
format_changes.append(f"font size: {op['font_size']}pt")
|
||||
if op.get('font_family') is not None:
|
||||
text_style['fontFamily'] = op['font_family']
|
||||
format_changes.append(f"font family: {op['font_family']}")
|
||||
|
||||
if text_style:
|
||||
requests.append({
|
||||
'updateTextStyle': {
|
||||
'range': {
|
||||
'startIndex': op['start_index'],
|
||||
'endIndex': op['end_index']
|
||||
},
|
||||
'textStyle': text_style,
|
||||
'fields': ','.join(text_style.keys())
|
||||
}
|
||||
})
|
||||
operation_descriptions.append(f"format text {op['start_index']}-{op['end_index']} ({', '.join(format_changes)})")
|
||||
|
||||
elif op_type == 'insert_table':
|
||||
if not op.get('rows') or not op.get('columns'):
|
||||
return f"Error: Operation {i+1} (insert_table) requires 'rows' and 'columns' fields."
|
||||
|
||||
requests.append({
|
||||
'insertTable': {
|
||||
'location': {'index': op['index']},
|
||||
'rows': op['rows'],
|
||||
'columns': op['columns']
|
||||
}
|
||||
})
|
||||
operation_descriptions.append(f"insert {op['rows']}x{op['columns']} table at {op['index']}")
|
||||
|
||||
elif op_type == 'insert_page_break':
|
||||
requests.append({
|
||||
'insertPageBreak': {
|
||||
'location': {'index': op['index']}
|
||||
}
|
||||
})
|
||||
operation_descriptions.append(f"insert page break at {op['index']}")
|
||||
|
||||
elif op_type == 'find_replace':
|
||||
requests.append({
|
||||
'replaceAllText': {
|
||||
'containsText': {
|
||||
'text': op['find_text'],
|
||||
'matchCase': op.get('match_case', False)
|
||||
},
|
||||
'replaceText': op['replace_text']
|
||||
}
|
||||
})
|
||||
operation_descriptions.append(f"find/replace '{op['find_text']}' → '{op['replace_text']}'")
|
||||
|
||||
else:
|
||||
return f"Error: Unsupported operation type '{op_type}' in operation {i+1}. Supported types: insert_text, delete_text, replace_text, format_text, insert_table, insert_page_break, find_replace."
|
||||
|
||||
except KeyError as e:
|
||||
return f"Error: Operation {i+1} ({op_type}) missing required field: {e}"
|
||||
except Exception as e:
|
||||
return f"Error: Operation {i+1} ({op_type}) failed validation: {str(e)}"
|
||||
|
||||
# Execute all operations in a single batch
|
||||
result = await asyncio.to_thread(
|
||||
service.documents().batchUpdate(
|
||||
documentId=document_id,
|
||||
body={'requests': requests}
|
||||
).execute
|
||||
)
|
||||
|
||||
# Extract results information
|
||||
replies_count = len(result.get('replies', []))
|
||||
|
||||
operations_summary = ', '.join(operation_descriptions[:3]) # Show first 3 operations
|
||||
if len(operation_descriptions) > 3:
|
||||
operations_summary += f" and {len(operation_descriptions) - 3} more"
|
||||
|
||||
link = f"https://docs.google.com/document/d/{document_id}/edit"
|
||||
return f"Successfully executed {len(operations)} operations ({operations_summary}) on document {document_id}. API replies: {replies_count}. Link: {link}"
|
||||
|
||||
|
||||
# Create comment management tools for documents
|
||||
_comment_tools = create_comment_tools("document", "document_id")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user