カテゴリー: GoogleDriveAPI

日付形式の問題解決とイベント抽出とフィルタリング

日付形式の問題解決とイベント抽出とフィルタリング

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

このような場合、統一的に処理するためのフォーマット変換を行う必要がある

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from datetime import datetime
import re
 
def convert_japanese_date(japanese_date):
    # 和暦の例: "令和6年10月26日"
    pattern = r"令和(\d+)年(\d+)月(\d+)日"
    match = re.match(pattern, japanese_date)
    if match:
        year = int(match.group(1)) + 2018  # 令和元年は2019年に相当
        month = int(match.group(2))
        day = int(match.group(3))
        return f"{year}-{month:02d}-{day:02d}"
    # ISO形式のチェック
    try:
        datetime.strptime(japanese_date, '%Y-%m-%d')
        return japanese_date
    except ValueError:
        return None  # 無効な日付形式

また
日付形式の変換後、無効な日付やイベント名が空のエントリをフィルタリングするように変更

eventsリストから有効なエントリのみを抽出

1
2
3
4
5
6
7
8
9
def filter_events(events):
    valid_events = []
    for event in events:
        date = convert_japanese_date(event['date'])
        if date and event['event']:
            valid_events.append({'date': date, 'event': event['event']})
        else:
            print(f"無効な日付形式: {event['date']}")
    return valid_events

日付の変換やイベントのフィルタリングを他のスクリプトでも使うため
モジュール化する

event_utils.pyというモジュールを作成し
これに記述する

1
touch event_utils.py

内容は

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from datetime import datetime
import re
 
def convert_japanese_date(japanese_date):
    pattern = r"令和(\d+)年(\d+)月(\d+)日"
    match = re.match(pattern, japanese_date)
    if match:
        year = int(match.group(1)) + 2018  # 令和元年は2019年に相当
        month = int(match.group(2))
        day = int(match.group(3))
        return f"{year}-{month:02d}-{day:02d}"
    # ISO形式のチェック
    try:
        datetime.strptime(japanese_date, '%Y-%m-%d')
        return japanese_date
    except ValueError:
        return None  # 無効な日付形式
 
def filter_events(events):
    valid_events = []
    for event in events:
        date = convert_japanese_date(event['date'])
        if date and event['event']:
            valid_events.append({'date': date, 'event': event['event']})
        else:
            print(f"無効な日付形式: {event['date']}")
    return valid_events

これをmain4.pyで
event_utils.pyをインポートして使用する

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
from event_utils import convert_japanese_date, filter_events
 
# フォルダIDを指定して処理を開始
folder_id = "" #Folder ID
texts = extract_texts_from_folder(folder_id)
 
if not texts:
    print("フォルダ内に解析するテキストがありません。")
else:
    for text_content in texts:
        raw_events = parse_text_with_ollama(text_content, model_name='elyza:jp8b')
        print("抽出されたイベント:")
        for event in raw_events:
            print(event)
 
        # イベントのフィルタリングとフォーマット
        events = filter_events(raw_events)
 
        print("有効なイベント:")
        for event in events:
            print(event)
 
        if events:
            add_events_to_calendar(events, calendar_id='primary', token_file='token.json', credentials_file='credentials.json')
        else:
            print("有効なイベントがありません。")

これで
eventsリストからNoneのイベントを削除するフィルタリング処理が追加
Noneを削除したイベントのみをカレンダーに追加

1
events = [event for event in events if event['event'] is not None]

の部分がその処理

また
eventsリストが空でないかを確認し、有効なイベントがある場合のみカレンダーに追加
もしイベントがなければ、カレンダーには何も追加されず、
「有効なイベントがありません。」というメッセージが表示される

このチェックによって、無駄な処理が減り、空のイベントがカレンダーに追加されるのを防ぐ

1
2
3
4
if events:
    add_events_to_calendar(events, calendar_id='primary', token_file='token.json', credentials_file='credentials.json')
