【Pythonの基礎知識】AI執事でGoogleタスクを自在に操る（参照・登録・削除・更新）

2025年5月4日

予定は「いつ・何をするか」を決めるものですが、タスクは「やるべきことを忘れずに処理する」ための“実行リスト”です。

スケジュールに書かれていない細かな作業、今日でなくてもいいけれど、いずれ必ずやるべきこと——
こうしたものはタスクとして管理することで頭のメモリを解放できます。

さらに、タスクをLINEから登録・確認できるようになれば、場所やタイミングに縛られず、思いついた瞬間に即記録。
リマインド処理との連携も視野に入り、「やるべきことを忘れないAI執事」が完成に近づきます。

本記事では、PythonでLINE×GoogleタスクAPIを連携し、タスクの参照・登録・削除・更新が行えるAI執事ボットの実装手順を解説します。

タスク連携の全体像と目的
使用ライブラリと認証設定
token.json を取得するために必要な2つの設定
完成ソースと構造の全体像
ChatGPTのプロンプト設計と制御

タスク連携の全体像と目的

AI執事における「タスク管理機能」は、Googleカレンダーとは異なるアプローチで予定や作業を扱います。本章では、Googleタスクとの違いや連携の目的、実際にできる処理について整理します。

タスクと予定の違いとは？

予定（Googleカレンダー）は「時間」が主軸ですが、タスクは「やること」が主軸です。つまり「いつやるか決まっていないが必ず完了したいもの」がタスクとなります。これにより、タスクは時間の空きに合わせて柔軟に消化でき、ユーザーの心理的負荷を減らす役割も担います。

本記事でのAI執事ができること（機能一覧）

このAI執事は、LINEで送信された自然文メッセージをChatGPTで解析し、Googleタスクに関する各種処理を実行します。現在対応している機能は以下の通りです。

カテゴリ	ユーザー発言の例	意図（インテント）	対応関数	処理内容
タスクの参照	タスク一覧を見せて	task_list	listTasks()	未完了タスクをすべて表示
	完了済のタスクを教えて	task_list_completed	listCompletedTasks()	完了フラグ付きのタスクを表示
	期限付きタスクだけ見せて	task_list_due	listTasksWithDue()	未完了かつ期限ありタスクを表示
タスクの登録	明日までに資料作成	task_register	registerTaskWithDue()	期限付きタスクを登録
タスクの登録	テストタスクを登録	task_register	registerTask()	期限なしタスクを登録
タスクの削除	資料作成タスクを削除して	task_delete	deleteTask()	タイトル一致のタスクを削除
タスクの完了	テストタスクを完了にして	task_complete	completeTask()	該当タスクに完了フラグを付与

※補足：
本AI執事では「更新機能」は実装されていません。
タスクの内容を変更したい場合は、ユーザー自身が該当タスクを削除し、修正内容で再登録する運用方式を採用しています。

なお、 Google Tasks API で使用するタスクリストID（ tasklist_id）は、本実装では Pythonコード内で自動取得しており、 .envファイルに定義する必要はありません。

このように、自然文の指示はすべてChatGPTによって「インテント」に分類され、対応する関数に処理が振り分けられます。具体的な処理の実装は chatgpt_logic.py → task_utils.py に記述されています。

LINEから実行できる３つの処理

ユーザーがLINEで以下のようなメッセージを送るだけで、AI執事が自動的にタスク処理を実行します。
処理はすべてChatGPTによって自然文が解析され、該当のコマンドに分類されます。

「明日までに企画書を作成するタスクを追加して」→ 登録
「タスク一覧を見せて」→ 参照
「企画書のタスク削除して」→ 削除

※タスクの更新機能はありません。タスク名や期限を変更したい場合は、該当タスクを削除してから再登録してください。

使用ライブラリと認証設定

Google Tasks APIと連携するためには、APIの有効化と認証設定が必要です。この記事では、必要なライブラリや.envファイルに記述する項目について明確に解説します。

Google Tasks APIの有効化

Google Cloud Consoleにアクセスし、対象プロジェクトで「Tasks API」を有効にします。
サービスアカウントの作成とJSONファイルのダウンロードも忘れずに行ってください。

APIとサービス → ライブラリ → Tasks API を検索して有効化
サービスアカウントを作成 → ロールは「編集者」または「オーナー」
認証情報 → JSON形式で鍵を生成 → プロジェクトに保存

.env に記述するキー一覧

以下の環境変数を .env ファイルに記述することで、ChatGPT・Google Tasksの認証が可能になります。

変数名	用途
OPENAI_API_KEY	ChatGPTとの連携に使用するAPIキー
LINE_CHANNEL_ACCESS_TOKEN	LINE Messaging APIのアクセストークン（v3 SDK用）
LINE_CHANNEL_SECRET	LINEのチャネルシークレット（Webhook署名検証に使用）
GOOGLE_CALENDAR_ID	予定の登録・取得・削除に使用するGoogleカレンダーID
GOOGLE_TOKEN_JSON	Google API（Calendar / Tasks）の認証に使用するOAuthトークンのJSONファイルパス

Google APIの認証方式は2種類ある

Googleが提供するAPI（カレンダーやタスクなど）には、主に下記の2つの認証方式があります。どちらの方式を使うかによって、コードの書き方や許可の仕組みが大きく異なります。

✅ ２つの認証方式

OAuth認証
サービスアカウント認証

OAuth認証とは？（token.json方式）

ユーザーがGoogleにログインして、自分のアカウントのデータにアクセスしてよいかを明示的に許可する方式です。初回だけ手動でログイン・許可が必要ですが、以降は「token.json」というファイルを使って自動化できます。

ユーザー本人のカレンダーやタスクなど、個人データを扱う用途向け
初回だけログイン許可が必要
token.json が本人確認の鍵になる

サービスアカウント認証とは？（credentials.json方式）

Google Cloud Consoleで発行できる「機械専用のアカウント」です。人間のログインは不要で、バックエンドのプログラムやサーバー処理向けに利用されます。

人の手を介さずに自動で動作させたい処理に最適
credentials.json（秘密鍵ファイル）を使って認証
カレンダーなど一部のAPIでは使用可能

本AI執事で使用している認証方式

本AI執事では、OAuth認証（token.json方式）のみを使用しています。
サービスアカウント（credentials.json）は使用しません。

ここで注意すべき最大のポイントは「スコープ（SCOPES）の指定」です。
token.json を生成する際に、どのGoogle APIへのアクセスを許可するかを事前に指定する必要があります。以下のように記述します。

スコープの指定が重要

Google APIへのアクセス権限は「スコープ」で制御されており、token.json を生成する際に以下のように明示する必要があります：

SCOPES = [
"https://www.googleapis.com/auth/calendar",
"https://www.googleapis.com/auth/tasks"
]

上記のように、カレンダーとタスク両方のスコープを含めておけば、1つの token.json で両方のAPIにアクセス可能です。

逆に、片方だけで生成した場合は、あとから追加することができず、403エラーが発生します。

スコープを間違えた場合の対処法

古い token.json を削除（またはリネーム）
正しいスコープを含めたコードで認証を再実行
新しい token.json が自動生成される

本AI執事の構成における補足

本実装では、 GOOGLE_TOKEN_JSON のみを使用し、 GOOGLE_TASKLIST_ID は .env では定義しません
タスクリストIDはコード側で自動的に検出・使用されます
ユーザーのGoogleアカウントで一度だけ認証を行えば、自動的にタスクや予定が処理されます

token.json を取得するために必要な2つの設定

PythonからGoogleのカレンダーやタスクにアクセスするためには、token.json というファイルが必要です。
このファイルは、あなたがGoogleログインでアクセスを許可したことを証明する「本人認証済みの通行証」となります。

この token.json を取得するには、以下の2ステップが必要です。

OAuthクライアントの認証情報（client_secret.json）を作成する
Google Cloud Console 上で OAuth用の認証情報を発行し、 client_secret.json をダウンロードします。
このファイルは「このアプリが誰によって作られたか」「どの情報にアクセスするか」をGoogleに伝える役割を持ちます。

Pythonスクリプトでログインし、token.json を生成する
先ほどの client_secret.json を元に、あなた自身がGoogleログインしてアクセスを許可すると、token.json が自動生成されます。

この token.json が生成されることで、以降は毎回ログインすることなく、Pythonスクリプトからあなたの Googleカレンダーや Googleタスクに自動でアクセスできるようになります。

OAuthクライアントの認証情報（client_secret.json）を作成する

OAuth認証による token.json の取得を行う前に、Google Cloud Console で「OAuth同意画面」を設定する必要があります。以下の手順は、初期状態から完全に設定が完了するまでを1ステップずつ解説しています。

① Google Cloud Console にアクセス

はじめにGoogle Cloud Console を開きます。

② プロジェクトを選択（または作成）

右上の「プロジェクトを選択」をクリックし、対象のプロジェクトを選択または新規作成します。
この後の設定は、選択したプロジェクトに対して反映されます。

③ 「OAuth 同意画面を構成してください」と表示されたら「構成」ボタンを押す

初回設定時は、「Google Auth Platform はまだ構成されていません」と表示されます。
その場合は画面右下の「構成」または「開始」ボタンをクリックして、同意画面の設定を開始します。

④ アプリ情報を入力

アプリ名： 任意の名前（例：BePro AI執事）
ユーザーサポートメール： 自分のGoogleアカウント

入力が終わったら「次へ」ボタンをクリックします。

⑤ ユーザータイプを選択（外部）

「外部」を選択することで、Googleアカウントを持つすべてのテストユーザーがアプリを使用できるようになります。
この設定により、OAuth認証で自分のGoogleアカウントを使うことが可能になります。

