github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-04-08 03:00:28 -04:00
Code Issues Packages Projects Releases Wiki Activity

run format

Browse Source
This commit is contained in:
Toran Bruce Richards
2024-07-21 11:58:29 +01:00
parent 4256ffcc78
commit 3e1f59814d
1 changed files with 53 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

93
rnd/autogpt_server/autogpt_server/blocks/google-sheets-block.py
View File

@@ -1,36 +1,42 @@
from autogpt_server.data.block import Block, BlockSchema, BlockOutput
from autogpt_server.data.model import SchemaField
import re
from typing import List, Union
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from typing import Union, List
import re
from autogpt_server.data.block import Block, BlockOutput, BlockSchema
from autogpt_server.data.model import SchemaField
def extract_spreadsheet_id(url: str) -> str:
"""Extract the spreadsheet ID from a Google Sheets URL."""
match = re.search(r'/d/([a-zA-Z0-9-_]+)', url) # This works for clean urls and those with extra parameters at the end
match = re.search(
r"/d/([a-zA-Z0-9-_]+)", url
) # This works for clean urls and those with extra parameters at the end
if match:
return match.group(1)
raise ValueError("Invalid Google Sheets URL")
class GoogleSheetsWriter(Block):
class Input(BlockSchema):
spreadsheet_url: str = SchemaField(
description="The link to the Google Sheet to write to.",
placeholder="https://docs.google.com/spreadsheets/d/1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms"
placeholder="https://docs.google.com/spreadsheets/d/1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms",
)
sheet_name: str = SchemaField(
description="The name of the sheet to append data to",
placeholder="Sheet1",
default="Sheet1"
default="Sheet1",
)
row_data: list = SchemaField(
description="The data to append as a single row",
placeholder="['John Doe', 'johndoe@example.com', '30']"
placeholder="['John Doe', 'johndoe@example.com', '30']",
)
access_token: str = SchemaField(
description="Google OAuth2 access token. Click 'Sign in with Google' below to automatically generate this token. Keep this secret!",
placeholder="Sign in with Google below",
secret=True
secret=True,
)
class Output(BlockSchema):
@@ -51,69 +57,70 @@ class GoogleSheetsWriter(Block):
sheet = service.spreadsheets()
# Get the current sheet data to find the next empty row
result = sheet.values().get(
spreadsheetId=spreadsheet_id,
range=f"{input_data.sheet_name}!A:A"
).execute()
values = result.get('values', [])
result = (
sheet.values()
.get(spreadsheetId=spreadsheet_id, range=f"{input_data.sheet_name}!A:A")
.execute()
)
values = result.get("values", [])
next_row = len(values) + 1
# Prepare the range for appending
append_range = f"{input_data.sheet_name}!A{next_row}"
body = {
'values': [input_data.row_data]
}
body = {"values": [input_data.row_data]}
result = sheet.values().append(
spreadsheetId=spreadsheet_id,
range=append_range,
valueInputOption='USER_ENTERED',
insertDataOption='INSERT_ROWS',
body=body
).execute()
result = (
sheet.values()
.append(
spreadsheetId=spreadsheet_id,
range=append_range,
valueInputOption="USER_ENTERED",
insertDataOption="INSERT_ROWS",
body=body,
)
.execute()
)
yield "result", {
"success": True,
"updated_range": result.get('updates', {}).get('updatedRange'),
"updated_rows": result.get('updates', {}).get('updatedRows'),
"updated_cells": result.get('updates', {}).get('updatedCells')
"updated_range": result.get("updates", {}).get("updatedRange"),
"updated_rows": result.get("updates", {}).get("updatedRows"),
"updated_cells": result.get("updates", {}).get("updatedCells"),
}
except Exception as e:
yield "result", {
"success": False,
"error": str(e)
}
yield "result", {"success": False, "error": str(e)}
class GoogleSheetsReader(Block):
class Input(BlockSchema):
spreadsheet_url: str = SchemaField(
description="The link to the Google Sheet to read from.",
placeholder="https://docs.google.com/spreadsheets/d/1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms"
placeholder="https://docs.google.com/spreadsheets/d/1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms",
)
read_single_row: bool = SchemaField(
description="Do you want to read just one row? Select 'True' for a single row, 'False' for a range.",
default=False
default=False,
)
sheet_name: str = SchemaField(
description="The name of the sheet you want to read from (e.g., 'Sheet1')",
placeholder="Sheet1",
default="Sheet1"
default="Sheet1",
)
row_number: int = SchemaField(
description="If reading a single row, which row number do you want? (e.g., 3 for the third row)",
placeholder="3"
placeholder="3",
)
cell_range: str = SchemaField(
description="If reading a range, what range of cells do you want? (e.g., 'A1:E10' for the first 10 rows of columns A to E)",
placeholder="A1:E10",
default="A1:Z1000"
default="A1:Z1000",
)
access_token: str = SchemaField(
placeholder="Sign in with Google below",
description="Google OAuth2 access token. Click 'Sign in with Google' below to automatically generate this token. Keep this secret!",
secret=True
secret=True,
)
class Output(BlockSchema):
@@ -140,11 +147,17 @@ class GoogleSheetsReader(Block):
else:
range_to_read = f"{input_data.sheet_name}!{input_data.cell_range}"
result = sheet.values().get(spreadsheetId=spreadsheet_id, range=range_to_read).execute()
result = (
sheet.values()
.get(spreadsheetId=spreadsheet_id, range=range_to_read)
.execute()
)
values = result.get("values", [])
if not values:
yield "data", {"message": "No data found in the specified range or row."}
yield "data", {
"message": "No data found in the specified range or row."
}
else:
if input_data.read_single_row:
# Return a single list for a single row
@@ -154,4 +167,4 @@ class GoogleSheetsReader(Block):
yield "data", values
except Exception as e:
yield "data", {"error": f"Oops! Something went wrong: {str(e)}"}
yield "data", {"error": f"Oops! Something went wrong: {str(e)}"}