# 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"
" f"➡️ 코드 {code}
", unsafe_allow_html=True ) # 저장 code_map[task_key] = { "domain": d, "domain_code": domain_code, "name": t, "cycle": cyc, "time": tm, "seq": seq, "code": code, } # 상태 갱신 st.session_state.seq_registry = seq_registry st.session_state.code_map = code_map # ---- 3) 결과 요약/그래프/다운로드 ---- st.divider() st.markdown("#### 📘 최종 코드 목록") rows = [] for task_key, meta in st.session_state.code_map.items(): d, t = task_key.split("::", 1) deps_names = st.session_state.dependencies.get(task_key, []) rows.append({ "domain": d, "domain_code": meta["domain_code"], "name": t, "cycle": meta["cycle"], "time": meta["time"], "seq": meta["seq"], "code": meta["code"], "depends_on": ", ".join(deps_names), "output": st.session_state.outputs.get(task_key, ""), }) df = pd.DataFrame(rows) if not df.empty: df = df.sort_values(by=["domain_code", "cycle", "seq", "name"]).reset_index(drop=True) st.dataframe( df[["domain", "domain_code", "name", "cycle", "time", "code", "depends_on", "output"]], use_container_width=True ) # 의존성 그래프(이름→코드 간선 매핑) name_to_code = {meta["name"]: meta["code"] for meta in st.session_state.code_map.values()} df_graph = df.copy() if not df_graph.empty: df_graph["depends_on"] = df_graph["depends_on"].apply( lambda s: ", ".join([name_to_code.get(x.strip(), x.strip()) for x in s.split(",") if x.strip()]) if s else "" ) html = draw_dependency_graph(df_graph.rename(columns={"code": "code", "name": "name"}) if not df_graph.empty else df_graph) st.components.v1.html(html, height=520, scrolling=True) c1, c2 = st.columns(2) c1.download_button("⬇️ CSV 다운로드", export_file(df, "csv"), "final_task_codes.csv", "text/csv") c2.download_button("⬇️ Excel 다운로드", export_file(df, "xlsx"), "final_task_codes.xlsx") if st.button("🔄 처음으로 돌아가기"): for k in ["page", "domains", "grouped_tasks", "dependencies", "outputs", "code_map", "seq_registry"]: if k == "domains": st.session_state[k] = [] elif k == "page": st.session_state[k] = "도메인 설정" else: st.session_state[k] = {} st.rerun() # 푸터 st.markdown("---") st.caption("© 2025 Crystal_MVP")