選択が完了したら、「次へ」をクリックして、連絡先情報の入力に進んでください。

⑥ 連絡先情報入力

連絡先情報：同じメールアドレスを入力して「作成」ボタンをクリック

⑦ OAuth クライアントの作成

OAuth同意画面の構成ウィザードをすべて完了した後に「OAuthの概要」画面が表示されますが、まだクライアントID（client_secret.json）は未作成の状態です。

「OAuth クライアントを作成」ボタンを押してください。

⑧ OAuth クライアント ID の作成（アプリ種別を選択し「作成」）

アプリケーションの種類 → 「デスクトップアプリ」を選択
名前 → 任意（例：bepro-auth-client）
「作成」ボタンを押すと client_secret_XXXX.json がダウンロード可能になります

「作成」ボタンをクリックすrとOAuthクライアントが作成されます。

⑨ OAuthクライアントIDを作成し、JSONをダウンロード

「OAuthクライアントを作成しました」という確認画面が表示されます。
この画面で表示される「JSONをダウンロード」をクリックし、 client_secret_xxxx.json ファイルを必ず保存してください。

このファイルは、後で token.json を生成するPythonスクリプトで使用します。
万が一紛失すると、クライアントIDの作成からやり直す必要があります。

」→「認証情報」をクリックします。

④ 「認証情報を作成」→「OAuth クライアント ID」を選択

画面上部の「認証情報を作成」ボタンをクリックし、「OAuth クライアント ID」を選択します。

client_secret.json のリネーム手順

Google Cloud Console からダウンロードした認証ファイルは、以下のように長いファイル名になっています：

client_secret_724670778229-xxxxxx.apps.googleusercontent.com.json

しかし、Pythonスクリプトでは client_secret.json という名前で読み込むよう記述されています。
そのため、以下のようにファイル名をリネームしてください。

mv client_secret_724670778229-xxxxxx.apps.googleusercontent.com.json client_secret.json

Pythonスクリプトでログインし、token.json を生成する

Googleカレンダーやタスクなど、ユーザーの Google アカウントにあるデータへアプリがアクセスするには、OAuth同意画面を通じてアクセス権限の許可を得る必要があります。

OAuth同意画面とは？

OAuth同意画面は、アプリがどの情報にアクセスしようとしているのかをユーザーに明示し、「このアプリにGoogleカレンダーやタスクへのアクセスを許可しますか？」と確認するための画面です。

OAuth同意画面は、あなたが開発したアプリが、ユーザー本人の Google アカウントにあるデータ（カレンダーやタスクなど）へアクセスす

たとえば、あなたのアプリが「Googleカレンダーに予定を書き込む」処理を行う場合、Googleはその操作に対して明確な許可を要求します。
その許可を取るための画面が「OAuth同意画面」です。

開発初期の段階では「アプリをGoogleに公開する」前提にはなっていないため、OAuth同意画面は テストユーザーとして登録したアカウントのみ に表示されます。
このテストモードでは、アクセスできるユーザーを限定することで、Googleはアプリが悪用されるリスクを防いでいます。

つまり、OAuth同意画面とは──

アプリの名称・スコープ・連絡先などをユーザーに明示するための画面
「このアプリに、あなたのアカウント情報へのアクセスを許可しますか？」と尋ねる仕組み
Googleがユーザーの明示的な許可を記録し、安全性を担保するためのもの

この認証が完了したあとに取得されるのが token.json です。
このファイルは、「あなた自身のGoogleアカウントがこのアプリにアクセスを許可した」という証拠として機能し、以後の操作に使用されます。

アクセス権限付与手順

はじめにGoogle Cloud Console から、「APIとサービス」を選択します。

① Google Cloud Console にアクセス
はじめにGoogle Cloud Console を開きます。

② OAuth同意認証を選択
表示されたAPIとサービス画面左メニューから「OAuth同意認証」をします。

※OAuth同意認証が未認証の場合は、自動的にOAuthの概要画面へリダイレクトします。

③ テストユーザーの追加
OAuthの概要画面左メニューから「対象」を選択します。
テストユーザーの「add users」ボタンをクリックし、対象となるアクセスユーザーのメールアドレスを追加します。

スクリプトで認証を行い、token.json を生成する

Google Tasks API を使用するには、OAuth認証を通じて token.json を生成する必要があります。以下は VPS上での手順です。
今回は全環境（Mac, Windows, VPS含む）で使用できる「ブラウザ認証＋コード入力方式（ run_console()）」を使用します。

① client_secret.json を VPS上のプロジェクトフォルダに配置

事前にダウンロードした認証情報ファイル（例: client_secret_XXXX.json）を、VPS上のプロジェクトフォルダに client_secret.json という名前で配置してください。

配置先（プロジェクフォルダ）は以下を想定しています：

/home/bepro/projects/ai_butler/client_secret.json

② 認証スクリプト get_token.py を作成

以下の内容で get_token.py というファイルを作成します。

from google_auth_oauthlib.flow import InstalledAppFlow
import os

SCOPES = [
    "https://www.googleapis.com/auth/calendar",
    "https://www.googleapis.com/auth/tasks"
]

flow = InstalledAppFlow.from_client_secrets_file("client_secret.json", SCOPES)
creds = flow.run_console()

with open("token.json", "w") as token:
    token.write(creds.to_json())

必要に応じて以下もインストールしておいてください：

pip install --upgrade google-auth-oauthlib

③ ブラウザ認証後に token.json が生成される

VPSのターミナルで下記を実行します：

python get_token.py
以下のURLをローカルPCのブラウザで開いてください：
https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=724670778229-7pchv3hinla49buaa3ek0jns3auu945g.apps.googleusercontent.com&redirect_uri=urn%3Aietf%3Awg%3Aoauth%3A2.0%3Aoob&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fcalendar+https%3A%2F%2Fwww.googleapis.com%2Fauth%2Ftasks&state=WZ6ideGlBXvdXw9njqfkqkqCc4CFkI&prompt=consent&access_type=offline
認証後に表示されるコードをここに貼り付けてください:

「認証後に表示されるコードをここに貼り付けてください:」で一旦ユーザー待ち受け状態で停止します。

「以下のURLをローカルPCのブラウザで開いてください：」の要求に対し、出力されたURLをコピーし、ローカルPCのブラウザで開きます。

Googleのログイン画面が表示されます
「このアプリはGoogleで確認されていません」と出たら「続行」をクリック
アクセス許可を確認 → 「認可コード（数十文字）」が表示されます
認証コードをコピーします。
生成された認証コードの右にコピーアイコンが表示されるのでクリックして認証コードをコピーします。
コピーしたコードをVPS側のターミナルに貼り付けます

入力後、カレントディレクトリに token.json が生成されます。これで認証完了です。
プロジェクトフォルダ配下へ「token.json」が存在することを確認します。

(.venv) bepro@vm-09d41400-18:~/projects/ai_butler$ ls -l
-rw-rw-r-- 1 bepro bepro 1937 Apr 20 16:28 app.py
-rw-r--r-- 1 bepro bepro 407 Apr 23 11:28 client_secret.json
-rw-r--r-- 1 bepro bepro 2374 Apr 18 16:44 credentials.json
-rw-rw-r-- 1 bepro bepro 665 Apr 23 12:34 get_token.py
drwxrwxr-x 3 bepro bepro 4096 Apr 23 11:51 logic
-rw-rw-r-- 1 bepro bepro 1141 Apr 20 16:27 requirements.txt
-rw-rw-r-- 1 bepro bepro 746 Apr 23 13:27 token.json　👈

🔒 token.json の取り扱いについて

このファイルは認証済みユーザーの鍵情報を含みます
絶対に第三者と共有しないでください
.gitignore に追加して、Gitにコミットしないようにしてください

完成ソースと構造の全体像

このセクションでは、Googleタスク連携処理を行うAI執事Botにおける全体構造を整理します。ファイル単位での役割分担を明示し、読者が再利用や拡張を行いやすいよう構造の意図を伝えます。

下記は、今回構築した AI執事ボットにおける、Google カレンダーおよび Google タスク機能を実装した際のファイル構成です。

Flask アプリとして構成されており、ロジックは logic ディレクトリにモジュールとして分離されています。

ai_butler/
├── app.py                    Flaskアプリのメイン起動スクリプト
├── get_token.py              token.json を生成する認証スクリプト
├── client_secret.json        Google OAuth クライアント認証情報（ダウンロードファイル）
├── token.json                認証後に生成されるトークン（API呼び出しに使用）
├── requirements.txt          必要なライブラリをまとめたファイル
└── logic/                    アプリの主要ロジックを分離
    ├── chatgpt_logic.py      ChatGPTとの連携・自然文の解析と分類
    ├── calendar_utils.py     Googleカレンダーの予定登録・削除・取得処理
    └── task_utils.py         Googleタスクの登録・参照・削除処理

ai_butler/

├── app.py Flaskアプリのメイン起動スクリプト

├── get_token.py token.json を生成する認証スクリプト

├── client_secret.json Google OAuth クライアント認証情報（ダウンロードファイル）

├── token.json 認証後に生成されるトークン（API呼び出しに使用）

├── requirements.txt 必要なライブラリをまとめたファイル

└── logic/ アプリの主要ロジックを分離

├── chatgpt_logic.py ChatGPTとの連携・自然文の解析と分類

├── calendar_utils.py Googleカレンダーの予定登録・削除・取得処理

└── task_utils.py Googleタスクの登録・参照・削除処理

このように、トークン管理・外部APIとの連携・ユーザーの入力解析をそれぞれ分離することで、構造が明確になり、保守性の高い設計となっています。

ファイル構成と役割

