diff --git a/anayasa_mcp_module/api_client.py b/anayasa_mcp_module/api_client.py new file mode 100644 index 0000000..f6f3209 --- /dev/null +++ b/anayasa_mcp_module/api_client.py @@ -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": "", "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_fragment}' + + 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:" 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=&type=) 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.") diff --git a/anayasa_mcp_module/bireysel_client.py b/anayasa_mcp_module/bireysel_client.py index 676c76b..8a35d26 100644 --- a/anayasa_mcp_module/bireysel_client.py +++ b/anayasa_mcp_module/bireysel_client.py @@ -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_input_for_markdown}" - 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.") \ No newline at end of file + await self.api.close() + logger.info("AnayasaBireyselBasvuruApiClient: HTTP client session closed.") diff --git a/anayasa_mcp_module/client.py b/anayasa_mcp_module/client.py index 22aec73..dc27ce5 100644 --- a/anayasa_mcp_module/client.py +++ b/anayasa_mcp_module/client.py @@ -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_input_for_markdown}" - 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

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.") \ No newline at end of file + await self.api.close() + logger.info("AnayasaMahkemesiApiClient (Norm Denetimi): HTTP client session closed.") diff --git a/anayasa_mcp_module/models.py b/anayasa_mcp_module/models.py index f5e74e1..eb06295 100644 --- a/anayasa_mcp_module/models.py +++ b/anayasa_mcp_module/models.py @@ -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.""" diff --git a/anayasa_mcp_module/unified_client.py b/anayasa_mcp_module/unified_client.py index 26543e1..5eb28f5 100644 --- a/anayasa_mcp_module/unified_client.py +++ b/anayasa_mcp_module/unified_client.py @@ -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() \ No newline at end of file + await self.norm_client.close_client_session() + await self.bireysel_client.close_client_session() diff --git a/mcp_server_main.py b/mcp_server_main.py index 1c5762c..9b78f9b 100644 --- a/mcp_server_main.py +++ b/mcp_server_main.py @@ -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 diff --git a/uv.lock b/uv.lock index c88a96a..efd9fa8 100644 --- a/uv.lock +++ b/uv.lock @@ -2245,7 +2245,7 @@ wheels = [ [[package]] name = "yargi-mcp" -version = "0.2.0" +version = "0.2.1" source = { editable = "." } dependencies = [ { name = "aiohttp" }, diff --git a/uyusmazlik_mcp_module/client.py b/uyusmazlik_mcp_module/client.py index ca2ffd0..890b250 100644 --- a/uyusmazlik_mcp_module/client.py +++ b/uyusmazlik_mcp_module/client.py @@ -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 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.") \ No newline at end of file diff --git a/uyusmazlik_mcp_module/models.py b/uyusmazlik_mcp_module/models.py index ac6c008..ac61b98 100644 --- a/uyusmazlik_mcp_module/models.py +++ b/uyusmazlik_mcp_module/models.py @@ -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.") \ No newline at end of file + """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.")