d1ed53cbd7
- Enum 값을 소문자로 변경 (material_missing, design_error, incoming_defect, complete) - nonconformity_name -> description 필드명 변경 - completion_date -> actual_completion_date 필드명 변경 - Excel 다운로드 시 한글 파일명 URL 인코딩 추가 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
361 lines
13 KiB
Python
361 lines
13 KiB
Python
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)
|