| 1 | from __future__ import annotations |
| 2 | |
| 3 | import time |
| 4 | from typing import Any |
| 5 | from urllib.parse import parse_qs, urlparse |
| 6 | |
| 7 | from plugins._oauth.helpers import state as state_store |
| 8 | |
| 9 | |
| 10 | def parse_manual_callback(raw: Any) -> dict[str, str | None] | None: |
| 11 | text = "" if raw is None else str(raw).strip() |
| 12 | if not text: |
| 13 | return None |
| 14 | |
| 15 | if text.startswith("http://") or text.startswith("https://"): |
| 16 | query = urlparse(text).query |
| 17 | elif text.startswith("?"): |
| 18 | query = text[1:] |
| 19 | elif "=" in text or "&" in text: |
| 20 | query = text |
| 21 | else: |
| 22 | return { |
| 23 | "code": text, |
| 24 | "state": None, |
| 25 | "error": None, |
| 26 | "error_description": None, |
| 27 | } |
| 28 | |
| 29 | parsed = parse_qs(query, keep_blank_values=True) |
| 30 | return { |
| 31 | "code": first_query_value(parsed, "code"), |
| 32 | "state": first_query_value(parsed, "state"), |
| 33 | "error": first_query_value(parsed, "error"), |
| 34 | "error_description": first_query_value(parsed, "error_description"), |
| 35 | } |
| 36 | |
| 37 | |
| 38 | def latest_attempt(provider_id: str): |
| 39 | state_store.cleanup_expired() |
| 40 | with state_store._lock: |
| 41 | attempts = [ |
| 42 | attempt |
| 43 | for attempt in state_store._attempts.values() |
| 44 | if attempt.provider_id == provider_id and not attempt.expired() |
| 45 | ] |
| 46 | if not attempts: |
| 47 | return None |
| 48 | return max(attempts, key=lambda attempt: attempt.created_at) |
| 49 | |
| 50 | |
| 51 | def models_from_payload(payload: Any) -> list[str]: |
| 52 | values: list[Any] |
| 53 | if isinstance(payload, dict) and isinstance(payload.get("data"), list): |
| 54 | values = payload["data"] |
| 55 | elif isinstance(payload, dict) and isinstance(payload.get("models"), list): |
| 56 | values = payload["models"] |
| 57 | elif isinstance(payload, list): |
| 58 | values = payload |
| 59 | else: |
| 60 | return [] |
| 61 | |
| 62 | models: list[str] = [] |
| 63 | seen: set[str] = set() |
| 64 | for value in values: |
| 65 | model_id = "" |
| 66 | if isinstance(value, str): |
| 67 | model_id = value |
| 68 | elif isinstance(value, dict): |
| 69 | model_id = str(value.get("id") or value.get("name") or "") |
| 70 | model_id = model_id.strip() |
| 71 | if model_id.startswith("models/"): |
| 72 | model_id = model_id.split("/", 1)[1] |
| 73 | if model_id and model_id not in seen: |
| 74 | seen.add(model_id) |
| 75 | models.append(model_id) |
| 76 | return models |
| 77 | |
| 78 | |
| 79 | def json_payload(response: Any) -> dict[str, Any]: |
| 80 | try: |
| 81 | payload = response.json() |
| 82 | except Exception: |
| 83 | payload = {} |
| 84 | if not isinstance(payload, dict): |
| 85 | return {} |
| 86 | return payload |
| 87 | |
| 88 | |
| 89 | def error_message(payload: dict[str, Any], fallback: str) -> str: |
| 90 | error = payload.get("error") |
| 91 | if isinstance(error, dict): |
| 92 | return str(error.get("message") or error.get("status") or fallback) |
| 93 | return str(payload.get("error_description") or payload.get("error") or fallback) |
| 94 | |
| 95 | |
| 96 | def first_query_value(parsed: dict[str, list[str]], key: str) -> str | None: |
| 97 | values = parsed.get(key) or [] |
| 98 | if not values: |
| 99 | return None |
| 100 | return values[0] |
| 101 | |
| 102 | |
| 103 | def as_optional_string(value: Any) -> str | None: |
| 104 | if isinstance(value, list): |
| 105 | value = value[0] if value else None |
| 106 | text = "" if value is None else str(value).strip() |
| 107 | return text or None |
| 108 | |
| 109 | |
| 110 | def expires_ms(payload: dict[str, Any]) -> int: |
| 111 | if payload.get("expires_at") is not None: |
| 112 | try: |
| 113 | value = float(payload["expires_at"]) |
| 114 | if value < 1_000_000_000_000: |
| 115 | value *= 1000 |
| 116 | return int(value) |
| 117 | except (TypeError, ValueError): |
| 118 | pass |
| 119 | try: |
| 120 | return int((time.time() + float(payload.get("expires_in") or 0)) * 1000) |
| 121 | except (TypeError, ValueError): |
| 122 | return 0 |
| 123 | |
| 124 | |
| 125 | def as_int(value: Any, default: int) -> int: |
| 126 | try: |
| 127 | return int(value) |
| 128 | except (TypeError, ValueError): |
| 129 | return default |