else:
    print("有効なイベントがありません。")

これで実行すると
運動会のイベントが失敗している

これは
OllamaからのレスポンスがJSON形式として期待されているのに対し、
不正なJSON形式
（複数のJSONオブジェクトがコンマで区切られている形式や、余分なテキストが含まれている形式）
が返ってきていることが考えられます。

これを解決するために、
Ollamaのレスポンスを受け取る際に、
例外処理や、JSON以外の余計な部分を取り除く前処理を追加

レスポンスを前処理する関数を追加し、JSON形式に整形
JSON解析エラーの例外処理を追加して、不正な形式のデータをスキップ

llamaからのレスポンスを受け取って前処理し、JSON形式に変換するには

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import json
import re
 
def parse_ollama_response(response_text):
    # 正規表現でJSON配列形式のみを抽出
    json_pattern = re.compile(r'\[.*?\]', re.DOTALL)
    match = json_pattern.search(response_text)
    if not match:
        print("JSON形式のレスポンスが見つかりません。")
        return []
 
    json_data = match.group(0)  # 最初のJSON形式の部分を取得
 
    try:
        events = json.loads(json_data)
        return events
    except json.JSONDecodeError as e:
        print("Ollamaからのレスポンスの解析に失敗しました:", str(e))
        return []

parse_ollama_response関数を
parse_text_with_ollama関数の後処理として使用し、Ollamaからのレスポンスを整形
これにより、Ollamaからの不正なレスポンス形式を取り除く

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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
import json
import re
 
# OllamaからのレスポンスをJSON形式に変換
def parse_ollama_response(response_text):
    # JSON配列形式のみを抽出する正規表現
    json_pattern = re.compile(r'\[.*?\]', re.DOTALL)
    match = json_pattern.search(response_text)
    if not match:
        print("JSON形式のレスポンスが見つかりません。")
        return []
 
    json_data = match.group(0)  # 最初のJSON形式の部分を取得
 
    try:
        events = json.loads(json_data)
        return events
    except json.JSONDecodeError as e:
        print("Ollamaからのレスポンスの解析に失敗しました:", str(e))
        return []
 
# 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でテキストを解析（モデル名を指定）
        raw_response = parse_text_with_ollama(text_content, model_name='elyza:jp8b')
         
        # Ollamaのレスポンスを前処理してJSONに変換
        events = parse_ollama_response(raw_response)
 
        print("抽出されたイベント:")
        for event in events:
            print(event)
 
        # Noneのイベントを除外
        events = [event for event in events if event['event']]
 
        # フィルタリング後のイベントを表示
        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("有効なイベントがありません。")

としたが

TypeError: expected string or bytes-like object というエラーは、
response_textのデータ型が文字列やバイト列ではない場合に発生

これは、response_textが予期せず辞書（dict）型や他の非文字列データ型である場合に起こることがよくある

response_textの型が文字列であることを確認し、
もし辞書型など別の型が返ってきている場合には、JSONエンコードして文字列に変換する処理を追加

レスポンスが辞書型か確認し、文字列に変換
parse_text_with_ollamaから返ってくるレスポンスが文字列であるかをチェックし、
もし辞書型など別の型であれば文字列に変換する処理を追加

型チェックを行う
parse_ollama_response関数内で
response_textが文字列であることをチェックし、文字列でない場合には変換

以下修正

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import json
import re
 
def parse_ollama_response(response_text):
    # レスポンスが辞書型なら文字列に変換
    if isinstance(response_text, dict):
        response_text = json.dumps(response_text)  # 辞書をJSON文字列に変換
    elif not isinstance(response_text, str):
        print("不明なデータ形式のレスポンスを受け取りました:", type(response_text))
        return []
     
    # JSON配列形式のみを抽出する正規表現
    json_pattern = re.compile(r'\[.*?\]', re.DOTALL)
    match = json_pattern.search(response_text)
    if not match:
        print("JSON形式のレスポンスが見つかりません。")
        return []
 
    json_data = match.group(0)  # 最初のJSON形式の部分を取得
 
    try:
        events = json.loads(json_data)
        return events
    except json.JSONDecodeError as e:
        print("Ollamaからのレスポンスの解析に失敗しました:", str(e))
        return []

