#!/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()