今回の処理は3つの主要ファイルによって構成されており、それぞれ明確な役割を担っています。

ファイル名	主な役割
app.py	Flask本体。LINEのWebhookを受信し、chatgpt_logic.pyへ処理を委譲
chatgpt_logic.py	ユーザー入力から意図を分類し、task_utils.pyへ適切な処理を振り分け
task_utils.py	Google Tasks APIとのやり取りを一手に担い、登録・取得・削除・更新を実行

chatgpt_logic.py（既存回収）

import os
import json
from datetime import datetime
from openai import OpenAI
from dateutil.parser import parse
from logic.calendar_utils import (
    registerSchedule,
    getScheduleByOffset,
    deleteEvent,
    updateEvent
)
from logic.task_utils import (
    registerTask,
    listTasks,
    completeTask,
    deleteTask,
    listCompletedTasks,
    registerTaskWithDue,
    listTasksWithDue
)

# 🔍 ユーザーの発言から意図を判定（登録・更新・削除・予定確認など）
def classifyIntent(user_input):
    user_input = user_input.lower()

    if "削除" in user_input:
        return "delete"
    elif "更新" in user_input or "変更" in user_input:
        return "update"
    elif "完了済" in user_input or "完了した" in user_input:
        return "task_list_completed"
    elif "期限付き" in user_input or "締め切り" in user_input or "期日" in user_input:
        return "task_list_due"
    elif "入れて" in user_input or "登録" in user_input or "追加" in user_input:
        return "register"
    elif "明後日" in user_input and "予定" in user_input:
        return "schedule+2"
    elif "明日" in user_input and "予定" in user_input:
        return "schedule+1"
    elif "今日" in user_input and "予定" in user_input:
        return "schedule+0"
    elif "予定" in user_input or "スケジュール" in user_input:
        return "schedule+0"
    elif "天気" in user_input:
        return "weather"
    elif "疲れた" in user_input or "やる気" in user_input:
        return "mental"
    elif "タスク" in user_input or "やること" in user_input:
        if "一覧" in user_input or "確認" in user_input:
            return "task_list"
        elif "完了" in user_input:
            return "task_complete"
        elif "削除" in user_input:
            return "task_delete"
        else:
            return "task_register"
    else:
        return "general"
    
# 📤 ChatGPTを使って予定のタイトルと（必要なら）開始時刻を抽出する
def extractNewEventDetails(user_input, require_time=True):
    today = datetime.now().strftime("%Y-%m-%d")

    if require_time:
        system_content = (
            f"あなたは自然文から予定の日時とタイトルを抽出するアシスタントです。\n"
            f"今日の日付は {today} です。『明日』『明後日』なども正しく認識してください。\n"
            f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"
            f"{{\"title\": \"予定名\", \"start_time\": \"2025-04-30 15:00:00\"}}\n"
            f"※形式が正しくないと処理ができません。"
        )
    else:
        system_content = (
            f"あなたは自然文から予定のタイトルだけを抽出するアシスタントです。\n"
            f"今日の日付は {today} です。『明日』『明後日』なども正しく認識してください。\n"
            f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"
            f"{{\"title\": \"予定名\"}}\n"
            f"※形式が正しくないと処理ができません。"
        )

    messages = [
        {"role": "system", "content": system_content},
        {"role": "user", "content": user_input}
    ]

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages
    )
    content = response.choices[0].message.content
    print("📤 ChatGPTの返答（予定抽出）：", content)

    try:
        parsed = json.loads(content)
    except json.JSONDecodeError:
        print("❌ JSON解析失敗：ChatGPT応答が不正な形式")
        raise ValueError("ChatGPTの応答が正しい形式ではありません。")

    # タイトルの正規化処理（ゆらぎ防止）
    title = parsed.get("title", "").strip()
    for junk in [
        "の予定を変更", "の予定を削除", "の予定を追加", "の予定を登録",
        "を変更", "を削除", "を追加", "を登録",
        "の予定", "の予約", "予約"
    ]:
        title = title.replace(junk, "")
    title = title.strip()

    if require_time:
        start_time = parsed.get("start_time")
        return {"title": title, "start_time": start_time}
    else:
        return {"title": title}

# 📤 ChatGPTを使ってタスク名を抽出する（余計な語句は除去）
def extractTaskTitle(user_input):
    today = datetime.now().strftime("%Y-%m-%d")

    system_content = (
        f"あなたは自然文からタスク名を抽出するアシスタントです。\n"
        f"今日の日付は {today} です。『明日までにやること』などの文脈を正しく判断してください。\n"
        f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"
        f"{{\"title\": \"タスク名\"}}\n"
        f"※形式が正しくないと処理ができません。"
    )

    messages = [
        {"role": "system", "content": system_content},
        {"role": "user", "content": user_input}
    ]

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages
    )
    content = response.choices[0].message.content
    print("📤 ChatGPTの返答（タスク抽出）：", content)

    try:
        parsed = json.loads(content)
    except json.JSONDecodeError:
        print("❌ JSON解析失敗：ChatGPT応答が不正な形式")
        raise ValueError("ChatGPTの応答が正しい形式ではありません。")

    title = parsed.get("title", "").strip()

    # ✅ 正規化：削除・完了などの余計な語句を取り除く
    for junk in [
        "を削除", "を登録", "を追加", "を変更", "を完了にする", "を完了にして",
        "を完了", "を実行", "してください", "して"
    ]:
        title = title.replace(junk, "")

    return {"title": title.strip()}

# 🗓️ 予定登録用：ChatGPTで抽出 → 登録処理 → 成功メッセージ返却
def registerScheduleFromText(user_message, client):
    try:
        new_event = extractNewEventDetails(user_message, require_time=True)
        title = new_event["title"]
        start_time = datetime.strptime(new_event["start_time"], "%Y-%m-%d %H:%M:%S")

        # ✅ 結果メッセージをそのまま返す
        result = registerSchedule(title, start_time)
        return result

    except Exception as error:
        print("❌ 予定登録エラー：", error)
        return "日付とタイトルの解析に失敗しました。"

# 📥 タスクのタイトル＋期限（due）を抽出する
def extractTaskDetails(user_input):
    today = datetime.now().strftime("%Y-%m-%d")

    system_content = (
        f"あなたは自然文からタスク名と期限日を抽出するアシスタントです。\n"
        f"今日の日付は {today} です。『明日までに』などの文脈も正しく解釈してください。\n"
        f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"
        f"{{\"title\": \"タスク名\", \"due\": \"2025-05-10T00:00:00.000Z\"}}\n"
        f"期限がない場合は \"due\": null を設定してください。\n"
        f"※形式が正しくないと処理ができません。"
    )

    messages = [
        {"role": "system", "content": system_content},
        {"role": "user", "content": user_input}
    ]

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages
    )

    content = response.choices[0].message.content
    print("📥 ChatGPTの返答（タスク抽出＋期限）:", content)

    try:
        parsed = json.loads(content)
    except json.JSONDecodeError as e:
        print("❌ JSON解析エラー：", e)
        raise ValueError("ChatGPTの応答が正しいJSON形式ではありません。")

    # タイトル正規化（不要な文言の除去）
    title = parsed.get("title", "").strip()
    for junk in [
        "のタスクを追加", "のタスクを登録", "を追加", "を登録",
        "を完了", "を削除", "を更新", "タスク", "追加", "登録"
    ]:
        title = title.replace(junk, "")
    title = title.strip()

    # due の正規化
    due = parsed.get("due")
    if isinstance(due, str) and due.lower() == "null":
        due = None

    return {"title": title, "due": due}

