안전관리 업무 자동화: 파이썬으로 만드는 보고서

이 글의 목적은 산업현장의 안전관리 데이터를 파이썬으로 수집·정제·분석·시각화하여 일일·주간·월간 보고서를 자동 생성하는 방법을 체계적으로 제시하고, 실무에서 즉시 적용 가능한 모듈 구성과 코드 예시를 제공하는 것이다.

자동화 목표와 시스템 아키텍처 개요

안전관리 보고서 자동화의 목표는 반복되는 수작업을 제거하고 데이터의 정확성과 재현성을 확보하는 데 있다. 핵심은 표준화된 입력, 검증 가능한 처리 로직, 일관된 출력 포맷을 유지하는 것이다. 시스템은 입력 계층, 처리 계층, 출력 계층, 스케줄러, 로깅·감사 모듈로 구성하는 것이 바람직하다.

“같은 입력에는 같은 결과가 나와야 품질이 유지된다.”라는 원칙을 자동화의 기준으로 삼아야 한다.

폴더 구조와 형상관리 전략

현장 적용을 위해서는 명확한 폴더 구조와 형상관리가 필수이다. 다음 구조를 권장한다.

ehs-reporting/
├─ data/
│  ├─ raw/                 # 원천데이터 수신 폴더
│  ├─ interim/             # 전처리 중간산출물
│  └─ processed/           # 검증 완료 데이터셋
├─ reports/
│  ├─ daily/
│  ├─ weekly/
│  └─ monthly/
├─ templates/              # 보고서 템플릿(docx, pptx, html)
├─ src/
│  ├─ config/              # yaml 설정
│  ├─ etl/                 # 수집·전처리
│  ├─ kpi/                 # 지표계산
│  ├─ viz/                 # 시각화
│  ├─ publish/             # 파일출력·메일발송
│  └─ utils/               # 공통함수(로깅, 검증)
├─ tests/                  # 단위테스트
└─ logs/                   # 실행로그 및 감사지표

Git으로 코드와 템플릿을 형상관리하고, 데이터는 개인정보 포함 여부에 따라 접근권한을 분리하는 것이 바람직하다.

데이터 소스 정의와 표준 스키마

안전관리 보고서의 주요 데이터 소스는 다음과 같다.

사고·아차사고 접수 시스템에서 추출한 이벤트 로그이다.
작업환경측정, 설비점검, 교육이수 기록 등 정형 CSV·엑셀 데이터이다.
IoT 센서에서 집계된 알람·노출량 요약값이다.

수집 단계에서 표준 스키마를 강제하여 품질을 확보한다.

필드명	형식	필수	설명
event_id	문자열	예	사고·아차사고 고유식별자이다.
event_dt	날짜시간	예	발생 일시(로컬타임존 기준)이다.
site	문자열	예	사업장 코드이다.
type	카테고리	예	사고유형(전도, 협착, 화학물질 누출 등)이다.
severity	카테고리	예	중상·경상·무상해 등급이다.
lost_time_hr	숫자	아니오	손실시간 합계이다.
desc	문자열	아니오	사건 요약 서술이다.

파이썬 기술 스택 선택 기준

역할	권장 라이브러리	선정 이유
데이터 처리	pandas, polars	대용량 테이블 처리 성능과 풍부한 I/O가 강점이다.
시각화	matplotlib, plotly	정적·인터랙티브 차트 모두 대응 가능하다.
문서 생성	python-docx, python-pptx, openpyxl	사내 양식 준수와 자동 서식 적용이 가능하다.
PDF 렌더	reportlab, wkhtmltopdf 연계	배포 표준 포맷 출력을 지원한다.
스케줄링	cron, Windows Task Scheduler, APScheduler	운영환경 제약을 고려한 선택이 편리하다.
로깅·감사	logging, loguru	레벨별 로그와 회귀분석을 위한 보관이 용이하다.
설정관리	PyYAML, pydantic	환경별 변수와 유효성 검증을 중앙집중화한다.
메일 발송	smtplib, M365·Gmail API	보고서 배포 자동화를 구현한다.

핵심 KPI 정의와 계산 로직

보고서 자동화의 품질은 지표정의의 일관성에 달려 있다. 대표 지표는 다음과 같다.

