diff --git a/pygitweb/main.py b/pygitweb/main.py
index 52cfed3..f94e4cc 100644
--- a/pygitweb/main.py
+++ b/pygitweb/main.py
@@ -10,6 +10,7 @@ import subprocess
import tempfile
import zipfile
from contextlib import suppress
+from datetime import UTC, datetime
from pathlib import Path
from typing import Annotated
from urllib.parse import quote
@@ -43,7 +44,7 @@ from pygitweb.actions import (
summary_ref_state,
)
from pygitweb.config import ACTIONS, check_loadavg, settings
-from pygitweb.formatting import esc_html
+from pygitweb.formatting import age_string, esc_html
from pygitweb.git_helpers import git_get_references, git_get_type
from pygitweb.projects import git_get_project_owner, git_get_projects_list
from pygitweb.settings import router as settings_router
@@ -55,6 +56,7 @@ from pygitweb.tasks import (
task_router,
)
from pygitweb.templates_env import POSTAMBLE, PREAMBLE, env
+from pygitweb.timeline_cache import get_all_timeline_events, warm_timeline_cache
from pygitweb.validation import is_valid_action, is_valid_pathname, is_valid_project
with open("pygitweb/README.md", encoding="utf-8") as _readme_file:
@@ -130,6 +132,11 @@ app.include_router(task_router, prefix="/tasks")
app.include_router(comment_router, prefix="/comments")
+@app.on_event("startup")
+async def warm_activity_timeline_cache() -> None:
+ warm_timeline_cache()
+
+
@app.middleware("http")
async def loadavg_middleware(request: Request, call_next):
try:
@@ -156,6 +163,26 @@ def _validate_project(project: str | None) -> str:
return project
+def _format_activity_time(epoch: int) -> str:
+ try:
+ age_seconds = datetime.now(UTC).timestamp() - float(epoch)
+ if age_seconds <= 0:
+ return "right now"
+ return age_string(age_seconds)
+ except (ValueError, OSError):
+ return ""
+
+
+def _truncate_activity_description(description: str | None, max_len: int = 80) -> str:
+ if not description:
+ return ""
+ if "\n" in description:
+ return description.splitlines()[0][:max_len].rstrip() + "..."
+ if len(description) > max_len:
+ return description[:max_len].rstrip() + "..."
+ return description
+
+
# ---------- Routes (no project) ----------
@@ -234,6 +261,63 @@ def git_project_index(
return PlainTextResponse("\n".join(lines), media_type="text/plain; charset=utf-8")
+@app.get("/activity", response_class=HTMLResponse)
+def activity_page(
+ request: Request,
+ page: Annotated[str | None, Query(alias="page")] = None,
+ pagecount: Annotated[str | None, Query(alias="pagecount")] = None,
+) -> HTMLResponse:
+ p, pc = parse_pagination(page, pagecount)
+ all_events = get_all_timeline_events()
+ skip = (p - 1) * pc
+ events_page = all_events[skip : skip + pc + 1]
+ has_next = len(events_page) > pc
+ if has_next:
+ events_page = events_page[:pc]
+ rows: list[list[str]] = []
+ for item in events_page:
+ project_name = item["project"]
+ event = item["event"]
+ oid = event["oid"]
+ if event["kind"] == "commit":
+ event_label = (
+ f"{esc_html(oid[:7])}"
+ )
+ else:
+ event_label = esc_html(event["kind"])
+ description = _truncate_activity_description(event.get("description"))
+ rows.append(
+ [
+ esc_html(_format_activity_time(event["timestamp"])),
+ f"{esc_html(project_name)}",
+ event_label,
+ esc_html(description),
+ ]
+ )
+ table = env.get_template("table.html").render(
+ cols=["Time", "Project Name", "Event", "Description"],
+ rows=rows,
+ )
+ prev_url = ""
+ next_url = ""
+ if p > 1:
+ prev_url = f"{request.url.path}?page={p - 1}&pagecount={pc}"
+ if has_next:
+ next_url = f"{request.url.path}?page={p + 1}&pagecount={pc}"
+ pagination_html = env.get_template("pagination.html").render(
+ current_page=p,
+ pagecount=pc,
+ has_prev=p > 1,
+ has_next=has_next,
+ prev_url=prev_url,
+ next_url=next_url,
+ total_pages=None,
+ page_links=None,
+ )
+ pre = PREAMBLE.render(title=f"{esc_html(settings.SITE_NAME)} - Activity", site_name=settings.SITE_NAME)
+ return HTMLResponse(f"{pre}Activity
{table}{pagination_html}{POSTAMBLE}")
+
+
@app.get("/opml", response_class=PlainTextResponse)
def git_opml():
"""OPML feed list. Port of git_opml (stub)."""
diff --git a/pygitweb/templates/preamble.html b/pygitweb/templates/preamble.html
index df4cce6..8345df8 100644
--- a/pygitweb/templates/preamble.html
+++ b/pygitweb/templates/preamble.html
@@ -42,7 +42,7 @@
-
+
Activity
diff --git a/pygitweb/timeline_cache.py b/pygitweb/timeline_cache.py
new file mode 100644
index 0000000..9bc2f7e
--- /dev/null
+++ b/pygitweb/timeline_cache.py
@@ -0,0 +1,166 @@
+"""
+Per-project timeline cache for recent repository activity.
+"""
+
+from __future__ import annotations
+
+from typing import Literal, TypedDict
+
+import pygit2
+
+from pygitweb.config import settings
+from pygitweb.git_helpers import open_repo
+from pygitweb.projects import git_get_projects_list
+
+TIMELINE_CACHE_SIZE: int = 10
+
+
+class TimelineEvent(TypedDict):
+ timestamp: int
+ oid: str
+ kind: Literal["commit", "ref"]
+ description: str | None
+
+
+class TimelineProjectEvent(TypedDict):
+ project: str
+ event: TimelineEvent
+
+
+PROJECT_TIMELINE_CACHE: dict[str, list[TimelineEvent]] = {}
+
+
+def _commit_description(commit: pygit2.Commit) -> str | None:
+ if not commit.message:
+ return None
+ first_line: str = commit.message.splitlines()[0].strip()
+ return first_line or None
+
+
+def _collect_commit_events(repo: pygit2.Repository) -> list[TimelineEvent]:
+ start_points: list[pygit2.Oid] = []
+ for ref_name in repo.references:
+ if not ref_name.startswith("refs/heads/"):
+ continue
+ try:
+ resolved: pygit2.Reference = repo.references[ref_name].resolve()
+ oid: pygit2.Oid = resolved.target
+ start_points.append(oid)
+ except (KeyError, pygit2.GitError, ValueError):
+ continue
+ if not start_points:
+ try:
+ head_ref: pygit2.Reference = repo.head
+ start_points.append(head_ref.target)
+ except (pygit2.GitError, ValueError, KeyError):
+ pass
+ if not start_points:
+ return []
+
+ walker: pygit2.Walker = repo.walk(
+ start_points[0],
+ pygit2.GIT_SORT_TIME | pygit2.GIT_SORT_TOPOLOGICAL,
+ )
+ for oid in start_points[1:]:
+ try:
+ walker.push(oid)
+ except (pygit2.GitError, ValueError):
+ continue
+
+ events: list[TimelineEvent] = []
+ seen: set[str] = set()
+ for commit in walker:
+ commit_oid: str = str(commit.id)
+ if commit_oid in seen:
+ continue
+ seen.add(commit_oid)
+ events.append(
+ {
+ "timestamp": commit.commit_time,
+ "oid": commit_oid,
+ "kind": "commit",
+ "description": _commit_description(commit),
+ }
+ )
+ return events
+
+
+def _collect_ref_events(repo: pygit2.Repository) -> list[TimelineEvent]:
+ events: list[TimelineEvent] = []
+ for ref_name in repo.references:
+ if not ref_name.startswith("refs/tags/"):
+ continue
+ try:
+ ref_obj: pygit2.Reference = repo.references[ref_name].resolve()
+ oid_str: str = str(ref_obj.target)
+ obj = repo[ref_obj.target]
+ except (KeyError, pygit2.GitError, ValueError):
+ continue
+
+ if isinstance(obj, pygit2.Tag):
+ tag_tagger = obj.tagger
+ timestamp: int = tag_tagger.time if tag_tagger else 0
+ description: str | None = ref_name.replace("refs/tags/", "", 1)
+ elif isinstance(obj, pygit2.Commit):
+ timestamp = obj.commit_time
+ description = ref_name.replace("refs/tags/", "", 1)
+ else:
+ continue
+
+ events.append(
+ {
+ "timestamp": timestamp,
+ "oid": oid_str,
+ "kind": "ref",
+ "description": description or None,
+ }
+ )
+ return events
+
+
+def build_project_timeline_cache(project: str, max_events: int = TIMELINE_CACHE_SIZE) -> list[TimelineEvent]:
+ try:
+ repo: pygit2.Repository = open_repo(project)
+ except (pygit2.GitError, OSError, ValueError):
+ return []
+
+ commit_events: list[TimelineEvent] = _collect_commit_events(repo)
+ ref_events: list[TimelineEvent] = _collect_ref_events(repo)
+ all_events: list[TimelineEvent] = commit_events + ref_events
+ all_events.sort(key=lambda event: (event["timestamp"], event["oid"]), reverse=True)
+ return all_events[:max_events]
+
+
+def warm_timeline_cache() -> None:
+ global PROJECT_TIMELINE_CACHE
+ new_cache: dict[str, list[TimelineEvent]] = {}
+ projects = git_get_projects_list(
+ filter_path="",
+ paranoid=settings.STRICT_EXPORT,
+ export_ok=settings.EXPORT_OK,
+ )
+ for project_entry in projects:
+ project_name = project_entry.get("path")
+ if not isinstance(project_name, str) or not project_name:
+ continue
+ try:
+ new_cache[project_name] = build_project_timeline_cache(project_name)
+ except (pygit2.GitError, OSError, ValueError, KeyError):
+ new_cache[project_name] = []
+ PROJECT_TIMELINE_CACHE = new_cache
+
+
+def get_project_timeline_cache(project: str) -> list[TimelineEvent]:
+ return PROJECT_TIMELINE_CACHE.get(project, [])
+
+
+def get_all_timeline_events() -> list[TimelineProjectEvent]:
+ events: list[TimelineProjectEvent] = []
+ for project_name, project_events in PROJECT_TIMELINE_CACHE.items():
+ for event in project_events:
+ events.append({"project": project_name, "event": event})
+ events.sort(
+ key=lambda item: (item["event"]["timestamp"], item["event"]["oid"], item["project"]),
+ reverse=True,
+ )
+ return events