354 lines
20 KiB
Python
354 lines
20 KiB
Python
# 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: 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: 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: 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: query_params.append(("IncelemeTuruKararSonuclar_id[]", outcome_enum_val.value))
|
||
if params.reason_for_final_outcome and params.reason_for_final_outcome.value:
|
||
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: query_params.append(("BasinDuyurusu", params.has_press_release.value))
|
||
if params.has_dissenting_opinion and params.has_dissenting_opinion.value: query_params.append(("KarsiOy", params.has_dissenting_opinion.value))
|
||
if params.has_different_reasoning and params.has_different_reasoning.value: 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(enable_plugins=False)
|
||
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", "<!doctype")):
|
||
tmp_file.write(f"<html><head><meta charset=\"UTF-8\"></head><body>{html_input_for_markdown}</body></html>")
|
||
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 <p> tags directly or in the WordSection1
|
||
# Look for patterns like "Esas No.: YYYY/NN" and "Karar No.: YYYY/NN"
|
||
|
||
esas_no_tag = karar_metni_div.find(lambda tag: tag.name == "p" and tag.find("b") and "Esas No.:" in tag.find("b").get_text())
|
||
karar_no_tag = karar_metni_div.find(lambda tag: tag.name == "p" and tag.find("b") and "Karar No.:" in tag.find("b").get_text())
|
||
karar_tarihi_tag = karar_metni_div.find(lambda tag: tag.name == "p" and tag.find("b") and "Karar tarihi:" in tag.find("b").get_text()) # Less common on Norm pages
|
||
resmi_gazete_tag = karar_metni_div.find(lambda tag: tag.name == "p" and ("Resmî Gazete tarih ve sayısı:" in tag.get_text() or "Resmi Gazete tarih/sayı:" in tag.get_text()))
|
||
|
||
|
||
if esas_no_tag and esas_no_tag.find("b") and karar_no_tag and karar_no_tag.find("b"):
|
||
esas_str = esas_no_tag.find("b").get_text(strip=True).replace('Esas No.:', '').strip()
|
||
karar_str = karar_no_tag.find("b").get_text(strip=True).replace('Karar No.:', '').strip()
|
||
decision_ek_no_from_page = f"E.{esas_str}, K.{karar_str}"
|
||
|
||
if karar_tarihi_tag and karar_tarihi_tag.find("b"):
|
||
decision_date_from_page = karar_tarihi_tag.find("b").get_text(strip=True).replace("Karar tarihi:", "").strip()
|
||
elif karar_metni_div: # Fallback for Karar Tarihi if not in specific tag
|
||
date_match = re.search(r"Karar Tarihi\s*:\s*([\d\.]+)", karar_metni_div.get_text()) # Norm pages often use DD.MM.YYYY
|
||
if date_match: decision_date_from_page = date_match.group(1).strip()
|
||
|
||
|
||
if resmi_gazete_tag:
|
||
# Try to get the bold part first if it exists
|
||
bold_rg_tag = resmi_gazete_tag.find("b")
|
||
rg_text_content = bold_rg_tag.get_text(strip=True) if bold_rg_tag else resmi_gazete_tag.get_text(strip=True)
|
||
official_gazette_from_page = rg_text_content.replace("Resmî Gazete tarih ve sayısı:", "").replace("Resmi Gazete tarih/sayı:", "").strip()
|
||
|
||
|
||
full_markdown_content = 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.") |