Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions tools/productivity/gsuite/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -1278,6 +1278,63 @@ def docs_read(
raise typer.Exit(1)


@docs_app.command("comments")
def docs_comments(
doc_id: str = typer.Argument(..., help="Document ID or Google Docs URL"),
limit: int = typer.Option(100, "--limit", "-n", help="Max comments"),
include_deleted: bool = typer.Option(
False,
"--include-deleted",
help="Include deleted comments and replies",
),
json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
):
"""Read comments and replies on a Google Doc.

Examples:
gsuite docs comments "1abc123"
gsuite docs comments "https://docs.google.com/document/d/1abc123/edit" --json
"""
from .client import docs_list_comments

try:
document_id = extract_doc_id(doc_id)
comments = docs_list_comments(
document_id,
max_results=limit,
include_deleted=include_deleted,
)
if json_output:
print(json.dumps(comments, indent=2, ensure_ascii=False))
return
if not comments:
console.print("[yellow]No comments found.[/]")
return

for comment in comments:
author = comment["author"]["display_name"] or "Unknown author"
status = (
"deleted" if comment["deleted"] else "resolved" if comment["resolved"] else "open"
)
console.print(f"Comment {comment['id']} by {author} [{status}]", markup=False)
quoted_text = comment["quoted_file_content"]["value"]
if quoted_text:
console.print(f" Quoted: {quoted_text}", markup=False)
if comment["content"]:
console.print(f" {comment['content']}", markup=False)
for reply in comment["replies"]:
reply_author = reply["author"]["display_name"] or "Unknown author"
reply_action = f" [{reply['action']}]" if reply["action"] else ""
console.print(
f" Reply {reply['id']} by {reply_author}{reply_action}: {reply['content']}",
markup=False,
)
console.print()
except Exception as e:
console.print(f"[red]Error: {e}[/]")
raise typer.Exit(1) from e


@docs_app.command("replace")
def docs_replace_cmd(
doc_id: str = typer.Argument(..., help="Document ID or Google Docs URL"),
Expand Down
117 changes: 117 additions & 0 deletions tools/productivity/gsuite/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1759,6 +1759,101 @@ def extract_text_from_content(content: list) -> str:
return extract_text_from_content(content)


DRIVE_COMMENT_FIELDS = (
"id,content,htmlContent,anchor,quotedFileContent,resolved,deleted,"
"createdTime,modifiedTime,assigneeEmailAddress,mentionedEmailAddresses,"
"author(displayName,photoLink,me),"
"replies(id,content,htmlContent,action,deleted,createdTime,modifiedTime,"
"assigneeEmailAddress,mentionedEmailAddresses,author(displayName,photoLink,me))"
)


def _normalize_drive_comment_user(user: dict) -> dict:
return {
"display_name": user.get("displayName", ""),
"photo_link": user.get("photoLink", ""),
"is_me": user.get("me", False),
}


def _normalize_drive_reply(reply: dict) -> dict:
return {
"id": reply.get("id", ""),
"content": reply.get("content", ""),
"html_content": reply.get("htmlContent", ""),
"action": reply.get("action", ""),
"deleted": reply.get("deleted", False),
"created_time": reply.get("createdTime", ""),
"modified_time": reply.get("modifiedTime", ""),
"author": _normalize_drive_comment_user(reply.get("author") or {}),
"assignee_email": reply.get("assigneeEmailAddress", ""),
"mentioned_emails": reply.get("mentionedEmailAddresses") or [],
}


def _normalize_drive_comment(comment: dict) -> dict:
quoted_content = comment.get("quotedFileContent") or {}
return {
"id": comment.get("id", ""),
"content": comment.get("content", ""),
"html_content": comment.get("htmlContent", ""),
"anchor": comment.get("anchor", ""),
"quoted_file_content": {
"mime_type": quoted_content.get("mimeType", ""),
"value": quoted_content.get("value", ""),
},
"resolved": comment.get("resolved", False),
"deleted": comment.get("deleted", False),
"created_time": comment.get("createdTime", ""),
"modified_time": comment.get("modifiedTime", ""),
"author": _normalize_drive_comment_user(comment.get("author") or {}),
"assignee_email": comment.get("assigneeEmailAddress", ""),
"mentioned_emails": comment.get("mentionedEmailAddresses") or [],
"replies": [_normalize_drive_reply(reply) for reply in comment.get("replies", [])],
}


def docs_list_comments(
document_id: str,
max_results: int = 100,
include_deleted: bool = False,
) -> list[dict]:
"""List comments and replies on a Google Doc.

Args:
document_id: The document ID
max_results: Maximum number of comments to return
include_deleted: Whether to include deleted comments and replies

Returns:
Comments with their quoted document content and replies
"""
if max_results < 1:
raise ValueError("max_results must be at least 1")

service = get_drive_service()
comments: list[dict] = []
page_token: str | None = None

while len(comments) < max_results:
request_args = {
"fileId": document_id,
"pageSize": min(100, max_results - len(comments)),
"includeDeleted": include_deleted,
"fields": f"nextPageToken,comments({DRIVE_COMMENT_FIELDS})",
}
if page_token:
request_args["pageToken"] = page_token

result = service.comments().list(**request_args).execute()
comments.extend(_normalize_drive_comment(comment) for comment in result.get("comments", []))
page_token = result.get("nextPageToken")
if not page_token:
break

return comments[:max_results]


def docs_append(
document_id: str,
text: str,
Expand Down Expand Up @@ -3226,6 +3321,28 @@ def docs_get_text(self, document_id: str) -> str:
"""
return docs_get_text(document_id)

def docs_list_comments(
self,
document_id: str,
max_results: int = 100,
include_deleted: bool = False,
) -> list[dict]:
"""List comments and replies on a Google Doc.

Args:
document_id: The document ID
max_results: Maximum number of comments to return
include_deleted: Whether to include deleted comments and replies

Returns:
Comments with their quoted document content and replies
"""
return docs_list_comments(
document_id,
max_results=max_results,
include_deleted=include_deleted,
)

def docs_append(
self,
document_id: str,
Expand Down
85 changes: 85 additions & 0 deletions tools/productivity/gsuite/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,91 @@ def test_docs_bullets_command_prints_verification_summary(monkeypatch):
assert "tab tab-2 paragraph 4:" in result.output


def test_docs_comments_command_accepts_url_and_outputs_json(monkeypatch):
calls: list[dict] = []
comments = [
{
"id": "comment-1",
"content": "Please clarify this section.",
"author": {"display_name": "Ada Lovelace"},
"quoted_file_content": {"value": "Draft language"},
"resolved": False,
"deleted": False,
"replies": [],
}
]
monkeypatch.setattr(
client,
"docs_list_comments",
lambda document_id, max_results, include_deleted: (
calls.append(
{
"document_id": document_id,
"max_results": max_results,
"include_deleted": include_deleted,
}
)
or comments
),
)

result = runner.invoke(
app,
[
"docs",
"comments",
"https://docs.google.com/document/d/doc-123/edit",
"--limit",
"25",
"--include-deleted",
"--json",
],
)

assert result.exit_code == 0
assert json.loads(result.output) == comments
assert calls == [
{
"document_id": "doc-123",
"max_results": 25,
"include_deleted": True,
}
]


def test_docs_comments_command_prints_threads_without_rich_markup(monkeypatch):
monkeypatch.setattr(
client,
"docs_list_comments",
lambda document_id, max_results, include_deleted: [
{
"id": "comment-1",
"content": "Use [draft] here.",
"author": {"display_name": "Ada Lovelace"},
"quoted_file_content": {"value": "Original [text]"},
"resolved": True,
"deleted": False,
"replies": [
{
"id": "reply-1",
"content": "Done [now].",
"action": "resolve",
"author": {"display_name": "Grace Hopper"},
}
],
}
],
)

result = runner.invoke(app, ["docs", "comments", "doc-123"])

assert result.exit_code == 0
assert "Comment comment-1 by Ada Lovelace [resolved]" in result.output
assert "Quoted: Original [text]" in result.output
assert "Use [draft] here." in result.output
assert "Reply reply-1 by Grace Hopper [resolve]: Done [now]." in result.output


def test_drive_revisions_command_accepts_sheets_url_and_outputs_json(monkeypatch):
calls: list[dict] = []
monkeypatch.setattr(
Expand Down
Loading