GitHub2

LLM

목록 보기

9/10

GitHub URL 목록 파일을 받아서 모든 저장소를 일괄 분석하는 스크립트를 작성해드리겠습니다.

#!/usr/bin/env python3
"""
GitHub 저장소 일괄 분석 스크립트
URL 목록 파일을 받아서 모든 저장소를 클론하고 커밋 로그를 분석합니다.
"""

import os
import subprocess
import tempfile
import shutil
from pathlib import Path
from typing import List, Dict, Optional
import pandas as pd
import argparse
from datetime import datetime
import json
import logging
from urllib.parse import urlparse
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

# 기존 CommitAnalyzer 클래스 import
from commit_analyzer import CommitAnalyzer

class BatchRepoAnalyzer:
    def __init__(self, openai_api_key: str = None, max_workers: int = 3):
        """
        일괄 저장소 분석기 초기화
        
        Args:
            openai_api_key: OpenAI API 키
            max_workers: 동시 처리할 저장소 수
        """
        self.commit_analyzer = CommitAnalyzer(openai_api_key)
        self.max_workers = max_workers
        self.temp_dir = None
        
        # 로깅 설정
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('batch_analysis.log', encoding='utf-8'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    def read_repo_urls(self, file_path: str) -> List[Dict[str, str]]:
        """
        저장소 URL 목록 파일을 읽습니다.
        
        지원 형식:
        1. 텍스트 파일 (한 줄에 하나씩 URL)
        2. JSON 파일 (배열 또는 객체)
        3. CSV 파일 (url 컬럼 필요)
        
        Args:
            file_path: 파일 경로
            
        Returns:
            저장소 정보 리스트 [{'name': '저장소명', 'url': 'URL', 'description': '설명'}]
        """
        file_path = Path(file_path)
        repos = []
        
        try:
            if file_path.suffix.lower() == '.json':
                with open(file_path, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    
                if isinstance(data, list):
                    for item in data:
                        if isinstance(item, str):
                            repos.append(self._parse_repo_info(item))
                        elif isinstance(item, dict):
                            repos.append({
                                'name': item.get('name', self._extract_repo_name(item.get('url', ''))),
                                'url': item.get('url', ''),
                                'description': item.get('description', '')
                            })
                            
            elif file_path.suffix.lower() == '.csv':
                df = pd.read_csv(file_path)
                for _, row in df.iterrows():
                    url = row.get('url', '') or row.get('URL', '') or row.get('repository', '')
                    repos.append({
                        'name': row.get('name', self._extract_repo_name(url)),
                        'url': url,
                        'description': row.get('description', '') or row.get('desc', '')
                    })
                    
            else:  # 텍스트 파일
                with open(file_path, 'r', encoding='utf-8') as f:
                    for line in f:
                        line = line.strip()
                        if line and not line.startswith('#'):
                            repos.append(self._parse_repo_info(line))
                            
        except Exception as e:
            self.logger.error(f"파일 읽기 오류 {file_path}: {e}")
            
        self.logger.info(f"{len(repos)}개의 저장소 URL을 읽었습니다.")
        return repos

    def _parse_repo_info(self, url: str) -> Dict[str, str]:
        """URL에서 저장소 정보를 추출합니다."""
        return {
            'name': self._extract_repo_name(url),
            'url': url.strip(),
            'description': ''
        }

    def _extract_repo_name(self, url: str) -> str:
        """URL에서 저장소 이름을 추출합니다."""
        try:
            parsed = urlparse(url)
            path_parts = parsed.path.strip('/').split('/')
            if len(path_parts) >= 2:
                return f"{path_parts[-2]}/{path_parts[-1]}"
            return path_parts[-1] if path_parts else url
        except:
            return url

    def clone_repository(self, repo_info: Dict[str, str], clone_dir: str) -> Optional[str]:
        """
        저장소를 클론합니다.
        
        Args:
            repo_info: 저장소 정보
            clone_dir: 클론할 디렉토리
            
        Returns:
            클론된 저장소 경로 또는 None
        """
        repo_url = repo_info['url']
        repo_name = repo_info['name'].replace('/', '_')
        repo_path = os.path.join(clone_dir, repo_name)
        
        try:
            self.logger.info(f"📥 저장소 클론 중: {repo_url}")
            
            # 이미 클론된 경우 pull
            if os.path.exists(repo_path):
                self.logger.info(f"기존 저장소 업데이트: {repo_path}")
                subprocess.run(['git', 'pull'], cwd=repo_path, check=True, 
                             capture_output=True, text=True)
            else:
                # 새로 클론
                subprocess.run(['git', 'clone', repo_url, repo_path], 
                             check=True, capture_output=True, text=True)
                
            return repo_path
            
        except subprocess.CalledProcessError as e:
            self.logger.error(f"저장소 클론 실패 {repo_url}: {e.stderr}")
            return None
        except Exception as e:
            self.logger.error(f"저장소 클론 오류 {repo_url}: {e}")
            return None

    def analyze_single_repository(self, repo_info: Dict[str, str], 
                                 clone_dir: str, 
                                 author: str = None, 
                                 since: str = None, 
                                 until: str = None) -> Optional[Dict]:
        """
        단일 저장소를 분석합니다.
        
        Args:
            repo_info: 저장소 정보
            clone_dir: 클론 디렉토리
            author: 특정 작성자 필터
            since: 시작 날짜
            until: 종료 날짜
            
        Returns:
            분석 결과 딕셔너리
        """
        repo_path = self.clone_repository(repo_info, clone_dir)
        if not repo_path:
            return None
            
        try:
            self.logger.info(f"📊 저장소 분석 중: {repo_info['name']}")
            
            # 커밋 로그 가져오기
            commits = self.commit_analyzer.get_commit_log(
                repo_path=repo_path,
                author=author,
                since=since,
                until=until
            )
            
            if not commits:
                self.logger.warning(f"커밋이 없습니다: {repo_info['name']}")
                return {
                    'repository': repo_info['name'],
                    'url': repo_info['url'],
                    'description': repo_info['description'],
                    'status': 'no_commits',
                    'commits_count': 0,
                    'total_time_minutes': 0,
                    'analysis_data': pd.DataFrame(),
                    'error': None
                }
            
            # 커밋 분석
            df = self.commit_analyzer.analyze_commits(commits)
            
            # 요약 통계 계산
            total_commits = df['커밋 수'].sum()
            total_time = df['개발작업 추정시간 (분)'].sum()
            
            result = {
                'repository': repo_info['name'],
                'url': repo_info['url'],
                'description': repo_info['description'],
                'status': 'success',
                'commits_count': total_commits,
                'total_time_minutes': total_time,
                'total_time_hours': total_time / 60,
                'avg_time_per_commit': total_time / total_commits if total_commits > 0 else 0,
                'date_range': f"{df['일자'].min()} ~ {df['일자'].max()}" if len(df) > 0 else "",
                'analysis_data': df,
                'error': None
            }
            
            self.logger.info(f"✅ 분석 완료: {repo_info['name']} ({total_commits}개 커밋, {total_time:.0f}분)")
            return result
            
        except Exception as e:
            self.logger.error(f"저장소 분석 오류 {repo_info['name']}: {e}")
            return {
                'repository': repo_info['name'],
                'url': repo_info['url'],
                'description': repo_info['description'],
                'status': 'error',
                'commits_count': 0,
                'total_time_minutes': 0,
                'analysis_data': pd.DataFrame(),
                'error': str(e)
            }

    def analyze_repositories(self, repos: List[Dict[str, str]], 
                           output_dir: str = "analysis_results",
                           author: str = None,
                           since: str = None,
                           until: str = None) -> Dict:
        """
        여러 저장소를 병렬로 분석합니다.
        
        Args:
            repos: 저장소 정보 리스트
            output_dir: 결과 저장 디렉토리
            author: 특정 작성자 필터
            since: 시작 날짜
            until: 종료 날짜
            
        Returns:
            전체 분석 결과
        """
        # 출력 디렉토리 생성
        os.makedirs(output_dir, exist_ok=True)
        
        # 임시 클론 디렉토리 생성
        self.temp_dir = tempfile.mkdtemp(prefix="batch_analysis_")
        
        try:
            results = []
            failed_repos = []
            
            # 병렬 처리
            with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                # 작업 제출
                future_to_repo = {
                    executor.submit(
                        self.analyze_single_repository, 
                        repo, self.temp_dir, author, since, until
                    ): repo for repo in repos
                }
                
                # 결과 수집
                for future in as_completed(future_to_repo):
                    repo = future_to_repo[future]
                    try:
                        result = future.result()
                        if result:
                            if result['status'] == 'success':
                                results.append(result)
                            else:
                                failed_repos.append(result)
                        else:
                            failed_repos.append({
                                'repository': repo['name'],
                                'url': repo['url'],
                                'status': 'failed',
                                'error': 'Failed to clone or analyze'
                            })
                    except Exception as e:
                        self.logger.error(f"작업 처리 오류 {repo['name']}: {e}")
                        failed_repos.append({
                            'repository': repo['name'],
                            'url': repo['url'],
                            'status': 'error',
                            'error': str(e)
                        })
            
            # 결과 저장
            self._save_results(results, failed_repos, output_dir)
            
            return {
                'successful': results,
                'failed': failed_repos,
                'total_repositories': len(repos),
                'successful_count': len(results),
                'failed_count': len(failed_repos)
            }
            
        finally:
            # 임시 디렉토리 정리
            if self.temp_dir and os.path.exists(self.temp_dir):
                shutil.rmtree(self.temp_dir)

    def _save_results(self, results: List[Dict], failed_repos: List[Dict], output_dir: str):
        """분석 결과를 파일로 저장합니다."""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        # 1. 전체 요약 보고서
        summary_data = []
        all_detailed_data = []
        
        for result in results:
            summary_data.append({
                '저장소': result['repository'],
                'URL': result['url'],
                '설명': result['description'],
                '커밋수': result['commits_count'],
                '총개발시간(분)': result['total_time_minutes'],
                '총개발시간(시간)': f"{result['total_time_hours']:.1f}",
                '커밋당평균시간(분)': f"{result['avg_time_per_commit']:.1f}",
                '분석기간': result['date_range'],
                '상태': result['status']
            })
            
            # 상세 데이터에 저장소 정보 추가
            detailed_df = result['analysis_data'].copy()
            if not detailed_df.empty:
                detailed_df.insert(0, '저장소', result['repository'])
                detailed_df.insert(1, 'URL', result['url'])
                all_detailed_data.append(detailed_df)
        
        # 요약 보고서 저장
        summary_df = pd.DataFrame(summary_data)
        summary_file = os.path.join(output_dir, f"summary_report_{timestamp}.csv")
        summary_df.to_csv(summary_file, index=False, encoding='utf-8-sig')
        
        # 전체 상세 데이터 저장
        if all_detailed_data:
            combined_df = pd.concat(all_detailed_data, ignore_index=True)
            detailed_file = os.path.join(output_dir, f"detailed_analysis_{timestamp}.csv")
            combined_df.to_csv(detailed_file, index=False, encoding='utf-8-sig')
        
        # 개별 저장소 상세 분석 저장
        individual_dir = os.path.join(output_dir, f"individual_{timestamp}")
        os.makedirs(individual_dir, exist_ok=True)
        
        for result in results:
            if not result['analysis_data'].empty:
                repo_name = result['repository'].replace('/', '_')
                individual_file = os.path.join(individual_dir, f"{repo_name}.csv")
                result['analysis_data'].to_csv(individual_file, index=False, encoding='utf-8-sig')
        
        # 실패한 저장소 목록 저장
        if failed_repos:
            failed_df = pd.DataFrame(failed_repos)
            failed_file = os.path.join(output_dir, f"failed_repositories_{timestamp}.csv")
            failed_df.to_csv(failed_file, index=False, encoding='utf-8-sig')
        
        # JSON 형태로도 저장
        full_results = {
            'timestamp': timestamp,
            'summary': summary_data,
            'failed_repositories': failed_repos,
            'statistics': {
                'total_repositories': len(results) + len(failed_repos),
                'successful_analyses': len(results),
                'failed_analyses': len(failed_repos),
                'total_commits': sum(r['commits_count'] for r in results),
                'total_development_time_minutes': sum(r['total_time_minutes'] for r in results),
                'total_development_time_hours': sum(r['total_time_minutes'] for r in results) / 60
            }
        }
        
        json_file = os.path.join(output_dir, f"full_results_{timestamp}.json")
        with open(json_file, 'w', encoding='utf-8') as f:
            json.dump(full_results, f, ensure_ascii=False, indent=2)
        
        self.logger.info(f"📄 결과 파일 저장 완료:")
        self.logger.info(f"   📊 요약 보고서: {summary_file}")
        if all_detailed_data:
            self.logger.info(f"   📋 상세 분석: {detailed_file}")
        self.logger.info(f"   📁 개별 분석: {individual_dir}")
        if failed_repos:
            self.logger.info(f"   ❌ 실패 목록: {failed_file}")
        self.logger.info(f"   📦 전체 결과: {json_file}")

def main():
    parser = argparse.ArgumentParser(description='GitHub 저장소 일괄 분석')
    parser.add_argument('urls_file', help='저장소 URL 목록 파일 (.txt, .json, .csv)')
    parser.add_argument('--output', '-o', default='analysis_results', help='결과 저장 디렉토리')
    parser.add_argument('--author', '-a', help='특정 작성자 필터링')
    parser.add_argument('--since', '-s', help='시작 날짜 (YYYY-MM-DD)')
    parser.add_argument('--until', '-u', help='종료 날짜 (YYYY-MM-DD)')
    parser.add_argument('--openai-key', help='OpenAI API 키')
    parser.add_argument('--workers', '-w', type=int, default=3, help='동시 처리 저장소 수 (기본값: 3)')
    
    args = parser.parse_args()
    
    # 분석기 초기화
    analyzer = BatchRepoAnalyzer(
        openai_api_key=args.openai_key,
        max_workers=args.workers
    )
    
    print("🚀 GitHub 저장소 일괄 분석을 시작합니다...")
    start_time = time.time()
    
    # URL 목록 읽기
    repos = analyzer.read_repo_urls(args.urls_file)
    if not repos:
        print("❌ 분석할 저장소가 없습니다.")
        return
    
    # 저장소 분석
    final_results = analyzer.analyze_repositories(
        repos=repos,
        output_dir=args.output,
        author=args.author,
        since=args.since,
        until=args.until
    )
    
    # 최종 결과 출력
    elapsed_time = time.time() - start_time
    
    print("\n" + "="*80)
    print("🎉 일괄 분석 완료!")
    print("="*80)
    print(f"⏱️  총 소요 시간: {elapsed_time:.1f}초")
    print(f"📊 전체 저장소: {final_results['total_repositories']}개")
    print(f"✅ 성공: {final_results['successful_count']}개")
    print(f"❌ 실패: {final_results['failed_count']}개")
    
    if final_results['successful']:
        total_commits = sum(r['commits_count'] for r in final_results['successful'])
        total_time = sum(r['total_time_minutes'] for r in final_results['successful'])
        print(f"📈 총 분석 커밋: {total_commits:,}개")
        print(f"🕐 총 개발 시간: {total_time:,.0f}분 ({total_time/60:.1f}시간)")
    
    print(f"💾 결과가 '{args.output}' 디렉토리에 저장되었습니다.")

if __name__ == "__main__":
    main()코드를 입력하세요

이제 URL 목록 파일의 예시 파일들도 생성해드리겠습니다.

# 1. repositories.txt (텍스트 파일 형식)
# GitHub 저장소 URL 목록 - 한 줄에 하나씩
https://github.com/sechan9999/sfguide-build-data-agents-using-snowflake-cortex-ai
https://github.com/username/project1
https://github.com/username/project2
https://github.com/organization/awesome-project

---

# 2. repositories.json (JSON 파일 형식)
[
  {
    "name": "Snowflake Cortex AI Guide",
    "url": "https://github.com/sechan9999/sfguide-build-data-agents-using-snowflake-cortex-ai",
    "description": "Snowflake Cortex AI를 사용한 데이터 에이전트 구축 가이드"
  },
  {
    "name": "Project 1",
    "url": "https://github.com/username/project1",
    "description": "첫 번째 프로젝트"
  },
  {
    "name": "Project 2", 
    "url": "https://github.com/username/project2",
    "description": "두 번째 프로젝트"
  }
]

---

# 3. repositories.csv (CSV 파일 형식)
name,url,description
"Snowflake Cortex AI Guide","https://github.com/sechan9999/sfguide-build-data-agents-using-snowflake-cortex-ai","Snowflake Cortex AI 데이터 에이전트 가이드"
"Project 1","https://github.com/username/project1","첫 번째 프로젝트"
"Project 2","https://github.com/username/project2","두 번째 프로젝트"
"Awesome Project","https://github.com/organization/awesome-project","멋진 프로젝트"

마지막으로 일괄 분석 스크립트의 사용 가이드를 작성해드리겠습니다.

GitHub 저장소 일괄 분석 스크립트

여러 GitHub 저장소를 한 번에 분석하여 개발 시간을 추정하고 상세한 보고서를 생성하는 스크립트입니다.

🌟 주요 기능

📋 다양한 형식 지원: TXT, JSON, CSV 형식의 URL 목록 파일
🚀 병렬 처리: 여러 저장소를 동시에 분석하여 시간 단축
📊 종합 보고서: 전체 요약 + 개별 상세 분석
🤖 AI 요약: OpenAI API로 커밋 내용 자동 요약
💾 다양한 출력: CSV, JSON 형식으로 결과 저장
🔄 자동 클론/업데이트: 저장소 자동 클론 및 업데이트

📋 필요 조건

Python 패키지 설치:

pip install pandas openai python-dateutil

Git 설치: 시스템에 Git이 설치되어 있어야 합니다.
OpenAI API 키 (선택사항): 커밋 요약 기능을 사용하려면 필요합니다.

📁 URL 목록 파일 형식

1. 텍스트 파일 (.txt)

https://github.com/user1/repo1
https://github.com/user2/repo2
https://github.com/user3/repo3
# 주석도 가능합니다

2. JSON 파일 (.json)

[
  {
    "name": "프로젝트 이름",
    "url": "https://github.com/user/repo",
    "description": "프로젝트 설명"
  }
]

3. CSV 파일 (.csv)

name,url,description
"프로젝트1","https://github.com/user1/repo1","첫 번째 프로젝트"
"프로젝트2","https://github.com/user2/repo2","두 번째 프로젝트"

🚀 사용 방법

기본 사용법

python batch_repo_analyzer.py repositories.txt

고급 사용법

# 출력 디렉토리 지정
python batch_repo_analyzer.py repositories.txt --output my_analysis

# 특정 작성자만 분석
python batch_repo_analyzer.py repositories.txt --author "김개발"

# 날짜 범위 지정
python batch_repo_analyzer.py repositories.txt --since 2024-01-01 --until 2024-12-31

# OpenAI API 키 지정
python batch_repo_analyzer.py repositories.txt --openai-key "sk-..."

# 동시 처리 저장소 수 조정 (기본값: 3)
python batch_repo_analyzer.py repositories.txt --workers 5

# 모든 옵션 조합
python batch_repo_analyzer.py repositories.json \
  --output ./results \
  --author "개발자" \
  --since 2024-01-01 \
  --openai-key "sk-..." \
  --workers 4

📊 출력 파일 구조

분석 완료 후 다음과 같은 파일들이 생성됩니다:

analysis_results/
├── summary_report_20241227_143022.csv          # 📊 전체 요약 보고서
├── detailed_analysis_20241227_143022.csv       # 📋 상세 분석 데이터
├── failed_repositories_20241227_143022.csv     # ❌ 실패한 저장소 목록
├── full_results_20241227_143022.json           # 📦 전체 결과 (JSON)
└── individual_20241227_143022/                 # 📁 개별 저장소 분석
    ├── user1_repo1.csv
    ├── user2_repo2.csv
    └── user3_repo3.csv

📊 요약 보고서 (summary_report.csv)

저장소	URL	설명	커밋수	총개발시간(분)	총개발시간(시간)	커밋당평균시간(분)	분석기간	상태
user/repo1	https://...	프로젝트 설명	45	3240	54.0	72.0	2024-01-01 ~ 2024-02-28	success

📋 상세 분석 (detailed_analysis.csv)

저장소	URL	일자	시간대	커밋 수	개발작업 추정시간 (분)	커밋내용요약
user/repo1	https://...	2024-01-15	오후	3	180	프로젝트 초기 설정 및 구조 생성

🔧 커맨드 라인 옵션

옵션	설명	기본값	예시
`urls_file`	URL 목록 파일 경로	필수	`repositories.txt`
`--output`, `-o`	결과 저장 디렉토리	`analysis_results`	`--output ./my_results`
`--author`, `-a`	특정 작성자 필터링	없음	`--author "김개발"`
`--since`, `-s`	시작 날짜	없음	`--since 2024-01-01`
`--until`, `-u`	종료 날짜	없음	`