from .google_api import try_get_google_services_oauth, DEFAULT_TOKEN_STEM

import pandas as pd

import os

import json

import csv

import uuid

from pathlib import Path

import inspect

import re

import io

from datetime import datetime, timedelta

from dateutil.parser import isoparse

from dateutil.tz import UTC

import mimetypes

from bs4 import BeautifulSoup

from googleapiclient.http import MediaFileUpload

from googleapiclient.http import MediaIoBaseDownload

from googleapiclient.discovery import build

import base64

from email.mime.text import MIMEText

from typing import Union

from googleapiclient.discovery import Resource

from typing import Optional, Any, Tuple

class GoogleApi:

"""

Encapsulates Google OAuth + service clients.

After init, check `self.google_auth` and `self.error`.

Services (when available): self.drive_service, self.docs_service, self.sheets_service,

self.calendar_service, self.tasks_service, self.forms_service, self.gmail_service

"""

def __init__(

self,

*,

oauth_client_file: Optional[str] = None,

oauth_token_stem: str = DEFAULT_TOKEN_STEM,

interactive: Optional[bool] = None,

auto_init: bool = True,

) -> None:

# config

self._oauth_client_file = oauth_client_file

self._oauth_token_stem = oauth_token_stem

self._interactive = interactive

# state

self.google_auth: bool = False

self.error: Optional[Exception] = None

# services

self.drive_service: Any = None

self.docs_service: Any = None

self.sheets_service: Any = None

self.calendar_service: Any = None

self.tasks_service: Any = None

self.forms_service: Any = None

self.gmail_service: Any = None

if auto_init:

self.init_auth()

def init_auth(self) -> bool:

"""

Attempts to obtain OAuth creds + build services.

Sets self.google_auth and self.error.

Returns True on success, False otherwise.

"""

res = try_get_google_services_oauth(

oauth_client_file=self._oauth_client_file,

oauth_token_stem=self._oauth_token_stem,

interactive=self._interactive,

)

self.google_auth = bool(res and res.ok)

self.error = getattr(res, "error", None)

if self.google_auth:

(self.drive_service, self.docs_service, self.sheets_service,

self.calendar_service, self.tasks_service, self.forms_service, self.gmail_service) = res.services

else:

# ensure all are None on failure

self.drive_service = self.docs_service = self.sheets_service = None

self.calendar_service = self.tasks_service = self.forms_service = None

self.gmail_service = None

return self.google_auth

def ensure_auth(self) -> bool:

"""

Lazy ensure: if not authenticated, try again once.

Useful if token file appeared after construction.

"""

return self.google_auth or self.init_auth()

def services_tuple(self) -> Optional[Tuple[Any, ...]]:

"""Return services as a tuple (matching your original order) or None."""

if not self.google_auth:

return None

return (self.drive_service, self.docs_service, self.sheets_service,

self.calendar_service, self.tasks_service, self.forms_service, self.gmail_service)

def send_email(self, sender: str, to: Union[str, list[str]], subject: str, body: str):

"""

Sends a plain text email using the Gmail API.

This method constructs a MIME message, encodes it in base64, and sends it using the Gmail API.

The `gmail_service` must be an authenticated Gmail API client.

Parameters:

sender (str): The sender's email address.

to (str | list[str]): One or more recipient email addresses.

subject (str): The subject of the email.

body (str): The plain text body of the email.

Returns:

dict: A dictionary containing:

- 'status' (str): 'success' if the email was sent, otherwise 'error'.

- 'response' (dict):

- 'meta_data' (dict): Includes the recipient(s) and message ID.

- 'data' (str): JSON-encoded string containing the message metadata.

- 'message' (str): A human-readable message about the result.

Example:

>>> send_email(gmail_service, 'me@gmail.com', ['you@example.com', 'them@example.com'], 'Test', 'Hello')

{

'status': 'success',

'response': {

'meta_data': {'to': ['you@example.com', 'them@example.com'], 'id': '17abcd123xyz'},

'data': '{"records": [{"to": ["you@example.com", "them@example.com"], "id": "17abcd123xyz"}]}',

'message': 'Email sent to 2 recipient(s) with ID: 17abcd123xyz'

}

}

Notes:

- The `to` field can be a single string or a list of email addresses.

- The email body is plain text. Use MIME multipart for HTML or attachments if needed.

- Gmail API requires 'https://www.googleapis.com/auth/gmail.send' scope.

"""

