ガントチャート自動生成アプリ

タスク管理って難しいですよね？やることは山積み、でも計画を立てるのが面倒…そんなあなたに朗報です！

本記事では、PythonのStreamlitを使って、OpenAI APIでタスクを自動生成し、ガントチャートを描画するアプリを作る方法を紹介します。

目標、現状、期日を記入し、タスク生成ボタンを押すとガントチャートが生成されます。

ガントチャートはタスクとサブタスクのつながりが分かるように色を揃えています。

実装

コードまとめ

import streamlit as st
import datetime
import json
import requests
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import pandas as pd
import re
import matplotlib.font_manager as fm
import seaborn as sns
import numpy as np
import io

# OpenAI API Key (環境変数または直接入力)
API_KEY = "OPENAI_API_KEY"

# OpenAI APIを使ってタスクを生成する関数
def clean_and_parse_json(content):
    """
    OpenAI API のレスポンスから JSON データを抽出し、不要な部分を削除する
    """
    try:
        # コードブロック (```json ... ```) を削除
        content = content.strip()
        content = re.sub(r"^```json", "", content)  # `json` コードブロックの開始を削除
        content = re.sub(r"```$", "", content)  # `json` コードブロックの終了を削除
        content = content.strip()

        # JSONデータをパース
        return json.loads(content)

    except json.JSONDecodeError as e:
        st.error(f"JSONデコードエラー: {e}")
        st.write("修正後のAPIレスポンス:")
        st.text(content)  # 修正後の JSON 文字列を出力
        return []


def generate_tasks(goal, current_state, deadline):
    """
    OpenAI API を使って目標達成のタスクスケジュールを取得（今日から開始）
    """
    today = datetime.date.today().strftime("%Y-%m-%d")  # 今日の日付を取得

    messages = [
        {"role": "system", "content": "あなたはユーザーの目標を達成するためのスケジュールプランナーです。"
                                      "ユーザーの目標と期日から、達成するためのタスクとサブタスクを提案してください。"
                                      "各タスクには開始日 (start_date) を今日の日付から設定し、"
                                      "サブタスクごとに適切な期日 (due_date) を設定してください。"
                                      "レスポンスは以下のJSONフォーマットにしてください。"},
        {"role": "user", "content": f"目標: {goal}\n"
                                    f"現状: {current_state}\n"
                                    f"期限: {deadline}\n"
                                    f"今日の日付: {today}\n\n"
                                    "以下のJSON形式でタスクスケジュールを作成してください。\n\n"
                                    "[\n"
                                    "  {\"task\": \"タスク名\", \"start_date\": \"YYYY-MM-DD\", \"due_date\": \"YYYY-MM-DD\", \"subtasks\": [\n"
                                    "    {\"subtask\": \"サブタスク名\", \"start_date\": \"YYYY-MM-DD\", \"due_date\": \"YYYY-MM-DD\"},\n"
                                    "    {\"subtask\": \"サブタスク名\", \"start_date\": \"YYYY-MM-DD\", \"due_date\": \"YYYY-MM-DD\"}\n"
                                    "  ]}\n"
                                    "]"}
    ]
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_KEY}"
    }
    payload = {
        "model": "gpt-4o",
        "messages": messages,
        "max_tokens": 3000,
        "temperature": 0.7
    }
    
    response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
    
    if response.status_code == 200:
        try:
            content = response.json()["choices"][0]["message"]["content"]
            
            # コードブロック (````json ... ````) を削除
            content_cleaned = re.sub(r"^```json\n|\n```$", "", content).strip()
            
            # JSONパース
            tasks = json.loads(content_cleaned)
            
            return tasks
        
        except (json.JSONDecodeError, KeyError) as e:
            st.error(f"APIレスポンスの解析中にエラーが発生しました: {e}")
            st.write("修正後のレスポンス内容:")
            st.text(content)  # デバッグ用にAPIレスポンスを表示
            return []
    else:
        st.error(f"APIリクエストエラー: {response.status_code}, {response.text}")
        return []


