from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from sqlalchemy import desc from io import BytesIO from datetime import datetime, timedelta from urllib.parse import quote from typing import Optional import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from database import get_db import models from auth import require_admin router = APIRouter(prefix="/api/export", tags=["export"]) NAVY = "0B1E3D" LIGHT = "D6EAF8" GREEN = "1B5E20" ORANGE = "E65100" PURPLE = "4A148C" TEAL = "004D40" def _hdr_cell(ws, row, col, value, color=NAVY): bd = Side(style="thin", color="AAAAAA") c = ws.cell(row=row, column=col, value=value) c.font = Font(bold=True, color="FFFFFF", size=10) c.fill = PatternFill("solid", fgColor=color) c.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) c.border = Border(left=bd, right=bd, top=bd, bottom=bd) def style_header(ws, headers, row=1, color=NAVY): for col, h in enumerate(headers, 1): _hdr_cell(ws, row, col, h, color) ws.row_dimensions[row].height = 20 def style_row(ws, row_num, num_cols, even=True): bd = Side(style="thin", color="DDDDDD") for col in range(1, num_cols + 1): cell = ws.cell(row=row_num, column=col) if even: cell.fill = PatternFill("solid", fgColor="F4F7FB") cell.border = Border(left=bd, right=bd, top=bd, bottom=bd) cell.alignment = Alignment(vertical="center", wrap_text=True) def fmt_dt(dt): return dt.strftime("%Y-%m-%d %H:%M") if dt else "" def fmt_d(d): return str(d) if d else "" def elapsed(start, end): if not start or not end: return "" diff = end - start total = int(diff.total_seconds()) if total < 0: return "" h, m = divmod(total // 60, 60) return f"{h}시간 {m}분" def set_col_widths(ws, widths): for i, w in enumerate(widths, 1): ws.column_dimensions[ws.cell(1, i).column_letter].width = w def make_response(wb: openpyxl.Workbook, korean_name: str) -> StreamingResponse: buf = BytesIO() wb.save(buf) buf.seek(0) date_str = datetime.now().strftime("%Y%m%d_%H%M") filename = f"{korean_name}_{date_str}.xlsx" encoded = quote(filename, safe="") cd = f"attachment; filename*=UTF-8''{encoded}" return StreamingResponse( buf, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": cd}, ) def _parse_dates(date_from, date_to): try: dt_from = datetime.strptime(date_from, "%Y-%m-%d") if date_from else None dt_to = (datetime.strptime(date_to, "%Y-%m-%d") + timedelta(days=1)) if date_to else None except ValueError: raise HTTPException(400, "날짜 형식 오류 (YYYY-MM-DD)") return dt_from, dt_to # ───────────────────────────────────────────── # 시트 빌더 — 공용 # ───────────────────────────────────────────── CLOSURE_LABEL = { "natural": "증상자연소거", "remote_reset": "원격리셋후증상소거", "false_alarm": "인지오류", "other": "기타", } SOURCE_LABEL = { "qr": "QR스캔", "admin": "관리자접수", "dashboard": "대시보드접수", } COST_STATUS_LABEL = { "pending": "미처리", "billed": "청구완료", "waived": "면제", "settled": "정산완료", } PARTY_LABEL = { "cpo": "CPO(운영사)", "manufacturer": "업체", "self": "자체부담", "user": "사용자과실", "other": "기타", } IMP_STATUS_LABEL = { "registered": "등록", "reviewing": "검토중", "developing": "개발중", "deployed": "배포완료", "done": "완료", } RESULT_LABEL = { "done": "완료", "in_progress": "진행중", "waiting": "부품대기", "revisit": "재방문", } def _ws_reports(wb, db, dt_from, dt_to): ws = wb.create_sheet("AS신고이력") ws.freeze_panes = "A2" headers = [ "신고번호", "충전기ID", "충전기종류", "충전기명", "충전소명", "CPO", "문제유형", "에러코드", "상세설명", "신고자연락처", "문제발생시각", "신고일시", "신고출처", "신고자", "처리상태", "담당정비사", "소속", "조치유형", "조치내용", "조치시작", "조치완료", "작업소요시간", "재조치횟수", "문제원인", "비고", "출장비부담", "출장비금액(원)", "출장비상태", "상황종료사유", "상황종료일시", ] style_header(ws, headers, color=NAVY) set_col_widths(ws, [10,14,14,14,18,12,22,12,26,14,16,16,10,14,12,12,12,16,26,16,16,12,8,24,24,14,12,12,18,16]) q = db.query(models.Report) if dt_from: q = q.filter(models.Report.reported_at >= dt_from) if dt_to: q = q.filter(models.Report.reported_at < dt_to) rows = q.order_by(desc(models.Report.reported_at)).all() for rn, r in enumerate(rows, 2): c = r.charger repair = r.repair_links[0].repair if r.repair_links else None cost = repair.cost if repair else None ws.append([ r.id, r.charger_id, c.charger_type.name if c and c.charger_type else "", c.name if c else "", c.station_name if c else "", c.cpo_name if c else "", ", ".join(r.issue_types or []), r.error_code or "", r.issue_detail or "", r.contact or "", fmt_dt(r.occurred_at), fmt_dt(r.reported_at), SOURCE_LABEL.get(r.source or "qr", r.source or ""), r.reporter.name if r.reporter else "", r.status, repair.mechanic.name if repair and repair.mechanic else "", repair.mechanic.company if repair and repair.mechanic else "", ", ".join(repair.repair_types or []) if repair else "", repair.description if repair else "", fmt_dt(repair.started_at) if repair else "", fmt_dt(repair.completed_at) if repair else "", elapsed(repair.started_at, repair.completed_at) if repair else "", r.re_dispatch_count or 0, cost.root_cause if cost else "", cost.admin_note if cost else "", PARTY_LABEL.get(cost.cost_party_type, cost.cost_party_type or "") if cost else "", cost.cost_amount if cost else "", COST_STATUS_LABEL.get(cost.cost_status, cost.cost_status or "") if cost else "", CLOSURE_LABEL.get(r.closure_type or "", ""), fmt_dt(r.closed_at), ]) style_row(ws, rn, len(headers), rn % 2 == 0) ws.row_dimensions[rn].height = 16 return len(rows) def _ws_repairs(wb, db, dt_from, dt_to): ws = wb.create_sheet("조치이력") ws.freeze_panes = "A2" headers = [ "조치번호", "연결신고번호", "충전기ID", "충전소명", "충전기종류", "정비사", "소속", "조치유형", "조치내용", "시작시각", "완료시각", "소요시간", "처리결과", "재조치요청", "승인완료", "승인자", "승인일시", ] style_header(ws, headers, color=GREEN) set_col_widths(ws, [10,16,14,18,14,12,14,18,30,16,16,12,10,10,10,12,16]) q = db.query(models.Repair) if dt_from: q = q.filter(models.Repair.completed_at >= dt_from) if dt_to: q = q.filter(models.Repair.completed_at < dt_to) rows = q.order_by(desc(models.Repair.completed_at)).all() for rn, rep in enumerate(rows, 2): rids = [rr.report_id for rr in rep.report_links] charger_id = station = ctype = "" if rids: r = db.query(models.Report).filter_by(id=rids[0]).first() if r and r.charger: charger_id = r.charger_id station = r.charger.station_name or "" ctype = r.charger.charger_type.name if r.charger.charger_type else "" ws.append([ rep.id, ", ".join(str(i) for i in rids), charger_id, station, ctype, rep.mechanic.name if rep.mechanic else "", rep.mechanic.company if rep.mechanic else "", ", ".join(rep.repair_types or []), rep.description or "", fmt_dt(rep.started_at), fmt_dt(rep.completed_at), elapsed(rep.started_at, rep.completed_at), RESULT_LABEL.get(rep.result_status or "", rep.result_status or ""), "예" if rep.re_dispatch_requested else "아니오", "예" if rep.approved_at else "아니오", rep.approver.name if rep.approver else "", fmt_dt(rep.approved_at), ]) style_row(ws, rn, len(headers), rn % 2 == 0) ws.row_dimensions[rn].height = 16 return len(rows) def _ws_improvements(wb, db, dt_from, dt_to): ws = wb.create_sheet("개선항목") ws.freeze_panes = "A2" headers = [ "번호", "제목", "분류", "우선순위", "개선내용", "관련부품", "담당업체", "담당자(대표)", "연락처", "연결AS건수", "연결AS번호", "진행상태", "SW배포목표일", "SW실제배포일", "제조사메모", "등록자", "등록일시", ] style_header(ws, headers, color=PURPLE) set_col_widths(ws, [8,26,10,10,32,14,16,14,14,10,20,12,14,14,26,12,16]) q = db.query(models.Improvement) if dt_from: q = q.filter(models.Improvement.created_at >= dt_from) if dt_to: q = q.filter(models.Improvement.created_at < dt_to) rows = q.order_by(desc(models.Improvement.created_at)).all() for rn, imp in enumerate(rows, 2): rids = [ir.report_id for ir in imp.report_links] ws.append([ imp.id, imp.title, imp.category, imp.priority, imp.description, imp.part_name or "", imp.manufacturer.name if imp.manufacturer else "", imp.manufacturer.representative_name if imp.manufacturer else "", imp.manufacturer.phone if imp.manufacturer else "", len(rids), ", ".join(str(i) for i in rids), IMP_STATUS_LABEL.get(imp.status, imp.status), fmt_d(imp.sw_deploy_target), fmt_d(imp.sw_deployed_at), imp.manufacturer_memo or "", imp.creator.name if imp.creator else "", fmt_dt(imp.created_at), ]) style_row(ws, rn, len(headers), rn % 2 == 0) ws.row_dimensions[rn].height = 16 return len(rows) def _ws_costs(wb, db, dt_from, dt_to): ws = wb.create_sheet("출장비정산") ws.freeze_panes = "A2" headers = [ "신고번호", "충전기ID", "충전소명", "충전기종류", "정비사", "소속", "조치완료일", "문제원인", "비고", "부담주체유형", "부담업체명", "부담기타", "수급주체유형", "수급업체명", "수급기타", "금액(원)", "처리상태", "처리담당자", "처리일시", ] style_header(ws, headers, color=ORANGE) set_col_widths(ws, [14,14,18,14,12,14,16,26,26,12,16,16,12,16,16,12,12,12,16]) q = db.query(models.RepairCost).join(models.Repair) if dt_from: q = q.filter(models.Repair.completed_at >= dt_from) if dt_to: q = q.filter(models.Repair.completed_at < dt_to) rows = q.order_by(desc(models.Repair.completed_at)).all() for rn, cost in enumerate(rows, 2): repair = cost.repair rids = [rr.report_id for rr in repair.report_links] charger_id = station = ctype = "" if rids: r = db.query(models.Report).filter_by(id=rids[0]).first() if r and r.charger: charger_id = r.charger_id station = r.charger.station_name or "" ctype = r.charger.charger_type.name if r.charger.charger_type else "" ws.append([ ", ".join(str(i) for i in rids), charger_id, station, ctype, repair.mechanic.name if repair.mechanic else "", repair.mechanic.company if repair.mechanic else "", fmt_dt(repair.completed_at), cost.root_cause or "", cost.admin_note or "", PARTY_LABEL.get(cost.cost_party_type or "", cost.cost_party_type or ""), cost.cost_manufacturer.name if cost.cost_manufacturer else "", cost.cost_party_custom or "", PARTY_LABEL.get(cost.recv_party_type or "", cost.recv_party_type or ""), cost.recv_manufacturer.name if cost.recv_manufacturer else "", cost.recv_party_custom or "", cost.cost_amount or 0, COST_STATUS_LABEL.get(cost.cost_status or "", cost.cost_status or ""), cost.reviewer.name if cost.reviewer else "", fmt_dt(cost.reviewed_at), ]) style_row(ws, rn, len(headers), rn % 2 == 0) ws.row_dimensions[rn].height = 16 return len(rows) def _ws_summary(wb, counts, dt_from, dt_to, date_from, date_to): ws = wb.create_sheet("요약", 0) # 맨 앞에 삽입 ws.sheet_view.showGridLines = False period = f"{date_from or '전체'} ~ {date_to or '전체'}" def lbl(row, col, text, bold=False, size=11, color="1A2B4A"): c = ws.cell(row=row, column=col, value=text) c.font = Font(bold=bold, size=size, color=color) c.alignment = Alignment(vertical="center") def val(row, col, text, color="1A2B4A", bold=True): c = ws.cell(row=row, column=col, value=text) c.font = Font(bold=bold, size=13, color=color) c.alignment = Alignment(vertical="center", horizontal="center") ws.row_dimensions[1].height = 16 ws.row_dimensions[2].height = 36 ws.row_dimensions[3].height = 14 c = ws.cell(row=2, column=2, value="EV AS 관리 통합 이력") c.font = Font(bold=True, size=18, color=NAVY) c.alignment = Alignment(vertical="center") ws.cell(row=2, column=5, value=f"조회 기간: {period}").font = Font(size=11, color="666666") sheet_names = ["AS신고이력", "조치이력", "개선항목", "출장비정산"] colors = [NAVY, GREEN, PURPLE, ORANGE] labels = ["AS 신고", "조치 이력", "개선항목", "출장비 정산"] for i, (name, color, label, cnt) in enumerate(zip(sheet_names, colors, labels, counts)): row = 5 + i * 3 ws.row_dimensions[row].height = 24 ws.row_dimensions[row+1].height = 22 cell = ws.cell(row=row, column=2, value=f"● {label}") cell.font = Font(bold=True, size=12, color=color) cell.alignment = Alignment(vertical="center") ws.cell(row=row, column=3, value=f"{cnt}건").font = Font(bold=True, size=14, color=color) ws.cell(row=row+1, column=2, value=f'자세한 내용은 "{name}" 시트를 확인하세요').font = Font(size=9, color="888888") ws.column_dimensions["A"].width = 4 ws.column_dimensions["B"].width = 22 ws.column_dimensions["C"].width = 14 ws.column_dimensions["D"].width = 10 ws.column_dimensions["E"].width = 30 # ───────────────────────────────────────────── # 1. AS 신고 목록 (개별) # ───────────────────────────────────────────── @router.get("/reports") def export_reports( date_from: Optional[str] = None, date_to: Optional[str] = None, db: Session = Depends(get_db), _=Depends(require_admin) ): dt_from, dt_to = _parse_dates(date_from, date_to) wb = openpyxl.Workbook() ws = wb.active ws.title = "AS신고목록" ws.freeze_panes = "A2" headers = [ "접수번호","충전기ID","충전기종류","충전기명","충전소명","CPO명","설치일", "신고위치(위도)","신고위치(경도)","문제유형","에러코드","상세설명", "신고자연락처","문제발생시각","신고일시","신고출처","신고자","처리상태", "담당정비사","정비사소속","조치유형","조치내용", "조치시작","조치완료","작업소요시간","신고→완료소요시간", "문제원인(관리자)","비고","출장비부담주체","출장비금액(원)","출장비상태", "처리담당자","처리일시","연결개선항목번호", "상황종료사유","상황종료메모","상황종료일시","상황종료처리자", "재조치횟수" ] style_header(ws, headers) set_col_widths(ws, [10,14,14,14,18,14,12,12,12,22,12,24,14,16,16,10,16,12, 12,14,16,24,16,16,12,18,24,24,16,12,12,12,16,18, 18,24,16,14,10]) q = db.query(models.Report) if dt_from: q = q.filter(models.Report.reported_at >= dt_from) if dt_to: q = q.filter(models.Report.reported_at < dt_to) reports = q.order_by(desc(models.Report.reported_at)).all() for row_num, r in enumerate(reports, 2): c = r.charger repair = r.repair_links[0].repair if r.repair_links else None cost = repair.cost if repair else None imp_ids = [ ir.improvement_id for ir in db.query(models.ImprovementReport).filter_by(report_id=r.id).all() ] row_data = [ r.id, r.charger_id, c.charger_type.name if c and c.charger_type else "", c.name if c else "", c.station_name if c else "", c.cpo_name if c else "", fmt_d(c.installed_at) if c else "", r.gps_lat or "", r.gps_lng or "", ", ".join(r.issue_types or []), r.error_code or "", r.issue_detail or "", r.contact or "", fmt_dt(r.occurred_at), fmt_dt(r.reported_at), SOURCE_LABEL.get(r.source or "qr", r.source or ""), r.reporter.name if r.reporter else "", r.status, repair.mechanic.name if repair and repair.mechanic else "", repair.mechanic.company if repair and repair.mechanic else "", ", ".join(repair.repair_types or []) if repair else "", repair.description if repair else "", fmt_dt(repair.started_at) if repair else "", fmt_dt(repair.completed_at) if repair else "", elapsed(repair.started_at, repair.completed_at) if repair else "", elapsed(r.occurred_at or r.reported_at, repair.completed_at if repair else None), cost.root_cause if cost else "", cost.admin_note if cost else "", PARTY_LABEL.get(cost.cost_party_type or "", cost.cost_party_type or "") if cost else "", cost.cost_amount if cost else "", COST_STATUS_LABEL.get(cost.cost_status or "", cost.cost_status or "") if cost else "", cost.reviewer.name if cost and cost.reviewer else "", fmt_dt(cost.reviewed_at) if cost else "", ", ".join(str(i) for i in imp_ids) if imp_ids else "", CLOSURE_LABEL.get(r.closure_type or "", ""), r.closure_note or "", fmt_dt(r.closed_at), r.closer.name if r.closer else "", r.re_dispatch_count or 0, ] for col, val in enumerate(row_data, 1): ws.cell(row=row_num, column=col, value=val) style_row(ws, row_num, len(headers), row_num % 2 == 0) ws.row_dimensions[row_num].height = 16 return make_response(wb, "AS신고목록") # ───────────────────────────────────────────── # 2. 출장비 목록 (개별) # ───────────────────────────────────────────── @router.get("/costs") def export_costs( date_from: Optional[str] = None, date_to: Optional[str] = None, db: Session = Depends(get_db), _=Depends(require_admin) ): dt_from, dt_to = _parse_dates(date_from, date_to) wb = openpyxl.Workbook() ws = wb.active ws.title = "출장비목록" ws.freeze_panes = "A2" headers = [ "신고번호","충전기ID","충전기종류","충전소명","조치완료일", "정비사","소속","문제원인","비고", "부담주체유형","부담업체명","부담기타", "수급주체유형","수급업체명","수급기타", "금액(원)","처리상태","처리담당자","처리일시" ] style_header(ws, headers) set_col_widths(ws, [14,14,14,18,16,12,14,26,26,12,16,16,12,16,16,12,12,12,16]) q = db.query(models.RepairCost).join(models.Repair) if dt_from: q = q.filter(models.Repair.completed_at >= dt_from) if dt_to: q = q.filter(models.Repair.completed_at < dt_to) costs = q.order_by(desc(models.Repair.completed_at)).all() for row_num, cost in enumerate(costs, 2): repair = cost.repair rids = [rr.report_id for rr in repair.report_links] charger_id = station = ctype = "" if rids: r = db.query(models.Report).filter_by(id=rids[0]).first() if r and r.charger: charger_id = r.charger_id station = r.charger.station_name or "" ctype = r.charger.charger_type.name if r.charger.charger_type else "" row_data = [ ", ".join(str(i) for i in rids), charger_id, ctype, station, fmt_dt(repair.completed_at), repair.mechanic.name if repair.mechanic else "", repair.mechanic.company if repair.mechanic else "", cost.root_cause or "", cost.admin_note or "", PARTY_LABEL.get(cost.cost_party_type or "", cost.cost_party_type or ""), cost.cost_manufacturer.name if cost.cost_manufacturer else "", cost.cost_party_custom or "", PARTY_LABEL.get(cost.recv_party_type or "", cost.recv_party_type or ""), cost.recv_manufacturer.name if cost.recv_manufacturer else "", cost.recv_party_custom or "", cost.cost_amount or 0, COST_STATUS_LABEL.get(cost.cost_status or "", cost.cost_status or ""), cost.reviewer.name if cost.reviewer else "", fmt_dt(cost.reviewed_at), ] for col, val in enumerate(row_data, 1): ws.cell(row=row_num, column=col, value=val) style_row(ws, row_num, len(headers), row_num % 2 == 0) ws.row_dimensions[row_num].height = 16 return make_response(wb, "출장비목록") # ───────────────────────────────────────────── # 3. 개선항목 목록 (개별) # ───────────────────────────────────────────── @router.get("/improvements") def export_improvements( date_from: Optional[str] = None, date_to: Optional[str] = None, db: Session = Depends(get_db), _=Depends(require_admin) ): dt_from, dt_to = _parse_dates(date_from, date_to) wb = openpyxl.Workbook() ws = wb.active ws.title = "개선항목목록" ws.freeze_panes = "A2" headers = [ "번호","제목","분류","우선순위","개선내용","관련부품", "담당업체","담당자(대표)","연락처","연결AS건수","연결AS번호", "진행상태","SW배포목표일","SW실제배포일","제조사메모", "등록관리자","등록일시" ] style_header(ws, headers) set_col_widths(ws, [8,26,10,10,32,14,16,14,14,10,20,12,14,14,26,12,16]) q = db.query(models.Improvement) if dt_from: q = q.filter(models.Improvement.created_at >= dt_from) if dt_to: q = q.filter(models.Improvement.created_at < dt_to) imps = q.order_by(desc(models.Improvement.created_at)).all() for row_num, imp in enumerate(imps, 2): rids = [ir.report_id for ir in imp.report_links] row_data = [ imp.id, imp.title, imp.category, imp.priority, imp.description, imp.part_name or "", imp.manufacturer.name if imp.manufacturer else "", imp.manufacturer.representative_name if imp.manufacturer else "", imp.manufacturer.phone if imp.manufacturer else "", len(rids), ", ".join(str(i) for i in rids), IMP_STATUS_LABEL.get(imp.status, imp.status), fmt_d(imp.sw_deploy_target), fmt_d(imp.sw_deployed_at), imp.manufacturer_memo or "", imp.creator.name if imp.creator else "", fmt_dt(imp.created_at), ] for col, val in enumerate(row_data, 1): ws.cell(row=row_num, column=col, value=val) style_row(ws, row_num, len(headers), row_num % 2 == 0) ws.row_dimensions[row_num].height = 16 return make_response(wb, "개선항목목록") # ───────────────────────────────────────────── # 4. 통합 다운로드 (멀티 시트) # ───────────────────────────────────────────── @router.get("/full") def export_full( date_from: Optional[str] = None, date_to: Optional[str] = None, db: Session = Depends(get_db), _=Depends(require_admin) ): dt_from, dt_to = _parse_dates(date_from, date_to) wb = openpyxl.Workbook() wb.remove(wb.active) # 기본 빈 시트 제거 c1 = _ws_reports(wb, db, dt_from, dt_to) c2 = _ws_repairs(wb, db, dt_from, dt_to) c3 = _ws_improvements(wb, db, dt_from, dt_to) c4 = _ws_costs(wb, db, dt_from, dt_to) _ws_summary(wb, [c1, c2, c3, c4], dt_from, dt_to, date_from, date_to) period = f"{date_from or '전체'}~{date_to or '전체'}" return make_response(wb, f"EV_AS_통합이력_{period}")