QHSE report 자동화: 스케줄러와 이메일 발송까지 완성하기

QHSE report automation image

지난 포스트에서는 QHSE report PDF를 생성하고 테스트 코드까지 붙여서
“클릭 한 번이면 리포트가 나오는” 시스템을 만들었죠.

하지만 현실에서는
관리자가 매번 사이트에 들어가 버튼을 눌러서 리포트를 생성하는 방식은
솔직히 절대 오래가지 않습니다.

그렇기 때문에 대부분의 QHSE 시스템은 이렇게 구성됩니다.

  • 매월 1일 09:00 → 자동으로 리포트 생성
  • 담당자·팀장·품질책임자에게 PDF를 이메일 발송
  • 사내 파일서버·회계 시스템에도 자동 백업

이번 편은 바로 이 단계까지 다룹니다.

Django + Celery + Celery Beat + 이메일 발송 → QHSE report 자동화 풀루틴

1. 전체 구조 한눈에 정리

QHSE report 자동화

자동화를 구성하는 요소는 네 가지입니다.

(1) PDF 생성 함수
(지난 편의 monthly_qhse_report 뷰와 동일한 로직)

2.其の2 비동기 작업 큐(Celery)
“리포트 생성 → 이메일 발송”을 서버에 부하 없이 처리

(三) Celery Beat (스케줄러)
매월 1일 09:00에 작업을 예약 실행

④の Email Backend 설정
SMTP 또는 사내 메일서버로 PDF 첨부 파일 발송

이 네 요소가 합쳐져서 자동 리포트 시스템이 됩니다.

2. Celery 기본 설정

2.1 Celery 설치

pip install celery

2.2 프로젝트 루트에 celery.py 生成

core/celery.py (코어 폴더 밑에 셀러리 파일 생성)

import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")

app = Celery("core")

app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

2.3 __init__.py 수정

core/__init__.py(코어 폴더 밑에 이닛 파일 생성)

from .celery import app as celery_app

__all__ = ("celery_app",)

3. 리포트 PDF 생성 + 메일 발송 Celery Task

이제 핵심 코드입니다.

reports/tasks.py

import datetime
from celery import shared_task
from django.core.mail import EmailMessage
from django.template.loader import render_to_string
from weasyprint import HTML

from vendors.models import Vendor, QhseRecord


@shared_task
def generate_and_send_monthly_report(year, month, recipients):
    """
    1) QHSE 리포트 PDF 생성
    2) 이메일로 발송
    """
    # 날짜 범위
    start_date = datetime.date(year, month, 1)
    if month == 12:
        end_date = datetime.date(year + 1, 1, 1)
    else:
        end_date = datetime.date(year, month + 1, 1)

    records = QhseRecord.objects.select_related("vendor").filter(
        occurred_at__gte=start_date,
        occurred_at__lt=end_date,
    )

    vendors = Vendor.objects.all()

    context = {
        "year": year,
        "month": month,
        "vendors": vendors,
        "total_records": records.count(),
        "open_records": records.filter(is_closed=False).count(),
        "by_type": {
            key: records.filter(record_type=key).count()
            for key, _ in QhseRecord.QHSE_TYPE_CHOICES
        },
        "records": records,
    }

    # HTML → PDF
    html_string = render_to_string("reports/monthly_qhse_report.html", context)
    pdf_bytes = HTML(string=html_string).write_pdf()

    # 이메일 구성
    subject = f"{year}년 {month}월 QHSE report"
    body = f"{year}년 {month}월 QHSE report 입니다. 첨부 파일을 확인해주세요."

    email = EmailMessage(
        subject,
        body,
        "noreply@company.com",    # 보내는 사람
        recipients,               # 받는 사람 리스트
    )
    email.attach(
        filename=f"QHSE_Report_{year}_{month:02d}.pdf",
        content=pdf_bytes,
        mimetype="application/pdf",
    )

    email.send()

    return f"Report sent to {recipients}"

이 Task는 다음 흐름을 자동으로 수행합니다.

데이터 조회 → PDF 생성 → 이메일 발송

4. Celery Beat 스케줄러 설정

4.1 settings.py에 스케줄러 추가