# 🎯 メイン処理：ユーザーの意図に応じて処理分岐し、結果を返す
def askChatgpt(user_message):
    try:
        client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        intent = classifyIntent(user_message)
        print(f"🎯 intent 判定: {intent}")

        # 📆 特定日の予定確認（今日・明日・明後日など）
        if intent.startswith("schedule+"):
            day_offset = int(intent.split("+")[1])
            return getScheduleByOffset(day_offset)

        # 📝 新規登録（予定 or タスクの自動判別）
        elif intent == "register":
            try:
                task_info = extractTaskDetails(user_message)
                title = task_info.get("title")
                due = task_info.get("due")

                # ✅ 修正ポイント：括弧を見直し、title が None なら予定扱い
                if due or (title and ("タスク" in title or "やること" in title)):
                    if not title:
                        return "タスク名がうまく抽出できませんでした。"
                    if due:
                        return registerTaskWithDue(title, due)
                    else:
                        return registerTask(title)
                else:
                    return registerScheduleFromText(user_message, client)

            except Exception as e:
                print("❌ 登録エラー（タスク/予定）：", e)
                return "登録中にエラーが発生しました。"

        # 🗑️ 予定の削除（タイトル＋開始時刻を厳密に抽出して削除）
        elif intent == "delete":
            try:
                new_event = extractNewEventDetails(user_message, require_time=True)
                title = new_event.get("title")
                start_time_raw = new_event.get("start_time")

                if not title or not start_time_raw:
                    return "削除対象の予定が正しく抽出できませんでした。予定名と時間を明記してください。"

                start_time = datetime.strptime(start_time_raw, "%Y-%m-%d %H:%M:%S")
                return deleteEvent(title, start_time)

            except Exception as e:
                print("❌ 削除エラー：", e)
                return "削除中にエラーが発生しました。"

        # ♻️ 予定の更新（旧予定を削除 → 新予定を登録）
        elif intent == "update":
            try:
                new_event = extractNewEventDetails(user_message, require_time=True)
                title = new_event.get("title")
                start_time_raw = new_event.get("start_time")

                if not title or not start_time_raw:
                    return "更新対象の予定が正しく抽出できませんでした。予定名と時間を明記してください。"

                start_time = datetime.strptime(start_time_raw, "%Y-%m-%d %H:%M:%S")
                return updateEvent(title, {"title": title, "start_time": start_time})
            except Exception as e:
                print("❌ 更新エラー：", e)
                return "更新中にエラーが発生しました。"

        # ✅ タスク登録
        elif intent == "task_register":
            try:
                new_task = extractTaskTitle(user_message)
                title = new_task.get("title")
                if not title:
                    return "タスク名がうまく抽出できませんでした。"
                return registerTask(title)
            except Exception as e:
                print("❌ タスク登録エラー：", e)
                return "タスク登録中にエラーが発生しました。"

        # 📋 タスク一覧表示
        elif intent == "task_list":
            return listTasks()

        # ✅ タスク完了処理
        elif intent == "task_complete":
            try:
                new_task = extractTaskTitle(user_message)
                title = new_task.get("title")
                if not title:
                    return "完了させたいタスク名が見つかりませんでした。"
                return completeTask(title)
            except Exception as e:
                print("❌ タスク完了エラー：", e)
                return "タスク完了中にエラーが発生しました。"
            
        # 🗑️ タスク削除
        elif intent == "task_delete":
            try:
                new_task = extractTaskTitle(user_message)
                title = new_task.get("title")
                if not title:
                    return "削除したいタスク名が見つかりませんでした。"
                return deleteTask(title)
            except Exception as e:
                print("❌ タスク削除エラー：", e)
                return "タスク削除中にエラーが発生しました。"
            
        # 完了済みタスク一覧
        elif intent == "task_list_completed":
            return listCompletedTasks()

        # ✅ タスク登録（期限付きも含む）
        elif intent == "task_register":
            try:
                task_info = extractTaskDetails(user_message)
                title = task_info.get("title")
                due = task_info.get("due")

                if not title:
                    return "タスク名がうまく抽出できませんでした。"

                if due:
                    return registerTaskWithDue(title, due)
                else:
                    return registerTask(title)

            except Exception as e:
                print("❌ タスク登録エラー：", e)
                return "タスク登録中にエラーが発生しました。"
            
        # 📅 期限付きタスクの一覧表示
        elif intent == "task_list_due":
            return listTasksWithDue()

        # 🤖 雑談など（ChatGPTへそのまま転送）
        messages = [
            {"role": "system", "content": "あなたは親切で柔軟なAIアシスタントです。"},
            {"role": "user", "content": user_message}
        ]
        response = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages
        )
        return response.choices[0].message.content

    except Exception as error:
        print("❌ ChatGPT応答全体エラー：", error)
        return "AI応答中にエラーが発生しました。"

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

import os

import json

from datetime import datetime

from openai import OpenAI

from dateutil.parser import parse

from logic.calendar_utils import (

registerSchedule,

getScheduleByOffset,

deleteEvent,

updateEvent

)

from logic.task_utils import (

registerTask,

listTasks,

completeTask,

deleteTask,

listCompletedTasks,

registerTaskWithDue,

listTasksWithDue

)

# 🔍 ユーザーの発言から意図を判定（登録・更新・削除・予定確認など）

def classifyIntent(user_input):

user_input = user_input.lower()

if "削除" in user_input:

return "delete"

elif "更新" in user_input or "変更" in user_input:

return "update"

elif "完了済" in user_input or "完了した" in user_input:

return "task_list_completed"

elif "期限付き" in user_input or "締め切り" in user_input or "期日" in user_input:

return "task_list_due"

elif "入れて" in user_input or "登録" in user_input or "追加" in user_input:

return "register"

elif "明後日" in user_input and "予定" in user_input:

return "schedule+2"

elif "明日" in user_input and "予定" in user_input:

return "schedule+1"

elif "今日" in user_input and "予定" in user_input:

return "schedule+0"

elif "予定" in user_input or "スケジュール" in user_input:

return "schedule+0"

elif "天気" in user_input:

return "weather"

elif "疲れた" in user_input or "やる気" in user_input:

return "mental"

elif "タスク" in user_input or "やること" in user_input:

if "一覧" in user_input or "確認" in user_input:

return "task_list"

elif "完了" in user_input:

return "task_complete"

elif "削除" in user_input:

return "task_delete"

else:

return "task_register"

else:

return "general"

# 📤 ChatGPTを使って予定のタイトルと（必要なら）開始時刻を抽出する

def extractNewEventDetails(user_input, require_time=True):

today = datetime.now().strftime("%Y-%m-%d")

if require_time:

system_content = (

f"あなたは自然文から予定の日時とタイトルを抽出するアシスタントです。\n"

f"今日の日付は {today} です。『明日』『明後日』なども正しく認識してください。\n"

f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"

f"{{\"title\": \"予定名\", \"start_time\": \"2025-04-30 15:00:00\"}}\n"

f"※形式が正しくないと処理ができません。"

)

else:

system_content = (

f"あなたは自然文から予定のタイトルだけを抽出するアシスタントです。\n"

f"今日の日付は {today} です。『明日』『明後日』なども正しく認識してください。\n"

f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"

f"{{\"title\": \"予定名\"}}\n"

f"※形式が正しくないと処理ができません。"

)

messages = [

{"role": "system", "content": system_content},

{"role": "user", "content": user_input}

]

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = client.chat.completions.create(

model="gpt-3.5-turbo",

messages=messages

)

content = response.choices[0].message.content

print("📤 ChatGPTの返答（予定抽出）：", content)

try:

parsed = json.loads(content)

except json.JSONDecodeError:

print("❌ JSON解析失敗：ChatGPT応答が不正な形式")

raise ValueError("ChatGPTの応答が正しい形式ではありません。")

# タイトルの正規化処理（ゆらぎ防止）

title = parsed.get("title", "").strip()

