작업 스케줄링
Python으로 정기적인 작업을 자동으로 실행할 수 있습니다. 일일 백업, 정기 리포트, 데이터 수집 등을 스케줄링해보겠습니다.
schedule 라이브러리
설치하기
pip install schedule
기본 사용법
import schedule
import time
def job():
print("작업 실행!")
# 10초마다 실행
schedule.every(10).seconds.do(job)
# 10분마다 실행
schedule.every(10).minutes.do(job)
# 1시간마다 실행
schedule.every().hour.do(job)
# 매일 실행
schedule.every().day.do(job)
# 매주 월요일 실행
schedule.every().monday.do(job)
# 매주 수요일 13:15에 실행
schedule.every().wednesday.at("13:15").do(job)
# 매일 10:30에 실행
schedule.every().day.at("10:30").do(job)
# 스케줄 실행
while True:
schedule.run_pending()
time.sleep(1)
다양한 스케줄 패턴
시간 기반 스케줄
import schedule
def morning_routine():
print("좋은 아침입니다!")
def lunch_reminder():
print("점심 시간입니다!")
def evening_report():
print("일일 보고서 생성 중...")
# 매일 오전 8시
schedule.every().day.at("08:00").do(morning_routine)
# 매일 정오
schedule.every().day.at("12:00").do(lunch_reminder)
# 매일 오후 6시
schedule.every().day.at("18:00").do(evening_report)
# 평일 오전 9시
schedule.every().monday.at("09:00").do(morning_routine)
schedule.every().tuesday.at("09:00").do(morning_routine)
schedule.every().wednesday.at("09:00").do(morning_routine)
schedule.every().thursday.at("09:00").do(morning_routine)
schedule.every().friday.at("09:00").do(morning_routine)
간격 기반 스케줄
import schedule
def check_server():
print("서버 상태 확인")
def collect_data():
print("데이터 수집")
# 5분마다
schedule.every(5).minutes.do(check_server)
# 30분마다
schedule.every(30).minutes.do(collect_data)
# 2시간마다
schedule.every(2).hours.do(check_server)
# 3일마다
schedule.every(3).days.do(collect_data)
매개변수가 있는 작업
import schedule
def greet(name):
print(f"안녕하세요, {name}님!")
def send_report(email, report_type):
print(f"{email}로 {report_type} 리포트 발송")
# 매개변수와 함께 실행
schedule.every().day.at("09:00").do(greet, name="홍길동")
schedule.every().day.at("18:00").do(send_report,
email="admin@example.com",
report_type="일일")
작업 취소
import schedule
def some_task():
print("작업 실행")
# 조건이 만족되면 작업 취소
return schedule.CancelJob
# 작업 등록
job = schedule.every().day.do(some_task)
# 나중에 수동으로 취소
schedule.cancel_job(job)
# 모든 작업 취소
schedule.clear()
# 특정 태그의 작업만 취소
schedule.clear('daily-tasks')
태그 사용하기
import schedule
def backup():
print("백업 실행")
def report():
print("리포트 생성")
# 태그 지정
schedule.every().day.at("02:00").do(backup).tag('backup', 'critical')
schedule.every().day.at("09:00").do(report).tag('report', 'daily')
# 특정 태그의 작업만 실행
schedule.run_all(delay_seconds=0) # 모든 작업 즉시 실행
schedule.run_all('backup') # 'backup' 태그 작업만 실행
# 특정 태그의 작업 취소
schedule.clear('daily')
실전 예제
1. 일일 백업 자동화
import schedule
import time
import shutil
import os
from datetime import datetime
import zipfile
def create_backup():
"""매일 자동 백업"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 백업 시작...")
# 백업 대상 폴더
source_dirs = [
'/Users/username/Documents',
'/Users/username/Projects'
]
# 백업 저장 위치
backup_dir = '/Users/username/Backups'
os.makedirs(backup_dir, exist_ok=True)
# 백업 파일명 (날짜 포함)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_file = os.path.join(backup_dir, f'backup_{timestamp}.zip')
try:
# ZIP 파일 생성
with zipfile.ZipFile(backup_file, 'w', zipfile.ZIP_DEFLATED) as zipf:
for source_dir in source_dirs:
if not os.path.exists(source_dir):
print(f"경고: {source_dir} 폴더가 없습니다.")
continue
print(f"백업 중: {source_dir}")
for root, dirs, files in os.walk(source_dir):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, os.path.dirname(source_dir))
zipf.write(file_path, arcname)
# 백업 파일 크기 확인
size_mb = os.path.getsize(backup_file) / (1024 * 1024)
print(f"백업 완료: {backup_file} ({size_mb:.2f} MB)")
# 오래된 백업 정리 (30일 이상 된 파일 삭제)
cleanup_old_backups(backup_dir, days=30)
except Exception as e:
print(f"백업 실패: {e}")
def cleanup_old_backups(backup_dir, days=30):
"""오래된 백업 파일 삭제"""
now = time.time()
cutoff = now - (days * 86400) # days를 초로 변환
for filename in os.listdir(backup_dir):
if filename.startswith('backup_') and filename.endswith('.zip'):
file_path = os.path.join(backup_dir, filename)
file_time = os.path.getmtime(file_path)
if file_time < cutoff:
os.remove(file_path)
print(f"삭제: {filename} (생성 후 {days}일 경과)")
# 매일 새벽 2시에 백업 실행
schedule.every().day.at("02:00").do(create_backup)
print("백업 스케줄러 시작")
print("매일 02:00에 백업이 실행됩니다.")
# 테스트를 위해 즉시 실행
# create_backup()
while True:
schedule.run_pending()
time.sleep(60) # 1분마다 체크
2. 이메일 리포트 자동 발송
import schedule
import time
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
import os
def send_daily_report():
"""일일 리포트 이메일 발송"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 리포트 생성 중...")
# 리포트 데이터 생성 (예시)
report_data = generate_report()
# 이메일 설정
sender_email = "your_email@gmail.com"
sender_password = "your_app_password"
recipient_email = "recipient@example.com"
# 이메일 메시지 생성
message = MIMEMultipart()
message['From'] = sender_email
message['To'] = recipient_email
message['Subject'] = f"일일 리포트 - {datetime.now().strftime('%Y-%m-%d')}"
# 이메일 본문
body = f"""
<html>
<body>
<h2>일일 리포트</h2>
<p>날짜: {datetime.now().strftime('%Y년 %m월 %d일')}</p>
<h3>주요 지표</h3>
<ul>
<li>총 방문자: {report_data['visitors']:,}명</li>
<li>신규 가입: {report_data['signups']:,}명</li>
<li>매출: {report_data['revenue']:,}원</li>
</ul>
<p>상세 내용은 첨부 파일을 확인해주세요.</p>
</body>
</html>
"""
message.attach(MIMEText(body, 'html'))
# 파일 첨부 (선택사항)
report_file = 'daily_report.csv'
if os.path.exists(report_file):
with open(report_file, 'rb') as f:
part = MIMEBase('application', 'octet-stream')
part.set_payload(f.read())
encoders.encode_base64(part)
part.add_header('Content-Disposition', f'attachment; filename={report_file}')
message.attach(part)
try:
# SMTP 서버 연결 및 전송
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as server:
server.login(sender_email, sender_password)
server.send_message(message)
print("리포트 발송 완료!")
except Exception as e:
print(f"이메일 발송 실패: {e}")
def generate_report():
"""리포트 데이터 생성 (예시)"""
# 실제로는 데이터베이스에서 가져옴
return {
'visitors': 1234,
'signups': 56,
'revenue': 5678900
}
# 매일 오전 9시에 리포트 발송
schedule.every().day.at("09:00").do(send_daily_report)
# 매주 월요일 오전 10시에 주간 리포트
def send_weekly_report():
print("주간 리포트 발송")
# 주간 리포트 로직
schedule.every().monday.at("10:00").do(send_weekly_report)
print("리포트 스케줄러 시작")
print("- 매일 09:00: 일일 리포트")
print("- 매주 월요일 10:00: 주간 리포트")
while True:
schedule.run_pending()
time.sleep(60)
3. 웹 데이터 정기 수집
import schedule
import time
from datetime import datetime
import requests
from bs4 import BeautifulSoup
import csv
import os
def collect_stock_prices():
"""주식 가격 정기 수집"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 데이터 수집 시작...")
stocks = ['AAPL', 'GOOGL', 'MSFT', 'AMZN']
data = []
for stock in stocks:
try:
# API 또는 웹 스크래핑으로 데이터 수집
# 여기서는 예시로 랜덤 데이터 사용
import random
price = random.uniform(100, 500)
data.append({
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'stock': stock,
'price': round(price, 2)
})
print(f"{stock}: ${price:.2f}")
except Exception as e:
print(f"{stock} 수집 실패: {e}")
# CSV 파일에 저장
save_to_csv(data)
def save_to_csv(data):
"""데이터를 CSV 파일에 추가"""
filename = f"stock_prices_{datetime.now().strftime('%Y%m')}.csv"
file_exists = os.path.exists(filename)
with open(filename, 'a', newline='', encoding='utf-8') as f:
fieldnames = ['timestamp', 'stock', 'price']
writer = csv.DictWriter(f, fieldnames=fieldnames)
# 파일이 없으면 헤더 추가
if not file_exists:
writer.writeheader()
writer.writerows(data)
print(f"데이터 저장 완료: {filename}")
def analyze_daily_data():
"""하루 수집한 데이터 분석"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 일일 분석 중...")
filename = f"stock_prices_{datetime.now().strftime('%Y%m')}.csv"
if not os.path.exists(filename):
print("분석할 데이터가 없습니다.")
return
# 데이터 읽기 및 분석
stocks_data = {}
with open(filename, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
# 오늘 날짜 데이터만 필터링
if row['timestamp'].startswith(datetime.now().strftime('%Y-%m-%d')):
stock = row['stock']
price = float(row['price'])
if stock not in stocks_data:
stocks_data[stock] = []
stocks_data[stock].append(price)
# 통계 출력
print("\n일일 주가 통계:")
for stock, prices in stocks_data.items():
if prices:
avg = sum(prices) / len(prices)
min_price = min(prices)
max_price = max(prices)
print(f"{stock}:")
print(f" 평균: ${avg:.2f}")
print(f" 최저: ${min_price:.2f}")
print(f" 최고: ${max_price:.2f}")
print(f" 변동: ${max_price - min_price:.2f}")
# 매 30분마다 데이터 수집
schedule.every(30).minutes.do(collect_stock_prices)
# 매일 오후 6시에 일일 분석
schedule.every().day.at("18:00").do(analyze_daily_data)
print("데이터 수집 스케줄러 시작")
print("- 30분마다: 주가 데이터 수집")
print("- 매일 18:00: 일일 분석")
# 즉시 한 번 실행
collect_stock_prices()
while True:
schedule.run_pending()
time.sleep(60)
4. 시스템 모니터링
import schedule
import time
from datetime import datetime
import psutil
import platform
def monitor_system():
"""시스템 리소스 모니터링"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 시스템 모니터링")
# CPU 사용률
cpu_percent = psutil.cpu_percent(interval=1)
print(f"CPU 사용률: {cpu_percent}%")
# 메모리 사용률
memory = psutil.virtual_memory()
print(f"메모리 사용률: {memory.percent}%")
print(f"사용 중: {memory.used / (1024**3):.2f} GB / {memory.total / (1024**3):.2f} GB")
# 디스크 사용률
disk = psutil.disk_usage('/')
print(f"디스크 사용률: {disk.percent}%")
print(f"남은 공간: {disk.free / (1024**3):.2f} GB")
# 경고 체크
alerts = []
if cpu_percent > 80:
alerts.append(f"⚠️ CPU 사용률 높음: {cpu_percent}%")
if memory.percent > 80:
alerts.append(f"⚠️ 메모리 사용률 높음: {memory.percent}%")
if disk.percent > 90:
alerts.append(f"⚠️ 디스크 공간 부족: {disk.percent}%")
# 경고 출력
if alerts:
print("\n경고:")
for alert in alerts:
print(alert)
# 알림 전송 (이메일, 메시지 등)
send_alert(alerts)
else:
print("✓ 모든 시스템 정상")
def send_alert(alerts):
"""경고 알림 전송"""
# 실제로는 이메일, Slack, SMS 등으로 알림
print("\n알림 전송됨:")
for alert in alerts:
print(f" {alert}")
def daily_system_report():
"""일일 시스템 보고서"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 일일 시스템 보고서")
# 시스템 정보
print(f"\n시스템 정보:")
print(f"OS: {platform.system()} {platform.release()}")
print(f"프로세서: {platform.processor()}")
# 부팅 시간
boot_time = datetime.fromtimestamp(psutil.boot_time())
print(f"부팅 시간: {boot_time.strftime('%Y-%m-%d %H:%M:%S')}")
uptime = datetime.now() - boot_time
print(f"가동 시간: {uptime.days}일 {uptime.seconds // 3600}시간")
# 프로세스 정보
print(f"\n실행 중인 프로세스: {len(psutil.pids())}개")
# 네트워크 정보
net_io = psutil.net_io_counters()
print(f"\n네트워크:")
print(f"송신: {net_io.bytes_sent / (1024**3):.2f} GB")
print(f"수신: {net_io.bytes_recv / (1024**3):.2f} GB")
# 5분마다 시스템 모니터링
schedule.every(5).minutes.do(monitor_system)
# 매일 오전 9시에 일일 보고서
schedule.every().day.at("09:00").do(daily_system_report)
print("시스템 모니터링 시작")
print("- 5분마다: 리소스 모니터링")
print("- 매일 09:00: 일일 시스템 보고서")
# 즉시 한 번 실행
monitor_system()
while True:
schedule.run_pending()
time.sleep(60)
5. 데 이터베이스 정리 작업
import schedule
import time
from datetime import datetime, timedelta
import sqlite3
def cleanup_old_logs():
"""오래된 로그 데이터 삭제"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 데이터베이스 정리 시작...")
# 30일 이상 된 데이터 삭제
cutoff_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
try:
conn = sqlite3.connect('app.db')
cursor = conn.cursor()
# 오래된 로그 삭제
cursor.execute("""
DELETE FROM logs
WHERE created_at < ?
""", (cutoff_date,))
deleted_count = cursor.rowcount
# 오래된 세션 삭제
cursor.execute("""
DELETE FROM sessions
WHERE last_activity < ?
""", (cutoff_date,))
deleted_count += cursor.rowcount
conn.commit()
conn.close()
print(f"정리 완료: {deleted_count}개 레코드 삭제")
# 데이터베이스 최적화
optimize_database()
except Exception as e:
print(f"정리 실패: {e}")
def optimize_database():
"""데이터베이스 최적화"""
print("데이터베이스 최적화 중...")
try:
conn = sqlite3.connect('app.db')
conn.execute("VACUUM")
conn.close()
print("최적화 완료")
except Exception as e:
print(f"최적화 실패: {e}")
def backup_database():
"""데이터베이스 백업"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 데이터베이스 백업 중...")
import shutil
import os
source = 'app.db'
backup_dir = 'backups'
os.makedirs(backup_dir, exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
destination = os.path.join(backup_dir, f'app_backup_{timestamp}.db')
try:
shutil.copy2(source, destination)
# 파일 크기 확인
size_mb = os.path.getsize(destination) / (1024 * 1024)
print(f"백업 완료: {destination} ({size_mb:.2f} MB)")
except Exception as e:
print(f"백업 실패: {e}")
# 매일 새벽 3시에 정리 작업
schedule.every().day.at("03:00").do(cleanup_old_logs)
# 매일 새벽 2시에 백업
schedule.every().day.at("02:00").do(backup_database)
# 매주 일요일 새벽 4시에 최적화
schedule.every().sunday.at("04:00").do(optimize_database)
print("데이터베이스 관리 스케줄러 시작")
print("- 매일 02:00: 데이터베이스 백업")
print("- 매일 03:00: 오래된 데이터 정리")
print("- 매주 일요일 04:00: 데이터베이스 최적화")
while True:
schedule.run_pending()
time.sleep(60)