기본 클래스 구조 설계
먼저 재사용 가능하고 확장성 있는 기본 클래스를 설계합니다.
pythonimport requests
import logging
import time
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
import json
import os
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class SnipeITAPI:
"""Snipe-IT API 통합 클래스"""
def __init__(self, base_url: str, token: str, timeout: int = 30):
self.base_url = base_url.rstrip('/')
self.api_base = f"{self.base_url}/api/v1"
self.timeout = timeout
# 세션 설정 (연결 풀링 및 재시도 로직)
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {token}',
'Accept': 'application/json',
'Content-Type': 'application/json'
})
# 재시도 전략 설정
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
# 로깅 설정
self.logger = self._setup_logging()
def _setup_logging(self) -> logging.Logger:
"""로깅 설정"""
logger = logging.getLogger('SnipeIT_API')
logger.setLevel(logging.INFO)
if not logger.handlers:
# 파일 핸들러
file_handler = logging.FileHandler(
f'snipe_it_api_{datetime.now().strftime("%Y%m%d")}.log'
)
file_handler.setLevel(logging.INFO)
# 콘솔 핸들러
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 포맷터
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
def _make_request(self, method: str, endpoint: str,
data: Dict = None, params: Dict = None) -> Dict:
"""통합 요청 처리 메소드"""
url = f"{self.api_base}/{endpoint.lstrip('/')}"
try:
self.logger.info(f"{method.upper()} 요청: {url}")
response = self.session.request(
method=method.upper(),
url=url,
json=data,
params=params,
timeout=self.timeout
)
# 응답 로깅
self.logger.info(f"응답 상태: {response.status_code}")
if response.status_code >= 400:
self.logger.error(f"오류 응답: {response.text}")
return self._handle_response(response)
except requests.exceptions.RequestException as e:
self.logger.error(f"요청 실패: {str(e)}")
raise
def _handle_response(self, response: requests.Response) -> Dict:
"""응답 처리 및 오류 핸들링"""
try:
json_response = response.json()
except json.JSONDecodeError:
json_response = {'raw_content': response.text}
result = {
'status_code': response.status_code,
'success': response.status_code < 400,
'data': json_response
}
if not result['success']:
error_msg = self._extract_error_message(json_response)
result['error'] = error_msg
self.logger.error(f"API 오류: {error_msg}")
return result
def _extract_error_message(self, response_data: Dict) -> str:
"""응답에서 오류 메시지 추출"""
if isinstance(response_data, dict):
# Snipe-IT 표준 오류 형식
if 'messages' in response_data:
messages = response_data['messages']
if isinstance(messages, dict):
return '; '.join([f"{k}: {v}" for k, v in messages.items()])
return str(messages)
if 'message' in response_data:
return response_data['message']
if 'error' in response_data:
return response_data['error']
return str(response_data)대량 자산 등록 스크립트
CSV 파일에서 자산 정보를 읽어 대량으로 등록하는 스크립트입니다.
pythonimport pandas as pd
from typing import List, Dict
import time
class BulkAssetManager(SnipeITAPI):
"""대량 자산 관리 클래스"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.batch_size = 10 # 배치 크기
self.delay_between_batches = 1 # 배치 간 대기 시간 (초)
def load_assets_from_csv(self, file_path: str) -> List[Dict]:
"""CSV 파일에서 자산 데이터 로드"""
try:
df = pd.read_csv(file_path)
# 필수 컬럼 확인
required_columns = ['name', 'asset_tag', 'model_id', 'status_id']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
raise ValueError(f"필수 컬럼 누락: {missing_columns}")
# NaN 값을 None으로 변환
df = df.where(pd.notnull(df), None)
assets = df.to_dict('records')
self.logger.info(f"CSV에서 {len(assets)}개의 자산 데이터 로드 완료")
return assets
except Exception as e:
self.logger.error(f"CSV 로드 실패: {str(e)}")
raise
def validate_asset_data(self, asset_data: Dict) -> Dict:
"""자산 데이터 유효성 검사 및 전처리"""
# 필수 필드 확인
required_fields = ['name', 'asset_tag', 'model_id', 'status_id']
for field in required_fields:
if not asset_data.get(field):
raise ValueError(f"필수 필드 누락: {field}")
# 날짜 형식 변환
if asset_data.get('purchase_date'):
try:
# 다양한 날짜 형식 지원
purchase_date = pd.to_datetime(asset_data['purchase_date']).strftime('%Y-%m-%d')
asset_data['purchase_date'] = purchase_date
except:
self.logger.warning(f"잘못된 날짜 형식: {asset_data['purchase_date']}")
asset_data['purchase_date'] = None
# 가격 형식 변환
if asset_data.get('purchase_cost'):
try:
# 문자열에서 숫자만 추출
cost_str = str(asset_data['purchase_cost']).replace(',', '').replace('원', '')
asset_data['purchase_cost'] = float(cost_str)
except:
self.logger.warning(f"잘못된 가격 형식: {asset_data['purchase_cost']}")
asset_data['purchase_cost'] = None
return asset_data
def create_assets_batch(self, assets: List[Dict]) -> Dict:
"""자산 배치 생성"""
results = {
'success': [],
'failed': [],
'total': len(assets)
}
for i in range(0, len(assets), self.batch_size):
batch = assets[i:i+self.batch_size]
self.logger.info(f"배치 {i//self.batch_size + 1} 처리 중 ({len(batch)}개 자산)")
for asset in batch:
try:
# 데이터 유효성 검사
validated_asset = self.validate_asset_data(asset.copy())
# API 호출
response = self._make_request('POST', '/hardware', data=validated_asset)
if response['success']:
results['success'].append({
'asset_tag': validated_asset['asset_tag'],
'id': response['data']['payload']['id']
})
self.logger.info(f"자산 생성 성공: {validated_asset['asset_tag']}")
else:
results['failed'].append({
'asset_tag': validated_asset['asset_tag'],
'error': response.get('error', 'Unknown error')
})
except Exception as e:
results['failed'].append({
'asset_tag': asset.get('asset_tag', 'Unknown'),
'error': str(e)
})
self.logger.error(f"자산 생성 실패: {str(e)}")
# 배치 간 대기
if i + self.batch_size < len(assets):
time.sleep(self.delay_between_batches)
# 결과 요약
success_count = len(results['success'])
failed_count = len(results['failed'])
self.logger.info(f"배치 처리 완료 - 성공: {success_count}, 실패: {failed_count}")
return results
def process_csv_file(self, file_path: str) -> Dict:
"""CSV 파일 전체 처리"""
try:
# CSV 로드
assets = self.load_assets_from_csv(file_path)
# 배치 처리
results = self.create_assets_batch(assets)
# 결과 저장
self.save_results(results, f"bulk_import_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
return results
except Exception as e:
self.logger.error(f"CSV 처리 실패: {str(e)}")
raise
def save_results(self, results: Dict, filename: str):
"""결과를 JSON 파일로 저장"""
try:
with open(filename, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
self.logger.info(f"결과 저장 완료: {filename}")
except Exception as e:
self.logger.error(f"결과 저장 실패: {str(e)}")자산 현황 리포트 자동 생성
정기적으로 자산 현황을 분석하고 리포트를 생성하는 스크립트입니다.
pythonimport pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import os
class AssetReportGenerator(SnipeITAPI):
"""자산 리포트 생성 클래스"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.report_date = datetime.now()
def get_all_assets(self) -> List[Dict]:
"""모든 자산 정보 조회"""
all_assets = []
limit = 500
offset = 0
while True:
params = {
'limit': limit,
'offset': offset,
'sort': 'id',
'order': 'asc'
}
response = self._make_request('GET', '/hardware', params=params)
if not response['success']:
self.logger.error("자산 조회 실패")
break
assets = response['data']['rows']
all_assets.extend(assets)
# 마지막 페이지 확인
if len(assets) < limit:
break
offset += limit
self.logger.info(f"자산 조회 진행 중... ({len(all_assets)}개)")
self.logger.info(f"총 {len(all_assets)}개 자산 조회 완료")
return all_assets
def get_users_with_assets(self) -> Dict:
"""사용자별 할당 자산 현황"""
users_response = self._make_request('GET', '/users')
if not users_response['success']:
return {}
users = users_response['data']['rows']
user_assets = {}
for user in users:
user_id = user['id']
assets_response = self._make_request('GET', f'/users/{user_id}/assets')
if assets_response['success']:
assets = assets_response['data']['rows']
if assets: # 할당된 자산이 있는 사용자만
user_assets[user['name']] = {
'user_info': user,
'assets': assets,
'asset_count': len(assets)
}
return user_assets
def analyze_asset_status(self, assets: List[Dict]) -> Dict:
"""자산 상태별 분석"""
status_analysis = {}
for asset in assets:
status = asset['status_label']['name']
if status not in status_analysis:
status_analysis[status] = {
'count': 0,
'total_value': 0,
'assets': []
}
status_analysis[status]['count'] += 1
status_analysis[status]['assets'].append(asset)
# 구매 금액 합계
if asset.get('purchase_cost'):
try:
cost = float(asset['purchase_cost'])
status_analysis[status]['total_value'] += cost
except:
pass
return status_analysis
def analyze_by_category(self, assets: List[Dict]) -> Dict:
"""카테고리별 분석"""
category_analysis = {}
for asset in assets:
if asset.get('model') and asset['model'].get('category'):
category = asset['model']['category']['name']
if category not in category_analysis:
category_analysis[category] = {
'count': 0,
'deployed': 0,
'available': 0,
'total_value': 0
}
category_analysis[category]['count'] += 1
# 상태별 집계
status_type = asset['status_label'].get('status_type', '')
if status_type == 'deployable':
category_analysis[category]['available'] += 1
elif status_type == 'deployed':
category_analysis[category]['deployed'] += 1
# 구매 금액
if asset.get('purchase_cost'):
try:
cost = float(asset['purchase_cost'])
category_analysis[category]['total_value'] += cost
except:
pass
return category_analysis
def generate_aging_report(self, assets: List[Dict]) -> Dict:
"""자산 노후화 분석"""
aging_data = {
'less_than_1_year': 0,
'1_to_3_years': 0,
'3_to_5_years': 0,
'more_than_5_years': 0,
'unknown': 0
}
current_date = datetime.now()
for asset in assets:
purchase_date = asset.get('purchase_date')
if not purchase_date:
aging_data['unknown'] += 1
continue
try:
purchase_dt = datetime.strptime(purchase_date.split('T')[0], '%Y-%m-%d')
age_years = (current_date - purchase_dt).days / 365.25
if age_years < 1:
aging_data['less_than_1_year'] += 1
elif age_years < 3:
aging_data['1_to_3_years'] += 1
elif age_years < 5:
aging_data['3_to_5_years'] += 1
else:
aging_data['more_than_5_years'] += 1
except:
aging_data['unknown'] += 1
return aging_data
def create_visualizations(self, analysis_data: Dict):
"""시각화 차트 생성"""
plt.style.use('seaborn-v0_8')
fig, axes = plt.subplots(2, 2, figsize=(15, 12))
fig.suptitle(f'자산 현황 분석 리포트 - {self.report_date.strftime("%Y-%m-%d")}', fontsize=16)
# 1. 상태별 자산 분포
status_data = analysis_data['status_analysis']
status_names = list(status_data.keys())
status_counts = [data['count'] for data in status_data.values()]
axes[0,0].pie(status_counts, labels=status_names, autopct='%1.1f%%')
axes[0,0].set_title('자산 상태별 분포')
# 2. 카테고리별 자산 수량
category_data = analysis_data['category_analysis']
if category_data:
category_names = list(category_data.keys())
category_counts = [data['count'] for data in category_data.values()]
axes[0,1].bar(category_names, category_counts)
axes[0,1].set_title('카테고리별 자산 수량')
axes[0,1].tick_params(axis='x', rotation=45)
# 3. 자산 노후화 현황
aging_data = analysis_data['aging_analysis']
aging_labels = ['1년 미만', '1-3년', '3-5년', '5년 이상', '미상']
aging_values = list(aging_data.values())
axes[1,0].bar(aging_labels, aging_values, color=['green', 'yellow', 'orange', 'red', 'gray'])
axes[1,0].set_title('자산 노후화 현황')
axes[1,0].tick_params(axis='x', rotation=45)
# 4. 월별 구매 트렌드 (최근 12개월)
# 이 부분은 실제 데이터에 따라 구현
axes[1,1].text(0.5, 0.5, '월별 구매 트렌드\n(데이터 준비 중)',
ha='center', va='center', transform=axes[1,1].transAxes)
plt.tight_layout()
# 차트 저장
chart_filename = f'asset_report_charts_{self.report_date.strftime("%Y%m%d")}.png'
plt.savefig(chart_filename, dpi=300, bbox_inches='tight')
plt.close()
self.logger.info(f"차트 저장 완료: {chart_filename}")
return chart_filename
def generate_excel_report(self, analysis_data: Dict) -> str:
"""Excel 리포트 생성"""
filename = f'snipe_it_asset_report_{self.report_date.strftime("%Y%m%d")}.xlsx'
with pd.ExcelWriter(filename, engine='openpyxl') as writer:
# 1. 요약 시트
summary_data = {
'구분': ['총 자산 수', '배포 가능', '배포됨', '수리 중', '폐기'],
'수량': [
analysis_data['total_assets'],
analysis_data['status_analysis'].get('Ready to Deploy', {}).get('count', 0),
analysis_data['status_analysis'].get('Deployed', {}).get('count', 0),
analysis_data['status_analysis'].get('Out for Repair', {}).get('count', 0),
analysis_data['status_analysis'].get('Archived', {}).get('count', 0)
]
}
summary_df = pd.DataFrame(summary_data)
summary_df.to_excel(writer, sheet_name='요약', index=False)
# 2. 상태별 상세
status_details = []
for status, data in analysis_data['status_analysis'].items():
status_details.append({
'상태': status,
'수량': data['count'],
'총 가치': data['total_value']
})
status_df = pd.DataFrame(status_details)
status_df.to_excel(writer, sheet_name='상태별 현황', index=False)
# 3. 카테고리별 현황
if analysis_data['category_analysis']:
category_details = []
for category, data in analysis_data['category_analysis'].items():
category_details.append({
'카테고리': category,
'총 수량': data['count'],
'배포됨': data['deployed'],
'사용 가능': data['available'],
'총 가치': data['total_value']
})
category_df = pd.DataFrame(category_details)
category_df.to_excel(writer, sheet_name='카테고리별 현황', index=False)
# 4. 사용자별 할당 현황
if analysis_data.get('user_assets'):
user_details = []
for user_name, data in analysis_data['user_assets'].items():
user_details.append({
'사용자': user_name,
'부서': data['user_info'].get('department', {}).get('name', ''),
'할당 자산 수': data['asset_count']
})
user_df = pd.DataFrame(user_details)
user_df.to_excel(writer, sheet_name='사용자별 할당', index=False)
self.logger.info(f"Excel 리포트 생성 완료: {filename}")
return filename
def generate_full_report(self) -> Dict:
"""전체 리포트 생성"""
try:
self.logger.info("자산 리포트 생성 시작")
# 1. 데이터 수집
assets = self.get_all_assets()
user_assets = self.get_users_with_assets()
# 2. 데이터 분석
status_analysis = self.analyze_asset_status(assets)
category_analysis = self.analyze_by_category(assets)
aging_analysis = self.generate_aging_report(assets)
analysis_data = {
'total_assets': len(assets),
'status_analysis': status_analysis,
'category_analysis': category_analysis,
'aging_analysis': aging_analysis,
'user_assets': user_assets
}
# 3. 시각화 생성
chart_file = self.create_visualizations(analysis_data)
# 4. Excel 리포트 생성
excel_file = self.generate_excel_report(analysis_data)
# 5. 결과 반환
return {
'success': True,
'report_date': self.report_date.isoformat(),
'total_assets': len(assets),
'files': {
'excel': excel_file,
'chart': chart_file
},
'analysis': analysis_data
}
except Exception as e:
self.logger.error(f"리포트 생성 실패: {str(e)}")
return {
'success': False,
'error': str(e)
}사용자별 할당 자산 조회 및 관리
pythonclass UserAssetManager(SnipeITAPI):
"""사용자별 자산 관리 클래스"""
def get_user_by_email(self, email: str) -> Optional[Dict]:
"""이메일로 사용자 검색"""
response = self._make_request('GET', '/users', params={'search': email})
if response['success'] and response['data']['rows']:
for user in response['data']['rows']:
if user.get('email', '').lower() == email.lower():
return user
return None
def get_user_assets_detailed(self, user_id: int) -> Dict:
"""사용자별 자산 상세 정보"""
response = self._make_request('GET', f'/users/{user_id}/assets')
if not response['success']:
return {'success': False, 'error': response.get('error')}
assets = response['data']['rows']
# 자산 정보 상세화
detailed_assets = []
total_value = 0
for asset in assets:
asset_detail = {
'id': asset['id'],
'name': asset['name'],
'asset_tag': asset['asset_tag'],
'model': asset.get('model', {}).get('name', ''),
'manufacturer': asset.get('model', {}).get('manufacturer', {}).get('name', ''),
'purchase_date': asset.get('purchase_date'),
'purchase_cost': asset.get('purchase_cost', 0),
'checkout_date': asset.get('assigned_to', {}).get('checkout_at'),
'expected_checkin': asset.get('expected_checkin'),
'status': asset.get('status_label', {}).get('name', '')
}
detailed_assets.append(asset_detail)
# 총 가치 계산
try:
cost = float(asset_detail['purchase_cost'] or 0)
total_value += cost
except:
pass
return {
'success': True,
'user_id': user_id,
'total_assets': len(detailed_assets),
'total_value': total_value,
'assets': detailed_assets
}
def bulk_checkout_assets(self, assignments: List[Dict]) -> Dict:
"""대량 자산 할당"""
results = {
'success': [],
'failed': [],
'total': len(assignments)
}
for assignment in assignments:
try:
asset_id = assignment['asset_id']
user_id = assignment['user_id']
checkout_data = {
'assigned_user': user_id,
'checkout_at': assignment.get('checkout_at', datetime.now().strftime('%Y-%m-%d %H:%M:%S')),
'expected_checkin': assignment.get('expected_checkin'),
'note': assignment.get('note', '')
}
response = self._make_request('POST', f'/hardware/{asset_id}/checkout', data=checkout_data)
if response['success']:
results['success'].append({
'asset_id': asset_id,
'user_id': user_id,
'checkout_date': checkout_data['checkout_at']
})
else:
results['failed'].append({
'asset_id': asset_id,
'user_id': user_id,
'error': response.get('error')
})
except Exception as e:
results['failed'].append({
'asset_id': assignment.get('asset_id', 'Unknown'),
'user_id': assignment.get('user_id', 'Unknown'),
'error': str(e)
})
return results
def generate_user_asset_report(self, user_email: str) -> Dict:
"""특정 사용자의 자산 리포트 생성"""
user = self.get_user_by_email(user_email)
if not user:
return {'success': False, 'error': f'사용자를 찾을 수 없습니다: {user_email}'}
assets_info = self.get_user_assets_detailed(user['id'])
if not assets_info['success']:
return assets_info
# 리포트 생성
report = {
'user_info': {
'name': user.get('name'),
'email': user.get('email'),
'employee_num': user.get('employee_num'),
'department': user.get('department', {}).get('name'),
'location': user.get('location', {}).get('name')
},
'asset_summary': {
'total_count': assets_info['total_assets'],
'total_value': assets_info['total_value']
},
'assets': assets_info['assets']
}
return {'success': True, 'report': report}완전한 실무 활용 예시
모든 기능을 통합한 실무용 스크립트입니다.
python#!/usr/bin/env python3
"""
Snipe-IT 자산 관리 자동화 메인 스크립트
"""
import os
import sys
import argparse
from dotenv import load_dotenv
from datetime import datetime
import schedule
import time
def main():
"""메인 함수"""
load_dotenv()
# 환경 변수 확인
required_env_vars = ['SNIPE_IT_URL', 'SNIPE_IT_TOKEN']
missing_vars = [var for var in required_env_vars if not os.getenv(var)]
if missing_vars:
print(f"❌ 필수 환경 변수가 설정되지 않았습니다: {missing_vars}")
sys.exit(1)
# API 클라이언트 초기화
api = SnipeITAPI(
base_url=os.getenv('SNIPE_IT_URL'),
token=os.getenv('SNIPE_IT_TOKEN'),
timeout=int(os.getenv('SNIPE_IT_TIMEOUT', 30))
)
# 명령행 인자 처리
parser = argparse.ArgumentParser(description='Snipe-IT 자산 관리 자동화')
parser.add_argument('--action', choices=['bulk-import', 'report', 'user-assets', 'schedule'],
required=True, help='실행할 작업')
parser.add_argument('--file', help='CSV 파일 경로 (bulk-import용)')
parser.add_argument('--email', help='사용자 이메일 (user-assets용)')
args = parser.parse_args()
try:
if args.action == 'bulk-import':
if not args.file:
print("❌ --file 옵션이 필요합니다")
sys.exit(1)
bulk_manager = BulkAssetManager(
base_url=os.getenv('SNIPE_IT_URL'),
token=os.getenv('SNIPE_IT_TOKEN')
)
print(f"📥 CSV 파일에서 자산 대량 등록 시작: {args.file}")
results = bulk_manager.process_csv_file(args.file)
print(f"✅ 처리 완료 - 성공: {len(results['success'])}, 실패: {len(results['failed'])}")
if results['failed']:
print("\n❌ 실패한 항목들:")
for failed in results['failed'][:5]: # 처음 5개만 표시
print(f" - {failed['asset_tag']}: {failed['error']}")
elif args.action == 'report':
report_generator = AssetReportGenerator(
base_url=os.getenv('SNIPE_IT_URL'),
token=os.getenv('SNIPE_IT_TOKEN')
)
print("📊 자산 현황 리포트 생성 시작...")
results = report_generator.generate_full_report()
if results['success']:
print(f"✅ 리포트 생성 완료!")
print(f"📄 Excel 파일: {results['files']['excel']}")
print(f"📈 차트 파일: {results['files']['chart']}")
print(f"📋 총 자산 수: {results['total_assets']}")
else:
print(f"❌ 리포트 생성 실패: {results['error']}")
elif args.action == 'user-assets':
if not args.email:
print("❌ --email 옵션이 필요합니다")
sys.exit(1)
user_manager = UserAssetManager(
base_url=os.getenv('SNIPE_IT_URL'),
token=os.getenv('SNIPE_IT_TOKEN')
)
print(f"👤 사용자 자산 조회: {args.email}")
results = user_manager.generate_user_asset_report(args.email)
if results['success']:
report = results['report']
print(f"✅ 사용자: {report['user_info']['name']}")
print(f"📦 할당된 자산: {report['asset_summary']['total_count']}개")
print(f"💰 총 자산 가치: {report['asset_summary']['total_value']:,.0f}원")
if report['assets']:
print("\n📋 자산 목록:")
for asset in report['assets'][:10]: # 처음 10개만 표시
print(f" - {asset['name']} ({asset['asset_tag']})")
else:
print(f"❌ 조회 실패: {results['error']}")
elif args.action == 'schedule':
print("⏰ 스케줄러 시작...")
setup_scheduler(api)
except KeyboardInterrupt:
print("\n⏹️ 사용자에 의해 중단되었습니다.")
except Exception as e:
print(f"❌ 오류 발생: {str(e)}")
sys.exit(1)
def setup_scheduler(api):
"""정기 작업 스케줄러 설정"""
def daily_report():
"""일일 리포트 생성"""
print(f"📊 일일 리포트 생성 시작 - {datetime.now()}")
report_generator = AssetReportGenerator(
base_url=os.getenv('SNIPE_IT_URL'),
token=os.getenv('SNIPE_IT_TOKEN')
)
results = report_generator.generate_full_report()
if results['success']:
print(f"✅ 일일 리포트 생성 완료: {results['files']['excel']}")
# 이메일 발송 (옵션)
if os.getenv('SMTP_SERVER'):
send_report_email(results['files']['excel'])
else:
print(f"❌ 일일 리포트 생성 실패: {results['error']}")
def weekly_maintenance():
"""주간 유지보수 작업"""
print(f"🔧 주간 유지보수 작업 시작 - {datetime.now()}")
# 예시: 오래된 로그 파일 정리
cleanup_old_logs()
# 예시: 만료 예정 자산 알림
check_expiring_assets(api)
# 스케줄 등록
schedule.every().day.at("09:00").do(daily_report)
schedule.every().monday.at("08:00").do(weekly_maintenance)
print("📅 스케줄 등록 완료:")
print(" - 일일 리포트: 매일 09:00")
print(" - 주간 유지보수: 매주 월요일 08:00")
print(" - Ctrl+C로 중단")
while True:
schedule.run_pending()
time.sleep(60)
def cleanup_old_logs():
"""오래된 로그 파일 정리"""
import glob
from datetime import timedelta
cutoff_date = datetime.now() - timedelta(days=30)
for log_file in glob.glob("snipe_it_api_*.log"):
try:
file_date = datetime.fromtimestamp(os.path.getctime(log_file))
if file_date < cutoff_date:
os.remove(log_file)
print(f"🗑️ 오래된 로그 파일 삭제: {log_file}")
except:
pass
def check_expiring_assets(api):
"""만료 예정 자산 확인"""
# 워런티 만료 예정 자산 확인 로직
# 실제 구현은 비즈니스 요구사항에 따라 달라짐
pass
def send_report_email(report_file):
"""리포트 이메일 발송"""
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email import encoders
try:
# SMTP 설정
smtp_server = os.getenv('SMTP_SERVER')
smtp_port = int(os.getenv('SMTP_PORT', 587))
smtp_user = os.getenv('SMTP_USER')
smtp_password = os.getenv('SMTP_PASSWORD')
# 이메일 구성
msg = MIMEMultipart()
msg['From'] = smtp_user
msg['To'] = os.getenv('REPORT_EMAIL_TO')
msg['Subject'] = f"Snipe-IT 자산 현황 리포트 - {datetime.now().strftime('%Y-%m-%d')}"
body = """
안녕하세요,
일일 자산 현황 리포트를 첨부파일로 보내드립니다.
감사합니다.
자산 관리 시스템
"""
msg.attach(MIMEText(body, 'plain', 'utf-8'))
# 파일 첨부
with open(report_file, "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= {os.path.basename(report_file)}',
)
msg.attach(part)
# 이메일 발송
server = smtplib.SMTP(smtp_server, smtp_port)
server.starttls()
server.login(smtp_user, smtp_password)
text = msg.as_string()
server.sendmail(smtp_user, os.getenv('REPORT_EMAIL_TO'), text)
server.quit()
print("📧 리포트 이메일 발송 완료")
except Exception as e:
print(f"❌ 이메일 발송 실패: {str(e)}")
if __name__ == "__main__":
main()설정 파일 예시
.env 파일:
bash# Snipe-IT 설정
SNIPE_IT_URL=https://your-snipe-it.domain.com
SNIPE_IT_TOKEN=your-api-token-here
SNIPE_IT_TIMEOUT=30
# 이메일 설정 (선택사항)
SMTP_SERVER=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@company.com
SMTP_PASSWORD=your-app-password
REPORT_EMAIL_TO=management@company.comrequirements.txt:
txtrequests>=2.28.0
pandas>=1.5.0
python-dotenv>=0.19.0
matplotlib>=3.6.0
seaborn>=0.11.0
openpyxl>=3.0.0
schedule>=1.2.0사용법 예시
bash# 가상환경 설정
python -m venv snipe_it_env
source snipe_it_env/bin/activate # Linux/Mac
# snipe_it_env\Scripts\activate # Windows
# 의존성 설치
pip install -r requirements.txt
# CSV에서 자산 대량 등록
python snipe_it_automation.py --action bulk-import --file assets.csv
# 자산 현황 리포트 생성
python snipe_it_automation.py --action report
# 특정 사용자 자산 조회
python snipe_it_automation.py --action user-assets --email john.doe@company.com
# 스케줄러 실행 (백그라운드 작업)
python snipe_it_automation.py --action schedule마무리 및 확장 가능성
이 Python 자동화 스크립트는 다음과 같은 확장이 가능합니다:
보안 강화:
- OAuth 2.0 인증 지원
- API 키 암호화 저장
- 감사 로그 강화
기능 확장:
- Slack/Teams 알림 연동
- 자동 QR 코드 생성
- 바코드 스캐너 연동
- 자산 위치 추적 (RFID/GPS)
성능 최적화:
- 비동기 처리 (asyncio)
- 캐싱 메커니즘
- 데이터베이스 연동
모니터링:
- Prometheus 메트릭 수집
- Grafana 대시보드 연동
- 헬스체크 엔드포인트
이제 여러분은 Snipe-IT API를 완전히 마스터했습니다! 이 시리즈의 코드를 기반으로 조직의 요구사항에 맞는 맞춤형 자산 관리 시스템을 구축할 수 있습니다.