| import logging |
| import re |
| from typing import Dict, Any, Optional |
|
|
| import requests |
| from requests.exceptions import RequestException |
|
|
| |
| _BASE_URL = "https://chat.notdiamond.ai" |
| _API_BASE_URL = "https://spuckhogycrxcbomznwo.supabase.co" |
| _USER_AGENT = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36' |
|
|
| class AuthManager: |
| """ |
| AuthManager类用于管理身份验证过程,包括获取API密钥、用户信息和处理刷新令牌等操作。 |
| """ |
| |
| def __init__(self, email: str, password: str): |
| self._email: str = email |
| self._password: str = password |
| self._api_key: str = "" |
| self._user_info: Dict[str, Any] = {} |
| self._refresh_token: str = "" |
| self._session: requests.Session = requests.session() |
| |
| self._logger: logging.Logger = logging.getLogger(__name__) |
| logging.basicConfig(level=logging.INFO) |
| |
| def login(self) -> None: |
| """使用电子邮件和密码进行用户登录,并获取用户信息。""" |
| url = f"{_API_BASE_URL}/auth/v1/token?grant_type=password" |
| headers = self._get_headers(with_content_type=True) |
| data = { |
| "email": self._email, |
| "password": self._password, |
| "gotrue_meta_security": {} |
| } |
|
|
| try: |
| response = self._make_request('POST', url, headers=headers, json=data) |
| self._user_info = response.json() |
| self._refresh_token = self._user_info.get('refresh_token', '') |
| self._log_values() |
| except RequestException as e: |
| self._logger.error(f"\033[91m登录请求错误: {e}\033[0m") |
|
|
| def refresh_user_token(self) -> None: |
| """使用刷新令牌来请求一个新的访问令牌并更新实例变量。""" |
| url = f"{_API_BASE_URL}/auth/v1/token?grant_type=refresh_token" |
| headers = self._get_headers(with_content_type=True) |
| data = {"refresh_token": self._refresh_token} |
|
|
| try: |
| response = self._make_request('POST', url, headers=headers, json=data) |
| self._user_info = response.json() |
| self._refresh_token = self._user_info.get('refresh_token', '') |
| self._log_values() |
| except RequestException as e: |
| self._logger.error(f"刷新令牌请求错误: {e}") |
|
|
| def get_jwt_value(self) -> str: |
| """返回访问令牌。""" |
| return self._user_info.get('access_token', '') |
|
|
| def _log_values(self) -> None: |
| """记录刷新令牌到日志中。""" |
| self._logger.info(f"\033[92mRefresh Token: {self._refresh_token}\033[0m") |
| |
| def _fetch_apikey(self) -> str: |
| """获取API密钥。""" |
| if self._api_key: |
| return self._api_key |
|
|
| try: |
| login_url = f"{_BASE_URL}/login" |
| response = self._make_request('GET', login_url) |
| |
| match = re.search(r'<script src="(/_next/static/chunks/app/layout-[^"]+\.js)"', response.text) |
| if not match: |
| raise ValueError("未找到匹配的脚本标签") |
|
|
| js_url = f"{_BASE_URL}{match.group(1)}" |
| js_response = self._make_request('GET', js_url) |
| |
| api_key_match = re.search(r'\("https://spuckhogycrxcbomznwo\.supabase\.co","([^"]+)"\)', js_response.text) |
| if not api_key_match: |
| raise ValueError("未能匹配API key") |
| |
| self._api_key = api_key_match.group(1) |
| return self._api_key |
|
|
| except (RequestException, ValueError) as e: |
| self._logger.error(f"获取API密钥时发生错误: {e}") |
| return "" |
|
|
| def _get_headers(self, with_content_type: bool = False) -> Dict[str, str]: |
| """生成请求头。""" |
| headers = { |
| 'apikey': self._fetch_apikey(), |
| 'user-agent': _USER_AGENT |
| } |
| if with_content_type: |
| headers['Content-Type'] = 'application/json' |
| return headers |
|
|
| def _make_request(self, method: str, url: str, **kwargs) -> requests.Response: |
| """发送HTTP请求并处理异常。""" |
| try: |
| response = self._session.request(method, url, **kwargs) |
| response.raise_for_status() |
| return response |
| except RequestException as e: |
| self._logger.error(f"请求错误 ({method} {url}): {e}") |
| raise |
|
|
