# app.py
import streamlit as st
import pandas as pd
import networkx as nx
from pyvis.network import Network
from io import BytesIO
import tempfile
import os
from streamlit_sortables import sort_items
import hashlib
# =========================
# 기본 설정 & 전역 스타일
# =========================
st.set_page_config(page_title="Team Task Structuring Assistant (P-D-E-R-O)", layout="wide")
st.markdown("""
""", unsafe_allow_html=True)
# =========================
# 세션 초기화
# =========================
def _init(key, default):
if key not in st.session_state:
st.session_state[key] = default
_init("page", "도메인 설정")
_init("domains", [])
_init("grouped_tasks", {}) # {domain: [task, ...]}
_init("dependencies", {}) # {"domain::task": [task_name, ...]} (같은 도메인 내 이름 목록)
_init("outputs", {}) # {"domain::task": output_text}
_init("code_map", {}) # {"domain::task": {...}}
_init("seq_registry", {}) # {(domain_code, cycle): max_seq}
_init("domain_defaults", {}) # {domain_code: {"cycle":"E","time":"T"}}
# =========================
# 유틸
# =========================
PALETTE = ["#E3F2FD", "#E8F5E9", "#FFF8E1", "#FCE4EC", "#E0F7FA"]
def goto(page: str):
st.session_state.page = page
st.rerun()
def safe_key(domain: str, task: str, prefix: str) -> str:
h = hashlib.md5(f"{domain}::{task}".encode()).hexdigest()[:8]
return f"{prefix}_{h}"
def export_file(df: pd.DataFrame, kind="csv"):
if kind == "csv":
return df.to_csv(index=False).encode("utf-8-sig")
bio = BytesIO()
with pd.ExcelWriter(bio, engine="openpyxl") as w:
df.to_excel(w, index=False, sheet_name="tasks")
bio.seek(0)
return bio.getvalue()
def draw_dependency_graph(df: pd.DataFrame):
if df is None or df.empty:
return "
그래프를 표시할 데이터가 없습니다.
"
G = nx.DiGraph()
color_map = {"P": "#A7C7E7", "D": "#FFE8A3", "E": "#A8E6CF", "R": "#FFD3B6", "O": "#FFAAA5"}
for _, row in df.iterrows():
code = row["code"]
label = row["name"]
lifecycle = row.get("cycle", "E")
color = color_map.get(lifecycle, "#CFCFCF")
G.add_node(code, label=label, color=color)
for dep in str(row.get("depends_on", "")).split(","):
dep = dep.strip()
if dep:
G.add_edge(dep, code)
nt = Network(height="550px", width="100%", directed=True, bgcolor="#FFFFFF", font_color="#222")
nt.from_nx(G)
tmp_path = tempfile.NamedTemporaryFile(delete=False, suffix=".html").name
nt.save_graph(tmp_path)
html = open(tmp_path, "r", encoding="utf-8").read()
os.remove(tmp_path)
return html
def map_domain_to_code(d: str) -> str:
table = {
"공통": "COMM", "전략": "COMM",
"CEO": "CEO", "CEO서베이": "CEO",
"리더십서베이": "MULTI", "리더십": "MULTI",
"리더십심층": "DEEP", "심층진단": "DEEP",
"수시": "ADHOC", "요청": "ADHOC", "수시요청진단": "ADHOC",
"교육운영": "EDU", "교육": "EDU",
"데이터분석": "DATA", "분석": "DATA",
"운영관리": "OPS", "운영": "OPS",
}
for k, v in table.items():
if k in d:
return v
return d[:4].upper()
def ensure_unique_list(seq):
seen, out = set(), []
for x in seq:
x = x.strip()
if x and x not in seen:
seen.add(x); out.append(x)
return out
# =========================
# 상단 진행 표시바
# =========================
STEPS = ["도메인 설정", "업무 발산", "그룹 조정", "의존성 판단", "산출물 정의", "최종 정리"]
idx = STEPS.index(st.session_state.page) if st.session_state.page in STEPS else 0
st.progress((idx + 1) / len(STEPS), text=f"단계 {idx+1}/{len(STEPS)} : {st.session_state.page}")
# =========================
# 1) 도메인 설정
# =========================
if st.session_state.page == "도메인 설정":
st.title("1️⃣ 도메인 설정")
st.markdown("팀의 주요 업무 도메인을 3~4개 정의하세요. (예: 공통, 리더십서베이, 교육운영, 데이터분석 등)")
cols = st.columns(4)
new_domains = []
for i in range(4):
with cols[i]:
d = st.text_input(f"도메인 {i+1}", st.session_state.domains[i] if i < len(st.session_state.domains) else "")
if d:
new_domains.append(d.strip())
st.session_state.domains = ensure_unique_list([d for d in new_domains if d])
if st.button("➡️ 다음: 업무 발산"):
if not st.session_state.domains:
st.warning("도메인을 최소 1개 이상 입력하세요.")
else:
goto("업무 발산")
# =========================
# 2) 업무 발산
# =========================
elif st.session_state.page == "업무 발산":
st.title("2️⃣ 업무 발산")
st.markdown("각 도메인별 실제 수행 중인 업무를 가능한 한 많이 적어보세요. (줄바꿈으로 구분)")
for d in st.session_state.domains + ["기타"]:
st.subheader(f"📂 {d}")
text = st.text_area(f"{d} 업무", key=f"tasks_{d}", height=150,
placeholder="예: 리더십 진단 리포트 작성\n교육 기획\n데이터 분석 등")
if text:
tasks = [t.strip() for t in text.split("\n") if t.strip()]
st.session_state.grouped_tasks[d] = ensure_unique_list(tasks)
if st.button("➡️ 다음: 그룹 조정"):
goto("그룹 조정")
# =========================
# 3) 그룹 조정 —— 콜백 기반으로 안정화
# =========================
elif st.session_state.page == "그룹 조정":
st.title("3️⃣ 업무 그룹 조정")
st.markdown("도메인별 업무를 정리하고 **드래그로 순서를 바꾸거나**, **이동/삭제/추가** 하세요.")
st.divider()
domains = st.session_state.domains + ["기타"]
# --- 콜백 함수들 ---
def cb_delete(domain: str, task: str):
lst = list(st.session_state.grouped_tasks.get(domain, []))
st.session_state.grouped_tasks[domain] = [x for x in lst if x != task]
def cb_move(src: str, task: str, sel_key: str):
dst = st.session_state.get(sel_key, "(이동)")
if dst and dst != "(이동)" and dst != src:
src_list = list(st.session_state.grouped_tasks.get(src, []))
dst_list = list(st.session_state.grouped_tasks.get(dst, []))
if task in src_list:
src_list.remove(task)
if task not in dst_list:
dst_list.append(task)
st.session_state.grouped_tasks[src] = src_list
st.session_state.grouped_tasks[dst] = dst_list
# 선택 초기화
st.session_state[sel_key] = "(이동)"
def cb_add(domain: str, input_key: str):
val = (st.session_state.get(input_key) or "").strip()
if val:
lst = list(st.session_state.grouped_tasks.get(domain, []))
if val not in lst:
lst.append(val)
st.session_state.grouped_tasks[domain] = ensure_unique_list(lst)
# 입력창 비우기
st.session_state[input_key] = ""
cols = st.columns(len(domains))
for i, d in enumerate(domains):
with cols[i]:
bg = PALETTE[i % len(PALETTE)]
st.markdown(f"", unsafe_allow_html=True)
st.markdown(f"
📦 {d}
", unsafe_allow_html=True)
# 현재 리스트 가져와서 드래그 정렬 반영
orig = list(st.session_state.grouped_tasks.get(d, []))
sorted_tasks = sort_items(orig, direction="vertical", key=f"sort_{d}") or []
if sorted_tasks != orig:
st.session_state.grouped_tasks[d] = ensure_unique_list(sorted_tasks)
# 각 업무 행 렌더링
for t in st.session_state.grouped_tasks.get(d, []):
if not t.strip():
continue
c1, c2, c3, c4 = st.columns([5, 1, 2, 1])
with c1:
st.markdown(f"
{t}
", unsafe_allow_html=True)
with c2:
st.button("🗑", key=safe_key(d, t, "del_btn"),
help="삭제", on_click=cb_delete, args=(d, t))
with c3:
sel_key = safe_key(d, t, "mv_sel")
st.selectbox(" ", ["(이동)"] + [x for x in domains if x != d],
key=sel_key, label_visibility="collapsed")
with c4:
st.button("↪", key=safe_key(d, t, "mv_btn"),
help="이동", on_click=cb_move, args=(d, t, sel_key))
# 새 업무 추가(입력+버튼)
add_key = f"add_input_{i}"
st.text_input("새 업무", key=add_key, label_visibility="collapsed", placeholder="새 업무 입력")
st.button(f"➕ 추가 ({d})", key=f"add_btn_{i}", on_click=cb_add, args=(d, add_key))
st.markdown("
", unsafe_allow_html=True)
c1, c2 = st.columns(2)
if c1.button("⬅️ 이전: 업무 발산"): goto("업무 발산")
if c2.button("➡️ 다음: 의존성 판단"): goto("의존성 판단")
# =========================
# 4) 의존성 판단 (같은 도메인 내에서만)
# =========================
elif st.session_state.page == "의존성 판단":
st.title("4️⃣ 의존성 판단")
st.markdown("각 업무 간 선후관계(의존성)를 **같은 도메인 내에서만** 설정하세요.")
st.divider()
deps = {}
domains = st.session_state.domains + ["기타"]
for i, d in enumerate(domains):
st.subheader(f"📂 {d}")
tasks_in_domain = st.session_state.grouped_tasks.get(d, [])
for t in tasks_in_domain:
key = safe_key(d, t, "deps")
candidates = [x for x in tasks_in_domain if x != t] # 동일 도메인만
default_vals = [x for x in st.session_state.dependencies.get(f"{d}::{t}", []) if x in candidates]
selected = st.multiselect(f"'{t}' 이전 업무 (동일 도메인)", candidates, default=default_vals, key=key)
deps[f"{d}::{t}"] = selected
st.session_state.dependencies = deps
c1, c2 = st.columns(2)
if c1.button("⬅️ 이전: 그룹 조정"): goto("그룹 조정")
if c2.button("➡️ 다음: 산출물 정의"): goto("산출물 정의")
# =========================
# 5) 산출물 정의 (간결 매핑 UI, placeholder 제거)
# =========================
elif st.session_state.page == "산출물 정의":
st.title("5️⃣ 산출물 정의")
st.markdown("각 업무에 대응되는 산출물을 표에서 직접 입력하세요.")
st.divider()
updated_outputs = {}
for i, (d, tasks) in enumerate(st.session_state.grouped_tasks.items()):
bg = PALETTE[i % len(PALETTE)]
st.markdown(f"", unsafe_allow_html=True)
st.markdown(f"### 📂 {d}")
rows = [{"업무명": t, "산출물": st.session_state.outputs.get(f"{d}::{t}", "")} for t in tasks]
df = pd.DataFrame(rows)
# 일부 버전 호환을 위해 TextColumn의 placeholder 미사용
try:
col_cfg = {
"업무명": st.column_config.TextColumn(disabled=True),
"산출물": st.column_config.TextColumn()
}
edited = st.data_editor(df, key=f"editor_{i}", hide_index=True, num_rows="fixed",
column_config=col_cfg, use_container_width=True)
except Exception:
edited = st.data_editor(df, key=f"editor_{i}", hide_index=True, use_container_width=True)
for _, row in edited.iterrows():
updated_outputs[f"{d}::{row['업무명']}"] = row["산출물"]
st.markdown("
Tip: 산출물은 보고서/대시보드/시스템/데이터/교육자료 등으로 간단히 적어도 충분합니다.
", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
st.session_state.outputs = updated_outputs
c1, c2 = st.columns(2)
if c1.button("⬅️ 이전: 의존성 판단"): goto("의존성 판단")
if c2.button("➡️ 다음: 최종 정리"): goto("최종 정리")
# =========================
# 6) 최종 정리 (도메인 코드 매핑 + 단순화된 UI + 코드 생성)
# =========================
elif st.session_state.page == "최종 정리":
st.title("6️⃣ 최종 정리 및 업무 코드 생성")
st.markdown(
"- ① **도메인 → 영문코드**를 확인/수정\n"
"- ② 각 업무에 **사이클(P/D/E/R/O)**, **시간기호(T/FE)** 선택\n"
"- ③ 자동 생성된 코드를 확인하고 다운로드"
)
st.divider()
# ---- 0) 도메인 → 영문코드 매핑 편집 ----
_init("domain_code_overrides", {})
domains = list(st.session_state.grouped_tasks.keys())
# 자동 제안 + 기존 오버라이드 반영
map_rows = []
for d in domains:
suggested = map_domain_to_code(d)
current = st.session_state.domain_code_overrides.get(d, suggested)
map_rows.append({"도메인": d, "영문코드": current})
st.markdown("#### 🔤 도메인 코드 매핑")
map_df = pd.DataFrame(map_rows)
map_edited = st.data_editor(
map_df, key="domain_code_editor", hide_index=True, use_container_width=True,
column_config={
"도메인": st.column_config.TextColumn(disabled=True),
"영문코드": st.column_config.TextColumn()
}
)
# 형식 검증: 영문/숫자, 2~8자
invalid = []
domain_code_overrides = {}
for _, r in map_edited.iterrows():
dname = str(r["도메인"])
code = str(r["영문코드"]).strip().upper()
if not code or not code.isalnum() or not (2 <= len(code) <= 8):
invalid.append(f"- {dname}: '{r['영문코드']}' (영문/숫자 2~8자)")
domain_code_overrides[dname] = code
if invalid:
st.error("도메인 코드 형식 오류가 있습니다:\n" + "\n".join(invalid))
st.stop()
st.session_state.domain_code_overrides = domain_code_overrides
st.caption("Tip: 코드 예) COMM, MULTI, CEO, DEEP, ADHOC, EDU, DATA, OPS 등")
st.divider()
# ---- 1) 시퀀스 집계(안정적 부여 준비) ----
seq_registry = dict(st.session_state.seq_registry)
code_map = dict(st.session_state.code_map)
for task_key, meta in code_map.items():
new_domain_code = st.session_state.domain_code_overrides.get(meta.get("domain", ""), meta.get("domain_code", ""))
pair = (new_domain_code, meta.get("cycle", ""))
if pair[0] and pair[1]:
seq_registry[pair] = max(seq_registry.get(pair, 0), int(meta.get("seq", 0)))
# ---- 2) 업무 단위 UI (최소 선택 중심) ----
cycle_opts = ["P", "D", "E", "R", "O"]
time_opts = ["T", "FE"]
def recommend_cycles(is_p, is_d, is_e, is_r, is_o):
recs = []
if is_p: recs.append("P")
if is_d: recs.append("D")
if is_e: recs.append("E")
if is_r: recs.append("R")
if is_o: recs.append("O")
return recs
for d in domains:
domain_code = st.session_state.domain_code_overrides[d]
st.subheader(f"📂 {d} → `{domain_code}`")
tasks = st.session_state.grouped_tasks.get(d, [])
for t in tasks:
task_key = f"{d}::{t}"
prev = code_map.get(task_key, {})
st.markdown(f"**🧩 {t}**")
# 핵심 선택 (간결 UI): 사이클 · 시간기호
c1, c2 = st.columns([1.6, 1.0])
cycle_key = safe_key(d, t, "cycle")
time_key = safe_key(d, t, "time")
default_cycle = prev.get("cycle", "E")
default_time = prev.get("time", "T")
with c1:
cyc = st.radio(
"사이클", cycle_opts, key=cycle_key, horizontal=True,
index=cycle_opts.index(default_cycle) if cycle_key not in st.session_state else
cycle_opts.index(st.session_state[cycle_key])
)
with c2:
tm = st.radio(
"시간기호", time_opts, key=time_key, horizontal=True,
index=time_opts.index(default_time) if time_key not in st.session_state else
time_opts.index(st.session_state[time_key])
)
# 필요한 사람만 여는 도움말: 체크 → 추천 적용
with st.expander("도움이 필요하세요? (질문형 보조로 추천받기)", expanded=False):
qcols = st.columns(5)
# ✅ 각 체크박스에 고유 key 부여 (중복 ID 방지)
with qcols[0]:
qP = st.checkbox("P: 목적/시점/대상", key=safe_key(d, t, "qP"))
with qcols[1]:
qD = st.checkbox("D: 세팅/설계", key=safe_key(d, t, "qD"))
with qcols[2]:
qE = st.checkbox("E: 집행/수집", key=safe_key(d, t, "qE"))
with qcols[3]:
qR = st.checkbox("R: 해석/평가", key=safe_key(d, t, "qR"))
with qcols[4]:
qO = st.checkbox("O: 배포/반영", key=safe_key(d, t, "qO"))
recs = recommend_cycles(qP, qD, qE, qR, qO)
if recs:
rcols = st.columns(len(recs))
for i, r in enumerate(recs):
if rcols[i].button(f"추천 적용 {r}", key=safe_key(d, t, f"apply_{r}")):
st.session_state[cycle_key] = r
st.rerun()
else:
st.caption("체크 결과가 없거나 모호합니다. 직접 선택을 유지하세요.")
# 최종 값 반영
cyc = st.session_state[cycle_key]
tm = st.session_state[time_key]
# 시퀀스 부여: 동일 (domain_code, cycle)이면 기존 유지, 아니면 다음 번호
pair = (domain_code, cyc)
if prev and prev.get("domain_code") == domain_code and prev.get("cycle") == cyc and int(prev.get("seq", 0)) > 0:
seq = int(prev["seq"])
else:
seq = int(seq_registry.get(pair, 0)) + 1
seq_registry[pair] = seq
code = f"{domain_code}-{cyc}{seq:02d}-{tm}"
st.markdown(
f"