status = ''

message = ''

meta_data = {}

try:

if isinstance(to, list):

to_str = ', '.join(to)

recipients = to

else:

to_str = to

recipients = [to]

mime_msg = MIMEText(body)

mime_msg['to'] = to_str

mime_msg['from'] = sender

mime_msg['subject'] = subject

raw_msg = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode()

send_result = self.gmail_service.users().messages().send(userId='me', body={'raw': raw_msg}).execute()

msg_id = send_result.get('id')

status = 'success'

message = f'Email sent to {len(recipients)} recipient(s) with ID: {msg_id}'

meta_data = {'to': recipients, 'id': msg_id}

except Exception as e:

status = 'error'

message = f'Error: {str(e)}'

meta_data = {'to': to, 'id': None}

response = {

'meta_data': meta_data,

'data': json.dumps({"records": [meta_data]}),

'message': message

}

return {

'status': status,

'response': response,

'message': message

}

def get_gdrive_folder_explorer(

self,

folder_id: str = "root",

query: str | None = None,

user_id: str | None = None, # kept for your signature

*,

mime_types: str | list[str] | tuple[str, ...] | None = None,

only_folders: bool = False,

shared_drive_id: str | None = None, # pass a drive ID to target a specific Shared Drive

page_size: int =10

):

"""

Fetch contents of a Google Drive folder with optional name and MIME-type filtering using the Google Drive API (v3) .

(Non-recursive.)

Args:

folder_id: Drive folder ID to explore; 'root' for My Drive root.

query: Plain search term for name filtering (uses "name contains '<term>'").

mime_types: One MIME type (str) or a list/tuple of MIME types to include.

only_folders: If True, include only folders.

shared_drive_id: If set, search that Shared Drive (Team Drive). When provided,

the call sets corpora="drive" and driveId=<shared_drive_id>.

page_size: Number of records to return. Default is 10.

Returns:

dict with 'status', 'response' (meta_data, data={'records':[...]}, message), and 'message'.

"""

status = "error"

meta_data = {}

records: list[dict] = []

message = ""

# ---- helpers -------------------------------------------------------------

def _escape_term(s: str) -> str:

# Drive 'q' strings are single-quoted with backslash escapes for quotes/backslashes.

return s.replace("\\", "\\\\").replace("'", "\\'")

def _normalize_mimes(m) -> list[str]:

if m is None:

return []

if isinstance(m, (list, tuple)):

return list(m)

if isinstance(m, str):

return [m]

raise TypeError("mime_types must be None, str, list[str], or tuple[str,...]")

try:

parts: list[str] = []

# Scope by parent (unless you intentionally want to search all of My Drive)

if folder_id and folder_id != "root":

parts.append(f"'{folder_id}' in parents")

# Name filter

if query:

parts.append(f"name contains '{_escape_term(query)}'")

# Trashed

parts.append("trashed = false")

# MIME type filters

if only_folders:

parts.append("mimeType = 'application/vnd.google-apps.folder'")

elif mime_types:

mts = _normalize_mimes(mime_types)

if mts:

or_block = " or ".join([f"mimeType = '{_escape_term(mt)}'" for mt in mts])

parts.append(f"({or_block})")

# else: include both files and folders (no filter)

q = " and ".join(parts)

# ---- list() params (Shared Drives friendly defaults) -----------------

