File name
Commit message
Commit date
File name
Commit message
Commit date
import smtplib
import os
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
from email.utils import formataddr
from typing import Dict, List, Tuple
import json
import fitz # PyMuPDF
import re
import urllib.parse
def extract_payslip_date_from_pdf(pdf_path: str) -> str:
"""PDF에서 급여명세서 날짜 정보 추출하여 다음달 표시 (예: "08월" → "09월")"""
# 기본값 설정: 현재 날짜 기준 다음달
from datetime import datetime
current_month = datetime.now().month
if current_month == 12:
next_month = 1
else:
next_month = current_month + 1
default_month = f"{next_month:02d}월"
try:
doc = fitz.open(pdf_path)
if doc.page_count == 0:
doc.close()
return default_month
# 첫 페이지에서 텍스트 추출
page = doc[0]
text = page.get_text()
doc.close()
# "YYYY년MM월분 급여명세서" 패턴 찾기
patterns = [
r'(\d{4})년(\d{1,2})월분\s*급여명세서',
r'(\d{4})년(\d{1,2})월\s*급여명세서',
r'급여명세서.*?(\d{4})년(\d{1,2})월',
]
for pattern in patterns:
match = re.search(pattern, text)
if match:
year, month = match.groups()
year_int = int(year)
month_int = int(month)
# 다음달 계산 (12월이면 1월로, 연도도 +1)
if month_int == 12:
next_month = 1
next_year = year_int + 1
else:
next_month = month_int + 1
next_year = year_int
month_str = f"{next_month:02d}월"
return month_str
# PDF에서 텍스트 추출 실패시 현재 날짜 기준 다음달 반환
return default_month
except Exception as e:
print(f"날짜 추출 실패 ({pdf_path}): {e}")
# 오류 발생시에도 현재 날짜 기준 다음달 반환
return default_month
class EmailConfig:
"""이메일 설정 관리 클래스"""
def __init__(self, config_file="email_config.json"):
self.config_file = config_file
self.config = self.load_config()
def load_config(self) -> Dict:
"""설정 파일 로드"""
default_config = {
"smtp_server": "smtp.worksmobile.com",
"smtp_port": 465,
"use_ssl": True,
"sender_email": "noreply@munjaon.co.kr",
"sender_name": "급여관리팀",
"email_template": {
"subject": "{month} 월급명세서 전달드립니다.",
"body": """안녕하세요.
늘 맡은 자리에서 애써주심에 깊이 감사드립니다.
다음 달 급여명세서를 PDF 파일로 첨부하여 보내드립니다.
※ 첨부 파일의 비밀번호는 본인 생년월일 8자리입니다.
예) 1990년 1월 15일 → 19900115
확인 중 궁금한 점이나 문의사항이 있으시면 언제든 편하게 연락 주시기 바랍니다.
감사합니다.
급여관리팀 드림"""
}
}
if os.path.exists(self.config_file):
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
loaded_config = json.load(f)
# 기본 설정과 병합
default_config.update(loaded_config)
except Exception as e:
print(f"설정 파일 로드 실패: {e}")
self.save_config(default_config)
return default_config
def save_config(self, config: Dict = None):
"""설정 파일 저장"""
if config is None:
config = self.config
try:
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(config, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"설정 파일 저장 실패: {e}")
def get_smtp_config(self) -> Dict:
"""SMTP 설정 반환"""
return {
"server": self.config.get("smtp_server", "smtp.worksmobile.com"),
"port": self.config.get("smtp_port", 465),
"use_ssl": self.config.get("use_ssl", True)
}
def get_sender_info(self) -> Dict:
"""발송자 정보 반환"""
return {
"email": self.config.get("sender_email", "noreply@munjaon.co.kr"),
"name": self.config.get("sender_name", "급여관리팀")
}
def get_email_template(self) -> Dict:
"""이메일 템플릿 반환"""
return self.config.get("email_template", {})
class PayslipEmailSender:
"""급여명세서 이메일 발송 클래스"""
def __init__(self, username: str = None, password: str = None):
self.config = EmailConfig()
self.username = username
self.password = password
self.smtp_config = self.config.get_smtp_config()
self.sender_info = self.config.get_sender_info()
self.template = self.config.get_email_template()
def set_credentials(self, username: str, password: str):
"""인증 정보 설정"""
self.username = username
self.password = password
def create_email_message(self, recipient_email: str, recipient_name: str,
pdf_path: str, custom_subject: str = None,
custom_body: str = None) -> MIMEMultipart:
"""이메일 메시지 생성"""
msg = MIMEMultipart()
# 발송자 정보
sender_email = self.sender_info["email"]
sender_name = self.sender_info["name"]
msg['From'] = formataddr((sender_name, sender_email))
msg['To'] = recipient_email
# 제목 - PDF에서 날짜 추출하여 동적 생성
if custom_subject:
subject = custom_subject
else:
month = extract_payslip_date_from_pdf(pdf_path) if pdf_path else "현재달"
subject_template = self.template.get("subject", "{month} 월급명세서 전달드립니다.")
subject = subject_template.format(month=month)
msg['Subject'] = subject
# 본문
body_template = custom_body or self.template.get("body", "안녕하세요, {name}님.\n\n월급명세서를 첨부하여 전달드립니다.")
body = body_template.format(name=recipient_name)
msg.attach(MIMEText(body, 'plain', 'utf-8'))
# 첨부파일
if pdf_path and os.path.exists(pdf_path):
try:
with open(pdf_path, "rb") as attachment:
# PDF 전용 MIME 타입 사용
part = MIMEBase('application', 'pdf')
part.set_payload(attachment.read())
encoders.encode_base64(part)
filename = os.path.basename(pdf_path)
# 한국어 파일명 인코딩 처리
try:
filename_encoded = filename.encode('ascii')
filename_header = f'attachment; filename="{filename}"'
except UnicodeEncodeError:
# 한국어 파일명의 경우 UTF-8로 인코딩
import urllib.parse
filename_encoded = urllib.parse.quote(filename)
filename_header = f"attachment; filename*=UTF-8''{filename_encoded}"
part.add_header('Content-Disposition', filename_header)
part.add_header('Content-Transfer-Encoding', 'base64')
msg.attach(part)
print(f"PDF 첨부 성공: {filename} ({os.path.getsize(pdf_path)} bytes)") # 디버그 로그
except Exception as e:
print(f"PDF 첨부 실패 ({pdf_path}): {e}") # 오류 로그
else:
print(f"PDF 파일 없음 또는 경로 문제: {pdf_path}")
return msg
def send_single_email(self, recipient_email: str, recipient_name: str,
pdf_path: str, custom_subject: str = None,
custom_body: str = None) -> Tuple[bool, str]:
"""단일 이메일 발송"""
if not self.username or not self.password:
return False, "인증 정보가 설정되지 않았습니다."
if not os.path.exists(pdf_path):
return False, f"첨부파일을 찾을 수 없습니다: {pdf_path}"
try:
# 이메일 메시지 생성
msg = self.create_email_message(recipient_email, recipient_name,
pdf_path, custom_subject, custom_body)
# SMTP 서버 연결 및 발송
if self.smtp_config["use_ssl"]:
server = smtplib.SMTP_SSL(self.smtp_config["server"], self.smtp_config["port"])
else:
server = smtplib.SMTP(self.smtp_config["server"], self.smtp_config["port"])
server.starttls()
server.login(self.username, self.password)
text = msg.as_string()
server.sendmail(self.sender_info["email"], recipient_email, text)
server.quit()
return True, f"'{recipient_name}'님에게 이메일이 성공적으로 발송되었습니다."
except Exception as e:
return False, f"이메일 발송 실패: {str(e)}"
def send_bulk_emails(self, email_data: List[Dict], progress_callback=None) -> Dict:
"""대량 이메일 발송
Args:
email_data: [{"name": "이름", "email": "이메일", "pdf_path": "파일경로"}]
progress_callback: 진행상황 콜백 함수
Returns:
{"success": [], "failed": []}
"""
if not self.username or not self.password:
return {
"success": [],
"failed": [{"name": "모든 직원", "error": "인증 정보가 설정되지 않았습니다."}]
}
results = {"success": [], "failed": []}
total_count = len(email_data)
try:
# SMTP 서버 연결 (연결 유지)
if self.smtp_config["use_ssl"]:
server = smtplib.SMTP_SSL(self.smtp_config["server"], self.smtp_config["port"])
else:
server = smtplib.SMTP(self.smtp_config["server"], self.smtp_config["port"])
server.starttls()
server.login(self.username, self.password)
for i, data in enumerate(email_data):
name = data.get("name", "")
email = data.get("email", "")
pdf_path = data.get("pdf_path", "")
try:
if not os.path.exists(pdf_path):
results["failed"].append({
"name": name,
"email": email,
"error": f"첨부파일을 찾을 수 없음: {pdf_path}"
})
continue
# 이메일 메시지 생성
msg = self.create_email_message(email, name, pdf_path)
# 발송
text = msg.as_string()
server.sendmail(self.sender_info["email"], email, text)
results["success"].append({
"name": name,
"email": email,
"pdf_path": pdf_path
})
except Exception as e:
results["failed"].append({
"name": name,
"email": email,
"error": str(e)
})
# 진행상황 업데이트
if progress_callback:
progress_callback(i + 1, total_count, name)
server.quit()
except Exception as e:
# 서버 연결 실패
for data in email_data:
if data.get("name") not in [item.get("name") for item in results["success"]]:
results["failed"].append({
"name": data.get("name", ""),
"email": data.get("email", ""),
"error": f"SMTP 서버 연결 실패: {str(e)}"
})
return results
def test_connection(self) -> Tuple[bool, str]:
"""SMTP 서버 연결 테스트"""
if not self.username or not self.password:
return False, "인증 정보가 설정되지 않았습니다."
try:
if self.smtp_config["use_ssl"]:
server = smtplib.SMTP_SSL(self.smtp_config["server"], self.smtp_config["port"])
else:
server = smtplib.SMTP(self.smtp_config["server"], self.smtp_config["port"])
server.starttls()
server.login(self.username, self.password)
server.quit()
return True, "SMTP 서버 연결 성공!"
except Exception as e:
return False, f"SMTP 서버 연결 실패: {str(e)}"