def plot_schedule(tasks):
    """
    Matplotlib の broken_barh を使用してタスクごとに色を変更し、サブタスクは薄く表示
    """
    # 日本語フォントを設定
    font_path = None
    jp_font_candidates = ["IPAexGothic", "Meiryo", "Noto Sans CJK JP", "Yu Gothic"]

    for font in fm.findSystemFonts():
        for jp_font in jp_font_candidates:
            if jp_font in font:
                font_path = font
                break
        if font_path:
            break

    if font_path:
        plt.rcParams['font.family'] = fm.FontProperties(fname=font_path).get_name()
    else:
        plt.rcParams['font.family'] = "IPAexGothic"

    # タスクデータの処理
    if not tasks:
        st.error("タスクデータが空です。APIから正しく取得されているか確認してください。")
        return

    data = []
    task_colors = sns.color_palette("husl", len(tasks))  # タスクごとに異なる色を生成
    task_color_map = {}  # タスクごとの色を保存
    
    for i, task in enumerate(tasks):
        start_date = datetime.datetime.strptime(task["start_date"], "%Y-%m-%d").date()
        task_due_date = datetime.datetime.strptime(task["due_date"], "%Y-%m-%d").date()
        duration = int((task_due_date - start_date).days)

        task_label = f"{task['task']}"  
        task_color_map[task["task"]] = task_colors[i]  # タスクごとの色をマップ

        data.append({"Task": task_label, "Start": start_date, "End": task_due_date, 
                     "Category": "Task", "Color": task_colors[i]})

        for subtask in task["subtasks"]:
            subtask_start_date = datetime.datetime.strptime(subtask["start_date"], "%Y-%m-%d").date()
            subtask_due_date = datetime.datetime.strptime(subtask["due_date"], "%Y-%m-%d").date()
            subtask_duration = int((subtask_due_date - subtask_start_date).days)

            subtask_label = f"{subtask['subtask']}"

            # サブタスクはタスクと同じ色で `alpha=0.5`
            subtask_color = (task_colors[i][0], task_colors[i][1], task_colors[i][2], 0.5)

            data.append({"Task": subtask_label, "Start": subtask_start_date, "End": subtask_due_date, 
                         "Category": "Subtask", "Color": subtask_color})

    # データフレームに変換
    df = pd.DataFrame(data)

    # **タスクの順序を API 出力順（元の順番）に揃える**
    y_pos = list(reversed(range(len(df))))  # **元の順番を維持しながら Y 軸を逆順に**

    # Matplotlib の broken_barh でガントチャートを作成
    fig, ax = plt.subplots(figsize=(15, len(df) * 0.5))  #　**タスク数に応じてサイズを動的調整**
    
    for index, row in df.iterrows():
        start_num = mdates.date2num(row["Start"])
        duration = (row["End"] - row["Start"]).days
        ax.broken_barh([(start_num, duration)], (y_pos[index] - 0.4, 0.8), facecolors=row["Color"])

        # 終了日だけを各バーの上に表示（開始日はラベルを付けない）
        ax.text(start_num + duration, y_pos[index], row["End"].strftime("%Y-%m-%d"),
                va='center', fontsize=10, color="black")

    # **X軸のラベルを開始日から終了日までで5つ均等に設定**
    min_date = df["Start"].min()
    max_date = df["End"].max()
    custom_ticks = np.linspace(mdates.date2num(min_date), mdates.date2num(max_date), num=5)  # 5つのラベルを均等に配置
    
    ax.set_xticks(custom_ticks)
    ax.set_xticklabels([mdates.num2date(tick).strftime("%Y-%m-%d") for tick in custom_ticks], rotation=45, ha="right")

    # **Y軸のラベルを `set_ylabel()` で設定**
    ax.set_yticks(y_pos, minor=False)
    ax.set_yticklabels(df["Task"], fontsize=12)  # **Y軸のラベルを直接設定**
    ax.set_ylim(-1, len(df) + 2)  # 📌 **Y軸の範囲を広げる**

    # グリッド線を追加
    ax.grid(True, linestyle="--", alpha=0.6)

    # 軸ラベルとタイトルを設定
    ax.set_title(f"{goal}", fontsize=20)

    # `tight_layout()` & `subplots_adjust` を適用
    plt.tight_layout()  # **レイアウトを自動調整**

    # 12. Streamlitに表示
    st.pyplot(fig)

    return fig



# Streamlitアプリの設定
st.title("ガントチャート自動生成アプリ")

# 入力フォーム
goal = st.text_input("目標を入力してください:")
current_state = st.text_area("現状を入力してください:")
deadline = st.date_input("期日を選択してください:", min_value=datetime.date.today())

# ボタンを押したらタスクを生成
if st.button("タスクを生成"):
    if not goal or not current_state:
        st.error("目標と現状を入力してください！")
    else:
        tasks = generate_tasks(goal, current_state, deadline)
        # st.write(tasks)

        if tasks:
            # **タスクをガントチャートで可視化し、Figure オブジェクトを取得**
            fig = plot_schedule(tasks)

            if fig:  # **エラーなく取得できた場合のみ保存**
                img_bytes = io.BytesIO()
                fig.savefig(img_bytes, format="png", bbox_inches="tight")
                img_bytes.seek(0)

                # **ガントチャートの画像をダウンロードできるようにする**
                st.download_button(
                    label="ガントチャートをダウンロード",
                    data=img_bytes,
                    file_name="gantt_chart.png",
                    mime="image/png"
                )

必要なライブラリのインポートとAPI設定

import streamlit as st
import datetime
import json
import requests
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import pandas as pd
import re
import matplotlib.font_manager as fm
import seaborn as sns
import numpy as np
import io