list_kwargs = dict(

q=q,

pageSize=page_size,

fields="nextPageToken, files(id,name,mimeType,parents,modifiedTime,webViewLink,iconLink)",

includeItemsFromAllDrives=True,

supportsAllDrives=True,

orderBy="recency desc",

)

if shared_drive_id:

# Search a specific shared drive rather than 'user' (My Drive).

list_kwargs.update(corpora="drive", driveId=shared_drive_id)

# ---- pagination ------------------------------------------------------

page_token = None

while True:

if page_token:

list_kwargs["pageToken"] = page_token

resp = self.drive_service.files().list(**list_kwargs).execute()

for f in resp.get("files", []):

records.append(

{

"id": f.get("id"),

"name": f.get("name"),

"mimeType": f.get("mimeType"),

"modifiedTime": f.get("modifiedTime"),

"webViewLink": f.get("webViewLink"),

"iconLink": f.get("iconLink"),

}

)

page_token = resp.get("nextPageToken")

if not page_token:

break

# ---- response assembly ----------------------------------------------

meta_data = {

"folder_id": folder_id,

"search": query,

"mime_types": _normalize_mimes(mime_types) if not (only_folders) else None,

"only_folders": only_folders,

"shared_drive_id": shared_drive_id,

"q": q, # handy for debugging

}

message = f"Found {len(records)} item(s)"

if query:

message += f" matching \"{query}\""

if only_folders:

message += " (folders only)"

elif mime_types:

message += f" (mime_types filter applied)"

# Pretty list

if records:

message += ":\n" + "\n".join(

f"- {r['name']} {'📁' if r['mimeType']=='application/vnd.google-apps.folder' else '📄'} (id: {r['id']}, mime_type: {r['mimeType']})"

for r in records

)

status = "success"

except Exception as e:

message = f"Error: {e!s}"

status = "error"

return {

"status": status,

"response": {

"meta_data": meta_data,

"data": json.dumps({"records": records}), # structured dict (not JSON string)

"message": message,

},

"message": message,

}

def create_gdrive_folder(self, name=None, parent_folder_id=None, user_id=None):

"""

Creates a new folder in Google Drive, optionally inside a specified parent folder.

This method uses the Google Drive API to create a new folder with the specified `name`.

If a `parent_folder_id` is provided, the new folder will be nested inside it.

Parameters:

name (str): The name of the folder to be created.

parent_folder_id (str, optional): The ID of the parent folder where the new folder will be created.

If not provided, the folder will be created at the root level.

Returns:

dict: A dictionary containing:

- 'status' (str): 'success' if the folder was created, otherwise 'error'.

- 'response' (dict):

- 'meta_data' (dict): Includes the name and ID of the created folder (if successful).

- 'data' (str): JSON-encoded string containing the folder metadata.

- 'message' (str): A human-readable message about the result.

Example:

>>> create_gdrive_folder(name='ProjectDocs', parent_folder_id='1a2b3c4d')

{

'status': 'success',

'response': {

'meta_data': {'folder': 'ProjectDocs', 'id': 'abc123xyz'},

'data': '{"records": {"folder": "ProjectDocs", "id": "abc123xyz"}}',

'message': 'Folder "ProjectDocs" created with ID: abc123xyz'

}

}

Notes:

- The `mimeType` used is `'application/vnd.google-apps.folder'` which is required for Drive folders.

- The Google Drive API service (`drive_service`) must be authenticated and accessible.

- Any exceptions during folder creation are caught and returned in the response message.

Raises:

Exception: Any errors from the Drive API are caught and returned as error messages in the response.

"""

status=''

message=''

meta_data = {}

try:

folder_metadata = {

'name': name,

'mimeType': 'application/vnd.google-apps.folder'

}

if parent_folder_id:

folder_metadata['parents'] = [parent_folder_id]

folder = self.drive_service.files().create(body=folder_metadata, fields='id').execute()

folder_id = folder.get('id')

