| """The grading rule: which files two sessions wrote within an hour, read from a git log. |
| |
| A line-for-line port of the web grader's grade.js; a shared fixture test keeps the two verdicts equal. |
| """ |
| import math |
| import re |
| from datetime import datetime |
|
|
| COMMAND = "git log --all --source --name-only --since='1 month ago' --format='%H|%an|%aI|%S'" |
| WINDOW_MIN = 60 |
| HEAD = re.compile(r"^[0-9a-f]{7,64}\|") |
|
|
|
|
| def branch(ref): |
| """A ref as git prints it (%S, or %D's decorations) to a branch name; "" when the commit carries none.""" |
| names = [s.strip() for s in (ref or "").split(",")] |
| names = [re.sub(r"^HEAD -> ", "", s) for s in names] |
| names = [s for s in names if s and s != "HEAD" and not s.startswith("tag: ") and not s.startswith("refs/tags/")] |
| if not names: |
| return "" |
| return re.sub(r"^refs/remotes/[^/]+/", "", re.sub(r"^refs/heads/", "", names[0])) |
|
|
|
|
| def millis(text): |
| """An ISO 8601 date with its offset to epoch milliseconds; None when it does not read as one.""" |
| try: |
| when = datetime.fromisoformat(text.strip().replace("Z", "+00:00")) |
| except ValueError: |
| return None |
| return int(round(when.timestamp() * 1000)) |
|
|
|
|
| def parse(text): |
| commits, cur = [], None |
| for line in re.split(r"\r?\n", str(text or "")): |
| if HEAD.match(line): |
| p = line.split("|") |
| when = millis(p[-2]) if len(p) >= 4 else None |
| if when is not None: |
| cur = {"hash": p[0], "author": "|".join(p[1:-2]), "when": when, "ref": branch(p[-1]), "files": []} |
| commits.append(cur) |
| continue |
| f = line.strip() |
| if cur is not None and f: |
| cur["files"].append(f) |
| return commits |
|
|
|
|
| def two_sessions(a, b): |
| """Two commits are two sessions when their authors differ, or both carry a branch and the branches differ.""" |
| return a["author"] != b["author"] or bool(a["ref"] and b["ref"] and a["ref"] != b["ref"]) |
|
|
|
|
| def grade(text): |
| commits, by_file = parse(text), {} |
| for c in commits: |
| for f in c["files"]: |
| by_file.setdefault(f, []).append(c) |
| files = [] |
| for name, writes in by_file.items(): |
| writes = sorted(writes, key=lambda c: c["when"]) |
| pairs, latest = [], 0 |
| for i in range(1, len(writes)): |
| j = i - 1 |
| while j >= 0 and writes[i]["when"] - writes[j]["when"] <= WINDOW_MIN * 60000: |
| if two_sessions(writes[i], writes[j]): |
| minutes = int(math.floor((writes[i]["when"] - writes[j]["when"]) / 60000 + 0.5)) |
| pairs.append({"a": writes[j], "b": writes[i], "minutes": minutes}) |
| latest = max(latest, writes[i]["when"]) |
| j -= 1 |
| if pairs: |
| pairs.sort(key=lambda p: p["minutes"]) |
| files.append({"file": name, "count": len(pairs), "latest": latest, "pairs": pairs[:3]}) |
| files.sort(key=lambda f: (-f["count"], -f["latest"], f["file"])) |
| times = [c["when"] for c in commits] |
| days = max(1, math.ceil((max(times) - min(times)) / 864e5)) if times else 0 |
| sessions = {(c["author"], c["ref"]) for c in commits} |
| return {"commits": len(commits), "sessions": len(sessions), "days": days, "files": files} |
|
|
|
|
| def score_line(report): |
| n = len(report["files"]) |
| head = "No file" if n == 0 else "1 file" if n == 1 else "%d files" % n |
| return "%s written by two sessions within an hour, in %d %s of history." % ( |
| head, report["days"], "day" if report["days"] == 1 else "days") |
|
|
