data-extraction.py

"""
Git activity extraction using PyDriller.

This script traverses commits in a repository (local path or Git URL),
collects per-file change metrics, and exports results to a CSV saved under ./data/.
"""

# =====================
# Imports & Constants
# =====================
from pydriller import Repository
from datetime import datetime
from pprint import pprint
import pandas as pd
import os
from urllib.parse import urlparse

OUTPUT_DIR = "data"
COLUMNS = [
  "hash",
  "date",
  "committer",
  "message",
  "filename",
  "added_lines",
  "deleted_lines",
  "token_count",
  "nloc",
]


# =====================
# Helpers
# =====================
def extract_repo_name(path_or_url: str) -> str:
  """Return a safe repo name from a local path or Git URL.

  Examples:
  - https://github.com/org/repo.git -> repo
  - C:\\code\\my-project -> my-project
  """

  parsed = urlparse(path_or_url)
  # If it's a URL, get the last non-empty segment of the path.
  if parsed.scheme and parsed.path:
    basename = os.path.basename(parsed.path.rstrip("/"))
  else:
    # Treat as local path.
    basename = os.path.basename(os.path.normpath(path_or_url))

  if basename.endswith(".git"):
    basename = basename[:-4]

  # Replace any characters unsafe for filenames.
  safe = "".join(ch if ch.isalnum() or ch in ("-", "_") else "-" for ch in basename)
  return safe 


# =====================
# Inputs
# =====================
repo_source = input("Enter repository local path or Git URL: ").strip()
start_date_str = input("Enter the start date (YYYY-MM-DD): ").strip()
end_date_str = input("Enter the end date (YYYY-MM-DD): ").strip()

# Validate and parse dates.
try:
  start_dt = datetime.strptime(start_date_str, "%Y-%m-%d")
  end_dt = datetime.strptime(end_date_str, "%Y-%m-%d")
except ValueError:
  raise SystemExit("Invalid date format. Please use YYYY-MM-DD.")

if end_dt < start_dt:
  raise SystemExit("End date must be on or after start date.")

start_time = datetime.now()

# =====================
# Extraction
# =====================
print("\n# =====================\n# Extraction\n# =====================")
print(f"Extracting activity for: {extract_repo_name(repo_source)}")
print(f"Date range: {start_dt.strftime('%Y-%m-%d')} to {end_dt.strftime('%Y-%m-%d')}")
print("Starting traversal...\n")


# =====================
# Repository Traversal
# =====================
repository = Repository(
  repo_source,
  since=start_dt,
  to=end_dt,
  num_workers=3,
)

rows = [] # List to hold all extracted data rows.

for commit in repository.traverse_commits():
  commit_info = {
    "hash": commit.hash,
    "date": commit.committer_date,
    "committer": getattr(commit.committer, "name", str(commit.committer)),
    "message": commit.msg,
  }

  for mod in commit.modified_files:
    file_data = {
      "filename": mod.filename,
      "added_lines": mod.added_lines,
      "deleted_lines": mod.deleted_lines,
      "token_count": mod.token_count,
      "nloc": mod.nloc,
    }
    rows.append({**commit_info, **file_data})


# =====================
# Output
# =====================
print("\n# =====================\n# Output\n# =====================")
os.makedirs(OUTPUT_DIR, exist_ok=True)
repo_name = extract_repo_name(repo_source)
date_range = f"{start_dt.strftime('%Y-%m-%d')}_to_{end_dt.strftime('%Y-%m-%d')}"
output_file = os.path.join(OUTPUT_DIR, f"{repo_name}_{date_range}_activity.csv")

df = pd.DataFrame(rows, columns=COLUMNS)
df.to_csv(output_file, index=False)

end_time = datetime.now()

pprint({
  "output": output_file,
  "rows": len(df),
  "execution_time": str(end_time - start_time),
})