지표	정의	계산식
TRIR	총 기록재해율이다.	(기록재해 건수 × 200,000) ÷ 총 근로시간이다.
LTIFR	손실시간 재해율이다.	(손실시간 재해 건수 × 1,000,000) ÷ 총 근로시간이다.
Near-Miss Rate	아차사고 보고율이다.	(아차사고 건수 ÷ 총 근로시간) × 200,000이다.
Action Closure	개선조치 완료율이다.	(기간 내 완료 조치 ÷ 전체 조치) × 100이다.
Training Compliance	교육이수 준수율이다.	(필수 이수 인원 ÷ 대상 인원) × 100이다.

ETL 파이프라인 예시 코드

다음 예시는 CSV 데이터에서 월간 KPI를 계산해 엑셀과 PDF로 출력하는 최소 구현 사례이다.

# src/etl/load.py
import pandas as pd
from pathlib import Path

def load_events(path: str) -> pd.DataFrame:
df = pd.read_csv(path, parse_dates=["event_dt"])
df.columns = [c.strip().lower() for c in df.columns]
return df

def load_hours(path: str) -> pd.DataFrame:
return pd.read_excel(path, sheet_name="hours")

# src/kpi/calc.py
import pandas as pd

def monthly_kpi(events: pd.DataFrame, hours: pd.DataFrame, month: str) -> pd.DataFrame:
    e = events[events["event_dt"].dt.strftime("%Y-%m") == month].copy()
    h = hours[hours["month"] == month].copy()
    total_hours = h["work_hours"].sum()

    recordables = e[e["severity"].isin(["중상","경상"])].shape[0]
    losttime = e[e["lost_time_hr"].fillna(0) > 0].shape[0]
    near_miss = e[e["type"] == "아차사고"].shape[0]

    trir = (recordables * 200_000) / total_hours if total_hours else 0
    ltifr = (losttime * 1_000_000) / total_hours if total_hours else 0
    nmr = (near_miss * 200_000) / total_hours if total_hours else 0

    return pd.DataFrame([{
        "month": month,
        "total_hours": total_hours,
        "recordables": recordables,
        "losttime": losttime,
        "near_miss": near_miss,
        "TRIR": round(trir, 2),
        "LTIFR": round(ltifr, 2),
        "NearMissRate": round(nmr, 2)
    }])

# src/publish/excel_out.py
import pandas as pd
from openpyxl import Workbook
from openpyxl.utils.dataframe import dataframe_to_rows

def to_excel(df: pd.DataFrame, out_path: str) -> None:
    wb = Workbook()
    ws = wb.active
    ws.title = "KPI"
    for r in dataframe_to_rows(df, index=False, header=True):
        ws.append(r)
    for col in ws.columns:
        ws.column_dimensions[col[0].column_letter].width = 16
    wb.save(out_path)

# src/publish/pdf_out.py
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas

def to_pdf(summary: dict, out_path: str) -> None:
    c = canvas.Canvas(out_path, pagesize=A4)
    w, h = A4
    y = h - 72
    c.setFont("Helvetica-Bold", 14)
    c.drawString(72, y, f"월간 안전 KPI 보고서 - {summary['month']}")
    c.setFont("Helvetica", 11); y -= 24
    for k, v in summary.items():
        c.drawString(72, y, f"{k}: {v}")
        y -= 18
    c.showPage(); c.save()

# run_monthly.py
from src.etl.load import load_events, load_hours
from src.kpi.calc import monthly_kpi
from src.publish.excel_out import to_excel
from src.publish.pdf_out import to_pdf

events = load_events("data/processed/events.csv")
hours = load_hours("data/processed/hours.xlsx")
kpi = monthly_kpi(events, hours, month="2025-08")
to_excel(kpi, "reports/monthly/2025-08_kpi.xlsx")
to_pdf(kpi.iloc[0].to_dict(), "reports/monthly/2025-08_kpi.pdf")

데이터 검증과 예외처리 규칙

스키마 검증을 사전 실행하여 필수 컬럼 누락시 즉시 실패하도록 한다.
시간대는 단일 표준으로 변환하고 리포트 표시만 로컬타임존으로 한다.
중복 이벤트는 event_id 기준으로 제거하고 로그에 기록한다.
총 근로시간 0일 때는 모든 비율형 KPI를 0으로 고정하고 경고를 남긴다.

# src/utils/validate.py
import pandas as pd