# OpenAI API Key (環境変数または直接入力)
API_KEY = "OPENAI_API_KEY"

OpenAI APIのレスポンスを解析する関数

def clean_and_parse_json(content):
    """
    OpenAI API のレスポンスから JSON データを抽出し、不要な部分を削除する
    """
    try:
        # コードブロック (```json ... ```) を削除
        content = content.strip()
        content = re.sub(r"^```json", "", content)  # `json` コードブロックの開始を削除
        content = re.sub(r"```$", "", content)  # `json` コードブロックの終了を削除
        content = content.strip()

        # JSONデータをパース
        return json.loads(content)

    except json.JSONDecodeError as e:
        st.error(f"JSONデコードエラー: {e}")
        st.write("修正後のAPIレスポンス:")
        st.text(content)  # 修正後の JSON 文字列を出力
        return []

OpenAI APIを使ってタスクスケジュールを生成する関数

def generate_tasks(goal, current_state, deadline):
    """
    OpenAI API を使って目標達成のタスクスケジュールを取得（今日から開始）
    """
    today = datetime.date.today().strftime("%Y-%m-%d")  # 今日の日付を取得

    messages = [
        {"role": "system", "content": "あなたはユーザーの目標を達成するためのスケジュールプランナーです。"
                                      "ユーザーの目標と期日から、達成するためのタスクとサブタスクを提案してください。"
                                      "各タスクには開始日 (start_date) を今日の日付から設定し、"
                                      "サブタスクごとに適切な期日 (due_date) を設定してください。"
                                      "レスポンスは以下のJSONフォーマットにしてください。"},
        {"role": "user", "content": f"目標: {goal}\n"
                                    f"現状: {current_state}\n"
                                    f"期限: {deadline}\n"
                                    f"今日の日付: {today}\n\n"
                                    "以下のJSON形式でタスクスケジュールを作成してください。\n\n"
                                    "[\n"
                                    "  {\"task\": \"タスク名\", \"start_date\": \"YYYY-MM-DD\", \"due_date\": \"YYYY-MM-DD\", \"subtasks\": [\n"
                                    "    {\"subtask\": \"サブタスク名\", \"start_date\": \"YYYY-MM-DD\", \"due_date\": \"YYYY-MM-DD\"},\n"
                                    "    {\"subtask\": \"サブタスク名\", \"start_date\": \"YYYY-MM-DD\", \"due_date\": \"YYYY-MM-DD\"}\n"
                                    "  ]}\n"
                                    "]"}
    ]
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_KEY}"
    }
    payload = {
        "model": "gpt-4o",
        "messages": messages,
        "max_tokens": 3000,
        "temperature": 0.7
    }
    
    response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
    
    if response.status_code == 200:
        try:
            content = response.json()["choices"][0]["message"]["content"]
            
            # コードブロック (````json ... ````) を削除
            content_cleaned = re.sub(r"^```json\n|\n```$", "", content).strip()
            
            # JSONパース
            tasks = json.loads(content_cleaned)
            
            return tasks
        
        except (json.JSONDecodeError, KeyError) as e:
            st.error(f"APIレスポンスの解析中にエラーが発生しました: {e}")
            st.write("修正後のレスポンス内容:")
            st.text(content)
            return []
    else:
        st.error(f"APIリクエストエラー: {response.status_code}, {response.text}")
        return []

ガントチャートの描画

def plot_schedule(tasks):
    """
    Matplotlib の broken_barh を使用してタスクごとに色を変更し、サブタスクは薄く表示
    """
    ...
    # (省略: Matplotlib を使ったガントチャート作成処理)
    ...

# Streamlitアプリの設定
st.title("ガントチャート自動生成アプリ")

# 入力フォーム
goal = st.text_input("目標を入力してください:")
current_state = st.text_area("現状を入力してください:")
deadline = st.date_input("期日を選択してください:", min_value=datetime.date.today())

# ボタンを押したらタスクを生成
if st.button("タスクを生成"):
    if not goal or not current_state:
        st.error("目標と現状を入力してください！")
    else:
        tasks = generate_tasks(goal, current_state, deadline)

        if tasks:
            fig = plot_schedule(tasks)

            if fig:  # エラーなく取得できた場合のみ保存
                img_bytes = io.BytesIO()
                fig.savefig(img_bytes, format="png", bbox_inches="tight")
                img_bytes.seek(0)

                # ガントチャートの画像をダウンロードできるようにする
                st.download_button(
                    label="ガントチャートをダウンロード",
                    data=img_bytes,
                    file_name="gantt_chart.png",
                    mime="image/png"
                )

まとめ

今回生成したタスクは扱いやすいようにJSON形式にしました。そのため、Notionなどと連携にして、自動的にNotionにタスクが振られるようにできればおもしろいかなと思っています。