Files
saidsurucuandClaude Opus 4.7 96a5a538b2 perf(server): unblock event loop on rate-limit waits and markitdown
Two complementary changes to mitigate intermittent TLS handshake
timeouts and "notifications/cancelled: Bad Request" seen against the
single-worker uvicorn deployment.

1. bedesten rate-limiter back-pressure
   - Add optional ``max_wait`` to ``_TokenBucket.acquire``: if the next
     wait would exceed it, raise ``BedestenRateLimited`` immediately
     instead of sleeping. After a server-side 429 the bucket pauses for
     up to 30s; previously a queued request sat in ``asyncio.sleep``
     for that whole window, holding the worker slot and pushing the
     MCP client past its cancellation timeout.
   - ``search_bedesten_unified`` / ``get_bedesten_document_markdown``
     catch ``BedestenRateLimited`` and reuse the existing structured
     429-style response, so callers get a fast, clean retry signal.
   - Tunable via ``BEDESTEN_RATE_MAX_WAIT_S`` (default 8.0s).

2. Offload sync markitdown conversions to a thread
   - Every ``markitdown.convert*`` call site is now wrapped in
     ``asyncio.to_thread(...)`` across 14 modules (bedesten, yargitay,
     danistay, anayasa norm + bireysel, uyusmazlik, emsal, rekabet,
     gib, kvkk, sayistay, bddk, sigorta_tahkim, kik_v2). PDF / large
     HTML parsing was stalling the event loop for seconds, which on a
     single-worker deployment delayed every other in-flight request
     and queued new TLS handshakes until they timed out.

Verified locally:
- ``ast.parse`` + ``importlib.import_module`` on all 15 modified files
- ``mcp_server_main.create_app()`` constructs successfully
- New ``_TokenBucket.acquire(max_wait=...)`` smoke-tested across 6
  paths: capacity-available, no-arg backward compat, max_wait raise,
  max_wait wait+succeed, ``penalize_until`` + max_wait fast-raise.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 14:31:23 +03:00

373 lines
15 KiB
Python

