yeonjae-universal-data-aggregator 1.0.5

Universal data aggregator module for processing and aggregating development data

Universal Data Aggregator

커밋 데이터와 diff 정보를 가공, 집계, 통계 처리하는 범용 모듈

🎯 주요 기능

• 개발자별 활동 집계: 커밋 수, 코드 변경량, 작업 패턴 분석
• 저장소별 통계: 기여자, 언어 사용량, 복잡도 메트릭
• 시간대별 분석: 피크 시간, 작업 패턴, 커밋 빈도
• 복잡도 메트릭: 코드 품질 및 복잡도 트렌드 분석
• 캐시 기능: 메모리 기반 캐싱으로 성능 최적화
• 타입 안전성: Pydantic 기반의 강력한 데이터 검증

📦 설치

# GitHub에서 직접 설치
pip install yeonjae-universal-data-aggregator

# 개발 의존성과 함께 설치
pip install yeonjae-universal-data-aggregator[dev]

🚀 빠른 시작

기본 사용법

from datetime import datetime
from yeonjae_universal_data_aggregator import (
    DataAggregatorService, AggregationInput, DateRange,
    CommitData, DiffInfo, DiffType
)

# 서비스 초기화
service = DataAggregatorService(enable_cache=True)

# 샘플 커밋 데이터 생성
diff_info = DiffInfo(
    file_path="src/main.py",
    diff_type=DiffType.ADDED,
    lines_added=50,
    lines_deleted=10,
    language="Python"
)

commit = CommitData(
    commit_id="abc123",
    author="John Doe",
    author_email="[e-mail]",
    timestamp=datetime.now(),
    message="Add new feature",
    repository="my-project",
    branch="main",
    diff_info=[diff_info]
)

# 집계 입력 데이터 구성
date_range = DateRange(start="2024-06-01", end="2024-06-01")
input_data = AggregationInput(
    commits=[commit],
    date_range=date_range
)

# 데이터 집계 수행
result = await service.aggregate_data(input_data)

print(f"총 개발자 수: {result.total_developers}")
print(f"총 커밋 수: {result.total_commits}")
print(f"총 저장소 수: {result.total_repositories}")

고급 사용법 - 필터링

# 특정 개발자만 집계
input_data = AggregationInput(
    commits=commits,
    date_range=date_range,
    developer_filter=["[e-mail]", "[e-mail]"],
    repository_filter=["project-a", "project-b"]
)

result = await service.aggregate_data(input_data)

# 개발자별 상세 통계
for email, stats in result.developer_stats.items():
    print(f"개발자: {stats.developer}")
    print(f"커밋 수: {stats.commit_count}")
    print(f"추가된 라인: {stats.lines_added}")
    print(f"삭제된 라인: {stats.lines_deleted}")
    print(f"변경된 파일: {stats.files_changed}")
    print(f"사용 언어: {', '.join(stats.languages_used)}")
    print(f"평균 복잡도: {stats.avg_complexity}")
    print("---")

📊 데이터 모델

입력 데이터

CommitData

class CommitData(BaseModel):
    commit_id: str
    author: str
    author_email: str
    timestamp: datetime
    message: str
    repository: str
    branch: str
    diff_info: List[DiffInfo]

DiffInfo

class DiffInfo(BaseModel):
    file_path: str
    diff_type: DiffType  # ADDED, MODIFIED, DELETED, RENAMED
    lines_added: int
    lines_deleted: int
    complexity_score: Optional[float]
    language: Optional[str]

출력 데이터

AggregationResult

class AggregationResult(BaseModel):
    developer_stats: Dict[str, DeveloperStats]
    repository_stats: Dict[str, RepositoryStats]
    time_analysis: TimeAnalysis
    complexity_metrics: ComplexityMetrics
    aggregation_time: datetime
    data_quality_score: float

DeveloperStats

class DeveloperStats(BaseModel):
    developer: str
    developer_email: str
    commit_count: int
    lines_added: int
    lines_deleted: int
    files_changed: int
    repositories: List[str]
    languages_used: List[str]
    avg_complexity: float
    peak_hours: List[int]

🔧 API 레퍼런스

DataAggregatorService

메서드