message = f'Folder "{name}" created with ID: {folder_id}'

status = 'success'

except Exception as e:

message = f'Error: {str(e)}'

meta_data = {'folder':name, 'id':folder_id}

response = {"meta_data": meta_data, "data":json.dumps({"records":[meta_data]}), "message":message}

response = {

'status': status,

'response':response,

"message":message

}

return response

def upload_file_to_drive(self, file_path, file_name=None, parent_folder_id=None, user_id=None):

"""

Uploads a file to Google Drive, optionally inside a specified parent folder.

This method uses the Google Drive API to upload a file. If a `parent_folder_id` is provided,

the file will be uploaded into that folder. If `file_name` is not specified, the original

filename from `file_path` will be used.

Parameters:

file_path (str): Path to the local file to be uploaded.

file_name (str, optional): Desired name of the file on Google Drive.

parent_folder_id (str, optional): ID of the folder to upload the file into.

user_id (str, optional): Optional user identifier for tracking/logging.

Returns:

dict: A dictionary with:

- 'status' (str): 'success' if the upload was successful, otherwise 'error'.

- 'response' (dict):

- 'meta_data' (dict): Includes `parent_folder_id`, `file_id`, and `file_name`.

- 'data' (str): JSON-encoded version of the metadata under the `records` key.

- 'message' (str): Human-readable message.

Example:

>>> upload_file_to_drive('/path/to/file.pdf', parent_folder_id='1AbCdEfGhIj')

{

'status': 'success',

'response': {

'meta_data': {

'parent_folder_id': '1AbCdEfGhIj',

'file_id': '1XyZ9aBcDeFgHiJkLmNo',

'file_name': 'file.pdf'

},

'data': '{"records":[{"parent_folder_id": "1AbCdEfGhIj", "file_id": "1XyZ9aBcDeFgHiJkLmNo", "file_name": "file.pdf"}]}',

'message': 'File "file.pdf" uploaded with ID: 1XyZ9aBcDeFgHiJkLmNo'

}

}

Notes:

- The Drive API service (`drive_service`) must be authenticated.

- Automatically detects MIME type using `mimetypes` module.

- Supports common file types (PDF, image, text, etc.).

Raises:

Exception: Any Drive API errors are caught and returned in the message.

"""

status = 'success'

message = ''

file_id = ''

meta_data = {}

try:

if not file_name:

file_name = os.path.basename(file_path)

mime_type, _ = mimetypes.guess_type(file_path)

if not mime_type:

mime_type = 'application/octet-stream'

file_metadata = {

'name': file_name

}

if parent_folder_id:

file_metadata['parents'] = [parent_folder_id]

media = MediaFileUpload(file_path, mimetype=mime_type, resumable=True)

uploaded_file = self.drive_service.files().create(

body=file_metadata,

media_body=media,

fields='id'

).execute()

file_id = uploaded_file.get('id')

message = f'File "{file_name}" uploaded with ID: {file_id}'

except Exception as e:

status = 'error'

message = f'Error: {str(e)}'

meta_data = {

'parent_folder_id': parent_folder_id,

'file_id': file_id,

'file_name': file_name

}

return {

'status': status,

'response': {

'meta_data': meta_data,

'data': json.dumps({"records": [meta_data]}),

'message': message

},

'message': message

}

def create_gdoc(self, title=None, parent_folder_id=None, user_id=None):

