""" SI 프로젝트 보고서 자동 생성 엔진 지원 형식: - Excel (.xlsx): WBS 현황 + 이슈 목록 + 산출물 현황 + KPI - HTML: 웹 보고서 (대시보드 내 표시) - PDF: HTML → PDF 변환 (weasyprint) - DOCX: Word 보고서 (python-docx) - PPTX: PowerPoint 보고서 (python-pptx) 보고서 유형: - daily: 일일 진행 현황 (당일 WBS 완료 + 이슈 발생) - weekly: 주간 보고서 (완료율 + 이슈 + 위험 요약) - monthly: 월간 보고서 (KPI + 예산 + 이달의 산출물) """ from __future__ import annotations import io import logging import os from datetime import date, datetime, timedelta from typing import Any, Dict, Optional logger = logging.getLogger(__name__) # ── 프로젝트 데이터 수집 ────────────────────────────────────────────────────── async def collect_project_data(project_id: int, db) -> Dict[str, Any]: """보고서에 필요한 전체 프로젝트 데이터 수집.""" from models import ( SiProject, WbsItem, WbsStatus, ProjectIssue, IssueStatus, ProjectMilestone, ProjectRisk, RiskLevel, Deliverable, DeliverableStatus, ) from sqlalchemy import select, func proj = await db.get(SiProject, project_id) if not proj: raise ValueError(f"프로젝트 {project_id}를 찾을 수 없습니다.") today = date.today() # WBS 항목 wbs_items = (await db.execute( select(WbsItem).where(WbsItem.project_id == project_id) .order_by(WbsItem.wbs_code) )).scalars().all() leaf_items = [w for w in wbs_items if w.is_leaf] completed = [w for w in leaf_items if w.completion_pct >= 100] delayed = [w for w in leaf_items if w.planned_end and w.planned_end < today and w.completion_pct < 100] # 이슈 issues = (await db.execute( select(ProjectIssue).where(ProjectIssue.project_id == project_id) .order_by(ProjectIssue.raised_date.desc()) )).scalars().all() open_issues = [i for i in issues if i.status in (IssueStatus.OPEN, IssueStatus.IN_PROGRESS)] closed_issues = [i for i in issues if i.status == IssueStatus.CLOSED] # 마일스톤 milestones = (await db.execute( select(ProjectMilestone).where(ProjectMilestone.project_id == project_id) .order_by(ProjectMilestone.target_date) )).scalars().all() upcoming_milestones = [ m for m in milestones if m.target_date and m.target_date >= today ][:3] # 위험 risks = (await db.execute( select(ProjectRisk).where(ProjectRisk.project_id == project_id) )).scalars().all() high_risks = [r for r in risks if r.risk_level in (RiskLevel.HIGH, RiskLevel.CRITICAL)] # 산출물 deliverables = (await db.execute( select(Deliverable).where(Deliverable.project_id == project_id) )).scalars().all() overdue_deliverables = [ d for d in deliverables if d.status == "PENDING" and d.due_date and d.due_date < today ] # 진척률 계산 overall_progress = ( sum(w.completion_pct for w in leaf_items) / len(leaf_items) if leaf_items else 0 ) # 예산 소진율 budget_pct = ( round(proj.budget_used / proj.budget_total * 100, 1) if proj.budget_total else 0.0 ) return { "project": proj, "today": today, "wbs_items": wbs_items, "leaf_items": leaf_items, "completed_wbs": completed, "delayed_wbs": delayed, "issues": issues, "open_issues": open_issues, "closed_issues": closed_issues, "milestones": milestones, "upcoming_milestones": upcoming_milestones, "risks": risks, "high_risks": high_risks, "deliverables": deliverables, "overdue_deliverables": overdue_deliverables, "overall_progress": round(overall_progress, 1), "budget_pct": budget_pct, "wbs_total": len(leaf_items), "wbs_done": len(completed), "wbs_delayed": len(delayed), "issue_open": len(open_issues), "issue_closed": len(closed_issues), } # ── Excel 보고서 ────────────────────────────────────────────────────────────── def generate_excel(data: Dict[str, Any], report_type: str = "weekly") -> bytes: """Excel (.xlsx) 보고서 생성.""" try: import openpyxl from openpyxl.styles import ( Font, PatternFill, Alignment, Border, Side, numbers ) from openpyxl.utils import get_column_letter except ImportError: raise ValueError("openpyxl 미설치: pip install openpyxl") proj = data["project"] today = data["today"] wb = openpyxl.Workbook() wb.remove(wb.active) # ── 공통 스타일 ──────────────────────────────────────────── HEADER_FILL = PatternFill("solid", fgColor="1E3A5F") HEADER_FONT = Font(bold=True, color="FFFFFF", size=11) TITLE_FONT = Font(bold=True, size=13, color="1E3A5F") OK_FILL = PatternFill("solid", fgColor="D5F5E3") WARN_FILL = PatternFill("solid", fgColor="FEF9E7") DANGER_FILL = PatternFill("solid", fgColor="FADBD8") CENTER = Alignment(horizontal="center", vertical="center", wrap_text=True) LEFT = Alignment(horizontal="left", vertical="center", wrap_text=True) thin = Side(style="thin", color="CCCCCC") BORDER = Border(left=thin, right=thin, top=thin, bottom=thin) def header_row(ws, headers, row=1): for col, h in enumerate(headers, 1): c = ws.cell(row=row, column=col, value=h) c.font = HEADER_FONT; c.fill = HEADER_FILL c.alignment = CENTER; c.border = BORDER def data_row(ws, values, row, fills=None): for col, val in enumerate(values, 1): c = ws.cell(row=row, column=col, value=val) c.alignment = LEFT; c.border = BORDER if fills and col-1 < len(fills) and fills[col-1]: c.fill = fills[col-1] # ── 시트 1: 프로젝트 개요 ───────────────────────────────── ws1 = wb.create_sheet("프로젝트 개요") ws1.column_dimensions["A"].width = 22 ws1.column_dimensions["B"].width = 35 ws1["A1"] = f"📋 {proj.project_name}" ws1["A1"].font = TITLE_FONT ws1.merge_cells("A1:D1") kpi_rows = [ ("프로젝트 코드", proj.project_code), ("현재 단계", proj.phase), ("전체 진척률", f"{data['overall_progress']}%"), ("건강 상태", proj.health_status), ("계획 기간", f"{proj.planned_start} ~ {proj.planned_end}"), ("PM", proj.pm_user or "—"), ("예산 소진율", f"{data['budget_pct']}%"), ("WBS 완료 수", f"{data['wbs_done']} / {data['wbs_total']}"), ("미결 이슈", f"{data['issue_open']}건"), ("지연 WBS", f"{data['wbs_delayed']}건"), ("보고일", str(today)), ] for r, (k, v) in enumerate(kpi_rows, 3): ws1.cell(row=r, column=1, value=k).font = Font(bold=True) ws1.cell(row=r, column=2, value=v) # ── 시트 2: WBS 현황 ────────────────────────────────────── ws2 = wb.create_sheet("WBS 현황") for col, w in zip("ABCDEFG", [10, 40, 20, 15, 15, 12, 25]): ws2.column_dimensions[col].width = w header_row(ws2, ["WBS 코드", "제목", "단계", "예정 시작", "예정 완료", "진척률(%)", "상태"]) for r, item in enumerate(data["wbs_items"], 2): status = "완료" if item.completion_pct >= 100 else ( "지연" if item.planned_end and item.planned_end < today and item.completion_pct < 100 else "진행중" ) fill = OK_FILL if status == "완료" else (DANGER_FILL if status == "지연" else None) fills = [None, fill, None, None, None, fill, fill] data_row(ws2, [ item.wbs_code, item.title, item.phase or "", str(item.planned_start or ""), str(item.planned_end or ""), item.completion_pct, status, ], r, fills) # ── 시트 3: 이슈 목록 ───────────────────────────────────── ws3 = wb.create_sheet("이슈 관리") for col, w in zip("ABCDEFG", [18, 15, 35, 20, 12, 18, 30]): ws3.column_dimensions[col].width = w header_row(ws3, ["이슈 ID", "유형", "제목", "담당자", "심각도", "상태", "발생일"]) for r, iss in enumerate(data["issues"], 2): severity_fill = DANGER_FILL if getattr(iss, "severity", "") in ("CRITICAL", "HIGH") else None data_row(ws3, [ iss.issue_id, iss.issue_type, iss.title, iss.assigned_to or "—", getattr(iss, "severity", "—"), iss.status, str(iss.raised_date or ""), ], r, [None, None, None, None, severity_fill, None, None]) # ── 시트 4: 산출물 현황 ─────────────────────────────────── ws4 = wb.create_sheet("산출물 현황") for col, w in zip("ABCDEF", [30, 15, 15, 15, 15, 20]): ws4.column_dimensions[col].width = w header_row(ws4, ["산출물명", "유형", "버전", "상태", "제출기한", "제출일"]) for r, dlv in enumerate(data["deliverables"], 2): status_fill = ( OK_FILL if dlv.status == "APPROVED" else DANGER_FILL if dlv.status == "REJECTED" else WARN_FILL if dlv.status == "REVIEWING" else DANGER_FILL if dlv.due_date and dlv.due_date < today and dlv.status == "PENDING" else None ) data_row(ws4, [ dlv.name, dlv.deliverable_type, dlv.version, dlv.status, str(dlv.due_date or ""), str(dlv.submitted_at.date() if dlv.submitted_at else "—"), ], r, [None, None, None, status_fill, None, None]) # ── 시트 5: 위험 관리 ───────────────────────────────────── ws5 = wb.create_sheet("위험 관리") for col, w in zip("ABCDE", [18, 35, 12, 12, 40]): ws5.column_dimensions[col].width = w header_row(ws5, ["위험 ID", "제목", "레벨", "상태", "대응 계획"]) for r, risk in enumerate(data["risks"], 2): fill = DANGER_FILL if getattr(risk, "risk_level", "") in ("HIGH", "CRITICAL") else WARN_FILL data_row(ws5, [ risk.risk_id, risk.title, getattr(risk, "risk_level", "—"), risk.status, risk.mitigation or "", ], r, [None, None, fill, None, None]) buf = io.BytesIO() wb.save(buf) return buf.getvalue() # ── HTML 보고서 ────────────────────────────────────────────────────────────── def generate_html(data: Dict[str, Any], report_type: str = "weekly") -> str: """HTML 보고서 생성 (대시보드 표시 + PDF 변환용).""" try: from jinja2 import Template except ImportError: raise ValueError("jinja2 미설치: pip install jinja2") proj = data["project"] today = data["today"] template_str = """

