# anayasa_mcp_module/client.py # This client is for Norm Denetimi: https://normkararlarbilgibankasi.anayasa.gov.tr import httpx from bs4 import BeautifulSoup from typing import List, Optional, Tuple import logging import html import re import io from urllib.parse import urljoin from markitdown import MarkItDown import math # For math.ceil for pagination from .models import ( AnayasaNormDenetimiSearchRequest, AnayasaDecisionSummary, AnayasaReviewedNormInfo, AnayasaSearchResult, AnayasaDocumentMarkdown, # Model for Norm Denetimi document ) logger = logging.getLogger(__name__) if not logger.hasHandlers(): logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') class AnayasaMahkemesiApiClient: BASE_URL = "https://normkararlarbilgibankasi.anayasa.gov.tr" SEARCH_PATH_SEGMENT = "Ara" DOCUMENT_MARKDOWN_CHUNK_SIZE = 5000 # Character limit per page 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 async def search_norm_denetimi_decisions( self, 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}") 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 )) return AnayasaSearchResult( decisions=processed_decisions, total_records_found=total_records, 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 ) -> 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}") 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 = 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] 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) ) 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 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.")