from __future__ import annotations from datetime import datetime from pathlib import Path from xml.sax.saxutils import escape import math import struct import zlib import zipfile BASE_DIR = Path(__file__).resolve().parent.parent REPORT_DIR = BASE_DIR / "reports" OUTPUT_PATH = REPORT_DIR / "my-intranet-app_architecture_report_20260522_v3.docx" PAGE_WIDTH = 11906 MARGIN = 620 CONTENT_WIDTH = PAGE_WIDTH - (MARGIN * 2) def esc(value: object) -> str: return escape(str(value), {'"': """}) def run(text: str, *, bold: bool = False, size: int | None = None, font: str | None = None) -> str: props: list[str] = [] if bold: props.append("") if size: props.append(f'') if font: props.append(f'') rpr = f"{''.join(props)}" if props else "" preserve = ' xml:space="preserve"' if text[:1].isspace() or text[-1:].isspace() else "" return f"{rpr}{esc(text)}" def para(text: str = "", *, style: str | None = None, bold: bool = False, size: int | None = None, font: str | None = None, after: int | None = None) -> str: pprops: list[str] = [] if style: pprops.append(f'') if after is not None: pprops.append(f'') ppr = f"{''.join(pprops)}" if pprops else "" return f"{ppr}{run(text, bold=bold, size=size, font=font)}" def bullet(text: str) -> str: return para("• " + text, after=60) def table(headers: list[str], rows: list[list[str]], widths: list[int] | None = None) -> str: if widths is None: widths = [CONTENT_WIDTH // len(headers)] * len(headers) grid = "".join(f'' for w in widths) def cell(text: str, width: int, header: bool = False) -> str: fill = '' if header else "" props = ( f'{fill}' '' '' ) return f"{props}{para(text, bold=header, size=16, after=0)}" rows_xml = ["" + "".join(cell(h, widths[i], True) for i, h in enumerate(headers)) + ""] rows_xml.extend( "" + "".join(cell(c, widths[i]) for i, c in enumerate(row)) + "" for row in rows ) return ( '' f'' "" f"{grid}{''.join(rows_xml)}" ) class PngCanvas: def __init__(self, width: int, height: int, bg: tuple[int, int, int] = (255, 255, 255)): self.width = width self.height = height self.px = bytearray(bg * width * height) def set(self, x: int, y: int, color: tuple[int, int, int]) -> None: if 0 <= x < self.width and 0 <= y < self.height: i = (y * self.width + x) * 3 self.px[i : i + 3] = bytes(color) def rect(self, x: int, y: int, w: int, h: int, fill: tuple[int, int, int], border: tuple[int, int, int], bw: int = 3) -> None: for yy in range(y, y + h): for xx in range(x, x + w): if x <= xx < x + w and y <= yy < y + h: self.set(xx, yy, fill) for n in range(bw): self.line(x + n, y + n, x + w - 1 - n, y + n, border) self.line(x + n, y + h - 1 - n, x + w - 1 - n, y + h - 1 - n, border) self.line(x + n, y + n, x + n, y + h - 1 - n, border) self.line(x + w - 1 - n, y + n, x + w - 1 - n, y + h - 1 - n, border) def line(self, x1: int, y1: int, x2: int, y2: int, color: tuple[int, int, int], width: int = 3) -> None: dx = abs(x2 - x1) dy = -abs(y2 - y1) sx = 1 if x1 < x2 else -1 sy = 1 if y1 < y2 else -1 err = dx + dy x, y = x1, y1 while True: r = width // 2 for yy in range(y - r, y + r + 1): for xx in range(x - r, x + r + 1): self.set(xx, yy, color) if x == x2 and y == y2: break e2 = 2 * err if e2 >= dy: err += dy x += sx if e2 <= dx: err += dx y += sy def arrow(self, x1: int, y1: int, x2: int, y2: int, color: tuple[int, int, int] = (75, 88, 99)) -> None: self.line(x1, y1, x2, y2, color, 4) ang = math.atan2(y2 - y1, x2 - x1) for a in (ang + 2.55, ang - 2.55): self.line(x2, y2, int(x2 + 22 * math.cos(a)), int(y2 + 22 * math.sin(a)), color, 4) def digit(self, x: int, y: int, digit: str, color: tuple[int, int, int] = (20, 39, 54), scale: int = 8) -> None: glyphs = { "0": ["111", "101", "101", "101", "111"], "1": ["010", "110", "010", "010", "111"], "2": ["111", "001", "111", "100", "111"], "3": ["111", "001", "111", "001", "111"], "4": ["101", "101", "111", "001", "001"], "5": ["111", "100", "111", "001", "111"], "6": ["111", "100", "111", "101", "111"], "7": ["111", "001", "010", "010", "010"], "8": ["111", "101", "111", "101", "111"], "9": ["111", "101", "111", "001", "111"], }[digit] for gy, row in enumerate(glyphs): for gx, v in enumerate(row): if v == "1": self.rect(x + gx * scale, y + gy * scale, scale - 1, scale - 1, color, color, 1) def number_badge(self, x: int, y: int, n: int) -> None: self.rect(x - 28, y - 28, 56, 56, (255, 255, 255), (47, 111, 163), 4) self.digit(x - 12, y - 18, str(n), scale=8) def png(self) -> bytes: rows = bytearray() stride = self.width * 3 for y in range(self.height): rows.append(0) rows.extend(self.px[y * stride : (y + 1) * stride]) def chunk(kind: bytes, data: bytes) -> bytes: return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF) return ( b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", self.width, self.height, 8, 2, 0, 0, 0)) + chunk(b"IDAT", zlib.compress(bytes(rows), 9)) + chunk(b"IEND", b"") ) def architecture_png() -> bytes: c = PngCanvas(1200, 640, (252, 254, 255)) blue, green, gold, purple, gray = (232, 242, 252), (237, 248, 237), (255, 248, 230), (246, 239, 250), (78, 91, 104) c.rect(55, 260, 170, 90, blue, (47, 111, 163)); c.number_badge(140, 305, 1) c.rect(310, 215, 245, 180, green, (63, 143, 77)); c.number_badge(432, 305, 2) c.rect(650, 45, 230, 90, gold, (176, 122, 26)); c.number_badge(765, 90, 3) c.rect(650, 185, 230, 105, blue, (47, 111, 163)); c.number_badge(765, 238, 4) c.rect(650, 340, 230, 105, purple, (127, 85, 160)); c.number_badge(765, 393, 5) c.rect(650, 500, 230, 90, gold, (176, 122, 26)); c.number_badge(765, 545, 6) c.rect(970, 230, 175, 145, green, (63, 143, 77)); c.number_badge(1058, 303, 7) c.arrow(225, 305, 310, 305, gray); c.arrow(555, 270, 650, 92, gray); c.arrow(555, 305, 650, 238, gray) c.arrow(555, 335, 650, 393, gray); c.arrow(555, 375, 650, 545, gray); c.arrow(880, 238, 970, 285, gray); c.arrow(880, 393, 970, 325, gray) return c.png() def flow_png() -> bytes: c = PngCanvas(1200, 640, (255, 255, 255)) colors = [(232, 242, 252), (237, 248, 237), (255, 248, 230), (246, 239, 250), (238, 242, 246)] border = [(47, 111, 163), (63, 143, 77), (176, 122, 26), (127, 85, 160), (90, 105, 120)] boxes = [(60, 90, 190, 105), (330, 90, 190, 105), (600, 90, 190, 105), (870, 90, 190, 105), (330, 350, 190, 105), (600, 350, 190, 105), (870, 350, 190, 105)] for i, (x, y, w, h) in enumerate(boxes, start=1): c.rect(x, y, w, h, colors[(i - 1) % len(colors)], border[(i - 1) % len(border)]) c.number_badge(x + w // 2, y + h // 2, i) gray = (78, 91, 104) c.arrow(250, 142, 330, 142, gray); c.arrow(520, 142, 600, 142, gray); c.arrow(790, 142, 870, 142, gray) c.arrow(965, 195, 965, 350, gray); c.arrow(870, 402, 790, 402, gray); c.arrow(600, 402, 520, 402, gray) c.arrow(425, 350, 425, 195, gray) return c.png() def image_paragraph(rel_id: str, width_emu: int = 6_950_000, height_emu: int = 3_700_000) -> str: return f""" """ def body_xml() -> str: now = datetime.now().strftime("%Y-%m-%d %H:%M") p: list[str] = [] p.append(para("my-intranet-app 아키텍처 분석 보고서", style="Title")) p.append(para(f"작성일: {now} / 기준 경로: {BASE_DIR}", style="Subtitle")) p.append(para("본 보고서의 주 목적은 현재 my-intranet-app의 아키텍처, 데이터 흐름, 주요 결합도와 개선 포인트를 분석하는 것입니다. WSL 제로 세팅 시 보존 범위는 마지막 운영 고려사항으로 덧붙였습니다.")) p.append(para("1. 분석 결론", style="Heading1")) p.append(bullet("현재 시스템은 FastAPI 단일 애플리케이션 안에 화면 렌더링, JSON API, SQLite 접근, 캐시, 백그라운드 작업, 외부 ERP/WEHAGO 연동이 결합된 내부 업무 앱입니다.")) p.append(bullet("업무 데이터의 중심은 SQLite data.db이며, 프로젝트/회계/전표비교/캐시/작업 이력이 같은 DB에 함께 저장됩니다.")) p.append(bullet("구조 안정성 측면의 가장 큰 리스크는 main.py의 과도한 책임 집중과 대용량 SQLite 파일에 운영 데이터와 캐시가 공존하는 점입니다.")) p.append(bullet("속도는 런타임 캐시, system_page_cache, WEHAGO query/export projection, background job으로 보완하고 있으나, 장기적으로는 DB 슬림화와 모듈 분리가 필요합니다.")) p.append(para("2. 아키텍처 개요 이미지", style="Heading1")) p.append(image_paragraph("rId10")) p.append(table( ["번호", "구성요소", "설명"], [ ["1", "Browser/UI", "Jinja2로 내려받은 HTML과 브라우저 fetch API가 화면 갱신을 담당"], ["2", "FastAPI Runtime(main.py)", "라우트, HTML 렌더링, JSON API, DB 초기화, 캐시, 작업 큐 진입점"], ["3", "Templates", "dashboard/projects/process_cost/wehago_compare 등 서버 렌더링 화면"], ["4", "SQLite data.db", "업무 데이터, 설정, 캐시, 작업 상태의 중심 저장소"], ["5", "WEHAGO Compare", "전표 비교 전문 로직. 파일 정규화, 매칭, 리뷰, 추천, export"], ["6", "Hanmac External", "pymysql 기반 외부 ERP DB 조회 및 집계"], ["7", "Workers/Jobs", "캐시 재생성, snapshot, export, 유지보수 작업"], ], [750, 2500, CONTENT_WIDTH - 3250], )) p.append(para("3. 런타임/기술 스택", style="Heading1")) p.append(table( ["영역", "현재 구성", "역할/관찰"], [ ["Web", "FastAPI, Uvicorn", "ASGI 기반 단일 서버. HTML과 JSON API를 같은 앱에서 제공"], ["UI", "Jinja2, CSS, Vanilla JS", "템플릿별 inline CSS/JS가 많고 fetch 기반 동적 로딩을 사용"], ["DB", "SQLite, SQLAlchemy", "로컬 data.db를 중심으로 업무/캐시/작업 데이터 저장"], ["Excel", "openpyxl", "업로드 파일 파싱, WEHAGO 상태별 xlsx 내보내기"], ["External DB", "pymysql", "Hanmac 외부 MySQL 접속/preview/aggregate"], ["DB Browser", "Datasette", "/db 및 /db-browser에서 내부 DB 조회 지원"], ], [1400, 2600, CONTENT_WIDTH - 4000], )) p.append(para("4. 코드 구조 분석", style="Heading1")) p.append(table( ["파일/디렉터리", "아키텍처상 책임", "개선 관점"], [ ["main.py", "FastAPI app 생성, 라우트, DB 초기화, 화면별 bootstrap, 저장 API, system job, Hanmac 연동", "routers/services/repositories/jobs로 단계 분리 필요"], ["wehago_compare.py", "WEHAGO/ERP 전표 비교 도메인. 매칭, 리뷰, 추천, projection/cache/export", "비교 도메인으로 분리된 점은 좋으나 내부 함수가 매우 크고 캐시 책임도 함께 큼"], ["templates/*.html", "서버 렌더링 화면과 화면별 대형 JS/CSS", "공통 fetch/job polling/table rendering을 static 모듈로 분리 가능"], ["scripts/", "서버 실행, WEHAGO 수집/검증/보정, Windows portproxy", "운영 자동화와 일회성 보정 스크립트 구분 필요"], ["data.db", "업무 데이터와 캐시/작업 이력 저장", "운영 데이터와 재생성 캐시 분리 또는 보존 정책 필요"], ["backups/", "수동/시점 백업", "복구 가치 기준으로 최신/중요 백업만 관리 권장"], ], [2000, 4300, CONTENT_WIDTH - 6300], )) p.append(para("5. 주요 화면/API 경계", style="Heading1")) p.append(table( ["화면/도메인", "대표 라우트", "핵심 데이터 흐름"], [ ["Dashboard", "/, /bootstrap-data, /dashboard/api/rebuild-cache", "transactions/project 집계 -> bootstrap/cache -> 차트/KPI"], ["Projects", "/projects, /projects/bootstrap-data, /projects/save-json", "project_* 조회/저장 -> 미계약/관련 프로젝트/비교 상세 API"], ["Process Cost", "/process-cost, /process-cost/bootstrap-data", "Hanmac/WEHAGO 소스 선택 -> 프로젝트별 수익/비용/진척/비율 계산"], ["Annual Summary", "/annual-summary, /annual-summary/bootstrap-data", "연도/월별 회계 집계 -> 차트 데이터"], ["WEHAGO Compare", "/wehago-compare/api/*", "원천 rows -> 비교 결과 -> 상태별 상세 -> 리뷰/매칭/export"], ["Hanmac Browser", "/hanmac-browser/api/*", "외부 MySQL 조회 -> preview/aggregate cache -> CSV export"], ["System Jobs", "/api/system-jobs/*", "무거운 cache rebuild/export 작업 생성 및 진행률 조회"], ], [1900, 3300, CONTENT_WIDTH - 5200], )) p.append(para("6. 데이터 흐름 이미지", style="Heading1")) p.append(image_paragraph("rId11")) p.append(table( ["번호", "흐름 단계", "설명"], [ ["1", "원천 데이터", "Excel 업로드, WEHAGO_DB 파일, Hanmac 외부 DB, 사용자 입력"], ["2", "수집/정규화", "main.py와 wehago_compare.py에서 날짜/금액/전표번호/프로젝트코드 정규화"], ["3", "영속 저장", "transactions, project_*, wehago_* 테이블에 저장"], ["4", "집계/비교 계산", "프로젝트 원가, 연도 집계, 전표 매칭, 상태별 metric 계산"], ["5", "캐시/작업", "system_page_cache, wehago query cache, background jobs로 무거운 조회 완화"], ["6", "API 응답", "bootstrap-data 및 상세 JSON API로 화면에 전달"], ["7", "화면 표시/export", "Jinja2 화면, fetch 갱신, xlsx/csv 다운로드"], ], [750, 2200, CONTENT_WIDTH - 2950], )) p.append(para("7. 데이터 아키텍처", style="Heading1")) p.append(table( ["테이블 그룹", "대표 테이블", "아키텍처 의미"], [ ["업무 원장", "transactions", "회계 전표/거래 행의 중심 원천"], ["프로젝트", "project_basic_info, project_status, project_contract_info, project_billing_entries, project_collection_entries", "프로젝트 기본/계약/청구/수금/상태"], ["프로젝트 분석", "project_exec_budget_entries, project_actual_input_entries, project_task_plan_entries, project_analysis_settings", "원가/투입/계획/분석 설정"], ["WEHAGO 비교 원천", "wehago_source_files, wehago_voucher_rows, wehago_ledger_rows", "ERP 전표와 WEHAGO 원장 정규화 데이터"], ["WEHAGO 비교 결과", "wehago_comparison_results, wehago_recheck_reviews, wehago_manual_pair_matches", "비교 결과와 사용자가 만든 검토/매칭 상태"], ["캐시/작업", "system_page_cache, system_jobs, wehago_*_cache, hanmac_*_cache", "속도 보완용. 일부는 재생성 가능"], ["설정/운영", "app_option_items, app_keyword_rules, hanmac_holidays, db_backup_history", "분류 규칙, 옵션, 휴일, 백업 이력"], ], [1800, 4300, CONTENT_WIDTH - 6100], )) p.append(para("8. 구조 안정성/속도 개선 포인트", style="Heading1")) p.append(table( ["개선 영역", "현재 리스크", "권장 방향"], [ ["main.py 책임 분리", "라우트/DB/worker/비즈니스 로직 집중", "도메인별 router, service, repository, job 모듈로 점진 분리"], ["DB 관리", "14GB 수준 SQLite에 운영 데이터와 캐시 공존", "캐시 보존 정책, VACUUM/ANALYZE, cache DB 분리 검토"], ["WEHAGO 비교", "projection/cache가 많고 상태별 경로가 복잡", "상태별 query path 정리, 캐시 키 문서화, 재계산 CLI 표준화"], ["Frontend", "템플릿별 inline JS/CSS가 큼", "공통 fetch/polling/render 유틸을 static JS/CSS로 이동"], ["작업 큐", "DB 테이블 기반 작업 상태와 런타임 worker 결합", "작업 타입/상태 전이 규칙 문서화 및 stale job 정리 강화"], ["테스트", "구조 변경 후 회귀 확인 경로 부족", "핵심 bootstrap API와 저장 API smoke test 추가"], ], [1800, 3300, CONTENT_WIDTH - 5100], )) p.append(para("9. WSL 제로 세팅 시 보존 범위(부가 운영 고려사항)", style="Heading1")) p.append(para("이 절은 이관 방법 보고서가 아니라, 현재 아키텍처를 보존 가능한 상태로 유지하려면 어떤 정보를 어느 수준까지 관리해야 하는지에 대한 부가 판단입니다.")) p.append(table( ["대상", "보존 수준", "이유"], [ ["코드", "필수", "main.py, wehago_compare.py, templates, scripts, requirements는 앱 동작의 본체"], ["SQLite DB", "필수", "업무 데이터와 사용자 검토/매칭/설정이 data.db에 존재. 코드만으로 복구 불가"], ["WAL/SHM", "조건부 필수", "실행 중 복사라면 data.db-wal 변경분 누락 위험. 서버 중지 또는 checkpoint/backup 필요"], ["DB dump", "필수 또는 강력 권장", "새 환경 복원 검증용. 현재 dump.sql은 0 bytes라 유효하지 않음"], ["원천 Excel/WEHAGO_DB", "강력 권장", "재검증/재처리/비교 로직 개선 시 기준 자료"], ["사용자 검토/수동매칭", "필수", "wehago_recheck_reviews, wehago_manual_pair_matches 등은 재생성 어려움"], ["캐시 테이블", "선택", "속도에는 도움되지만 구조 개선 후 재생성 가능. DB 슬림화 대상"], ["backups", "선별", "최신 정상본과 구조 변경 직전본 위주로 보존"], [".venv/__pycache__", "불필요", "새 WSL에서 재생성"], ], [2200, 1600, CONTENT_WIDTH - 3800], )) p.append(para("10. 최종 권고", style="Heading1")) p.append(bullet("아키텍처 개선의 1순위는 기능 추가보다 책임 분리와 DB/캐시 관리 기준 정립입니다.")) p.append(bullet("WSL 제로 세팅을 하더라도 목표는 '코드 이관'이 아니라 '동일 업무 상태를 복원 가능한 형태로 보존'하는 것입니다.")) p.append(bullet("구조 개선 시작 전에는 유효한 SQLite 백업/dump를 새로 만들고, 원천파일과 사용자 검토 데이터의 보존 여부를 반드시 확인해야 합니다.")) p.append(para("Appendix. 관찰된 현재 상태", style="Heading1")) p.append(bullet("data.db 약 14GB, data.db-wal 약 183MB, dump.sql 0 bytes 상태를 확인했습니다.")) p.append(bullet("현재 git working tree에는 기존 수정 파일과 미추적 파일이 존재합니다. 구조 변경 전 기준점을 별도로 고정하는 것이 좋습니다.")) p.append(bullet("검토 파일: requirements.txt, main.py, wehago_compare.py, templates/*.html, data.db sqlite_master schema.")) sect = f'' return "".join(p) + sect def styles_xml() -> str: return """ """ def write_docx() -> None: created = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") document = f""" {body_xml()}""" files = { "[Content_Types].xml": """""", "_rels/.rels": """""", "word/_rels/document.xml.rels": """""", "word/document.xml": document, "word/styles.xml": styles_xml(), "word/settings.xml": """""", "word/media/architecture.png": architecture_png(), "word/media/dataflow.png": flow_png(), "docProps/core.xml": f"""my-intranet-app 아키텍처 분석 보고서CodexCodex{created}{created}""", "docProps/app.xml": """Codex OOXML Generator""", } REPORT_DIR.mkdir(exist_ok=True) with zipfile.ZipFile(OUTPUT_PATH, "w", compression=zipfile.ZIP_DEFLATED) as docx: for name, content in files.items(): docx.writestr(name, content) if __name__ == "__main__": write_docx() print(OUTPUT_PATH)