• __init__(enable_cache: bool = True) - 서비스 초기화
• async aggregate_data(input_data: AggregationInput) -> AggregationResult - 메인 집계 메서드
• aggregate_by_developer(commits: List[CommitData], input_data: AggregationInput) -> Dict[str, DeveloperStats] - 개발자별 집계
• generate_time_analysis(commits: List[CommitData]) -> TimeAnalysis - 시간대별 분석
• calculate_complexity_metrics(commits: List[CommitData]) -> ComplexityMetrics - 복잡도 계산

📈 분석 기능

시간대별 분석

# 시간대별 활동 패턴 분석
time_analysis = result.time_analysis
print(f"피크 시간대: {time_analysis.peak_hours}")
print(f"작업 패턴: {time_analysis.work_pattern}")
print(f"평균 커밋 간격: {time_analysis.avg_commit_interval}분")

# 시간대별 커밋 빈도
for hour, count in time_analysis.commit_frequency.items():
    print(f"{hour}시: {count}개 커밋")

복잡도 메트릭

# 코드 복잡도 분석
complexity = result.complexity_metrics
print(f"평균 복잡도: {complexity.avg_complexity}")
print(f"최대 복잡도: {complexity.max_complexity}")
print(f"복잡도 트렌드: {complexity.complexity_trend}")
print(f"고복잡도 파일: {complexity.high_complexity_files}")

저장소별 통계

# 저장소별 상세 정보
for repo_name, repo_stats in result.repository_stats.items():
    print(f"저장소: {repo_name}")
    print(f"총 커밋: {repo_stats.total_commits}")
    print(f"기여자 수: {len(repo_stats.contributors)}")
    print(f"사용 언어: {', '.join(repo_stats.languages)}")
    print("---")

⚡ 성능 최적화

캐시 활용

# 캐시 활성화 (기본값)
service = DataAggregatorService(enable_cache=True)

# 캐시 비활성화 (메모리 절약)
service = DataAggregatorService(enable_cache=False)

대용량 데이터 처리

# 배치 단위로 처리
async def process_large_dataset(commits: List[CommitData], batch_size: int = 1000):
    results = []
    
    for i in range(0, len(commits), batch_size):
        batch = commits[i:i + batch_size]
        input_data = AggregationInput(
            commits=batch,
            date_range=DateRange(start="2024-06-01", end="2024-06-01")
        )
        
        result = await service.aggregate_data(input_data)
        results.append(result)
    
    return results

🧪 테스트

# 테스트 실행
pytest

# 커버리지와 함께 실행
pytest --cov=universal_data_aggregator

# 특정 테스트만 실행
pytest tests/test_service.py::TestDataAggregatorService::test_aggregate_data_success

🔧 개발자 가이드

개발 환경 설정

# 저장소 클론
git clone https://github.com/yeonjae-work/universal-modules.git
cd universal-modules/packages/yeonjae-universal-data-aggregator

# 개발 의존성 설치
pip install -e .[dev]

# 코드 포맷팅
black src/ tests/
isort src/ tests/

# 린팅
flake8 src/ tests/
mypy src/

커스텀 집계 로직 추가

class CustomDataAggregatorService(DataAggregatorService):
    """커스텀 집계 서비스"""
    
    def calculate_custom_metrics(self, commits: List[CommitData]) -> Dict[str, Any]:
        """사용자 정의 메트릭 계산"""
        # 커스텀 로직 구현
        return {
            "custom_metric": "value"
        }

🤝 기여하기

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

📄 라이선스

이 프로젝트는 MIT 라이선스 하에 배포됩니다. 자세한 내용은 LICENSE 파일을 참조하세요.

🙋‍♂️ 지원

• 이슈 리포트: GitHub Issues
• 문서: GitHub Wiki
• 토론: GitHub Discussions

📈 로드맵

• 실시간 스트리밍 데이터 처리
• 머신러닝 기반 패턴 분석
• 그래프 데이터베이스 연동
• 대시보드 시각화 지원
• 다중 저장소 병합 분석
• 성능 벤치마크 도구

---

Universal Data Aggregator로 개발 활동을 체계적으로 분석하고 인사이트를 발견하세요! 📊