という処理を追加

response_textが辞書型かを確認します。辞書型の場合はjson.dumps()を使って文字列に変換

次に、正規表現を使用してレスポンス内のJSON配列形式を抽出

json.loadsを使って抽出した部分をJSONとしてパースし、
もしエラーが発生した場合にはエラーメッセージを表示

逆にエラーになった

このため一度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
from event_utils import convert_japanese_date, filter_events
 
# フォルダIDを指定して処理を開始
folder_id = ""
texts = extract_texts_from_folder(folder_id)
 
if not texts:
    print("フォルダ内に解析するテキストがありません。")
else:
    for text_content in texts:
        raw_events = parse_text_with_ollama(text_content, model_name='elyza:jp8b')
        print("抽出されたイベント:")
        for event in raw_events:
            print(event)
 
        # イベントのフィルタリングとフォーマット
        events = filter_events(raw_events)
 
        print("有効なイベント:")
        for event in events:
            print(event)
 
        if events:
            add_events_to_calendar(events, calendar_id='primary', token_file='token.json', credentials_file='credentials.json')
        else:
            print("有効なイベントがありません。")

に戻したら
となって無事にイベントが追加された

次は
メールから読み込んで実行できるようにする

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

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

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

1
drive_pdf_extractor.py

を作成する

1
touch drive_pdf_extractor.py

でファイルを作成し

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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

として保存

1
touch main3.py

で

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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 を変更

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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

というように

1
drive_pdf_extractor.py

を修正

修正したのは

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

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

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

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

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

GoogleDriveの指定フォルダからファイルを取得し処理

GoogleDriveの指定フォルダからファイルを取得し処理

Google Drive API でフォルダのIDを取得できたので
次にここにpdfファイルを配置して
これを取得しテキストを抽出する

PyPDF2 と pdfminer.six はどちらも PDF ファイルを操作するための Python ライブラリ

PyPDF2 には PDF ファイルの結合、分割、ページの回転、ページの追加・削除など、PDF のページ操作機能が充実

pdfminer.six は PDF を読み取ってテキストデータを抽出することが主目的のため、ページ操作や編集の機能はない

PyPDF2 は比較的軽量で、簡単なテキスト抽出やページ操作には高速に動作しますが、PDFの内容解析が簡易的

pdfminer.six はテキスト抽出においては非常に詳細な処理を行うため、
特に長い PDF や複雑なレイアウトのファイルでは処理時間が長くなる

ということでpdfminer.sixを使いテキストを取り出す

そしてその処理結果を渡すというモジュールを作成する

Google Drive の
School フォルダの ID: ここからpdfファイルを取得し
pdfminer.sixを使ってPDFからテキストを抽出したい

Google Drive の School フォルダの ID:
ここからpdfファイルを取得し pdfminer.sixを使ってPDFからテキストを抽出したい
個人で使うので token.jsonを使ったコードにする
抽出したテキストを処理する関数に渡すようにするため 抽出したテキストをOllamaに送信は不要。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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
 
# テキストを処理する関数
def process_text(text):
    """抽出したテキストを処理します"""
    print("Extracted Text:\n", text)
    # テキストの処理ロジックをここに追加します
 
# フォルダ内のPDFを処理
def process_pdfs_in_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
 
    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:
            process_text(pdf_text)
        else:
            print(f"{pdf_file['name']} からテキストを抽出できませんでした。")
 
# フォルダIDを指定して処理を開始
folder_id = ""  # SchoolフォルダのID
process_pdfs_in_folder(folder_id)

これを