REQUIRED = {"event_id","event_dt","site","type","severity"}

def validate_events(df: pd.DataFrame) -> None:
missing = REQUIRED - set(df.columns)
if missing:
raise ValueError(f"필수 컬럼 누락: {sorted(list(missing))}")
if df["event_id"].duplicated().any():
raise ValueError("중복 event_id 발견")

시각화 자동 생성과 레이아웃

보고서 가독성을 위해 월간 추세와 사업장별 분포를 병행 제시하는 것이 효과적이다. PNG 차트를 사전 생성하여 문서에 삽입한다.

# src/viz/plots.py
import pandas as pd
import matplotlib.pyplot as plt

def plot_monthly_trir(df: pd.DataFrame, out_path: str) -> None:
pivot = df.pivot_table(index="month", values="TRIR", aggfunc="mean").reset_index()
plt.figure()
plt.plot(pivot["month"], pivot["TRIR"], marker="o")
plt.title("월간 TRIR 추세")
plt.xlabel("월"); plt.ylabel("TRIR")
plt.xticks(rotation=45); plt.tight_layout()
plt.savefig(out_path, dpi=150); plt.close()

문서 템플릿 자동화

사내 양식 준수를 위해 템플릿 기반 채우기를 사용한다. python-docx로 자리표시자를 채우는 방식이 안정적이다.

# src/publish/docx_out.py
from docx import Document

def fill_docx(template_path: str, context: dict, out_path: str) -> None:
doc = Document(template_path)
for p in doc.paragraphs:
for k, v in context.items():
p.text = p.text.replace(f"{{{{{k}}}}}", str(v))
doc.save(out_path)

스케줄링과 배포

운영환경에 맞춰 스케줄러를 설정한다. 리눅스는 cron, 윈도우는 작업 스케줄러를 사용한다. 성공·실패 로그는 중앙에 적재한다.

주기	작업	출력	배포
일일	전일 이벤트 집계이다.	daily.xlsx, daily.pdf	메일 그룹, 공유드라이브이다.
주간	주간 KPI와 트렌드이다.	weekly.pptx	경영회의 배포이다.
월간	월간 KPI와 개선현황이다.	monthly.docx, monthly.pdf	이사회 보고이다.

# cron 예시(매월 1일 07:10)
10 7 1 * * /usr/bin/python3 /opt/ehs-reporting/run_monthly.py >> /opt/ehs-reporting/logs/cron.log 2>&1

품질보증: 테스트, 로깅, 감사지표

자동화의 신뢰성은 테스트와 로깅에서 나온다. 핵심 KPI 계산과 변환 함수에 단위테스트를 작성한다.

# tests/test_kpi.py
from src.kpi.calc import monthly_kpi
import pandas as pd

def test_zero_hours():
e = pd.DataFrame([{"event_dt":"2025-08-10","severity":"중상","type":"사고"}])
e["event_dt"] = pd.to_datetime(e["event_dt"])
h = pd.DataFrame([{"month":"2025-08","work_hours":0}])
df = monthly_kpi(e, h, "2025-08")
assert df.loc[0,"TRIR"] == 0

로깅에는 실행시각, 데이터건수, 경고·오류 메시지, 산출물 경로를 포함해야 한다. 감사지표로 재실행 재현율, 실패율, 평균 실행시간을 관리한다.

개인정보 보호와 보안

개인 식별정보는 보고서 단계에서 익명화한다.
접근권한은 최소권한 원칙으로 부여한다.
민감 데이터는 전송·저장 모두 암호화를 적용한다.
API 자격증명은 환경변수나 비밀관리소에 보관한다.

현장 적용 체크리스트

항목	체크방법	상태
데이터 소스 명세서 존재 여부	문서 확인	준비
스키마 검증 자동화	테스트 실행	준비
지표 정의 승인	위원회 회의록	진행
템플릿 승인	양식 번호 확인	진행
스케줄러 등록	작업 목록 확인	미정
로그 보관 정책	보존기간 명시	준비

실무 팁: 장애 대응 시나리오

원천 CSV 컬럼 추가로 실패하는 경우에는 유연한 컬럼 매핑 테이블을 도입한다.
근로시간 지연 수신 시에는 이전 월 추정치를 임시 적용하고 보고서에는 추정 사용을 명시한다.
메일 발송 실패 시에는 파일서버 업로드와 링크 공지를 대체경로로 사용한다.

