1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
"""Client-side reference manifest for board task @ / # lookups."""
from __future__ import annotations
import os
from datetime import UTC, datetime, timedelta, timezone
from typing import TypedDict
from urllib.parse import quote
from pygit2 import Oid, Repository, Tag
from pygittools.merge import MR_REF_PREFIX, get_merge_request_by_oid
from pygitweb.config import settings
from pygitweb.git_helpers import get_commit_history
from pygitweb.tasks import TASK_REF_PREFIX, get_board_tasks_grouped
BOARD_REF_MANIFEST_COMMIT_LIMIT = 100
class BoardRefManifestEntry(TypedDict):
name: str
insert: str
url: str
tags: str
snippet: str
created: str
class BoardRefManifest(TypedDict):
commits: list[BoardRefManifestEntry]
refs: list[BoardRefManifestEntry]
def _format_epoch(epoch: int | None, tz_offset: int | None = None) -> str:
if epoch is None:
return ""
try:
tz = timezone(timedelta(seconds=tz_offset * 60)) if tz_offset is not None else UTC
return datetime.fromtimestamp(epoch, tz=tz).strftime("%Y-%m-%d %H:%M:%S")
except (ValueError, OSError):
return ""
def _project_url(project: str) -> str:
return f"/project/{quote(project, safe='/')}"
def _board_task_url(project: str, board_name: str, task_ref: str) -> str:
base = f"{_project_url(project)}/board/"
qs = f"b={quote(board_name, safe='')}&task={quote(task_ref, safe='')}"
return f"{base}?{qs}" if board_name != "Tasks" else f"{base}?task={quote(task_ref, safe='')}"
def _task_display_id(ref: str) -> str:
slug = ref.removeprefix(TASK_REF_PREFIX)
if slug.startswith("task_"):
return slug.removeprefix("task_")
return slug
def _manifest_commits(project: str) -> list[BoardRefManifestEntry]:
project_url = _project_url(project)
entries: list[BoardRefManifestEntry] = []
for commit in get_commit_history(project, ref=None, max_count=BOARD_REF_MANIFEST_COMMIT_LIMIT):
oid = str(commit.get("oid") or "")
if not oid:
continue
short = oid[:7]
subject = str(commit.get("subject") or "")
author = str(commit.get("author") or "")
created = _format_epoch(commit.get("author_epoch"), commit.get("author_tz"))
search_tags = " ".join(part.lower() for part in (oid, short, subject, author) if part)
entries.append({
"name": f"{short} {subject}".strip(),
"insert": short,
"url": f"{project_url}?a=commit&h={quote(oid, safe='')}",
"tags": search_tags,
"snippet": author,
"created": created,
})
return entries
def _manifest_board_tasks(project: str, board_name: str) -> list[BoardRefManifestEntry]:
entries: list[BoardRefManifestEntry] = []
seen_refs: set[str] = set()
try:
columns = get_board_tasks_grouped(project, board_name)
except KeyError:
return entries
for column in columns:
for task in column["tasks"]:
task_ref = str(task.get("ref") or "")
if not task_ref or task_ref in seen_refs:
continue
seen_refs.add(task_ref)
task_id = str(task.get("id") or _task_display_id(task_ref))
title = str(task.get("title") or "Untitled")
slug = task_ref.removeprefix("refs/tags/")
created = str(task.get("updated_at") or task.get("created_at") or "")
search_tags = " ".join(
part.lower() for part in (task_id, slug, task_ref, title, str(task.get("status") or "")) if part
)
entries.append({
"name": title,
"insert": task_id,
"url": _board_task_url(project, board_name, task_ref),
"tags": search_tags,
"snippet": f"task {task_id}",
"created": created,
})
return entries
def _manifest_merge_requests(project: str) -> list[BoardRefManifestEntry]:
project_url = _project_url(project)
repo = Repository(os.path.join(settings.PROJECTROOT, project))
entries: list[BoardRefManifestEntry] = []
for ref_name in repo.references:
if not ref_name.startswith(MR_REF_PREFIX):
continue
try:
oid = repo.references[ref_name].resolve().target
obj = repo[oid]
if not isinstance(obj, Tag):
continue
mr = get_merge_request_by_oid(repo, Oid(hex=str(oid)))
except (ValueError, KeyError, TypeError):
continue
slug = ref_name.removeprefix(MR_REF_PREFIX)
title = mr.title or slug
created = mr.updated_at.isoformat() if mr.updated_at else ""
search_tags = " ".join(
part.lower() for part in (slug, ref_name, title, mr.description or "", mr.ours or "") if part
)
entries.append({
"name": title,
"insert": slug,
"url": f"{project_url}?a=tag&h={quote(ref_name, safe='')}",
"tags": search_tags,
"snippet": f"merge request {slug}",
"created": created,
})
return entries
def board_ref_manifest(project: str, board_name: str) -> BoardRefManifest:
refs = _manifest_board_tasks(project, board_name)
refs.extend(_manifest_merge_requests(project))
refs.sort(key=lambda entry: (entry["created"], entry["name"]), reverse=True)
return {
"commits": _manifest_commits(project),
"refs": refs,
}