[응용#9] 전체 자동화 워크플로우 통합 (End-to-End 시스템 구축)

한줄 요약:
크롤링 → 요약/키워드 → LLM 리포트 → 워드프레스 자동 포스팅 → 메일/알림까지 하나의 파이프라인으로 자동 실행!

---

1. 목표

• 개별 스크립트(응용#5~#8)를 하나의 오케스트레이터로 묶기
• 환경 변수(.env) 단일 관리, 로그/리트라이/중복방지 적용
• 스케줄링과 알림(이메일/슬랙/텔레그램 택1) 추가

---

2. 아키텍처 개요

[URL 목록] 
   └─▶ crawler_cleaner (응용#5) ─▶ report.csv / report.md
          └─▶ llm_report (응용#7) ─▶ final_report.md
                 └─▶ wp_poster (응용#8) ─▶ 블로그 게시
                        └─▶ notifier ─▶ 이메일/슬랙/텔레그램

---

3. 디렉터리 & 설정

project/
 ├─ .env                         # 모든 키/URL/토큰
 ├─ urls.txt                     # 대상 URL 목록
 ├─ auto_report.py               # 응용#5
 ├─ llm_report.py                # 응용#7
 ├─ auto_post_wp.py              # 응용#8
 ├─ orchestrator.py              # ★ 이번 강좌 핵심
 └─ logs/
     └─ run_YYYYMMDD.log

.env 예시(통합)

# WordPress
WP_URL=https://yourblog.com
WP_USER=admin
WP_APP_PASS=xxxx xxxx xxxx xxxx

# OpenAI
OPENAI_API_KEY=sk-...

# Mail (yagmail 사용 시)
MAIL_SENDER=your_email@gmail.com
MAIL_APP_PASS=your_app_password
MAIL_TO=receiver@example.com

# Slack(선택)
SLACK_WEBHOOK=https://hooks.slack.com/services/...

# Scheduler
RUN_HOUR=07
RUN_MINUTE=00

---

4. 오케스트레이터(복붙) — orchestrator.py

import os, sys, time, hashlib, json, traceback
from datetime import datetime
from dotenv import load_dotenv
import subprocess
import logging
import requests

# ── 1) 공통 설정/로그 ─────────────────────────────────────────
load_dotenv()
LOG_DIR = "logs"
os.makedirs(LOG_DIR, exist_ok=True)
log_path = os.path.join(LOG_DIR, f"run_{datetime.now():%Y%m%d}.log")

logging.basicConfig(
    filename=log_path,
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
console = logging.StreamHandler(sys.stdout)
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
logging.getLogger().addHandler(console)

SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK", "").strip()
MAIL_TO = os.getenv("MAIL_TO", "").strip()

def notify(text: str):
    """필요 시 Slack 웹훅 + 콘솔 출력"""
    logging.info(text)
    if SLACK_WEBHOOK:
        try:
            requests.post(SLACK_WEBHOOK, json={"text": text}, timeout=8)
        except Exception:
            logging.warning("Slack 통신 실패")

# ── 2) 유틸: 실행/리트라이/중복방지 ────────────────────────────
def run_py(script, args=None, retries=2):
    cmd = ["python", script] + (args or [])
    for attempt in range(retries + 1):
        try:
            logging.info(f"▶ 실행: {' '.join(cmd)} (시도 {attempt+1})")
            subprocess.check_call(cmd)
            return
        except subprocess.CalledProcessError as e:
            logging.error(f"실패({script}): {e}")
            time.sleep(3)
    raise RuntimeError(f"재시도 초과: {script}")

def file_hash(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()[:16]

def is_duplicate_today(ref_path="final_report.md"):
    """금일 생성물 중복 게시 방지: 해시 기록 비교"""
    stamp_path = os.path.join(LOG_DIR, f"posted_{datetime.now():%Y%m%d}.json")
    if not os.path.exists(ref_path):
        return False
    h = file_hash(ref_path)
    if os.path.exists(stamp_path):
        try:
            data = json.load(open(stamp_path, "r", encoding="utf-8"))
            if data.get("hash") == h:
                return True
        except Exception:
            pass
    json.dump({"hash": h, "ts": time.time()}, open(stamp_path, "w", encoding="utf-8"))
    return False

# ── 3) 파이프라인 실행 ────────────────────────────────────────
def main():
    start = time.time()
    notify("🚀 파이프라인 시작")

    try:
        # 3-1) 크롤링/요약/키워드 (응용#5)
        run_py("auto_report.py")               # report.csv, report.md

        # 3-2) LLM 리포트 생성 (응용#7)
        run_py("llm_report.py")                # final_report.md

        # 3-3) 중복 방지 체크
        if is_duplicate_today("final_report.md"):
            notify("♻️ 금일 동일 리포트 감지 → 게시 생략")
            return

        # 3-4) 워드프레스 포스팅 (응용#8)
        run_py("auto_post_wp.py")              # 게시

        elapsed = round(time.time() - start, 2)
        notify(f"✅ 파이프라인 성공 (소요 {elapsed}s)")
    except Exception:
        err = traceback.format_exc(limit=5)
        notify(f"❌ 파이프라인 실패\n```{err}```")
        raise

if __name__ == "__main__":
    main()

핵심 포인트

• run_py()로 각 스텝을 격리 실행 + 리트라이
• 산출물 final_report.md의 SHA-256 해시로 중복 게시 방지
• Slack 웹훅(선택)으로 성공/실패 알림

---

5. 스케줄링

파이썬 schedule로 상시 실행(간단)

# scheduler.py
import os, time
from dotenv import load_dotenv
import schedule
load_dotenv()

H = os.getenv("RUN_HOUR", "07")
M = os.getenv("RUN_MINUTE", "00")

def job():
    os.system("python orchestrator.py")

schedule.every().day.at(f"{H}:{M}").do(job)
print(f"⏰ 매일 {H}:{M} 자동 실행")
while True:
    schedule.run_pending()
    time.sleep(30)

OS 스케줄러(권장)

• Windows 작업 스케줄러: python orchestrator.py
• macOS/Linux crontab: 0 7 * * * /usr/bin/python3 /path/orchestrator.py

---

6. 로그 & 장애 복구

• 로그는 logs/run_YYYYMMDD.log로 일자별 분리
• 실패 시 notify()로 Slack/콘솔에 스택트레이스 제공
• 리트라이는 각 스텝 최대 2회(필요 시 조정)

---

7. 품질 & 오탐(탐지) 줄이는 합법적 베스트 프랙티스

• 원저작/출처 명시: 원문 링크·도메인·생성 일자 표기
• 근거 제시: 표·수치·코드·로그 스니펫 등 검증 가능한 증거 포함
• AI 사용 고지: “요약/초안에 AI 활용, 최종 검수는 사람” 명시
• 개인화/맥락화: 운영자 코멘트(해석·의견·경험) 최소 1단락 추가
• 중복 콘텐츠 회피: 동일 URL·제목 해시 기반 중복 방지(위 코드 적용)

---

8. 자주 겪는 문제

문제	원인	해결
게시가 간헐적으로 누락	네트워크·토큰 만료	오케스트레이터의 리트라이/에러 알림 확인
같은 글이 반복 업로드	산출물 변경 없음	is_duplicate_today() 해시 체크 활성화
LLM 응답 지연/에러	레이트 리밋	배치 요약 + 재시도 백오프, 모델/요약길이 최적화
워드프레스 401/403	권한/보안 플러그인	앱 비밀번호/REST 허용, WAF 예외 등록

---

9. 체크리스트

• .env 통합키 세팅 완료
• urls.txt·auto_report.py 정상 동작
• llm_report.py로 final_report.md 생성 확인
• auto_post_wp.py 게시 성공
• orchestrator.py 실행/로그/알림 정상
• 스케줄러 등록 완료

---

10. 요약 한 줄

오케스트레이터 한 번으로 수집→요약→리포트→게시→알림 완전 자동화!
중복 방지·로그·리트라이까지 넣어 실서비스 수준으로 안정화 ✅

---

이전 강좌 👈 [응용#8] 워드프레스 자동 포스팅: REST API
다음 강좌 👉 [운영#1] 모니터링/알림/코스트 관리(로그 대시보드 & 에러율 트래킹)