#!/usr/bin/env python3 """ organization-structure.json 최적화 스크립트 작성자: 엔키 (Enki) - 개발5팀 백엔드 개발자 목적: 메인 파일을 Read tool 25,000 토큰 제한 이하로 축소 """ import json import shutil from pathlib import Path # 경로 설정 MEMORY_DIR = Path("/home/jay/workspace/memory") ORG_FILE = MEMORY_DIR / "organization-structure.json" DETAIL_DIR = MEMORY_DIR / "org-details" BACKUP_FILE = DETAIL_DIR / "organization-structure.backup.json" # expertise에서 유지할 필드 (나머지는 제거) EXPERTISE_KEEP = {"primary", "style"} # 멤버에서 유지할 필드 (나머지는 제거) MEMBER_KEEP = {"id", "name", "role", "model", "engine", "status", "persona", "expertise", "note", "caller", "specialty"} # 멤버에서 반드시 제거할 필드 MEMBER_REMOVE = {"invoke"} def prune_expertise(expertise: object) -> tuple[dict, dict]: """ expertise 딕셔너리에서 primary, style만 유지하고 나머지를 분리. Returns: (pruned_expertise, removed_fields) """ if not isinstance(expertise, dict): return {}, {} pruned = {} removed = {} for key, val in expertise.items(): if key in EXPERTISE_KEEP: pruned[key] = val else: removed[key] = val return pruned, removed def prune_member(member: dict) -> tuple[dict, dict]: """ 멤버 객체에서 필요 필드만 유지. Returns: (pruned_member, detail_data) """ pruned = {} detail = {} # invoke는 detail에 저장 if "invoke" in member: detail["invoke"] = member["invoke"] for key, val in member.items(): if key in MEMBER_REMOVE: continue # invoke 제거 if key == "expertise": if isinstance(val, dict): pruned_exp, _ = prune_expertise(val) pruned["expertise"] = pruned_exp # detail에 원본 expertise 전체 저장 detail["expertise_full"] = val else: pruned["expertise"] = val elif key in MEMBER_KEEP: pruned[key] = val # MEMBER_KEEP에 없는 다른 필드는 유지 (specialty 등 이미 포함됨) return pruned, detail def process_team_members(members: list) -> tuple[list, dict]: """ 팀 멤버 목록 처리. 축약된 멤버 목록과 상세 데이터 반환. """ pruned_members = [] members_detail = {} for member in members: member_id = member.get("id", "unknown") pruned, detail = prune_member(member) pruned_members.append(pruned) if detail: members_detail[member_id] = detail return pruned_members, members_detail def process_lead(lead: dict) -> tuple[dict, dict]: """ 팀장 객체 처리. 축약된 팀장 데이터와 상세 데이터 반환. """ if not lead or not isinstance(lead, dict): return lead, {} # 팀장도 invoke 제거 및 expertise 축약 pruned, detail = prune_member(lead) return pruned, detail def main(): print("=" * 60) print("organization-structure.json 최적화 시작") print("=" * 60) # 1. 원본 파일 로드 print(f"\n[1] 원본 파일 로드: {ORG_FILE}") with open(ORG_FILE, "r", encoding="utf-8") as f: data = json.load(f) orig_size = ORG_FILE.stat().st_size print(f" 원본 크기: {orig_size:,} bytes") # 2. 상세 디렉토리 생성 및 백업 DETAIL_DIR.mkdir(parents=True, exist_ok=True) print(f"\n[2] 백업 생성: {BACKUP_FILE}") shutil.copy2(ORG_FILE, BACKUP_FILE) print(f" 백업 완료: {BACKUP_FILE.stat().st_size:,} bytes") # 3. 데이터 처리 및 상세 파일 생성 print("\n[3] 팀별 상세 파일 생성 및 메인 파일 축약") teams = data["structure"]["columns"]["teams"] for team_idx, team in enumerate(teams): team_id = team.get("team_id", f"team-{team_idx}") team_members_detail = {} print(f"\n 처리 중: {team_id}") if team_id == "development-office": # development-office는 sub_teams 처리 for sub_idx, sub_team in enumerate(team.get("sub_teams", [])): sub_id = sub_team.get("sub_team_id", f"sub-{sub_idx}") # sub_team lead 처리 if "lead" in sub_team: lead_orig = sub_team["lead"] lead_id = lead_orig.get("id", "unknown") pruned_lead, lead_detail = process_lead(lead_orig) sub_team["lead"] = pruned_lead if lead_detail: team_members_detail[lead_id] = lead_detail # sub_team members 처리 if "members" in sub_team: pruned_members, sub_detail = process_team_members(sub_team["members"]) sub_team["members"] = pruned_members team_members_detail.update(sub_detail) # sub_team에 detail_path 추가 sub_team["detail_path"] = f"memory/org-details/{sub_id}.json" # development-office의 office_lead, direct_worker도 처리 if "office_lead" in team: lead_orig = team["office_lead"] if isinstance(lead_orig, dict) and "id" in lead_orig: lead_id = lead_orig.get("id", "unknown") pruned_lead, lead_detail = process_lead(lead_orig) team["office_lead"] = pruned_lead if lead_detail: team_members_detail[lead_id] = lead_detail if "direct_worker" in team: dw = team["direct_worker"] if isinstance(dw, dict) and "id" in dw: dw_id = dw.get("id", "unknown") pruned_dw, dw_detail = process_lead(dw) team["direct_worker"] = pruned_dw if dw_detail: team_members_detail[dw_id] = dw_detail # 상세 파일 생성 (development-office 전체) detail_file = DETAIL_DIR / f"{team_id}.json" detail_data = { "team_id": team_id, "extracted_from": "organization-structure.json", "members_detail": team_members_detail } with open(detail_file, "w", encoding="utf-8") as f: json.dump(detail_data, f, ensure_ascii=False, indent=2) print(f" -> 상세 파일 생성: {detail_file.name} ({detail_file.stat().st_size:,} bytes)") # development-office에 detail_path 추가 team["detail_path"] = f"memory/org-details/{team_id}.json" else: # 일반 팀 처리 # team lead 처리 (expertise가 있는 경우만) if "lead" in team: lead_orig = team["lead"] if isinstance(lead_orig, dict) and "expertise" in lead_orig: lead_id = lead_orig.get("id", "unknown") pruned_lead, lead_detail = process_lead(lead_orig) team["lead"] = pruned_lead if lead_detail: team_members_detail[lead_id] = lead_detail # team members 처리 if "members" in team: pruned_members, mem_detail = process_team_members(team["members"]) team["members"] = pruned_members team_members_detail.update(mem_detail) # 상세 파일 생성 detail_file = DETAIL_DIR / f"{team_id}.json" detail_data = { "team_id": team_id, "extracted_from": "organization-structure.json", "members_detail": team_members_detail } with open(detail_file, "w", encoding="utf-8") as f: json.dump(detail_data, f, ensure_ascii=False, indent=2) print(f" -> 상세 파일 생성: {detail_file.name} ({detail_file.stat().st_size:,} bytes)") # 팀에 detail_path 추가 team["detail_path"] = f"memory/org-details/{team_id}.json" # 4. centers 처리 print("\n 센터 처리:") centers = data["structure"]["rows"]["centers"] for center in centers: center_id = center.get("center_id", "unknown") center_members_detail = {} # center lead 처리 if "lead" in center: lead_orig = center["lead"] if isinstance(lead_orig, dict) and "expertise" in lead_orig: lead_id = lead_orig.get("id", "unknown") pruned_lead, lead_detail = process_lead(lead_orig) center["lead"] = pruned_lead if lead_detail: center_members_detail[lead_id] = lead_detail # center members 처리 if "members" in center: pruned_members, mem_detail = process_team_members(center["members"]) center["members"] = pruned_members center_members_detail.update(mem_detail) # 상세 파일 생성 detail_file = DETAIL_DIR / f"{center_id}.json" detail_data = { "team_id": center_id, "extracted_from": "organization-structure.json", "members_detail": center_members_detail } with open(detail_file, "w", encoding="utf-8") as f: json.dump(detail_data, f, ensure_ascii=False, indent=2) print(f" -> {detail_file.name} ({detail_file.stat().st_size:,} bytes)") # center에 detail_path 추가 center["detail_path"] = f"memory/org-details/{center_id}.json" # 5. 메인 파일 저장 (in-place) print(f"\n[4] 메인 파일 저장 (in-place): {ORG_FILE}") with open(ORG_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) # 5.5 개별 sub-team 및 팀 상세 파일 생성 (공통 모듈 사용) print("\n[5] 팀별 개별 상세 파일 생성") try: import sys sys.path.insert(0, str(Path("/home/jay/workspace/teams/shared"))) from generate_org_details import generate_all_sub_team_details # pyright: ignore[reportMissingImports] generated = generate_all_sub_team_details( org_file=ORG_FILE, backup_file=BACKUP_FILE, output_dir=DETAIL_DIR, ) for fp in generated: fname = Path(fp).name fsize = Path(fp).stat().st_size print(f" -> {fname} ({fsize:,} bytes)") print(f" 총 {len(generated)}개 개별 상세 파일 생성 완료") except ImportError as e: print(f" 경고: 공통 모듈 로드 실패 ({e}). 개별 상세 파일 미생성.") except Exception as e: print(f" 에러: 개별 상세 파일 생성 실패: {e}") # 6. 결과 검증 print("\n" + "=" * 60) print("검증 결과") print("=" * 60) new_size = ORG_FILE.stat().st_size reduction = (1 - new_size / orig_size) * 100 estimated_tokens = new_size / 3.2 # 줄 수 계산 with open(ORG_FILE, "r", encoding="utf-8") as f: lines = f.readlines() line_count = len(lines) print(f"\n파일 크기:") print(f" 원본: {orig_size:,} bytes") print(f" 최적화: {new_size:,} bytes") print(f" 감소율: {reduction:.1f}%") print(f"\n줄 수: {line_count:,} 줄") print(f"추정 토큰 수: {estimated_tokens:,.0f} tokens (문자수/3.2)") print(f"목표 (25,000): {'달성' if estimated_tokens <= 25000 else '미달성'}") # JSON 유효성 확인 print("\n[검증 1] JSON 유효성:") try: with open(ORG_FILE, "r", encoding="utf-8") as f: verify_data = json.load(f) print(" OK - JSON 파싱 성공") except json.JSONDecodeError as e: print(f" FAIL - JSON 파싱 실패: {e}") return # 핵심 필드 접근 테스트 print("\n[검증 2] 핵심 필드 접근 테스트:") # development-office sub_teams[0].members[0].expertise.primary (동적 탐색) dev_office = next((t for t in verify_data["structure"]["columns"]["teams"] if t.get("team_id") == "development-office"), None) try: val = dev_office["sub_teams"][0]["members"][0]["expertise"]["primary"] # type: ignore[index] print(f" OK - development-office.sub_teams[0].members[0].expertise.primary = '{val}'") except (KeyError, IndexError, TypeError) as e: print(f" FAIL - development-office.sub_teams[0].members[0].expertise.primary: {e}") # centers[0].lead.expertise.primary try: val = verify_data["structure"]["rows"]["centers"][0]["lead"]["expertise"]["primary"] print(f" OK - centers[0].lead.expertise.primary = '{val}'") except (KeyError, IndexError, TypeError) as e: print(f" FAIL - centers[0].lead.expertise.primary: {e}") # expertise가 객체인지 확인 (string으로 변환 금지) print("\n[검증 3] expertise 타입 확인 (객체여야 함):") dev_members = dev_office["sub_teams"][0]["members"] if dev_office else [] # type: ignore[index] for m in dev_members[:2]: if "expertise" in m: exp_type = type(m["expertise"]).__name__ print(f" {m['id']}.expertise 타입: {exp_type} {'OK' if exp_type == 'dict' else 'FAIL'}") # invoke 필드 제거 확인 print("\n[검증 4] invoke 필드 제거 확인:") invoke_count = 0 def count_invoke(obj): nonlocal invoke_count if isinstance(obj, dict): if "invoke" in obj: invoke_count += 1 for v in obj.values(): count_invoke(v) elif isinstance(obj, list): for item in obj: count_invoke(item) count_invoke(verify_data) print(f" 메인 파일 내 invoke 필드 수: {invoke_count} (0이어야 함) {'OK' if invoke_count == 0 else 'FAIL'}") # detail_path 추가 확인 print("\n[검증 5] detail_path 추가 확인:") for team in verify_data["structure"]["columns"]["teams"]: tid = team.get("team_id", "?") if tid == "development-office": for sub in team.get("sub_teams", []): sid = sub.get("sub_team_id", "?") has_dp = "detail_path" in sub print(f" sub_team {sid}: detail_path {'OK' if has_dp else 'FAIL'}") else: has_dp = "detail_path" in team print(f" team {tid}: detail_path {'OK' if has_dp else 'FAIL'}") for center in verify_data["structure"]["rows"]["centers"]: cid = center.get("center_id", "?") has_dp = "detail_path" in center print(f" center {cid}: detail_path {'OK' if has_dp else 'FAIL'}") # 상세 파일 목록 및 개별 sub-team 파일 존재 확인 print("\n[검증 6] 생성된 상세 파일 및 개별 sub-team 파일 확인:") detail_files = sorted(DETAIL_DIR.glob("*.json")) aggregate_files = [] subteam_files = [] for df in detail_files: if df.name == "organization-structure.backup.json": continue size = df.stat().st_size print(f" {df.name}: {size:,} bytes") # development-office sub-teams are named dev1~devN, others are aggregate team files if df.stem.startswith("dev") and df.stem != "development-office": subteam_files.append(df) else: aggregate_files.append(df) print(f"\n 집계 팀 파일 수: {len(aggregate_files)}") print(f" 개별 sub-team 파일 수: {len(subteam_files)}") # 개별 sub-team 파일 존재 여부 검증 print("\n 개별 sub-team 파일 존재 확인:") for team in verify_data["structure"]["columns"]["teams"]: if team.get("team_id") == "development-office": for sub in team.get("sub_teams", []): sid = sub.get("sub_team_id", "?") expected = DETAIL_DIR / f"{sid}.json" exists = expected.exists() print(f" {sid}.json: {'OK' if exists else 'FAIL (없음)'}") print("\n" + "=" * 60) print("최적화 완료!") print("=" * 60) if __name__ == "__main__": main()