summaryrefslogtreecommitdiff
path: root/src/srht_contrib/services
diff options
context:
space:
mode:
Diffstat (limited to 'src/srht_contrib/services')
-rw-r--r--src/srht_contrib/services/__init__.py1
-rw-r--r--src/srht_contrib/services/aggregator.py100
-rw-r--r--src/srht_contrib/services/git.py202
-rw-r--r--src/srht_contrib/services/srht_client.py77
-rw-r--r--src/srht_contrib/services/todo.py559
5 files changed, 939 insertions, 0 deletions
diff --git a/src/srht_contrib/services/__init__.py b/src/srht_contrib/services/__init__.py
new file mode 100644
index 0000000..02dea84
--- /dev/null
+++ b/src/srht_contrib/services/__init__.py
@@ -0,0 +1 @@
+"""Service layer."""
diff --git a/src/srht_contrib/services/aggregator.py b/src/srht_contrib/services/aggregator.py
new file mode 100644
index 0000000..800329a
--- /dev/null
+++ b/src/srht_contrib/services/aggregator.py
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from srht_contrib.models import ContributionEvent
+from srht_contrib.schemas import ContributionCalendarResponse, ContributionDay, ContributionStatsResponse
+from srht_contrib.utils.dates import date_range, date_to_utc_bounds
+
+
+@dataclass(slots=True)
+class DailyAggregate:
+ date: date
+ count: int
+ score: float
+
+
+class ContributionAggregator:
+ def build_calendar(self, db: Session, actor: str, start: date, end: date) -> ContributionCalendarResponse:
+ aggregates = self._query_daily_aggregates(db, actor, start, end)
+ by_day = {row.date: row for row in aggregates}
+ days = [
+ ContributionDay(
+ date=day,
+ count=by_day.get(day, DailyAggregate(date=day, count=0, score=0.0)).count,
+ score=by_day.get(day, DailyAggregate(date=day, count=0, score=0.0)).score,
+ )
+ for day in date_range(start, end)
+ ]
+ return ContributionCalendarResponse(actor=actor, from_date=start, to_date=end, days=days)
+
+ def build_stats(self, db: Session, actor: str, start: date, end: date) -> ContributionStatsResponse:
+ calendar = self.build_calendar(db, actor, start, end)
+ active_days = [day for day in calendar.days if day.count > 0]
+ streaks = self._streak_lengths(calendar.days)
+ current_streak = self._current_streak(calendar.days)
+
+ return ContributionStatsResponse(
+ actor=actor,
+ from_date=start,
+ to_date=end,
+ total_events=sum(day.count for day in calendar.days),
+ total_score=round(sum(day.score for day in calendar.days), 2),
+ active_days=len(active_days),
+ longest_streak=max(streaks, default=0),
+ current_streak=current_streak,
+ )
+
+ def _query_daily_aggregates(self, db: Session, actor: str, start: date, end: date) -> list[DailyAggregate]:
+ start_dt, _ = date_to_utc_bounds(start)
+ _, end_dt = date_to_utc_bounds(end)
+
+ stmt = (
+ select(
+ func.date(ContributionEvent.occurred_at).label("day"),
+ func.count(ContributionEvent.id).label("count"),
+ func.coalesce(func.sum(ContributionEvent.weight), 0.0).label("score"),
+ )
+ .where(ContributionEvent.actor == actor)
+ .where(ContributionEvent.occurred_at >= start_dt)
+ .where(ContributionEvent.occurred_at <= end_dt)
+ .group_by(func.date(ContributionEvent.occurred_at))
+ .order_by(func.date(ContributionEvent.occurred_at))
+ )
+ rows = db.execute(stmt).all()
+ return [
+ DailyAggregate(
+ date=date.fromisoformat(str(row.day)),
+ count=int(row.count),
+ score=round(float(row.score), 2),
+ )
+ for row in rows
+ ]
+
+ @staticmethod
+ def _streak_lengths(days: list[ContributionDay]) -> list[int]:
+ streaks: list[int] = []
+ current = 0
+ for day in days:
+ if day.count > 0:
+ current += 1
+ elif current > 0:
+ streaks.append(current)
+ current = 0
+ if current > 0:
+ streaks.append(current)
+ return streaks
+
+ @staticmethod
+ def _current_streak(days: list[ContributionDay]) -> int:
+ streak = 0
+ for day in reversed(days):
+ if day.count > 0:
+ streak += 1
+ else:
+ break
+ return streak
diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py
new file mode 100644
index 0000000..20a18fe
--- /dev/null
+++ b/src/srht_contrib/services/git.py
@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+import logging
+from typing import Any
+
+from srht_contrib.config import Settings
+from srht_contrib.schemas import NormalizedEvent
+from srht_contrib.services.srht_client import SourceHutGraphQLClient
+from srht_contrib.utils.dates import ensure_utc, parse_datetime
+from srht_contrib.utils.identity import ActorIdentityResolver
+
+
+logger = logging.getLogger(__name__)
+
+
+REPOSITORY_LOG_QUERY = """
+query RepositoryLog($username: String!, $repoName: String!, $cursor: Cursor) {
+ user(username: $username) {
+ repository(name: $repoName) {
+ name
+ owner {
+ canonicalName
+ }
+ log(cursor: $cursor) {
+ results {
+ id
+ shortId
+ author {
+ name
+ email
+ time
+ }
+ committer {
+ name
+ email
+ time
+ }
+ message
+ }
+ cursor
+ }
+ }
+ }
+}
+""".strip()
+
+
+@dataclass(slots=True)
+class GitPollResult:
+ events: list[NormalizedEvent]
+ cursor: str
+
+
+class GitIngestionService:
+ """Polls tracked git.sr.ht repositories and normalizes commits for one actor."""
+
+ service_name = "git"
+
+ def __init__(self, client: SourceHutGraphQLClient, settings: Settings) -> None:
+ self.client = client
+ self.settings = settings
+ self.identity_resolver = ActorIdentityResolver(settings.actor_aliases_json)
+
+ def fetch_recent_events(
+ self,
+ actor: str,
+ since: datetime | None = None,
+ repositories: list[str] | None = None,
+ ) -> GitPollResult:
+ since_dt = ensure_utc(since or (datetime.now(tz=UTC) - timedelta(days=30)))
+ tracked_repositories = repositories or self._tracked_repositories(actor)
+ if not tracked_repositories:
+ logger.info("git poll skipped for actor=%s because no tracked repositories are configured", actor)
+ return GitPollResult(events=[], cursor=datetime.now(tz=UTC).isoformat())
+
+ events: list[NormalizedEvent] = []
+ for repository in tracked_repositories:
+ owner, repo_name = self._split_repository(actor, repository)
+ repo_events = self._fetch_repository_commits(actor=actor, owner=owner, repo_name=repo_name, since=since_dt)
+ events.extend(repo_events)
+
+ logger.info("git poll complete for actor=%s normalized_events=%s", actor, len(events))
+ return GitPollResult(events=events, cursor=datetime.now(tz=UTC).isoformat())
+
+ def _tracked_repositories(self, actor: str) -> list[str]:
+ repositories = self.settings.git_tracked_repositories
+ return repositories
+
+ @staticmethod
+ def _split_repository(default_actor: str, repository: str) -> tuple[str, str]:
+ if "/" in repository:
+ owner, repo_name = repository.split("/", 1)
+ canonical_owner = owner if owner.startswith("~") else f"~{owner}"
+ return canonical_owner.lstrip("~"), repo_name
+ return default_actor.lstrip("~"), repository
+
+ def _fetch_repository_commits(
+ self,
+ *,
+ actor: str,
+ owner: str,
+ repo_name: str,
+ since: datetime,
+ ) -> list[NormalizedEvent]:
+ events: list[NormalizedEvent] = []
+ cursor: str | None = None
+
+ for _ in range(50):
+ data = self.client.execute(
+ REPOSITORY_LOG_QUERY,
+ {"username": owner, "repoName": repo_name, "cursor": cursor},
+ )
+ user = data.get("user") or {}
+ repository = user.get("repository") or {}
+ log_page = repository.get("log") or {}
+ commits = log_page.get("results") or []
+ cursor = log_page.get("cursor")
+ logger.info(
+ "git repository=%s/%s commit page count=%s next_cursor=%s",
+ owner,
+ repo_name,
+ len(commits),
+ bool(cursor),
+ )
+
+ stop_paging = False
+ for commit in commits:
+ if not isinstance(commit, dict):
+ continue
+ commit_time = parse_datetime((commit.get("author") or {}).get("time"))
+ if commit_time < since:
+ stop_paging = True
+ logger.info(
+ "git commit %s skipped because commit_time=%s is before since=%s",
+ commit.get("shortId") or commit.get("id"),
+ commit_time.isoformat(),
+ since.isoformat(),
+ )
+ continue
+
+ normalized = self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)
+ if normalized is not None:
+ logger.info(
+ "git commit accepted repo=%s shortId=%s author=%s email=%s",
+ repo_name,
+ commit.get("shortId"),
+ (commit.get("author") or {}).get("name"),
+ (commit.get("author") or {}).get("email"),
+ )
+ events.append(normalized)
+ else:
+ logger.info(
+ "git commit skipped repo=%s shortId=%s author=%s email=%s",
+ repo_name,
+ commit.get("shortId"),
+ (commit.get("author") or {}).get("name"),
+ (commit.get("author") or {}).get("email"),
+ )
+
+ if stop_paging or not cursor:
+ break
+
+ return events
+
+ def _normalize_commit(
+ self,
+ *,
+ actor: str,
+ repo_name: str,
+ commit: dict[str, Any],
+ ) -> NormalizedEvent | None:
+ author = commit.get("author") or {}
+ candidate_aliases = [
+ actor,
+ author.get("email", ""),
+ author.get("name", ""),
+ ]
+ matched_actor = None
+ for candidate in candidate_aliases:
+ canonical = self.identity_resolver.canonicalize(candidate)
+ if canonical == actor:
+ matched_actor = canonical
+ break
+
+ if matched_actor is None:
+ return None
+
+ commit_id = str(commit["id"])
+ commit_time = parse_datetime(author["time"])
+ return NormalizedEvent(
+ service=self.service_name,
+ event_type="commit",
+ actor=matched_actor,
+ repo_name=repo_name,
+ resource_id=commit_id,
+ external_uid=f"git:{repo_name}:{commit_id}",
+ occurred_at=commit_time,
+ weight=self.settings.event_weights["commit"],
+ raw_payload_json=commit,
+ )
diff --git a/src/srht_contrib/services/srht_client.py b/src/srht_contrib/services/srht_client.py
new file mode 100644
index 0000000..c8e2649
--- /dev/null
+++ b/src/srht_contrib/services/srht_client.py
@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+
+
+logger = logging.getLogger(__name__)
+
+
+class SourceHutClientError(RuntimeError):
+ """Raised when a SourceHut GraphQL request fails."""
+
+
+class SourceHutGraphQLClient:
+ def __init__(
+ self,
+ endpoint: str,
+ token: str,
+ *,
+ timeout: float = 15.0,
+ max_retries: int = 2,
+ transport: httpx.BaseTransport | None = None,
+ ) -> None:
+ self.endpoint = endpoint
+ self.timeout = timeout
+ self.max_retries = max_retries
+ headers = {
+ "Authorization": f"Bearer {token}",
+ "Content-Type": "application/json",
+ }
+ self._client = httpx.Client(headers=headers, timeout=timeout, transport=transport)
+
+ def execute(self, query: str, variables: dict[str, Any] | None = None) -> dict[str, Any]:
+ payload = {"query": query, "variables": variables or {}}
+ attempts = self.max_retries + 1
+
+ for attempt in range(1, attempts + 1):
+ try:
+ response = self._client.post(self.endpoint, json=payload)
+ response.raise_for_status()
+ body = response.json()
+ except httpx.HTTPStatusError as exc:
+ response_text = exc.response.text[:500]
+ logger.warning(
+ "SourceHut HTTP failure from %s on attempt %s/%s: %s %s",
+ self.endpoint,
+ attempt,
+ attempts,
+ exc.response.status_code,
+ response_text,
+ )
+ if exc.response.status_code >= 500 and attempt < attempts:
+ continue
+ raise SourceHutClientError(
+ f"HTTP error from SourceHut: {exc.response.status_code} {response_text}".strip()
+ ) from exc
+ except httpx.HTTPError as exc:
+ logger.warning(
+ "SourceHut network failure from %s on attempt %s/%s",
+ self.endpoint,
+ attempt,
+ attempts,
+ )
+ if attempt < attempts:
+ continue
+ raise SourceHutClientError("Network error while contacting SourceHut") from exc
+
+ if "errors" in body:
+ raise SourceHutClientError(f"GraphQL errors returned by SourceHut: {body['errors']}")
+ return body.get("data", {})
+
+ raise SourceHutClientError("SourceHut request exhausted retries")
+
+ def close(self) -> None:
+ self._client.close()
diff --git a/src/srht_contrib/services/todo.py b/src/srht_contrib/services/todo.py
new file mode 100644
index 0000000..2d25acf
--- /dev/null
+++ b/src/srht_contrib/services/todo.py
@@ -0,0 +1,559 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+import logging
+from typing import Any
+
+from srht_contrib.config import Settings
+from srht_contrib.schemas import NormalizedEvent
+from srht_contrib.services.srht_client import SourceHutGraphQLClient
+from srht_contrib.utils.dates import ensure_utc, parse_datetime
+
+
+logger = logging.getLogger(__name__)
+
+
+TODO_ACTIVITY_QUERY = """
+query TodoActivity($cursor: Cursor) {
+ me {
+ canonicalName
+ }
+ events(cursor: $cursor) {
+ results {
+ id
+ created
+ ticket {
+ id
+ ref
+ status
+ resolution
+ tracker {
+ name
+ }
+ }
+ changes {
+ __typename
+ eventType
+ ticket {
+ id
+ }
+ ... on Created {
+ author {
+ canonicalName
+ }
+ }
+ ... on Comment {
+ author {
+ canonicalName
+ }
+ }
+ ... on StatusChange {
+ editor {
+ canonicalName
+ }
+ oldStatus
+ newStatus
+ oldResolution
+ newResolution
+ }
+ }
+ }
+ cursor
+ }
+}
+""".strip()
+
+TODO_TRACKERS_QUERY = """
+query TodoTrackers($cursor: Cursor) {
+ me {
+ canonicalName
+ trackers(cursor: $cursor) {
+ results {
+ id
+ rid
+ name
+ }
+ cursor
+ }
+ }
+}
+""".strip()
+
+TODO_TRACKER_TICKETS_QUERY = """
+query TodoTrackerTickets($trackerRid: ID!, $cursor: Cursor) {
+ tracker(rid: $trackerRid) {
+ id
+ name
+ tickets(cursor: $cursor) {
+ results {
+ id
+ ref
+ created
+ updated
+ status
+ resolution
+ submitter {
+ canonicalName
+ }
+ }
+ cursor
+ }
+ }
+}
+""".strip()
+
+TODO_TICKET_EVENTS_QUERY = """
+query TodoTicketEvents($trackerRid: ID!, $ticketId: Int!, $cursor: Cursor) {
+ tracker(rid: $trackerRid) {
+ id
+ name
+ ticket(id: $ticketId) {
+ id
+ ref
+ status
+ resolution
+ events(cursor: $cursor) {
+ results {
+ id
+ created
+ changes {
+ __typename
+ eventType
+ ticket {
+ id
+ }
+ ... on Created {
+ author {
+ canonicalName
+ }
+ }
+ ... on Comment {
+ author {
+ canonicalName
+ }
+ }
+ ... on StatusChange {
+ editor {
+ canonicalName
+ }
+ oldStatus
+ newStatus
+ oldResolution
+ newResolution
+ }
+ }
+ }
+ cursor
+ }
+ }
+ }
+}
+""".strip()
+
+
+TICKET_CLOSED_STATUSES = {"RESOLVED"}
+TICKET_CLOSED_RESOLUTIONS = {
+ "CLOSED",
+ "FIXED",
+ "IMPLEMENTED",
+ "WONT_FIX",
+ "BY_DESIGN",
+ "INVALID",
+ "DUPLICATE",
+ "NOT_OUR_BUG",
+}
+
+
+class TodoSchemaError(RuntimeError):
+ """Raised when SourceHut returns an unexpected todo event shape."""
+
+
+def _safe_nested_name(entity: dict[str, Any] | None) -> str | None:
+ if not entity:
+ return None
+ return entity.get("canonicalName") or entity.get("name")
+
+
+def _repo_name_from_event(event: dict[str, Any]) -> str | None:
+ ticket = event.get("ticket") or {}
+ tracker = ticket.get("tracker") or {}
+ return tracker.get("name")
+
+
+def _resource_id_from_event(event: dict[str, Any]) -> str:
+ ticket = event.get("ticket") or {}
+ return str(ticket.get("ref") or ticket.get("id") or event["id"])
+
+
+def _change_ticket_id(change: dict[str, Any], event: dict[str, Any]) -> str:
+ ticket = change.get("ticket") or event.get("ticket") or {}
+ return str(ticket.get("id") or event["id"])
+
+
+def _normalize_event_change(
+ *,
+ settings: Settings,
+ actor: str,
+ event: dict[str, Any],
+ change: dict[str, Any],
+ occurred_at: datetime,
+) -> NormalizedEvent | None:
+ change_type = change.get("__typename")
+ event_id = str(event["id"])
+ resource_id = _resource_id_from_event(event)
+ repo_name = _repo_name_from_event(event)
+ ticket_id = _change_ticket_id(change, event)
+
+ if change_type == "Created" and _safe_nested_name(change.get("author")) == actor:
+ return NormalizedEvent(
+ service="todo",
+ event_type="ticket_created",
+ actor=actor,
+ repo_name=repo_name,
+ resource_id=resource_id,
+ external_uid=f"todo:event:{event_id}:created:{ticket_id}",
+ occurred_at=occurred_at,
+ weight=settings.event_weights["ticket_created"],
+ raw_payload_json={"event": event, "change": change},
+ )
+
+ if change_type == "Comment" and _safe_nested_name(change.get("author")) == actor:
+ return NormalizedEvent(
+ service="todo",
+ event_type="ticket_comment",
+ actor=actor,
+ repo_name=repo_name,
+ resource_id=resource_id,
+ external_uid=f"todo:event:{event_id}:comment:{ticket_id}",
+ occurred_at=occurred_at,
+ weight=settings.event_weights["ticket_comment"],
+ raw_payload_json={"event": event, "change": change},
+ )
+
+ if change_type == "StatusChange" and _safe_nested_name(change.get("editor")) == actor:
+ new_status = change.get("newStatus")
+ new_resolution = change.get("newResolution")
+ if new_status in TICKET_CLOSED_STATUSES or new_resolution in TICKET_CLOSED_RESOLUTIONS:
+ return NormalizedEvent(
+ service="todo",
+ event_type="ticket_closed",
+ actor=actor,
+ repo_name=repo_name,
+ resource_id=resource_id,
+ external_uid=f"todo:event:{event_id}:closed:{ticket_id}",
+ occurred_at=occurred_at,
+ weight=settings.event_weights["ticket_closed"],
+ raw_payload_json={"event": event, "change": change},
+ )
+
+ return None
+
+
+def _extract_event_cursor_page(data: dict[str, Any]) -> tuple[str | None, list[dict[str, Any]], str]:
+ me = data.get("me") or {}
+ canonical_actor = me.get("canonicalName")
+ if not canonical_actor:
+ raise TodoSchemaError("todo.sr.ht response did not include me.canonicalName")
+
+ events = data.get("events") or {}
+ results = events.get("results") or []
+ if not isinstance(results, list):
+ raise TodoSchemaError("todo.sr.ht response did not include events.results")
+
+ return events.get("cursor"), [event for event in results if isinstance(event, dict)], canonical_actor
+
+
+@dataclass(slots=True)
+class TodoPollResult:
+ events: list[NormalizedEvent]
+ cursor: str
+
+
+class TodoIngestionService:
+ """Fetches todo.sr.ht activity from the authenticated event feed and normalizes it."""
+
+ service_name = "todo"
+
+ def __init__(self, client: SourceHutGraphQLClient, settings: Settings) -> None:
+ self.client = client
+ self.settings = settings
+
+ def fetch_recent_events(self, actor: str, since: datetime | None = None) -> TodoPollResult:
+ since_dt = ensure_utc(since or (datetime.now(tz=UTC) - timedelta(days=30)))
+ cursor_time = datetime.now(tz=UTC).isoformat()
+ feed_result = self._fetch_from_activity_feed(actor=actor, since=since_dt)
+ if feed_result.events:
+ return TodoPollResult(events=feed_result.events, cursor=cursor_time)
+
+ logger.info(
+ "todo activity feed returned no normalized events for actor=%s; falling back to tracker crawl",
+ actor,
+ )
+ tracker_events = self._fetch_from_trackers(actor=actor, since=since_dt)
+ return TodoPollResult(events=tracker_events, cursor=cursor_time)
+
+ def _fetch_from_activity_feed(self, actor: str, since: datetime) -> TodoPollResult:
+ events: list[NormalizedEvent] = []
+ cursor: str | None = None
+ effective_actor = actor
+
+ for _ in range(10):
+ data = self.client.execute(TODO_ACTIVITY_QUERY, {"cursor": cursor})
+ cursor, page_events, canonical_actor = _extract_event_cursor_page(data)
+ effective_actor = actor or canonical_actor
+ logger.info(
+ "todo page fetched for actor=%s canonical_actor=%s events=%s next_cursor=%s since=%s",
+ actor,
+ canonical_actor,
+ len(page_events),
+ bool(cursor),
+ since.isoformat(),
+ )
+
+ stop_paging = False
+ for event in page_events:
+ occurred_at = parse_datetime(event["created"])
+ event_id = str(event.get("id"))
+ resource_id = _resource_id_from_event(event)
+ repo_name = _repo_name_from_event(event)
+ change_list = event.get("changes") or []
+ logger.info(
+ "todo event id=%s resource=%s repo=%s occurred_at=%s changes=%s",
+ event_id,
+ resource_id,
+ repo_name,
+ occurred_at.isoformat(),
+ len(change_list) if isinstance(change_list, list) else "unknown",
+ )
+ if occurred_at < since:
+ stop_paging = True
+ logger.info(
+ "todo event id=%s skipped because occurred_at=%s is before since=%s",
+ event_id,
+ occurred_at.isoformat(),
+ since.isoformat(),
+ )
+ continue
+
+ for change in change_list:
+ if not isinstance(change, dict):
+ logger.info("todo event id=%s skipped non-dict change payload", event_id)
+ continue
+ change_type = change.get("__typename")
+ author = _safe_nested_name(change.get("author"))
+ editor = _safe_nested_name(change.get("editor"))
+ logger.info(
+ "todo change event_id=%s type=%s eventType=%s author=%s editor=%s newStatus=%s newResolution=%s",
+ event_id,
+ change_type,
+ change.get("eventType"),
+ author,
+ editor,
+ change.get("newStatus"),
+ change.get("newResolution"),
+ )
+ normalized = _normalize_event_change(
+ settings=self.settings,
+ actor=effective_actor,
+ event=event,
+ change=change,
+ occurred_at=occurred_at,
+ )
+ if normalized is not None:
+ logger.info(
+ "todo change accepted event_id=%s normalized_type=%s external_uid=%s",
+ event_id,
+ normalized.event_type,
+ normalized.external_uid,
+ )
+ events.append(normalized)
+ else:
+ logger.info(
+ "todo change skipped event_id=%s for actor=%s",
+ event_id,
+ effective_actor,
+ )
+
+ if stop_paging or not cursor:
+ break
+
+ logger.info("todo poll complete for actor=%s normalized_events=%s", effective_actor, len(events))
+ return TodoPollResult(events=events, cursor=datetime.now(tz=UTC).isoformat())
+
+ def _fetch_from_trackers(self, actor: str, since: datetime) -> list[NormalizedEvent]:
+ data = self.client.execute(TODO_TRACKERS_QUERY, {"cursor": None})
+ me = data.get("me") or {}
+ canonical_actor = me.get("canonicalName") or actor
+ trackers = ((me.get("trackers") or {}).get("results")) or []
+ logger.info("todo tracker crawl actor=%s trackers=%s", canonical_actor, len(trackers))
+
+ events: list[NormalizedEvent] = []
+ for tracker in trackers:
+ if not isinstance(tracker, dict):
+ continue
+ tracker_id = tracker.get("id")
+ tracker_rid = tracker.get("rid")
+ tracker_name = tracker.get("name")
+ tracker_events = self._fetch_tracker_tickets(
+ actor=canonical_actor,
+ tracker_id=str(tracker_id),
+ tracker_rid=str(tracker_rid),
+ tracker_name=tracker_name,
+ since=since,
+ )
+ events.extend(tracker_events)
+
+ logger.info("todo tracker crawl complete actor=%s normalized_events=%s", canonical_actor, len(events))
+ return events
+
+ def _fetch_tracker_tickets(
+ self,
+ *,
+ actor: str,
+ tracker_id: str,
+ tracker_rid: str,
+ tracker_name: str | None,
+ since: datetime,
+ ) -> list[NormalizedEvent]:
+ events: list[NormalizedEvent] = []
+ cursor: str | None = None
+
+ for _ in range(50):
+ data = self.client.execute(
+ TODO_TRACKER_TICKETS_QUERY,
+ {"trackerRid": tracker_rid, "cursor": cursor},
+ )
+ tracker = data.get("tracker") or {}
+ tickets_page = tracker.get("tickets") or {}
+ tickets = tickets_page.get("results") or []
+ cursor = tickets_page.get("cursor")
+ logger.info(
+ "todo tracker=%s ticket page count=%s next_cursor=%s",
+ tracker_name or tracker_id,
+ len(tickets),
+ bool(cursor),
+ )
+
+ stop_paging = False
+ for ticket in tickets:
+ if not isinstance(ticket, dict):
+ continue
+ updated_at = parse_datetime(ticket["updated"])
+ if updated_at < since:
+ stop_paging = True
+ logger.info(
+ "todo ticket ref=%s skipped because updated_at=%s is before since=%s",
+ ticket.get("ref"),
+ updated_at.isoformat(),
+ since.isoformat(),
+ )
+ continue
+ events.extend(
+ self._fetch_ticket_events(
+ actor=actor,
+ tracker_id=tracker_id,
+ tracker_rid=tracker_rid,
+ tracker_name=tracker_name,
+ ticket=ticket,
+ since=since,
+ )
+ )
+
+ if stop_paging or not cursor:
+ break
+
+ return events
+
+ def _fetch_ticket_events(
+ self,
+ *,
+ actor: str,
+ tracker_id: str,
+ tracker_rid: str,
+ tracker_name: str | None,
+ ticket: dict[str, Any],
+ since: datetime,
+ ) -> list[NormalizedEvent]:
+ events: list[NormalizedEvent] = []
+ cursor: str | None = None
+ ticket_id = int(ticket["id"])
+ ticket_ref = str(ticket.get("ref") or ticket_id)
+
+ for _ in range(50):
+ data = self.client.execute(
+ TODO_TICKET_EVENTS_QUERY,
+ {"trackerRid": tracker_rid, "ticketId": ticket_id, "cursor": cursor},
+ )
+ tracker = data.get("tracker") or {}
+ ticket_payload = (tracker.get("ticket") or {}) if isinstance(tracker, dict) else {}
+ event_page = ticket_payload.get("events") or {}
+ page_events = event_page.get("results") or []
+ cursor = event_page.get("cursor")
+ logger.info(
+ "todo ticket events ref=%s tracker=%s count=%s next_cursor=%s",
+ ticket_ref,
+ tracker_name or tracker_id,
+ len(page_events),
+ bool(cursor),
+ )
+
+ stop_paging = False
+ for event in page_events:
+ if not isinstance(event, dict):
+ continue
+ event["ticket"] = {
+ "id": ticket_payload.get("id", ticket.get("id")),
+ "ref": ticket_payload.get("ref", ticket_ref),
+ "status": ticket_payload.get("status", ticket.get("status")),
+ "resolution": ticket_payload.get("resolution", ticket.get("resolution")),
+ "tracker": {"name": tracker_name},
+ }
+ occurred_at = parse_datetime(event["created"])
+ event_id = str(event.get("id"))
+ change_list = event.get("changes") or []
+ logger.info(
+ "todo ticket event ref=%s id=%s occurred_at=%s changes=%s",
+ ticket_ref,
+ event_id,
+ occurred_at.isoformat(),
+ len(change_list) if isinstance(change_list, list) else "unknown",
+ )
+ if occurred_at < since:
+ stop_paging = True
+ continue
+
+ for change in change_list:
+ if not isinstance(change, dict):
+ continue
+ logger.info(
+ "todo ticket change ref=%s event_id=%s type=%s eventType=%s author=%s editor=%s newStatus=%s newResolution=%s",
+ ticket_ref,
+ event_id,
+ change.get("__typename"),
+ change.get("eventType"),
+ _safe_nested_name(change.get("author")),
+ _safe_nested_name(change.get("editor")),
+ change.get("newStatus"),
+ change.get("newResolution"),
+ )
+ normalized = _normalize_event_change(
+ settings=self.settings,
+ actor=actor,
+ event=event,
+ change=change,
+ occurred_at=occurred_at,
+ )
+ if normalized is not None:
+ logger.info(
+ "todo ticket change accepted ref=%s event_id=%s normalized_type=%s",
+ ticket_ref,
+ event_id,
+ normalized.event_type,
+ )
+ events.append(normalized)
+
+ if stop_paging or not cursor:
+ break
+
+ return events
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-04 12:50:25 +0000