collect_data_test.py

import github3
import requests
import logging
from datetime import datetime
from pathlib import Path
import hashlib

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

# 配置 GitHub API 访问
def get_github_token():
    token_file = Path.home() / '.github_token'
    if token_file.exists():
        return token_file.read_text().strip()
    return None

github_token = get_github_token()
if not github_token:
    logging.error('GitHub token not found. Please create ~/.github_token file with your token')
    exit(1)

try:
    github = github3.login(token=github_token)
    if not github:
        raise Exception('Failed to authenticate with GitHub')
    logging.info('Successfully authenticated with GitHub')
except Exception as e:
    logging.error(f'Error authenticating with GitHub: {str(e)}')
    exit(1)

# 项目列表
projects = [
    {'name': 'THUDM/CodeGeeX', 'size': 'L'}
]

# 存储收集到的数据
data = []

def generate_task_id(commit_sha, repo_name):
    """生成唯一的task_id"""
    return hashlib.md5(f"{repo_name}_{commit_sha}".encode()).hexdigest()[:8]

def get_commit_details(repo, commit_sha):
    commit = repo.commit(commit_sha)
    return {
        'commit_sha': commit.sha,
        'commit_message': commit.commit.message,
        'committer_date': commit.commit.committer.date,
        'affected_files': [file.filename for file in commit.files],
        'diff': [file.patch for file in commit.files if file.patch]
    }

def get_test_files(repo, commit_sha):
    commit = repo.commit(commit_sha)
    test_files = []
    for file in commit.files:
        if 'test' in file.filename or 'spec' in file.filename:
            test_files.append({
                'path': file.filename,
                'content': file.patch if file.patch else None
            })
    return test_files

def collect_data(repo, project_size):
    try:
        logging.info(f'Collecting data for repository: {repo.full_name}')
        repo_commits = repo.commits()
        if not repo_commits:
            logging.warning(f'No commits found in repository: {repo.full_name}')
            return
            
        for commit in repo_commits:
            commit_details = get_commit_details(repo, commit.sha)
            test_files = get_test_files(repo, commit.sha)
            
            # 生成唯一的task_id
            task_id = generate_task_id(commit.sha, repo.full_name)
            
            data.append({
                'task_id': task_id,
                'project': {
                    'repo_url': repo.html_url,
                    'size_category': project_size,
                    'language': 'Python'
                },
                'change': {
                    'description': commit_details['commit_message'],
                    'branch': commit.committer['branch'] if 'branch' in commit.committer else 'master',
                    'affected_files': commit_details['affected_files'],
                    'diff': commit_details['diff'],
                    'commit': {
                        'before': None,  # 需要更多逻辑来获取变更前的 commit
                        'after': commit_details['commit_sha']
                    },
                    'tests': {
                        'test_list': test_files,
                        'test_setup_code': None  # 如果需要可以添加测试设置代码
                    },
                    'timestamp': commit_details['committer_date'].isoformat()
                }
            })
            
            logging.info(f'Processed commit: {commit.sha}')
            
    except Exception as e:
        logging.error(f'Error collecting data from repository {repo.full_name}: {str(e)}')

for project in projects:
    try:
        owner, repo_name = project['name'].split('/')
        logging.info(f'Processing project: {project["name"]}')
        repo = github.repository(owner, repo_name)
        if not repo:
            logging.error(f'Repository not found: {project["name"]}')
            continue
        collect_data(repo, project['size'])
    except Exception as e:
        logging.error(f'Error processing project {project["name"]}: {str(e)}')

# 保存数据到文件
import json
try:
    output_file = 'collected_data_test.json'
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    logging.info(f'Successfully saved data to {output_file}')
    logging.info(f'Total records collected: {len(data)}')
except Exception as e:
    logging.error(f'Error saving data to file: {str(e)}')
    exit(1)