1
0
Fork 0

refactor!: change the entire purpose of this script

This commit is contained in:
Arthur K. 2026-03-01 19:32:10 +03:00
parent 217e176975
commit 71d1050adb
Signed by: wzray
GPG key ID: B97F30FDC4636357
20 changed files with 1124 additions and 872 deletions

View file

@ -0,0 +1,68 @@
import logging
from typing import Callable
from typing import Any
from playwright.async_api import BrowserContext
from providers.base import Provider, ProviderTokens
from email_providers import BaseProvider
from email_providers import TempMailOrgProvider
from .tokens import load_tokens, save_tokens, get_valid_tokens
from .usage import get_usage_percent
from .registration import register_chatgpt_account
logger = logging.getLogger(__name__)
class ChatGPTProvider(Provider):
"""ChatGPT account provider"""
def __init__(
self,
email_provider_factory: Callable[[BrowserContext], BaseProvider] | None = None,
):
self.email_provider_factory = email_provider_factory or TempMailOrgProvider
@property
def name(self) -> str:
return "chatgpt"
async def get_token(self) -> str | None:
"""Get valid access token, refreshing if needed"""
tokens = await get_valid_tokens()
if not tokens:
logger.info("No valid tokens, registering new account")
success = await self.register_new_account()
if not success:
return None
tokens = await get_valid_tokens()
if not tokens:
return None
return tokens.access_token
async def register_new_account(self) -> bool:
"""Register a new ChatGPT account"""
return await register_chatgpt_account(
email_provider_factory=self.email_provider_factory,
)
async def get_usage_info(self, access_token: str) -> dict[str, Any]:
"""Get usage information for the current token"""
usage_percent = get_usage_percent(access_token)
if usage_percent < 0:
return {"error": "Failed to get usage"}
remaining = max(0, 100 - usage_percent)
return {
"used_percent": usage_percent,
"remaining_percent": remaining,
"exhausted": usage_percent >= 100,
}
def load_tokens(self) -> ProviderTokens | None:
"""Load tokens from storage"""
return load_tokens()
def save_tokens(self, tokens: ProviderTokens) -> None:
"""Save tokens to storage"""
save_tokens(tokens)