# kvkk_mcp_module/client.py
import asyncio
import httpx
from bs4 import BeautifulSoup
from typing import List, Optional, Dict, Any
import logging
import os
import re
import io
import math
from urllib.parse import urljoin, urlparse, parse_qs
from markitdown import MarkItDown
from pydantic import HttpUrl
from .models import (
KvkkSearchRequest,
KvkkDecisionSummary,
KvkkSearchResult,
KvkkDocumentMarkdown
)
logger = logging.getLogger(__name__)
if not logger.hasHandlers():
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
class KvkkApiClient:
"""
API client for searching and retrieving KVKK (Personal Data Protection Authority) decisions
using Brave Search API for discovery and direct HTTP requests for content retrieval.
"""
BRAVE_API_URL = "https://api.search.brave.com/res/v1/web/search"
KVKK_BASE_URL = "https://www.kvkk.gov.tr"
DOCUMENT_MARKDOWN_CHUNK_SIZE = 5000 # Character limit per page
def __init__(self, request_timeout: float = 60.0):
"""Initialize the KVKK API client."""
self.brave_api_token = os.getenv("BRAVE_API_TOKEN")
if not self.brave_api_token:
# Fallback to provided free token
self.brave_api_token = "BSAuaRKB-dvSDSQxIN0ft1p2k6N82Kq"
logger.info("Using fallback Brave API token (limited free token)")
else:
logger.info("Using Brave API token from environment variable")
self.http_client = httpx.AsyncClient(
headers={
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Accept-Language": "tr-TR,tr;q=0.9,en-US;q=0.8,en;q=0.7",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
},
timeout=request_timeout,
verify=True,
follow_redirects=True
)
def _construct_search_query(self, keywords: str) -> str:
"""Construct the search query for Brave API."""
base_query = 'site:kvkk.gov.tr "karar özeti"'
if keywords.strip():
return f"{base_query} {keywords.strip()}"
return base_query
def _extract_decision_id_from_url(self, url: str) -> Optional[str]:
"""Extract decision ID from KVKK decision URL."""
try:
# Example URL: https://www.kvkk.gov.tr/Icerik/7288/2021-1303
parsed_url = urlparse(url)
path_parts = parsed_url.path.strip('/').split('/')
if len(path_parts) >= 3 and path_parts[0] == 'Icerik':
# Extract the decision ID from the path
decision_id = '/'.join(path_parts[1:]) # e.g., "7288/2021-1303"
return decision_id
except Exception as e:
logger.debug(f"Could not extract decision ID from URL {url}: {e}")
return None
def _extract_decision_metadata_from_title(self, title: str) -> Dict[str, Optional[str]]:
"""Extract decision metadata from title string."""
metadata = {
"decision_date": None,
"decision_number": None
}
if not title:
return metadata
# Extract decision date (DD/MM/YYYY format)
date_match = re.search(r'(\d{1,2}/\d{1,2}/\d{4})', title)
if date_match:
metadata["decision_date"] = date_match.group(1)
# Extract decision number (YYYY/XXXX format)
number_match = re.search(r'(\d{4}/\d+)', title)
if number_match:
metadata["decision_number"] = number_match.group(1)
return metadata
async def search_decisions(self, params: KvkkSearchRequest) -> KvkkSearchResult:
"""Search for KVKK decisions using Brave API."""
search_query = self._construct_search_query(params.keywords)
logger.info(f"KvkkApiClient: Searching with query: {search_query}")
try:
# Calculate offset for pagination
offset = (params.page - 1) * params.pageSize
response = await self.http_client.get(
self.BRAVE_API_URL,
headers={
"Accept": "application/json",
"Accept-Encoding": "gzip",
"x-subscription-token": self.brave_api_token
},
params={
"q": search_query,
"country": "TR",
"search_lang": "tr",
"ui_lang": "tr-TR",
"offset": offset,
"count": params.pageSize
}
)
response.raise_for_status()
data = response.json()
# Extract search results
decisions = []
web_results = data.get("web", {}).get("results", [])
for result in web_results:
title = result.get("title", "")
url = result.get("url", "")
description = result.get("description", "")
# Extract metadata from title
metadata = self._extract_decision_metadata_from_title(title)
# Extract decision ID from URL
decision_id = self._extract_decision_id_from_url(url)
decision = KvkkDecisionSummary(
title=title,
url=HttpUrl(url) if url else None,
description=description,
decision_id=decision_id,
publication_date=metadata.get("decision_date"),
decision_number=metadata.get("decision_number")
)
decisions.append(decision)
# Get total results if available
total_results = None
query_info = data.get("query", {})
if "total_results" in query_info:
total_results = query_info["total_results"]
return KvkkSearchResult(
decisions=decisions,
total_results=total_results,
page=params.page,
pageSize=params.pageSize,
query=search_query
)
except httpx.RequestError as e:
logger.error(f"KvkkApiClient: HTTP request error during search: {e}")
return KvkkSearchResult(
decisions=[],
total_results=0,
page=params.page,
pageSize=params.pageSize,
query=search_query
)
except Exception as e:
logger.error(f"KvkkApiClient: Unexpected error during search: {e}")
return KvkkSearchResult(
decisions=[],
total_results=0,
page=params.page,
pageSize=params.pageSize,
query=search_query
)
def _extract_decision_content_from_html(self, html: str, url: str) -> Dict[str, Any]:
"""Extract decision content from KVKK decision page HTML."""
try:
soup = BeautifulSoup(html, 'html.parser')
# Extract title
title = None
title_element = soup.find('h3', class_='blog-post-title')
if title_element:
title = title_element.get_text(strip=True)
elif soup.title:
title = soup.title.get_text(strip=True)
# Extract decision content from the main content div
content_div = soup.find('div', class_='blog-post-inner')
if not content_div:
# Fallback to other possible content containers
content_div = soup.find('div', style='text-align:justify;')
if not content_div:
logger.warning(f"Could not find decision content div in {url}")
return {
"title": title,
"decision_date": None,
"decision_number": None,
"subject_summary": None,
"html_content": None
}
# Extract decision metadata from table
decision_date = None
decision_number = None
subject_summary = None
table = content_div.find('table')
if table:
rows = table.find_all('tr')
for row in rows:
cells = row.find_all('td')
if len(cells) >= 3:
field_name = cells[0].get_text(strip=True)
field_value = cells[2].get_text(strip=True)
if 'Karar Tarihi' in field_name:
decision_date = field_value
elif 'Karar No' in field_name:
decision_number = field_value
elif 'Konu Özeti' in field_name:
subject_summary = field_value
return {
"title": title,
"decision_date": decision_date,
"decision_number": decision_number,
"subject_summary": subject_summary,
"html_content": str(content_div)
}
except Exception as e:
logger.error(f"Error extracting content from HTML for {url}: {e}")
return {
"title": None,
"decision_date": None,
"decision_number": None,
"subject_summary": None,
"html_content": None
}
def _convert_html_to_markdown(self, html_content: str) -> Optional[str]:
"""Convert HTML content to Markdown using MarkItDown with BytesIO to avoid filename length issues."""
if not html_content:
return None
try:
# Convert HTML string to bytes and create BytesIO stream
html_bytes = html_content.encode('utf-8')
html_stream = io.BytesIO(html_bytes)
# Pass BytesIO stream to MarkItDown to avoid temp file creation
md_converter = MarkItDown(enable_plugins=False)
result = md_converter.convert(html_stream)
return result.text_content
except Exception as e:
logger.error(f"Error converting HTML to Markdown: {e}")
return None
async def get_decision_document(self, decision_url: str, page_number: int = 1) -> KvkkDocumentMarkdown:
"""Retrieve and convert a KVKK decision document to paginated Markdown."""
logger.info(f"KvkkApiClient: Getting decision document from: {decision_url}, page: {page_number}")
try:
# Fetch the decision page
response = await self.http_client.get(decision_url)
response.raise_for_status()
# Extract content from HTML
extracted_data = self._extract_decision_content_from_html(response.text, decision_url)
# Convert HTML content to Markdown
full_markdown_content = None
if extracted_data["html_content"]:
full_markdown_content = await asyncio.to_thread(self._convert_html_to_markdown, extracted_data["html_content"])
if not full_markdown_content:
return KvkkDocumentMarkdown(
source_url=HttpUrl(decision_url),
title=extracted_data["title"],
decision_date=extracted_data["decision_date"],
decision_number=extracted_data["decision_number"],
subject_summary=extracted_data["subject_summary"],
markdown_chunk=None,
current_page=page_number,
total_pages=0,
is_paginated=False,
error_message="Could not convert document content to Markdown"
)
# Calculate pagination
content_length = len(full_markdown_content)
total_pages = math.ceil(content_length / self.DOCUMENT_MARKDOWN_CHUNK_SIZE)
if total_pages == 0:
total_pages = 1
# Clamp page number to valid range
current_page_clamped = max(1, min(page_number, total_pages))
# Extract the requested chunk
start_index = (current_page_clamped - 1) * self.DOCUMENT_MARKDOWN_CHUNK_SIZE
end_index = start_index + self.DOCUMENT_MARKDOWN_CHUNK_SIZE
markdown_chunk = full_markdown_content[start_index:end_index]
return KvkkDocumentMarkdown(
source_url=HttpUrl(decision_url),
title=extracted_data["title"],
decision_date=extracted_data["decision_date"],
decision_number=extracted_data["decision_number"],
subject_summary=extracted_data["subject_summary"],
markdown_chunk=markdown_chunk,
current_page=current_page_clamped,
total_pages=total_pages,
is_paginated=(total_pages > 1),
error_message=None
)
except httpx.HTTPStatusError as e:
error_msg = f"HTTP error {e.response.status_code} when fetching decision document"
logger.error(f"KvkkApiClient: {error_msg}")
return KvkkDocumentMarkdown(
source_url=HttpUrl(decision_url),
title=None,
decision_date=None,
decision_number=None,
subject_summary=None,
markdown_chunk=None,
current_page=page_number,
total_pages=0,
is_paginated=False,
error_message=error_msg
)
except Exception as e:
error_msg = f"Unexpected error when fetching decision document: {str(e)}"
logger.error(f"KvkkApiClient: {error_msg}")
return KvkkDocumentMarkdown(
source_url=HttpUrl(decision_url),
title=None,
decision_date=None,
decision_number=None,
subject_summary=None,
markdown_chunk=None,
current_page=page_number,
total_pages=0,
is_paginated=False,
error_message=error_msg
)
async def close_client_session(self):
"""Close the HTTP client session."""
if hasattr(self, 'http_client') and self.http_client and not self.http_client.is_closed:
await self.http_client.aclose()
logger.info("KvkkApiClient: HTTP client session closed.")