{{ report_title }}

프로젝트: {{ proj.project_name }} ({{ proj.project_code }}) | 보고일: {{ today }} | 단계: {{ proj.phase }}

📊 KPI 현황

{{ data.overall_progress }}%
전체 진척률
{{ data.wbs_done }}/{{ data.wbs_total }}
WBS 완료
{{ data.wbs_delayed }}
지연 항목
{{ data.issue_open }}
미결 이슈
{{ data.budget_pct }}%
예산 소진율
{{ proj.health_status }}
건강 상태

📋 WBS 지연 현황 ({{ data.wbs_delayed }}건)

{% if data.delayed_wbs %} {% for item in data.delayed_wbs %} {% endfor %}
WBS 코드제목예정 완료진척률
{{ item.wbs_code }} {{ item.title }} {{ item.planned_end }} {{ item.completion_pct }}%
{% else %}

✅ 지연 WBS 없음

{% endif %}

🔴 미결 이슈 ({{ data.issue_open }}건)

{% if data.open_issues %} {% for iss in data.open_issues[:10] %} {% endfor %}
이슈 ID제목유형담당자발생일
{{ iss.issue_id }} {{ iss.title }} {{ iss.issue_type }} {{ iss.assigned_to or '—' }} {{ iss.raised_date }}
{% else %}

