main.py

import os
import json
import airbyte as ab
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from pydantic import BaseModel, Field
from datetime import datetime
import pandas as pd


# If modifying these scopes, delete the file token.json.
SCOPES = ["https://www.googleapis.com/auth/drive"]


def get_credentials():
    """Authenticate user and return credentials."""
    creds = None
    # Check if token.json exists and load credentials
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)

    # If credentials are not valid, refresh or request new ones
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
            creds = flow.run_local_server(port=0)
        # Save new credentials
        with open("token.json", "w") as token:
            token.write(creds.to_json())
    return creds


def load_credentials_from_token():
    """Extract client_id, client_secret, and refresh_token from token.json."""
    if not os.path.exists("token.json"):
        raise FileNotFoundError("token.json not found. Run authentication first.")

    with open("token.json", "r") as token_file:
        data = json.load(token_file)

    return {
        "auth_type": "Client",
        "client_id": data.get("client_id"),
        "client_secret": data.get("client_secret"),
        "refresh_token": data.get("refresh_token"),
    }


def sync_google_drive():
    """Use Airbyte to sync Google Drive folder."""
    try:
        credentials = load_credentials_from_token()
        folder_url = "https://drive.google.com/drive/folders/10DrawuhFx85xmr8v8vVB6PPOTVWPu7nI"  # Extracted from folder URL

        source = ab.get_source(
            "source-google-drive",
            install_if_missing=True,
            config={
                "folder_url": folder_url,
                "credentials": credentials,
                "streams": [
                {
                    "name": "Csv_data",
                    "format": {
                    "filetype": "csv"
                    },
                   "globs": ["**/*.csv"],
                },
                {
                    "name":"Unstructured_data",
                    "format":{
                        "filetype":"unstructured"
                    }
                }

                ]
            }
        )
        source.check()

        source.select_streams("Unstructured_data")  # Select all streams from the Google Drive source
        read_result = source.read()  # Read the data from the selected streams
        # documents_list = []

        # Convert the read data into document objects 
        for stream_name, cached_dataset in read_result.items():
            print(f"Stream Name: {stream_name}") 
            print(cached_dataset)  

            df = cached_dataset.to_pandas()
            print(df.to_string())  
            
            

    except HttpError as error:
        print(f"An error occurred: {error}")
    except FileNotFoundError as e:
        print(e)


def main():
    """Authenticate and sync Google Drive files using Airbyte."""
    get_credentials()  # Ensure authentication
    sync_google_drive()  # Sync with Airbyte


if __name__ == "__main__":
    main()