# anayasa_mcp_module/bireysel_client.py # This client is for Bireysel Başvuru: https://kararlarbilgibankasi.anayasa.gov.tr import httpx from bs4 import BeautifulSoup, Tag 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 ( AnayasaBireyselReportSearchRequest, AnayasaBireyselReportDecisionDetail, AnayasaBireyselReportDecisionSummary, AnayasaBireyselReportSearchResult, AnayasaBireyselBasvuruDocumentMarkdown, # Model for Bireysel Başvuru document ) logger = logging.getLogger(__name__) if not logger.hasHandlers(): logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') class AnayasaBireyselBasvuruApiClient: BASE_URL = "https://kararlarbilgibankasi.anayasa.gov.tr" SEARCH_PATH = "/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_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 async def search_bireysel_basvuru_report( self, 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}") 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 None) alti_cizili_div = decision_div.find("div", class_="AltiCizili") ref_no, dec_type, body, app_date, dec_date, url_path = None, None, None, None, None, None 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 None current_idx += 1 body = parts[current_idx] if len(parts) > current_idx else None current_idx += 1 app_date_raw = parts[current_idx] if len(parts) > current_idx else None current_idx += 1 dec_date_raw = parts[current_idx] if len(parts) > current_idx else None 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 None 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 None, mudahale_iddiasi=cells[1].get_text(strip=True) or None, sonuc=cells[2].get_text(strip=True) or None, giderim=cells[3].get_text(strip=True) or None, )) full_decision_page_url = urljoin(self.BASE_URL, url_path) if url_path else None 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 )) return AnayasaBireyselReportSearchResult( decisions=processed_decisions, total_records_found=total_records, 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 temp_file_path = None try: md_converter = MarkItDown() with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".html", encoding="utf-8") as tmp_file: 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"AnayasaBireyselBasvuruApiClient: 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_path: str, # e.g. /BB/2021/20295 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}") 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 = 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] 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) ) 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 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.")