for junk in [

"の予定を変更", "の予定を削除", "の予定を追加", "の予定を登録",

"を変更", "を削除", "を追加", "を登録",

"の予定", "の予約", "予約"

title = title.replace(junk, "")

title = title.strip()

if require_time:

start_time = parsed.get("start_time")

return {"title": title, "start_time": start_time}

else:

return {"title": title}

# 📤 ChatGPTを使ってタスク名を抽出する（余計な語句は除去）

def extractTaskTitle(user_input):

today = datetime.now().strftime("%Y-%m-%d")

system_content = (

f"あなたは自然文からタスク名を抽出するアシスタントです。\n"

f"今日の日付は {today} です。『明日までにやること』などの文脈を正しく判断してください。\n"

f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"

f"{{\"title\": \"タスク名\"}}\n"

f"※形式が正しくないと処理ができません。"

)

messages = [

{"role": "system", "content": system_content},

{"role": "user", "content": user_input}

]

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = client.chat.completions.create(

model="gpt-3.5-turbo",

messages=messages

)

content = response.choices[0].message.content

print("📤 ChatGPTの返答（タスク抽出）：", content)

try:

parsed = json.loads(content)

except json.JSONDecodeError:

print("❌ JSON解析失敗：ChatGPT応答が不正な形式")

raise ValueError("ChatGPTの応答が正しい形式ではありません。")

title = parsed.get("title", "").strip()

# ✅ 正規化：削除・完了などの余計な語句を取り除く

for junk in [

"を削除", "を登録", "を追加", "を変更", "を完了にする", "を完了にして",

"を完了", "を実行", "してください", "して"

title = title.replace(junk, "")

return {"title": title.strip()}

# 🗓️ 予定登録用：ChatGPTで抽出 → 登録処理 → 成功メッセージ返却

def registerScheduleFromText(user_message, client):

try:

new_event = extractNewEventDetails(user_message, require_time=True)

title = new_event["title"]

start_time = datetime.strptime(new_event["start_time"], "%Y-%m-%d %H:%M:%S")

# ✅ 結果メッセージをそのまま返す

result = registerSchedule(title, start_time)

return result

except Exception as error:

print("❌ 予定登録エラー：", error)

return "日付とタイトルの解析に失敗しました。"

# 📥 タスクのタイトル＋期限（due）を抽出する

def extractTaskDetails(user_input):

today = datetime.now().strftime("%Y-%m-%d")

system_content = (

f"あなたは自然文からタスク名と期限日を抽出するアシスタントです。\n"

f"今日の日付は {today} です。『明日までに』などの文脈も正しく解釈してください。\n"

f"絶対に自然文では返さず、以下の形式のJSONだけを返してください：\n"

f"{{\"title\": \"タスク名\", \"due\": \"2025-05-10T00:00:00.000Z\"}}\n"

f"期限がない場合は \"due\": null を設定してください。\n"

f"※形式が正しくないと処理ができません。"

)

messages = [

{"role": "system", "content": system_content},

{"role": "user", "content": user_input}

]

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = client.chat.completions.create(

model="gpt-3.5-turbo",

messages=messages

)

content = response.choices[0].message.content

print("📥 ChatGPTの返答（タスク抽出＋期限）:", content)

try:

parsed = json.loads(content)

except json.JSONDecodeError as e:

print("❌ JSON解析エラー：", e)

raise ValueError("ChatGPTの応答が正しいJSON形式ではありません。")

# タイトル正規化（不要な文言の除去）

title = parsed.get("title", "").strip()

for junk in [

"のタスクを追加", "のタスクを登録", "を追加", "を登録",

"を完了", "を削除", "を更新", "タスク", "追加", "登録"

title = title.replace(junk, "")

title = title.strip()

# due の正規化

due = parsed.get("due")

if isinstance(due, str) and due.lower() == "null":

due = None

return {"title": title, "due": due}

# 🎯 メイン処理：ユーザーの意図に応じて処理分岐し、結果を返す

def askChatgpt(user_message):

try:

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

intent = classifyIntent(user_message)

print(f"🎯 intent 判定: {intent}")

# 📆 特定日の予定確認（今日・明日・明後日など）

if intent.startswith("schedule+"):

day_offset = int(intent.split("+")[1])

return getScheduleByOffset(day_offset)

# 📝 新規登録（予定 or タスクの自動判別）

elif intent == "register":

try:

task_info = extractTaskDetails(user_message)

title = task_info.get("title")

due = task_info.get("due")

# ✅ 修正ポイント：括弧を見直し、title が None なら予定扱い

if due or (title and ("タスク" in title or "やること" in title)):

if not title:

return "タスク名がうまく抽出できませんでした。"

if due:

return registerTaskWithDue(title, due)

else:

return registerTask(title)

else:

return registerScheduleFromText(user_message, client)

except Exception as e:

print("❌ 登録エラー（タスク/予定）：", e)

return "登録中にエラーが発生しました。"

# 🗑️ 予定の削除（タイトル＋開始時刻を厳密に抽出して削除）

elif intent == "delete":

try:

new_event = extractNewEventDetails(user_message, require_time=True)

title = new_event.get("title")

start_time_raw = new_event.get("start_time")

if not title or not start_time_raw:

return "削除対象の予定が正しく抽出できませんでした。予定名と時間を明記してください。"

start_time = datetime.strptime(start_time_raw, "%Y-%m-%d %H:%M:%S")

return deleteEvent(title, start_time)

except Exception as e:

print("❌ 削除エラー：", e)

return "削除中にエラーが発生しました。"

# ♻️ 予定の更新（旧予定を削除 → 新予定を登録）

elif intent == "update":

try:

new_event = extractNewEventDetails(user_message, require_time=True)

title = new_event.get("title")

start_time_raw = new_event.get("start_time")

if not title or not start_time_raw:

return "更新対象の予定が正しく抽出できませんでした。予定名と時間を明記してください。"

start_time = datetime.strptime(start_time_raw, "%Y-%m-%d %H:%M:%S")

return updateEvent(title, {"title": title, "start_time": start_time})

except Exception as e:

print("❌ 更新エラー：", e)

return "更新中にエラーが発生しました。"

# ✅ タスク登録

elif intent == "task_register":

try:

new_task = extractTaskTitle(user_message)

title = new_task.get("title")

if not title:

return "タスク名がうまく抽出できませんでした。"

return registerTask(title)

except Exception as e:

print("❌ タスク登録エラー：", e)

return "タスク登録中にエラーが発生しました。"

# 📋 タスク一覧表示

elif intent == "task_list":

return listTasks()

# ✅ タスク完了処理

elif intent == "task_complete":

try:

new_task = extractTaskTitle(user_message)

title = new_task.get("title")

if not title:

return "完了させたいタスク名が見つかりませんでした。"

return completeTask(title)

except Exception as e:

print("❌ タスク完了エラー：", e)

return "タスク完了中にエラーが発生しました。"

# 🗑️ タスク削除

elif intent == "task_delete":

try:

new_task = extractTaskTitle(user_message)

title = new_task.get("title")

if not title:

return "削除したいタスク名が見つかりませんでした。"

return deleteTask(title)

except Exception as e:

print("❌ タスク削除エラー：", e)

return "タスク削除中にエラーが発生しました。"

# 完了済みタスク一覧

elif intent == "task_list_completed":

return listCompletedTasks()

# ✅ タスク登録（期限付きも含む）

elif intent == "task_register":

try:

task_info = extractTaskDetails(user_message)

title = task_info.get("title")

due = task_info.get("due")

if not title:

return "タスク名がうまく抽出できませんでした。"

if due:

return registerTaskWithDue(title, due)

else:

return registerTask(title)

except Exception as e:

print("❌ タスク登録エラー：", e)

return "タスク登録中にエラーが発生しました。"

# 📅 期限付きタスクの一覧表示

elif intent == "task_list_due":

return listTasksWithDue()

# 🤖 雑談など（ChatGPTへそのまま転送）

messages = [

{"role": "system", "content": "あなたは親切で柔軟なAIアシスタントです。"},

{"role": "user", "content": user_message}

]

response = client.chat.completions.create(

model="gpt-3.5-turbo",

messages=messages

)

return response.choices[0].message.content

except Exception as error:

print("❌ ChatGPT応答全体エラー：", error)

return "AI応答中にエラーが発生しました。"

chatgpt_logic.py における処理振り分けの追加

本記事では、ChatGPTの自然言語解析によって取得された「タスク名」と「期限情報」を Google Tasks に登録する機能を実装するために、 chatgpt_logic.py を改修しています。

具体的には、ChatGPT APIの応答結果として以下のような辞書データ（title と due を含む）が得られた場合、それを task_utils.py 側の処理に振り分ける構成としました。

{"title": "テストタスク9", "due": "2025-05-04T00:00:00.000Z"}

このような応答を受け取った際、 chatgpt_logic.py 内では次のような処理が実行されます。

from logic.task_utils import registerTaskWithDue

# ChatGPTの返答内容を解析し、タイトルと期限が含まれていれば登録処理へ
if isinstance(chatgpt_result, dict) and "title" in chatgpt_result and "due" in chatgpt_result:
    title = chatgpt_result["title"]
    due = chatgpt_result["due"]
    return registerTaskWithDue(title, due)

このように、ChatGPTから返された内容に応じて、必要な処理（ここでは期限付きタスク登録）を task_utils.py に振り分けています。

なお、今回の実装では intent（意図）に基づくルーティングや createTask / updateTask などの汎用関数は使用していません。あくまで「期限付きタスクの登録」に特化したシンプルなルート処理のみを実装しています。

task_utils.py完成ソース（新規追加）

import os<br />from google.oauth2.credentials import Credentials<br />from googleapiclient.discovery import build<br />from google.auth.transport.requests import Request<br />from dotenv import load_dotenv<br />from datetime import datetime<br /><br /># .envファイルから環境変数を読み込む<br />load_dotenv()<br /><br /># ✅ Google認証情報を取得（token.jsonベース）<br />def getCredentials():<br />    token_path = os.getenv("GOOGLE_TOKEN_JSON") or "/home/bepro/projects/ai_butler/token.json"<br />    if not token_path:<br />        raise ValueError("GOOGLE_TOKEN_JSON が未設定です")<br /><br />    # token.json を元に認証情報を構築<br />    creds = Credentials.from_authorized_user_file(<br />        token_path,<br />        scopes=["https://www.googleapis.com/auth/tasks"]<br />    )<br /><br />    # トークンが期限切れならリフレッシュ<br />    if creds and creds.expired and creds.refresh_token:<br />        creds.refresh(Request())<br />        with open(token_path, "w") as token_file:<br />            token_file.write(creds.to_json())<br /><br />    print("✅ GOOGLE_TOKEN_JSON:", token_path)<br />    return creds<br /><br /># ✅ 「geeksさんのリスト」のIDをリスト一覧から検索<br />def getDefaultTasklistId(service):<br />    results = service.tasklists().list().execute()<br />    for item in results.get("items", []):<br />        print("🧩 リスト検出:", item["title"], "→", item["id"])<br />        if item["title"].strip() == "geeksさんのリスト":<br />            return item["id"]<br />    raise ValueError("『geeksさんのリスト』が見つかりませんでした。")<br /><br /># ✅ タスク登録処理（タイトルのみ登録）<br />def registerTask(title):<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br /><br />        task = {<br />            "title": title<br />        }<br /><br />        # タスク登録実行<br />        result = service.tasks().insert(tasklist=tasklist_id, body=task).execute()<br />        print("✅ 登録タスク:", result.get("title"))<br />        return f"タスク『{title}』を登録しました。"<br /><br />    except Exception as e:<br />        print("❌ タスク登録エラー：", e)<br />        return "タスク登録中にエラーが発生しました。"<br /><br /># ✅ タスク一覧を取得し、整形して返す<br />def listTasks():<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id:", tasklist_id)<br /><br />        results = service.tasks().list(tasklist=tasklist_id, showCompleted=True).execute()<br />        tasks = results.get("items", [])<br />        print("📦 取得タスク数:", len(tasks))<br />        print("📦 取得タスク内容:", tasks)<br /><br />        if not tasks:<br />            return "現在、タスクは登録されていません。"<br /><br />        response = "現在のタスク一覧です：\n"<br />        for task in tasks:<br />            title = task.get("title", "").strip()<br />            status = task.get("status", "")<br />            due_str = task.get("due", None)<br /><br />            # ✅ フィルタ：完了・空タイトルを除外<br />            if not title or status != "needsAction":<br />                continue<br /><br />            # ✅ ゾンビ対策：過去すぎるタスクは除外（UI準拠）<br />            if due_str:<br />                try:<br />                    due = datetime.strptime(due_str[:10], "%Y-%m-%d")<br />                    if due.year < 2015:<br />                        continue<br />                except Exception as e:<br />                    print("⚠️ 日付パース失敗:", e)<br /><br />            response += f"・{title}\n"<br /><br />        if response.strip() == "現在のタスク一覧です：":<br />            return "現在、タイトルのあるタスクは登録されていません。"<br /><br />        return response<br /><br />    except Exception as e:<br />        print("❌ タスク一覧取得エラー：", e)<br />        return "タスクの一覧取得中にエラーが発生しました。"<br /><br /># ✅ 指定タイトルのタスクを「完了」に変更<br />def completeTask(title):<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        results = service.tasks().list(tasklist=tasklist_id).execute()<br />        tasks = results.get("items", [])<br /><br />        for task in tasks:<br />            task_title = task.get("title", "").strip()<br />            if task_title == title:<br />                # ステータス変更 → update<br />                task["status"] = "completed"<br />                service.tasks().update(<br />                    tasklist=tasklist_id,<br />                    task=task["id"],<br />                    body=task<br />                ).execute()<br />                print("✅ タスク完了:", title)<br />                return f"タスク『{title}』を完了にしました。"<br /><br />        return f"タスク『{title}』が見つかりませんでした。"<br /><br />    except Exception as e:<br />        print("❌ タスク完了エラー：", e)<br />        return "タスク完了中にエラーが発生しました。"<br /><br /># ✅ 指定タイトルのタスクを削除（先頭一致1件）<br />def deleteTask(target_title):<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        results = service.tasks().list(tasklist=tasklist_id, showCompleted=True).execute()<br />        tasks = results.get("items", [])<br /><br />        for task in tasks:<br />            title = task.get("title", "").strip()<br />            task_id = task.get("id")<br /><br />            if title == target_title:<br />                service.tasks().delete(tasklist=tasklist_id, task=task_id).execute()<br />                print(f"✅ タスク削除成功：{title}")<br />                return f"タスク『{title}』を削除しました。"<br /><br />        return f"指定されたタスク『{target_title}』は見つかりませんでした。"<br /><br />    except Exception as e:<br />        print("❌ タスク削除エラー：", e)<br />        return "タスク削除中にエラーが発生しました。"<br /><br /># ✅ 指定されたタイトルのタスクを完了状態にする<br />def completeTask(target_title):<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id:", tasklist_id)<br /><br />        # 未完了タスクのみ取得（完了済みは対象外）<br />        results = service.tasks().list(tasklist=tasklist_id, showCompleted=False).execute()<br />        tasks = results.get("items", [])<br /><br />        # タイトルが一致するタスクを探して完了に変更<br />        for task in tasks:<br />            title = task.get("title", "").strip()<br />            if title == target_title:<br />                task["status"] = "completed"<br />                service.tasks().update(tasklist=tasklist_id, task=task["id"], body=task).execute()<br />                print(f"✅ 完了マークを付けたタスク: {title}")<br />                return f"タスク『{title}』を完了にしました。"<br /><br />        return f"指定されたタスク『{target_title}』は見つかりませんでした。"<br /><br />    except Exception as e:<br />        print("❌ タスク完了エラー：", e)<br />        return "タスクの完了処理中にエラーが発生しました。"<br /><br /># ✅ 完了済みタスク一覧を返す関数<br />def listCompletedTasks():<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id（完了済み確認）:", tasklist_id)<br /><br />        results = service.tasks().list(<br />            tasklist=tasklist_id,<br />            showCompleted=True<br />        ).execute()<br /><br />        tasks = results.get("items", [])<br />        completed_tasks = [task for task in tasks if task.get("status") == "completed"]<br /><br />        print("📦 完了済みタスク数:", len(completed_tasks))<br />        print("📦 完了済みタスク内容:", completed_tasks)<br /><br />        if not completed_tasks:<br />            return "完了済みのタスクはありません。"<br /><br />        response = "✅ 完了済みタスク一覧です：\n"<br />        for task in completed_tasks:<br />            title = task.get("title", "").strip()<br />            if title:<br />                response += f"・{title}\n"<br /><br />        return response<br /><br />    except Exception as e:<br />        print("❌ 完了済みタスク取得エラー：", e)<br />        return "完了済みタスク一覧の取得中にエラーが発生しました。"<br /><br /># ✅ 完了済みタスク一覧を取得して整形して返す<br />def listCompletedTasks():<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id（完了）:", tasklist_id)<br /><br />        # 完了タスクのみ取得（showCompleted=True + statusで絞り込み）<br />        results = service.tasks().list(<br />            tasklist=tasklist_id,<br />            showCompleted=True,<br />            showHidden=True<br />        ).execute()<br /><br />        tasks = results.get("items", [])<br />        print("📦 取得タスク数（完了）:", len(tasks))<br /><br />        completed_tasks = [<br />            task for task in tasks if task.get("status") == "completed"<br />        ]<br /><br />        if not completed_tasks:<br />            return "現在、完了済みのタスクはありません。"<br /><br />        response = "✅ 完了済みのタスク一覧です：\n"<br />        for task in completed_tasks:<br />            title = task.get("title", "").strip()<br />            if title:<br />                response += f"・{title}\n"<br /><br />        return response<br /><br />    except Exception as e:<br />        print("❌ 完了済みタスク一覧取得エラー：", e)<br />        return "完了済みタスクの一覧取得中にエラーが発生しました。"<br /><br /># 📌 期限付きタスクを登録する<br />def registerTaskWithDue(title, due):<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br /><br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id:", tasklist_id)<br /><br />        task_body = {<br />            "title": title<br />        }<br /><br />        if due:<br />            # 文字列からdatetimeへ変換し、UTCのISO形式にする<br />            from datetime import datetime, timezone<br />            due_dt = datetime.strptime(due, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)<br />            task_body["due"] = due_dt.isoformat()<br /><br />        result = service.tasks().insert(tasklist=tasklist_id, body=task_body).execute()<br />        print("✅ 登録されたタスク:", result)<br />        return f"✅ タスク『{title}』を登録しました。期限: {due if due else '指定なし'}"<br /><br />    except Exception as e:<br />        print("❌ タスク登録（期限付き）エラー：", e)<br />        return "タスク登録中にエラーが発生しました。"<br /><br /># ✅ 期限付きのタスク（未完了）だけを抽出して一覧表示<br />def registerTaskWithDue(title, due_raw):<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id:", tasklist_id)<br /><br />        # ISO 8601形式かどうかを検証＆変換（例：2025-05-03 → 2025-05-03T00:00:00.000Z）<br />        try:<br />            if "T" not in due_raw:<br />                # もし日付だけだったら00:00:00にしてUTC形式に<br />                dt = datetime.strptime(due_raw, "%Y-%m-%d")<br />                due = dt.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z")<br />            else:<br />                # そのままISOとして使う（形式崩れていたら下でエラーになる）<br />                dt = datetime.fromisoformat(due_raw.replace("Z", "+00:00"))<br />                due = dt.isoformat().replace("+00:00", "Z")<br />        except Exception as e:<br />            print("❌ 期限形式エラー：", e)<br />            return "期限の形式が正しくありません（例：2025-05-03）。"<br /><br />        task_body = {<br />            "title": title,<br />            "due": due<br />        }<br /><br />        result = service.tasks().insert(tasklist=tasklist_id, body=task_body).execute()<br />        print("✅ 登録されたタスク:", result)<br />        <br />  # 🔧 ここでフォーマット変換（末尾の"Z"は除去）<br />        formatted_due = datetime.strptime(due.replace("Z", ""), "%Y-%m-%dT%H:%M:%S").strftime("%Y-%m-%d")<br /><br />        return f"✅ タスク『{title}』を登録しました（期限: {formatted_due}）"<br />        #return f"✅ タスク『{title}』を登録しました （期限: {due}）\n🔗 {result.get('webViewLink')}"<br /><br />    except Exception as e:<br />        print("❌ タスク登録（期限付き）エラー：", e)<br />        return "タスク登録中にエラーが発生しました。"<br /><br /># ✅ 期限付きタスク（未完了）を一覧で返す<br />def listTasksWithDue():<br />    try:<br />        creds = getCredentials()<br />        service = build("tasks", "v1", credentials=creds)<br />        tasklist_id = getDefaultTasklistId(service)<br />        print("📦 使用中のtasklist_id:", tasklist_id)<br /><br />        results = service.tasks().list(tasklist=tasklist_id, showCompleted=True).execute()<br />        tasks = results.get("items", [])<br /><br />        response = "期限付きタスク一覧：\n"<br />        for task in tasks:<br />            title = task.get("title", "").strip()<br />            due = task.get("due")<br />            status = task.get("status")<br /><br />            if due and status != "completed":<br />                due_date = due.split("T")[0]<br />                response += f"・{title}：期限 {due_date}\n"<br /><br />        if response.strip() == "期限付きタスク一覧：":<br />            return "現在、期限付きのタスクは登録されていません。"<br /><br />        return response<br /><br />    except Exception as e:<br />        print("❌ 期限付きタスク一覧取得エラー：", e)<br />        return "期限付きタスク一覧の取得中にエラーが発生しました。"

import os from google.oauth2.credentials import Credentials from googleapiclient.discovery import build from google.auth.transport.requests import Request from dotenv import load_dotenv from datetime import datetime # .envファイルから環境変数を読み込む load_dotenv() # ✅ Google認証情報を取得（token.jsonベース） def getCredentials(): token_path = os.getenv("GOOGLE_TOKEN_JSON") or "/home/bepro/projects/ai_butler/token.json" if not token_path: raise ValueError("GOOGLE_TOKEN_JSON が未設定です") # token.json を元に認証情報を構築 creds = Credentials.from_authorized_user_file( token_path, scopes=["https://www.googleapis.com/auth/tasks"] ) # トークンが期限切れならリフレッシュ if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) with open(token_path, "w") as token_file: token_file.write(creds.to_json()) print("✅ GOOGLE_TOKEN_JSON:", token_path) return creds # ✅ 「geeksさんのリスト」のIDをリスト一覧から検索 def getDefaultTasklistId(service): results = service.tasklists().list().execute() for item in results.get("items", []): print("🧩 リスト検出:", item["title"], "→", item["id"]) if item["title"].strip() == "geeksさんのリスト": return item["id"] raise ValueError("『geeksさんのリスト』が見つかりませんでした。") # ✅ タスク登録処理（タイトルのみ登録） def registerTask(title): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) task = { "title": title } # タスク登録実行 result = service.tasks().insert(tasklist=tasklist_id, body=task).execute() print("✅ 登録タスク:", result.get("title")) return f"タスク『{title}』を登録しました。" except Exception as e: print("❌ タスク登録エラー：", e) return "タスク登録中にエラーが発生しました。" # ✅ タスク一覧を取得し、整形して返す def listTasks(): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id:", tasklist_id) results = service.tasks().list(tasklist=tasklist_id, showCompleted=True).execute() tasks = results.get("items", []) print("📦 取得タスク数:", len(tasks)) print("📦 取得タスク内容:", tasks) if not tasks: return "現在、タスクは登録されていません。" response = "現在のタスク一覧です：\n" for task in tasks: title = task.get("title", "").strip() status = task.get("status", "") due_str = task.get("due", None) # ✅ フィルタ：完了・空タイトルを除外 if not title or status != "needsAction": continue # ✅ ゾンビ対策：過去すぎるタスクは除外（UI準拠） if due_str: try: due = datetime.strptime(due_str[:10], "%Y-%m-%d") if due.year < 2015: continue except Exception as e: print("⚠️ 日付パース失敗:", e) response += f"・{title}\n" if response.strip() == "現在のタスク一覧です：": return "現在、タイトルのあるタスクは登録されていません。" return response except Exception as e: print("❌ タスク一覧取得エラー：", e) return "タスクの一覧取得中にエラーが発生しました。" # ✅ 指定タイトルのタスクを「完了」に変更 def completeTask(title): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) results = service.tasks().list(tasklist=tasklist_id).execute() tasks = results.get("items", []) for task in tasks: task_title = task.get("title", "").strip() if task_title == title: # ステータス変更 → update task["status"] = "completed" service.tasks().update( tasklist=tasklist_id, task=task["id"], body=task ).execute() print("✅ タスク完了:", title) return f"タスク『{title}』を完了にしました。" return f"タスク『{title}』が見つかりませんでした。" except Exception as e: print("❌ タスク完了エラー：", e) return "タスク完了中にエラーが発生しました。" # ✅ 指定タイトルのタスクを削除（先頭一致1件） def deleteTask(target_title): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) results = service.tasks().list(tasklist=tasklist_id, showCompleted=True).execute() tasks = results.get("items", []) for task in tasks: title = task.get("title", "").strip() task_id = task.get("id") if title == target_title: service.tasks().delete(tasklist=tasklist_id, task=task_id).execute() print(f"✅ タスク削除成功：{title}") return f"タスク『{title}』を削除しました。" return f"指定されたタスク『{target_title}』は見つかりませんでした。" except Exception as e: print("❌ タスク削除エラー：", e) return "タスク削除中にエラーが発生しました。" # ✅ 指定されたタイトルのタスクを完了状態にする def completeTask(target_title): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id:", tasklist_id) # 未完了タスクのみ取得（完了済みは対象外） results = service.tasks().list(tasklist=tasklist_id, showCompleted=False).execute() tasks = results.get("items", []) # タイトルが一致するタスクを探して完了に変更 for task in tasks: title = task.get("title", "").strip() if title == target_title: task["status"] = "completed" service.tasks().update(tasklist=tasklist_id, task=task["id"], body=task).execute() print(f"✅ 完了マークを付けたタスク: {title}") return f"タスク『{title}』を完了にしました。" return f"指定されたタスク『{target_title}』は見つかりませんでした。" except Exception as e: print("❌ タスク完了エラー：", e) return "タスクの完了処理中にエラーが発生しました。" # ✅ 完了済みタスク一覧を返す関数 def listCompletedTasks(): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id（完了済み確認）:", tasklist_id) results = service.tasks().list( tasklist=tasklist_id, showCompleted=True ).execute() tasks = results.get("items", []) completed_tasks = [task for task in tasks if task.get("status") == "completed"] print("📦 完了済みタスク数:", len(completed_tasks)) print("📦 完了済みタスク内容:", completed_tasks) if not completed_tasks: return "完了済みのタスクはありません。" response = "✅ 完了済みタスク一覧です：\n" for task in completed_tasks: title = task.get("title", "").strip() if title: response += f"・{title}\n" return response except Exception as e: print("❌ 完了済みタスク取得エラー：", e) return "完了済みタスク一覧の取得中にエラーが発生しました。" # ✅ 完了済みタスク一覧を取得して整形して返す def listCompletedTasks(): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id（完了）:", tasklist_id) # 完了タスクのみ取得（showCompleted=True + statusで絞り込み） results = service.tasks().list( tasklist=tasklist_id, showCompleted=True, showHidden=True ).execute() tasks = results.get("items", []) print("📦 取得タスク数（完了）:", len(tasks)) completed_tasks = [ task for task in tasks if task.get("status") == "completed" ] if not completed_tasks: return "現在、完了済みのタスクはありません。" response = "✅ 完了済みのタスク一覧です：\n" for task in completed_tasks: title = task.get("title", "").strip() if title: response += f"・{title}\n" return response except Exception as e: print("❌ 完了済みタスク一覧取得エラー：", e) return "完了済みタスクの一覧取得中にエラーが発生しました。" # 📌 期限付きタスクを登録する def registerTaskWithDue(title, due): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id:", tasklist_id) task_body = { "title": title } if due: # 文字列からdatetimeへ変換し、UTCのISO形式にする from datetime import datetime, timezone due_dt = datetime.strptime(due, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc) task_body["due"] = due_dt.isoformat() result = service.tasks().insert(tasklist=tasklist_id, body=task_body).execute() print("✅ 登録されたタスク:", result) return f"✅ タスク『{title}』を登録しました。期限: {due if due else '指定なし'}" except Exception as e: print("❌ タスク登録（期限付き）エラー：", e) return "タスク登録中にエラーが発生しました。" # ✅ 期限付きのタスク（未完了）だけを抽出して一覧表示 def registerTaskWithDue(title, due_raw): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id:", tasklist_id) # ISO 8601形式かどうかを検証＆変換（例：2025-05-03 → 2025-05-03T00:00:00.000Z） try: if "T" not in due_raw: # もし日付だけだったら00:00:00にしてUTC形式に dt = datetime.strptime(due_raw, "%Y-%m-%d") due = dt.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z") else: # そのままISOとして使う（形式崩れていたら下でエラーになる） dt = datetime.fromisoformat(due_raw.replace("Z", "+00:00")) due = dt.isoformat().replace("+00:00", "Z") except Exception as e: print("❌ 期限形式エラー：", e) return "期限の形式が正しくありません（例：2025-05-03）。" task_body = { "title": title, "due": due } result = service.tasks().insert(tasklist=tasklist_id, body=task_body).execute() print("✅ 登録されたタスク:", result) # 🔧 ここでフォーマット変換（末尾の"Z"は除去） formatted_due = datetime.strptime(due.replace("Z", ""), "%Y-%m-%dT%H:%M:%S").strftime("%Y-%m-%d") return f"✅ タスク『{title}』を登録しました（期限: {formatted_due}）" #return f"✅ タスク『{title}』を登録しました（期限: {due}）\n🔗 {result.get('webViewLink')}" except Exception as e: print("❌ タスク登録（期限付き）エラー：", e) return "タスク登録中にエラーが発生しました。" # ✅ 期限付きタスク（未完了）を一覧で返す def listTasksWithDue(): try: creds = getCredentials() service = build("tasks", "v1", credentials=creds) tasklist_id = getDefaultTasklistId(service) print("📦 使用中のtasklist_id:", tasklist_id) results = service.tasks().list(tasklist=tasklist_id, showCompleted=True).execute() tasks = results.get("items", []) response = "期限付きタスク一覧：\n" for task in tasks: title = task.get("title", "").strip() due = task.get("due") status = task.get("status") if due and status != "completed": due_date = due.split("T")[0] response += f"・{title}：期限 {due_date}\n" if response.strip() == "期限付きタスク一覧：": return "現在、期限付きのタスクは登録されていません。" return response except Exception as e: print("❌ 期限付きタスク一覧取得エラー：", e) return "期限付きタスク一覧の取得中にエラーが発生しました。"

task_utils.py に追加した関数の解説

本記事では、ChatGPTの応答を元にGoogle Tasksへタスクを登録するため、以下の関数を task_utils.py に新規実装しました。

registerTaskWithDue(title, due)

ChatGPTから取得した「タイトル」と「期限」を受け取り、指定された GOOGLE_TASKLIST_ID に期限付きのタスクを登録します。

def registerTaskWithDue(title, due):
    service = getTasksService()
    task = {
        "title": title,
        "due": due
    }
    result = service.tasks().insert(tasklist=tasklist_id, body=task).execute()
    return f"✅ タスク『{title}』を登録しました（期限: {due}）"

getTasksService()

Google Tasks APIへアクセスするための認証済みサービスオブジェクトを生成します。
これは token.json を用いて、ユーザーのタスクリストへアクセスするための必須処理です。

def getTasksService():
    creds = getGoogleCredentials()
    service = build("tasks", "v1", credentials=creds)
    return service

getGoogleCredentials()

token.json を読み込み、Google API へのアクセスに必要なクレデンシャル（認証情報）を取得します。
認証済みであればそのまま返却し、期限切れなどがあれば例外を出力します。

def getGoogleCredentials():
    creds = None
    if os.path.exists("token.json"):
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
    return creds

補足：

今回の機能では「削除して再登録」という考え方を採用しているため、タスクの更新用関数は存在しません。名称や期限の変更が必要な場合は、タスクを一度削除してから再度登録し直すことで対応します。

※予定の「更新」機能は、既存の予定を一度削除してから新しい予定を再登録することで代替しています。専用の更新処理は設けていません。Google Tasksの更新APIは仕様的に扱いづらく、誤更新リスクがあるため、確実な挙動を重視して「削除＋再登録」で代替しています。

LINE→Flask→ChatGPT→Googleタスクの流れ

ユーザーがLINEに自然文でメッセージを送信すると、以下のような流れで処理が進みます。

LINEがWebhookとしてFlask（app.py）を呼び出す
app.pyがchatgpt_logic.pyの ask_chatgpt()を実行
chatgpt_logic.pyが意図を分類し、 task_utils.py内の関数を呼び出す
Google Tasks APIとやり取りし、処理結果をLINEへ返す

以下は簡略化した構成図です：

LINE → Flask（app.py）
　　└→ chatgpt_logic.py（意図分類）
　　　　　└→ task_utils.py（Googleタスク操作）
　　　　　　　└→ 結果をLINEへ返答

今回の構成が目指す理想的なUX

この構成では「思いついたらすぐLINEに送るだけ」でタスクの登録や確認が完了する体験を提供します。特に重要なのは以下の3点です：

ユーザーがコマンドを覚える必要がない（自然文でOK）
スピード感のあるやりとりが可能（即レスポンス）
メモ帳代わりに頭の中のToDoをそのまま書き出せる

このようなUXが、AI執事を単なるBotではなく「生活導線に溶け込むツール」に進化させる鍵となります。

ChatGPTのプロンプト設計と制御

タスク処理の自動化を実現するには、ChatGPTへの適切なプロンプト設計が不可欠です。自然文からタスクの構造を正しく生成させるための設計と処理の流れを以下にまとめます。

タスク処理のインテント分類と関数対応

ユーザー発言の例	インテント	🔧 処理関数	補足
タスク一覧を見せて	task_get	getTasks()	未完了タスクを表示（期限の有無問わず）
完了済タスクを見せて	task_get_completed	getCompletedTasks()	完了フラグのついたタスクのみ
期限付きタスクを見せて	task_get_due	getTasksWithDue()	未完了かつ期限あり
テストタスクを登録して	task_create	createTask()	期限なし登録
テストタスクを明日期限で登録して	task_create	createTask()	期限付き登録
テストタスクを削除して	task_delete	deleteTask()	タイトル一致で削除
テストタスクを完了にして	task_complete	completeTask()	完了フラグ付与

ChatGPTに指示するプロンプト例

自然文から「タイトル」と「期限」を抽出するには、ChatGPTへのプロンプトに出力形式を明示する必要があります。

role: system
content: あなたは自然文からタスク名と期日を抽出し、JSON形式で出力するアシスタントです。形式は {"title": "タスク内容", "due": "2025-05-10"} としてください。

role: user
content: 来週月曜に部長に報告書を提出する

想定される出力：

{"title": "部長に報告書を提出する", "due": "2025-05-13"}

ChatGPT応答のパース処理（Python側）

ChatGPTの出力はJSON形式の文字列として返されるため、Python側でパース処理が必要です。

import json
response_text = '{ "title": "部長に報告書を提出する", "due": "2025-05-13" }'
task = json.loads(response_text)
print(task["title"]) # → 部長に報告書を提出する

エラーとAPI制限の対処法

ChatGPT APIは、429エラー（レート制限）などの問題が発生する可能性があります。

try:
    response = client.chat.completions.create(...)
    return response.choices[0].message.content
except Exception as e:
    print("❌ ChatGPT応答エラー：", e)
    return "現在AI応答が制限されています。"

なお、OpenAIの利用プランによっては課金後も「Tier2条件（$50支払い＋7日経過）」が必要です。商用運用前には十分な検証が推奨されます。

今後の応用と拡張ポイント

AI執事は現時点で「タスクの基本的な参照・登録・削除・更新」に対応していますが、実用化を進めるにはさらに応用的な機能の追加が鍵となります。以下では、拡張に向けた展望を3つ紹介します。

チェックリスト形式への対応

現在は単一のタスクに対して1件ずつ処理を行っていますが、「複数タスクの一括登録」や「チェックボックス付きリスト形式」の扱いも将来的には想定できます。
この機能が実装されれば、買い物リストや作業手順リストなど、細かなToDo管理にも対応できるようになります。

LINE通知によるリマインダー

登録されたタスクの「期限が近づいたら通知する」機能も、ユーザー体験を大きく向上させる要素です。
GoogleタスクAPIでは期日（due）を設定できるため、それをもとに通知をLINEで送る機能との連携も可能です。

タスク完了報告機能との連携

ユーザーが「完了」と入力すれば、該当のタスクをチェック済みに更新するような双方向の処理も考えられます。
これにより、単なるメモ的な使い方から、「行動を促進する相棒」としての位置付けに進化できます。

まとめ：予定だけじゃないAI執事の真価

AI執事は、カレンダーとの連携によって予定の登録・削除・更新をこなすだけでなく、タスク管理にまで対応することで「行動の自動化と見える化」を実現します。

仕事のToDo、日常のやること、習慣づくりなど、日々の小さな行動も蓄積・管理できるAIは、まさに現代の「秘書」そのもの。
LINEという誰もが使うプラットフォームで、自分だけの執事を構築するという体験が、より多くの人に届くよう、今後も機能拡張を進めていきます。

よく読まれている記事

1: 【IT入門】シェルとシェルスクリプトの違いとは？シェルの種類について

IT入門シリーズ 🟢 STEP 1: ITの基礎を知る（ITとは何か？）　📌 IT初心者が最初に学ぶべき基本知識。ITの概念、ネットワーク、OS、クラウドの仕組みを学ぶ ...

2: 【Linuxの基礎知識】Linuxとは？基本概要と仕組みをわかりやすく解説！

「私たちが日々利用しているスマートフォンやインターネット、そしてスーパーコンピュータやクラウドサービス――これらの多くがLinuxの力で動いていることをご存じですか？無料で使えるだけでなく、高い柔軟性 ...

3: 【Shellの基礎知識】Shellスクリプト入門｜初心者が押さえる基本

この記事は、Linuxについて勉強している初心者の方向けに「Shellスクリプト」について解説します。最後まで読んで頂けましたら、Shellスクリプトはどのような役割を担っているのか？を理解出来るよう ...

-Pythonの基礎知識（実践編）

【Pythonの基礎知識】LINEとGoogleカレンダーで作るAI執事ボット

この世界の多くのプログラムは「会社の業務を回すため」に書かれています。でも、あなたの生活を便利にするプログラムは、誰も用意してくれません。 PCはあなたの道具であるはずなのに、なぜ他人の仕事ばかりを処理しているのでしょうか？この連載では、「生活を変えるプログラミング」をテーマに、 LINEとPythonを使って、自分だけの“AI執事ボット”を構築していきます。第1回は、「LINEに“今日の予定”と送ると、Googleカレンダーから予定を返してくれる」 ──そんな第一歩を、あなたと一緒に作ってみましょ ...

【Pythonの基礎知識】ChatGPT連携で賢く返信するAI執事ボットを作ろう（第2弾）

LINEに「疲れた」と話しかけたら、気の利いた言葉を返してくれる――そんな“気の利く執事ボット”を作ってみませんか？本記事では、前回作成したLINE×Googleカレンダー連携ボットに「ChatGPTによる自然な返答」を加え、より賢く・柔軟に返してくれるAI執事へと進化させていきます。予定の取得だけでなく、言葉の意図を読み取って対応を変える“文脈理解型ボット”の作り方を、実装付きで解説します。もくじ ChatGPT連携の前に知っておきたいことChatGPT APIはWeb版とは別物APIの利用には課金が ...

【Pythonの基礎知識】AI執事ボットのためのVPS環境構築マニュアル（LINE対応付き）

LINEで動くチャットボットを作ってみたい。そんな人の最初の関門になるのが、LINE Messaging APIの仕組みとWebhookの設定です。この記事では、VPS上でPythonとFlaskを使って、ユーザーがLINEで話しかけた内容に応答する「簡易AI執事ボット」を構築してみます。予定管理やタスク応答などは最低限にとどめつつ、LINEとサーバをつなぐ構成の理解に集中した内容になっています。「まずは動かす」「仕組みをつかむ」を目的とした入門編として、LINE Bot構築の第一歩にぜひ活用してくださ ...

【Pythonの基礎知識】AI執事で実現する！Googleカレンダーの予定削除・更新

AI執事ボットを活用して、Googleカレンダーの予定をPythonで効率的に管理する方法を解説します。本記事では、同一時間帯に予定が重複しないよう自動判定する仕組みや、既存予定の更新に対応する構成の考え方を中心に紹介します。AIを取り入れた実用的なスケジュール管理を学びましょう。もくじ予定削除・更新の概要予定削除と更新が必要な理由重複を防ぐ仕様と更新の考え方更新処理の実際完成ソースコードと説明AI執事ボットに予定の削除・更新機能を追加予定削除機能の概要削除処理のイメージコード予定更新機能の概要更新処 ...

【Pythonの基礎知識】LINEに送るだけで予定が登録されるAI執事

LINEに「明日の13時に歯医者」と送るだけで、Googleカレンダーにその予定が自動で登録される。そんな“自分専用のAI執事”を、Pythonと無料のAPIだけで実現するシリーズ第4弾です。これまでの構成では、「LINEから予定を確認する」まででしたが、今回はついに“予定を追加できる”ステージに進みます。自然な文章から日時と予定名を抽出し、Googleカレンダーに直接登録してくれるこの仕組みは、日々の小さなタスク管理を圧倒的に効率化します。スマホを開かずに予定を登録したい。 PCや手帳に向かう時 ...

【Pythonの基礎知識】AI執事で実現する！Googleカレンダーの予定削除・更新