GoogleDriveの指定フォルダからファイルを取得し処理のモジュール化と処理結果を ollamaへ渡す

GoogleDriveの指定フォルダからファイルを取得し処理のモジュール化と処理結果を ollamaへ渡す

Google DriveからPDFを取得し、テキストを抽出する機能をモジュールに分ける

drive_pdf_extractor.py

を作成する

touch drive_pdf_extractor.py

でファイルを作成し

import os
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from googleapiclient.http import MediaIoBaseDownload
from pdfminer.high_level import extract_text
from io import BytesIO

# Google Drive APIの認証
def authenticate_drive():
    SCOPES = ['https://www.googleapis.com/auth/drive']
    creds = None
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    else:
        print("認証トークンが見つかりません。認証を実行してください。")
        return None
    return build('drive', 'v3', credentials=creds)

# フォルダ内のPDFファイルリストを取得
def list_pdf_files_in_folder(service, folder_id):
    """Google Driveフォルダ内のPDFファイルのリストを取得します"""
    query = f"'{folder_id}' in parents and mimeType='application/pdf'"
    results = service.files().list(q=query, fields="files(id, name)").execute()
    files = results.get('files', [])
    return files

# Google DriveからPDFファイルをダウンロード
def download_pdf_from_drive(service, file_id):
    """Google DriveからPDFファイルをダウンロードし、バイナリデータを返します"""
    request = service.files().get_media(fileId=file_id)
    file_data = BytesIO()
    downloader = MediaIoBaseDownload(file_data, request)
    done = False
    while not done:
        status, done = downloader.next_chunk()
        print(f"Download Progress: {int(status.progress() * 100)}%")
    file_data.seek(0)
    return file_data

# PDFからテキストを抽出
def extract_text_from_pdf(pdf_data):
    """PDFファイルデータからテキストを抽出します"""
    text = extract_text(pdf_data)
    return text

# 指定したフォルダ内のすべてのPDFファイルのテキストを抽出
def extract_texts_from_folder(folder_id):
    """フォルダ内のPDFファイルからテキストを抽出し、リストとして返します"""
    service = authenticate_drive()
    if not service:
        return []

    pdf_files = list_pdf_files_in_folder(service, folder_id)
    if not pdf_files:
        print("指定されたフォルダにPDFファイルが見つかりません。")
        return []

    texts = []
    for pdf_file in pdf_files:
        print(f"Processing file: {pdf_file['name']}")
        pdf_data = download_pdf_from_drive(service, pdf_file['id'])
        pdf_text = extract_text_from_pdf(pdf_data)
        if pdf_text:
            texts.append(pdf_text)
        else:
            print(f"{pdf_file['name']} からテキストを抽出できませんでした。")
    return texts

として保存

 touch main3.py

で

from drive_pdf_extractor import extract_texts_from_folder
from ollama_module import parse_text_with_ollama
from google_calendar_module import add_events_to_calendar

# Google DriveのSchoolフォルダID
folder_id = ""

# フォルダ内のPDFファイルからテキストを抽出
texts = extract_texts_from_folder(folder_id)

# テキストが抽出できているか確認
if not texts:
    print("フォルダ内に解析するテキストがありません。")
else:
    for text_content in texts:
        # Ollamaでテキストを解析（モデル名を指定）
        events = parse_text_with_ollama(text_content, model_name='elyza:jp8b')

        # 抽出されたイベントを表示
        print("抽出されたイベント:")
        for event in events:
            print(event)

        # eventがNoneのものを削除
        events = [event for event in events if event['event'] is not None]

        # フィルタリング後のイベントを表示
        print("有効なイベント:")
        for event in events:
            print(event)

        # 有効なイベントがある場合のみGoogleカレンダーに追加
        if events:
            add_events_to_calendar(events, calendar_id='primary', token_file='token.json', credentials_file='credentials.json')
        else:
            print("有効なイベントがありません。")

として保存

しかし認証エラーになる

これは google drive API の時に作成した token.jsonが
Google calendar API の権限と異なるため
なので参照する token.json を変更

import os
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from googleapiclient.http import MediaIoBaseDownload
from pdfminer.high_level import extract_text
from io import BytesIO

# Google Drive APIの認証
def authenticate_drive():
    SCOPES = ['https://www.googleapis.com/auth/drive.readonly']
    token_path = 'g_drive/token.json'  # token.jsonのパスを変更
    creds = None
    if os.path.exists(token_path):
        creds = Credentials.from_authorized_user_file(token_path, SCOPES)
    else:
        print("認証トークンが見つかりません。認証を実行してください。")
        return None
    return build('drive', 'v3', credentials=creds)

# 他の関数も引き続き同様
# フォルダ内のPDFファイルリストを取得
def list_pdf_files_in_folder(service, folder_id):
    query = f"'{folder_id}' in parents and mimeType='application/pdf'"
    results = service.files().list(q=query, fields="files(id, name)").execute()
    files = results.get('files', [])
    return files

# Google DriveからPDFファイルをダウンロード
def download_pdf_from_drive(service, file_id):
    request = service.files().get_media(fileId=file_id)
    file_data = BytesIO()
    downloader = MediaIoBaseDownload(file_data, request)
    done = False
    while not done:
        status, done = downloader.next_chunk()
        print(f"Download Progress: {int(status.progress() * 100)}%")
    file_data.seek(0)
    return file_data

# PDFからテキストを抽出
def extract_text_from_pdf(pdf_data):
    text = extract_text(pdf_data)
    return text

# 指定したフォルダ内のすべてのPDFファイルのテキストを抽出
def extract_texts_from_folder(folder_id):
    service = authenticate_drive()
    if not service:
        return []

    pdf_files = list_pdf_files_in_folder(service, folder_id)
    if not pdf_files:
        print("指定されたフォルダにPDFファイルが見つかりません。")
        return []

    texts = []
    for pdf_file in pdf_files:
        print(f"Processing file: {pdf_file['name']}")
        pdf_data = download_pdf_from_drive(service, pdf_file['id'])
        pdf_text = extract_text_from_pdf(pdf_data)
        if pdf_text:
            texts.append(pdf_text)
        else:
            print(f"{pdf_file['name']} からテキストを抽出できませんでした。")
    return texts

というように

drive_pdf_extractor.py

を修正

修正したのは

token_path = 'g_drive/token.json'  # token.jsonのパスを変更

というように参照先を変えた

これで
Main3.pyを実行したが

レスポンスに含まれている日付形式が「令和6年10月26日」などの和暦表記や、
「2024-10-08」などのISO表記と混在しているため

統一的に処理するためのフォーマット変換を行い
日付形式の変換後、無効な日付やイベント名が空のエントリをフィルタリングするように変更する必要がある