diff --git a/src/contributor_stats/main.py b/src/contributor_stats/main.py index c5bd9d8..1ba62b0 100644 --- a/src/contributor_stats/main.py +++ b/src/contributor_stats/main.py @@ -1,10 +1,11 @@ import sys import os +import subprocess import unicodedata import logging from pathlib import Path -from typing import Dict, Tuple, Any, MutableMapping -from collections import OrderedDict +from typing import Any, Dict, Iterable, List, MutableMapping, Optional, Set, Tuple +from collections import OrderedDict, defaultdict from contextlib import contextmanager original_cwd = os.getcwd() @@ -17,9 +18,20 @@ AuthorInfo = MutableMapping[str, dict] Table = MutableMapping[str, AuthorInfo] +# Author name -> commit email -> number of commits made with it +Emails = MutableMapping[str, MutableMapping[str, int]] zero_row = OrderedDict(commits=0, active_days=[], lines_added=0, lines_removed=0, blame=0) +# Emails that identify a git client default or a shared/anonymous address rather +# than a person, and must therefore never be used to merge two authors together. +GENERIC_EMAILS = { + "noreply@github.com", + "you@example.com", + "root@localhost", + "git@localhost", +} + def foldername(path) -> str: if os.path.isdir(path): @@ -39,25 +51,159 @@ def merge_author(a1: MutableMapping, a2: MutableMapping) -> AuthorInfo: return a1 -def get_authorInfos(data) -> AuthorInfo: +def normalize_name(name: str) -> str: + """ + Normalize an author name, so that different spellings of the same name compare equal. + """ + # Run the following and be amazed by the power of Unicode: + # bool('å' == "å") # False + # This weird unicode char was in Måns name, so now we have to unicode normalize everything. + # Never done this before, so thanks for making me learn Måns, or perhaps should I write Måns. + new_name = unicodedata.normalize("NFKC", name) + if new_name != name: + logger.info("Name '{}' was normalized to '{}'".format(name, new_name)) + name = new_name + + return name.replace("å", "å") + + +def normalize_email(email: str) -> str: + """ + Normalize a commit email, returning an empty string for the emails that can't be + used to identify an author (malformed, or a well-known generic address). + """ + email = email.strip().lower() + if "@" not in email or email in GENERIC_EMAILS: + return "" + return email + + +def git_author_emails(path) -> Emails: + """ + Maps each author name in the history to the emails they have committed with, + and how many commits they made with each of them. + + Uses the same (mailmap-resolved) names as gitstats, so that the keys line up + with the names in the tables. + """ + output = subprocess.check_output( + ["git", "-C", str(path), "log", "--format=%aN%x09%aE", "HEAD"], text=True + ) + emails: Emails = defaultdict(lambda: defaultdict(int)) + for line in output.splitlines(): + name, _, email = line.partition("\t") + email = normalize_email(email) + if not name or not email: + continue + emails[normalize_name(name)][email] += 1 + return emails + + +def merge_emails(emails: Iterable[Emails]) -> Emails: + """Combines the author -> email mappings of several repos into one.""" + merged: Emails = defaultdict(lambda: defaultdict(int)) + for repo_emails in emails: + for name, counts in repo_emails.items(): + for email, commits in counts.items(): + merged[name][email] += commits + return merged + + +def group_names_by_email(emails: Emails) -> List[List[str]]: + """ + Groups the author names that share at least one commit email. + + Grouping is transitive: if "a" and "b" share an email, and "b" and "c" share + another, then all three end up in the same group. + """ + parent = {name: name for name in emails} + + def find(name: str) -> str: + while parent[name] != name: + parent[name] = parent[parent[name]] + name = parent[name] + return name + + def union(name: str, other: str) -> None: + root, other_root = find(name), find(other) + if root != other_root: + parent[other_root] = root + + names_by_email: Dict[str, List[str]] = defaultdict(list) + for name, counts in emails.items(): + for email in counts: + names_by_email[email].append(name) + + for group in names_by_email.values(): + for other in group[1:]: + union(group[0], other) + + groups: Dict[str, List[str]] = defaultdict(list) + for name in parent: + groups[find(name)].append(name) + return list(groups.values()) + + +def author_aliases(emails: Emails) -> Dict[str, str]: + """ + Maps every author name that has an alias (another name used with one of the + same commit emails) to the single name to display them under. + + The name with the most commits wins, ties broken by name to keep the result + deterministic (which also happens to prefer "Brayo" over "brayo", since + uppercase sorts first). The mapping is computed across all repos at once, so + that the same person gets the same name in every table. + """ + aliases = {} + for group in group_names_by_email(emails): + if len(group) < 2: + continue + keep = min(group, key=lambda name: (-sum(emails[name].values()), name)) + logger.info("Merging {} into '{}' (shared commit email)".format( + sorted(name for name in group if name != keep), keep)) + for name in group: + if name != keep: + aliases[name] = keep + return aliases + + +def get_authorInfos(data, aliases: Optional[Dict[str, str]] = None) -> Tuple[AuthorInfo, Dict[str, str]]: + """ + Returns the stats per author, with the aliases of a person merged into a single + entry, and a mapping from every name in the history to the name it was merged + into (used to attribute blame lines to the merged author). + """ names = data.getAuthors() + aliases = dict(aliases or {}) - authorInfos = {} + authorInfos: AuthorInfo = {} for name in names: _authorInfo = data.getAuthorInfo(name) - - # Run the following and be amazed by the power of Unicode: - # bool('å' == "å") # False - # This weird unicode char was in Måns name, so now we have to unicode normalize everything. - # Never done this before, so thanks for making me learn Måns, or perhaps should I write Måns. - _new_name = unicodedata.normalize("NFKC", name) - if _new_name != name: - logger.info("Name '{}' was normalized to '{}'".format(name, _new_name)) - name = _new_name - - name = name.replace("å", "å") + name = normalize_name(name) + if name in authorInfos: + # Two spellings of the name normalized into the same one + _authorInfo = merge_author(dict(authorInfos[name]), _authorInfo) authorInfos[name] = _authorInfo + # Every name maps to itself, until merged into another one + merged_into = {name: name for name in authorInfos} + + def merge_into(keep: str, alias: str) -> None: + merged: Any = authorInfos.pop(alias) + if keep in authorInfos: + merged = merge_author(dict(authorInfos[keep]), merged) + authorInfos[keep] = merged + for name, into in merged_into.items(): + if into == alias: + merged_into[name] = keep + + # Merge the names that committed with the same email + for alias in sorted(authorInfos): + keep = aliases.get(alias, alias) + if keep != alias: + merge_into(keep, alias) + + # Manual merges, for the aliases that don't share a commit email author_merges = [ ("Erik Bjäreholt", ["Erik Bjäreholt", "Erik Bjareholt"]), ("Johan Bjäreholt", ["johan-bjareholt"]), @@ -69,16 +215,12 @@ def get_authorInfos(data) -> AuthorInfo: ("Otto-AA", ["A_A"]), ("Brayo", ["brayo"]) ] - for name, aliases in author_merges: - for alias in aliases: + for name, _aliases in author_merges: + for alias in _aliases: if alias in authorInfos: - to_keep = authorInfos.pop(alias) - if name in authorInfos: - to_merge_with = authorInfos.pop(name) - to_keep = merge_author(to_merge_with, to_keep) - authorInfos[name] = to_keep + merge_into(name, alias) - return authorInfos + return authorInfos, merged_into def git_blame_stats(path) -> dict[str, int]: @@ -86,7 +228,6 @@ def git_blame_stats(path) -> dict[str, int]: Uses the following command to get stats of lines last touched by each author: git ls-tree --name-only -z -r HEAD -- $1 | xargs -0 -n1 git blame --line-porcelain | grep "^author "|sort|uniq -c|sort -nr """ - import subprocess os.chdir(path) output = subprocess.check_output( "git ls-tree --name-only -z -r HEAD -- . | xargs -0 -n1 git blame --line-porcelain | grep '^author '|sort|uniq -c|sort -nr", shell=True, text=True @@ -98,8 +239,10 @@ def git_blame_stats(path) -> dict[str, int]: return {" ".join(line.split()[2:]): int(line.split()[0]) for line in output.split("\n") if line} -def generate_from_repo(path: Path) -> Tuple[str, Table]: +def generate_from_repo(path: Path, aliases: Optional[Dict[str, str]] = None) -> Tuple[str, Table]: path = Path(path).resolve() + if aliases is None: + aliases = author_aliases(git_author_emails(path)) # TODO: Could use caching to speed up (not much point since it usually runs in CI) data = gitstats.GitDataCollector() @@ -117,10 +260,17 @@ def generate_from_repo(path: Path) -> Tuple[str, Table]: print("Generated stats for: {}".format(data.projectname)) rows = {} - authorInfos = get_authorInfos(data) + authorInfos, merged_into = get_authorInfos(data, aliases) + + # Attribute the blame lines of every alias to the author it was merged into + blame_by_author: Dict[str, int] = defaultdict(int) + for blamed_name, lines in blame.items(): + name = normalize_name(blamed_name) + blame_by_author[merged_into.get(name, name)] += lines + for name, info in authorInfos.items(): rows[name] = merge_author(zero_row.copy(), info) - rows[name]["blame"] = blame.get(name, 0) # type: ignore[assignment] + rows[name]["blame"] = blame_by_author.get(name, 0) # type: ignore[assignment] for name in rows: rows[name]["blame_percent"] = rows[name]["blame"] / blame_lines * 100 # type: ignore @@ -242,8 +392,12 @@ def main(): print("Found repos: {}".format([str(r) for r in repos])) + # Resolve aliases across all repos at once, so that a person gets the same + # name in every table (and thus a single row in the merged "total" table) + aliases = author_aliases(merge_emails(git_author_emails(path) for path in repos)) + for path in repos: - repo_name, rows = generate_from_repo(str(path)) + repo_name, rows = generate_from_repo(str(path), aliases) tables[repo_name] = rows tables["total"] = merge_tables(tables) diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..3ad53b8 --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,152 @@ +from contributor_stats.main import ( + author_aliases, + get_authorInfos, + group_names_by_email, + merge_emails, + normalize_email, + normalize_name, +) + + +class FakeData: + """Stands in for gitstats.GitDataCollector, which needs an actual repo.""" + + def __init__(self, authors): + self.authors = authors + + def getAuthors(self): + return list(self.authors) + + def getAuthorInfo(self, author): + return dict(self.authors[author]) + + +def author(commits, active_days, lines_added=0, lines_removed=0): + return dict( + commits=commits, + active_days=active_days, + lines_added=lines_added, + lines_removed=lines_removed, + ) + + +def test_normalize_name(): + # NFD and NFC spellings of the same name are the same author + assert normalize_name("Måns") == normalize_name("Måns") + + +def test_normalize_email(): + assert normalize_email(" Erik@Bjareho.lt ") == "erik@bjareho.lt" + # Not usable to identify an author + assert normalize_email("not an email") == "" + assert normalize_email("noreply@github.com") == "" + + +def test_group_names_by_email(): + groups = group_names_by_email( + { + "Brayo": {"vukubrian@gmail.com": 2, "brayo@laptop.local": 1}, + "brayo": {"vukubrian@gmail.com": 1}, + "Erik": {"erik@bjareho.lt": 1}, + } + ) + assert sorted(sorted(group) for group in groups) == [["Brayo", "brayo"], ["Erik"]] + + +def test_group_names_by_email_is_transitive(): + groups = group_names_by_email( + { + "a": {"one@example.com": 1}, + "b": {"one@example.com": 1, "two@example.com": 1}, + "c": {"two@example.com": 1}, + } + ) + assert [sorted(group) for group in groups] == [["a", "b", "c"]] + + +def test_author_aliases_picks_the_name_with_most_commits(): + aliases = author_aliases( + {"Brayo": {"vukubrian@gmail.com": 5}, "brayo": {"vukubrian@gmail.com": 1}} + ) + assert aliases == {"brayo": "Brayo"} + + +def test_author_aliases_ties_are_deterministic(): + emails = {"Brayo": {"vukubrian@gmail.com": 1}, "brayo": {"vukubrian@gmail.com": 1}} + assert author_aliases(emails) == {"brayo": "Brayo"} + assert author_aliases(dict(reversed(list(emails.items())))) == {"brayo": "Brayo"} + + +def test_author_aliases_keeps_authors_without_a_shared_email_apart(): + assert author_aliases( + {"Brayo": {"vukubrian@gmail.com": 1}, "Erik": {"erik@bjareho.lt": 1}} + ) == {} + + +def test_merge_emails(): + merged = merge_emails( + [ + {"Brayo": {"vukubrian@gmail.com": 2}}, + {"Brayo": {"vukubrian@gmail.com": 3}, "brayo": {"vukubrian@gmail.com": 1}}, + ] + ) + assert merged == { + "Brayo": {"vukubrian@gmail.com": 5}, + "brayo": {"vukubrian@gmail.com": 1}, + } + + +def test_get_authorInfos_merges_aliases(): + data = FakeData( + { + "Brayo": author(commits=2, active_days=["2025-01-01", "2025-01-02"], lines_added=10), + "brayo": author(commits=1, active_days=["2025-01-02"], lines_added=5), + "Erik": author(commits=1, active_days=["2025-01-03"], lines_added=1), + } + ) + authorInfos, merged_into = get_authorInfos(data, {"brayo": "Brayo"}) + + assert set(authorInfos) == {"Brayo", "Erik"} + assert authorInfos["Brayo"]["commits"] == 3 + assert authorInfos["Brayo"]["lines_added"] == 15 + assert authorInfos["Brayo"]["active_days"] == {"2025-01-01", "2025-01-02"} + # so that blame lines of an alias are attributed to the merged author + assert merged_into["brayo"] == "Brayo" + assert merged_into["Brayo"] == "Brayo" + + +def test_get_authorInfos_merges_into_a_name_not_in_the_repo(): + data = FakeData({"brayo": author(commits=1, active_days=["2025-01-01"])}) + authorInfos, merged_into = get_authorInfos(data, {"brayo": "Brayo"}) + + assert set(authorInfos) == {"Brayo"} + assert authorInfos["Brayo"]["commits"] == 1 + assert merged_into["brayo"] == "Brayo" + + +def test_get_authorInfos_still_applies_the_manual_merges(): + # These aliases don't share a commit email, so they need the manual list + data = FakeData( + { + "dependabot[bot]": author(commits=1, active_days=["2025-01-01"]), + "dependabot-preview[bot]": author(commits=2, active_days=["2025-01-02"]), + } + ) + authorInfos, merged_into = get_authorInfos(data) + + assert set(authorInfos) == {"dependabot[bot]"} + assert authorInfos["dependabot[bot]"]["commits"] == 3 + assert merged_into["dependabot-preview[bot]"] == "dependabot[bot]" + + +def test_get_authorInfos_merges_names_that_normalize_to_the_same_one(): + data = FakeData( + { + "Måns": author(commits=1, active_days=["2025-01-01"]), + "Måns": author(commits=2, active_days=["2025-01-02"]), + } + ) + authorInfos, _ = get_authorInfos(data) + + assert len(authorInfos) == 1 + assert list(authorInfos.values())[0]["commits"] == 3