64c27c3509
G-1: 메신저 Webhook Relay + _send_to_room 실제 httpx 호출 구현 G-2: POST /api/tasks/bulk SR 대량작업 엔드포인트 (최대 100건) G-3: 라이선스 만료 알림 스케줄러 (매일 09:00 KST) G-4: 체험판 upgrade_banner 필드 + license.py 배너 로직 G-5: core/auto_rca.py + incidents/problem auto-rca 엔드포인트 G-6: core/deploy_impact.py + vibe impact-analysis 엔드포인트 G-7: core/ticket_classifier.py + SR 생성 시 AI 분류 + ai-suggestion API G-8: VulnPatchRecord 모델 + vuln_scan 패치추적 4개 엔드포인트 G-9: core/jira_sync.py + gateway Jira/Confluence 연동 엔드포인트 G-10: core/push_notify.py + routers/push.py + PushSubscription 모델 G-11: approvals 다중승인 (위임/서명/기한초과/마감연장) G-12: alembic.ini + migrations/ + cicd/migrate_to_postgres.sh 하네스: guardia-orchestrator 확장기능 Phase 반영 봇명령어: /sr /status /license /bulk 슬래시 명령어 추가 설치스크립트: setup/ (Ubuntu, CentOS, RHEL, Windows) --test 옵션 포함 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
553 lines
21 KiB
Python
553 lines
21 KiB
Python
"""
|
|
정기 PM (예방정비) 관리 API.
|
|
|
|
엔드포인트:
|
|
Templates:
|
|
GET /api/pm/templates — 체크리스트 템플릿 목록
|
|
POST /api/pm/templates — 템플릿 생성
|
|
PATCH /api/pm/templates/{id} — 템플릿 수정
|
|
DELETE /api/pm/templates/{id} — 템플릿 삭제
|
|
|
|
Schedules:
|
|
GET /api/pm/schedules — PM 반복 스케줄 목록
|
|
POST /api/pm/schedules — 스케줄 등록
|
|
PATCH /api/pm/schedules/{id} — 스케줄 수정
|
|
DELETE /api/pm/schedules/{id} — 스케줄 삭제
|
|
POST /api/pm/schedules/{id}/trigger — 즉시 실행 (WorkTimetable 생성)
|
|
|
|
Results:
|
|
GET /api/pm/results/{timetable_id} — 점검 결과 목록
|
|
POST /api/pm/results/{timetable_id}/init — 체크리스트 초기화 (템플릿 → 결과 행 생성)
|
|
PATCH /api/pm/results/{result_id} — 단일 결과 업데이트
|
|
GET /api/pm/results/{timetable_id}/report — Excel 보고서 다운로드
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import logging
|
|
from datetime import datetime, date
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy import select, and_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.auth import get_current_user
|
|
from database import get_db
|
|
from models import (
|
|
Institution, Server, WorkTimetable,
|
|
PmFrequency, PmItemResult,
|
|
PmTemplate, PmTemplateCreate, PmTemplateOut,
|
|
PmSchedule, PmScheduleCreate, PmScheduleOut,
|
|
PmResult, PmResultOut, PmResultUpdate,
|
|
User, UserRole,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/pm", tags=["pm"])
|
|
|
|
RESULT_COLOR = {
|
|
"PASS": "C6EFCE",
|
|
"FAIL": "FFC7CE",
|
|
"WARNING": "FFEB9C",
|
|
"NA": "F4F4F4",
|
|
}
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PM 체크리스트 템플릿
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/templates", response_model=List[PmTemplateOut])
|
|
async def list_templates(
|
|
server_role: Optional[str] = Query(None),
|
|
category: Optional[str] = Query(None),
|
|
active_only: bool = Query(True),
|
|
db: AsyncSession = Depends(get_db),
|
|
_u: User = Depends(get_current_user),
|
|
):
|
|
"""체크리스트 템플릿 목록. server_role / category 필터 지원."""
|
|
q = select(PmTemplate)
|
|
if active_only:
|
|
q = q.where(PmTemplate.is_active.is_(True))
|
|
if server_role:
|
|
q = q.where(PmTemplate.server_role == server_role)
|
|
if category:
|
|
q = q.where(PmTemplate.category == category)
|
|
q = q.order_by(PmTemplate.server_role, PmTemplate.category, PmTemplate.item_order)
|
|
result = await db.execute(q)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/templates", response_model=PmTemplateOut, status_code=201)
|
|
async def create_template(
|
|
payload: PmTemplateCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
tmpl = PmTemplate(**payload.model_dump())
|
|
db.add(tmpl)
|
|
await db.commit()
|
|
await db.refresh(tmpl)
|
|
return tmpl
|
|
|
|
|
|
@router.patch("/templates/{template_id}", response_model=PmTemplateOut)
|
|
async def update_template(
|
|
template_id: int,
|
|
payload: PmTemplateCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
r = await db.execute(select(PmTemplate).where(PmTemplate.id == template_id))
|
|
tmpl = r.scalars().first()
|
|
if not tmpl:
|
|
raise HTTPException(404, "템플릿을 찾을 수 없습니다.")
|
|
for k, v in payload.model_dump(exclude_unset=True).items():
|
|
setattr(tmpl, k, v)
|
|
await db.commit()
|
|
await db.refresh(tmpl)
|
|
return tmpl
|
|
|
|
|
|
@router.delete("/templates/{template_id}", status_code=204)
|
|
async def delete_template(
|
|
template_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
r = await db.execute(select(PmTemplate).where(PmTemplate.id == template_id))
|
|
tmpl = r.scalars().first()
|
|
if not tmpl:
|
|
raise HTTPException(404, "템플릿을 찾을 수 없습니다.")
|
|
tmpl.is_active = False # 소프트 삭제
|
|
await db.commit()
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PM 반복 스케줄
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/schedules", response_model=List[PmScheduleOut])
|
|
async def list_schedules(
|
|
inst_id: Optional[int] = Query(None),
|
|
active_only: bool = Query(True),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
q = select(PmSchedule)
|
|
if active_only:
|
|
q = q.where(PmSchedule.is_active.is_(True))
|
|
if inst_id:
|
|
q = q.where(PmSchedule.inst_id == inst_id)
|
|
|
|
if current_user.role == UserRole.CUSTOMER and current_user.inst_code:
|
|
r_i = await db.execute(
|
|
select(Institution).where(Institution.inst_code == current_user.inst_code)
|
|
)
|
|
own = r_i.scalars().first()
|
|
if own:
|
|
q = q.where(PmSchedule.inst_id == own.id)
|
|
|
|
q = q.order_by(PmSchedule.next_scheduled)
|
|
result = await db.execute(q)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/schedules", response_model=PmScheduleOut, status_code=201)
|
|
async def create_schedule(
|
|
payload: PmScheduleCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
|
|
if payload.frequency == PmFrequency.CUSTOM and not payload.cron_expr:
|
|
raise HTTPException(422, "CUSTOM 주기에는 cron_expr이 필요합니다.")
|
|
|
|
sched = PmSchedule(
|
|
**payload.model_dump(),
|
|
created_by=current_user.username,
|
|
)
|
|
# next_scheduled 초기 계산
|
|
from core.scheduler import _calc_next
|
|
sched.next_scheduled = _calc_next(sched, datetime.now())
|
|
|
|
db.add(sched)
|
|
await db.commit()
|
|
await db.refresh(sched)
|
|
return sched
|
|
|
|
|
|
@router.patch("/schedules/{schedule_id}", response_model=PmScheduleOut)
|
|
async def update_schedule(
|
|
schedule_id: int,
|
|
payload: PmScheduleCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
r = await db.execute(select(PmSchedule).where(PmSchedule.id == schedule_id))
|
|
sched = r.scalars().first()
|
|
if not sched:
|
|
raise HTTPException(404, "스케줄을 찾을 수 없습니다.")
|
|
for k, v in payload.model_dump(exclude_unset=True).items():
|
|
setattr(sched, k, v)
|
|
sched.updated_at = datetime.now()
|
|
await db.commit()
|
|
await db.refresh(sched)
|
|
return sched
|
|
|
|
|
|
@router.delete("/schedules/{schedule_id}", status_code=204)
|
|
async def delete_schedule(
|
|
schedule_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "ADMIN 또는 PM 권한이 필요합니다.")
|
|
r = await db.execute(select(PmSchedule).where(PmSchedule.id == schedule_id))
|
|
sched = r.scalars().first()
|
|
if not sched:
|
|
raise HTTPException(404, "스케줄을 찾을 수 없습니다.")
|
|
sched.is_active = False
|
|
await db.commit()
|
|
|
|
|
|
@router.post("/schedules/{schedule_id}/trigger", status_code=201)
|
|
async def trigger_pm_now(
|
|
schedule_id: int,
|
|
scheduled_at: Optional[datetime] = Query(None, description="예정 일시 (미입력 시 즉시)"),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""PM 스케줄을 즉시 또는 지정 일시로 WorkTimetable에 등록."""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
|
|
raise HTTPException(403, "권한이 없습니다.")
|
|
|
|
r = await db.execute(select(PmSchedule).where(PmSchedule.id == schedule_id))
|
|
sched = r.scalars().first()
|
|
if not sched:
|
|
raise HTTPException(404, "스케줄을 찾을 수 없습니다.")
|
|
|
|
target_dt = scheduled_at or datetime.now()
|
|
wt = WorkTimetable(
|
|
work_type = "PM",
|
|
title = f"{sched.schedule_name} PM (수동 실행)",
|
|
inst_id = sched.inst_id,
|
|
server_id = sched.server_id,
|
|
scheduled_at = target_dt,
|
|
content = (
|
|
f"PM 수동 실행\n"
|
|
f"스케줄: {sched.schedule_name} ({sched.frequency})\n"
|
|
f"담당자: {sched.assignee or '미지정'}"
|
|
),
|
|
result_status = "PENDING",
|
|
assignee = sched.assignee,
|
|
reviewer = sched.reviewer,
|
|
created_by = current_user.username,
|
|
)
|
|
db.add(wt)
|
|
sched.last_generated = datetime.now()
|
|
await db.commit()
|
|
await db.refresh(wt)
|
|
|
|
# 템플릿 ID가 설정된 경우 PmResult 행 초기화
|
|
timetable_id = wt.id
|
|
if sched.template_ids:
|
|
template_id_list = [
|
|
int(t.strip()) for t in sched.template_ids.split(",") if t.strip().isdigit()
|
|
]
|
|
if template_id_list:
|
|
tmpl_result = await db.execute(
|
|
select(PmTemplate).where(
|
|
PmTemplate.id.in_(template_id_list),
|
|
PmTemplate.is_active.is_(True),
|
|
).order_by(PmTemplate.item_order)
|
|
)
|
|
templates = tmpl_result.scalars().all()
|
|
for tmpl in templates:
|
|
pr = PmResult(
|
|
timetable_id = timetable_id,
|
|
template_id = tmpl.id,
|
|
item_title = tmpl.item_title,
|
|
item_desc = tmpl.item_desc,
|
|
check_command = tmpl.check_command,
|
|
result = PmItemResult.NA,
|
|
)
|
|
db.add(pr)
|
|
await db.commit()
|
|
|
|
return {
|
|
"timetable_id": timetable_id,
|
|
"scheduled_at": target_dt.isoformat(),
|
|
"message": f"PM WorkTimetable 생성 완료 (ID: {timetable_id})",
|
|
}
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PM 점검 결과 (체크리스트)
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/results/{timetable_id}", response_model=List[PmResultOut])
|
|
async def get_results(
|
|
timetable_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_u: User = Depends(get_current_user),
|
|
):
|
|
"""타임테이블 ID에 해당하는 PM 체크리스트 항목 목록."""
|
|
result = await db.execute(
|
|
select(PmResult)
|
|
.where(PmResult.timetable_id == timetable_id)
|
|
.order_by(PmResult.id)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/results/{timetable_id}/init", status_code=201)
|
|
async def init_checklist(
|
|
timetable_id: int,
|
|
server_role: Optional[str] = Query(None, description="템플릿 필터 — 서버 역할"),
|
|
category: Optional[str] = Query(None, description="템플릿 필터 — 카테고리"),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
템플릿에서 PmResult 행을 일괄 초기화.
|
|
이미 결과가 있으면 건너뜀.
|
|
"""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
|
|
raise HTTPException(403, "권한이 없습니다.")
|
|
|
|
# WorkTimetable 확인
|
|
wt = (await db.execute(
|
|
select(WorkTimetable).where(WorkTimetable.id == timetable_id)
|
|
)).scalars().first()
|
|
if not wt:
|
|
raise HTTPException(404, "타임테이블을 찾을 수 없습니다.")
|
|
|
|
# 기존 결과 있으면 건너뜀
|
|
existing = (await db.execute(
|
|
select(PmResult).where(PmResult.timetable_id == timetable_id)
|
|
)).scalars().all()
|
|
if existing:
|
|
return {"created": 0, "message": "이미 체크리스트가 초기화되어 있습니다.", "total": len(existing)}
|
|
|
|
# 템플릿 조회
|
|
q = select(PmTemplate).where(PmTemplate.is_active.is_(True))
|
|
if server_role:
|
|
q = q.where(PmTemplate.server_role == server_role)
|
|
if category:
|
|
q = q.where(PmTemplate.category == category)
|
|
q = q.order_by(PmTemplate.item_order)
|
|
templates = (await db.execute(q)).scalars().all()
|
|
|
|
for tmpl in templates:
|
|
pr = PmResult(
|
|
timetable_id = timetable_id,
|
|
template_id = tmpl.id,
|
|
item_title = tmpl.item_title,
|
|
item_desc = tmpl.item_desc,
|
|
check_command = tmpl.check_command,
|
|
result = PmItemResult.NA,
|
|
)
|
|
db.add(pr)
|
|
await db.commit()
|
|
|
|
return {
|
|
"created": len(templates),
|
|
"message": f"{len(templates)}개 체크리스트 항목이 초기화되었습니다.",
|
|
"total": len(templates),
|
|
}
|
|
|
|
|
|
@router.patch("/results/{result_id}", response_model=PmResultOut)
|
|
async def update_result(
|
|
result_id: int,
|
|
payload: PmResultUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""단일 PM 점검 항목 결과 업데이트."""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
|
|
raise HTTPException(403, "권한이 없습니다.")
|
|
|
|
r = await db.execute(select(PmResult).where(PmResult.id == result_id))
|
|
item = r.scalars().first()
|
|
if not item:
|
|
raise HTTPException(404, "점검 결과 항목을 찾을 수 없습니다.")
|
|
|
|
for k, v in payload.model_dump(exclude_unset=True).items():
|
|
setattr(item, k, v)
|
|
item.checked_by = current_user.username
|
|
item.checked_at = datetime.now()
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.get("/results/{timetable_id}/report")
|
|
async def download_report(
|
|
timetable_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""PM 점검 보고서 Excel 다운로드."""
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
|
|
raise HTTPException(403, "권한이 없습니다.")
|
|
|
|
try:
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
|
from openpyxl.utils import get_column_letter
|
|
except ImportError:
|
|
raise HTTPException(500, "openpyxl 라이브러리가 필요합니다. pip install openpyxl")
|
|
|
|
# 타임테이블 정보
|
|
wt = (await db.execute(
|
|
select(WorkTimetable).where(WorkTimetable.id == timetable_id)
|
|
)).scalars().first()
|
|
if not wt:
|
|
raise HTTPException(404, "타임테이블을 찾을 수 없습니다.")
|
|
|
|
results = (await db.execute(
|
|
select(PmResult).where(PmResult.timetable_id == timetable_id).order_by(PmResult.id)
|
|
)).scalars().all()
|
|
|
|
# 기관/서버명 조회
|
|
inst_name = ""
|
|
if wt.inst_id:
|
|
inst = (await db.execute(
|
|
select(Institution).where(Institution.id == wt.inst_id)
|
|
)).scalars().first()
|
|
if inst:
|
|
inst_name = inst.inst_name
|
|
|
|
server_name = ""
|
|
if wt.server_id:
|
|
srv = (await db.execute(
|
|
select(Server).where(Server.id == wt.server_id)
|
|
)).scalars().first()
|
|
if srv:
|
|
server_name = srv.server_name
|
|
|
|
# Excel 생성
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "PM 점검결과"
|
|
ws.freeze_panes = "A3"
|
|
|
|
hdr_fill = PatternFill("solid", fgColor="1E3A5F")
|
|
hdr_font = Font(name="맑은 고딕", bold=True, color="FFFFFF", size=10)
|
|
thin = Side(style="thin", color="CCCCCC")
|
|
bdr = Border(left=thin, right=thin, top=thin, bottom=thin)
|
|
|
|
# 타이틀 행
|
|
ws.merge_cells("A1:H1")
|
|
title_cell = ws["A1"]
|
|
title_cell.value = (
|
|
f"GUARDiA ITSM — PM 점검 결과 | {wt.title} | "
|
|
f"{wt.scheduled_at.strftime('%Y-%m-%d') if wt.scheduled_at else ''}"
|
|
)
|
|
title_cell.font = Font(name="맑은 고딕", bold=True, size=12)
|
|
title_cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
ws.row_dimensions[1].height = 25
|
|
|
|
# 헤더
|
|
headers = [
|
|
("No", 5), ("카테고리", 14), ("점검 항목", 35), ("확인 명령어", 30),
|
|
("기준값", 18), ("실제값", 20), ("결과", 10), ("비고", 25),
|
|
]
|
|
for ci, (h, w) in enumerate(headers, 1):
|
|
cell = ws.cell(2, ci, h)
|
|
cell.font = hdr_font; cell.fill = hdr_fill
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
cell.border = bdr
|
|
ws.column_dimensions[get_column_letter(ci)].width = w
|
|
ws.row_dimensions[2].height = 20
|
|
|
|
# 데이터 행
|
|
for ri, item in enumerate(results, 3):
|
|
tmpl_cat = ""
|
|
if item.template_id:
|
|
tmpl_obj = (await db.execute(
|
|
select(PmTemplate).where(PmTemplate.id == item.template_id)
|
|
)).scalars().first()
|
|
if tmpl_obj:
|
|
tmpl_cat = tmpl_obj.category or ""
|
|
|
|
row_vals = [
|
|
ri - 2,
|
|
tmpl_cat,
|
|
item.item_title,
|
|
item.check_command or "",
|
|
(await db.execute(
|
|
select(PmTemplate.expected_value).where(PmTemplate.id == item.template_id)
|
|
)).scalar_one_or_none() if item.template_id else "",
|
|
item.actual_value or "",
|
|
item.result or "NA",
|
|
item.result_note or "",
|
|
]
|
|
result_fill_color = RESULT_COLOR.get(item.result or "NA", "F4F4F4")
|
|
for ci, val in enumerate(row_vals, 1):
|
|
cell = ws.cell(ri, ci, val)
|
|
cell.border = bdr
|
|
cell.alignment = Alignment(
|
|
horizontal="center" if ci in (1, 7) else "left",
|
|
vertical="center",
|
|
wrap_text=True,
|
|
)
|
|
if ci == 7:
|
|
cell.fill = PatternFill("solid", fgColor=result_fill_color)
|
|
|
|
# 통계
|
|
ws.append([])
|
|
pass_cnt = sum(1 for r in results if r.result == PmItemResult.PASS)
|
|
fail_cnt = sum(1 for r in results if r.result == PmItemResult.FAIL)
|
|
warn_cnt = sum(1 for r in results if r.result == PmItemResult.WARNING)
|
|
na_cnt = sum(1 for r in results if r.result == PmItemResult.NA)
|
|
summary_row = ws.max_row + 1
|
|
ws.cell(summary_row, 1, "합계")
|
|
ws.cell(summary_row, 2, f"총 {len(results)}건")
|
|
ws.cell(summary_row, 3, f"PASS: {pass_cnt} FAIL: {fail_cnt} WARNING: {warn_cnt} NA: {na_cnt}")
|
|
for ci in range(1, 4):
|
|
ws.cell(summary_row, ci).font = Font(name="맑은 고딕", bold=True)
|
|
|
|
# 메타 시트
|
|
ws2 = wb.create_sheet("점검정보")
|
|
meta = [
|
|
("타임테이블 ID", timetable_id),
|
|
("작업명", wt.title),
|
|
("기관", inst_name),
|
|
("서버", server_name),
|
|
("예정일", wt.scheduled_at.strftime("%Y-%m-%d") if wt.scheduled_at else ""),
|
|
("담당자", wt.assignee or ""),
|
|
("검토자", wt.reviewer or ""),
|
|
("결과상태", wt.result_status),
|
|
]
|
|
for r, (k, v) in enumerate(meta, 1):
|
|
ws2.cell(r, 1, k).font = Font(bold=True)
|
|
ws2.cell(r, 2, str(v))
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
|
|
from urllib.parse import quote
|
|
fname = f"PM점검결과_{wt.scheduled_at.strftime('%Y%m%d') if wt.scheduled_at else 'NO_DATE'}.xlsx"
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(fname)}"},
|
|
)
|