Task_Progress_Heatmap.py

# app.py
# ---------------------------------------------------------
# ToDo進捗ヒートマップ（編集・削除対応／日本語フォント＆0-10固定）
# - 編集／削除機能つき（保存後は自動リロード）
# - 背景なし・文字色自動切替（黒/白）
# - 目に優しい配色（YlGnBu）／カラーバー 0〜10（1刻み）
# - 週次レポート（MarkdownDL）／メンバー別ヒートマップ
# ---------------------------------------------------------
import os
import math
import calendar
from datetime import date, timedelta
from typing import Dict, List, Tuple

import numpy as np
import pandas as pd
import matplotlib
from matplotlib import font_manager
import matplotlib.pyplot as plt
import streamlit as st

# ====== 日本語フォント ======
def set_japanese_font():
    candidates = [
        "Yu Gothic UI", "Yu Gothic", "Meiryo", "MS Gothic",
        "Noto Sans CJK JP", "Noto Sans JP", "IPAGothic",
    ]
    available = {f.name for f in font_manager.fontManager.ttflist}
    for name in candidates:
        if name in available:
            matplotlib.rcParams["font.family"] = name
            matplotlib.rcParams["axes.unicode_minus"] = False
            return name
    return None
_ = set_japanese_font()

# ====== 基本設定 ======
st.set_page_config(page_title="ToDo進捗ヒートマップ", page_icon="📊", layout="wide")
CSV_PATH = "tasks.csv"
REQUIRED_COLS = ["date", "task", "effort", "project", "member", "done"]
PLACEHOLDER = "（未設定）"

# ====== データ ======
def _coerce_df(df: pd.DataFrame) -> pd.DataFrame:
    for c in REQUIRED_COLS:
        if c not in df.columns:
            df[c] = np.nan
    df["date"] = pd.to_datetime(df["date"], errors="coerce").dt.date
    df["task"] = df["task"].astype(str).fillna("")
    df["project"] = df["project"].astype(str).fillna("")
    df["member"] = df["member"].astype(str).fillna("")
    df["effort"] = pd.to_numeric(df["effort"], errors="coerce").fillna(0).astype(int).clip(lower=0)
    df["done"] = df["done"].map(lambda x: str(x).lower() in ["true", "1", "t", "y"])
    df = df.dropna(subset=["date"])
    return df[REQUIRED_COLS].copy()

@st.cache_data
def load_data() -> pd.DataFrame:
    if os.path.exists(CSV_PATH):
        try:
            return _coerce_df(pd.read_csv(CSV_PATH))
        except Exception:
            return pd.DataFrame(columns=REQUIRED_COLS)
    return pd.DataFrame(columns=REQUIRED_COLS)

def save_row(d: date, t: str, e: int, project: str, member: str, done: bool):
    df = load_data().copy()
    new = pd.DataFrame([{
        "date": d, "task": t.strip(), "effort": int(e),
        "project": project.strip(), "member": member.strip(), "done": bool(done),
    }])
    df = pd.concat([df, new], ignore_index=True)
    df.to_csv(CSV_PATH, index=False, encoding="utf-8")
    st.cache_data.clear()