"""

Creates a new Google Docs document in Google Drive, optionally inside a specified parent folder.

This method uses the Google Drive API to create a blank Google Docs document with a specified title.

If a `parent_folder_id` is provided, the document will be created inside that folder.

Parameters:

title (str): The title of the Google Docs document to be created.

parent_folder_id (str, optional): The ID of the parent folder where the document should be created.

If not provided, the document will be created at the root level.

Returns:

dict: A dictionary with:

- 'status' (str): 'success' if the document was created, otherwise 'error'.

- 'response' (dict):

- 'meta_data' (dict): Includes `parent_folder_id`, `doc_id`, and `title` of the created document.

- 'data' (str): JSON-encoded version of the document metadata under the `records` key.

- 'message' (str): Human-readable message about the result.

Example:

>>> create_gdoc(title='Meeting Notes', parent_folder_id='1AbCdEfGhIj')

{

'status': 'success',

'response': {

'meta_data': {

'parent_folder_id': '1AbCdEfGhIj',

'doc_id': '1XyZ9aBcDeFgHiJkLmNo',

'title': 'Meeting Notes'

},

'data': '{"records":[{"parent_folder_id": "1AbCdEfGhIj", "doc_id": "1XyZ9aBcDeFgHiJkLmNo", "title": "Meeting Notes"}]}',

'message': 'Document "Meeting Notes" created with ID: 1XyZ9aBcDeFgHiJkLmNo'

}

}

Notes:

- Uses MIME type `'application/vnd.google-apps.document'` to specify a Google Docs file.

- The Drive API service (`drive_service`) must be authenticated.

- If an error occurs, the `doc_id` will remain empty and the error will be reflected in the message.

Raises:

Exception: Any Drive API errors are caught and returned in the message.

"""

status = 'success'

message = ''

meta_data = {}

doc_id=''

try:

doc_metadata = {'name': title, 'mimeType': 'application/vnd.google-apps.document'}

if parent_folder_id:

doc_metadata['parents'] = [parent_folder_id]

doc = self.drive_service.files().create(body=doc_metadata, fields='id').execute()

message=f'Document "{title}" created with ID: {doc.get("id")}'

doc_id = doc.get('id')

except Exception as e:

message = f'Error: {str(e)}'

meta_data = {

'parent_folder_id': parent_folder_id,

'doc_id': doc_id,

'title': title

}

return {

'status': status,

'response': {

'meta_data': meta_data,

'data': json.dumps({"records":[meta_data]}),

'message': message

}

}

def move_gdrive_file_to_folder(self, file_id, folder_id, user_id=None):

"""

Moves a file in Google Drive to a specified folder.

This method uses the Google Drive API to move a file (`file_id`) from its current parent folder(s)

to a new destination folder (`folder_id`). It does so by:

- Retrieving the file's current parent folders.

- Adding the new parent.

- Removing the previous parents.

Parameters:

file_id (str): The ID of the file to move.

folder_id (str): The ID of the destination folder.

Returns:

dict: A dictionary with:

- 'status' (str): 'success' if the operation was successful, 'error' otherwise.

- 'response' (dict):

- 'meta_data' (dict): Contains `file_id`, list of new parents, and previous parents.

- 'data' (str): JSON-encoded string containing the metadata.

- 'message' (str): Human-readable description of the operation outcome.

Example:

>>> move_gdrive_file_to_folder('1AbCdEfG123', '9XyZtUvW456')

{

'status': 'success',

'response': {

'meta_data': {

'file_id': '1AbCdEfG123',

'new_parents': ['9XyZtUvW456'],

'previous_parents': ['4PrEvIoUs456']

},

'data': '{"records": [{"file_id": "1AbCdEfG123", ...}]}',

'message': 'File 1AbCdEfG123 successfully moved to folder 9XyZtUvW456.'

}

}

Notes:

- This method replaces all previous parent folders with the new one.

- The file will no longer be visible in any of its former folders.

- Requires that the authenticated Drive API user has `writer` or higher permissions on both the file and the destination folder.

Raises:

Exception: Any Drive API errors encountered during retrieval or update are caught and returned in the error response.

"""

try:

# First get the existing parents

file = self.drive_service.files().get(fileId=file_id, fields='parents').execute()

previous_parents = ",".join(file.get('parents', []))

# Move the file to the new folder

updated_file = self.drive_service.files().update(

fileId=file_id,

addParents=folder_id,

removeParents=previous_parents,

fields='id, parents'

).execute()

