ayrisdev/yargi-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(aym,uyusmazlik): adapt to rebuilt AYM and Uyuşmazlık sites

Browse Source 
Both sites were rebuilt and their old endpoints now 404:
- AYM moved to a single-page app backed by a JSON API
  (POST /api/core/public/search, kararTipi NormDenetimi/BireyselBasvuru;
  full text via {id, size:1} -> "icerik" HTML). Old /Ara, /ND/, /BB/ gone.
- Uyuşmazlık moved to ASP.NET WebForms (viewstate postback to /, GridView
  results, decisions served as /Uploads/{EsasNo}.pdf). Old /Arama/Search gone.

Changes:
- New anayasa_mcp_module/api_client.py: shared KBB JSON client, base64url
  document-id codec, HTML->Markdown + HTML text stripping helpers.
- Rewrite anayasa client/bireysel_client/unified_client over the new API,
  preserving public method names and response models.
- Rewrite uyusmazlik client for the postback flow + GridView parse + PDF
  document conversion; simplify request model to text + scope + paging.
- Trim search_anayasa_unified and search_uyusmazlik_decisions tool signatures
  to parameters the new APIs actually support; drop dead enums.

Verified live via FastMCP client: search + document retrieval work for AYM
norm/bireysel and Uyuşmazlık.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
saidsurucu
2026-06-30 19:22:49 +03:00
co-authored by Claude Opus 4.8
parent 6bbc656dc6
commit fadc3b0bc0
9 changed files with 634 additions and 1116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
anayasa_mcp_module/api_client.py
+201
View File
@@ -0,0 +1,201 @@
# anayasa_mcp_module/api_client.py
# Low-level client for the new Anayasa Mahkemesi "Kararlar Bilgi Bankası" (KBB) JSON API.
#
# Both the Norm Denetimi host (normkararlarbilgibankasi.anayasa.gov.tr) and the
# Bireysel Başvuru host (kararlarbilgibankasi.anayasa.gov.tr) share the SAME
# backend, exposed at POST /api/core/public/search. The request differs only by
# the "kararTipi" discriminator:
#
# {"kararTipi": "NormDenetimi", "query": "mülkiyet", "page": 1, "size": 10}
# -> {"total": N, "page": 1, "data": [...summary records...], "page_size": 10}
#
# {"kararTipi": "NormDenetimi", "id": "<uuid>", "page": 1, "size": 1}
# -> data[0] additionally includes "icerik" = full decision HTML
#
# The previous HTML-scraping endpoints (/Ara, /ND/.., /BB/..) were retired when
# the sites were rebuilt as a single-page app; they now return HTTP 404.
import base64
import html as html_module
import io
import logging
import re
from typing import Any, Dict, Optional, Tuple
from urllib.parse import urlparse, parse_qs, quote
import httpx
from bs4 import BeautifulSoup
from markitdown import MarkItDown
logger = logging.getLogger(__name__)
# Markdown pagination chunk size (characters), shared across AYM document tools.
DOCUMENT_MARKDOWN_CHUNK_SIZE = 5000
def strip_html_text(value: Optional[str]) -> str:
"""Return plain text from a possibly-HTML field (e.g. kararKonusu)."""
if not value:
return ""
text = BeautifulSoup(html_module.unescape(value), "html.parser").get_text(" ", strip=True)
return re.sub(r"\s+", " ", text).strip()
def convert_icerik_to_markdown(icerik_html: Optional[str]) -> Optional[str]:
"""Convert the "icerik" decision HTML returned by the KBB API to Markdown.
The icerik field is a self-contained HTML fragment (the rendered decision
body). Scripts/styles are stripped before handing it to MarkItDown.
"""
if not icerik_html:
return None
processed_html = html_module.unescape(icerik_html)
soup = BeautifulSoup(processed_html, "html.parser")
for tag in soup.find_all(["script", "style"]):
tag.decompose()
body = soup.find("body")
html_fragment = str(body) if body else str(soup)
if not html_fragment.strip().lower().startswith(("<html", "<!doctype")):
html_fragment = f'<html><head><meta charset="UTF-8"></head><body>{html_fragment}</body></html>'
try:
html_stream = io.BytesIO(html_fragment.encode("utf-8"))
conversion_result = MarkItDown().convert(html_stream)
return conversion_result.text_content
except Exception as e: # pragma: no cover - defensive
logger.error("AnayasaApiClient: MarkItDown conversion error: %s", e)
return None
# kararTipi discriminator values accepted by the API.
KARAR_TIPI_NORM = "NormDenetimi"
KARAR_TIPI_BIREYSEL = "BireyselBasvuru"
NORM_HOST = "https://normkararlarbilgibankasi.anayasa.gov.tr"
BIREYSEL_HOST = "https://kararlarbilgibankasi.anayasa.gov.tr"
SEARCH_PATH = "/api/core/public/search"
# Map kararTipi -> the host whose SPA can display the decision (cosmetic only;
# either host's API answers for any kararTipi).
_HOST_FOR_TIPI = {
KARAR_TIPI_NORM: NORM_HOST,
KARAR_TIPI_BIREYSEL: BIREYSEL_HOST,
}
def encode_document_token(uuid: str) -> str:
"""Encode a raw decision UUID into the base64url token the SPA uses in its URLs.
The SPA addresses decisions as base64url("kbb:" + uuid) (no padding).
"""
raw = f"kbb:{uuid}".encode("utf-8")
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
def decode_document_token(token: str) -> Optional[str]:
"""Decode a base64url SPA token back into the raw decision UUID.
Returns None if the token is not a valid "kbb:<uuid>" token.
"""
try:
padded = token + "=" * (-len(token) % 4)
decoded = base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8")
except Exception:
return None
if decoded.startswith("kbb:"):
return decoded[len("kbb:"):]
return None
def build_document_url(karar_tipi: str, uuid: str) -> str:
"""Build a clickable SPA URL for a decision, used as its document_url."""
host = _HOST_FOR_TIPI.get(karar_tipi, BIREYSEL_HOST)
token = encode_document_token(uuid)
return f"{host}/kbb/pages/search/{karar_tipi}?id={quote(token)}&type={karar_tipi}"
def parse_document_url(document_url: str) -> Tuple[Optional[str], Optional[str]]:
"""Extract (karar_tipi, uuid) from a document URL.
Handles the new SPA URLs (?id=<token>&type=<kararTipi>) and is lenient about
older /ND/ and /BB/ style paths so historical references still resolve.
Returns (None, None) if neither the type nor id can be determined.
"""
parsed = urlparse(document_url)
qs = parse_qs(parsed.query)
karar_tipi = None
type_param = qs.get("type", [None])[0]
path = parsed.path or ""
if type_param in (KARAR_TIPI_NORM, KARAR_TIPI_BIREYSEL):
karar_tipi = type_param
elif "/ND/" in path or "NormDenetimi" in path:
karar_tipi = KARAR_TIPI_NORM
elif "/BB/" in path or "BireyselBasvuru" in path:
karar_tipi = KARAR_TIPI_BIREYSEL
uuid = None
id_param = qs.get("id", [None])[0]
if id_param:
# The id may be the raw uuid or the base64url SPA token.
uuid = decode_document_token(id_param) or id_param
return karar_tipi, uuid
class AnayasaApiClient:
"""Thin async wrapper around the KBB /api/core/public/search endpoint."""
def __init__(self, request_timeout: float = 60.0):
self.http_client = httpx.AsyncClient(
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"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 _search_url(self, karar_tipi: str) -> str:
host = _HOST_FOR_TIPI.get(karar_tipi, BIREYSEL_HOST)
return f"{host}{SEARCH_PATH}"
async def search(
self,
karar_tipi: str,
query: str = "",
page: int = 1,
size: int = 10,
) -> Dict[str, Any]:
"""Run a list search and return the parsed JSON envelope.
Envelope shape: {"total": int, "page": int, "data": [..], "page_size": int}.
"""
body: Dict[str, Any] = {"kararTipi": karar_tipi, "page": page, "size": size}
if query:
body["query"] = query
logger.info("AnayasaApiClient: search kararTipi=%s query=%r page=%s size=%s",
karar_tipi, query, page, size)
response = await self.http_client.post(self._search_url(karar_tipi), json=body)
response.raise_for_status()
return response.json()
async def get_decision(self, karar_tipi: str, uuid: str) -> Optional[Dict[str, Any]]:
"""Fetch a single decision record (including the "icerik" HTML) by UUID."""
body = {"kararTipi": karar_tipi, "id": uuid, "page": 1, "size": 1}
logger.info("AnayasaApiClient: get_decision kararTipi=%s id=%s", karar_tipi, uuid)
response = await self.http_client.post(self._search_url(karar_tipi), json=body)
response.raise_for_status()
payload = response.json()
data = payload.get("data") or []
return data[0] if data else None
async def close(self):
if self.http_client and not self.http_client.is_closed:
await self.http_client.aclose()
logger.info("AnayasaApiClient: HTTP client session closed.")
anayasa_mcp_module/bireysel_client.py
+83 -317
View File
@@ -1,24 +1,27 @@
# anayasa_mcp_module/bireysel_client.py
# This client is for Bireysel Başvuru: https://kararlarbilgibankasi.anayasa.gov.tr
# Bireysel Başvuru client backed by the new KBB JSON API (see api_client.py).
#
# Same backend as Norm Denetimi, distinguished by kararTipi="BireyselBasvuru".
# The legacy /Ara report-scraping endpoint was retired and now returns HTTP 404.
import asyncio
import httpx
from bs4 import BeautifulSoup, Tag
from typing import Dict, Any, List, Optional, Tuple
import logging
import html
import re
import io
from urllib.parse import urlencode, urljoin, quote
from markitdown import MarkItDown
import math # For math.ceil for pagination
import math
from typing import List, Optional
from .api_client import (
AnayasaApiClient,
KARAR_TIPI_BIREYSEL,
DOCUMENT_MARKDOWN_CHUNK_SIZE,
build_document_url,
parse_document_url,
convert_icerik_to_markdown,
strip_html_text,
)
from .models import (
AnayasaBireyselReportSearchRequest,
AnayasaBireyselReportDecisionDetail,
AnayasaBireyselReportDecisionSummary,
AnayasaBireyselReportSearchResult,
AnayasaBireyselBasvuruDocumentMarkdown, # Model for Bireysel Başvuru document
AnayasaBireyselBasvuruDocumentMarkdown,
)
logger = logging.getLogger(__name__)
@@ -27,330 +30,93 @@ if not logger.hasHandlers():
class AnayasaBireyselBasvuruApiClient:
BASE_URL = "https://kararlarbilgibankasi.anayasa.gov.tr"
SEARCH_PATH = "/Ara"
DOCUMENT_MARKDOWN_CHUNK_SIZE = 5000 # Character limit per page
"""Bireysel Başvuru search/document client over the KBB JSON API."""
def __init__(self, request_timeout: float = 60.0):
self.http_client = httpx.AsyncClient(
base_url=self.BASE_URL,
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 _build_query_params_for_bireysel_report(self, params: AnayasaBireyselReportSearchRequest) -> List[Tuple[str, str]]:
query_params: List[Tuple[str, str]] = []
query_params.append(("KararBulteni", "1")) # Specific to this report type
if params.keywords:
for kw in params.keywords:
query_params.append(("KelimeAra[]", kw))
if params.page_to_fetch and params.page_to_fetch > 1:
query_params.append(("page", str(params.page_to_fetch)))
return query_params
self.api = AnayasaApiClient(request_timeout)
async def search_bireysel_basvuru_report(
self,
params: AnayasaBireyselReportSearchRequest
params: AnayasaBireyselReportSearchRequest,
) -> AnayasaBireyselReportSearchResult:
final_query_params = self._build_query_params_for_bireysel_report(params)
request_url = self.SEARCH_PATH
logger.info(f"AnayasaBireyselBasvuruApiClient: Performing Bireysel Başvuru Report search. Path: {request_url}, Params: {final_query_params}")
query = " ".join(t for t in (params.keywords or []) if t).strip()
payload = await self.api.search(
karar_tipi=KARAR_TIPI_BIREYSEL,
query=query,
page=params.page_to_fetch,
size=getattr(params, "results_per_page", 10),
)
try:
response = await self.http_client.get(request_url, params=final_query_params)
response.raise_for_status()
html_content = response.text
except httpx.RequestError as e:
logger.error(f"AnayasaBireyselBasvuruApiClient: HTTP request error during Bireysel Başvuru Report search: {e}")
raise
except Exception as e:
logger.error(f"AnayasaBireyselBasvuruApiClient: Error processing Bireysel Başvuru Report search request: {e}")
raise
soup = BeautifulSoup(html_content, 'html.parser')
total_records = None
bulunan_karar_div = soup.find("div", class_="bulunankararsayisi")
if bulunan_karar_div:
match_records = re.search(r'(\d+)\s*Karar Bulundu', bulunan_karar_div.get_text(strip=True))
if match_records:
total_records = int(match_records.group(1))
processed_decisions: List[AnayasaBireyselReportDecisionSummary] = []
report_content_area = soup.find("div", class_="HaberBulteni")
if not report_content_area:
logger.warning("HaberBulteni div not found, attempting to parse decision divs from the whole page.")
report_content_area = soup
decision_divs = report_content_area.find_all("div", class_="KararBulteniBirKarar")
if not decision_divs:
logger.warning("No KararBulteniBirKarar divs found.")
for decision_div in decision_divs:
title_tag = decision_div.find("h4")
title_text = title_tag.get_text(strip=True) if title_tag and title_tag.strong else (title_tag.get_text(strip=True) if title_tag else "")
alti_cizili_div = decision_div.find("div", class_="AltiCizili")
ref_no, dec_type, body, app_date, dec_date, url_path = "", "", "", "", "", ""
if alti_cizili_div:
link_tag = alti_cizili_div.find("a", href=True)
if link_tag:
ref_no = link_tag.get_text(strip=True)
url_path = link_tag['href']
parts_text = alti_cizili_div.get_text(separator="|", strip=True)
parts = [part.strip() for part in parts_text.split("|")]
# Clean ref_no from the first part if it was extracted from link
if ref_no and parts and parts[0].strip().startswith(ref_no):
parts[0] = parts[0].replace(ref_no, "").strip()
if not parts[0]: parts.pop(0) # Remove empty string if ref_no was the only content
# Assign parts based on typical order, adjusting for missing ref_no at start
current_idx = 0
if not ref_no and len(parts) > current_idx and re.match(r"\d+/\d+", parts[current_idx]): # Check if first part is ref_no
ref_no = parts[current_idx]
current_idx += 1
dec_type = parts[current_idx] if len(parts) > current_idx else ""
current_idx += 1
body = parts[current_idx] if len(parts) > current_idx else ""
current_idx += 1
app_date_raw = parts[current_idx] if len(parts) > current_idx else ""
current_idx += 1
dec_date_raw = parts[current_idx] if len(parts) > current_idx else ""
if app_date_raw and "Başvuru Tarihi :" in app_date_raw:
app_date = app_date_raw.replace("Başvuru Tarihi :", "").strip()
elif app_date_raw: # If label is missing but format matches
app_date_match = re.search(r'(\d{1,2}/\d{1,2}/\d{4})', app_date_raw)
if app_date_match: app_date = app_date_match.group(1)
if dec_date_raw and "Karar Tarihi :" in dec_date_raw:
dec_date = dec_date_raw.replace("Karar Tarihi :", "").strip()
elif dec_date_raw: # If label is missing but format matches
dec_date_match = re.search(r'(\d{1,2}/\d{1,2}/\d{4})', dec_date_raw)
if dec_date_match: dec_date = dec_date_match.group(1)
subject_div = decision_div.find(lambda tag: tag.name == 'div' and not tag.has_attr('class') and tag.get_text(strip=True).startswith("BAŞVURU KONUSU :"))
subject_text = subject_div.get_text(strip=True).replace("BAŞVURU KONUSU :", "").strip() if subject_div else ""
details_list: List[AnayasaBireyselReportDecisionDetail] = []
karar_detaylari_div = decision_div.find_next_sibling("div", id="KararDetaylari") # Corrected: was KararDetaylari
if karar_detaylari_div:
table = karar_detaylari_div.find("table", class_="table")
if table and table.find("tbody"):
for row in table.find("tbody").find_all("tr"):
cells = row.find_all("td")
if len(cells) == 4: # Hak, Müdahale İddiası, Sonuç, Giderim
details_list.append(AnayasaBireyselReportDecisionDetail(
hak=cells[0].get_text(strip=True) or "",
mudahale_iddiasi=cells[1].get_text(strip=True) or "",
sonuc=cells[2].get_text(strip=True) or "",
giderim=cells[3].get_text(strip=True) or "",
))
full_decision_page_url = urljoin(self.BASE_URL, url_path) if url_path else ""
processed_decisions.append(AnayasaBireyselReportDecisionSummary(
title=title_text,
decision_reference_no=ref_no,
decision_page_url=full_decision_page_url,
decision_type_summary=dec_type,
decision_making_body=body,
application_date_summary=app_date,
decision_date_summary=dec_date,
application_subject_summary=subject_text,
details=details_list
total_records = int(payload.get("total") or 0)
decisions: List[AnayasaBireyselReportDecisionSummary] = []
for item in payload.get("data") or []:
decisions.append(AnayasaBireyselReportDecisionSummary(
title=item.get("basvuruAdi") or "",
decision_reference_no=item.get("basvuruNo") or "",
decision_page_url=build_document_url(KARAR_TIPI_BIREYSEL, item.get("id", "")),
decision_type_summary=item.get("kararTuruBasvuruSonucuLabel") or "",
decision_making_body=item.get("kararVerenBirimLabel") or "",
application_date_summary=item.get("basvuruTarihi") or "",
decision_date_summary=item.get("kararTarihi") or "",
application_subject_summary=strip_html_text(item.get("kararKonusu")),
details=[],
))
return AnayasaBireyselReportSearchResult(
decisions=processed_decisions,
decisions=decisions,
total_records_found=total_records,
retrieved_page_number=params.page_to_fetch
retrieved_page_number=params.page_to_fetch,
)
def _convert_html_to_markdown_bireysel(self, full_decision_html_content: str) -> Optional[str]:
if not full_decision_html_content:
return None
processed_html = html.unescape(full_decision_html_content)
soup = BeautifulSoup(processed_html, "html.parser")
html_input_for_markdown = ""
karar_tab_content = soup.find("div", id="Karar")
if karar_tab_content:
karar_html_span = karar_tab_content.find("span", class_="kararHtml")
if karar_html_span:
word_section = karar_html_span.find("div", class_="WordSection1")
if word_section:
for s in word_section.select('script, style, .item.col-xs-12.col-sm-12, center:has(b)'):
s.decompose()
html_input_for_markdown = str(word_section)
else:
logger.warning("AnayasaBireyselBasvuruApiClient: WordSection1 not found in span.kararHtml. Using span.kararHtml content.")
for s in karar_html_span.select('script, style, .item.col-xs-12.col-sm-12, center:has(b)'):
s.decompose()
html_input_for_markdown = str(karar_html_span)
else:
logger.warning("AnayasaBireyselBasvuruApiClient: span.kararHtml not found in div#Karar. Using div#Karar content.")
for s in karar_tab_content.select('script, style, .item.col-xs-12.col-sm-12, center:has(b)'):
s.decompose()
html_input_for_markdown = str(karar_tab_content)
else:
logger.warning("AnayasaBireyselBasvuruApiClient: div#Karar (KARAR tab) not found. Trying WordSection1 fallback.")
word_section_fallback = soup.find("div", class_="WordSection1")
if word_section_fallback:
for s in word_section_fallback.select('script, style, .item.col-xs-12.col-sm-12, center:has(b)'):
s.decompose()
html_input_for_markdown = str(word_section_fallback)
else:
body_tag = soup.find("body")
if body_tag:
for s in body_tag.select('script, style, .item.col-xs-12.col-sm-12, center:has(b), .banner, .footer, .yazdirmaalani, .filtreler, .menu, .altmenu, .geri, .arabuton, .temizlebutonu, form#KararGetir, .TabBaslik, #KararDetaylari, .share-button-container'):
s.decompose()
html_input_for_markdown = str(body_tag)
else:
html_input_for_markdown = processed_html
markdown_text = None
try:
# Ensure the content is wrapped in basic HTML structure if it's not already
if not html_input_for_markdown.strip().lower().startswith(("<html", "<!doctype")):
html_content = f"<html><head><meta charset=\"UTF-8\"></head><body>{html_input_for_markdown}</body></html>"
else:
html_content = html_input_for_markdown
# 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()
conversion_result = md_converter.convert(html_stream)
markdown_text = conversion_result.text_content
except Exception as e:
logger.error(f"AnayasaBireyselBasvuruApiClient: MarkItDown conversion error: {e}")
return markdown_text
async def get_decision_document_as_markdown(
self,
document_url_path: str, # e.g. /BB/2021/20295
page_number: int = 1
document_url_path: str,
page_number: int = 1,
) -> AnayasaBireyselBasvuruDocumentMarkdown:
full_url = urljoin(self.BASE_URL, document_url_path)
logger.info(f"AnayasaBireyselBasvuruApiClient: Fetching Bireysel Başvuru document for Markdown (page {page_number}) from URL: {full_url}")
karar_tipi, uuid = parse_document_url(document_url_path)
if karar_tipi is None:
karar_tipi = KARAR_TIPI_BIREYSEL
basvuru_no_from_page = None
karar_tarihi_from_page = None
basvuru_tarihi_from_page = None
karari_veren_birim_from_page = None
karar_turu_from_page = None
resmi_gazete_info_from_page = None
try:
response = await self.http_client.get(full_url)
response.raise_for_status()
html_content_from_api = response.text
if not isinstance(html_content_from_api, str) or not html_content_from_api.strip():
logger.warning(f"AnayasaBireyselBasvuruApiClient: Received empty HTML from {full_url}.")
return AnayasaBireyselBasvuruDocumentMarkdown(
source_url=full_url, markdown_chunk=None, current_page=page_number, total_pages=0, is_paginated=False
)
soup = BeautifulSoup(html_content_from_api, 'html.parser')
meta_desc_tag = soup.find("meta", attrs={"name": "description"})
if meta_desc_tag and meta_desc_tag.get("content"):
content = meta_desc_tag["content"]
bn_match = re.search(r"B\.\s*No:\s*([\d\/]+)", content)
if bn_match: basvuru_no_from_page = bn_match.group(1).strip()
date_match = re.search(r"(\d{1,2}\/\d{1,2}\/\d{4}),\s*§", content)
if date_match: karar_tarihi_from_page = date_match.group(1).strip()
karar_detaylari_tab = soup.find("div", id="KararDetaylari")
if karar_detaylari_tab:
table = karar_detaylari_tab.find("table", class_="table")
if table:
rows = table.find_all("tr")
for row in rows:
cells = row.find_all("td")
if len(cells) == 2:
key = cells[0].get_text(strip=True)
value = cells[1].get_text(strip=True)
if "Kararı Veren Birim" in key: karari_veren_birim_from_page = value
elif "Karar Türü (Başvuru Sonucu)" in key: karar_turu_from_page = value
elif "Başvuru No" in key and not basvuru_no_from_page: basvuru_no_from_page = value
elif "Başvuru Tarihi" in key: basvuru_tarihi_from_page = value
elif "Karar Tarihi" in key and not karar_tarihi_from_page: karar_tarihi_from_page = value
elif "Resmi Gazete Tarih / Sayı" in key: resmi_gazete_info_from_page = value
full_markdown_content = await asyncio.to_thread(self._convert_html_to_markdown_bireysel, html_content_from_api)
if not full_markdown_content:
return AnayasaBireyselBasvuruDocumentMarkdown(
source_url=full_url,
basvuru_no_from_page=basvuru_no_from_page,
karar_tarihi_from_page=karar_tarihi_from_page,
basvuru_tarihi_from_page=basvuru_tarihi_from_page,
karari_veren_birim_from_page=karari_veren_birim_from_page,
karar_turu_from_page=karar_turu_from_page,
resmi_gazete_info_from_page=resmi_gazete_info_from_page,
markdown_chunk=None,
current_page=page_number,
total_pages=0,
is_paginated=False
)
content_length = len(full_markdown_content)
total_pages = math.ceil(content_length / self.DOCUMENT_MARKDOWN_CHUNK_SIZE)
if total_pages == 0: total_pages = 1
current_page_clamped = max(1, min(page_number, total_pages))
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]
record = await self.api.get_decision(karar_tipi, uuid) if uuid else None
if not record:
logger.warning("AnayasaBireyselBasvuruApiClient: No record for %s", document_url_path)
return AnayasaBireyselBasvuruDocumentMarkdown(
source_url=full_url,
basvuru_no_from_page=basvuru_no_from_page,
karar_tarihi_from_page=karar_tarihi_from_page,
basvuru_tarihi_from_page=basvuru_tarihi_from_page,
karari_veren_birim_from_page=karari_veren_birim_from_page,
karar_turu_from_page=karar_turu_from_page,
resmi_gazete_info_from_page=resmi_gazete_info_from_page,
markdown_chunk=markdown_chunk,
current_page=current_page_clamped,
total_pages=total_pages,
is_paginated=(total_pages > 1)
source_url=document_url_path, markdown_chunk=None,
current_page=page_number, total_pages=0, is_paginated=False,
)
except httpx.RequestError as e:
logger.error(f"AnayasaBireyselBasvuruApiClient: HTTP error fetching Bireysel Başvuru document from {full_url}: {e}")
raise
except Exception as e:
logger.error(f"AnayasaBireyselBasvuruApiClient: General error processing Bireysel Başvuru document from {full_url}: {e}")
raise
rg_tarihi = record.get("resmiGazeteTarihi") or ""
rg_sayisi = record.get("resmiGazeteSayisi")
official_gazette = f"{rg_tarihi} / {rg_sayisi}".strip(" /") if (rg_tarihi or rg_sayisi) else None
full_markdown = convert_icerik_to_markdown(record.get("icerik"))
common = dict(
source_url=document_url_path,
basvuru_no_from_page=record.get("basvuruNo"),
karar_tarihi_from_page=record.get("kararTarihi"),
basvuru_tarihi_from_page=record.get("basvuruTarihi"),
karari_veren_birim_from_page=record.get("kararVerenBirimLabel"),
karar_turu_from_page=record.get("kararTuruBasvuruSonucuLabel"),
resmi_gazete_info_from_page=official_gazette,
)
if not full_markdown:
return AnayasaBireyselBasvuruDocumentMarkdown(
**common, markdown_chunk=None, current_page=page_number,
total_pages=0, is_paginated=False,
)
total_pages = max(1, math.ceil(len(full_markdown) / DOCUMENT_MARKDOWN_CHUNK_SIZE))
current_page = max(1, min(page_number, total_pages))
start = (current_page - 1) * DOCUMENT_MARKDOWN_CHUNK_SIZE
chunk = full_markdown[start:start + DOCUMENT_MARKDOWN_CHUNK_SIZE]
return AnayasaBireyselBasvuruDocumentMarkdown(
**common, markdown_chunk=chunk, current_page=current_page,
total_pages=total_pages, is_paginated=(total_pages > 1),
)
async def close_client_session(self):
if hasattr(self, 'http_client') and self.http_client and not self.http_client.is_closed:
await self.http_client.aclose()
logger.info("AnayasaBireyselBasvuruApiClient: HTTP client session closed.")
await self.api.close()
logger.info("AnayasaBireyselBasvuruApiClient: HTTP client session closed.")
anayasa_mcp_module/client.py
+111 -318
View File
@@ -1,357 +1,150 @@
# anayasa_mcp_module/client.py
# This client is for Norm Denetimi: https://normkararlarbilgibankasi.anayasa.gov.tr
# Norm Denetimi client backed by the new KBB JSON API (see api_client.py).
#
# The Anayasa Mahkemesi sites were rebuilt as a single-page app; the old
# HTML-scraping endpoints on normkararlarbilgibankasi.anayasa.gov.tr/Ara now
# return HTTP 404. This client maps the rich legacy request model onto the new
# free-text "query" search and rebuilds the legacy response models from the JSON
# payload so existing tooling keeps working.
import asyncio
import httpx
from bs4 import BeautifulSoup
from typing import Dict, Any, List, Optional, Tuple
import logging
import html
import re
import io
from urllib.parse import urlencode, urljoin, quote
from markitdown import MarkItDown
import math # For math.ceil for pagination
import math
from typing import List, Optional
from .api_client import (
AnayasaApiClient,
KARAR_TIPI_NORM,
DOCUMENT_MARKDOWN_CHUNK_SIZE,
build_document_url,
parse_document_url,
convert_icerik_to_markdown,
strip_html_text,
)
from .models import (
AnayasaNormDenetimiSearchRequest,
AnayasaDecisionSummary,
AnayasaReviewedNormInfo,
AnayasaSearchResult,
AnayasaDocumentMarkdown, # Model for Norm Denetimi document
AnayasaDocumentMarkdown,
)
logger = logging.getLogger(__name__)
if not logger.hasHandlers():
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
def _build_query(params: AnayasaNormDenetimiSearchRequest) -> str:
"""Derive the free-text query string the new API expects from the legacy model.
The new endpoint only supports a single full-text "query" field, so the
keyword lists are flattened. Esas/Karar numbers are appended when no keyword
is provided so number-based lookups still return results.
"""
terms: List[str] = []
for bucket in (params.keywords_all, params.keywords_any):
if bucket:
terms.extend(t for t in bucket if t)
if not terms:
for value in (params.case_number_esas, params.decision_number_karar):
if value:
terms.append(value)
return " ".join(terms).strip()
class AnayasaMahkemesiApiClient:
BASE_URL = "https://normkararlarbilgibankasi.anayasa.gov.tr"
SEARCH_PATH_SEGMENT = "Ara"
DOCUMENT_MARKDOWN_CHUNK_SIZE = 5000 # Character limit per page
"""Norm Denetimi search/document client over the KBB JSON API."""
def __init__(self, request_timeout: float = 60.0):
self.http_client = httpx.AsyncClient(
base_url=self.BASE_URL,
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 _build_search_query_params_for_aym(self, params: AnayasaNormDenetimiSearchRequest) -> List[Tuple[str, str]]:
query_params: List[Tuple[str, str]] = []
if params.keywords_all:
for kw in params.keywords_all: query_params.append(("KelimeAra[]", kw))
if params.keywords_any:
for kw in params.keywords_any: query_params.append(("HerhangiBirKelimeAra[]", kw))
if params.keywords_exclude:
for kw in params.keywords_exclude: query_params.append(("BulunmayanKelimeAra[]", kw))
if params.period and params.period and params.period != "ALL": query_params.append(("Donemler_id", params.period))
if params.case_number_esas: query_params.append(("EsasNo", params.case_number_esas))
if params.decision_number_karar: query_params.append(("KararNo", params.decision_number_karar))
if params.first_review_date_start: query_params.append(("IlkIncelemeTarihiIlk", params.first_review_date_start))
if params.first_review_date_end: query_params.append(("IlkIncelemeTarihiSon", params.first_review_date_end))
if params.decision_date_start: query_params.append(("KararTarihiIlk", params.decision_date_start))
if params.decision_date_end: query_params.append(("KararTarihiSon", params.decision_date_end))
if params.application_type and params.application_type and params.application_type != "ALL": query_params.append(("BasvuruTurler_id", params.application_type))
if params.applicant_general_name: query_params.append(("BasvuranGeneller_id", params.applicant_general_name))
if params.applicant_specific_name: query_params.append(("BasvuranOzeller_id", params.applicant_specific_name))
if params.attending_members_names:
for name in params.attending_members_names: query_params.append(("Uyeler_id[]", name))
if params.rapporteur_name: query_params.append(("Raportorler_id", params.rapporteur_name))
if params.norm_type and params.norm_type and params.norm_type != "ALL": query_params.append(("NormunTurler_id", params.norm_type))
if params.norm_id_or_name: query_params.append(("NormunNumarasiAdlar_id", params.norm_id_or_name))
if params.norm_article: query_params.append(("NormunMaddeNumarasi", params.norm_article))
if params.review_outcomes:
for outcome_val in params.review_outcomes:
if outcome_val and outcome_val != "ALL": query_params.append(("IncelemeTuruKararSonuclar_id[]", outcome_val))
if params.reason_for_final_outcome and params.reason_for_final_outcome and params.reason_for_final_outcome != "ALL":
query_params.append(("KararSonucununGerekcesi", params.reason_for_final_outcome))
if params.basis_constitution_article_numbers:
for article_no in params.basis_constitution_article_numbers: query_params.append(("DayanakHukmu[]", article_no))
if params.official_gazette_date_start: query_params.append(("ResmiGazeteTarihiIlk", params.official_gazette_date_start))
if params.official_gazette_date_end: query_params.append(("ResmiGazeteTarihiSon", params.official_gazette_date_end))
if params.official_gazette_number_start: query_params.append(("ResmiGazeteSayisiIlk", params.official_gazette_number_start))
if params.official_gazette_number_end: query_params.append(("ResmiGazeteSayisiSon", params.official_gazette_number_end))
if params.has_press_release and params.has_press_release and params.has_press_release != "ALL": query_params.append(("BasinDuyurusu", params.has_press_release))
if params.has_dissenting_opinion and params.has_dissenting_opinion and params.has_dissenting_opinion != "ALL": query_params.append(("KarsiOy", params.has_dissenting_opinion))
if params.has_different_reasoning and params.has_different_reasoning and params.has_different_reasoning != "ALL": query_params.append(("FarkliGerekce", params.has_different_reasoning))
# Add pagination and sorting parameters as query params instead of URL path
if params.results_per_page and params.results_per_page != 10:
query_params.append(("SatirSayisi", str(params.results_per_page)))
if params.sort_by_criteria and params.sort_by_criteria != "KararTarihi":
query_params.append(("Siralama", params.sort_by_criteria))
if params.page_to_fetch and params.page_to_fetch > 1:
query_params.append(("page", str(params.page_to_fetch)))
return query_params
self.api = AnayasaApiClient(request_timeout)
async def search_norm_denetimi_decisions(
self,
params: AnayasaNormDenetimiSearchRequest
params: AnayasaNormDenetimiSearchRequest,
) -> AnayasaSearchResult:
# Use simple /Ara endpoint - the complex path structure seems to cause 404s
request_path = f"/{self.SEARCH_PATH_SEGMENT}"
final_query_params = self._build_search_query_params_for_aym(params)
logger.info(f"AnayasaMahkemesiApiClient: Performing Norm Denetimi search. Path: {request_path}, Params: {final_query_params}")
query = _build_query(params)
payload = await self.api.search(
karar_tipi=KARAR_TIPI_NORM,
query=query,
page=params.page_to_fetch,
size=params.results_per_page,
)
try:
response = await self.http_client.get(request_path, params=final_query_params)
response.raise_for_status()
html_content = response.text
except httpx.RequestError as e:
logger.error(f"AnayasaMahkemesiApiClient: HTTP request error during Norm Denetimi search: {e}")
raise
except Exception as e:
logger.error(f"AnayasaMahkemesiApiClient: Error processing Norm Denetimi search request: {e}")
raise
soup = BeautifulSoup(html_content, 'html.parser')
total_records = None
bulunan_karar_div = soup.find("div", class_="bulunankararsayisi")
if not bulunan_karar_div: # Fallback for mobile view
bulunan_karar_div = soup.find("div", class_="bulunankararsayisiMobil")
if bulunan_karar_div:
match_records = re.search(r'(\d+)\s*Karar Bulundu', bulunan_karar_div.get_text(strip=True))
if match_records:
total_records = int(match_records.group(1))
processed_decisions: List[AnayasaDecisionSummary] = []
decision_divs = soup.find_all("div", class_="birkarar")
for decision_div in decision_divs:
link_tag = decision_div.find("a", href=True)
doc_url_path = link_tag['href'] if link_tag else None
decision_page_url_str = urljoin(self.BASE_URL, doc_url_path) if doc_url_path else None
title_div = decision_div.find("div", class_="bkararbaslik")
ek_no_text_raw = title_div.get_text(strip=True, separator=" ").replace('\xa0', ' ') if title_div else ""
ek_no_match = re.search(r"(E\.\s*\d+/\d+\s*,\s*K\.\s*\d+/\d+)", ek_no_text_raw)
ek_no_text = ek_no_match.group(1) if ek_no_match else ek_no_text_raw.split("Sayılı Karar")[0].strip()
keyword_count_div = title_div.find("div", class_="BulunanKelimeSayisi") if title_div else None
keyword_count_text = keyword_count_div.get_text(strip=True).replace("Bulunan Kelime Sayısı", "").strip() if keyword_count_div else None
keyword_count = int(keyword_count_text) if keyword_count_text and keyword_count_text.isdigit() else None
info_div = decision_div.find("div", class_="kararbilgileri")
info_parts = [part.strip() for part in info_div.get_text(separator="|").split("|")] if info_div else []
app_type_summary = info_parts[0] if len(info_parts) > 0 else None
applicant_summary = info_parts[1] if len(info_parts) > 1 else None
outcome_summary = info_parts[2] if len(info_parts) > 2 else None
dec_date_raw = info_parts[3] if len(info_parts) > 3 else None
decision_date_summary = dec_date_raw.replace("Karar Tarihi:", "").strip() if dec_date_raw else None
reviewed_norms_list: List[AnayasaReviewedNormInfo] = []
details_table_container = decision_div.find_next_sibling("div", class_=re.compile(r"col-sm-12")) # The details table is in a sibling div
if details_table_container:
details_table = details_table_container.find("table", class_="table")
if details_table and details_table.find("tbody"):
for row in details_table.find("tbody").find_all("tr"):
cells = row.find_all("td")
if len(cells) == 6:
reviewed_norms_list.append(AnayasaReviewedNormInfo(
norm_name_or_number=cells[0].get_text(strip=True) or None,
article_number=cells[1].get_text(strip=True) or None,
review_type_and_outcome=cells[2].get_text(strip=True) or None,
outcome_reason=cells[3].get_text(strip=True) or None,
basis_constitution_articles_cited=[a.strip() for a in cells[4].get_text(strip=True).split(',') if a.strip()] if cells[4].get_text(strip=True) else [],
postponement_period=cells[5].get_text(strip=True) or None
))
processed_decisions.append(AnayasaDecisionSummary(
decision_reference_no=ek_no_text,
decision_page_url=decision_page_url_str,
keywords_found_count=keyword_count,
application_type_summary=app_type_summary,
applicant_summary=applicant_summary,
decision_outcome_summary=outcome_summary,
decision_date_summary=decision_date_summary,
reviewed_norms=reviewed_norms_list
total_records = int(payload.get("total") or 0)
decisions: List[AnayasaDecisionSummary] = []
for item in payload.get("data") or []:
esas_no = item.get("esasNo") or ""
karar_no = item.get("kararNo") or ""
if esas_no and karar_no:
reference = f"E.{esas_no}, K.{karar_no}"
else:
reference = esas_no or karar_no or ""
decisions.append(AnayasaDecisionSummary(
decision_reference_no=reference,
decision_page_url=build_document_url(KARAR_TIPI_NORM, item.get("id", "")),
keywords_found_count=item.get("highlightCount") or 0,
application_type_summary=item.get("basvuruTuruLabel") or "",
applicant_summary=item.get("basvuranGenelLabel") or "",
decision_outcome_summary=strip_html_text(item.get("kararKonusu")),
decision_date_summary=item.get("kararTarihi") or "",
reviewed_norms=[],
))
return AnayasaSearchResult(
decisions=processed_decisions,
decisions=decisions,
total_records_found=total_records,
retrieved_page_number=params.page_to_fetch
retrieved_page_number=params.page_to_fetch,
)
def _convert_html_to_markdown_norm_denetimi(self, full_decision_html_content: str) -> Optional[str]:
"""Converts direct HTML content from an Anayasa Mahkemesi Norm Denetimi decision page to Markdown."""
if not full_decision_html_content:
return None
processed_html = html.unescape(full_decision_html_content)
soup = BeautifulSoup(processed_html, "html.parser")
html_input_for_markdown = ""
karar_tab_content = soup.find("div", id="Karar") # "KARAR" tab content
if karar_tab_content:
karar_metni_div = karar_tab_content.find("div", class_="KararMetni")
if karar_metni_div:
# Remove scripts and styles
for script_tag in karar_metni_div.find_all("script"): script_tag.decompose()
for style_tag in karar_metni_div.find_all("style"): style_tag.decompose()
# Remove "Künye Kopyala" button and other non-content divs
for item_div in karar_metni_div.find_all("div", class_="item col-sm-12"): item_div.decompose()
for modal_div in karar_metni_div.find_all("div", class_="modal fade"): modal_div.decompose() # If any modals
word_section = karar_metni_div.find("div", class_="WordSection1")
html_input_for_markdown = str(word_section) if word_section else str(karar_metni_div)
else:
html_input_for_markdown = str(karar_tab_content)
else:
# Fallback if specific structure is not found
word_section_fallback = soup.find("div", class_="WordSection1")
if word_section_fallback:
html_input_for_markdown = str(word_section_fallback)
else:
# Last resort: use the whole body or the raw HTML
body_tag = soup.find("body")
html_input_for_markdown = str(body_tag) if body_tag else processed_html
markdown_text = None
try:
# Ensure the content is wrapped in basic HTML structure if it's not already
if not html_input_for_markdown.strip().lower().startswith(("<html", "<!doctype")):
html_content = f"<html><head><meta charset=\"UTF-8\"></head><body>{html_input_for_markdown}</body></html>"
else:
html_content = html_input_for_markdown
# 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()
conversion_result = md_converter.convert(html_stream)
markdown_text = conversion_result.text_content
except Exception as e:
logger.error(f"AnayasaMahkemesiApiClient: MarkItDown conversion error: {e}")
return markdown_text
async def get_decision_document_as_markdown(
self,
document_url: str,
page_number: int = 1
page_number: int = 1,
) -> AnayasaDocumentMarkdown:
"""
Retrieves a specific Anayasa Mahkemesi (Norm Denetimi) decision,
converts its content to Markdown, and returns the requested page/chunk.
"""
full_url = urljoin(self.BASE_URL, document_url) if not document_url.startswith("http") else document_url
logger.info(f"AnayasaMahkemesiApiClient: Fetching Norm Denetimi document for Markdown (page {page_number}) from URL: {full_url}")
karar_tipi, uuid = parse_document_url(document_url)
if karar_tipi is None:
karar_tipi = KARAR_TIPI_NORM
decision_ek_no_from_page = None
decision_date_from_page = None
official_gazette_from_page = None
try:
# Use a new client instance for document fetching if headers/timeout needs to be different,
# or reuse self.http_client if settings are compatible. For now, self.http_client.
get_response = await self.http_client.get(full_url, headers={"Accept": "text/html"})
get_response.raise_for_status()
html_content_from_api = get_response.text
if not isinstance(html_content_from_api, str) or not html_content_from_api.strip():
logger.warning(f"AnayasaMahkemesiApiClient: Received empty or non-string HTML from URL {full_url}.")
return AnayasaDocumentMarkdown(
source_url=full_url, markdown_chunk=None, current_page=page_number, total_pages=0, is_paginated=False
)
# Extract metadata from the page content (E.K. No, Date, RG)
soup = BeautifulSoup(html_content_from_api, "html.parser")
karar_metni_div = soup.find("div", class_="KararMetni") # Usually within div#Karar
if not karar_metni_div: # Fallback if not in KararMetni
karar_metni_div = soup.find("div", class_="WordSection1")
# Initialize with empty string defaults
decision_ek_no_from_page = ""
decision_date_from_page = ""
official_gazette_from_page = ""
if karar_metni_div:
# Attempt to find E.K. No (Esas No, Karar No)
# Norm Denetimi pages often have this in bold <p> tags directly or in the WordSection1
# Look for patterns like "Esas No.: YYYY/NN" and "Karar No.: YYYY/NN"
esas_no_tag = karar_metni_div.find(lambda tag: tag.name == "p" and tag.find("b") and "Esas No.:" in tag.find("b").get_text())
karar_no_tag = karar_metni_div.find(lambda tag: tag.name == "p" and tag.find("b") and "Karar No.:" in tag.find("b").get_text())
karar_tarihi_tag = karar_metni_div.find(lambda tag: tag.name == "p" and tag.find("b") and "Karar tarihi:" in tag.find("b").get_text()) # Less common on Norm pages
resmi_gazete_tag = karar_metni_div.find(lambda tag: tag.name == "p" and ("Resmî Gazete tarih ve sayısı:" in tag.get_text() or "Resmi Gazete tarih/sayı:" in tag.get_text()))
if esas_no_tag and esas_no_tag.find("b") and karar_no_tag and karar_no_tag.find("b"):
esas_str = esas_no_tag.find("b").get_text(strip=True).replace('Esas No.:', '').strip()
karar_str = karar_no_tag.find("b").get_text(strip=True).replace('Karar No.:', '').strip()
decision_ek_no_from_page = f"E.{esas_str}, K.{karar_str}"
if karar_tarihi_tag and karar_tarihi_tag.find("b"):
decision_date_from_page = karar_tarihi_tag.find("b").get_text(strip=True).replace("Karar tarihi:", "").strip()
elif karar_metni_div: # Fallback for Karar Tarihi if not in specific tag
date_match = re.search(r"Karar Tarihi\s*:\s*([\d\.]+)", karar_metni_div.get_text()) # Norm pages often use DD.MM.YYYY
if date_match: decision_date_from_page = date_match.group(1).strip()
if resmi_gazete_tag:
# Try to get the bold part first if it exists
bold_rg_tag = resmi_gazete_tag.find("b")
rg_text_content = bold_rg_tag.get_text(strip=True) if bold_rg_tag else resmi_gazete_tag.get_text(strip=True)
official_gazette_from_page = rg_text_content.replace("Resmî Gazete tarih ve sayısı:", "").replace("Resmi Gazete tarih/sayı:", "").strip()
full_markdown_content = await asyncio.to_thread(self._convert_html_to_markdown_norm_denetimi, html_content_from_api)
if not full_markdown_content:
return AnayasaDocumentMarkdown(
source_url=full_url,
decision_reference_no_from_page=decision_ek_no_from_page,
decision_date_from_page=decision_date_from_page,
official_gazette_info_from_page=official_gazette_from_page,
markdown_chunk=None,
current_page=page_number,
total_pages=0,
is_paginated=False
)
content_length = len(full_markdown_content)
total_pages = math.ceil(content_length / self.DOCUMENT_MARKDOWN_CHUNK_SIZE)
if total_pages == 0: total_pages = 1
current_page_clamped = max(1, min(page_number, total_pages))
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]
record = await self.api.get_decision(karar_tipi, uuid) if uuid else None
if not record:
logger.warning("AnayasaMahkemesiApiClient: No record for document_url %s", document_url)
return AnayasaDocumentMarkdown(
source_url=full_url,
decision_reference_no_from_page=decision_ek_no_from_page,
decision_date_from_page=decision_date_from_page,
official_gazette_info_from_page=official_gazette_from_page,
markdown_chunk=markdown_chunk,
current_page=current_page_clamped,
total_pages=total_pages,
is_paginated=(total_pages > 1)
source_url=document_url, markdown_chunk=None,
current_page=page_number, total_pages=0, is_paginated=False,
)
except httpx.RequestError as e:
logger.error(f"AnayasaMahkemesiApiClient: HTTP error fetching Norm Denetimi document from {full_url}: {e}")
raise
except Exception as e:
logger.error(f"AnayasaMahkemesiApiClient: General error processing Norm Denetimi document from {full_url}: {e}")
raise
esas_no = record.get("esasNo") or ""
karar_no = record.get("kararNo") or ""
reference = f"E.{esas_no}, K.{karar_no}" if (esas_no and karar_no) else (esas_no or karar_no or "")
rg_tarihi = record.get("resmiGazeteTarihi") or ""
rg_sayisi = record.get("resmiGazeteSayisi")
official_gazette = f"{rg_tarihi} / {rg_sayisi}".strip(" /") if (rg_tarihi or rg_sayisi) else ""
full_markdown = convert_icerik_to_markdown(record.get("icerik"))
if not full_markdown:
return AnayasaDocumentMarkdown(
source_url=document_url,
decision_reference_no_from_page=reference,
decision_date_from_page=record.get("kararTarihi") or "",
official_gazette_info_from_page=official_gazette,
markdown_chunk=None, current_page=page_number, total_pages=0, is_paginated=False,
)
total_pages = max(1, math.ceil(len(full_markdown) / DOCUMENT_MARKDOWN_CHUNK_SIZE))
current_page = max(1, min(page_number, total_pages))
start = (current_page - 1) * DOCUMENT_MARKDOWN_CHUNK_SIZE
chunk = full_markdown[start:start + DOCUMENT_MARKDOWN_CHUNK_SIZE]
return AnayasaDocumentMarkdown(
source_url=document_url,
decision_reference_no_from_page=reference,
decision_date_from_page=record.get("kararTarihi") or "",
official_gazette_info_from_page=official_gazette,
markdown_chunk=chunk,
current_page=current_page,
total_pages=total_pages,
is_paginated=(total_pages > 1),
)
async def close_client_session(self):
if hasattr(self, 'http_client') and self.http_client and not self.http_client.is_closed:
await self.http_client.aclose()
logger.info("AnayasaMahkemesiApiClient (Norm Denetimi): HTTP client session closed.")
await self.api.close()
logger.info("AnayasaMahkemesiApiClient (Norm Denetimi): HTTP client session closed.")
anayasa_mcp_module/models.py
+12 -18
View File
@@ -140,9 +140,10 @@ class AnayasaDocumentMarkdown(BaseModel):
# --- Models for Anayasa Mahkemesi - Bireysel Başvuru Karar Raporu ---
class AnayasaBireyselReportSearchRequest(BaseModel):
"""Model for Anayasa Mahkemesi (Bireysel Başvuru) 'Karar Arama Raporu' search request."""
keywords: Optional[List[str]] = Field(default_factory=list, description="Keywords for AND logic (KelimeAra[]).")
"""Model for Anayasa Mahkemesi (Bireysel Başvuru) search request."""
keywords: Optional[List[str]] = Field(default_factory=list, description="Keywords joined into the full-text query.")
page_to_fetch: int = Field(1, ge=1, description="Page number to fetch for the report (page). Default is 1.")
results_per_page: int = Field(10, ge=1, le=100, description="Results per page.")
class AnayasaBireyselReportDecisionDetail(BaseModel):
"""Details of a specific right/claim within a Bireysel Başvuru decision summary in a report."""
@@ -191,26 +192,19 @@ class AnayasaBireyselBasvuruDocumentMarkdown(BaseModel):
# --- Unified Models ---
class AnayasaUnifiedSearchRequest(BaseModel):
"""Unified search request for both Norm Denetimi and Bireysel Başvuru."""
"""Unified search request for both Norm Denetimi and Bireysel Başvuru.
The KBB API only exposes a single free-text "query" field plus pagination,
so the keyword lists below are flattened into that query.
"""
decision_type: Literal["norm_denetimi", "bireysel_basvuru"] = Field(..., description="Decision type: norm_denetimi or bireysel_basvuru")
# Common parameters
keywords: List[str] = Field(default_factory=list, description="Keywords to search for")
keywords: List[str] = Field(default_factory=list, description="Keywords to search for (joined into a single full-text query)")
keywords_all: List[str] = Field(default_factory=list, description="Additional keywords to include in the query")
keywords_any: List[str] = Field(default_factory=list, description="Additional alternative keywords to include in the query")
page_to_fetch: int = Field(1, ge=1, le=100, description="Page number to fetch (1-100)")
results_per_page: int = Field(10, ge=1, le=100, description="Results per page (1-100)")
# Norm Denetimi specific parameters (ignored for bireysel_basvuru)
keywords_all: List[str] = Field(default_factory=list, description="All keywords must be present (norm_denetimi only)")
keywords_any: List[str] = Field(default_factory=list, description="Any of these keywords (norm_denetimi only)")
decision_type_norm: Literal["ALL", "1", "2", "3"] = Field("ALL", description="Decision type for norm denetimi")
application_date_start: str = Field("", description="Application start date (norm_denetimi only)")
application_date_end: str = Field("", description="Application end date (norm_denetimi only)")
# Bireysel Başvuru specific parameters (ignored for norm_denetimi)
decision_start_date: str = Field("", description="Decision start date (bireysel_basvuru only)")
decision_end_date: str = Field("", description="Decision end date (bireysel_basvuru only)")
norm_type: Literal["ALL", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "0"] = Field("ALL", description="Norm type (bireysel_basvuru only)")
subject_category: str = Field("", description="Subject category (bireysel_basvuru only)")
class AnayasaUnifiedSearchResult(BaseModel):
"""Unified search result containing decisions from either system."""
anayasa_mcp_module/unified_client.py
+57 -111
View File
@@ -1,172 +1,118 @@
# anayasa_mcp_module/unified_client.py
# Unified client for both Norm Denetimi and Bireysel Başvuru
# Unified client for both Norm Denetimi and Bireysel Başvuru, backed by the new
# KBB JSON API. Routing between the two is by the "decision_type" discriminator
# on search, and by the document URL (?type=...) on document retrieval.
import logging
from typing import Optional, Tuple
from urllib.parse import urlparse, urlunparse
from .models import (
AnayasaUnifiedSearchRequest,
AnayasaUnifiedSearchResult,
AnayasaUnifiedSearchResult,
AnayasaUnifiedDocumentMarkdown,
# Removed AnayasaDecisionTypeEnum - now using string literals
AnayasaNormDenetimiSearchRequest,
AnayasaBireyselReportSearchRequest
AnayasaBireyselReportSearchRequest,
)
from .client import AnayasaMahkemesiApiClient
from .bireysel_client import AnayasaBireyselBasvuruApiClient
from .api_client import (
KARAR_TIPI_NORM,
KARAR_TIPI_BIREYSEL,
parse_document_url,
)
logger = logging.getLogger(__name__)
# Canonical hosts per decision type. Norm Denetimi (/ND/) documents live on the
# "norm" subdomain; Bireysel Başvuru (/BB/) documents on the plain subdomain.
# Callers (or upstream search links) sometimes supply the wrong host for a given
# path, which makes the AYM server return 404. We re-key the host off the path.
_NORM_HOST = "normkararlarbilgibankasi.anayasa.gov.tr"
_BIREYSEL_HOST = "kararlarbilgibankasi.anayasa.gov.tr"
def normalize_anayasa_document_url(document_url: str) -> Tuple[Optional[str], str]:
"""Detect the AYM decision type from the URL path and force the correct host.
"""Detect the AYM decision type from a document URL.
Detection is path-based (``/ND/`` vs ``/BB/``) because the path is
unambiguous, whereas the supplied host may be wrong. Query params and
fragment are preserved (they are harmless for document fetches).
Returns ``(decision_type, normalized_url)`` where ``decision_type`` is
Returns ``(decision_type, document_url)`` where ``decision_type`` is
``"norm_denetimi"``, ``"bireysel_basvuru"``, or ``None`` if it cannot be
determined (URL returned unchanged in that case).
determined. The URL is returned unchanged (kept for backwards compatibility
with callers that expect a possibly-normalized URL).
"""
parsed = urlparse(document_url)
path = parsed.path or ""
if "/ND/" in path:
decision_type, host = "norm_denetimi", _NORM_HOST
elif "/BB/" in path:
decision_type, host = "bireysel_basvuru", _BIREYSEL_HOST
else:
# Fall back to host-based detection when the path is uninformative.
if "normkararlarbilgibankasi" in parsed.netloc:
return "norm_denetimi", document_url
if "kararlarbilgibankasi" in parsed.netloc:
return "bireysel_basvuru", document_url
return None, document_url
normalized = urlunparse((
parsed.scheme or "https",
host,
parsed.path,
parsed.params,
parsed.query,
parsed.fragment,
))
return decision_type, normalized
karar_tipi, _ = parse_document_url(document_url)
if karar_tipi == KARAR_TIPI_NORM:
return "norm_denetimi", document_url
if karar_tipi == KARAR_TIPI_BIREYSEL:
return "bireysel_basvuru", document_url
return None, document_url
class AnayasaUnifiedClient:
"""Unified client that handles both Norm Denetimi and Bireysel Başvuru searches."""
def __init__(self, request_timeout: float = 60.0):
self.norm_client = AnayasaMahkemesiApiClient(request_timeout)
self.bireysel_client = AnayasaBireyselBasvuruApiClient(request_timeout)
async def search_unified(self, params: AnayasaUnifiedSearchRequest) -> AnayasaUnifiedSearchResult:
"""Unified search that routes to appropriate client based on decision_type."""
"""Unified search that routes to the appropriate client based on decision_type."""
if params.decision_type == "norm_denetimi":
# Convert to norm denetimi request
norm_params = AnayasaNormDenetimiSearchRequest(
keywords_all=params.keywords_all or params.keywords,
keywords_any=params.keywords_any,
application_type=params.decision_type_norm,
page_to_fetch=params.page_to_fetch,
results_per_page=params.results_per_page
results_per_page=params.results_per_page,
)
result = await self.norm_client.search_norm_denetimi_decisions(norm_params)
# Convert to unified format
decisions_list = [decision.model_dump() for decision in result.decisions]
return AnayasaUnifiedSearchResult(
decision_type="norm_denetimi",
decisions=decisions_list,
decisions=[d.model_dump() for d in result.decisions],
total_records_found=result.total_records_found,
retrieved_page_number=result.retrieved_page_number
retrieved_page_number=result.retrieved_page_number,
)
elif params.decision_type == "bireysel_basvuru":
# Convert to bireysel başvuru request
bireysel_params = AnayasaBireyselReportSearchRequest(
keywords=params.keywords,
decision_start_date=params.decision_start_date,
decision_end_date=params.decision_end_date,
norm_type=params.norm_type,
subject_category=params.subject_category,
keywords=params.keywords or params.keywords_all,
page_to_fetch=params.page_to_fetch,
results_per_page=params.results_per_page
results_per_page=params.results_per_page,
)
result = await self.bireysel_client.search_bireysel_basvuru_report(bireysel_params)
# Convert to unified format
decisions_list = [decision.model_dump() for decision in result.decisions]
return AnayasaUnifiedSearchResult(
decision_type="bireysel_basvuru",
decisions=decisions_list,
decisions=[d.model_dump() for d in result.decisions],
total_records_found=result.total_records_found,
retrieved_page_number=result.retrieved_page_number
retrieved_page_number=result.retrieved_page_number,
)
else:
raise ValueError(f"Unsupported decision type: {params.decision_type}")
raise ValueError(f"Unsupported decision type: {params.decision_type}")
async def get_document_unified(self, document_url: str, page_number: int = 1) -> AnayasaUnifiedDocumentMarkdown:
"""Unified document retrieval that auto-detects the appropriate client."""
# Auto-detect decision type from the path and force the correct host.
# This repairs malformed URLs (e.g. a /ND/ path on the bireysel host),
# which otherwise 404 against the AYM server.
decision_type, normalized_url = normalize_anayasa_document_url(document_url)
if normalized_url != document_url:
logger.info(
f"AnayasaUnifiedClient: Normalized document URL "
f"'{document_url}' -> '{normalized_url}'"
)
"""Unified document retrieval that auto-detects the decision type from the URL."""
if decision_type == "norm_denetimi":
result = await self.norm_client.get_decision_document_as_markdown(normalized_url, page_number)
return AnayasaUnifiedDocumentMarkdown(
decision_type="norm_denetimi",
source_url=result.source_url,
document_data=result.model_dump(),
markdown_chunk=result.markdown_chunk,
current_page=result.current_page,
total_pages=result.total_pages,
is_paginated=result.is_paginated
)
elif decision_type == "bireysel_basvuru":
result = await self.bireysel_client.get_decision_document_as_markdown(normalized_url, page_number)
decision_type, _ = normalize_anayasa_document_url(document_url)
if decision_type == "bireysel_basvuru":
result = await self.bireysel_client.get_decision_document_as_markdown(document_url, page_number)
return AnayasaUnifiedDocumentMarkdown(
decision_type="bireysel_basvuru",
source_url=result.source_url,
document_data=result.model_dump(),
document_data=result.model_dump(mode="json"),
markdown_chunk=result.markdown_chunk,
current_page=result.current_page,
total_pages=result.total_pages,
is_paginated=result.is_paginated
is_paginated=result.is_paginated,
)
else:
raise ValueError(f"Cannot determine document type from URL: {document_url}")
# Default to norm_denetimi (also covers explicit norm_denetimi detection).
result = await self.norm_client.get_decision_document_as_markdown(document_url, page_number)
return AnayasaUnifiedDocumentMarkdown(
decision_type="norm_denetimi",
source_url=result.source_url,
document_data=result.model_dump(mode="json"),
markdown_chunk=result.markdown_chunk,
current_page=result.current_page,
total_pages=result.total_pages,
is_paginated=result.is_paginated,
)
async def close_client_session(self):
"""Close both client sessions."""
if hasattr(self.norm_client, 'close_client_session'):
await self.norm_client.close_client_session()
if hasattr(self.bireysel_client, 'close_client_session'):
await self.bireysel_client.close_client_session()
await self.norm_client.close_client_session()
await self.bireysel_client.close_client_session()
mcp_server_main.py
+17 -82
View File
@@ -285,7 +285,7 @@ from emsal_mcp_module.models import (
)
from uyusmazlik_mcp_module.client import UyusmazlikApiClient
from uyusmazlik_mcp_module.models import (
UyusmazlikSearchRequest, UyusmazlikBolumEnum, UyusmazlikTuruEnum, UyusmazlikKararSonucuEnum
UyusmazlikSearchRequest
)
from anayasa_mcp_module.client import AnayasaMahkemesiApiClient
from anayasa_mcp_module.bireysel_client import AnayasaBireyselBasvuruApiClient
@@ -696,65 +696,24 @@ async def get_emsal_document_markdown(id: str) -> Dict[str, Any]:
}
)
async def search_uyusmazlik_decisions(
icerik: str = Field("", description="Keyword or content for main text search."),
bolum: Literal["ALL", "Ceza Bölümü", "Genel Kurul Kararları", "Hukuk Bölümü"] = Field("ALL", description="Select the department (Bölüm). Use 'ALL' for all departments."),
uyusmazlik_turu: Literal["ALL", "Görev Uyuşmazlığı", "Hüküm Uyuşmazlığı"] = Field("ALL", description="Select the type of dispute. Use 'ALL' for all types."),
karar_sonuclari: List[Literal["Hüküm Uyuşmazlığı Olmadığına Dair", "Hüküm Uyuşmazlığı Olduğuna Dair"]] = Field(default_factory=list, description="List of desired 'Karar Sonucu' types."),
esas_yil: str = Field("", description="Case year ('Esas Yılı')."),
esas_sayisi: str = Field("", description="Case number ('Esas Sayısı')."),
karar_yil: str = Field("", description="Decision year ('Karar Yılı')."),
karar_sayisi: str = Field("", description="Decision number ('Karar Sayısı')."),
kanun_no: str = Field("", description="Relevant Law Number."),
karar_date_begin: str = Field("", description="Decision start date (DD.MM.YYYY)."),
karar_date_end: str = Field("", description="Decision end date (DD.MM.YYYY)."),
resmi_gazete_sayi: str = Field("", description="Official Gazette number."),
resmi_gazete_date: str = Field("", description="Official Gazette date (DD.MM.YYYY)."),
tumce: str = Field("", description="Exact phrase search."),
wild_card: str = Field("", description="Search for phrase and its inflections."),
hepsi: str = Field("", description="Search for texts containing all specified words."),
herhangi_birisi: str = Field("", description="Search for texts containing any of the specified words."),
not_hepsi: str = Field("", description="Exclude texts containing these specified words.")
icerik: str = Field("", description="Search text. Searches full decision text, or matches a case/decision number depending on search_scope."),
search_scope: Literal["All", "EsasNo", "KararNo"] = Field("All", description="Search scope: 'All' (full text), 'EsasNo' (by case number), 'KararNo' (by decision number)."),
case_sensitive: bool = Field(False, description="Whether the search is case sensitive."),
page_number: int = Field(1, ge=1, description="Result page number.")
) -> Dict[str, Any]:
"""Search Court of Jurisdictional Disputes decisions."""
# Convert string literals to enums
# Map "ALL" to TUMU for backward compatibility
if bolum == "ALL":
bolum_enum = UyusmazlikBolumEnum.TUMU
else:
bolum_enum = UyusmazlikBolumEnum(bolum) if bolum else UyusmazlikBolumEnum.TUMU
if uyusmazlik_turu == "ALL":
uyusmazlik_turu_enum = UyusmazlikTuruEnum.TUMU
else:
uyusmazlik_turu_enum = UyusmazlikTuruEnum(uyusmazlik_turu) if uyusmazlik_turu else UyusmazlikTuruEnum.TUMU
karar_sonuclari_enums = [UyusmazlikKararSonucuEnum(ks) for ks in karar_sonuclari]
"""Search Court of Jurisdictional Disputes (Uyuşmazlık Mahkemesi) decisions."""
search_params = UyusmazlikSearchRequest(
icerik=icerik,
bolum=bolum_enum,
uyusmazlik_turu=uyusmazlik_turu_enum,
karar_sonuclari=karar_sonuclari_enums,
esas_yil=esas_yil,
esas_sayisi=esas_sayisi,
karar_yil=karar_yil,
karar_sayisi=karar_sayisi,
kanun_no=kanun_no,
karar_date_begin=karar_date_begin,
karar_date_end=karar_date_end,
resmi_gazete_sayi=resmi_gazete_sayi,
resmi_gazete_date=resmi_gazete_date,
tumce=tumce,
wild_card=wild_card,
hepsi=hepsi,
herhangi_birisi=herhangi_birisi,
not_hepsi=not_hepsi
search_scope=search_scope,
case_sensitive=case_sensitive,
page_number=page_number,
)
logger.info("Tool 'search_uyusmazlik_decisions' called.")
try:
result = await uyusmazlik_client_instance.search_decisions(search_params)
return result.model_dump()
return result.model_dump(mode="json")
except Exception:
logger.exception("Error in tool 'search_uyusmazlik_decisions'.")
raise
@@ -830,47 +789,23 @@ async def get_uyusmazlik_document_markdown_from_url(
)
async def search_anayasa_unified(
decision_type: Literal["norm_denetimi", "bireysel_basvuru"] = Field(..., description="Decision type: norm_denetimi (norm control) or bireysel_basvuru (individual applications)"),
keywords: List[str] = Field(default_factory=list, description="Keywords to search for (common parameter)"),
keywords: List[str] = Field(default_factory=list, description="Keywords for full-text search (joined into a single query)"),
page_to_fetch: int = Field(1, ge=1, le=100, description="Page number to fetch (1-100)"),
# results_per_page: int = Field(10, ge=1, le=100, description="Results per page (1-100)"),
# Norm Denetimi specific parameters (ignored for bireysel_basvuru)
keywords_all: List[str] = Field(default_factory=list, description="All keywords must be present (norm_denetimi only)"),
keywords_any: List[str] = Field(default_factory=list, description="Any of these keywords (norm_denetimi only)"),
decision_type_norm: Literal["ALL", "1", "2", "3"] = Field("ALL", description="Decision type for norm denetimi"),
application_date_start: str = Field("", description="Application start date (norm_denetimi only)"),
application_date_end: str = Field("", description="Application end date (norm_denetimi only)"),
# Bireysel Başvuru specific parameters (ignored for norm_denetimi)
decision_start_date: str = Field("", description="Decision start date (bireysel_basvuru only)"),
decision_end_date: str = Field("", description="Decision end date (bireysel_basvuru only)"),
norm_type: Literal["ALL", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "0"] = Field("ALL", description="Norm type (bireysel_basvuru only)"),
subject_category: str = Field("", description="Subject category (bireysel_basvuru only)")
results_per_page: int = Field(10, ge=1, le=100, description="Results per page (1-100)")
) -> str:
logger.info(f"Tool 'search_anayasa_unified' called for decision_type: {decision_type}")
results_per_page = 10 # Default value
try:
request = AnayasaUnifiedSearchRequest(
decision_type=decision_type,
keywords=keywords,
page_to_fetch=page_to_fetch,
results_per_page=results_per_page,
keywords_all=keywords_all,
keywords_any=keywords_any,
decision_type_norm=decision_type_norm,
application_date_start=application_date_start,
application_date_end=application_date_end,
decision_start_date=decision_start_date,
decision_end_date=decision_end_date,
norm_type=norm_type,
subject_category=subject_category
)
result = await anayasa_unified_client_instance.search_unified(request)
return json.dumps(result.model_dump(), ensure_ascii=False, indent=2)
except Exception:
logger.exception("Error in tool 'search_anayasa_unified'.")
raise
uv.lock
Generated
+1 -1
View File
@@ -2245,7 +2245,7 @@ wheels = [
[[package]]
name = "yargi-mcp"
version = "0.2.0"
version = "0.2.1"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
uyusmazlik_mcp_module/client.py
+125 -197
View File
@@ -1,251 +1,179 @@
# uyusmazlik_mcp_module/client.py
#
# Client for the rebuilt Uyuşmazlık Mahkemesi search site
# (https://kararlar.uyusmazlik.gov.tr). The site is an ASP.NET WebForms app:
# searching is a form postback against "/" that returns an HTML page with a
# GridView of results, and each decision is a PDF served from /Uploads/.
#
# The previous AJAX endpoint (/Arama/Search) was retired and now returns 404.
import asyncio
import io
import logging
import re
from typing import Dict, List, Optional
from urllib.parse import urljoin
import httpx
from bs4 import BeautifulSoup
from typing import Dict, Any, List, Optional, Union, Tuple
import logging
import html
import re
import io
from markitdown import MarkItDown
from urllib.parse import urljoin
from .models import (
UyusmazlikSearchRequest,
UyusmazlikApiDecisionEntry,
UyusmazlikSearchResponse,
UyusmazlikDocumentMarkdown,
UyusmazlikBolumEnum,
UyusmazlikTuruEnum,
UyusmazlikKararSonucuEnum
)
logger = logging.getLogger(__name__)
if not logger.hasHandlers():
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# --- Mappings from user-friendly Enum values to API IDs ---
BOLUM_ENUM_TO_ID_MAP = {
UyusmazlikBolumEnum.CEZA_BOLUMU: "f6b74320-f2d7-4209-ad6e-c6df180d4e7c",
UyusmazlikBolumEnum.GENEL_KURUL_KARARLARI: "e4ca658d-a75a-4719-b866-b2d2f1c3b1d9",
UyusmazlikBolumEnum.HUKUK_BOLUMU: "96b26fc4-ef8e-4a4f-a9cc-a3de89952aa1",
UyusmazlikBolumEnum.TUMU: "", # Represents "...Seçiniz..." or all - empty string for API
"ALL": "" # Also map the new "ALL" literal to empty string for backward compatibility
}
# ASP.NET hidden fields that must be round-tripped on every postback.
_HIDDEN_FIELDS = ("__VIEWSTATE", "__VIEWSTATEGENERATOR", "__EVENTVALIDATION")
UYUSMAZLIK_TURU_ENUM_TO_ID_MAP = {
UyusmazlikTuruEnum.GOREV_UYUSMAZLIGI: "7b1e2cd3-8f09-418a-921c-bbe501e1740c",
UyusmazlikTuruEnum.HUKUM_UYUSMAZLIGI: "19b88402-172b-4c1d-8339-595c942a89f5",
UyusmazlikTuruEnum.TUMU: "", # Represents "...Seçiniz..." or all - empty string for API
"ALL": "" # Also map the new "ALL" literal to empty string for backward compatibility
}
KARAR_SONUCU_ENUM_TO_ID_MAP = {
# These IDs are from the form HTML provided by the user
UyusmazlikKararSonucuEnum.HUKUM_UYUSMAZLIGI_OLMADIGINA_DAIR: "6f47d87f-dcb5-412e-9878-000385dba1d9",
UyusmazlikKararSonucuEnum.HUKUM_UYUSMAZLIGI_OLDUGUNA_DAIR: "5a01742a-c440-4c4a-ba1f-da20837cffed",
# Add all other 'Karar Sonucu' enum members and their corresponding GUIDs
# by inspecting the 'KararSonucuList' checkboxes in the provided form HTML.
}
# --- End Mappings ---
class UyusmazlikApiClient:
BASE_URL = "https://kararlar.uyusmazlik.gov.tr"
SEARCH_ENDPOINT = "/Arama/Search"
# Individual documents are fetched by their full URLs obtained from search results.
SEARCH_PATH = "/"
def __init__(self, request_timeout: float = 30.0):
self.request_timeout = request_timeout
# Create shared httpx client for all requests
# A persistent cookie-aware client so ASP.NET session/viewstate are kept.
self.http_client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "tr-TR,tr;q=0.9,en-US;q=0.8,en;q=0.7",
"X-Requested-With": "XMLHttpRequest",
"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",
"Origin": self.BASE_URL,
"Referer": self.BASE_URL + "/",
},
timeout=request_timeout,
verify=False
verify=False,
follow_redirects=True,
)
@staticmethod
def _extract_hidden_fields(html_content: str) -> Dict[str, str]:
soup = BeautifulSoup(html_content, "html.parser")
fields: Dict[str, str] = {}
for name in _HIDDEN_FIELDS:
tag = soup.find("input", attrs={"name": name})
fields[name] = tag["value"] if tag and tag.has_attr("value") else ""
return fields
async def search_decisions(
self,
params: UyusmazlikSearchRequest
) -> UyusmazlikSearchResponse:
bolum_id_for_api = BOLUM_ENUM_TO_ID_MAP.get(params.bolum, "")
uyusmazlik_id_for_api = UYUSMAZLIK_TURU_ENUM_TO_ID_MAP.get(params.uyusmazlik_turu, "")
form_data_list: List[Tuple[str, str]] = []
@staticmethod
def _parse_results(html_content: str, base_url: str) -> UyusmazlikSearchResponse:
soup = BeautifulSoup(html_content, "html.parser")
def add_to_form_data(key: str, value: Optional[str]):
# API expects empty strings for omitted optional fields based on user payload example
form_data_list.append((key, value or ""))
decisions: List[UyusmazlikApiDecisionEntry] = []
grid = soup.find("table", id="GridView1")
if grid:
rows = grid.find_all("tr")
for row in rows[1:]: # skip header row
cells = row.find_all("td")
if len(cells) < 4:
continue
# The İşlemler cell holds the PDF "Görüntüle" link. Pager rows also
# contain <a> tags (javascript:__doPostBack ...), so require a real
# document link and skip everything else.
link_tag = cells[3].find(
"a", href=lambda h: h and not h.strip().lower().startswith("javascript:")
)
if not link_tag:
continue
href = link_tag["href"].strip()
if "uploads" not in href.lower() and not href.lower().endswith(".pdf"):
continue
document_url = urljoin(base_url + "/", href)
decisions.append(UyusmazlikApiDecisionEntry(
esas_sayisi=cells[0].get_text(strip=True) or None,
karar_sayisi=cells[1].get_text(strip=True) or None,
karar_tarihi=cells[2].get_text(strip=True) or None,
document_url=document_url,
))
add_to_form_data("BolumId", bolum_id_for_api)
add_to_form_data("UyusmazlikId", uyusmazlik_id_for_api)
if params.karar_sonuclari:
for enum_member in params.karar_sonuclari:
api_id = KARAR_SONUCU_ENUM_TO_ID_MAP.get(enum_member)
if api_id: # Only add if a valid ID is found
form_data_list.append(('KararSonucuList', api_id))
add_to_form_data("EsasYil", params.esas_yil)
add_to_form_data("EsasSayisi", params.esas_sayisi)
add_to_form_data("KararYil", params.karar_yil)
add_to_form_data("KararSayisi", params.karar_sayisi)
add_to_form_data("KanunNo", params.kanun_no)
add_to_form_data("KararDateBegin", params.karar_date_begin)
add_to_form_data("KararDateEnd", params.karar_date_end)
add_to_form_data("ResmiGazeteSayi", params.resmi_gazete_sayi)
add_to_form_data("ResmiGazeteDate", params.resmi_gazete_date)
add_to_form_data("Icerik", params.icerik)
add_to_form_data("Tumce", params.tumce)
add_to_form_data("WildCard", params.wild_card)
add_to_form_data("Hepsi", params.hepsi)
add_to_form_data("Herhangibirisi", params.herhangi_birisi)
add_to_form_data("NotHepsi", params.not_hepsi)
# Try to read a "N kayıt/sonuç/karar bulundu" style count if present.
total_records: Optional[int] = None
count_match = re.search(r'(\d+)\s*(?:adet\s*)?(?:kayıt|sonuç|karar)\b', html_content, re.IGNORECASE)
if count_match:
total_records = int(count_match.group(1))
# Convert form data to dict for httpx
form_data_dict = {}
for key, value in form_data_list:
if key in form_data_dict:
# Handle multiple values (like KararSonucuList)
if not isinstance(form_data_dict[key], list):
form_data_dict[key] = [form_data_dict[key]]
form_data_dict[key].append(value)
else:
form_data_dict[key] = value
return UyusmazlikSearchResponse(decisions=decisions, total_records_found=total_records)
logger.info(f"UyusmazlikApiClient (httpx): Performing search to {self.SEARCH_ENDPOINT} with form_data: {form_data_dict}")
try:
# Use shared httpx client
response = await self.http_client.post(
self.SEARCH_ENDPOINT,
data=form_data_dict,
headers={"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
async def search_decisions(self, params: UyusmazlikSearchRequest) -> UyusmazlikSearchResponse:
# 1. Load the landing page to obtain a fresh viewstate + session cookie.
landing = await self.http_client.get(self.SEARCH_PATH)
landing.raise_for_status()
form_data = self._extract_hidden_fields(landing.text)
# 2. Submit the search form.
form_data.update({
"txtSearch": params.icerik or "",
"rblSearchScope": params.search_scope,
"btnSearch": "Ara",
})
if params.case_sensitive:
form_data["chkCaseSensitive"] = "on"
logger.info("UyusmazlikApiClient: search icerik=%r scope=%s page=%s",
params.icerik, params.search_scope, params.page_number)
response = await self.http_client.post(
self.SEARCH_PATH,
data=form_data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
html_content = response.text
# 3. Navigate the GridView pager if a later page is requested.
if params.page_number > 1:
page_fields = self._extract_hidden_fields(html_content)
page_fields.update({
"txtSearch": params.icerik or "",
"rblSearchScope": params.search_scope,
"__EVENTTARGET": "GridView1",
"__EVENTARGUMENT": f"Page${params.page_number}",
})
if params.case_sensitive:
page_fields["chkCaseSensitive"] = "on"
page_response = await self.http_client.post(
self.SEARCH_PATH,
data=page_fields,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
html_content = response.text
logger.debug("UyusmazlikApiClient (httpx): Received HTML response for search.")
except httpx.HTTPError as e:
logger.error(f"UyusmazlikApiClient (httpx): HTTP client error during search: {e}")
raise # Re-raise to be handled by the MCP tool
except Exception as e:
logger.error(f"UyusmazlikApiClient (httpx): Error processing search request: {e}")
raise
page_response.raise_for_status()
html_content = page_response.text
# --- HTML Parsing (remains the same as previous version) ---
soup = BeautifulSoup(html_content, 'html.parser')
total_records_text_div = soup.find("div", class_="pull-right label label-important")
total_records = None
if total_records_text_div:
match_records = re.search(r'(\d+)\s*adet kayıt bulundu', total_records_text_div.get_text(strip=True))
if match_records:
total_records = int(match_records.group(1))
result_table = soup.find("table", class_="table-hover")
processed_decisions: List[UyusmazlikApiDecisionEntry] = []
if result_table:
rows = result_table.find_all("tr")
if len(rows) > 1: # Skip header row
for row in rows[1:]:
cols = row.find_all('td')
if len(cols) >= 5:
try:
popover_div = cols[0].find("div", attrs={"data-rel": "popover"})
popover_content_raw = popover_div["data-content"] if popover_div and popover_div.has_attr("data-content") else None
link_tag = cols[0].find('a')
doc_relative_url = link_tag['href'] if link_tag and link_tag.has_attr('href') else None
if not doc_relative_url: continue
document_url_str = urljoin(self.BASE_URL, doc_relative_url)
return self._parse_results(html_content, self.BASE_URL)
pdf_link_tag = cols[5].find('a', href=re.compile(r'\.pdf$', re.IGNORECASE)) if len(cols) > 5 else None
pdf_url_str = urljoin(self.BASE_URL, pdf_link_tag['href']) if pdf_link_tag and pdf_link_tag.has_attr('href') else None
decision_data_parsed = {
"karar_sayisi": cols[0].get_text(strip=True),
"esas_sayisi": cols[1].get_text(strip=True),
"bolum": cols[2].get_text(strip=True),
"uyusmazlik_konusu": cols[3].get_text(strip=True),
"karar_sonucu": cols[4].get_text(strip=True),
"popover_content": html.unescape(popover_content_raw) if popover_content_raw else None,
"document_url": document_url_str,
"pdf_url": pdf_url_str
}
decision_model = UyusmazlikApiDecisionEntry(**decision_data_parsed)
processed_decisions.append(decision_model)
except Exception as e:
logger.warning(f"UyusmazlikApiClient: Could not parse decision row. Row content: {row.get_text(strip=True, separator=' | ')}, Error: {e}")
return UyusmazlikSearchResponse(
decisions=processed_decisions,
total_records_found=total_records
)
def _convert_html_to_markdown_uyusmazlik(self, full_decision_html_content: str) -> Optional[str]:
"""Converts direct HTML content (from an Uyuşmazlık decision page) to Markdown."""
if not full_decision_html_content:
return None
processed_html = html.unescape(full_decision_html_content)
# As per user request, pass the full (unescaped) HTML to MarkItDown
html_input_for_markdown = processed_html
markdown_text = None
def _convert_pdf_to_markdown(self, pdf_bytes: bytes) -> Optional[str]:
try:
# Convert HTML string to bytes and create BytesIO stream
html_bytes = html_input_for_markdown.encode('utf-8')
html_stream = io.BytesIO(html_bytes)
# Pass BytesIO stream to MarkItDown to avoid temp file creation
md_converter = MarkItDown()
conversion_result = md_converter.convert(html_stream)
markdown_text = conversion_result.text_content
logger.info("UyusmazlikApiClient: HTML to Markdown conversion successful.")
pdf_stream = io.BytesIO(pdf_bytes)
conversion_result = MarkItDown().convert(pdf_stream, file_extension=".pdf")
return conversion_result.text_content
except Exception as e:
logger.error(f"UyusmazlikApiClient: Error during MarkItDown HTML to Markdown conversion: {e}")
return markdown_text
logger.error("UyusmazlikApiClient: PDF to Markdown conversion error: %s", e)
return None
async def get_decision_document_as_markdown(self, document_url: str) -> UyusmazlikDocumentMarkdown:
"""
Retrieves a specific Uyuşmazlık decision from its full URL and returns content as Markdown.
"""
logger.info(f"UyusmazlikApiClient (httpx for docs): Fetching Uyuşmazlık document for Markdown from URL: {document_url}")
"""Fetch an Uyuşmazlık decision PDF and return its content as Markdown."""
logger.info("UyusmazlikApiClient: Fetching document PDF from %s", document_url)
try:
# Using a new httpx.AsyncClient instance for this GET request for simplicity
async with httpx.AsyncClient(verify=False, timeout=self.request_timeout) as doc_fetch_client:
get_response = await doc_fetch_client.get(document_url, headers={"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"})
get_response.raise_for_status()
html_content_from_api = get_response.text
if not isinstance(html_content_from_api, str) or not html_content_from_api.strip():
logger.warning(f"UyusmazlikApiClient: Received empty or non-string HTML from URL {document_url}.")
return UyusmazlikDocumentMarkdown(source_url=document_url, markdown_content=None)
markdown_content = await asyncio.to_thread(self._convert_html_to_markdown_uyusmazlik, html_content_from_api)
response = await self.http_client.get(
document_url,
headers={"Accept": "application/pdf,*/*"},
)
response.raise_for_status()
markdown_content = await asyncio.to_thread(self._convert_pdf_to_markdown, response.content)
return UyusmazlikDocumentMarkdown(source_url=document_url, markdown_content=markdown_content)
except httpx.RequestError as e:
logger.error(f"UyusmazlikApiClient (httpx for docs): HTTP error fetching Uyuşmazlık document from {document_url}: {e}")
raise
except Exception as e:
logger.error(f"UyusmazlikApiClient (httpx for docs): General error processing Uyuşmazlık document from {document_url}: {e}")
except httpx.HTTPError as e:
logger.error("UyusmazlikApiClient: HTTP error fetching document from %s: %s", document_url, e)
raise
async def close_client_session(self):
"""Close the shared httpx client session."""
if hasattr(self, 'http_client') and self.http_client:
if hasattr(self, "http_client") and self.http_client and not self.http_client.is_closed:
await self.http_client.aclose()
logger.info("UyusmazlikApiClient: HTTP client session closed.")
else:
logger.info("UyusmazlikApiClient: No persistent client session from __init__ to close.")
uyusmazlik_mcp_module/models.py
+27 -72
View File
@@ -1,86 +1,41 @@
# uyusmazlik_mcp_module/models.py
from pydantic import BaseModel, Field, HttpUrl
from typing import List, Optional
from enum import Enum
from typing import List, Optional, Literal
# Enum definitions for user-friendly input based on the provided HTML form
class UyusmazlikBolumEnum(str, Enum):
"""User-friendly names for 'BolumId'."""
TUMU = "ALL" # Represents "...Seçiniz..." or all
CEZA_BOLUMU = "Ceza Bölümü"
GENEL_KURUL_KARARLARI = "Genel Kurul Kararları"
HUKUK_BOLUMU = "Hukuk Bölümü"
# The Uyuşmazlık Mahkemesi search site was rebuilt as an ASP.NET WebForms app.
# It now offers only a single free-text search with a scope selector; the old
# Bölüm / Uyuşmazlık Türü / Karar Sonucu / Esas-Karar year filters no longer exist.
class UyusmazlikTuruEnum(str, Enum):
"""User-friendly names for 'UyusmazlikId'."""
TUMU = "ALL" # Represents "...Seçiniz..." or all
GOREV_UYUSMAZLIGI = "Görev Uyuşmazlığı"
HUKUM_UYUSMAZLIGI = "Hüküm Uyuşmazlığı"
UyusmazlikSearchScope = Literal["All", "EsasNo", "KararNo"]
class UyusmazlikKararSonucuEnum(str, Enum): # Based on checkbox text in the form
"""User-friendly names for 'KararSonucuList' items."""
HUKUM_UYUSMAZLIGI_OLMADIGINA_DAIR = "Hüküm Uyuşmazlığı Olmadığına Dair"
HUKUM_UYUSMAZLIGI_OLDUGUNA_DAIR = "Hüküm Uyuşmazlığı Olduğuna Dair"
# Add other "Karar Sonucu" options from the form's checkboxes as Enum members
# Example: GOREVLI_YARGI_YERI_ADLI = "Görevli Yargı Yeri Belirlenmesine Dair (Adli Yargı)"
# The client will map these enum values (which are strings) to their respective IDs.
class UyusmazlikSearchRequest(BaseModel): # This is the model the MCP tool will accept
"""Model for Uyuşmazlık Mahkemesi search request using user-friendly terms."""
icerik: Optional[str] = Field("", description="Search text")
bolum: Optional[UyusmazlikBolumEnum] = Field(
UyusmazlikBolumEnum.TUMU,
description="Department"
class UyusmazlikSearchRequest(BaseModel):
"""Model for the Uyuşmazlık Mahkemesi search request."""
icerik: str = Field("", description="Search text (txtSearch).")
search_scope: UyusmazlikSearchScope = Field(
"All",
description="Search scope: 'All' (full text), 'EsasNo' (by case number), 'KararNo' (by decision number).",
)
uyusmazlik_turu: Optional[UyusmazlikTuruEnum] = Field(
UyusmazlikTuruEnum.TUMU,
description="Dispute type"
)
# User provides a list of user-friendly names for Karar Sonucu
karar_sonuclari: Optional[List[UyusmazlikKararSonucuEnum]] = Field( # Changed to list of Enums
default_factory=list,
description="Decision types"
)
esas_yil: Optional[str] = Field("", description="Case year")
esas_sayisi: Optional[str] = Field("", description="Case no")
karar_yil: Optional[str] = Field("", description="Decision year")
karar_sayisi: Optional[str] = Field("", description="Decision no")
kanun_no: Optional[str] = Field("", description="Law no")
karar_date_begin: Optional[str] = Field("", description="Start date (DD.MM.YYYY)")
karar_date_end: Optional[str] = Field("", description="End date (DD.MM.YYYY)")
resmi_gazete_sayi: Optional[str] = Field("", description="Gazette no")
resmi_gazete_date: Optional[str] = Field("", description="Gazette date (DD.MM.YYYY)")
# Detailed text search fields from the "icerikDetail" section of the form
tumce: Optional[str] = Field("", description="Exact phrase")
wild_card: Optional[str] = Field("", description="Wildcard search")
hepsi: Optional[str] = Field("", description="All words")
herhangi_birisi: Optional[str] = Field("", description="Any word")
not_hepsi: Optional[str] = Field("", description="Exclude words")
case_sensitive: bool = Field(False, description="Whether the search is case sensitive (chkCaseSensitive).")
page_number: int = Field(1, ge=1, description="Result page number (GridView pager).")
class UyusmazlikApiDecisionEntry(BaseModel):
"""Model for an individual decision entry parsed from Uyuşmazlık API's HTML search response."""
karar_sayisi: Optional[str] = Field(None)
esas_sayisi: Optional[str] = Field(None)
bolum: Optional[str] = Field(None)
uyusmazlik_konusu: Optional[str] = Field(None)
karar_sonucu: Optional[str] = Field(None)
popover_content: Optional[str] = Field(None, description="Summary")
document_url: HttpUrl # Full URL to the decision document HTML page
pdf_url: Optional[HttpUrl] = Field(None, description="PDF URL")
"""A single decision row parsed from the Uyuşmazlık GridView results."""
esas_sayisi: Optional[str] = Field(None, description="Case number (Esas No).")
karar_sayisi: Optional[str] = Field(None, description="Decision number (Karar No).")
karar_tarihi: Optional[str] = Field(None, description="Decision date (DD/MM/YYYY).")
document_url: HttpUrl = Field(..., description="Full URL to the decision PDF document.")
class UyusmazlikSearchResponse(BaseModel): # This is what the MCP tool will return
"""Response model for Uyuşmazlık Mahkemesi search results for the MCP tool."""
class UyusmazlikSearchResponse(BaseModel):
"""Response model for Uyuşmazlık Mahkemesi search results."""
decisions: List[UyusmazlikApiDecisionEntry]
total_records_found: Optional[int] = Field(None, description="Total number of records found for the query, if available.")
total_records_found: Optional[int] = Field(None, description="Total number of records found, if reported.")
class UyusmazlikDocumentMarkdown(BaseModel):
"""Model for an Uyuşmazlık decision document, containing only Markdown content."""
source_url: HttpUrl # The URL from which the content was fetched
markdown_content: Optional[str] = Field(None, description="The decision content converted to Markdown.")
"""Model for an Uyuşmazlık decision document, containing Markdown content."""
source_url: HttpUrl
markdown_content: Optional[str] = Field(None, description="The decision PDF content converted to Markdown.")