from __future__ import annotations

import json

import logging

import os

import re

import shutil

import subprocess

import tempfile

from dataclasses import asdict, dataclass

from datetime import datetime, timezone

from pathlib import Path

from typing import Any, Dict, List, Optional, Tuple

import sys

try:

from dotenv import load_dotenv # type: ignore

load_dotenv()

except Exception:

pass

_ROOT_DIR = Path(__file__).resolve().parent

_SRC_DIR = _ROOT_DIR / "src"

if str(_SRC_DIR) not in sys.path:

sys.path.insert(0, str(_SRC_DIR))

logging.basicConfig(

level=logging.INFO,

format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',

handlers=[

logging.StreamHandler(),

logging.FileHandler('commit_to_dataset.log', mode='w')

]

)

logger = logging.getLogger(__name__)

try:

# Reuse commit analysis utilities

from collect.analysis.commits import PerfCommitAnalyzer

except Exception:

PerfCommitAnalyzer = None # type: ignore

try:

# Timing helpers (we reuse parse_times signature/regex)

from collect.execute.evaluate import parse_times

except Exception:

parse_times = None # type: ignore

try:

from test_scripts.generate_test_generators import process_extraction_file, LLMClient # type: ignore

except Exception:

process_extraction_file = None # type: ignore

LLMClient = None # type: ignore

try:

import yaml # type: ignore

except Exception:

yaml = None # type: ignore

def run(cmd: List[str], cwd: Optional[Path] = None, env: Optional[Dict[str, str]] = None) -> str:

return result.stdout

_API_MANIFESTS: Dict[str, Dict[str, Any]] = {}

def _read_text_safe(path: Path) -> str:

return ""

def _guess_project_name_from_pyproject(repo_path: Path) -> Optional[str]:

return None

def _candidate_import_names(repo_path: Path) -> List[str]:

return unique

def _select_import_name_via_venv(venv_python: Path, repo_path: Path, prefer: Optional[str] = None) -> Optional[str]:

return None

def _write_api_dump_script(script_path: Path) -> None:

script_path.write_text(script)

def snapshot_public_api(venv_python: Path, repo_path: Path, commit_hash: str, prefer_import_name: Optional[str] = None, work_dir: Optional[Path] = None) -> Optional[Tuple[Path, Dict[str, Any]]]:

return None

def _load_manifest_sets(manifest_path: Path) -> Optional[Tuple[str, set, Dict[str, List[str]]]]:

return None

def _parse_imports_and_aliases(code: str) -> Tuple[Dict[str, str], Dict[str, List[Tuple[str, Optional[str]]]]]:

return alias_to_module, from_imports

def _extract_dotted_refs(code: str, alias_to_module: Dict[str, str], package_root: str) -> List[Tuple[str, str]]:

return refs

def _choose_best_qual(target_leaf: str, origin_module: str, candidates: List[str]) -> Optional[str]:

return best

def _insert_additional_imports(code: str, new_import_lines: List[str]) -> str:

return "\n".join(lines[:insert_idx] + new_import_lines + lines[insert_idx:])

def _load_param_info_from_manifest(manifest_path: Path) -> Dict[str, Dict[str, Any]]:

return {}

def _add_api_probing_helpers(code: str, package_root: str) -> str:

return "\n".join(lines[:insert_idx] + [helpers] + lines[insert_idx:])

def _rewrite_api_calls_in_code(code: str, param_info: Dict[str, Dict[str, Any]]) -> str:

return modified_code

def rewrite_test_script_with_api_probing(test_script: Path, manifest_path: Path) -> Dict[str, Any]:

return summary

def rewrite_test_script_against_manifest(test_script: Path, manifest_path: Path) -> Dict[str, Any]:

return summary

def _extract_hf_repo_id(hf_repo: Optional[str], default_repo_name: str = "omni_commit_dataset") -> Optional[str]:

return f"{s}/{default_repo_name}"

def clone_or_update_repo(repo_url: str, dest_dir: Path) -> Path:

return dest_dir

def checkout_commit(repo_path: Path, commit_hash: str) -> None:

run(["git", "fetch", "--all", "--tags"], cwd=repo_path)

run(["git", "checkout", "--force", commit_hash], cwd=repo_path)

def git_unified_diff(repo_path: Path, base_commit: str, head_commit: str) -> str:

return run(["git", "diff", "-p", base_commit, head_commit], cwd=repo_path)

def get_main_branch_head(repo_path: Path) -> str:

raise RuntimeError("Unable to determine main branch HEAD (tried origin/main, origin/master, main, master)")

def is_test_path(path: str) -> bool:

)

def write_tests_to_tmp(tests: List[str], tmp_dir: Path) -> List[Path]:

return paths

def run_tests_locally(repo_path: Path, commit_hash: str, test_entry: Path) -> List[float]:

return [float('inf')]

def check_capability_requirements(test_script: Path) -> bool:

return True

def detect_capabilities() -> Dict[str, Any]:

return capabilities

def run_tests_with_commit_hopping(