# Construct metadata and message

meta_data = {

"file_id": file_id,

"new_parents": updated_file.get('parents', []),

"previous_parents": previous_parents.split(",") if previous_parents else []

}

message = f"File {file_id} successfully moved to folder {folder_id}."

data_json = json.dumps({"records": [meta_data]})

return {

"status": "success",

"response": {

"meta_data": meta_data,

"data": data_json,

"message": message

}

}

except Exception as e:

return {

"status": "error",

"response": {

"message": str(e)

}

}

def copy_file_to_gdrive_folder(self, file_id=None, new_folder_id=None, new_name=None, batch=None, user_id=None):

"""

Copies a file to a specified folder in Google Drive, with overwrite protection and optional batching.

This method performs a smart file copy to a target folder in Drive. It:

- Checks if a file with the same name already exists in the destination folder.

- Compares modification timestamps.

- Overwrites older versions if needed.

- Optionally performs the copy as part of a batch request.

Parameters:

file_id (str): The ID of the source file to copy.

new_folder_id (str): The ID of the destination folder.

new_name (str, optional): The name of the new file. If not provided, the source file’s name is reused.

batch (googleapiclient.http.BatchHttpRequest, optional): A batch object to which the copy request is added for batched execution.

Returns:

dict: A dictionary containing:

- 'status' (str):

- 'success' if the file was copied,

- 'skipped' if the destination file is newer or the same,

- 'error' in case of failure.

- 'response' (dict):

- 'meta_data' (dict): Metadata about the operation including `folder_id`, `file_id`, and `new_name`.

- 'data' (str): JSON-encoded metadata in the format `{"records": [...]}`.

- 'message' (str): A human-readable description of what occurred (e.g., skipped, copied, added to batch).

Example:

>>> copy_file_to_gdrive_folder(

file_id='1AbCdEfGh',

new_folder_id='9XyZtUvW',

new_name='Report_Copy.docx'

)

{

'status': 'success',

'response': {

'meta_data': {

'folder_id': '9XyZtUvW',

'file_id': '7LmNoPqRs',

'new_name': 'Report_Copy.docx'

},

'data': '{"records": [{"folder_id": "9XyZtUvW", ...}]}',

'message': 'Copied file "Report_Copy.docx" (ID: 7LmNoPqRs)'

}

}

Logic Summary:

1. Retrieves the file name if not provided.

2. Checks for existing files in the destination folder with the same name.

3. Compares modification timestamps:

- If destination is newer or same, skip copy.

- If destination is older, delete it and proceed to copy.

4. Copies the file using the Google Drive API.

5. Adds the request to a batch if `batch` is provided; otherwise executes immediately.

Notes:

- The MIME type is inferred from the original file and retained.

- File overwrite is handled safely based on timestamp comparison (`modifiedTime`).

- Batch copying enables faster execution of multiple copy operations.

- Requires appropriate access permissions on both source file and destination folder.

Raises:

Exception: Any API errors during metadata retrieval, deletion, or copy will be caught and returned in the response.

"""

status = 'success'

message = ''

meta_data = {}

new_file_id = ''

# Step 1: Determine name to check

name_to_check = new_name

if not name_to_check:

file_metadata = self.drive_service.files().get(fileId=file_id, fields='name').execute()

name_to_check = file_metadata['name']

# Step 2: Check if file already exists

query = (

f"'{new_folder_id}' in parents and "

f"name='{name_to_check}' and trashed=false"

)

existing_files = self.drive_service.files().list(q=query, fields="files(id, modifiedTime)").execute().get('files', [])

if existing_files:

# Step 3: Compare modified times

source_file_metadata = self.drive_service.files().get(

fileId=file_id, fields='modifiedTime'

).execute()

source_modified = source_file_metadata['modifiedTime']

dest_modified = existing_files[0]['modifiedTime']

src = isoparse(source_modified)

dst = isoparse(dest_modified)

