UsageKiosk/app/providers/devin.py

import math
import time
from datetime import datetime
from typing import Any
from urllib.parse import quote
from zoneinfo import ZoneInfo

import httpx
from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt, StrictStr, ValidationError

from app.config import Settings
from app.models import ProviderUsage, UsageLimit
from app.providers.base import GENERIC_ERROR, UsageProvider, utc_now


class Cycle(BaseModel):
    model_config = ConfigDict(extra='ignore')
    after: StrictInt
    before: StrictInt


class CyclesResponse(BaseModel):
    model_config = ConfigDict(extra='ignore')
    items: list[Cycle]
    has_next_page: StrictBool = False
    end_cursor: StrictStr | None = None


class DailyResponse(BaseModel):
    model_config = ConfigDict(extra='ignore')
    total_acus: Any
    consumption_by_date: list[Any] = Field(default_factory=list)


class OrgResponse(BaseModel):
    model_config = ConfigDict(extra='ignore')
    max_cycle_acu_limit: Any = None


class SessionSummary(BaseModel):
    model_config = ConfigDict(extra='ignore')
    session_id: StrictStr
    created_at: StrictInt
    acus_consumed: Any


class SessionsResponse(BaseModel):
    model_config = ConfigDict(extra='ignore')
    items: list[SessionSummary]
    has_next_page: StrictBool = False
    end_cursor: StrictStr | None = None


def _acu(value: Any) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError
    number = float(value)
    if not math.isfinite(number) or number < 0:
        raise ValueError
    return number


class DevinProvider(UsageProvider):
    def __init__(self, settings: Settings, client: httpx.AsyncClient | None = None):
        source = 'Devin Organization Sessions API' if settings.devin_mode == 'organization_sessions' else 'Devin API v3'
        super().__init__('devin', settings.provider_cache_ttl, source)
        self.settings = settings
        self._client = client or httpx.AsyncClient(base_url=settings.devin_api_url.rstrip('/'), timeout=settings.devin_timeout,
                                                   headers={'Authorization': f'Bearer {settings.devin_api_key.get_secret_value()}'})
        self._owns_client = client is None

    async def _fetch_usage(self) -> ProviderUsage:
        if self.settings.devin_mode == 'organization_sessions':
            return await self._fetch_organization_sessions()
        return await self._fetch_enterprise_usage()

    async def _fetch_enterprise_usage(self) -> ProviderUsage:
        now = int(time.time())
        params = {'first': 100}
        seen_cursors: set[str] = set()
        cycle = None
        for page in range(100):
            cycles_response = await self._client.get('/v3/enterprise/consumption/cycles', params=params)
            cycles_response.raise_for_status()
            try:
                cycles_page = CyclesResponse.model_validate(cycles_response.json())
            except (ValidationError, ValueError, TypeError):
                raise ValueError
            cycle = next((item for item in cycles_page.items if item.after <= now < item.before), None)
            if cycle is not None:
                break
            if not cycles_page.has_next_page:
                break
            cursor = cycles_page.end_cursor
            if not cursor or cursor in seen_cursors:
                raise ValueError
            seen_cursors.add(cursor)
            params = {'first': 100, 'after': cursor}
        else:
            raise ValueError
        if cycle is None:
            raise ValueError
        user_id = self.settings.devin_user_id.strip()
        org_id = self.settings.devin_org_id.strip()
        if user_id:
            path = f'/v3/enterprise/consumption/daily/users/{quote(user_id, safe="")}'
            org_scope = False
        elif org_id:
            path = f'/v3/enterprise/consumption/daily/organizations/{quote(org_id, safe="")}'
            org_scope = True
        else:
            path = '/v3/enterprise/consumption/daily'
            org_scope = False
        daily_response = await self._client.get(path, params={'time_after': cycle.after, 'time_before': min(now, cycle.before)})
        daily_response.raise_for_status()
        try:
            daily = DailyResponse.model_validate(daily_response.json())
            used = _acu(daily.total_acus)
        except (ValidationError, ValueError, TypeError):
            raise ValueError
        limit = remaining = used_percent = remaining_percent = None
        if org_scope:
            org_path = f'/v3/enterprise/organizations/{quote(org_id, safe="")}'
            try:
                org_response = await self._client.get(org_path)
                if org_response.is_success:
                    org_data = OrgResponse.model_validate(org_response.json())
                    if org_data.max_cycle_acu_limit is not None:
                        limit = _acu(org_data.max_cycle_acu_limit)
                        if limit > 0:
                            remaining = max(limit - used, 0)
                            used_percent = min(used / limit * 100, 100)
                            remaining_percent = max(100 - used_percent, 0)
                        else:
                            remaining = 0
                            used_percent = 100 if used else 0
                            remaining_percent = 0 if used else 100
            except (httpx.HTTPError, ValidationError, ValueError, TypeError):
                pass
        updated = utc_now()
        return ProviderUsage(provider='devin', status='ok', limits=[UsageLimit(
            id='cycle-acu-usage', name='Cycle ACU Usage', used=used, remaining=remaining, limit=limit,
            used_percent=used_percent, remaining_percent=remaining_percent, unit='ACUs',
            reset_at=datetime.fromtimestamp(cycle.before, tz=ZoneInfo(self.settings.timezone)),
            window='billing cycle', updated_at=updated)], last_successful_update=updated, source=self.source)

    async def _fetch_organization_sessions(self) -> ProviderUsage:
        org_id = self.settings.devin_org_id.strip()
        if not org_id:
            raise ValueError
        user_id = self.settings.devin_user_id.strip()
        now = int(time.time())
        days = self.settings.devin_session_window_days
        params: dict[str, Any] = {
            'first': 200,
            'created_after': now - days * 86400,
            'created_before': now,
        }
        if user_id:
            params['user_ids'] = [user_id]
        seen_cursors: set[str] = set()
        sessions: list[SessionSummary] = []
        for page in range(100):
            response = await self._client.get(
                f'/v3/organizations/{quote(org_id, safe="")}/sessions', params=params)
            response.raise_for_status()
            try:
                sessions_page = SessionsResponse.model_validate(response.json())
            except (ValidationError, ValueError, TypeError):
                raise ValueError
            sessions.extend(sessions_page.items)
            if not sessions_page.has_next_page:
                break
            cursor = sessions_page.end_cursor
            if not cursor or cursor in seen_cursors:
                raise ValueError
            seen_cursors.add(cursor)
            params = {'first': 200, 'after': cursor, 'created_after': now - days * 86400,
                      'created_before': now}
            if user_id:
                params['user_ids'] = [user_id]
        else:
            raise ValueError
        total_acus = sum(_acu(session.acus_consumed) for session in sessions)
        updated = utc_now()
        window = f'created in the last {days} days'
        return ProviderUsage(provider='devin', status='ok', limits=[
            UsageLimit(id='organization-session-acus', name=f'{days}-Day Session Usage', used=total_acus,
                       unit='ACUs', window=f'sessions {window}', updated_at=updated),
            UsageLimit(id='organization-session-count', name='Sessions Included', used=float(len(sessions)),
                       unit='sessions', window=window, updated_at=updated),
        ], last_successful_update=updated, source=self.source)

    async def aclose(self) -> None:
        if self._owns_client:
            await self._client.aclose()