# 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 Dict, Any, List, Optional, Tuple import logging import html import re import tempfile import os from urllib.parse import urlencode, urljoin, quote 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.value and params.period.value != "ALL": query_params.append(("Donemler_id", params.period.value)) 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.value and params.application_type.value != "ALL": query_params.append(("BasvuruTurler_id", params.application_type.value)) 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.value and params.norm_type.value != "ALL": query_params.append(("NormunTurler_id", params.norm_type.value)) 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_enum_val in params.review_outcomes: if outcome_enum_val.value and outcome_enum_val.value != "ALL": query_params.append(("IncelemeTuruKararSonuclar_id[]", outcome_enum_val.value)) if params.reason_for_final_outcome and params.reason_for_final_outcome.value and params.reason_for_final_outcome.value != "ALL": query_params.append(("KararSonucununGerekcesi", params.reason_for_final_outcome.value)) 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.value and params.has_press_release.value != "ALL": query_params.append(("BasinDuyurusu", params.has_press_release.value)) if params.has_dissenting_opinion and params.has_dissenting_opinion.value and params.has_dissenting_opinion.value != "ALL": query_params.append(("KarsiOy", params.has_dissenting_opinion.value)) if params.has_different_reasoning and params.has_different_reasoning.value and params.has_different_reasoning.value != "ALL": query_params.append(("FarkliGerekce", params.has_different_reasoning.value)) 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: path_segments = [] if params.results_per_page and params.results_per_page != 10: # Default is 10 path_segments.append(f"SatirSayisi/{params.results_per_page}") if params.sort_by_criteria and params.sort_by_criteria != "KararTarihi": # Default is KararTarihi # Ensure correct quoting for criteria that might have Turkish chars or spaces path_segments.append(f"Siralama/{quote(params.sort_by_criteria)}") path_segments.append(self.SEARCH_PATH_SEGMENT) request_path = "/" + "/".join(path_segments) 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 temp_file_path = None try: md_converter = MarkItDown() with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".html", encoding="utf-8") as tmp_file: # 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: tmp_file.write(html_input_for_markdown) temp_file_path = tmp_file.name conversion_result = md_converter.convert(temp_file_path) markdown_text = conversion_result.text_content except Exception as e: logger.error(f"AnayasaMahkemesiApiClient: MarkItDown conversion error: {e}") finally: if temp_file_path and os.path.exists(temp_file_path): os.remove(temp_file_path) 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") 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.")