from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from sqlalchemy import func, and_, or_ from datetime import datetime, date from typing import List import io from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from database.database import get_db from database.models import Issue, DailyWork, IssueStatus, IssueCategory, User, UserRole, Project, ReviewStatus from database import schemas from routers.auth import get_current_user router = APIRouter(prefix="/api/reports", tags=["reports"]) @router.post("/summary", response_model=schemas.ReportSummary) async def generate_report_summary( report_request: schemas.ReportRequest, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """보고서 요약 생성""" start_date = report_request.start_date end_date = report_request.end_date # 일일 공수 합계 daily_works = db.query(DailyWork).filter( DailyWork.date >= start_date.date(), DailyWork.date <= end_date.date() ).all() total_hours = sum(w.total_hours for w in daily_works) # 이슈 통계 issues_query = db.query(Issue).filter( Issue.report_date >= start_date, Issue.report_date <= end_date ) # 일반 사용자는 자신의 이슈만 if current_user.role == UserRole.user: issues_query = issues_query.filter(Issue.reporter_id == current_user.id) issues = issues_query.all() # 카테고리별 통계 category_stats = schemas.CategoryStats() completed_issues = 0 total_resolution_time = 0 resolved_count = 0 for issue in issues: # 카테고리별 카운트 if issue.category == IssueCategory.material_missing: category_stats.material_missing += 1 elif issue.category == IssueCategory.design_error: category_stats.dimension_defect += 1 elif issue.category == IssueCategory.incoming_defect: category_stats.incoming_defect += 1 # 완료된 이슈 if issue.status == IssueStatus.complete: completed_issues += 1 if issue.work_hours > 0: total_resolution_time += issue.work_hours resolved_count += 1 # 평균 해결 시간 average_resolution_time = total_resolution_time / resolved_count if resolved_count > 0 else 0 return schemas.ReportSummary( start_date=start_date, end_date=end_date, total_hours=total_hours, total_issues=len(issues), category_stats=category_stats, completed_issues=completed_issues, average_resolution_time=average_resolution_time ) @router.get("/issues") async def get_report_issues( start_date: datetime, end_date: datetime, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """보고서용 이슈 상세 목록""" query = db.query(Issue).filter( Issue.report_date >= start_date, Issue.report_date <= end_date ) # 일반 사용자는 자신의 이슈만 if current_user.role == UserRole.user: query = query.filter(Issue.reporter_id == current_user.id) issues = query.order_by(Issue.report_date.desc()).all() return [{ "id": issue.id, "photo_path": issue.photo_path, "category": issue.category, "description": issue.description, "status": issue.status, "reporter_name": issue.reporter.full_name or issue.reporter.username, "report_date": issue.report_date, "work_hours": issue.work_hours, "detail_notes": issue.detail_notes } for issue in issues] @router.get("/daily-works") async def get_report_daily_works( start_date: datetime, end_date: datetime, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """보고서용 일일 공수 목록""" works = db.query(DailyWork).filter( DailyWork.date >= start_date.date(), DailyWork.date <= end_date.date() ).order_by(DailyWork.date).all() return [{ "date": work.date, "worker_count": work.worker_count, "regular_hours": work.regular_hours, "overtime_workers": work.overtime_workers, "overtime_hours": work.overtime_hours, "overtime_total": work.overtime_total, "total_hours": work.total_hours } for work in works] @router.post("/daily-export") async def export_daily_report( request: schemas.DailyReportRequest, current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ): """품질팀용 일일보고서 엑셀 내보내기""" # 권한 확인 (품질팀만 접근 가능) if current_user.role != UserRole.admin: raise HTTPException(status_code=403, detail="품질팀만 접근 가능합니다") # 프로젝트 확인 project = db.query(Project).filter(Project.id == request.project_id).first() if not project: raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다") # 관리함 데이터 조회 (진행 중 + 완료됨) issues_query = db.query(Issue).filter( Issue.project_id == request.project_id, Issue.review_status.in_([ReviewStatus.in_progress, ReviewStatus.completed]) ).order_by(Issue.report_date.desc()) issues = issues_query.all() # 통계 계산 stats = calculate_project_stats(issues) # 엑셀 파일 생성 wb = Workbook() ws = wb.active ws.title = "일일보고서" # 스타일 정의 header_font = Font(bold=True, color="FFFFFF") header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid") stats_font = Font(bold=True, size=12) stats_fill = PatternFill(start_color="E7E6E6", end_color="E7E6E6", fill_type="solid") border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') ) center_alignment = Alignment(horizontal='center', vertical='center') # 제목 및 기본 정보 ws.merge_cells('A1:L1') ws['A1'] = f"{project.project_name} - 일일보고서" ws['A1'].font = Font(bold=True, size=16) ws['A1'].alignment = center_alignment ws.merge_cells('A2:L2') ws['A2'] = f"생성일: {datetime.now().strftime('%Y년 %m월 %d일')}" ws['A2'].alignment = center_alignment # 프로젝트 통계 (4행부터) ws.merge_cells('A4:L4') ws['A4'] = "프로젝트 현황" ws['A4'].font = stats_font ws['A4'].fill = stats_fill ws['A4'].alignment = center_alignment # 통계 데이터 stats_row = 5 ws[f'A{stats_row}'] = "총 신고 수량" ws[f'B{stats_row}'] = stats.total_count ws[f'D{stats_row}'] = "관리처리 현황" ws[f'E{stats_row}'] = stats.management_count ws[f'G{stats_row}'] = "완료 현황" ws[f'H{stats_row}'] = stats.completed_count ws[f'J{stats_row}'] = "지연 중" ws[f'K{stats_row}'] = stats.delayed_count # 통계 스타일 적용 for col in ['A', 'D', 'G', 'J']: ws[f'{col}{stats_row}'].font = Font(bold=True) ws[f'{col}{stats_row}'].fill = PatternFill(start_color="FFF2CC", end_color="FFF2CC", fill_type="solid") # 데이터 테이블 헤더 (7행부터) headers = [ "번호", "프로젝트", "부적합명", "상세내용", "원인분류", "해결방안", "담당부서", "담당자", "마감일", "상태", "신고일", "완료일" ] header_row = 7 for col, header in enumerate(headers, 1): cell = ws.cell(row=header_row, column=col, value=header) cell.font = header_font cell.fill = header_fill cell.alignment = center_alignment cell.border = border # 데이터 입력 current_row = header_row + 1 for issue in issues: # 완료됨 항목의 첫 내보내기 여부 확인 (실제로는 DB에 플래그를 저장해야 함) # 지금은 모든 완료됨 항목을 포함 ws.cell(row=current_row, column=1, value=issue.id) ws.cell(row=current_row, column=2, value=project.project_name) ws.cell(row=current_row, column=3, value=issue.description or "") ws.cell(row=current_row, column=4, value=issue.detail_notes or "") ws.cell(row=current_row, column=5, value=get_category_text(issue.category)) ws.cell(row=current_row, column=6, value=issue.solution or "") ws.cell(row=current_row, column=7, value=get_department_text(issue.responsible_department)) ws.cell(row=current_row, column=8, value=issue.responsible_person or "") ws.cell(row=current_row, column=9, value=issue.expected_completion_date.strftime('%Y-%m-%d') if issue.expected_completion_date else "") ws.cell(row=current_row, column=10, value=get_status_text(issue.review_status)) ws.cell(row=current_row, column=11, value=issue.report_date.strftime('%Y-%m-%d') if issue.report_date else "") ws.cell(row=current_row, column=12, value=issue.actual_completion_date.strftime('%Y-%m-%d') if issue.actual_completion_date else "") # 상태별 색상 적용 status_color = get_status_color(issue.review_status) if status_color: for col in range(1, len(headers) + 1): ws.cell(row=current_row, column=col).fill = PatternFill( start_color=status_color, end_color=status_color, fill_type="solid" ) # 테두리 적용 for col in range(1, len(headers) + 1): ws.cell(row=current_row, column=col).border = border ws.cell(row=current_row, column=col).alignment = Alignment(vertical='center') current_row += 1 # 열 너비 자동 조정 for col in range(1, len(headers) + 1): column_letter = get_column_letter(col) ws.column_dimensions[column_letter].width = 15 # 특정 열 너비 조정 ws.column_dimensions['C'].width = 20 # 부적합명 ws.column_dimensions['D'].width = 30 # 상세내용 ws.column_dimensions['F'].width = 25 # 해결방안 # 엑셀 파일을 메모리에 저장 excel_buffer = io.BytesIO() wb.save(excel_buffer) excel_buffer.seek(0) # 파일명 생성 today = date.today().strftime('%Y%m%d') filename = f"{project.project_name}_일일보고서_{today}.xlsx" # 한글 파일명을 위한 URL 인코딩 from urllib.parse import quote encoded_filename = quote(filename.encode('utf-8')) return StreamingResponse( io.BytesIO(excel_buffer.read()), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={ "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}" } ) def calculate_project_stats(issues: List[Issue]) -> schemas.DailyReportStats: """프로젝트 통계 계산""" stats = schemas.DailyReportStats() today = date.today() for issue in issues: stats.total_count += 1 if issue.review_status == ReviewStatus.in_progress: stats.management_count += 1 # 지연 여부 확인 if issue.expected_completion_date and issue.expected_completion_date < today: stats.delayed_count += 1 elif issue.review_status == ReviewStatus.completed: stats.completed_count += 1 return stats def get_category_text(category: IssueCategory) -> str: """카테고리 한글 변환""" category_map = { IssueCategory.material_missing: "자재 누락", IssueCategory.design_error: "설계 미스", IssueCategory.incoming_defect: "입고 불량", IssueCategory.inspection_miss: "검사 미스", IssueCategory.etc: "기타" } return category_map.get(category, str(category)) def get_department_text(department) -> str: """부서 한글 변환""" if not department: return "" department_map = { "production": "생산", "quality": "품질", "purchasing": "구매", "design": "설계", "sales": "영업" } return department_map.get(department, str(department)) def get_status_text(status: ReviewStatus) -> str: """상태 한글 변환""" status_map = { ReviewStatus.pending_review: "검토 대기", ReviewStatus.in_progress: "진행 중", ReviewStatus.completed: "완료됨", ReviewStatus.disposed: "폐기됨" } return status_map.get(status, str(status)) def get_status_color(status: ReviewStatus) -> str: """상태별 색상 반환""" color_map = { ReviewStatus.in_progress: "FFF2CC", # 연한 노랑 ReviewStatus.completed: "E2EFDA", # 연한 초록 ReviewStatus.disposed: "F2F2F2" # 연한 회색 } return color_map.get(status, None)