확장: 대시보드와 양방향 피드백

보고서 자동생성 후에는 웹 대시보드와 연동하여 실시간 탐색을 지원할 수 있다. 예산과 보안정책을 고려하여 사내 인트라넷 환경에서 plotly 기반 웹앱을 배포하면 효과적이다. 대시보드에는 KPI 타임라인, 사업장 비교, 아차사고 워드클라우드, 개선조치 만기경보 위젯을 포함하는 것이 바람직하다.

파일명과 버전 규칙

파일명은 <조직>_<리포트유형>_<YYYY-MM-DD>_v<주차·수정횟수>.확장자 규칙을 권장한다. 예시는 EHS_MonthlyKPI_2025-08-31_v2.pdf 형태이다. 자동증분 버전을 스크립트에서 관리하여 재현성을 높인다.

샘플 엔드투엔드 파이프라인

# main.py
from pathlib import Path
from src.etl.load import load_events, load_hours
from src.utils.validate import validate_events
from src.kpi.calc import monthly_kpi
from src.viz.plots import plot_monthly_trir
from src.publish.docx_out import fill_docx
from src.publish.excel_out import to_excel
from src.publish.pdf_out import to_pdf

MONTH = "2025-08"
events = load_events("data/processed/events.csv")
validate_events(events)
hours = load_hours("data/processed/hours.xlsx")

kpi_df = monthly_kpi(events, hours, MONTH)
plot_monthly_trir(kpi_df, f"reports/monthly/{MONTH}_trir.png")

ctx = {
"month": MONTH,
"TRIR": kpi_df.loc[0,"TRIR"],
"LTIFR": kpi_df.loc[0,"LTIFR"],
"NearMissRate": kpi_df.loc[0,"NearMissRate"]
}
fill_docx("templates/monthly_template.docx", ctx, f"reports/monthly/{MONTH}_kpi.docx")
to_excel(kpi_df, f"reports/monthly/{MONTH}_kpi.xlsx")
to_pdf(kpi_df.iloc[0].to_dict(), f"reports/monthly/{MONTH}_kpi.pdf")

도입 로드맵

요구사항 수집과 지표 정의 합의이다.
데이터 소스 연결과 스키마 표준화이다.
프로토타입 파이프라인 구현과 검증이다.
템플릿 확정과 경영 보고 승인이다.
스케줄러 배포와 운영 모니터링이다.
대시보드 확장과 사용자 피드백 반영이다.

예상 효과와 한계

보고서 생성 시간이 대폭 단축되고 수기 오류가 감소한다.
지표정의가 표준화되어 부서 간 비교 가능성이 높아진다.
반면, 원천데이터 품질이 낮으면 자동화의 신뢰성도 낮아진다.
지나친 커스텀 스크립트는 유지보수 비용을 증가시킨다.

FAQ

엑셀 양식이 자주 바뀌면 어떻게 대응하나?

컬럼 매핑 설정파일을 도입하여 필드명을 동적으로 연결하고, 스키마 검증 단계에서 미스매치를 조기 차단하는 것이 효과적이다.

PDF 대신 PPT 요약을 자동으로 만들 수 있나?

python-pptx로 핵심 KPI와 차트를 슬라이드에 배치하면 가능하다. 표준 슬라이드 마스터를 템플릿으로 사용하면 일관성이 유지된다.

여러 사업장을 한 번에 비교하려면 어떻게 하나?

사업장 코드를 기준으로 그룹바이를 수행하고, 박스플롯과 히트맵을 조합하여 분포와 이상치를 함께 제시하는 것이 바람직하다.

메일 발송 자동화에서 보안은 어떻게 보장하나?

앱 비밀번호 또는 서비스 계정을 사용하고, 자격증명은 환경변수로 주입하며, 첨부파일 암호화와 링크 공유 만료 정책을 병행한다.

데이터 누락이 있을 때 보고서를 중단해야 하나?

치명적 누락이면 실패로 처리하고 알림을 전송한다. 비치명적이면 추정치를 표시하되 보고서 본문에 추정 사용을 명시하고 후속 보정 절차를 안내한다.

SafeChem EHS Platform

안전보건관리책임자 지정 기준과 실무 절차 총정리