diff options
Diffstat (limited to 'src/srht_contrib')
23 files changed, 1769 insertions, 0 deletions
diff --git a/src/srht_contrib/__init__.py b/src/srht_contrib/__init__.py new file mode 100644 index 0000000..9069785 --- /dev/null +++ b/src/srht_contrib/__init__.py @@ -0,0 +1,5 @@ +"""srht_contrib package.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/src/srht_contrib/api/__init__.py b/src/srht_contrib/api/__init__.py new file mode 100644 index 0000000..1ce04c3 --- /dev/null +++ b/src/srht_contrib/api/__init__.py @@ -0,0 +1 @@ +"""API routes.""" diff --git a/src/srht_contrib/api/dependencies.py b/src/srht_contrib/api/dependencies.py new file mode 100644 index 0000000..8449380 --- /dev/null +++ b/src/srht_contrib/api/dependencies.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from fastapi import Depends, Header, HTTPException, Request, status +from sqlalchemy.orm import Session + +from srht_contrib.config import Settings +from srht_contrib.db import get_db_session +from srht_contrib.jobs.poller import PollerService +from srht_contrib.utils.identity import ActorIdentityResolver + + +def get_db(request: Request): + yield from get_db_session(request) + + +def get_settings(request: Request) -> Settings: + settings = getattr(request.app.state, "settings", None) + if settings is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Settings not configured.") + return settings + + +def get_actor_identity_resolver(request: Request) -> ActorIdentityResolver: + resolver = getattr(request.app.state, "actor_identity_resolver", None) + if resolver is None: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Actor identity resolver not configured.", + ) + return resolver + + +def get_poller(request: Request) -> PollerService: + poller = getattr(request.app.state, "poller", None) + if poller is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Poller not configured.") + return poller + + +def require_api_key( + x_api_key: str | None = Header(default=None, alias="X-API-Key"), + settings: Settings = Depends(get_settings), +) -> None: + if not settings.api_key or x_api_key != settings.api_key: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key.") diff --git a/src/srht_contrib/api/routes_contributions.py b/src/srht_contrib/api/routes_contributions.py new file mode 100644 index 0000000..0682aa6 --- /dev/null +++ b/src/srht_contrib/api/routes_contributions.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from datetime import date + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session + +from srht_contrib.api.dependencies import get_actor_identity_resolver, get_db, get_poller, require_api_key +from srht_contrib.jobs.poller import PollerService +from srht_contrib.schemas import ContributionCalendarResponse, ContributionStatsResponse, PollResponse +from srht_contrib.services.aggregator import ContributionAggregator +from srht_contrib.services.srht_client import SourceHutClientError +from srht_contrib.utils.dates import parse_date, year_bounds +from srht_contrib.utils.identity import ActorIdentityResolver + +router = APIRouter(prefix="/api/contributions", tags=["contributions"], dependencies=[Depends(require_api_key)]) + + +def _resolve_range(year: int | None, from_date: str | None, to_date: str | None) -> tuple[date, date]: + if year is not None and (from_date or to_date): + raise HTTPException(status_code=400, detail="Use either `year` or `from`/`to`, not both.") + + try: + if year is not None: + return year_bounds(year) + if from_date and to_date: + start = parse_date(from_date) + end = parse_date(to_date) + if start > end: + raise HTTPException(status_code=400, detail="`from` must be on or before `to`.") + return start, end + except ValueError as exc: + raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD.") from exc + + raise HTTPException(status_code=400, detail="Provide `year` or both `from` and `to`.") + + [email protected]("/{actor}", response_model=ContributionCalendarResponse) +def get_contributions( + actor: str, + year: int | None = Query(default=None, ge=1970, le=3000), + from_date: str | None = Query(default=None, alias="from"), + to_date: str | None = Query(default=None, alias="to"), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> ContributionCalendarResponse: + start, end = _resolve_range(year, from_date, to_date) + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + return ContributionAggregator().build_calendar(db, canonical_actor, start, end) + + [email protected]("/{actor}/stats", response_model=ContributionStatsResponse) +def get_contribution_stats( + actor: str, + year: int | None = Query(default=None, ge=1970, le=3000), + from_date: str | None = Query(default=None, alias="from"), + to_date: str | None = Query(default=None, alias="to"), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> ContributionStatsResponse: + start, end = _resolve_range(year, from_date, to_date) + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + return ContributionAggregator().build_stats(db, canonical_actor, start, end) + + [email protected]("/poll", response_model=PollResponse) +def trigger_manual_poll( + actor: str, + poller: PollerService = Depends(get_poller), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> PollResponse: + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + try: + inserted = poller.poll_all(db, canonical_actor) + except SourceHutClientError as exc: + raise HTTPException(status_code=502, detail=f"SourceHut polling failed: {exc}") from exc + return PollResponse(actor=canonical_actor, inserted_events=inserted, services=["todo", "git"]) diff --git a/src/srht_contrib/api/routes_health.py b/src/srht_contrib/api/routes_health.py new file mode 100644 index 0000000..32db759 --- /dev/null +++ b/src/srht_contrib/api/routes_health.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +from fastapi import APIRouter + +from srht_contrib.schemas import HealthResponse + +router = APIRouter(tags=["health"]) + + [email protected]("/health", response_model=HealthResponse) +def health() -> HealthResponse: + return HealthResponse(status="ok") diff --git a/src/srht_contrib/api/routes_repositories.py b/src/srht_contrib/api/routes_repositories.py new file mode 100644 index 0000000..93bc098 --- /dev/null +++ b/src/srht_contrib/api/routes_repositories.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from srht_contrib.api.dependencies import get_actor_identity_resolver, get_db, require_api_key +from srht_contrib.models import TrackedRepository +from srht_contrib.schemas import ( + TrackedRepositoryCreateRequest, + TrackedRepositoryResponse, + TrackedRepositoryUpdateRequest, +) +from srht_contrib.utils.identity import ActorIdentityResolver +from srht_contrib.utils.repositories import canonicalize_repository_name + +router = APIRouter(prefix="/api/repositories", tags=["repositories"], dependencies=[Depends(require_api_key)]) + + +def _to_response(repository: TrackedRepository) -> TrackedRepositoryResponse: + return TrackedRepositoryResponse( + id=repository.id, + service=repository.service, + actor=repository.actor, + repo_name=repository.repo_name, + ) + + +def _get_repository_or_404(db: Session, repository_id: int) -> TrackedRepository: + repository = db.scalar(select(TrackedRepository).where(TrackedRepository.id == repository_id)) + if repository is None: + raise HTTPException(status_code=404, detail="Tracked repository not found.") + return repository + + [email protected]("", response_model=list[TrackedRepositoryResponse]) +def list_tracked_repositories( + actor: str | None = Query(default=None), + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> list[TrackedRepositoryResponse]: + stmt = select(TrackedRepository).where(TrackedRepository.service == "git").order_by(TrackedRepository.repo_name) + if actor: + canonical_actor = actor_identity_resolver.canonicalize(actor, db=db) + stmt = stmt.where(TrackedRepository.actor == canonical_actor) + + repositories = db.scalars(stmt).all() + return [_to_response(repository) for repository in repositories] + + [email protected]("/{repository_id}", response_model=TrackedRepositoryResponse) +def get_tracked_repository( + repository_id: int, + db: Session = Depends(get_db), +) -> TrackedRepositoryResponse: + return _to_response(_get_repository_or_404(db, repository_id)) + + [email protected]("", response_model=TrackedRepositoryResponse, status_code=status.HTTP_201_CREATED) +def create_tracked_repository( + payload: TrackedRepositoryCreateRequest, + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> TrackedRepositoryResponse: + canonical_actor = actor_identity_resolver.canonicalize(payload.actor, db=db) + repository = TrackedRepository( + service="git", + actor=canonical_actor, + repo_name=canonicalize_repository_name(canonical_actor, payload.repo_name), + ) + db.add(repository) + try: + db.commit() + except IntegrityError as exc: + db.rollback() + raise HTTPException(status_code=409, detail="Tracked repository already exists.") from exc + + db.refresh(repository) + return _to_response(repository) + + [email protected]("/{repository_id}", response_model=TrackedRepositoryResponse) +def update_tracked_repository( + repository_id: int, + payload: TrackedRepositoryUpdateRequest, + db: Session = Depends(get_db), + actor_identity_resolver: ActorIdentityResolver = Depends(get_actor_identity_resolver), +) -> TrackedRepositoryResponse: + repository = _get_repository_or_404(db, repository_id) + + if payload.actor is not None: + repository.actor = actor_identity_resolver.canonicalize(payload.actor, db=db) + + if payload.repo_name is not None: + repository.repo_name = canonicalize_repository_name(repository.actor, payload.repo_name) + + db.add(repository) + try: + db.commit() + except IntegrityError as exc: + db.rollback() + raise HTTPException(status_code=409, detail="Tracked repository already exists.") from exc + + db.refresh(repository) + return _to_response(repository) + + [email protected]("/{repository_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_tracked_repository( + repository_id: int, + db: Session = Depends(get_db), +) -> Response: + repository = _get_repository_or_404(db, repository_id) + db.delete(repository) + db.commit() + return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/src/srht_contrib/config.py b/src/srht_contrib/config.py new file mode 100644 index 0000000..56da6ff --- /dev/null +++ b/src/srht_contrib/config.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from functools import lru_cache + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + case_sensitive=True, + ) + + app_name: str = "srht-contrib" + app_env: str = "development" + api_host: str = "127.0.0.1" + api_port: int = 8000 + enable_scheduler: bool = Field(default=False, alias="ENABLE_SCHEDULER") + api_key: str = Field(default="", alias="API_KEY") + + srht_token: str = Field(default="", alias="SRHT_TOKEN") + todo_srht_endpoint: str = Field( + default="https://todo.sr.ht/query", + alias="TODO_SRHT_ENDPOINT", + ) + git_srht_endpoint: str = Field( + default="https://git.sr.ht/query", + alias="GIT_SRHT_ENDPOINT", + ) + database_url: str = Field( + default="sqlite:///./srht_contrib.db", + alias="DATABASE_URL", + ) + default_actor: str = Field(default="~unknown", alias="DEFAULT_ACTOR") + poll_interval_seconds: int = Field(default=900, alias="POLL_INTERVAL_SECONDS") + actor_aliases_json: dict[str, list[str]] = Field( + default_factory=dict, + alias="ACTOR_ALIASES_JSON", + ) + git_tracked_repositories: list[str] = Field( + default_factory=list, + alias="GIT_TRACKED_REPOSITORIES", + ) + + event_weights: dict[str, float] = { + "commit": 1.0, + "ticket_created": 1.0, + "ticket_comment": 0.5, + "ticket_closed": 0.75, + "build_started": 0.25, + "build_passed": 0.25, + } + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + return Settings() diff --git a/src/srht_contrib/db.py b/src/srht_contrib/db.py new file mode 100644 index 0000000..d96de57 --- /dev/null +++ b/src/srht_contrib/db.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from collections.abc import Generator + +from fastapi import HTTPException, Request, status +from sqlalchemy import Engine, create_engine, text +from sqlalchemy.pool import StaticPool +from sqlalchemy.orm import Session, declarative_base, sessionmaker + +from srht_contrib.config import Settings + +Base = declarative_base() + + +def make_engine(settings: Settings) -> Engine: + connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {} + engine_kwargs = {"future": True, "connect_args": connect_args} + if settings.database_url in {"sqlite://", "sqlite:///:memory:"}: + engine_kwargs["poolclass"] = StaticPool + return create_engine(settings.database_url, **engine_kwargs) + + +def make_session_factory(settings: Settings) -> sessionmaker[Session]: + engine = make_engine(settings) + return sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False) + + +def validate_db(bind: Engine) -> None: + with bind.connect() as connection: + connection.execute(text("SELECT 1")) + + +def get_db() -> Generator[Session, None, None]: + raise RuntimeError("Use get_db(request) dependency injection with a Request parameter.") + + +def get_session_factory(request: Request) -> sessionmaker[Session]: + session_factory = getattr(request.app.state, "session_factory", None) + if session_factory is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database not configured.") + return session_factory + + +def get_db_session(request: Request) -> Generator[Session, None, None]: + session_factory = get_session_factory(request) + db = session_factory() + try: + yield db + finally: + db.close() diff --git a/src/srht_contrib/jobs/__init__.py b/src/srht_contrib/jobs/__init__.py new file mode 100644 index 0000000..75ca0d0 --- /dev/null +++ b/src/srht_contrib/jobs/__init__.py @@ -0,0 +1 @@ +"""Background jobs.""" diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py new file mode 100644 index 0000000..c3ed069 --- /dev/null +++ b/src/srht_contrib/jobs/poller.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from srht_contrib.models import ContributionEvent, SyncState, TrackedRepository +from srht_contrib.schemas import NormalizedEvent +from srht_contrib.services.git import GitIngestionService +from srht_contrib.services.todo import TodoIngestionService +from srht_contrib.utils.repositories import canonicalize_repository_name + + +logger = logging.getLogger(__name__) +SYNC_OVERLAP = timedelta(hours=24) + + +class PollerService: + def __init__(self, todo_service: TodoIngestionService, git_service: GitIngestionService) -> None: + self.todo_service = todo_service + self.git_service = git_service + + def poll_all(self, db: Session, actor: str) -> int: + inserted = 0 + inserted += self._poll_service(db, actor, self.todo_service.service_name, self.todo_service.fetch_recent_events) + self._sync_tracked_repositories(db, actor) + git_repositories = self._tracked_repositories_for_actor(db, actor) + inserted += self._poll_service( + db, + actor, + self.git_service.service_name, + lambda actor, since: self.git_service.fetch_recent_events( + actor=actor, + since=since, + repositories=git_repositories, + ), + ) + db.commit() + return inserted + + def _poll_service(self, db: Session, actor: str, service_name: str, fetcher) -> int: + state = db.scalar( + select(SyncState).where(SyncState.service == service_name).where(SyncState.actor == actor) + ) + since = datetime.now(tz=UTC) - timedelta(days=30) + if state and state.cursor_value: + since = datetime.fromisoformat(state.cursor_value.replace("Z", "+00:00")).astimezone(UTC) - SYNC_OVERLAP + logger.info( + "Using sync cursor for %s actor=%s with overlap; since=%s", + service_name, + actor, + since.isoformat(), + ) + + result = fetcher(actor=actor, since=since) + inserted = self._insert_events(db, result.events) + self._upsert_sync_state(db, service_name, actor, result.cursor) + logger.info("Polled %s for %s: inserted=%s", service_name, actor, inserted) + return inserted + + @staticmethod + def _insert_events(db: Session, events: list[NormalizedEvent]) -> int: + inserted = 0 + for event in events: + try: + with db.begin_nested(): + model = ContributionEvent(**event.model_dump()) + db.add(model) + db.flush() + inserted += 1 + except IntegrityError: + logger.info("Skipping duplicate event %s for service %s", event.external_uid, event.service) + return inserted + + @staticmethod + def _upsert_sync_state(db: Session, service: str, actor: str, cursor_value: str) -> None: + state = db.scalar(select(SyncState).where(SyncState.service == service).where(SyncState.actor == actor)) + now = datetime.now(tz=UTC) + if state is None: + db.add(SyncState(service=service, actor=actor, cursor_value=cursor_value, updated_at=now)) + db.flush() + return + + state.cursor_value = cursor_value + state.updated_at = now + db.add(state) + db.flush() + + def _sync_tracked_repositories(self, db: Session, actor: str) -> None: + configured = self.git_service.settings.git_tracked_repositories + for repo_name in configured: + canonical_repo_name = canonicalize_repository_name(actor, repo_name) + existing = db.scalar( + select(TrackedRepository) + .where(TrackedRepository.service == self.git_service.service_name) + .where(TrackedRepository.actor == actor) + .where(TrackedRepository.repo_name == canonical_repo_name) + ) + if existing is None: + db.add( + TrackedRepository( + service=self.git_service.service_name, + repo_name=canonical_repo_name, + actor=actor, + ) + ) + db.flush() + + def _tracked_repositories_for_actor(self, db: Session, actor: str) -> list[str]: + rows = db.scalars( + select(TrackedRepository.repo_name) + .where(TrackedRepository.service == self.git_service.service_name) + .where(TrackedRepository.actor == actor) + .order_by(TrackedRepository.repo_name) + ).all() + return list(rows) diff --git a/src/srht_contrib/logging.py b/src/srht_contrib/logging.py new file mode 100644 index 0000000..315bb89 --- /dev/null +++ b/src/srht_contrib/logging.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +import logging + + +def configure_logging() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s [%(name)s] %(message)s", + ) diff --git a/src/srht_contrib/main.py b/src/srht_contrib/main.py new file mode 100644 index 0000000..37c3492 --- /dev/null +++ b/src/srht_contrib/main.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from apscheduler.schedulers.background import BackgroundScheduler +from fastapi import FastAPI +from sqlalchemy import Engine +from sqlalchemy.orm import Session, sessionmaker + +from srht_contrib.api.routes_contributions import router as contributions_router +from srht_contrib.api.routes_health import router as health_router +from srht_contrib.api.routes_repositories import router as repositories_router +from srht_contrib.config import Settings, get_settings +from srht_contrib.db import make_engine, validate_db +from srht_contrib.jobs.poller import PollerService +from srht_contrib.logging import configure_logging +from srht_contrib.services.git import GitIngestionService +from srht_contrib.services.srht_client import SourceHutGraphQLClient +from srht_contrib.services.todo import TodoIngestionService +from srht_contrib.utils.identity import ActorIdentityResolver + + +def build_poller(settings: Settings) -> PollerService: + todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token) + git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token) + todo_service = TodoIngestionService(todo_client, settings) + git_service = GitIngestionService(git_client, settings) + return PollerService(todo_service=todo_service, git_service=git_service) + + +def create_app( + settings: Settings | None = None, + *, + engine: Engine | None = None, + session_factory: sessionmaker[Session] | None = None, + poller: PollerService | None = None, +) -> FastAPI: + app_settings = settings or get_settings() + configure_logging() + app_engine = engine or make_engine(app_settings) + app_session_factory = session_factory or sessionmaker( + bind=app_engine, + autoflush=False, + autocommit=False, + expire_on_commit=False, + ) + actor_identity_resolver = ActorIdentityResolver(app_settings.actor_aliases_json) + + @asynccontextmanager + async def lifespan(app: FastAPI) -> AsyncIterator[None]: + validate_db(app_engine) + app_poller = poller or build_poller(app_settings) + app.state.poller = app_poller + app.state.settings = app_settings + app.state.engine = app_engine + app.state.session_factory = app_session_factory + app.state.actor_identity_resolver = actor_identity_resolver + scheduler: BackgroundScheduler | None = None + if app_settings.enable_scheduler: + scheduler = BackgroundScheduler() + scheduler.add_job( + _scheduled_poll, + "interval", + seconds=app_settings.poll_interval_seconds, + args=[app], + id="srht-poller", + replace_existing=True, + ) + scheduler.start() + app.state.scheduler = scheduler + try: + yield + finally: + if scheduler is not None: + scheduler.shutdown(wait=False) + app_poller.todo_service.client.close() + app_poller.git_service.client.close() + + app = FastAPI(title=app_settings.app_name, lifespan=lifespan) + + app.include_router(health_router) + app.include_router(contributions_router) + app.include_router(repositories_router) + return app + + +def _scheduled_poll(app: FastAPI) -> None: + settings: Settings = app.state.settings + poller: PollerService = app.state.poller + session_factory: sessionmaker[Session] = app.state.session_factory + db = session_factory() + try: + poller.poll_all(db, settings.default_actor) + finally: + db.close() + + +app = create_app() diff --git a/src/srht_contrib/models.py b/src/srht_contrib/models.py new file mode 100644 index 0000000..b87f464 --- /dev/null +++ b/src/srht_contrib/models.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import JSON, DateTime, Float, Index, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from srht_contrib.db import Base + + +class ContributionEvent(Base): + __tablename__ = "contribution_events" + __table_args__ = ( + UniqueConstraint("service", "external_uid", name="uq_contribution_event_service_uid"), + Index("ix_contribution_events_actor_occurred_at", "actor", "occurred_at"), + Index("ix_contribution_events_service_occurred_at", "service", "occurred_at"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(32), nullable=False) + event_type: Mapped[str] = mapped_column(String(64), nullable=False) + actor: Mapped[str] = mapped_column(String(255), nullable=False) + repo_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + resource_id: Mapped[str] = mapped_column(String(255), nullable=False) + external_uid: Mapped[str] = mapped_column(String(255), nullable=False) + occurred_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + weight: Mapped[float] = mapped_column(Float, nullable=False) + raw_payload_json: Mapped[dict | None] = mapped_column(JSON, nullable=True) + + +class SyncState(Base): + __tablename__ = "sync_states" + __table_args__ = (UniqueConstraint("service", "actor", name="uq_sync_state_service_actor"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(32), nullable=False) + actor: Mapped[str] = mapped_column(String(255), nullable=False) + cursor_value: Mapped[str | None] = mapped_column(Text, nullable=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + +class TrackedRepository(Base): + __tablename__ = "tracked_repositories" + __table_args__ = ( + UniqueConstraint("service", "actor", "repo_name", name="uq_tracked_repository_service_actor_name"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + service: Mapped[str] = mapped_column(String(32), nullable=False) + repo_name: Mapped[str] = mapped_column(String(255), nullable=False) + actor: Mapped[str] = mapped_column(String(255), nullable=False) + + +class ActorAlias(Base): + __tablename__ = "actor_aliases" + __table_args__ = (UniqueConstraint("alias", name="uq_actor_alias_alias"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + canonical_actor: Mapped[str] = mapped_column(String(255), nullable=False) + alias: Mapped[str] = mapped_column(String(255), nullable=False) diff --git a/src/srht_contrib/schemas.py b/src/srht_contrib/schemas.py new file mode 100644 index 0000000..2a00996 --- /dev/null +++ b/src/srht_contrib/schemas.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from datetime import date, datetime + +from pydantic import BaseModel, Field, model_validator + + +class NormalizedEvent(BaseModel): + service: str + event_type: str + actor: str + repo_name: str | None = None + resource_id: str + external_uid: str + occurred_at: datetime + weight: float + raw_payload_json: dict | None = None + + +class ContributionDay(BaseModel): + date: date + count: int + score: float + + +class ContributionCalendarResponse(BaseModel): + actor: str + from_date: date = Field(alias="from") + to_date: date = Field(alias="to") + days: list[ContributionDay] + + model_config = {"populate_by_name": True} + + +class ContributionStatsResponse(BaseModel): + actor: str + from_date: date = Field(alias="from") + to_date: date = Field(alias="to") + total_events: int + total_score: float + active_days: int + longest_streak: int + current_streak: int + + model_config = {"populate_by_name": True} + + +class HealthResponse(BaseModel): + status: str + + +class PollResponse(BaseModel): + actor: str + inserted_events: int + services: list[str] + + +class TrackedRepositoryCreateRequest(BaseModel): + actor: str + repo_name: str + + +class TrackedRepositoryUpdateRequest(BaseModel): + actor: str | None = None + repo_name: str | None = None + + @model_validator(mode="after") + def validate_any_field_present(self) -> "TrackedRepositoryUpdateRequest": + if self.actor is None and self.repo_name is None: + raise ValueError("Provide `actor`, `repo_name`, or both.") + return self + + +class TrackedRepositoryResponse(BaseModel): + id: int + service: str + actor: str + repo_name: str 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 diff --git a/src/srht_contrib/utils/__init__.py b/src/srht_contrib/utils/__init__.py new file mode 100644 index 0000000..e9b9a78 --- /dev/null +++ b/src/srht_contrib/utils/__init__.py @@ -0,0 +1 @@ +"""Utility helpers.""" diff --git a/src/srht_contrib/utils/dates.py b/src/srht_contrib/utils/dates.py new file mode 100644 index 0000000..ad72968 --- /dev/null +++ b/src/srht_contrib/utils/dates.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from datetime import UTC, date, datetime, time, timedelta + + +def ensure_utc(dt: datetime) -> datetime: + if dt.tzinfo is None: + return dt.replace(tzinfo=UTC) + return dt.astimezone(UTC) + + +def parse_date(value: str) -> date: + return date.fromisoformat(value) + + +def parse_datetime(value: str) -> datetime: + normalized = value.replace("Z", "+00:00") + return ensure_utc(datetime.fromisoformat(normalized)) + + +def date_range(start: date, end: date) -> list[date]: + days: list[date] = [] + current = start + while current <= end: + days.append(current) + current += timedelta(days=1) + return days + + +def year_bounds(year: int) -> tuple[date, date]: + return date(year, 1, 1), date(year, 12, 31) + + +def date_to_utc_bounds(value: date) -> tuple[datetime, datetime]: + start = datetime.combine(value, time.min, tzinfo=UTC) + end = datetime.combine(value, time.max, tzinfo=UTC) + return start, end diff --git a/src/srht_contrib/utils/identity.py b/src/srht_contrib/utils/identity.py new file mode 100644 index 0000000..adc9a9d --- /dev/null +++ b/src/srht_contrib/utils/identity.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from srht_contrib.models import ActorAlias + + +class ActorIdentityResolver: + def __init__(self, configured_aliases: dict[str, list[str]] | None = None) -> None: + self.configured_aliases = configured_aliases or {} + + def canonicalize(self, actor: str, db: Session | None = None) -> str: + normalized = actor.strip() + if not normalized: + return normalized + + for canonical, aliases in self.configured_aliases.items(): + if normalized == canonical or normalized in aliases: + return canonical + + if db is not None: + alias = db.scalar(select(ActorAlias).where(ActorAlias.alias == normalized)) + if alias is not None: + return alias.canonical_actor + + return normalized diff --git a/src/srht_contrib/utils/repositories.py b/src/srht_contrib/utils/repositories.py new file mode 100644 index 0000000..f8e65b5 --- /dev/null +++ b/src/srht_contrib/utils/repositories.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from fastapi import HTTPException, status + + +def canonicalize_repository_name(actor: str, repo_name: str) -> str: + normalized_actor = actor.strip() + normalized_repo_name = repo_name.strip() + + if not normalized_actor: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Actor must not be blank.") + if not normalized_repo_name: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Repository name must not be blank.") + + if "/" in normalized_repo_name: + owner, name = normalized_repo_name.split("/", 1) + owner = owner.strip() + name = name.strip() + if not owner or not name or "/" in name: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail="Repository name must be `name` or `~owner/name`.", + ) + canonical_owner = owner if owner.startswith("~") else f"~{owner}" + return f"{canonical_owner}/{name}" + + if "/" in normalized_actor: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Actor must be a canonical sr.ht user.") + + canonical_actor = normalized_actor if normalized_actor.startswith("~") else f"~{normalized_actor}" + return f"{canonical_actor}/{normalized_repo_name}" |