This commit is contained in:
saidsurucu
2025-05-21 20:33:43 +03:00
parent 2935c686ea
commit 672337f9d4
19 changed files with 2439 additions and 0 deletions
View File
+356
View File
@@ -0,0 +1,356 @@
# 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(enable_plugins=False)
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", "<!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"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.")
+354
View File
@@ -0,0 +1,354 @@
# 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.")
+212
View File
@@ -0,0 +1,212 @@
# anayasa_mcp_module/models.py
from pydantic import BaseModel, Field, HttpUrl
from typing import List, Optional, Dict, Any
from enum import Enum
# --- Enums (AnayasaDonemEnum, AnayasaBasvuruTuruEnum, etc. - same as before) ---
class AnayasaDonemEnum(str, Enum):
TUMU = ""
DONEM_1961 = "1"
DONEM_1982 = "2"
class AnayasaBasvuruTuruEnum(str, Enum):
TUMU = ""
IPTAL = "1"
ITIRAZ = "2"
DIGER = "3"
class AnayasaVarYokEnum(str, Enum):
TUMU = ""
YOK = "0"
VAR = "1"
class AnayasaNormTuruEnum(str, Enum):
TUMU = ""
ANAYASA = "1"
ANAYASA_DEGISTIREN_KANUN = "2"
CUMHURBASKANLIGI_KARARNAMESI = "14"
ICTUZUK = "3"
KANUN = "4"
KANUN_HUKMUNDE_KARARNAME = "5"
KARAR = "6"
NIZAMNAME = "7"
TALIMATNAME = "8"
TARIFE = "9"
TBMM_KARARI = "10"
TEZKERE = "11"
TUZUK = "12"
YOK_SECENEGI = "0"
YONETMELIK = "13"
class AnayasaIncelemeSonucuEnum(str, Enum):
TUMU = ""
ESAS_ACILMAMIS_SAYILMA = "1"
ESAS_IPTAL = "2"
ESAS_KARAR_YER_OLMADIGI = "3"
ESAS_RET = "4"
ILK_ACILMAMIS_SAYILMA = "5"
ILK_ISIN_GERI_CEVRILMESI = "6"
ILK_KARAR_YER_OLMADIGI = "7"
ILK_RET = "8"
KANUN_6216_M43_4_IPTAL = "12"
class AnayasaSonucGerekcesiEnum(str, Enum):
TUMU = ""
ANAYASAYA_AYKIRI_DEGIL = "29"
ANAYASAYA_ESAS_YONUNDEN_AYKIRILIK = "1"
ANAYASAYA_ESAS_YONUNDEN_UYGUNLUK = "2"
ANAYASAYA_SEKIL_ESAS_UYGUNLUK = "30"
ANAYASAYA_SEKIL_YONUNDEN_AYKIRILIK = "3"
ANAYASAYA_SEKIL_YONUNDEN_UYGUNLUK = "4"
AYKIRILIK_ANAYASAYA_ESAS_YONUNDEN_DUPLICATE = "27"
BASVURU_KARARI = "5"
DENETIM_DISI = "6"
DIGER_GEREKCE_1 = "7"
DIGER_GEREKCE_2 = "8"
EKSIKLIGIN_GIDERILMEMESI = "9"
GEREKCE = "10"
GOREV = "11"
GOREV_YETKI = "12"
GOREVLI_MAHKEME = "13"
GORULMEKTE_OLAN_DAVA = "14"
MAHKEME = "15"
NORMDA_DEGISIKLIK_YAPILMASI = "16"
NORMUN_YURURLUKTEN_KALDIRILMASI = "17"
ON_YIL_YASAGI = "18"
SURE = "19"
USULE_UYMAMA = "20"
UYGULANACAK_NORM = "21"
UYGULANAMAZ_HALE_GELME = "22"
YETKI = "23"
YETKI_SURE = "24"
YOK_HUKMUNDE_OLMAMA = "25"
YOKLUK = "26"
# --- End Enums ---
class AnayasaNormDenetimiSearchRequest(BaseModel):
"""Model for Anayasa Mahkemesi (Norm Denetimi) search request for the MCP tool."""
keywords_all: Optional[List[str]] = Field(default_factory=list, description="Keywords for AND logic (KelimeAra[]).")
keywords_any: Optional[List[str]] = Field(default_factory=list, description="Keywords for OR logic (HerhangiBirKelimeAra[]).")
keywords_exclude: Optional[List[str]] = Field(default_factory=list, description="Keywords to exclude (BulunmayanKelimeAra[]).")
period: Optional[AnayasaDonemEnum] = Field(default=AnayasaDonemEnum.TUMU, description="Constitutional period (Donemler_id).")
case_number_esas: Optional[str] = Field(None, description="Case registry number (EsasNo), e.g., '2023/123'.")
decision_number_karar: Optional[str] = Field(None, description="Decision number (KararNo), e.g., '2023/456'.")
first_review_date_start: Optional[str] = Field(None, description="First review start date (IlkIncelemeTarihiIlk), format DD/MM/YYYY.")
first_review_date_end: Optional[str] = Field(None, description="First review end date (IlkIncelemeTarihiSon), format DD/MM/YYYY.")
decision_date_start: Optional[str] = Field(None, description="Decision start date (KararTarihiIlk), format DD/MM/YYYY.")
decision_date_end: Optional[str] = Field(None, description="Decision end date (KararTarihiSon), format DD/MM/YYYY.")
application_type: Optional[AnayasaBasvuruTuruEnum] = Field(default=AnayasaBasvuruTuruEnum.TUMU, description="Type of application (BasvuruTurler_id).")
applicant_general_name: Optional[str] = Field(None, description="General applicant name (BasvuranGeneller_id).")
applicant_specific_name: Optional[str] = Field(None, description="Specific applicant name (BasvuranOzeller_id).")
official_gazette_date_start: Optional[str] = Field(None, description="Official Gazette start date (ResmiGazeteTarihiIlk), format DD/MM/YYYY.")
official_gazette_date_end: Optional[str] = Field(None, description="Official Gazette end date (ResmiGazeteTarihiSon), format DD/MM/YYYY.")
official_gazette_number_start: Optional[str] = Field(None, description="Official Gazette starting number (ResmiGazeteSayisiIlk).")
official_gazette_number_end: Optional[str] = Field(None, description="Official Gazette ending number (ResmiGazeteSayisiSon).")
has_press_release: Optional[AnayasaVarYokEnum] = Field(default=AnayasaVarYokEnum.TUMU, description="Press release available (BasinDuyurusu).")
has_dissenting_opinion: Optional[AnayasaVarYokEnum] = Field(default=AnayasaVarYokEnum.TUMU, description="Dissenting opinion exists (KarsiOy).")
has_different_reasoning: Optional[AnayasaVarYokEnum] = Field(default=AnayasaVarYokEnum.TUMU, description="Different reasoning exists (FarkliGerekce).")
attending_members_names: Optional[List[str]] = Field(default_factory=list, description="List of attending members' exact names (Uyeler_id[]).")
rapporteur_name: Optional[str] = Field(None, description="Rapporteur's exact name (Raportorler_id).")
norm_type: Optional[AnayasaNormTuruEnum] = Field(default=AnayasaNormTuruEnum.TUMU, description="Type of the reviewed norm (NormunTurler_id).")
norm_id_or_name: Optional[str] = Field(None, description="Number or name of the norm (NormunNumarasiAdlar_id).")
norm_article: Optional[str] = Field(None, description="Article number of the norm (NormunMaddeNumarasi).")
review_outcomes: Optional[List[AnayasaIncelemeSonucuEnum]] = Field(default_factory=list, description="List of review types and outcomes (IncelemeTuruKararSonuclar_id[]).")
reason_for_final_outcome: Optional[AnayasaSonucGerekcesiEnum] = Field(default=AnayasaSonucGerekcesiEnum.TUMU, description="Main reason for the decision outcome (KararSonucununGerekcesi).")
basis_constitution_article_numbers: Optional[List[str]] = Field(default_factory=list, description="List of supporting Constitution article numbers (DayanakHukmu[]).")
results_per_page: Optional[int] = Field(10, description="Number of results per page. Options: 10, 20, 30, 40, 50.")
page_to_fetch: Optional[int] = Field(1, ge=1, description="Page number to fetch for results list.")
sort_by_criteria: Optional[str] = Field("KararTarihi", description="Sort criteria. Options: 'KararTarihi', 'YayinTarihi', 'Toplam' (keyword count).")
class AnayasaReviewedNormInfo(BaseModel):
"""Details of a norm reviewed within an AYM decision summary."""
norm_name_or_number: Optional[str] = None
article_number: Optional[str] = None
review_type_and_outcome: Optional[str] = None
outcome_reason: Optional[str] = None
basis_constitution_articles_cited: List[str] = Field(default_factory=list)
postponement_period: Optional[str] = None
class AnayasaDecisionSummary(BaseModel):
"""Model for a single Anayasa Mahkemesi (Norm Denetimi) decision summary from search results."""
decision_reference_no: Optional[str] = None
decision_page_url: Optional[HttpUrl] = None
keywords_found_count: Optional[int] = None
application_type_summary: Optional[str] = None
applicant_summary: Optional[str] = None
decision_outcome_summary: Optional[str] = None
decision_date_summary: Optional[str] = None
reviewed_norms: List[AnayasaReviewedNormInfo] = Field(default_factory=list)
class AnayasaSearchResult(BaseModel):
"""Model for the overall search result for Anayasa Mahkemesi Norm Denetimi decisions."""
decisions: List[AnayasaDecisionSummary]
total_records_found: Optional[int] = None
retrieved_page_number: Optional[int] = None
class AnayasaDocumentMarkdown(BaseModel):
"""
Model for an Anayasa Mahkemesi (Norm Denetimi) decision document, containing a chunk of Markdown content
and pagination information.
"""
source_url: HttpUrl
decision_reference_no_from_page: Optional[str] = Field(None, description="E.K. No parsed from the document page.")
decision_date_from_page: Optional[str] = Field(None, description="Decision date parsed from the document page.")
official_gazette_info_from_page: Optional[str] = Field(None, description="Official Gazette info parsed from the document page.")
markdown_chunk: Optional[str] = Field(None, description="A 5,000 character chunk of the Markdown content.") # Corrected chunk size
current_page: int = Field(description="The current page number of the markdown chunk (1-indexed).")
total_pages: int = Field(description="Total number of pages for the full markdown content.")
is_paginated: bool = Field(description="True if the full markdown content is split into multiple pages.")
# --- 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[]).")
page_to_fetch: int = Field(1, ge=1, description="Page number to fetch for the report (page). Default is 1.")
class AnayasaBireyselReportDecisionDetail(BaseModel):
"""Details of a specific right/claim within a Bireysel Başvuru decision summary in a report."""
hak: Optional[str] = Field(None, description="İhlal edildiği iddia edilen hak (örneğin, Mülkiyet hakkı).")
mudahale_iddiasi: Optional[str] = Field(None, description="İhlale neden olan müdahale iddiası.")
sonuc: Optional[str] = Field(None, description="İnceleme sonucu (örneğin, İhlal, Düşme).")
giderim: Optional[str] = Field(None, description="Kararlaştırılan giderim (örneğin, Yeniden yargılama).")
class AnayasaBireyselReportDecisionSummary(BaseModel):
"""Model for a single Anayasa Mahkemesi (Bireysel Başvuru) decision summary from a 'Karar Arama Raporu'."""
title: Optional[str] = Field(None, description="Başvurunun başlığı (e.g., 'HASAN DURMUŞ Başvurusuna İlişkin Karar').")
decision_reference_no: Optional[str] = Field(None, description="Başvuru Numarası (e.g., '2019/19126').")
decision_page_url: Optional[HttpUrl] = Field(None, description="URL to the full decision page.")
decision_type_summary: Optional[str] = Field(None, description="Karar Türü (Başvuru Sonucu) (e.g., 'Esas (İhlal)').")
decision_making_body: Optional[str] = Field(None, description="Kararı Veren Birim (e.g., 'Genel Kurul', 'Birinci Bölüm').")
application_date_summary: Optional[str] = Field(None, description="Başvuru Tarihi (DD/MM/YYYY).")
decision_date_summary: Optional[str] = Field(None, description="Karar Tarihi (DD/MM/YYYY).")
application_subject_summary: Optional[str] = Field(None, description="Başvuru konusunun özeti.")
details: List[AnayasaBireyselReportDecisionDetail] = Field(default_factory=list, description="İncelenen haklar ve sonuçlarına ilişkin detaylar.")
class AnayasaBireyselReportSearchResult(BaseModel):
"""Model for the overall search result for Anayasa Mahkemesi 'Karar Arama Raporu'."""
decisions: List[AnayasaBireyselReportDecisionSummary]
total_records_found: Optional[int] = Field(None, description="Raporda bulunan toplam karar sayısı.")
retrieved_page_number: int = Field(description="Alınan rapor sayfa numarası.")
class AnayasaBireyselBasvuruDocumentMarkdown(BaseModel):
"""
Model for an Anayasa Mahkemesi (Bireysel Başvuru) decision document, containing a chunk of Markdown content
and pagination information. Fetched from /BB/YYYY/NNNN paths.
"""
source_url: HttpUrl
basvuru_no_from_page: Optional[str] = Field(None, description="Başvuru Numarası (B.No) parsed from the document page.")
karar_tarihi_from_page: Optional[str] = Field(None, description="Decision date parsed from the document page.")
basvuru_tarihi_from_page: Optional[str] = Field(None, description="Application date parsed from the document page.")
karari_veren_birim_from_page: Optional[str] = Field(None, description="Deciding body (Bölüm/Genel Kurul) parsed from the document page.")
karar_turu_from_page: Optional[str] = Field(None, description="Decision type (Başvuru Sonucu) parsed from the document page.")
resmi_gazete_info_from_page: Optional[str] = Field(None, description="Official Gazette info parsed from the document page, if available.")
markdown_chunk: Optional[str] = Field(None, description="A 5,000 character chunk of the Markdown content.")
current_page: int = Field(description="The current page number of the markdown chunk (1-indexed).")
total_pages: int = Field(description="Total number of pages for the full markdown content.")
is_paginated: bool = Field(description="True if the full markdown content is split into multiple pages.")
# --- End Models for Bireysel Başvuru ---