if src <= dst:

message = f"Skipping '{name_to_check}' — destination is newer or same."

meta_data = {

'folder_id': new_folder_id,

'file_id': file_id,

'new_name': new_name

}

return {

'status': 'skipped',

'response': {

'meta_data': meta_data,

'data': json.dumps({"records":[meta_data]}),

'message': message

}

}

# Step 4: Delete older destination file

self.drive_service.files().delete(fileId=existing_files[0]['id']).execute()

message = f"Overwriting '{name_to_check}' — source is newer."

# Step 5: Prepare metadata and copy file

copied_file_metadata = {

'parents': [new_folder_id],

'name': name_to_check

}

if new_name:

copied_file_metadata['name'] = new_name

def callback(request_id, response, exception):

if exception:

print(f"Batch error copying file: {exception}")

else:

print(f"Batch copied file '{response['name']}' to folder ID {new_folder_id}")

if batch:

request = self.drive_service.files().copy(

fileId=file_id,

body=copied_file_metadata,

fields='id, name'

)

batch.add(request, callback=callback)

message = f"Copy request added to batch for file '{name_to_check}'"

else:

copy_response = self.drive_service.files().copy(

fileId=file_id,

body=copied_file_metadata,

fields='id, name'

).execute()

new_file_id = copy_response['id']

message = f"Copied file '{copy_response['name']}' (ID: {new_file_id})"

meta_data = {

'folder_id': new_folder_id,

'file_id': new_file_id or file_id,

'new_name': new_name or name_to_check

}

return {

'status': status,

'response': {

'meta_data': meta_data,

'data': json.dumps({"records":[meta_data]}),

'message': message

}

}

def copy_gdrive_folder_recursive(self, source_folder_id=None, destination_parent_folder_id=None, new_folder_name=None, user_id=None):

"""

Recursively copies a folder from one location in Google Drive to another, preserving structure and contents.

This method:

- Copies an entire folder (and its subfolders/files) from `source_folder_id` into a destination parent folder.

- Skips copying if the destination contains a newer version of a file with the same name.

- Uses batch operations for efficient copying of multiple files.

- Reuses existing destination folders if one with the same name already exists.

Parameters:

source_folder_id (str): The ID of the Google Drive folder to copy.

destination_parent_folder_id (str): The ID of the parent folder where the copied structure will be placed.

new_folder_name (str, optional): Optional custom name for the new root folder. Defaults to the source folder's name.

Returns:

dict: A dictionary containing:

- 'status' (str): 'success' if the operation completed without exception, otherwise 'error'.

- 'response' (dict):

- 'meta_data' (dict): Includes `source_folder_id` and `new_folder_id`.

- 'data' (str): JSON-encoded metadata.

- 'message' (str): A detailed multi-line message log of all actions performed (copied, skipped, errors, etc.).

Example:

>>> copy_gdrive_folder_recursive(

source_folder_id='1SourceAbCdEf',

destination_parent_folder_id='1DestXyZ123',

new_folder_name='Backup_2025'

)

{

'status': 'success',

'response': {

'meta_data': {

'source_folder_id': '1SourceAbCdEf',

'new_folder_id': '3NewBackupFolderId'

},

'data': '{"records": [{"source_folder_id": "1SourceAbCdEf", "new_folder_id": "3NewBackupFolderId"}]}',

'message': 'Created new folder "Backup_2025" with ID: ...\nCopied file "report.pdf"...'

}

}

Features:

- Automatically reuses existing folders if they match the `new_folder_name`.

- Skips overwriting if destination file is newer or the same.

- Deletes older destination files before replacing them.

- Uses batch API for efficient file copying.

- Fully recursive — all nested folders and files are handled.

Notes:

- Folder creation uses `mimeType='application/vnd.google-apps.folder'`.

- Only non-trashed files/folders are considered.

- Requires permission to read from the source and write to the destination.

Raises:

Exception: Any errors during metadata retrieval, folder creation, or copying are caught and returned in the `message`.

Known Limitation:

- The function assumes `drive_service` is accessible as a global or instance variable. Consider passing it explicitly.

- Large folder trees might hit rate limits; exponential backoff or delay strategies can be added if needed.

"""

