374 lines
14 KiB
Python
374 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GitHub Issues Creator from CSV
|
|
CSV 파일을 읽어서 GitHub 이슈를 자동으로 생성하는 스크립트
|
|
"""
|
|
|
|
import os
|
|
import csv
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Dict, Optional
|
|
import requests
|
|
from dataclasses import dataclass
|
|
import argparse
|
|
|
|
|
|
# 설정 상수
|
|
GITHUB_OWNER = "baron-consultant" # 실제 GitHub 사용자명으로 변경 필요
|
|
GITHUB_REPO = "llm-gateway-plan" # 실제 리포지토리명으로 변경 필요
|
|
|
|
# 마일스톤 설정 (순서대로 1, 2, 3)
|
|
MILESTONES = [
|
|
{
|
|
"title": "마일스톤 1: 통합 AI 게이트웨이 및 기본 로깅 백엔드 구축",
|
|
"description": "모든 LLM 및 내부 분석 모델을 위한 안정적인 단일 엔드포인트 제공 및 로깅 시스템 구축",
|
|
"weeks": 12
|
|
},
|
|
{
|
|
"title": "마일스톤 2: API 기반 거버넌스 및 분석 기능 강화",
|
|
"description": "플랫폼 관리자를 위한 API 거버넌스 및 사용량 분석 기능 구현",
|
|
"weeks": 24
|
|
},
|
|
{
|
|
"title": "마일스톤 3: 성능 및 지능 최적화 (LLM 중심)",
|
|
"description": "시맨틱 캐싱, A/B 테스트, 성능 최적화 기능 구현",
|
|
"weeks": 40
|
|
}
|
|
]
|
|
|
|
|
|
@dataclass
|
|
class IssueData:
|
|
title: str
|
|
body: str
|
|
labels: List[str]
|
|
assignees: List[str]
|
|
milestone: int
|
|
priority: str
|
|
estimated_weeks: int
|
|
|
|
|
|
class GitHubIssueCreator:
|
|
def __init__(self, token: str, owner: str = GITHUB_OWNER, repo: str = GITHUB_REPO):
|
|
self.token = token
|
|
self.owner = owner
|
|
self.repo = repo
|
|
self.base_url = "https://api.github.com"
|
|
self.headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/vnd.github.v3+json",
|
|
"X-GitHub-Api-Version": "2022-11-28"
|
|
}
|
|
self.milestones = {}
|
|
self.rate_limit_remaining = 5000 # 초기값
|
|
|
|
def check_rate_limit(self):
|
|
"""API 사용량 확인"""
|
|
url = f"{self.base_url}/rate_limit"
|
|
response = requests.get(url, headers=self.headers)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
self.rate_limit_remaining = data['rate']['remaining']
|
|
print(f"남은 API 호출 수: {self.rate_limit_remaining}")
|
|
|
|
def wait_if_needed(self):
|
|
"""API 제한을 고려한 대기"""
|
|
if self.rate_limit_remaining < 10:
|
|
print("API 사용량이 부족합니다. 1분 대기...")
|
|
time.sleep(60)
|
|
self.check_rate_limit()
|
|
else:
|
|
time.sleep(0.5) # 기본 대기 시간
|
|
|
|
def get_existing_milestones(self) -> Dict[str, int]:
|
|
"""기존 마일스톤 조회"""
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/milestones"
|
|
response = requests.get(url, headers=self.headers)
|
|
|
|
if response.status_code == 200:
|
|
milestones = response.json()
|
|
return {m['title']: m['number'] for m in milestones}
|
|
else:
|
|
print(f"마일스톤 조회 실패: {response.status_code}")
|
|
return {}
|
|
|
|
def create_milestone(self, title: str, description: str, due_date: str) -> Optional[int]:
|
|
"""마일스톤 생성"""
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/milestones"
|
|
data = {
|
|
"title": title,
|
|
"description": description,
|
|
"due_on": due_date
|
|
}
|
|
|
|
response = requests.post(url, headers=self.headers, json=data)
|
|
|
|
if response.status_code == 201:
|
|
milestone = response.json()
|
|
print(f"✅ 마일스톤 생성 완료: {title}")
|
|
return milestone['number']
|
|
else:
|
|
print(f"❌ 마일스톤 생성 실패: {title}, Status: {response.status_code}")
|
|
if response.status_code == 422:
|
|
print(f" 오류 내용: {response.text}")
|
|
return None
|
|
|
|
def setup_milestones(self):
|
|
"""마일스톤들 설정"""
|
|
print("🔄 마일스톤 설정 중...")
|
|
self.milestones = self.get_existing_milestones()
|
|
|
|
for i, milestone_data in enumerate(MILESTONES):
|
|
title = milestone_data["title"]
|
|
if title not in self.milestones:
|
|
due_date = (datetime.now() + timedelta(weeks=milestone_data["weeks"])).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
milestone_number = self.create_milestone(
|
|
title,
|
|
milestone_data["description"],
|
|
due_date
|
|
)
|
|
if milestone_number:
|
|
self.milestones[title] = milestone_number
|
|
self.wait_if_needed()
|
|
else:
|
|
print(f"✅ 기존 마일스톤 사용: {title}")
|
|
|
|
def get_milestone_number(self, milestone_index: int) -> Optional[int]:
|
|
"""마일스톤 번호 조회 (1, 2, 3 형태의 인덱스를 받음)"""
|
|
if 1 <= milestone_index <= len(MILESTONES):
|
|
milestone_title = MILESTONES[milestone_index - 1]["title"]
|
|
return self.milestones.get(milestone_title)
|
|
return None
|
|
|
|
def create_labels(self):
|
|
"""필요한 라벨들 생성"""
|
|
print("🔄 라벨 설정 중...")
|
|
|
|
labels_to_create = [
|
|
{"name": "priority:high", "color": "d73a4a", "description": "높은 우선순위"},
|
|
{"name": "priority:medium", "color": "fbca04", "description": "중간 우선순위"},
|
|
{"name": "priority:low", "color": "0e8a16", "description": "낮은 우선순위"},
|
|
{"name": "milestone-1", "color": "1f77b4", "description": "마일스톤 1 관련"},
|
|
{"name": "milestone-2", "color": "ff7f0e", "description": "마일스톤 2 관련"},
|
|
{"name": "milestone-3", "color": "2ca02c", "description": "마일스톤 3 관련"},
|
|
{"name": "infrastructure", "color": "8c564b", "description": "인프라 관련"},
|
|
{"name": "backend", "color": "e377c2", "description": "백엔드 개발"},
|
|
{"name": "database", "color": "7f7f7f", "description": "데이터베이스 관련"},
|
|
{"name": "api", "color": "bcbd22", "description": "API 개발"},
|
|
{"name": "integration", "color": "17becf", "description": "통합 작업"},
|
|
{"name": "documentation", "color": "aec7e8", "description": "문서화"},
|
|
{"name": "testing", "color": "ffbb78", "description": "테스트 관련"},
|
|
{"name": "optimization", "color": "ff9896", "description": "성능 최적화"},
|
|
{"name": "monitoring", "color": "c5b0d5", "description": "모니터링"},
|
|
]
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/labels"
|
|
|
|
# 기존 라벨 조회
|
|
existing_labels = requests.get(url, headers=self.headers)
|
|
existing_label_names = []
|
|
if existing_labels.status_code == 200:
|
|
existing_label_names = [label['name'] for label in existing_labels.json()]
|
|
|
|
created_count = 0
|
|
for label in labels_to_create:
|
|
if label['name'] not in existing_label_names:
|
|
response = requests.post(url, headers=self.headers, json=label)
|
|
if response.status_code == 201:
|
|
print(f"✅ 라벨 생성: {label['name']}")
|
|
created_count += 1
|
|
else:
|
|
print(f"❌ 라벨 생성 실패: {label['name']}")
|
|
self.wait_if_needed()
|
|
|
|
if created_count == 0:
|
|
print("✅ 모든 라벨이 이미 존재합니다.")
|
|
|
|
def create_issue(self, issue_data: IssueData) -> bool:
|
|
"""이슈 생성"""
|
|
url = f"{self.base_url}/repos/{self.owner}/{self.repo}/issues"
|
|
|
|
milestone_number = self.get_milestone_number(issue_data.milestone)
|
|
|
|
data = {
|
|
"title": issue_data.title,
|
|
"body": issue_data.body,
|
|
"labels": issue_data.labels,
|
|
}
|
|
|
|
# assignees가 비어있지 않은 경우에만 추가
|
|
if issue_data.assignees:
|
|
data["assignees"] = issue_data.assignees
|
|
|
|
if milestone_number:
|
|
data["milestone"] = milestone_number
|
|
|
|
response = requests.post(url, headers=self.headers, json=data)
|
|
|
|
if response.status_code == 201:
|
|
issue = response.json()
|
|
print(f"✅ 이슈 생성 완료: #{issue['number']} - {issue_data.title}")
|
|
return True
|
|
else:
|
|
print(f"❌ 이슈 생성 실패: {issue_data.title}")
|
|
print(f" Status: {response.status_code}")
|
|
if response.status_code == 422:
|
|
error_data = response.json()
|
|
print(f" 오류: {error_data}")
|
|
return False
|
|
|
|
|
|
def parse_csv(csv_file: str) -> List[IssueData]:
|
|
"""CSV 파일을 파싱하여 IssueData 리스트 반환"""
|
|
issues = []
|
|
|
|
with open(csv_file, 'r', encoding='utf-8') as file:
|
|
reader = csv.DictReader(file)
|
|
|
|
for row in reader:
|
|
# 라벨 파싱 (쉼표 구분) - None 값 처리 개선
|
|
labels_str = row.get('labels', '') or ''
|
|
labels = [label.strip() for label in labels_str.split(',') if label.strip()]
|
|
|
|
# 담당자 파싱 (쉼표 구분) - None 값 처리 개선
|
|
assignees_str = row.get('assignees', '') or ''
|
|
assignees = [assignee.strip() for assignee in assignees_str.split(',') if assignee.strip()]
|
|
|
|
# 마일스톤 파싱 (숫자)
|
|
try:
|
|
milestone_str = row.get('milestone', '1') or '1'
|
|
milestone = int(milestone_str)
|
|
except (ValueError, TypeError):
|
|
milestone = 1
|
|
|
|
# 예상 작업 시간 파싱
|
|
try:
|
|
weeks_str = row.get('estimated_weeks', '1') or '1'
|
|
estimated_weeks = int(weeks_str)
|
|
except (ValueError, TypeError):
|
|
estimated_weeks = 1
|
|
|
|
# 제목과 본문도 None 값 처리
|
|
title = row.get('title', '') or ''
|
|
body = row.get('body', '') or ''
|
|
priority = row.get('priority', 'Medium') or 'Medium'
|
|
|
|
issue = IssueData(
|
|
title=title.strip(),
|
|
body=body.strip(),
|
|
labels=labels,
|
|
assignees=assignees,
|
|
milestone=milestone,
|
|
priority=priority.strip(),
|
|
estimated_weeks=estimated_weeks
|
|
)
|
|
|
|
if issue.title: # 제목이 있는 경우에만 추가
|
|
issues.append(issue)
|
|
|
|
return issues
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='CSV에서 GitHub 이슈 생성')
|
|
parser.add_argument('csv_file', help='이슈 데이터가 포함된 CSV 파일 경로')
|
|
parser.add_argument('--token', help='GitHub 개인 액세스 토큰 (환경변수 GITHUB_TOKEN 사용 가능)')
|
|
parser.add_argument('--dry-run', action='store_true', help='실제 생성 없이 미리보기만 실행')
|
|
parser.add_argument('--skip-labels', action='store_true', help='라벨 생성 건너뛰기')
|
|
parser.add_argument('--skip-milestones', action='store_true', help='마일스톤 생성 건너뛰기')
|
|
parser.add_argument('--debug', action='store_true', help='CSV 파일 디버그 정보 표시')
|
|
args = parser.parse_args()
|
|
|
|
# GitHub 토큰 확인
|
|
token = args.token or os.environ.get('GITHUB_TOKEN')
|
|
if not token:
|
|
print("❌ GitHub 토큰이 필요합니다. --token 또는 GITHUB_TOKEN 환경변수를 설정하세요.")
|
|
return
|
|
|
|
# CSV 파일 확인
|
|
if not os.path.exists(args.csv_file):
|
|
print(f"❌ CSV 파일을 찾을 수 없습니다: {args.csv_file}")
|
|
return
|
|
|
|
# 디버그 모드로 CSV 구조 확인
|
|
if args.debug:
|
|
print(f"🔍 CSV 파일 디버그 정보: {args.csv_file}")
|
|
with open(args.csv_file, 'r', encoding='utf-8') as file:
|
|
reader = csv.DictReader(file)
|
|
print(f"📋 컬럼: {reader.fieldnames}")
|
|
|
|
for i, row in enumerate(reader):
|
|
if i >= 3: # 처음 3개 행만 표시
|
|
break
|
|
print(f"🔍 행 {i+1}: {dict(row)}")
|
|
return
|
|
|
|
print(f"📁 CSV 파일 읽는 중: {args.csv_file}")
|
|
|
|
try:
|
|
issues = parse_csv(args.csv_file)
|
|
except Exception as e:
|
|
print(f"❌ CSV 파일 파싱 중 오류 발생: {e}")
|
|
print("💡 --debug 옵션을 사용하여 CSV 파일 구조를 확인해보세요.")
|
|
return
|
|
|
|
if not issues:
|
|
print("❌ 생성할 이슈가 없습니다.")
|
|
return
|
|
|
|
print(f"📊 총 {len(issues)}개의 이슈를 발견했습니다.")
|
|
|
|
if args.dry_run:
|
|
print("\n🔍 [DRY RUN] 생성될 이슈들:")
|
|
for i, issue in enumerate(issues, 1):
|
|
print(f" {i:2d}. [{issue.milestone}] {issue.title}")
|
|
print(f" 라벨: {', '.join(issue.labels)}")
|
|
print(f" 담당자: {', '.join(issue.assignees)}")
|
|
print(f" 우선순위: {issue.priority}")
|
|
print()
|
|
return
|
|
|
|
# GitHub API 클라이언트 생성
|
|
creator = GitHubIssueCreator(token)
|
|
|
|
print(f"🎯 GitHub 리포지토리: {GITHUB_OWNER}/{GITHUB_REPO}")
|
|
|
|
# API 사용량 확인
|
|
creator.check_rate_limit()
|
|
|
|
# 마일스톤 설정
|
|
if not args.skip_milestones:
|
|
creator.setup_milestones()
|
|
|
|
# 라벨 설정
|
|
if not args.skip_labels:
|
|
creator.create_labels()
|
|
|
|
# 이슈 생성
|
|
print(f"\n🚀 {len(issues)}개 이슈 생성 시작...")
|
|
success_count = 0
|
|
fail_count = 0
|
|
|
|
for i, issue in enumerate(issues, 1):
|
|
print(f"\n[{i}/{len(issues)}] 이슈 생성 중...")
|
|
|
|
if creator.create_issue(issue):
|
|
success_count += 1
|
|
else:
|
|
fail_count += 1
|
|
|
|
creator.wait_if_needed()
|
|
|
|
print(f"\n📈 완료: 성공 {success_count}개, 실패 {fail_count}개")
|
|
|
|
if fail_count > 0:
|
|
print("\n⚠️ 일부 이슈 생성에 실패했습니다. 다음 사항을 확인하세요:")
|
|
print(" - GitHub 토큰 권한")
|
|
print(" - 리포지토리 접근 권한")
|
|
print(" - 담당자 사용자명")
|
|
print(" - API 사용량 제한")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |