Coverage for src/qdrant_loader_core/graph/extractor/jira.py: 98%
95 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:12 +0000
1from __future__ import annotations
3import re
4from typing import TYPE_CHECKING
5from urllib.parse import urlparse
7if TYPE_CHECKING:
8 from qdrant_loader.core.document import Document
10from qdrant_loader_core.graph.extractor.base_extractor import (
11 BaseEntityExtractor,
12)
13from qdrant_loader_core.graph.models import (
14 CoreEdgeType,
15 CoreNodeLabel,
16 GraphEdge,
17 GraphNode,
18 PersonInfo,
19)
22class JiraEntityExtractor(BaseEntityExtractor):
23 """
24 Jira graph extractor.
26 Extracts:
27 - Document node (status, priority, issue_type)
28 - Person nodes (reporter, assignee)
29 - Container node (jira project)
30 - Label nodes
31 - Linked issue relationships
32 - Parent/subtask relationships
33 - Cross-source links from URLs found in description
34 """
36 # NOTE:
37 # `parent_key` covers both subtask-of-parent and story/task-of-epic relations
38 # (Jira surfaces both through the same "parent" field); the PART_OF edge's
39 # "kind" is derived from issue_type to distinguish the two.
41 source_type = "jira"
43 CONFLUENCE_URL_RE = re.compile(
44 r"https?://[^\s]*confluence[^\s]+",
45 re.IGNORECASE,
46 )
48 GIT_URL_RE = re.compile(
49 r"https?://(?:github|gitlab|bitbucket)[^\s]+",
50 re.IGNORECASE,
51 )
53 # ------------------------------------------------------------------
54 # Project
55 # ------------------------------------------------------------------
57 def _project(
58 self,
59 doc: Document,
60 ) -> str | None:
61 return doc.metadata.get("project_key")
63 # ------------------------------------------------------------------
64 # Document
65 # ------------------------------------------------------------------
67 def _build_document_node(
68 self,
69 doc: Document,
70 project: str | None,
71 ) -> GraphNode:
72 metadata = doc.metadata
74 return GraphNode(
75 id=metadata.get("key"),
76 label=CoreNodeLabel.DOCUMENT.value,
77 project=project,
78 properties={
79 "title": doc.title,
80 "source_type": self.source_type,
81 "status": metadata.get("status"),
82 "priority": metadata.get("priority"),
83 "issue_type": metadata.get("issue_type"),
84 },
85 )
87 # ------------------------------------------------------------------
88 # People
89 # ------------------------------------------------------------------
91 def _extract_people(
92 self,
93 doc: Document,
94 ) -> list[tuple[PersonInfo, str]]:
95 people = []
97 reporter = doc.metadata.get("reporter")
99 if reporter:
100 if isinstance(reporter, dict):
101 person_id = (
102 reporter.get("email_address")
103 or reporter.get("account_id")
104 or reporter.get("display_name")
105 )
106 display_name = reporter.get("display_name", person_id)
107 else:
108 person_id = str(reporter)
109 display_name = person_id
111 if person_id:
112 people.append(
113 (PersonInfo(id=person_id, display_name=display_name), "reporter")
114 )
116 assignee = doc.metadata.get("assignee")
118 if assignee:
119 if isinstance(assignee, dict):
120 person_id = (
121 assignee.get("email_address")
122 or assignee.get("account_id")
123 or assignee.get("display_name")
124 )
125 display_name = assignee.get("display_name", person_id)
126 else:
127 person_id = str(assignee)
128 display_name = person_id
130 if person_id:
131 people.append(
132 (PersonInfo(id=person_id, display_name=display_name), "assignee")
133 )
135 return people
137 # ------------------------------------------------------------------
138 # Container
139 # ------------------------------------------------------------------
141 def _extract_container(
142 self,
143 doc: Document,
144 ) -> GraphNode | None:
145 project_key = doc.metadata.get("project_key")
147 if not project_key:
148 return None
150 return GraphNode(
151 id=project_key,
152 label=CoreNodeLabel.CONTAINER.value,
153 project=project_key,
154 properties={
155 "kind": "jira_project",
156 "project_key": project_key,
157 },
158 )
160 # ------------------------------------------------------------------
161 # Labels
162 # ------------------------------------------------------------------
164 def _extract_labels(
165 self,
166 doc: Document,
167 ) -> list[GraphNode]:
168 project = self._project(doc)
169 metadata = doc.metadata
170 nodes: list[GraphNode] = []
172 for label in metadata.get("labels", []):
173 nodes.append(
174 GraphNode(
175 id=label,
176 label=CoreNodeLabel.LABEL.value,
177 project=project,
178 properties={
179 "name": label,
180 },
181 )
182 )
184 return nodes
186 # ------------------------------------------------------------------
187 # Cross-source URL links
188 # ------------------------------------------------------------------
189 def _extract_links(
190 self,
191 doc: Document,
192 ) -> tuple[list[GraphNode], list[GraphEdge]]:
193 project = self._project(doc)
194 metadata = doc.metadata
195 text = metadata.get("description") or getattr(doc, "content", "") or ""
197 nodes: list[GraphNode] = []
198 edges: list[GraphEdge] = []
200 def _is_url(value: str) -> bool:
201 try:
202 parsed = urlparse(value)
203 return bool(parsed.scheme and parsed.netloc)
204 except Exception:
205 return False
207 def _handle_url(url: str, kind: str) -> None:
208 target = url.strip()
210 if _is_url(target):
211 nodes.append(
212 GraphNode(
213 id=target,
214 label=CoreNodeLabel.URL.value,
215 project=project,
216 properties={"url": target},
217 )
218 )
220 edges.append(
221 GraphEdge(
222 source=metadata.get("key"),
223 target=target,
224 edge_type=CoreEdgeType.LINKS_TO.value,
225 project=project,
226 properties={"kind": kind},
227 )
228 )
230 for url in self.CONFLUENCE_URL_RE.findall(text):
231 _handle_url(url, "confluence")
233 for url in self.GIT_URL_RE.findall(text):
234 _handle_url(url, "git")
236 return nodes, edges
238 # ------------------------------------------------------------------
239 # Jira-specific relationships
240 # ------------------------------------------------------------------
242 def _extract_source_specific(
243 self,
244 doc: Document,
245 ) -> tuple[list[GraphNode], list[GraphEdge]]:
246 project = self._project(doc)
248 edges: list[GraphEdge] = []
250 metadata = doc.metadata
252 # --------------------------------------------------------------
253 # Linked Issues
254 # --------------------------------------------------------------
256 # linked_issue_details carries relation/direction; linked_issues is the
257 # legacy plain-key list kept for documents indexed before that field existed.
258 links = metadata.get("linked_issue_details") or metadata.get(
259 "linked_issues", []
260 )
261 for link in links:
262 if isinstance(link, dict):
263 target_key = link.get("key")
264 kind = link.get("relation") or link.get("link_type") or "related"
265 direction = link.get("direction")
266 else:
267 # Backward-compat: plain issue key string, no type/direction info.
268 target_key = link
269 kind = "related"
270 direction = None
272 if not target_key:
273 continue
275 properties = {"kind": kind}
276 if direction:
277 properties["direction"] = direction
279 edges.append(
280 GraphEdge(
281 source=metadata.get("key"),
282 target=target_key,
283 edge_type=CoreEdgeType.LINKS_TO.value,
284 project=project,
285 properties=properties,
286 )
287 )
288 # --------------------------------------------------------------
289 # Parent Issue
290 # --------------------------------------------------------------
292 parent_issue = metadata.get("parent_key")
294 if parent_issue:
295 # Jira's "parent" field means "subtask of" for issue_type=Sub-task, but
296 # for other types (Story/Task/Bug) it means "child of an Epic" instead —
297 # derive the relation from issue_type rather than assuming subtask.
298 issue_type = (metadata.get("issue_type") or "").strip().lower()
299 kind = "subtask" if issue_type in {"sub-task", "subtask"} else "child"
301 edges.append(
302 GraphEdge(
303 source=metadata.get("key"),
304 target=parent_issue,
305 edge_type=CoreEdgeType.PART_OF.value,
306 project=project,
307 properties={
308 "kind": kind,
309 },
310 )
311 )
313 return [], edges