1
python get_pdf_read.py

で実行するとかなりの精度で取り出すことができた

次はこれをモジュールにしてollamaに処理結果を渡せるようにする

なおコード解説は
コードの説明
1. authenticate_drive関数: token.jsonを使用してGoogle Drive APIに認証し、Driveサービスを取得
2. list_pdf_files_in_folder関数: 指定したフォルダID内のPDFファイルをリスト化
3. download_pdf_from_drive関数: 各PDFファイルをダウンロードし、バイナリデータを返す
4. extract_text_from_pdf関数: pdfminer.sixを使ってPDFデータからテキストを抽出
5. process_text関数: 抽出したテキストを処理します。ここに任意の処理ロジックを追加
6. process_pdfs_in_folder関数: フォルダ内の全PDFファイルを処理し、各PDFのテキストを抽出後にprocess_text関数に渡す
これで、抽出されたテキストを直接処理する

なおGPTで生成したコードでは

というスコープだが
これだと認証エラーになる

で回避可能

これは google カレンダーの時も同様で

なら認証エラーにならない

Google Drive APIで新規フォルダ作成

Google Drive API を使い、my drive のなかに Schoolフォルダを作成したい
さらにSchoolフォルダのなかにReadフォルダを作成したい

ここへPDFファイルを置いて
このなかにファイルがあるならOCRしてカレンダーに予定を書き込み
処理終了したら ReadフォルダにPDFファイルを移動する

この時に同じフォルダがあると同じフォルダが複数できるらしいので
チェック機能を追加する

なので
Google Drive API を使用して My Drive 内に「School」フォルダを作成（または既存のものを使用）し、その中に「Read」フォルダを作成する

1
pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib

で必要なライブラリを入れる

Google Cloud Console でプロジェクトを作成し、Google Drive API を有効化も忘れずに

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from __future__ import print_function
import os.path
from google.oauth2 import credentials
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
 
# 認証と API クライアントのセットアップ
SCOPES = ['https://www.googleapis.com/auth/drive']
 
def main():
    """Google Drive API に接続し、フォルダを作成または取得します。"""
    creds = None
    # token.json はユーザーのアクセストークンとリフレッシュトークンを保存します。初回実行時に自動的に作成されます。
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    # 資格情報がないか、無効または期限切れの場合は再ログインします。
    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)
        # 認証情報を保存します。
        with open('token.json', 'w') as token:
            token.write(creds.to_json())
 
    # Drive API クライアントを作成
    service = build('drive', 'v3', credentials=creds)
 
    # 1. "School" フォルダを取得または作成
    school_folder_id = get_or_create_folder(service, 'School', parent_id=None)
 
    # 2. "Read" フォルダを "School" フォルダ内に作成
    read_folder_id = get_or_create_folder(service, 'Read', parent_id=school_folder_id)
 
    print('School フォルダの ID: %s' % school_folder_id)
    print('Read フォルダの ID: %s' % read_folder_id)
 
def get_or_create_folder(service, folder_name, parent_id=None):
    """
    フォルダを取得または作成します。
 
    Parameters:
        service: Drive API サービス インスタンス。
        folder_name (str): フォルダの名前。
        parent_id (str): 親フォルダの ID（省略可能）。
 
    Returns:
        str: フォルダの ID。
    """
    # フォルダを検索
    query = "name='{}' and mimeType='application/vnd.google-apps.folder' and trashed=false".format(folder_name)
    if parent_id:
        query += " and '{}' in parents".format(parent_id)
 
    response = service.files().list(
        q=query,
        spaces='drive',
        fields='files(id, name)',
    ).execute()
    files = response.get('files', [])
 
    if files:
        # 既存のフォルダが見つかった場合
        folder_id = files[0]['id']
        print('既存の "{}" フォルダの ID を使用します: {}'.format(folder_name, folder_id))
    else:
        # フォルダが存在しない場合、新規作成
        file_metadata = {
            'name': folder_name,
            'mimeType': 'application/vnd.google-apps.folder',
        }
        if parent_id:
            file_metadata['parents'] = [parent_id]
 
        file = service.files().create(body=file_metadata, fields='id').execute()
        folder_id = file.get('id')
        print('新規に "{}" フォルダを作成しました。ID: {}'.format(folder_name, folder_id))
 
    return folder_id
 