status = 'success'

messages = []

new_folder_id = ''

try:

if not new_folder_name:

source_folder = self.drive_service.files().get(fileId=source_folder_id, fields='name').execute()

new_folder_name = source_folder['name']

# Check if destination folder already exists

query = (

f"mimeType='application/vnd.google-apps.folder' and "

f"'{destination_parent_folder_id}' in parents and "

f"name='{new_folder_name}' and trashed=false"

)

results = self.drive_service.files().list(q=query, fields="files(id)").execute()

existing = results.get('files', [])

if existing:

new_folder_id = existing[0]['id']

messages.append(f"Using existing folder '{new_folder_name}' with ID: {new_folder_id}")

else:

new_folder_metadata = {

'name': new_folder_name,

'mimeType': 'application/vnd.google-apps.folder',

'parents': [destination_parent_folder_id]

}

new_folder = self.drive_service.files().create(body=new_folder_metadata, fields='id').execute()

new_folder_id = new_folder['id']

messages.append(f"Created new folder '{new_folder_name}' with ID: {new_folder_id}")

# List contents of source folder

query = f"'{source_folder_id}' in parents and trashed=false"

response = self.drive_service.files().list(q=query, spaces='drive', fields="files(id, name, mimeType)").execute()

items = response.get('files', [])

batch = self.drive_service.new_batch_http_request()

def callback(request_id, response, exception):

nonlocal messages

if exception:

messages.append(f"Error copying file: {exception}")

else:

messages.append(f"Copied file '{response['name']}' to folder ID {new_folder_id}")

for item in items:

item_id = item['id']

item_name = item['name']

item_type = item['mimeType']

if item_type == 'application/vnd.google-apps.folder':

messages.append(f"Recursively copying folder: {item_name}")

sub_result = self.copy_gdrive_folder_recursive(

source_folder_id=item_id,

destination_parent_folder_id=new_folder_id,

new_folder_name=item_name

)

messages.append(sub_result['response']['message'])

else:

# Check for duplicate

file_query = (

f"'{new_folder_id}' in parents and "

f"name='{item_name}' and trashed=false"

)

existing_files = self.drive_service.files().list(

q=file_query, fields="files(id, modifiedTime)"

).execute().get('files', [])

if existing_files:

source_file_metadata = self.drive_service.files().get(

fileId=item_id, fields='modifiedTime'

).execute()

source_modified = source_file_metadata['modifiedTime']

dest_modified = existing_files[0]['modifiedTime']

if source_modified <= dest_modified:

messages.append(f"Skipping '{item_name}' — destination is newer or same.")

continue

self.drive_service.files().delete(fileId=existing_files[0]['id']).execute()

messages.append(f"Overwriting '{item_name}' — source is newer.")

copied_file_metadata = {

'parents': [new_folder_id],

'name': item_name

}

request = self.drive_service.files().copy(

fileId=item_id,

body=copied_file_metadata,

fields='id, name'

)

batch.add(request, callback=callback)

batch.execute()

except Exception as e:

status = 'error'

messages.append(f'Error: {str(e)}')

meta_data = {

'source_folder_id': source_folder_id,

'new_folder_id': new_folder_id

}

return {

'status': status,

'response': {

'meta_data': meta_data,

'data': json.dumps({"records":[meta_data]}),

'message': "\n".join(messages)

}

}

def parse_markdown(self, text=None, user_id=None):

"""Extended markdown parser for headings, bold, italic, italic+bold, and hyperlinks."""

elements = []

lines = text.split('\n')

index = 1 # Google Docs indexes start at 1

for line in lines:

original_index = index

content = line