Spaces:
Sleeping
Sleeping
File size: 5,263 Bytes
494c89b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """
Accounts API - List, switch, delete accounts
"""
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
from services.token_service import TokenService
from services.kiro_service import KiroService
router = APIRouter()
token_service = TokenService()
kiro_service = KiroService()
class AccountResponse(BaseModel):
filename: str
accountName: str
email: Optional[str] = None
provider: str
authMethod: str
region: str
isActive: bool
isExpired: bool
expiresAt: Optional[str] = None
expiresIn: Optional[str] = None
class AccountListResponse(BaseModel):
accounts: List[AccountResponse]
total: int
valid: int
expired: int
activeAccount: Optional[str] = None
@router.get("", response_model=AccountListResponse)
async def list_accounts():
"""Get all accounts"""
tokens = token_service.list_tokens()
current = token_service.get_current_token()
current_refresh = current.raw_data.get('refreshToken') if current else None
accounts = []
valid_count = 0
expired_count = 0
active_account = None
for token in tokens:
is_active = (current_refresh and
token.raw_data.get('refreshToken') == current_refresh)
if is_active:
active_account = token.account_name
if token.is_expired:
expired_count += 1
else:
valid_count += 1
# Calculate expires in
expires_in = None
if token.expires_at:
try:
# Handle timezone-aware and naive datetimes
now = datetime.now(token.expires_at.tzinfo) if token.expires_at.tzinfo else datetime.now()
delta = token.expires_at - now
if delta.total_seconds() > 0:
hours = int(delta.total_seconds() // 3600)
if hours < 24:
expires_in = f"{hours}h"
else:
expires_in = f"{hours // 24}d"
else:
expires_in = "expired"
except Exception:
expires_in = "—"
accounts.append(AccountResponse(
filename=token.path.name,
accountName=token.account_name,
email=token.raw_data.get('email') if token.raw_data else None,
provider=token.provider,
authMethod=token.auth_method,
region=token.region,
isActive=is_active,
isExpired=token.is_expired,
expiresAt=token.expires_at.isoformat() if token.expires_at else None,
expiresIn=expires_in
))
return AccountListResponse(
accounts=accounts,
total=len(accounts),
valid=valid_count,
expired=expired_count,
activeAccount=active_account
)
@router.post("/{filename}/switch")
async def switch_account(filename: str):
"""Switch to a specific account"""
token = token_service.get_token(filename)
if not token:
raise HTTPException(status_code=404, detail="Account not found")
success = token_service.activate_token(token)
if not success:
raise HTTPException(status_code=500, detail="Failed to switch account")
return {"success": True, "message": f"Switched to {token.account_name}"}
@router.delete("/{filename}")
async def delete_account(filename: str):
"""Delete an account"""
token = token_service.get_token(filename)
if not token:
raise HTTPException(status_code=404, detail="Account not found")
try:
token_service.delete_token(filename)
return {"success": True, "message": "Account deleted"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/{filename}/refresh")
async def refresh_account(filename: str):
"""Refresh account token"""
token = token_service.get_token(filename)
if not token:
raise HTTPException(status_code=404, detail="Account not found")
try:
updated = token_service.refresh_and_save(token)
return {
"success": True,
"expiresAt": updated.expires_at.isoformat() if updated.expires_at else None
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/expired/all")
async def delete_expired():
"""Delete all expired accounts"""
tokens = token_service.list_tokens()
deleted = 0
for token in tokens:
if token.is_expired:
try:
token_service.delete_token(token.path.name)
deleted += 1
except Exception:
pass
return {"success": True, "deleted": deleted}
@router.get("/current")
async def get_current():
"""Get current active account"""
token = token_service.get_current_token()
if not token:
return {"active": False}
return {
"active": True,
"accountName": token.account_name,
"email": token.raw_data.get('email'),
"provider": token.provider,
"isExpired": token.is_expired
}
|