if __name__ == '__main__':
    main()

を実行したが

1
2
3
4
5
6
7
8
9
10
11
12
Traceback (most recent call last):
  File "/Users/snowpool/aw10s/week_calendar_voice/quickstart.py", line 59, in <module>
    main()
  File "/Users/snowpool/aw10s/week_calendar_voice/quickstart.py", line 26, in main
    creds.refresh(Request())
  File "/Users/snowpool/.pyenv/versions/3.10.6/lib/python3.10/site-packages/google/oauth2/credentials.py", line 335, in refresh
    ) = reauth.refresh_grant(
  File "/Users/snowpool/.pyenv/versions/3.10.6/lib/python3.10/site-packages/google/oauth2/reauth.py", line 351, in refresh_grant
    _client._handle_error_response(response_data, retryable_error)
  File "/Users/snowpool/.pyenv/versions/3.10.6/lib/python3.10/site-packages/google/oauth2/_client.py", line 73, in _handle_error_response
    raise exceptions.RefreshError(
google.auth.exceptions.RefreshError: ('invalid_scope: Bad Request', {'error': 'invalid_scope', 'error_description': 'Bad Request'})

となる

認証関連でエラーになるので
一度検証

1
2
3
4
mkdir g_drive
cd g_drive
cp ../credentials.json .
cp ../create_folder.py .

で

1
python create_folder.py

で実行すると
認証画面になるので進めていく

これで Google Drive APIが使えて
新規フォルダ作成までできた

次はPDFファイルを読み込んでテキストを抽出する

それができたら
抽出した内容をollamaに渡してカレンダーに入れる

一度テキストファイルに書き出すかGPTで調べたが
内容の多さを基準にする場合、以下の点を目安に考えると良いでしょう。

### 1. **APIの入力制限**
– Ollama APIや多くのテキスト処理APIは、入力テキストの長さに制限があります。
一般的に**数千文字以内**であれば直接渡しても問題ないことが多いです。
– 確認しているAPIの制限が**4,000〜8,000文字**程度であれば、
1〜2ページ程度の短めのPDFであれば直接渡せます。
内容がこの文字数を超える場合は、分割するかファイルで管理した方が良いでしょう。

### 2. **テキストの内容**
– **数百文字〜数千文字程度**であれば、直接渡してもスムーズに処理できます。
一般的には、A4サイズのPDF1〜2ページ程度です。
– 内容が**5ページ以上**ある場合や、**10,000文字以上の大量のテキスト**になる場合は、
テキストファイルに保存し分割しながら処理した方が管理が簡単です。

### 3. **利用シナリオ**
– **リアルタイム性が重要**な場合：数千文字までで、
可能であれば直接APIに渡す方がシンプルです。
– **大規模なドキュメントや長文の内容を確認しながら処理したい**場合：
ファイルに書き出して保存し、分割して処理する方が確認がしやすくなります。

### 具体的な基準
– **3,000文字以下**：直接Ollamaに渡す
– **3,000〜10,000文字程度**：可能であれば直接渡すが、ファイルに書き出して保存も検討
– **10,000文字以上**：テキストファイルに書き出し、必要に応じて分割して処理

例えば、10,000文字はおおよそA4サイズのPDFファイルで5ページ前後に相当します。
この文字数を基準に、リアルタイム性や内容の確認のしやすさに応じて直接渡すか、
ファイルに保存してから渡すかを判断すると良いでしょう。

ということなので
PDFの処理結果を直接渡すことにする