def sanitize_for_display(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty: return df
    df = df.copy()
    for col in ["task", "project", "member"]:
        df[col] = df[col].replace("", PLACEHOLDER).fillna(PLACEHOLDER)
    return df

# ====== カレンダー行列 ======
def month_matrix(year: int, month: int, values_by_day: Dict[int, int]) -> Tuple[np.ndarray, List[List[int]]]:
    cal = calendar.Calendar(firstweekday=6)  # 日曜始まり
    weeks = cal.monthdayscalendar(year, month)  # 0は当月外
    mat = []
    for w in weeks:
        row = [values_by_day.get(d, 0) if d != 0 else np.nan for d in w]
        mat.append(row)
    return np.array(mat), weeks

# ====== ヒートマップ描画 ======
def plot_heatmap(mat: np.ndarray, weeks: List[List[int]], values_by_day: Dict[int, int], title: str):
    fig, ax = plt.subplots(figsize=(6.8, 4.8))
    im = ax.imshow(
        np.ma.masked_invalid(mat),
        aspect="auto",
        cmap=plt.cm.YlGnBu,
        vmin=0, vmax=10,  # カラースケール固定
    )

    ax.set_xticks(range(7))
    ax.set_xticklabels(["日", "月", "火", "水", "木", "金", "土"])
    ax.set_yticks(range(len(weeks)))
    ax.set_yticklabels([f"{i+1}週" for i in range(len(weeks))])

    # 背景（値）に応じて文字色を自動切替：5以下=黒、6以上=白
    for i, week in enumerate(weeks):
        for j, day in enumerate(week):
            if day == 0:
                continue
            val = values_by_day.get(day, 0)
            txt_color = "white" if val > 5 else "black"
            ax.text(j, i, f"{day}\n{val}", ha="center", va="center", fontsize=10, color=txt_color)

    ax.set_title(title)
    ax.set_xlabel("曜日")
    ax.set_ylabel("週")

    # カラーバー：0〜10（1刻み）
    cbar = plt.colorbar(im, ax=ax, fraction=0.046, pad=0.04, ticks=range(0, 11))
    cbar.set_label("進捗ポイント", rotation=270, labelpad=15)

    st.pyplot(fig)

# ====== 週次レポート ======
def week_range(any_day: date):
    monday = any_day - timedelta(days=any_day.weekday())  # 月
    sunday = monday + timedelta(days=6)                   # 日
    return monday, sunday

def build_weekly_report_markdown(df_week: pd.DataFrame, wk_from: date, wk_to: date) -> str:
    df_week = sanitize_for_display(df_week)
    total_tasks = len(df_week)
    total_effort = int(df_week["effort"].sum()) if not df_week.empty else 0
    days = (wk_to - wk_from).days + 1
    avg_per_day = round(total_effort / days, 1) if days > 0 else 0.0
    done_count = int(df_week["done"].sum()) if not df_week.empty else 0
    comp_rate = (done_count / total_tasks * 100) if total_tasks > 0 else 0.0

    def topn(series: pd.Series, n=5): return [] if series.empty else series.value_counts().head(n).items()
    def topn_effort(df: pd.DataFrame, key: str, n=5):
        if df.empty: return []
        s = df.groupby(key)["effort"].sum().sort_values(ascending=False).head(n)
        return s.items()

    sections = [
        ("主要タスク（出現回数 上位）", topn(df_week["task"]), False),
        ("工数（effort）上位タスク", topn_effort(df_week, "task"), True),
        ("案件（出現回数 上位）", topn(df_week["project"]), False),
        ("工数（effort）上位案件", topn_effort(df_week, "project"), True),
        ("メンバー（出現回数 上位）", topn(df_week["member"]), False),
        ("工数（effort）上位メンバー", topn_effort(df_week, "member"), True),
    ]

    md = []
    md.append(f"# 週次レポート（{wk_from:%Y/%m/%d}〜{wk_to:%Y/%m/%d}）\n")
    md += [
        f"- 登録タスク数：**{total_tasks}**",
        f"- 合計ポイント：**{total_effort}**",
        f"- 1日平均ポイント：**{avg_per_day}**",
        f"- 完了数：**{done_count}**（完了率 **{comp_rate:.1f}%**）\n",
    ]
    for title, items, is_effort in sections:
        md.append(f"## {title}")
        if items:
            for k, v in items:
                md.append(f"- {k}：{int(v)}P" if is_effort else f"- {k}：{v}件")
        else:
            md.append("- なし")
        md.append("")
    # 稼働ゼロ日
    day_set = set(df_week["date"].unique().tolist())
    zeros, d = [], wk_from
    while d <= wk_to:
        if d not in day_set: zeros.append(d.strftime("%m/%d"))
        d += timedelta(days=1)
    md.append("## 稼働ゼロ日"); md.append("- " + ", ".join(zeros) if zeros else "- なし"); md.append("")
    return "\n".join(md)

# ====== 入力フォーム ======
st.title("📊 ToDo進捗ヒートマップ")
with st.expander("📝 データ入力（タスク追加）", expanded=True):
    c1, c2, c3 = st.columns([2, 1, 1])
    task = c1.text_input("タスク名", placeholder="例：Java学習、レビュー、バグ修正")
    d = c2.date_input("日付", value=date.today())
    effort = c3.number_input("進捗ポイント", min_value=0, max_value=24, value=1, step=1)
    c4, c5, c6 = st.columns([1.5, 1.5, 1])
    project = c4.text_input("案件名 / クライアント", placeholder="例：A社Web改修")
    member = c5.text_input("担当者", placeholder="例：山中")
    done = c6.checkbox("完了")
    if st.button("➕ 追加する", use_container_width=True):
        if not task.strip(): st.warning("タスク名を入力してください。")
        else:
            save_row(d, task, int(effort), project, member, done)
            st.success("追加しました！")

df = load_data()

# ====== サイドバー：フィルタ ======
with st.sidebar:
    st.header("🔎 フィルタ")
    today = date.today()
    year = st.number_input("年", min_value=2000, max_value=2100, value=today.year, step=1)
    month = st.number_input("月", min_value=1, max_value=12, value=today.month, step=1)
    projects = sorted([p for p in df["project"].dropna().unique().tolist() if p])
    members = sorted([m for m in df["member"].dropna().unique().tolist() if m]) if not df.empty else []
    selected_projects = st.multiselect("案件", options=projects, default=projects)
    selected_members = st.multiselect("メンバー", options=members, default=members)
    status = st.selectbox("ステータス", ["すべて", "未完のみ", "完了のみ"], index=0)

def apply_filters(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty: return df
    m = np.ones(len(df), dtype=bool)
    m &= [(d.year == int(year) and d.month == int(month)) for d in df["date"]]
    if selected_projects: m &= df["project"].isin(selected_projects)
    if selected_members: m &= df["member"].isin(selected_members)
    if status == "未完のみ": m &= ~df["done"]
    elif status == "完了のみ": m &= df["done"]
    return df.loc[m].copy()

df_month = apply_filters(df)

# ====== サマリー ======
st.subheader(f"📅 {int(year)}年{int(month)}月 のサマリー")
cA, cB, cC, cD = st.columns(4)
total_tasks = len(df_month)
done_count = int(df_month["done"].sum()) if not df_month.empty else 0
total_effort = int(df_month["effort"].sum()) if not df_month.empty else 0
completion_rate = (done_count / total_tasks) if total_tasks > 0 else 0.0
cA.metric("登録タスク数", f"{total_tasks}")
cB.metric("完了タスク数", f"{done_count}")
cC.metric("完了率", f"{completion_rate*100:.1f}%")
cD.metric("合計ポイント", f"{total_effort}")
st.progress(min(completion_rate, 1.0))
st.divider()

# ====== タブ：ヒートマップ / 週次レポート ======
tab1, tab2 = st.tabs(["🗺️ ヒートマップ", "🗒️ 週次レポート"])

with tab1:
    # チーム合計ヒートマップ
    values_by_day = {}
    if not df_month.empty:
        tmp = df_month.copy()
        tmp["day"] = [d.day for d in tmp["date"]]
        values_by_day = tmp.groupby("day")["effort"].sum().to_dict()
    mat, weeks = month_matrix(int(year), int(month), values_by_day)
    st.subheader("チーム合計ヒートマップ")
    plot_heatmap(mat, weeks, values_by_day, title="日別ポイント合計")

    # メンバー別ヒートマップ
    if selected_members or members:
        st.subheader("メンバー別ヒートマップ")
        show_members = selected_members if selected_members else members
        n = len(show_members); cols_per_row = 2; rows = math.ceil(n / cols_per_row)
        i = 0
        for _ in range(rows):
            cols = st.columns(cols_per_row)
            for c in range(cols_per_row):
                if i >= n: break
                mname = show_members[i]
                with cols[c]:
                    df_m = df_month[df_month["member"] == mname].copy()
                    vbd = {}
                    if not df_m.empty:
                        df_m["day"] = [d.day for d in df_m["date"]]
                        vbd = df_m.groupby("day")["effort"].sum().to_dict()
                    mat_m, weeks_m = month_matrix(int(year), int(month), vbd)
                    plot_heatmap(mat_m, weeks_m, vbd, title=f"{mname} の日別ポイント")
                i += 1

with tab2:
    st.subheader("週次レポート生成（Markdown）")
    base_day = st.date_input("週の基準日", value=date.today())
    wk_from, wk_to = week_range(base_day)
    df_week = df[(df["date"] >= wk_from) & (df["date"] <= wk_to)].copy()
    md = build_weekly_report_markdown(df_week, wk_from, wk_to)
    st.markdown(md)
    st.download_button(
        "📝 週次レポートをMarkdownでダウンロード",
        data=md.encode("utf-8"),
        file_name=f"weekly_report_{wk_from}_{wk_to}.md",
        mime="text/markdown",
        use_container_width=True,
    )

st.divider()

# ====== 編集・削除テーブル ======
st.subheader("📚 全体明細（編集・削除）")

orig_df = load_data()
edit_df = orig_df.copy().reset_index().rename(columns={"index": "ID"})
for col in ["task", "project", "member"]:
    edit_df[col] = edit_df[col].fillna("")
edit_df["削除"] = False

edited = st.data_editor(
    edit_df,
    use_container_width=True,
    num_rows="dynamic",
    hide_index=True,
    column_config={
        "ID": st.column_config.NumberColumn(disabled=True),
        "date": st.column_config.DateColumn("date", help="YYYY-MM-DD"),
        "task": st.column_config.TextColumn("task"),
        "effort": st.column_config.NumberColumn("effort", min_value=0, max_value=100, step=1),
        "project": st.column_config.TextColumn("project"),
        "member": st.column_config.TextColumn("member"),
        "done": st.column_config.CheckboxColumn("done", default=False),
        "削除": st.column_config.CheckboxColumn("削除", default=False),
    },
    key="editor",
)

left, right = st.columns([1, 3])
with left:
    if st.button("💾 変更を保存", use_container_width=True):
        try:
            df2 = edited.copy()
            if "削除" not in df2.columns: df2["削除"] = False
            df2["削除"] = df2["削除"].fillna(False).astype(bool)
            df2["done"] = df2["done"].fillna(False).astype(bool)

            # 追加テンプレの空行を除外
            is_empty = (
                df2["task"].fillna("").eq("") &
                df2["project"].fillna("").eq("") &
                df2["member"].fillna("").eq("") &
                pd.to_numeric(df2["effort"], errors="coerce").fillna(0).eq(0) &
                df2["done"].eq(False) &
                pd.to_datetime(df2["date"], errors="coerce").isna()
            )
            df2 = df2.loc[~is_empty].copy()

            # 削除チェックがついた行を落とす
            keep_df = df2.loc[~df2["削除"]].copy()

            # 型整形して保存
            keep_df["date"] = pd.to_datetime(keep_df["date"], errors="coerce")
            keep_df = keep_df.dropna(subset=["date"])
            keep_df["date"] = keep_df["date"].dt.date
            for col in ["task", "project", "member"]:
                keep_df[col] = keep_df[col].fillna("").astype(str)
            keep_df["effort"] = pd.to_numeric(keep_df["effort"], errors="coerce").fillna(0).astype(int).clip(0, 100)
            keep_df["done"] = keep_df["done"].fillna(False).astype(bool)

            out_df = keep_df[["date", "task", "effort", "project", "member", "done"]]
            out_df.to_csv(CSV_PATH, index=False, encoding="utf-8")

            st.success("変更を保存しました！")
            st.cache_data.clear()
            st.rerun()
        except Exception as e:
            st.error(f"保存時にエラーが発生しました：{e}")

with right:
    st.caption("※ 表の値を直接編集できます。『削除』にチェックをつけた行は保存時に削除されます。")

# 確認用表示
st.dataframe(
    sanitize_for_display(load_data()).sort_values(["date", "member", "project", "task"]).reset_index(drop=True),
    use_container_width=True,
)

# ====== CSVエクスポート ======
st.download_button(
    "📥 tasks.csv をダウンロード",
    data=load_data().to_csv(index=False).encode("utf-8"),
    file_name="tasks.csv",
    mime="text/csv",
    use_container_width=True,
)