파이썬으로 만드는 실용적인 자동화 프로그램 - 업무 효율을 10배 높이는 방법
파이썬으로 만드는 실용적인 자동화 프로그램 - 업무 효율을 10배 높이는 방법
파이썬으로 만드는 실용적인 자동화 프로그램 - 업무 효율을 10배 높이는 방법
반복적이고 지루한 작업에 시간을 낭비하고 계신가요? 파이썬은 이러한 일상적인 작업들을 자동화하여 생산성을 극적으로 향상시킬 수 있는 최고의 도구입니다. 파이썬의 간결한 문법과 풍부한 라이브러리 생태계는 프로그래밍 초보자도 쉽게 자동화 프로그램을 만들 수 있게 해줍니다. 이번 포스트에서는 실무에서 바로 활용할 수 있는 다양한 파이썬 자동화 사례들을 살펴보고, 각 분야별로 어떤 프로그램을 만들 수 있는지 구체적으로 알아보겠습니다.
📁 파일 및 폴더 자동화
1. 파일 정리 자동화 프로그램
다운로드 폴더나 작업 디렉토리에 파일이 무질서하게 쌓여있다면, 파이썬으로 자동으로 파일을 분류하고 정리하는 프로그램을 만들 수 있습니다. 파일 확장자, 생성 날짜, 파일명 패턴 등을 기준으로 자동으로 폴더를 생성하고 파일을 이동시킬 수 있습니다.
import os
import shutil
from pathlib import Path
from datetime import datetime
def organize_files(directory):
"""파일 확장자별로 자동 정리"""
file_types = {
'images': ['.jpg', '.jpeg', '.png', '.gif', '.webp'],
'documents': ['.pdf', '.docx', '.xlsx', '.txt', '.pptx'],
'videos': ['.mp4', '.avi', '.mkv', '.mov'],
'music': ['.mp3', '.wav', '.flac'],
'archives': ['.zip', '.rar', '.7z', '.tar']
}
for filename in os.listdir(directory):
file_path = Path(directory) / filename
if file_path.is_file():
ext = file_path.suffix.lower()
for category, extensions in file_types.items():
if ext in extensions:
category_path = Path(directory) / category
category_path.mkdir(exist_ok=True)
shutil.move(str(file_path), str(category_path / filename))
print(f"Moved {filename} to {category}/")
break
# 사용 예시
organize_files("/Users/username/Downloads")
이 프로그램은 매일 실행되도록 스케줄러(cron, Task Scheduler)에 등록하면, 다운로드 폴더가 항상 깔끔하게 유지됩니다. 파일 정리에 소요되는 시간을 완전히 제거할 수 있습니다.
2. 대량 파일명 변경 자동화
수백 개의 사진이나 문서 파일의 이름을 일괄적으로 변경해야 할 때, 수동으로 하나씩 바꾸는 것은 매우 비효율적입니다. 파이썬으로 패턴에 맞춰 자동으로 파일명을 변경하는 프로그램을 만들 수 있습니다.
import os
from datetime import datetime
def batch_rename_files(directory, prefix="IMG", start_number=1):
"""파일명을 일괄 변경 (번호 순서대로)"""
files = [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))]
files.sort()
for index, filename in enumerate(files, start=start_number):
file_ext = os.path.splitext(filename)[1]
new_name = f"{prefix}_{index:04d}{file_ext}"
old_path = os.path.join(directory, filename)
new_path = os.path.join(directory, new_name)
os.rename(old_path, new_path)
print(f"Renamed: {filename} → {new_name}")
def rename_with_date(directory):
"""파일 생성일을 기준으로 파일명 변경"""
for filename in os.listdir(directory):
filepath = os.path.join(directory, filename)
if os.path.isfile(filepath):
timestamp = os.path.getctime(filepath)
date_str = datetime.fromtimestamp(timestamp).strftime("%Y%m%d")
file_ext = os.path.splitext(filename)[1]
new_name = f"{date_str}_{filename}"
os.rename(filepath, os.path.join(directory, new_name))
# 사용 예시
batch_rename_files("./photos", prefix="Vacation2026", start_number=1)
이 방법으로 여행 사진, 스캔 문서, 프로젝트 파일 등을 체계적으로 관리할 수 있습니다.
🌐 웹 스크래핑 및 데이터 수집 자동화
3. 가격 모니터링 자동화
온라인 쇼핑몰의 상품 가격을 정기적으로 확인하여, 가격이 특정 금액 이하로 떨어지면 알림을 받는 프로그램을 만들 수 있습니다. Beautiful Soup이나 Selenium을 활용하면 웹사이트에서 원하는 정보를 자동으로 추출할 수 있습니다.
import requests
from bs4 import BeautifulSoup
import smtplib
from email.mime.text import MIMEText
import time
class PriceMonitor:
def __init__(self, url, target_price):
self.url = url
self.target_price = target_price
def get_current_price(self):
"""웹사이트에서 현재 가격 추출"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(self.url, headers=headers)
soup = BeautifulSoup(response.content, 'html.parser')
# 가격 요소 선택 (사이트마다 다름)
price_element = soup.select_one('.price')
if price_element:
price_text = price_element.text.strip()
# 숫자만 추출
price = int(''.join(filter(str.isdigit, price_text)))
return price
return None
def send_alert(self, current_price):
"""가격 하락 알림 이메일 전송"""
message = f"""
가격 알림!
목표 가격: {self.target_price:,}원
현재 가격: {current_price:,}원
상품 링크: {self.url}
"""
# 이메일 발송 로직
print(f"🔔 알림: 가격이 {current_price:,}원으로 하락했습니다!")
def monitor(self, check_interval=3600):
"""주기적으로 가격 확인"""
print(f"가격 모니터링 시작: 목표 가격 {self.target_price:,}원")
while True:
current_price = self.get_current_price()
if current_price and current_price <= self.target_price:
self.send_alert(current_price)
break
print(f"현재 가격: {current_price:,}원 (체크 시간: {time.strftime('%Y-%m-%d %H:%M:%S')})")
time.sleep(check_interval)
# 사용 예시
monitor = PriceMonitor(
url="https://example.com/product/12345",
target_price=50000
)
monitor.monitor(check_interval=3600) # 1시간마다 체크
이 프로그램으로 원하는 상품을 최저가에 구매할 수 있고, 쇼핑몰을 매번 확인하는 수고를 덜 수 있습니다.
4. 뉴스 및 콘텐츠 수집 자동화
관심 분야의 최신 뉴스나 블로그 글을 매일 자동으로 수집하여 요약본을 받아볼 수 있습니다.
import feedparser
from datetime import datetime, timedelta
class NewsAggregator:
def __init__(self):
self.feeds = {
'기술': [
'https://news.ycombinator.com/rss',
'https://www.reddit.com/r/programming/.rss'
],
'비즈니스': [
'https://feeds.bloomberg.com/markets/news.rss'
]
}
def fetch_latest_news(self, hours=24):
"""최근 N시간 이내 뉴스 수집"""
cutoff_time = datetime.now() - timedelta(hours=hours)
all_news = []
for category, urls in self.feeds.items():
for url in urls:
feed = feedparser.parse(url)
for entry in feed.entries[:10]:
all_news.append({
'category': category,
'title': entry.title,
'link': entry.link,
'published': entry.get('published', 'Unknown')
})
return all_news
def generate_daily_digest(self):
"""일일 다이제스트 생성"""
news_list = self.fetch_latest_news(24)
digest = "📰 오늘의 뉴스 다이제스트\n"
digest += "=" * 50 + "\n\n"
for item in news_list:
digest += f"[{item['category']}] {item['title']}\n"
digest += f"🔗 {item['link']}\n\n"
return digest
# 사용 예시
aggregator = NewsAggregator()
daily_digest = aggregator.generate_daily_digest()
print(daily_digest)
매일 아침 정해진 시간에 이메일이나 슬랙으로 요약본을 받아볼 수 있습니다.
📊 엑셀 및 데이터 처리 자동화
5. 엑셀 보고서 자동 생성
매주 반복되는 엑셀 보고서 작성을 자동화할 수 있습니다. pandas와 openpyxl을 사용하면 데이터를 분석하고, 차트를 생성하며, 서식이 적용된 보고서를 자동으로 만들 수 있습니다.
import pandas as pd
from openpyxl import load_workbook
from openpyxl.chart import BarChart, Reference
from openpyxl.styles import Font, PatternFill, Alignment
from datetime import datetime
class ExcelReportGenerator:
def __init__(self, data_file):
self.df = pd.read_excel(data_file)
def analyze_data(self):
"""데이터 분석 수행"""
summary = {
'total_sales': self.df['sales'].sum(),
'average_sales': self.df['sales'].mean(),
'top_products': self.df.nlargest(5, 'sales')[['product', 'sales']].to_dict('records'),
'monthly_trend': self.df.groupby('month')['sales'].sum().to_dict()
}
return summary
def create_report(self, output_file):
"""서식이 적용된 보고서 생성"""
summary = self.analyze_data()
# 새로운 워크북 생성
with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
# 요약 시트
summary_df = pd.DataFrame([summary])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# 상세 데이터 시트
self.df.to_excel(writer, sheet_name='Detail', index=False)
# 월별 트렌드 시트
trend_df = pd.DataFrame(list(summary['monthly_trend'].items()),
columns=['Month', 'Sales'])
trend_df.to_excel(writer, sheet_name='Trend', index=False)
# 스타일 적용
self._apply_formatting(output_file)
print(f"✅ 보고서가 생성되었습니다: {output_file}")
def _apply_formatting(self, file_path):
"""엑셀 파일에 서식 적용"""
wb = load_workbook(file_path)
ws = wb['Summary']
# 헤더 스타일
header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
header_font = Font(color="FFFFFF", bold=True)
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal='center')
wb.save(file_path)
# 사용 예시
generator = ExcelReportGenerator('sales_data.xlsx')
generator.create_report(f'weekly_report_{datetime.now().strftime("%Y%m%d")}.xlsx')
이 프로그램으로 주간/월간 보고서 작성 시간을 몇 시간에서 몇 분으로 단축할 수 있습니다.
📧 이메일 자동화
6. 대량 이메일 발송 자동화
고객 맞춤형 이메일을 대량으로 발송해야 할 때, 파이썬으로 개인화된 내용을 자동으로 생성하고 발송할 수 있습니다.
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 pandas as pd
class EmailAutomation:
def __init__(self, smtp_server, smtp_port, email, password):
self.smtp_server = smtp_server
self.smtp_port = smtp_port
self.email = email
self.password = password
def send_personalized_emails(self, recipients_file, template):
"""개인화된 이메일 대량 발송"""
recipients = pd.read_excel(recipients_file)
server = smtplib.SMTP(self.smtp_server, self.smtp_port)
server.starttls()
server.login(self.email, self.password)
for _, recipient in recipients.iterrows():
# 템플릿에 개인 정보 삽입
personalized_content = template.format(
name=recipient['name'],
company=recipient['company'],
product=recipient['product']
)
msg = MIMEMultipart()
msg['From'] = self.email
msg['To'] = recipient['email']
msg['Subject'] = f"{recipient['name']}님께 특별한 제안이 있습니다"
msg.attach(MIMEText(personalized_content, 'html'))
server.send_message(msg)
print(f"✉️ 이메일 발송 완료: {recipient['email']}")
server.quit()
def send_with_attachment(self, to_email, subject, body, attachment_path):
"""첨부파일이 있는 이메일 발송"""
msg = MIMEMultipart()
msg['From'] = self.email
msg['To'] = to_email
msg['Subject'] = subject
msg.attach(MIMEText(body, 'plain'))
# 첨부파일 추가
with open(attachment_path, 'rb') as attachment:
part = MIMEBase('application', 'octet-stream')
part.set_payload(attachment.read())
encoders.encode_base64(part)
part.add_header(
'Content-Disposition',
f'attachment; filename= {attachment_path.split("/")[-1]}'
)
msg.attach(part)
with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
server.starttls()
server.login(self.email, self.password)
server.send_message(msg)
# 사용 예시
email_template = """
<html>
<body>
<h2>안녕하세요, {name}님!</h2>
<p>{company}에서 사용하시는 {product}에 대한 특별 업그레이드 제안을 드립니다.</p>
<p>지금 바로 확인해보세요!</p>
</body>
</html>
"""
automation = EmailAutomation(
smtp_server='smtp.gmail.com',
smtp_port=587,
email='your-email@gmail.com',
password='your-app-password'
)
automation.send_personalized_emails('recipients.xlsx', email_template)
이메일 마케팅, 뉴스레터 발송, 고객 후속 조치 등에 활용할 수 있습니다.
🖼️ 이미지 및 미디어 처리 자동화
7. 이미지 일괄 처리 자동화
수백 장의 이미지를 리사이징하거나, 워터마크를 추가하거나, 포맷을 변환해야 할 때 Pillow 라이브러리를 활용할 수 있습니다.
from PIL import Image, ImageDraw, ImageFont
import os
from pathlib import Path
class ImageProcessor:
def __init__(self, input_dir, output_dir):
self.input_dir = Path(input_dir)
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
def batch_resize(self, max_width=1920, max_height=1080):
"""이미지 일괄 리사이징"""
for img_file in self.input_dir.glob('*.{jpg,jpeg,png}'):
with Image.open(img_file) as img:
# 비율 유지하며 리사이징
img.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
output_path = self.output_dir / f"resized_{img_file.name}"
img.save(output_path, quality=95, optimize=True)
print(f"✓ Resized: {img_file.name}")
def add_watermark(self, watermark_text="© 2026 MyCompany"):
"""워터마크 일괄 추가"""
for img_file in self.input_dir.glob('*.{jpg,jpeg,png}'):
with Image.open(img_file) as img:
# 반투명 워터마크 추가
draw = ImageDraw.Draw(img)
# 폰트 설정 (시스템 폰트 사용)
try:
font = ImageFont.truetype("Arial.ttf", 40)
except:
font = ImageFont.load_default()
# 워터마크 위치 (오른쪽 하단)
text_bbox = draw.textbbox((0, 0), watermark_text, font=font)
text_width = text_bbox[2] - text_bbox[0]
text_height = text_bbox[3] - text_bbox[1]
x = img.width - text_width - 20
y = img.height - text_height - 20
# 반투명 배경
draw.rectangle([x-10, y-10, x+text_width+10, y+text_height+10],
fill=(0, 0, 0, 128))
draw.text((x, y), watermark_text, fill=(255, 255, 255, 200), font=font)
output_path = self.output_dir / f"watermarked_{img_file.name}"
img.save(output_path)
print(f"✓ Watermarked: {img_file.name}")
def convert_format(self, target_format='webp', quality=85):
"""이미지 포맷 일괄 변환"""
for img_file in self.input_dir.glob('*.{jpg,jpeg,png}'):
with Image.open(img_file) as img:
# RGB로 변환 (WebP는 RGBA를 지원하지만 호환성을 위해)
if img.mode in ('RGBA', 'P'):
img = img.convert('RGB')
output_path = self.output_dir / f"{img_file.stem}.{target_format}"
img.save(output_path, format=target_format.upper(), quality=quality)
print(f"✓ Converted: {img_file.name} → {output_path.name}")
# 사용 예시
processor = ImageProcessor('./raw_images', './processed_images')
processor.batch_resize(max_width=1920, max_height=1080)
processor.add_watermark("© 2026 MyBrand")
processor.convert_format('webp', quality=90)
블로그 이미지 최적화, 상품 사진 처리, SNS 업로드용 이미지 준비 등에 사용할 수 있습니다.
💾 백업 및 데이터 관리 자동화
8. 자동 백업 시스템
중요한 파일과 폴더를 자동으로 백업하는 시스템을 구축할 수 있습니다. 날짜별로 백업본을 관리하고, 오래된 백업은 자동으로 삭제하여 저장공간을 효율적으로 사용할 수 있습니다.
import shutil
import os
from datetime import datetime, timedelta
from pathlib import Path
import zipfile
class AutoBackup:
def __init__(self, source_dirs, backup_root):
self.source_dirs = [Path(d) for d in source_dirs]
self.backup_root = Path(backup_root)
self.backup_root.mkdir(exist_ok=True)
def create_backup(self):
"""전체 백업 수행"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"backup_{timestamp}"
backup_path = self.backup_root / backup_name
print(f"🔄 백업 시작: {backup_name}")
# ZIP 파일로 압축 백업
zip_path = self.backup_root / f"{backup_name}.zip"
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for source_dir in self.source_dirs:
if source_dir.exists():
for file_path in source_dir.rglob('*'):
if file_path.is_file():
# 상대 경로로 저장
arcname = file_path.relative_to(source_dir.parent)
zipf.write(file_path, arcname)
print(f" ✓ {file_path.name}")
file_size = zip_path.stat().st_size / (1024 * 1024) # MB
print(f"✅ 백업 완료: {zip_path.name} ({file_size:.2f} MB)")
return zip_path
def cleanup_old_backups(self, keep_days=30):
"""오래된 백업 파일 삭제"""
cutoff_date = datetime.now() - timedelta(days=keep_days)
deleted_count = 0
for backup_file in self.backup_root.glob('backup_*.zip'):
file_time = datetime.fromtimestamp(backup_file.stat().st_mtime)
if file_time < cutoff_date:
backup_file.unlink()
deleted_count += 1
print(f"🗑️ 삭제: {backup_file.name}")
print(f"✅ {deleted_count}개의 오래된 백업 삭제 완료")
def incremental_backup(self, last_backup_time):
"""증분 백업 (마지막 백업 이후 변경된 파일만)"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"incremental_{timestamp}"
zip_path = self.backup_root / f"{backup_name}.zip"
changed_files = []
for source_dir in self.source_dirs:
if source_dir.exists():
for file_path in source_dir.rglob('*'):
if file_path.is_file():
file_mtime = datetime.fromtimestamp(file_path.stat().st_mtime)
if file_mtime > last_backup_time:
changed_files.append(file_path)
if changed_files:
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for file_path in changed_files:
arcname = file_path.relative_to(file_path.parent.parent)
zipf.write(file_path, arcname)
print(f"✅ 증분 백업 완료: {len(changed_files)}개 파일")
else:
print("ℹ️ 변경된 파일이 없습니다.")
# 사용 예시
backup_system = AutoBackup(
source_dirs=[
'/Users/username/Documents',
'/Users/username/Projects',
'/Users/username/Photos'
],
backup_root='/Volumes/ExternalDrive/Backups'
)
# 백업 실행
backup_system.create_backup()
backup_system.cleanup_old_backups(keep_days=30)
이 시스템을 cron이나 Task Scheduler에 등록하면 매일 자동으로 백업이 실행됩니다.
🤖 API 연동 및 소셜 미디어 자동화
9. 소셜 미디어 자동 포스팅
여러 소셜 미디어 플랫폼에 동시에 콘텐츠를 게시하는 자동화 시스템을 만들 수 있습니다.
import tweepy
import requests
from datetime import datetime, time
import schedule
class SocialMediaAutomation:
def __init__(self, twitter_credentials, linkedin_token):
# Twitter API 설정
auth = tweepy.OAuthHandler(
twitter_credentials['api_key'],
twitter_credentials['api_secret']
)
auth.set_access_token(
twitter_credentials['access_token'],
twitter_credentials['access_secret']
)
self.twitter_api = tweepy.API(auth)
self.linkedin_token = linkedin_token
def post_to_twitter(self, content, image_path=None):
"""트위터에 포스팅"""
try:
if image_path:
media = self.twitter_api.media_upload(image_path)
self.twitter_api.update_status(content, media_ids=[media.media_id])
else:
self.twitter_api.update_status(content)
print("✅ Twitter 포스팅 완료")
except Exception as e:
print(f"❌ Twitter 포스팅 실패: {e}")
def post_to_linkedin(self, content):
"""LinkedIn에 포스팅"""
url = "https://api.linkedin.com/v2/ugcPosts"
headers = {
'Authorization': f'Bearer {self.linkedin_token}',
'Content-Type': 'application/json'
}
payload = {
"author": "urn:li:person:YOUR_PERSON_ID",
"lifecycleState": "PUBLISHED",
"specificContent": {
"com.linkedin.ugc.ShareContent": {
"shareCommentary": {
"text": content
},
"shareMediaCategory": "NONE"
}
},
"visibility": {
"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"
}
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 201:
print("✅ LinkedIn 포스팅 완료")
else:
print(f"❌ LinkedIn 포스팅 실패: {response.text}")
def schedule_posts(self, post_schedule):
"""예약 포스팅 설정"""
for post_info in post_schedule:
schedule.every().day.at(post_info['time']).do(
self.cross_post,
content=post_info['content'],
platforms=post_info['platforms']
)
print("📅 포스팅 스케줄 설정 완료")
# 스케줄 실행
while True:
schedule.run_pending()
time.sleep(60)
def cross_post(self, content, platforms=['twitter', 'linkedin']):
"""여러 플랫폼에 동시 포스팅"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"\n📤 포스팅 시작 ({timestamp})")
if 'twitter' in platforms:
self.post_to_twitter(content)
if 'linkedin' in platforms:
self.post_to_linkedin(content)
# 사용 예시
twitter_creds = {
'api_key': 'YOUR_API_KEY',
'api_secret': 'YOUR_API_SECRET',
'access_token': 'YOUR_ACCESS_TOKEN',
'access_secret': 'YOUR_ACCESS_SECRET'
}
automation = SocialMediaAutomation(twitter_creds, 'LINKEDIN_TOKEN')
# 예약 포스팅 설정
post_schedule = [
{
'time': '09:00',
'content': '좋은 아침입니다! 오늘의 기술 팁을 공유합니다. #Python #Automation',
'platforms': ['twitter', 'linkedin']
},
{
'time': '18:00',
'content': '오늘 하루도 수고하셨습니다! 내일 뵙겠습니다.',
'platforms': ['twitter']
}
]
automation.schedule_posts(post_schedule)
블로그 글 발행 시 자동으로 SNS에 공유하거나, 정기적인 콘텐츠 배포를 자동화할 수 있습니다.
🖥️ 시스템 모니터링 및 알림 자동화
10. 서버 및 웹사이트 모니터링
서버나 웹사이트의 상태를 지속적으로 모니터링하고, 문제가 발생하면 즉시 알림을 받을 수 있습니다.
import requests
import psutil
import time
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
class SystemMonitor:
def __init__(self, alert_email=None):
self.alert_email = alert_email
self.alerts_sent = {}
def check_website_status(self, url, timeout=10):
"""웹사이트 응답 상태 확인"""
try:
response = requests.get(url, timeout=timeout)
response_time = response.elapsed.total_seconds()
status = {
'url': url,
'status_code': response.status_code,
'response_time': response_time,
'is_up': response.status_code == 200,
'timestamp': datetime.now()
}
if not status['is_up']:
self.send_alert(f"⚠️ 웹사이트 다운: {url} (상태 코드: {response.status_code})")
elif response_time > 3.0:
self.send_alert(f"🐌 느린 응답: {url} (응답 시간: {response_time:.2f}초)")
return status
except requests.RequestException as e:
self.send_alert(f"❌ 연결 실패: {url} - {str(e)}")
return {'url': url, 'is_up': False, 'error': str(e)}
def check_system_resources(self):
"""시스템 리소스 사용률 확인"""
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
status = {
'cpu_percent': cpu_percent,
'memory_percent': memory.percent,
'memory_available_gb': memory.available / (1024**3),
'disk_percent': disk.percent,
'disk_free_gb': disk.free / (1024**3),
'timestamp': datetime.now()
}
# 임계값 체크
if cpu_percent > 90:
self.send_alert(f"🔥 CPU 사용률 높음: {cpu_percent}%")
if memory.percent > 90:
self.send_alert(f"💾 메모리 사용률 높음: {memory.percent}%")
if disk.percent > 90:
self.send_alert(f"📀 디스크 사용률 높음: {disk.percent}%")
return status
def send_alert(self, message):
"""알림 발송 (중복 방지)"""
# 같은 알림을 10분 이내에 재발송하지 않음
current_time = datetime.now()
if message in self.alerts_sent:
last_sent = self.alerts_sent[message]
if (current_time - last_sent).seconds < 600:
return
self.alerts_sent[message] = current_time
print(f"🚨 {message}")
# 이메일 알림 (선택적)
if self.alert_email:
self._send_email_alert(message)
def _send_email_alert(self, message):
"""이메일로 알림 전송"""
# 이메일 발송 로직
pass
def monitor_loop(self, urls, check_interval=60):
"""지속적인 모니터링 루프"""
print(f"🔍 모니터링 시작 (체크 간격: {check_interval}초)")
while True:
print(f"\n📊 상태 체크 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# 웹사이트 체크
for url in urls:
status = self.check_website_status(url)
if status.get('is_up'):
print(f" ✅ {url}: {status['response_time']:.2f}초")
else:
print(f" ❌ {url}: 다운")
# 시스템 리소스 체크
system_status = self.check_system_resources()
print(f" 💻 CPU: {system_status['cpu_percent']:.1f}% | "
f"메모리: {system_status['memory_percent']:.1f}% | "
f"디스크: {system_status['disk_percent']:.1f}%")
time.sleep(check_interval)
# 사용 예시
monitor = SystemMonitor(alert_email='admin@example.com')
# 모니터링할 웹사이트 목록
websites = [
'https://mywebsite.com',
'https://api.myservice.com/health',
'https://blog.mysite.com'
]
# 모니터링 시작 (60초마다 체크)
monitor.monitor_loop(websites, check_interval=60)
서버 다운타임을 최소화하고, 문제를 조기에 발견하여 대응할 수 있습니다.
⏰ 작업 스케줄링 자동화
11. 스케줄 기반 작업 실행기
정해진 시간에 자동으로 작업을 실행하는 스케줄러를 만들 수 있습니다. schedule 라이브러리를 사용하면 간단하게 구현할 수 있습니다.
import schedule
import time
from datetime import datetime
import subprocess
class TaskScheduler:
def __init__(self):
self.tasks = []
def add_daily_task(self, time_str, task_func, description=""):
"""매일 특정 시간에 실행되는 작업 추가"""
schedule.every().day.at(time_str).do(task_func)
self.tasks.append({
'type': 'daily',
'time': time_str,
'description': description
})
print(f"📅 일일 작업 추가: {description} at {time_str}")
def add_interval_task(self, interval_minutes, task_func, description=""):
"""일정 간격으로 실행되는 작업 추가"""
schedule.every(interval_minutes).minutes.do(task_func)
self.tasks.append({
'type': 'interval',
'interval': f"{interval_minutes}분",
'description': description
})
print(f"⏱️ 주기 작업 추가: {description} every {interval_minutes}분")
def add_weekly_task(self, day, time_str, task_func, description=""):
"""매주 특정 요일에 실행되는 작업 추가"""
getattr(schedule.every(), day.lower()).at(time_str).do(task_func)
self.tasks.append({
'type': 'weekly',
'day': day,
'time': time_str,
'description': description
})
print(f"📆 주간 작업 추가: {description} on {day} at {time_str}")
def list_tasks(self):
"""등록된 작업 목록 출력"""
print("\n📋 등록된 작업 목록:")
print("=" * 60)
for i, task in enumerate(self.tasks, 1):
print(f"{i}. {task['description']}")
if task['type'] == 'daily':
print(f" ⏰ 매일 {task['time']}")
elif task['type'] == 'interval':
print(f" ⏰ {task['interval']}마다")
elif task['type'] == 'weekly':
print(f" ⏰ 매주 {task['day']} {task['time']}")
print("=" * 60)
def run(self):
"""스케줄러 실행"""
print(f"\n🚀 스케줄러 시작: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
self.list_tasks()
try:
while True:
schedule.run_pending()
time.sleep(1)
except KeyboardInterrupt:
print("\n⏹️ 스케줄러 종료")
# 실제 작업 함수들
def backup_databases():
"""데이터베이스 백업"""
print(f"[{datetime.now()}] 💾 데이터베이스 백업 시작...")
# 백업 로직
print("✅ 백업 완료")
def send_daily_report():
"""일일 보고서 발송"""
print(f"[{datetime.now()}] 📊 일일 보고서 생성 및 발송...")
# 보고서 생성 로직
print("✅ 보고서 발송 완료")
def clean_temp_files():
"""임시 파일 정리"""
print(f"[{datetime.now()}] 🧹 임시 파일 정리...")
# 파일 정리 로직
print("✅ 정리 완료")
def check_server_health():
"""서버 상태 체크"""
print(f"[{datetime.now()}] 🏥 서버 상태 체크...")
# 상태 확인 로직
print("✅ 서버 정상")
# 사용 예시
scheduler = TaskScheduler()
# 매일 새벽 2시에 데이터베이스 백업
scheduler.add_daily_task("02:00", backup_databases, "데이터베이스 자동 백업")
# 매일 오전 9시에 일일 보고서 발송
scheduler.add_daily_task("09:00", send_daily_report, "일일 보고서 자동 발송")
# 매일 자정에 임시 파일 정리
scheduler.add_daily_task("00:00", clean_temp_files, "임시 파일 자동 정리")
# 10분마다 서버 상태 체크
scheduler.add_interval_task(10, check_server_health, "서버 상태 모니터링")
# 매주 일요일 오후 3시에 주간 보고서
scheduler.add_weekly_task("Sunday", "15:00",
lambda: print("📈 주간 보고서 생성"),
"주간 보고서 자동 생성")
# 스케줄러 실행
scheduler.run()
이 스케줄러를 백그라운드 프로세스로 실행하면 정해진 시간에 자동으로 모든 작업이 수행됩니다.
🛠️ 자동화 프로그램 구축 시 유용한 팁
1. 필수 라이브러리 모음
파이썬 자동화에 자주 사용되는 핵심 라이브러리들입니다:
# 파일 및 시스템
pip install pathlib shutil psutil
# 웹 스크래핑
pip install requests beautifulsoup4 selenium
# 데이터 처리
pip install pandas openpyxl xlsxwriter
# 이미지 처리
pip install Pillow
# 스케줄링
pip install schedule APScheduler
# 이메일
pip install smtplib email
# API 연동
pip install tweepy python-linkedin-v2
# 데이터베이스
pip install sqlalchemy pymongo redis
# 로깅 및 모니터링
pip install loguru
2. 오류 처리 및 로깅
안정적인 자동화 프로그램을 위해서는 적절한 오류 처리와 로깅이 필수입니다:
import logging
from datetime import datetime
import traceback
# 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(f'automation_{datetime.now().strftime("%Y%m%d")}.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def safe_automation(func):
"""자동화 함수를 안전하게 실행하는 데코레이터"""
def wrapper(*args, **kwargs):
try:
logger.info(f"작업 시작: {func.__name__}")
result = func(*args, **kwargs)
logger.info(f"작업 완료: {func.__name__}")
return result
except Exception as e:
logger.error(f"작업 실패: {func.__name__}")
logger.error(f"오류 내용: {str(e)}")
logger.error(f"스택 트레이스:\n{traceback.format_exc()}")
# 오류 알림 발송
send_error_notification(func.__name__, str(e))
return None
return wrapper
@safe_automation
def critical_task():
"""중요한 자동화 작업"""
# 작업 로직
pass
3. 환경 변수 및 설정 관리
민감한 정보(API 키, 비밀번호 등)는 환경 변수나 설정 파일로 관리하세요:
import os
from dotenv import load_dotenv
import json
# .env 파일에서 환경 변수 로드
load_dotenv()
class Config:
"""설정 관리 클래스"""
# 환경 변수에서 읽기
SMTP_SERVER = os.getenv('SMTP_SERVER', 'smtp.gmail.com')
SMTP_PORT = int(os.getenv('SMTP_PORT', 587))
EMAIL = os.getenv('EMAIL')
EMAIL_PASSWORD = os.getenv('EMAIL_PASSWORD')
# API 키
TWITTER_API_KEY = os.getenv('TWITTER_API_KEY')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
# 경로 설정
BACKUP_DIR = os.getenv('BACKUP_DIR', '/path/to/backups')
LOG_DIR = os.getenv('LOG_DIR', './logs')
@classmethod
def validate(cls):
"""필수 설정값 확인"""
required = ['EMAIL', 'EMAIL_PASSWORD']
missing = [key for key in required if not getattr(cls, key)]
if missing:
raise ValueError(f"필수 환경 변수가 설정되지 않음: {', '.join(missing)}")
# .env 파일 예시
"""
SMTP_SERVER=smtp.gmail.com
SMTP_PORT=587
EMAIL=your-email@gmail.com
EMAIL_PASSWORD=your-app-password
TWITTER_API_KEY=your-twitter-key
BACKUP_DIR=/Volumes/Backup
"""
4. 실행 환경 설정
macOS/Linux - cron 설정
# crontab 편집
crontab -e
# 매일 오전 9시에 스크립트 실행
0 9 * * * /usr/bin/python3 /path/to/your/automation.py
# 매시간 실행
0 * * * * /usr/bin/python3 /path/to/your/monitor.py
# 매주 일요일 새벽 2시에 백업
0 2 * * 0 /usr/bin/python3 /path/to/your/backup.py
Windows - Task Scheduler
# PowerShell에서 작업 스케줄러 생성
$action = New-ScheduledTaskAction -Execute "python.exe" -Argument "C:\path\to\automation.py"
$trigger = New-ScheduledTaskTrigger -Daily -At 9am
Register-ScheduledTask -Action $action -Trigger $trigger -TaskName "DailyAutomation"
5. 성능 최적화
대량의 데이터를 처리할 때는 병렬 처리를 활용하세요:
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import multiprocessing
def process_file(file_path):
"""파일 처리 함수"""
# 처리 로직
return result
# 멀티스레딩 (I/O 바운드 작업)
with ThreadPoolExecutor(max_workers=10) as executor:
results = executor.map(process_file, file_list)
# 멀티프로세싱 (CPU 바운드 작업)
with ProcessPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
results = executor.map(heavy_computation, data_list)
📊 자동화 도입 효과
실제로 이러한 자동화 프로그램들을 도입하면 다음과 같은 효과를 기대할 수 있습니다:
| 작업 유형 | 수동 작업 시간 | 자동화 후 시간 | 절약 시간 | 생산성 향상 |
|---|---|---|---|---|
| 파일 정리 | 30분/일 | 0분 | 3.5시간/주 | 100% |
| 이미지 처리 (100장) | 2시간 | 5분 | 1시간 55분 | 96% |
| 데이터 수집 | 1시간/일 | 5분/일 | 6.5시간/주 | 92% |
| 보고서 작성 | 3시간/주 | 10분/주 | 2시간 50분/주 | 94% |
| 백업 작업 | 20분/일 | 0분 | 2.3시간/주 | 100% |
월간 절약 시간: 약 64시간 (8 근무일 상당)
🎯 결론
파이썬 자동화는 단순히 시간을 절약하는 것을 넘어서, 반복적인 작업에서 해방되어 더 창의적이고 가치 있는 일에 집중할 수 있게 해줍니다. 이 글에서 소개한 11가지 자동화 프로그램은 실무에서 즉시 활용할 수 있는 예시들이며, 여러분의 필요에 맞게 커스터마이징할 수 있습니다.
시작하기 위한 행동 계획
- 현재 업무 분석: 가장 많은 시간을 소비하는 반복 작업을 찾으세요
- 작은 것부터 시작: 파일 정리나 이름 변경 같은 간단한 작업부터 자동화하세요
- 점진적 확장: 성공 경험을 바탕으로 더 복잡한 자동화로 확장하세요
- 문서화: 작성한 스크립트를 문서화하여 팀과 공유하세요
- 지속적 개선: 자동화 프로그램을 계속 개선하고 최적화하세요
파이썬의 강력함과 풍부한 생태계를 활용하면, 여러분만의 맞춤형 자동화 도구를 만들 수 있습니다. 오늘 당장 작은 것부터 시작해보세요. 여러분의 일상이 얼마나 편해질 수 있는지 직접 경험하게 될 것입니다!
추가 학습 자료:
GitHub 저장소: 이 글의 모든 예제 코드는 GitHub에서 확인할 수 있습니다.