CELERY_BEAT_SCHEDULE = {
    "send-monthly-qhse-report": {
        "task": "reports.tasks.generate_and_send_monthly_report",
        "schedule": {
            "type": "crontab",
            "minute": 0,
            "hour": 9,
            "day_of_month": "1",
        },
        "args": (2025, 12, ["manager@company.com", "qhse@company.com"]),
    },
}

✔ 주의: 여기서 year/month를 고정하면 안 되므로, 실제 서비스에서는
현재 날짜 기반으로 처리하는 wrapper task를 둡니다.

例:

@shared_task
def scheduled_monthly_report():
    today = datetime.date.today()
    year = today.year
    month = today.month - 1 or 12

    recipients = ["manager@company.com", "qhse@company.com"]

    return generate_and_send_monthly_report(year, month, recipients)

그리고 Celery Beat 스케줄은 이렇게 연결합니다.

"send-monthly-qhse-report": {
    "task": "reports.tasks.scheduled_monthly_report",
    "schedule": crontab(minute=0, hour=9, day_of_month="1"),
},

5. 이메일 설정 (settings.py)

SMTP는 회사마다 다르지만 예시를 넣어보면

EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = "smtp.company.com"
EMAIL_PORT = 587
EMAIL_HOST_USER = "noreply@company.com"
EMAIL_HOST_PASSWORD = "비밀번호"
EMAIL_USE_TLS = True

사내망이라면 사내 SMTP를 사용하면 됩니다.

6. 테스트: 스케줄러+메일 발송 흐름 검증

스케줄러+메일 발송 흐름 이미지

아주 간단한 테스트로도 “동작 여부”는 체크할 수 있습니다.

from django.test import TestCase
from reports.tasks import generate_and_send_monthly_report
from unittest.mock import patch


class MonthlyReportTaskTests(TestCase):

    @patch("django.core.mail.EmailMessage.send")
    def test_task_sends_email(self, mock_send):
        generate_and_send_monthly_report(
            2025,
            11,
            ["manager@company.com"]
        )
        self.assertTrue(mock_send.called)

이 테스트는

  • 리포트가 PDF로 생성되었는지
  • 이메일 전송 함수가 호출되었는지

를 기본적으로 보장해 줍니다.

7. OpenCode로 스케줄러/테스트 자동 생성시키기

이제 자동화의 마지막 단계로
터미널에서 AI에게 맡기기です。

7.1 스케줄러 코드 자동 리팩터링 프롬프트

/ask Celery beat 스케줄러 설정을 개선하고 싶어.
현재 설정에서 year/month가 고정되어 있는데,
지난달 기준으로 자동 계산하는 wrapper task를 추가해줘.
그리고 예외 상황(예: 데이터가 하나도 없을 때)의 처리도 포함해줘.

7.2 PDF + 이메일 발송 테스트 보완 프롬프트

/ask generate_and_send_monthly_report 테스트 코드를 보완해줘.
- PDF가 실제로 생성되었는지 확인
- 이메일 첨부 파일 이름이 올바른지 확인
두 가지를 검증하는 새로운 테스트 함수를 작성해줘.

7.3 관리자 모드 대시보드용 텍스트 자동 생성

/ask 관리자 페이지에서 보여줄
'이번 달 리포트 생성 로그'를 위한 설명 문장을 3줄로 정리해줘.
nuclear SME 관리자 관점에서 읽기 쉬운 스타일로.

整理する

이번 포스터를 통해서 QHSE report 자동 생성 시스템을 실제처럼 운영 가능한 단계로 끌어올렸습니다.

핵심 요약

  • Celery로 PDF 리포트 생성과 이메일 발송을 비동기 처리
  • Celery Beat로 매월 1일 09:00 자동 실행
  • SMTP 설정으로 PDF 첨부 메일 발송
  • 테스트 코드로 안정성 확보
  • OpenCode를 활용해 스케줄러·테스트·문구 자동화까지 가능

이제 QHSE 시스템은
“기능이 있는 서비스” → “스스로 움직이는 서비스”로 진화했습니다.

다음 포스트에서는
대시보드에 그래프(막대/라인/도넛)까지 붙여서 시각화를 강화하는 실전 편으로 이어가겠습니다.

類似の投稿