"""X.com transaction ID üretimi için yardımcı sınıflar.""" from __future__ import annotations import sys import threading from dataclasses import dataclass from pathlib import Path from typing import Optional, TYPE_CHECKING from urllib.parse import urlparse import bs4 import requests def _ensure_project_root_on_path() -> None: """Projeyi içeren kök klasörü ``sys.path`` listesine ekle.""" project_root = Path(__file__).resolve().parents[1] project_root_str = str(project_root) if project_root_str not in sys.path: sys.path.insert(0, project_root_str) _ensure_project_root_on_path() from x_client_transaction import ClientTransaction from x_client_transaction.utils import ( generate_headers, get_ondemand_file_url, handle_x_migration, ) if TYPE_CHECKING: from xGen.xgen import XGenRequestContext class TransactionIdError(RuntimeError): """Transaction ID üretimi sırasında oluşan hatayı temsil eder.""" def _extract_header_value(headers, *names: str) -> str: if not headers: return "" for name in names: try: value = headers.get(name) if value: return str(value).strip() except Exception: pass try: for key, value in headers.items(): key_text = str(key).lower() if any(key_text == n.lower() for n in names): if value: return str(value).strip() except Exception: pass return "" @dataclass class _TransactionResources: """ClientTransaction oluşturmak için gerekli ham veriler.""" home_page: bs4.BeautifulSoup ondemand_file: str class TransactionIdProvider: """X-Client-Transaction-Id üretimi için yönetici.""" def __init__( self, context: "XGenRequestContext", *, user_agent: str = "", accept_language: str = "", ) -> None: if not hasattr(context, "session"): raise TypeError("context bir oturum nesnesi sağlamalıdır") self._context = context self._lock = threading.RLock() self._client: Optional[ClientTransaction] = None self._resources: Optional[_TransactionResources] = None self._user_agent = (user_agent or "").strip() self._accept_language = (accept_language or "").strip() session_headers = getattr(getattr(context, "session", None), "headers", None) if not self._user_agent: self._user_agent = _extract_header_value(session_headers, "user-agent", "User-Agent") if not self._accept_language: self._accept_language = _extract_header_value(session_headers, "accept-language", "Accept-Language") self._http_session = self._create_http_session() @property def user_agent(self) -> str: return self._user_agent @property def accept_language(self) -> str: return self._accept_language def matches(self, user_agent: str = "", accept_language: str = "") -> bool: ua = (user_agent or "").strip() lang = (accept_language or "").strip() if ua and ua != self._user_agent: return False if lang and lang != self._accept_language: return False return True # -- Genel API --------------------------------------------------------- def generate(self, method: str, path: str) -> str: method = (method or "").upper() if not method: raise TransactionIdError("HTTP method boş olamaz") path = self._normalise_path(path) if not path.startswith("/"): path = "/" + path attempts = 0 last_error: Optional[BaseException] = None while attempts < 2: attempts += 1 try: client = self._ensure_client() return client.generate_transaction_id(method=method, path=path) except Exception as exc: # pragma: no cover - dış kütüphane hataları last_error = exc self._reset() raise TransactionIdError(f"Transaction ID üretimi başarısız: {last_error}") # -- İç yardımcılar ---------------------------------------------------- def _ensure_client(self) -> ClientTransaction: with self._lock: if self._client is not None: return self._client resources = self._load_resources() self._client = ClientTransaction( home_page_response=resources.home_page, ondemand_file_response=resources.ondemand_file, ) return self._client def _load_resources(self) -> _TransactionResources: if self._resources is not None: return self._resources with self._lock: if self._resources is not None: return self._resources home_page = self._fetch_home_page() ondemand_file = self._fetch_ondemand_file(home_page) resources = _TransactionResources( home_page=home_page, ondemand_file=ondemand_file, ) self._resources = resources return resources def _fetch_home_page(self) -> bs4.BeautifulSoup: try: home_page = handle_x_migration(self._http_session) except Exception as exc: # pragma: no cover - ağ hataları raise TransactionIdError(f"Ana sayfa alınamadı: {exc}") from exc if not isinstance(home_page, bs4.BeautifulSoup): raise TransactionIdError("Geçersiz ana sayfa yanıtı alındı") return home_page def _fetch_ondemand_file(self, home_page: bs4.BeautifulSoup) -> str: file_url = get_ondemand_file_url(response=home_page) if not file_url: raise TransactionIdError("ondemand.s dosya URL'si bulunamadı") try: response = self._http_session.request("GET", file_url) except Exception as exc: # pragma: no cover - ağ hataları raise TransactionIdError(f"ondemand.s isteği başarısız: {exc}") from exc text: Optional[str] = None if hasattr(response, "text"): text_attr = response.text text = text_attr() if callable(text_attr) else text_attr if not text and hasattr(response, "content"): content = response.content if callable(content): content = content() if isinstance(content, (bytes, bytearray)): text = content.decode("utf-8", errors="ignore") elif content: text = str(content) if not isinstance(text, str) or not text.strip(): raise TransactionIdError("ondemand.s içeriği okunamadı") return text def _normalise_path(self, value: str) -> str: if not value: return "/" parsed = urlparse(value) if parsed.scheme or parsed.netloc: return parsed.path or "/" if value.startswith("/"): return value return "/" + value def _reset(self) -> None: with self._lock: self._client = None self._resources = None def _create_http_session(self) -> requests.Session: session = requests.Session() session.headers.update( generate_headers( user_agent=self._user_agent, accept_language=self._accept_language, ) ) return session __all__ = ["TransactionIdProvider", "TransactionIdError"]