"""TLS Client oturumları için yapılandırma temelli yardımcı araçlar (rnet).""" from __future__ import annotations import asyncio import inspect import logging import math import re from pathlib import Path from typing import Any, Callable, Dict, Iterable, Mapping, Optional, Tuple from http.cookiejar import CookieJar from requests.cookies import RequestsCookieJar, cookiejar_from_dict from requests.structures import CaseInsensitiveDict from requests.utils import dict_from_cookiejar try: # pragma: no cover - isteğe bağlı bağımlılık from rnet import Client as _RNETAsyncClient, Emulation, Method, Proxy try: # EmulationOS may not exist in every rnet version from rnet import EmulationOS except ImportError: EmulationOS = None # type: ignore[assignment] from rnet.blocking import Client as _RNETClient _RNET_AVAILABLE = True except ImportError: # pragma: no cover - ortamda rnet yok EmulationOS = None # type: ignore[assignment] Emulation = Method = Proxy = _RNETClient = _RNETAsyncClient = None # type: ignore[assignment] _RNET_AVAILABLE = False SessionType = Any class TLSClientConfigError(RuntimeError): """TLS istemcisi hazırlanırken oluşan hataları temsil eder.""" # ---- Varsayılan emülasyon: Chrome141 (UA otomatik emülasyondan gelir) ---- _DEFAULT_EMULATION: Optional[Any] = ( getattr(Emulation, "Chrome141", None) if (_RNET_AVAILABLE and Emulation is not None) else None ) _DEFAULT_EMULATION_OS: Optional[Any] = None if _RNET_AVAILABLE and EmulationOS is not None: _DEFAULT_EMULATION_OS = getattr(EmulationOS, "Windows", None) # ---- User-Agent'tan emülasyon çıkaran yardımcı ---- _CHROME_VERSION_PATTERN = re.compile(r"(?:Chrome|CriOS)/(\d+)", re.IGNORECASE) def resolve_emulation_from_user_agent(user_agent: str) -> Tuple[Optional[Any], Optional[str]]: """Kullanıcı ajanındaki Chrome sürümünü yakalayarak uygun emülasyonu döndür.""" if not user_agent or not isinstance(user_agent, str): return None, None if not (_RNET_AVAILABLE and Emulation is not None): # pragma: no cover - rnet mevcut değil return None, None match = _CHROME_VERSION_PATTERN.search(user_agent) if not match: return None, None major_version = match.group(1) attr_name = f"Chrome{major_version}" emulation_value = getattr(Emulation, attr_name, None) if emulation_value is None: return None, None return emulation_value, attr_name def _get_emulation_os_value(*names: str) -> Tuple[Optional[Any], Optional[str]]: if EmulationOS is None: # pragma: no cover - rnet mevcut degil veya EmulationOS yok return None, None for name in names: value = getattr(EmulationOS, name, None) if value is not None: return value, name return None, None def resolve_emulation_os_from_user_agent(user_agent: str) -> Tuple[Optional[Any], Optional[str]]: """Kullanici ajanindan uygun emulation OS degerini secer.""" if not user_agent or not isinstance(user_agent, str): return None, None if not (_RNET_AVAILABLE and EmulationOS is not None): # pragma: no cover return None, None ua = user_agent.lower() if "android" in ua: return _get_emulation_os_value("Android") if "iphone" in ua or "ipad" in ua or "ios" in ua: return _get_emulation_os_value("iOS", "IOS", "IPhone", "iPhone") if "mac os x" in ua or "macintosh" in ua: return _get_emulation_os_value("MacOS", "Macos", "Mac") if "windows" in ua: return _get_emulation_os_value("Windows") if "linux" in ua: return _get_emulation_os_value("Linux") return None, None # ============================== Response Adaptörü =============================== class RNetResponseAdapter: """rnet ``Response`` nesnesini requests yanıtı ile uyumlu hale getir.""" def __init__( self, response: Any, *, url: str, loop: Optional[asyncio.AbstractEventLoop] = None, ) -> None: self._response = response self._loop = loop self.url = getattr(response, "url", url) self.status_code = self._extract_status_code(response) self.headers = self._extract_headers(response) self.history = list(getattr(response, "history", []) or []) self.cookies: RequestsCookieJar = self._extract_cookies(response) self._text: Optional[str] = None self._content: Optional[bytes] = None # requests.Response ile uyumluluk için nitelikler self.reason = None self.ok = self.status_code is not None and 200 <= self.status_code < 400 @staticmethod def _extract_status_code(response: Any) -> Optional[int]: status = getattr(response, "status", None) if status is None: status = getattr(response, "status_code", None) if status is None: return None if callable(getattr(status, "as_int", None)): try: return int(status.as_int()) except Exception: pass try: return int(status) except Exception: return None def _extract_headers(self, response: Any) -> CaseInsensitiveDict[str]: """Yanıt nesnesinden başlıkları çıkarmak için çeşitli stratejileri dener.""" headers = self._try_extract_from_raw_pairs(response) if headers: return headers headers = self._try_extract_from_map(response) if headers: return headers headers = self._try_extract_from_keys(response) return headers @staticmethod def _b2s(value: Any) -> str: """Baytları dizeye dönüştürür.""" if isinstance(value, (bytes, bytearray)): try: return value.decode("latin-1") except UnicodeDecodeError: return value.decode("utf-8", "replace") return str(value) def _try_extract_from_raw_pairs(self, response: Any) -> Optional[CaseInsensitiveDict[str]]: """Strateji 1: 'raw_headers' gibi ham anahtar-değer çiftlerinden başlıkları çıkarır.""" for attr in ("raw_headers", "header_pairs", "pairs", "_pairs"): pairs = getattr(response, attr, None) if not pairs: continue try: headers = CaseInsensitiveDict() for key, value in list(pairs): name, val = self._b2s(key), self._b2s(value) # Birden fazla aynı adlı başlık için değerleri birleştir headers[name] = f"{headers[name]}, {val}" if name in headers else val if headers: return headers except Exception: continue # Bir sonraki özniteliği dene return None def _try_extract_from_map(self, response: Any) -> Optional[CaseInsensitiveDict[str]]: """Strateji 2: 'response.headers' gibi bir eşlem (map) nesnesinden başlıkları çıkarır.""" header_map = getattr(response, "headers", None) if not hasattr(header_map, "items"): return None try: headers = CaseInsensitiveDict() for key, value in header_map.items(): name = self._b2s(key) val = self._flatten_header_value(value) headers[name] = f"{headers[name]}, {val}" if name in headers else val return headers except Exception: return None def _flatten_header_value(self, value: Any) -> str: """Bir başlık değerini tek bir dizeye düzleştirir.""" if isinstance(value, (str, bytes, bytearray)): return self._b2s(value) # Değer bir liste veya demet ise, elemanları birleştir if isinstance(value, (list, tuple)): return ", ".join(self._b2s(part) for part in value if part is not None) return self._b2s(value) def _try_extract_from_keys(self, response: Any) -> CaseInsensitiveDict[str]: """Strateji 3: 'keys()' ve 'get'/'get_all' metodlarını kullanarak geri dönüş (fallback) olarak başlıkları çıkarır.""" header_map = getattr(response, "headers", None) if not callable(getattr(header_map, "keys", None)): return CaseInsensitiveDict() headers = CaseInsensitiveDict() try: keys = list(header_map.keys()) except Exception: return headers get_all = getattr(header_map, "get_all", None) get = getattr(header_map, "get", None) for key in keys: key_str = self._b2s(key) value = None try: if callable(get_all): value = get_all(key_str) elif callable(get): value = get(key_str) else: value = header_map[key] except Exception: continue if value is not None: headers[key_str] = self._flatten_header_value(value) return headers @staticmethod def _extract_cookies(response: Any) -> RequestsCookieJar: jar = RequestsCookieJar() cookies_attr = getattr(response, "cookies", None) if not cookies_attr: return jar if isinstance(cookies_attr, RequestsCookieJar): jar.update(cookies_attr) return jar if isinstance(cookies_attr, dict): # Sözlükten çerezleri yükle return cookiejar_from_dict({str(k): str(v) for k, v in cookies_attr.items()}) # Çerez nesneleri listesini işle try: for cookie in iter(cookies_attr): name = getattr(cookie, "name", None) value = getattr(cookie, "value", None) if name is None or value is None: continue rest = { "domain": getattr(cookie, "domain", ""), "path": getattr(cookie, "path", "/"), } jar.set(str(name), str(value), **rest) except TypeError: # cookies_attr yinelenebilir değilse görmezden gel pass return jar def json(self) -> Any: json_fn = getattr(self._response, "json", None) if callable(json_fn): value = json_fn() return self._await_value_sync(value) raise AttributeError("JSON yanıtı okunamadı") async def json_async(self) -> Any: json_fn = getattr(self._response, "json", None) if callable(json_fn): value = json_fn() return await self._await_value(value) raise AttributeError("JSON yanıtı okunamadı") @property def text(self) -> str: return self._get_text_sync() async def text_async(self) -> str: return await self._get_text_async() @property def content(self) -> bytes: return self._get_content_sync() async def content_async(self) -> bytes: return await self._get_content_async() def close(self) -> None: close_fn = getattr(self._response, "close", None) if callable(close_fn): close_fn() def _call_member(self, name: str) -> Any: member = getattr(self._response, name, None) if callable(member): return member() return member @staticmethod async def _await_value(value: Any) -> Any: if inspect.isawaitable(value): return await value return value def _await_value_sync(self, value: Any) -> Any: if not inspect.isawaitable(value): return value async def _consume() -> Any: return await value # type: ignore[misc] try: asyncio.get_running_loop() except RuntimeError: pass else: raise RuntimeError( "Asenkron yanıtı senkron olarak okumaya çalıştınız. " "Lütfen 'await response.text_async()', 'await response.content_async()' " "veya 'await response.json_async()' kullanın." ) loop = getattr(self, "_loop", None) if loop is not None: try: if not loop.is_closed() and not loop.is_running(): return loop.run_until_complete(_consume()) except Exception: pass return asyncio.run(_consume()) def _get_text_sync(self) -> str: if self._text is not None: return self._text value = self._call_member("text") value = self._await_value_sync(value) if isinstance(value, str): self._text = value elif value is None: self._text = self._get_content_sync().decode("utf-8", "replace") else: self._text = str(value) return self._text async def _get_text_async(self) -> str: if self._text is not None: return self._text value = self._call_member("text") value = await self._await_value(value) if isinstance(value, str): self._text = value elif value is None: content = await self._get_content_async() self._text = content.decode("utf-8", "replace") else: self._text = str(value) return self._text def _get_content_sync(self) -> bytes: if self._content is not None: return self._content value = self._call_member("bytes") value = self._await_value_sync(value) if isinstance(value, (bytes, bytearray)): self._content = bytes(value) elif value is None: self._content = b"" else: self._content = bytes(str(value), "utf-8") return self._content async def _get_content_async(self) -> bytes: if self._content is not None: return self._content value = self._call_member("bytes") value = await self._await_value(value) if isinstance(value, (bytes, bytearray)): self._content = bytes(value) elif value is None: self._content = b"" else: self._content = bytes(str(value), "utf-8") return self._content # ============================== Request Hazırlayıcı =============================== class _RequestKwargsBuilder: """requests tarzı argümanları rnet istemcisi için hazırlar.""" # Otomatik enjeksiyon yapan "browser default" başlıkları _AUTO_BROWSER_HEADERS = { "sec-fetch-site", "sec-fetch-mode", "sec-fetch-dest", "sec-fetch-user", "priority", "sec-ch-ua", "sec-ch-ua-mobile", "sec-ch-ua-platform", } def __init__(self, session: Any, kwargs: Mapping[str, Any]): self.session = session self.options = dict(kwargs) self.request_kwargs: Dict[str, Any] = {"default_headers": False} self.provided_header_keys_lc: set[str] = set() def build(self) -> Dict[str, Any]: """Tüm argümanları işleyerek son rnet kwargs sözlüğünü oluşturur.""" self._process_redirects() self._process_timeout() self._process_verify() self._process_headers() self._process_cookies() self._process_proxy() self._process_body() return self.request_kwargs def _process_redirects(self): allow_redirects = self.options.pop("allow_redirects", True) self.request_kwargs["allow_redirects"] = bool(allow_redirects) max_redirects = self.options.pop("max_redirects", None) if max_redirects is not None: try: self.request_kwargs["max_redirects"] = int(max_redirects) except (ValueError, TypeError): pass def _process_timeout(self): timeout = self.options.pop("timeout", None) timeout_seconds = self.options.pop("timeout_seconds", timeout) normalized = self.session._normalise_timeout(timeout_seconds) if normalized is not None: self.request_kwargs["timeout"] = int(math.ceil(normalized)) def _process_verify(self): insecure_skip_verify = bool(self.options.pop("insecure_skip_verify", False)) verify = self.options.pop("verify", True) verify_param = self.session._resolve_verify(verify, insecure_skip_verify) if verify_param is not None: self.request_kwargs["verify"] = verify_param def _process_headers(self): # Oturum ve istek başlıklarını birleştir merged_headers = CaseInsensitiveDict(self.session.headers) request_headers = self.options.pop("headers", None) if request_headers: self.provided_header_keys_lc = {str(k).lower() for k in getattr(request_headers, "keys", lambda: [])()} merged_headers.update(request_headers) # İsteğe özel ayarlara göre otomatik başlıkları kaldır strip_auto = bool(self.options.pop("strip_auto_headers", getattr(self.session, "_strip_auto_headers", True))) if strip_auto: self._strip_auto_headers(merged_headers) if not merged_headers: return self.request_kwargs["headers"] = dict(merged_headers.items()) # Başlık sırasını ve beyaz listeyi yönet strict = bool(self.options.pop("strict_headers", getattr(self.session, "_strict_headers", False))) if strict: self.request_kwargs["orig_headers"] = list(merged_headers.keys()) elif self.session._orig_headers: self.request_kwargs["orig_headers"] = list(self.session._orig_headers) def _strip_auto_headers(self, merged: CaseInsensitiveDict[str]): """Kullanıcı tarafından açıkça sağlanmadıysa, otomatik tarayıcı başlıklarını kaldırır.""" remove_keys = [ k for k in merged if k.lower() in self._AUTO_BROWSER_HEADERS and k.lower() not in self.provided_header_keys_lc ] for k in remove_keys: merged.pop(k, None) def _process_cookies(self): request_cookies = self.options.pop("cookies", None) cookie_jar = self.session._merge_cookies(self.session.cookies, request_cookies) if cookie_jar: self.request_kwargs["cookies"] = dict_from_cookiejar(cookie_jar) def _process_proxy(self): proxy_obj = self.session._build_proxy() if proxy_obj is not None: self.request_kwargs["proxy"] = proxy_obj def _process_body(self): params = self.options.pop("params", None) if params: self.request_kwargs["query"] = dict(params) if isinstance(params, Mapping) else params json_data = self.options.pop("json", None) if json_data is not None: self.request_kwargs["json"] = json_data return data = self.options.pop("data", None) if data is not None: if isinstance(data, Mapping): self.request_kwargs["form"] = dict(data) else: self.request_kwargs["body"] = data def _prepare_request_arguments( session: Any, method: str, url: str, kwargs: Mapping[str, Any] ) -> Tuple[Any, Dict[str, Any]]: """`_RequestKwargsBuilder` kullanarak istek argümanlarını hazırlar.""" if not method: raise ValueError("HTTP method belirtilmelidir") if not url: raise ValueError("URL belirtilmelidir") method_enum = session._METHOD_MAP.get(method.upper()) if method_enum is None: raise ValueError(f"Desteklenmeyen HTTP methodu: {method}") builder = _RequestKwargsBuilder(session, kwargs) request_kwargs = builder.build() # Kalan kwargs'ları temizle (uyumluluk) kwargs.clear() return method_enum, request_kwargs # ============================== Session Adaptörleri =============================== class RNetSessionAdapter: """rnet ``Client`` nesnesini requests benzeri API ile uyumlu hale getir.""" _METHOD_MAP = { "GET": Method.GET if Method else None, "POST": Method.POST if Method else None, "PUT": Method.PUT if Method else None, "DELETE": Method.DELETE if Method else None, "HEAD": Method.HEAD if Method else None, "OPTIONS": Method.OPTIONS if Method else None, "PATCH": Method.PATCH if Method else None, "TRACE": Method.TRACE if Method else None, } def __init__(self, client: Any, *, orig_headers: Optional[Iterable[str]] = None, strict_headers: bool = False, strip_auto_headers: bool = True) -> None: self._client = client self.headers: CaseInsensitiveDict[str] = CaseInsensitiveDict() self.proxies: Dict[str, str] = {} self.cookies: RequestsCookieJar = RequestsCookieJar() self._orig_headers = list(orig_headers or []) self._strict_headers = bool(strict_headers) self._strip_auto_headers = bool(strip_auto_headers) # -- requests tarzı API ------------------------------------------------- def request(self, method: str, url: str, **kwargs: Any) -> RNetResponseAdapter: method_enum, request_kwargs = _prepare_request_arguments(self, method, url, kwargs) response = self._client.request(method_enum, url, **request_kwargs) return self._adapt_response(response, url) # -- requests tarzı kısayollar ----------------------------------------- def get(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("GET", url, **kwargs) def post(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("POST", url, **kwargs) def put(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("PUT", url, **kwargs) def delete(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("DELETE", url, **kwargs) def head(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("HEAD", url, **kwargs) def options(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("OPTIONS", url, **kwargs) def patch(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return self.request("PATCH", url, **kwargs) def close(self) -> None: close_fn = getattr(self._client, "close", None) if callable(close_fn): close_fn() def _adapt_response(self, response: Any, url: str) -> RNetResponseAdapter: try: loop = asyncio.get_running_loop() except RuntimeError: loop = None adapted = RNetResponseAdapter(response, url=url, loop=loop) self.cookies.update(adapted.cookies) return adapted # -- Yardımcılar -------------------------------------------------------- @staticmethod def _normalise_timeout(timeout: Any) -> Optional[float]: if timeout is None: return None if isinstance(timeout, (list, tuple)): numeric = [value for value in timeout if isinstance(value, (int, float))] if not numeric: return None return max(numeric) if isinstance(timeout, (int, float)): return float(timeout) return None @staticmethod def _merge_cookies(base: CookieJar, extra: Any) -> RequestsCookieJar: merged = RequestsCookieJar() merged.update(base) if not extra: return merged if isinstance(extra, RequestsCookieJar): merged.update(extra) return merged try: data = dict(extra) except Exception: return merged merged.update(cookiejar_from_dict(data)) return merged def _build_proxy(self) -> Optional[Any]: if not self.proxies: return None proxy_url = self.proxies.get("https") or self.proxies.get("http") if not proxy_url or Proxy is None: return None try: # Tek uçtan tüm protokoller için uygun temsil return Proxy.all(str(proxy_url)) except Exception: return None @staticmethod def _resolve_verify(verify: Any, insecure_skip_verify: bool) -> Optional[Any]: if insecure_skip_verify: return False if isinstance(verify, (str, Path)): return Path(verify) if isinstance(verify, bool): return verify return None class AsyncRNetSessionAdapter(RNetSessionAdapter): """Asenkron ``Client`` nesnesi için requests benzeri API.""" async def request(self, method: str, url: str, **kwargs: Any) -> RNetResponseAdapter: method_enum, request_kwargs = _prepare_request_arguments(self, method, url, kwargs) response = self._client.request(method_enum, url, **request_kwargs) if inspect.isawaitable(response): response = await response return self._adapt_response(response, url) async def get(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("GET", url, **kwargs) async def post(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("POST", url, **kwargs) async def put(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("PUT", url, **kwargs) async def delete(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("DELETE", url, **kwargs) async def head(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("HEAD", url, **kwargs) async def options(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("OPTIONS", url, **kwargs) async def patch(self, url: str, **kwargs: Any) -> RNetResponseAdapter: return await self.request("PATCH", url, **kwargs) async def close(self) -> None: close_fn = getattr(self._client, "close", None) if callable(close_fn): result = close_fn() if inspect.isawaitable(result): await result # ============================== TLSClient Fabrikası ============================== class _RnetClientParamsBuilder: """TLSClient için rnet istemci parametrelerini oluşturur.""" def __init__(self, emulation: Any, overrides: Mapping[str, Any]): self.base_emulation = emulation self.overrides = overrides self.user_agent = self._extract_user_agent(overrides) self.client_params: Dict[str, Any] = {} @staticmethod def _extract_user_agent(overrides: Mapping[str, Any]) -> str: """Override'lardan User-Agent başlığını çıkarır.""" headers = overrides.get("headers") if isinstance(headers, Mapping): for key, value in headers.items(): if str(key).lower() == "user-agent": return str(value).strip() return "" def build(self) -> Dict[str, Any]: """Parametreleri derleyip son sözlüğü döndürür.""" self._set_default_params() self._determine_emulation() self._determine_emulation_os() # Kullanıcı override'ları her zaman önceliklidir self.client_params.update(self.overrides) return self.client_params def _set_default_params(self): """Temel rnet istemci parametrelerini ayarlar.""" self.client_params.update({ "cookie_store": True, "http2_only": False, "tls": {"random_extension": True}, }) def _determine_emulation(self): """User-Agent'a göre en uygun emülasyonu seçer.""" if self.user_agent: detected_emulation, _ = resolve_emulation_from_user_agent(self.user_agent) if detected_emulation is not None: self.client_params["emulation"] = detected_emulation return self.client_params["emulation"] = self.base_emulation def _determine_emulation_os(self): """Mevcutsa, emulation_os'u User-Agent veya varsayılan değere göre ayarlar.""" if "emulation_os" in self.overrides: return # Kullanıcı zaten bir override sağladı os_to_set = None if self.user_agent: detected_os, _ = resolve_emulation_os_from_user_agent(self.user_agent) if detected_os is not None: os_to_set = detected_os if os_to_set is None and _DEFAULT_EMULATION_OS is not None: os_to_set = _DEFAULT_EMULATION_OS if os_to_set is not None: self.client_params["emulation_os"] = os_to_set class TLSClient: """rnet Client'i baslatip requests-benzeri bir session adapter'i dondurur.""" def __init__( self, *, emulation: Optional[Any] = None, auto_initialize: bool = True, async_mode: bool = False, client_factory: Optional[Callable[..., Any]] = None, # ileri kullanım için opsiyonel client kwargs (örn. http2_only, tls_options vs.) client_overrides: Optional[Mapping[str, Any]] = None, # ek denetimler: strict_headers: bool = False, strip_auto_headers: bool = True, ) -> None: if not _RNET_AVAILABLE: raise TLSClientConfigError("rnet kütüphanesi yüklü değil.") self.async_mode = bool(async_mode) self.emulation = emulation or _DEFAULT_EMULATION if self.emulation is None: raise TLSClientConfigError( "Varsayılan rnet emülasyon profili (Chrome141) kullanılabilir değil." ) self._strict_headers = bool(strict_headers) self._strip_auto_headers = bool(strip_auto_headers) self._client_overrides = dict(client_overrides or {}) self.session: Optional[SessionType] = None if client_factory is None: client_factory = _RNETAsyncClient if self.async_mode else _RNETClient if client_factory is None: mode_name = "asenkron" if self.async_mode else "blocking" raise TLSClientConfigError( f"{mode_name} rnet.Client kullanılamıyor çünkü kütüphane eksik." ) self._client_factory = client_factory if auto_initialize: self.session = self.create_session() def create_session(self) -> SessionType: """Hazır TLS profilini kullanarak yeni bir rnet oturumu oluştur.""" if not _RNET_AVAILABLE: raise TLSClientConfigError("rnet kütüphanesi yüklü değil.") # Parametre oluşturma mantığını builder'a devret builder = _RnetClientParamsBuilder(self.emulation, self._client_overrides) client_params = builder.build() session_impl = self._client_factory(**client_params) adapter_class = AsyncRNetSessionAdapter if self.async_mode else RNetSessionAdapter return adapter_class( session_impl, strict_headers=self._strict_headers, strip_auto_headers=self._strip_auto_headers, ) def refresh_session( self, reload_config: bool = False, ) -> SessionType: """Var olan oturumu güncelle ve döndür.""" if reload_config: # Uyumluluk amacıyla parametre korunuyor. pass self.session = self.create_session() return self.session __all__ = [ "TLSClient", "TLSClientConfigError", "RNetSessionAdapter", "AsyncRNetSessionAdapter", "RNetResponseAdapter", "resolve_emulation_from_user_agent", "resolve_emulation_os_from_user_agent", ]