✅ 미결 이슈 없음

{% endif %}

📦 산출물 제출 현황

{% for dlv in data.deliverables %} {% endfor %}
산출물명유형버전상태제출기한
{{ dlv.name }} {{ dlv.deliverable_type }} {{ dlv.version }} {{ dlv.status }} {{ dlv.due_date or '—' }}
{% if data.upcoming_milestones %}

🏁 다가오는 마일스톤

{% for m in data.upcoming_milestones %} {% endfor %}
마일스톤목표일상태
{{ m.name }}{{ m.target_date }}{{ m.status }}
{% endif %} {% if data.high_risks %}

⚠️ 고위험 요소 ({{ data.high_risks|length }}건)

{% for r in data.high_risks %} {% endfor %}
위험 ID제목레벨대응 계획
{{ r.risk_id }} {{ r.title }} {{ r.risk_level }} {{ r.mitigation or '—' }}
{% endif %}

생성: GUARDiA ITSM | {{ today }}

""" title_map = { "daily": f"일일 보고서 ({today})", "weekly": f"주간 보고서 ({today} 기준)", "monthly": f"월간 보고서 ({today.strftime('%Y년 %m월')})", } t = Template(template_str) return t.render( proj=proj, data=data, today=today, report_title=title_map.get(report_type, "보고서"), ) # ── PDF 보고서 ─────────────────────────────────────────────────────────────── def generate_pdf(html_content: str) -> bytes: """HTML → PDF 변환.""" try: import weasyprint pdf_bytes = weasyprint.HTML(string=html_content).write_pdf() return pdf_bytes except ImportError: logger.warning("weasyprint 미설치 — HTML 반환") return html_content.encode("utf-8") except Exception as e: logger.error("PDF 변환 오류: %s", e) return html_content.encode("utf-8") # ── DOCX 보고서 ────────────────────────────────────────────────────────────── def generate_docx(data: Dict[str, Any], report_type: str = "weekly") -> bytes: """Word (.docx) 보고서 생성.""" try: from docx import Document from docx.shared import Pt, RGBColor, Cm from docx.enum.text import WD_ALIGN_PARAGRAPH except ImportError: raise ValueError("python-docx 미설치: pip install python-docx") proj = data["project"] today = data["today"] doc = Document() # 제목 rtype_kor = {"daily": "일일", "weekly": "주간", "monthly": "월간"}.get(report_type, "") title = doc.add_heading( f"{proj.project_name} — {rtype_kor}보고서", level=0 ) title.alignment = WD_ALIGN_PARAGRAPH.CENTER doc.add_paragraph(f"보고일: {today} | 단계: {proj.phase} | PM: {proj.pm_user or '—'}") # KPI 테이블 doc.add_heading("KPI 현황", level=1) kpi_table = doc.add_table(rows=1, cols=4) kpi_table.style = "Table Grid" for cell, text in zip(kpi_table.rows[0].cells, ["항목", "계획", "실적", "상태"]): cell.text = text kpi_rows = [ ("전체 진척률", "—", f"{data['overall_progress']}%", "양호" if data['overall_progress'] >= 80 else "주의"), ("WBS 완료율", f"{data['wbs_total']}건", f"{data['wbs_done']}건 ({round(data['wbs_done']/data['wbs_total']*100 if data['wbs_total'] else 0, 1)}%)", "양호" if data['wbs_delayed'] == 0 else "지연"), ("예산 소진율", "100%", f"{data['budget_pct']}%", "양호" if data['budget_pct'] <= 80 else "주의"), ("미결 이슈", "0건", f"{data['issue_open']}건", "양호" if data['issue_open'] == 0 else "주의"), ] for row_data in kpi_rows: row = kpi_table.add_row() for cell, val in zip(row.cells, row_data): cell.text = val # 지연 WBS if data["delayed_wbs"]: doc.add_heading(f"WBS 지연 현황 ({data['wbs_delayed']}건)", level=1) t = doc.add_table(rows=1, cols=4) t.style = "Table Grid" for c, h in zip(t.rows[0].cells, ["WBS 코드", "제목", "예정 완료", "진척률"]): c.text = h for item in data["delayed_wbs"]: r = t.add_row() for c, v in zip(r.cells, [item.wbs_code, item.title, str(item.planned_end), f"{item.completion_pct}%"]): c.text = v # 이슈 if data["open_issues"]: doc.add_heading(f"미결 이슈 ({data['issue_open']}건)", level=1) t = doc.add_table(rows=1, cols=4) t.style = "Table Grid" for c, h in zip(t.rows[0].cells, ["이슈 ID", "제목", "담당자", "발생일"]): c.text = h for iss in data["open_issues"][:10]: r = t.add_row() for c, v in zip(r.cells, [iss.issue_id, iss.title, iss.assigned_to or "—", str(iss.raised_date or "")]): c.text = v doc.add_paragraph(f"\n생성: GUARDiA ITSM | {today}") buf = io.BytesIO() doc.save(buf) return buf.getvalue() # ── PPTX 보고서 ────────────────────────────────────────────────────────────── def generate_pptx(data: Dict[str, Any], report_type: str = "weekly") -> bytes: """PowerPoint (.pptx) 보고서 생성.""" try: from pptx import Presentation from pptx.util import Inches, Pt, Emu from pptx.dml.color import RGBColor from pptx.enum.text import PP_ALIGN except ImportError: raise ValueError("python-pptx 미설치: pip install python-pptx") proj = data["project"] today = data["today"] prs = Presentation() prs.slide_width = Inches(13.33) prs.slide_height = Inches(7.5) DARK_BLUE = RGBColor(0x1E, 0x3A, 0x5F) WHITE = RGBColor(0xFF, 0xFF, 0xFF) def add_slide(layout_idx=5): layout = prs.slide_layouts[layout_idx] return prs.slides.add_slide(layout) def set_bg(slide, color=RGBColor(0xF7, 0xFA, 0xFC)): from pptx.util import Emu fill = slide.background.fill fill.solid() fill.fore_color.rgb = color def add_text_box(slide, text, left, top, width, height, font_size=14, bold=False, color=None): txBox = slide.shapes.add_textbox(Inches(left), Inches(top), Inches(width), Inches(height)) tf = txBox.text_frame tf.word_wrap = True p = tf.paragraphs[0] run = p.add_run() run.text = text run.font.size = Pt(font_size) run.font.bold = bold if color: run.font.color.rgb = color return txBox # ── 슬라이드 1: 표지 ────────────────────────────────────── slide1 = add_slide(6) # blank set_bg(slide1, DARK_BLUE) add_text_box(slide1, proj.project_name, 1, 2, 11, 1.5, font_size=32, bold=True, color=WHITE) rtype_name = {"daily": "일일", "weekly": "주간", "monthly": "월간"}.get(report_type, "") add_text_box(slide1, f"{rtype_name} 보고서", 1, 3.8, 11, 0.8, font_size=20, color=RGBColor(0xBD, 0xE3, 0xFF)) add_text_box(slide1, f"보고일: {today} | 단계: {proj.phase}", 1, 5, 11, 0.6, font_size=14, color=RGBColor(0xA0, 0xC4, 0xFF)) # ── 슬라이드 2: KPI 요약 ────────────────────────────────── slide2 = add_slide(6) set_bg(slide2) add_text_box(slide2, "KPI 현황", 0.5, 0.3, 12, 0.6, font_size=22, bold=True, color=DARK_BLUE) kpis = [ ("전체 진척률", f"{data['overall_progress']}%"), ("WBS 완료", f"{data['wbs_done']}/{data['wbs_total']}"), ("지연 항목", f"{data['wbs_delayed']}건"), ("미결 이슈", f"{data['issue_open']}건"), ("예산 소진", f"{data['budget_pct']}%"), ("건강 상태", proj.health_status), ] for i, (label, value) in enumerate(kpis): x = 0.5 + (i % 3) * 4.2 y = 1.5 + (i // 3) * 2.2 box = slide2.shapes.add_shape( 1, Inches(x), Inches(y), Inches(3.8), Inches(1.8) ) box.fill.solid(); box.fill.fore_color.rgb = RGBColor(0xEB, 0xF8, 0xFF) box.line.color.rgb = RGBColor(0x90, 0xCF, 0xF8) tf = box.text_frame tf.word_wrap = True p1 = tf.paragraphs[0] p1.alignment = PP_ALIGN.CENTER r1 = p1.add_run(); r1.text = value r1.font.size = Pt(28); r1.font.bold = True r1.font.color.rgb = DARK_BLUE p2 = tf.add_paragraph() p2.alignment = PP_ALIGN.CENTER r2 = p2.add_run(); r2.text = label r2.font.size = Pt(12); r2.font.color.rgb = RGBColor(0x71, 0x80, 0x96) # ── 슬라이드 3: WBS 현황 ────────────────────────────────── if data["delayed_wbs"]: slide3 = add_slide(6) set_bg(slide3) add_text_box(slide3, f"WBS 지연 현황 ({data['wbs_delayed']}건)", 0.5, 0.3, 12, 0.6, 22, True, DARK_BLUE) from pptx.util import Inches as I tbl = slide3.shapes.add_table( min(len(data["delayed_wbs"]) + 1, 8), 4, I(0.5), I(1.2), I(12), I(4.5) ).table headers = ["WBS 코드", "제목", "예정 완료", "진척률"] for col, h in enumerate(headers): tbl.cell(0, col).text = h for r, item in enumerate(data["delayed_wbs"][:6], 1): vals = [item.wbs_code, item.title[:30], str(item.planned_end), f"{item.completion_pct}%"] for col, v in enumerate(vals): tbl.cell(r, col).text = v # ── 슬라이드 4: 이슈 요약 ───────────────────────────────── if data["open_issues"]: slide4 = add_slide(6) set_bg(slide4) add_text_box(slide4, f"미결 이슈 ({data['issue_open']}건)", 0.5, 0.3, 12, 0.6, 22, True, DARK_BLUE) from pptx.util import Inches as I tbl = slide4.shapes.add_table( min(len(data["open_issues"]) + 1, 8), 4, I(0.5), I(1.2), I(12), I(4.5) ).table headers = ["이슈 ID", "제목", "유형", "담당자"] for col, h in enumerate(headers): tbl.cell(0, col).text = h for r, iss in enumerate(data["open_issues"][:6], 1): vals = [iss.issue_id, iss.title[:30], iss.issue_type, iss.assigned_to or "—"] for col, v in enumerate(vals): tbl.cell(r, col).text = v buf = io.BytesIO() prs.save(buf) return buf.getvalue() # ── 보고서 유형 통합 생성 ───────────────────────────────────────────────────── async def generate_report( project_id: int, report_type: str, # daily | weekly | monthly output_fmt: str, # excel | html | pdf | docx | pptx db, ) -> tuple[bytes, str, str]: """ 보고서 생성 통합 함수. Returns: (content_bytes, media_type, filename) """ data = await collect_project_data(project_id, db) proj = data["project"] today = data["today"] rtype = {"daily": "일일", "weekly": "주간", "monthly": "월간"}.get(report_type, report_type) base_name = f"{proj.project_code}_{rtype}보고서_{today}" if output_fmt == "excel": content = generate_excel(data, report_type) mime = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" filename = f"{base_name}.xlsx" elif output_fmt == "html": content = generate_html(data, report_type).encode("utf-8") mime = "text/html; charset=utf-8" filename = f"{base_name}.html" elif output_fmt == "pdf": html_str = generate_html(data, report_type) content = generate_pdf(html_str) mime = "application/pdf" filename = f"{base_name}.pdf" elif output_fmt == "docx": content = generate_docx(data, report_type) mime = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" filename = f"{base_name}.docx" elif output_fmt == "pptx": content = generate_pptx(data, report_type) mime = "application/vnd.openxmlformats-officedocument.presentationml.presentation" filename = f"{base_name}.pptx" else: raise ValueError(f"지원하지 않는 형식: {output_fmt}") return content, mime, filename