46537ee11a
- [P1] migration runner 도입: schema_migrations 추적, advisory lock, 단일 트랜잭션 실행, SQL 검증 (기존 DB 업그레이드 대응) - [P1] eml extract 큐 조건 분기: extract_worker 미지원 포맷 큐 스킵 - [P2] iCalendar escape_ical_text() 추가: RFC 5545 준수 - [P2] 이메일 charset 감지: get_content_charset() 사용 + payload None 방어 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
139 lines
3.7 KiB
Python
139 lines
3.7 KiB
Python
"""공통 유틸리티 — v1 pkm_utils.py에서 AppleScript 제거, 나머지 포팅"""
|
|
|
|
import hashlib
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
|
|
def setup_logger(name: str, log_dir: str = "logs") -> logging.Logger:
|
|
"""로거 설정"""
|
|
Path(log_dir).mkdir(exist_ok=True)
|
|
logger = logging.getLogger(name)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
if not logger.handlers:
|
|
# 파일 핸들러
|
|
fh = logging.FileHandler(f"{log_dir}/{name}.log", encoding="utf-8")
|
|
fh.setFormatter(logging.Formatter(
|
|
"%(asctime)s [%(levelname)s] %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
))
|
|
logger.addHandler(fh)
|
|
|
|
# 콘솔 핸들러
|
|
ch = logging.StreamHandler()
|
|
ch.setFormatter(logging.Formatter("[%(levelname)s] %(message)s"))
|
|
logger.addHandler(ch)
|
|
|
|
return logger
|
|
|
|
|
|
def file_hash(path: str | Path) -> str:
|
|
"""파일 SHA-256 해시 계산"""
|
|
sha256 = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(8192), b""):
|
|
sha256.update(chunk)
|
|
return sha256.hexdigest()
|
|
|
|
|
|
def count_log_errors(log_path: str) -> int:
|
|
"""로그 파일에서 ERROR 건수 카운트"""
|
|
try:
|
|
with open(log_path, encoding="utf-8") as f:
|
|
return sum(1 for line in f if "[ERROR]" in line)
|
|
except FileNotFoundError:
|
|
return 0
|
|
|
|
|
|
# ─── CalDAV 헬퍼 ───
|
|
|
|
|
|
def escape_ical_text(text: str | None) -> str:
|
|
"""iCalendar TEXT 값 이스케이프 (RFC 5545 §3.3.11).
|
|
SUMMARY, DESCRIPTION, LOCATION 등 TEXT 프로퍼티에 사용.
|
|
"""
|
|
if not text:
|
|
return ""
|
|
text = text.replace("\r\n", "\n").replace("\r", "\n") # CRLF 정규화
|
|
text = text.replace("\\", "\\\\") # 백슬래시 먼저
|
|
text = text.replace("\n", "\\n")
|
|
text = text.replace(",", "\\,")
|
|
text = text.replace(";", "\\;")
|
|
return text
|
|
|
|
|
|
def create_caldav_todo(
|
|
caldav_url: str,
|
|
username: str,
|
|
password: str,
|
|
title: str,
|
|
description: str = "",
|
|
due_days: int = 7,
|
|
) -> str | None:
|
|
"""Synology Calendar에 VTODO 생성, UID 반환"""
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import caldav
|
|
|
|
try:
|
|
client = caldav.DAVClient(url=caldav_url, username=username, password=password)
|
|
principal = client.principal()
|
|
calendars = principal.calendars()
|
|
if not calendars:
|
|
return None
|
|
|
|
calendar = calendars[0]
|
|
uid = str(uuid.uuid4())
|
|
due = datetime.now(timezone.utc) + timedelta(days=due_days)
|
|
due_str = due.strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
vtodo = f"""BEGIN:VCALENDAR
|
|
VERSION:2.0
|
|
BEGIN:VTODO
|
|
UID:{uid}
|
|
SUMMARY:{escape_ical_text(title)}
|
|
DESCRIPTION:{escape_ical_text(description)}
|
|
DUE:{due_str}
|
|
STATUS:NEEDS-ACTION
|
|
PRIORITY:5
|
|
END:VTODO
|
|
END:VCALENDAR"""
|
|
|
|
calendar.save_event(vtodo)
|
|
return uid
|
|
except Exception as e:
|
|
logging.getLogger("caldav").error(f"CalDAV VTODO 생성 실패: {e}")
|
|
return None
|
|
|
|
|
|
# ─── SMTP 헬퍼 ───
|
|
|
|
|
|
def send_smtp_email(
|
|
host: str,
|
|
port: int,
|
|
username: str,
|
|
password: str,
|
|
subject: str,
|
|
body: str,
|
|
to_addr: str | None = None,
|
|
):
|
|
"""Synology MailPlus SMTP로 이메일 발송"""
|
|
import smtplib
|
|
from email.mime.text import MIMEText
|
|
|
|
to_addr = to_addr or username
|
|
msg = MIMEText(body, "plain", "utf-8")
|
|
msg["Subject"] = subject
|
|
msg["From"] = username
|
|
msg["To"] = to_addr
|
|
|
|
try:
|
|
with smtplib.SMTP_SSL(host, port, timeout=30) as server:
|
|
server.login(username, password)
|
|
server.send_message(msg)
|
|
except Exception as e:
|
|
logging.getLogger("smtp").error(f"SMTP 발송 실패: {e}")
|