saidsurucuandClaude Opus 4.7
4c06a5926b
Introduces two tools backed by the gib.gov.tr public JSON API (reverse-engineered from the Next.js SPA chunks): - search_gib_ozelge: keyword, ozelgeNo, kanunNo, date-range, paging over 18k+ Revenue Administration tax rulings. Simple YYYY-MM-DD dates are auto-expanded to ISO 8601 to satisfy the backend. - get_gib_ozelge_document_markdown: fetch a single ruling by numeric id and return 5000-char paginated Markdown with a metadata header block (title, ozelgeNo, tarih, kanun, kaynak). Also prunes stale auth/Fly.io-era entries from uv.lock. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
355 lines
13 KiB
Python
355 lines
13 KiB
Python
# gib_mcp_module/client.py
|
||
|
||
import httpx
|
||
import io
|
||
import logging
|
||
import math
|
||
from typing import Optional, Any, Dict
|
||
from markitdown import MarkItDown
|
||
|
||
from .models import (
|
||
GibSearchRequest,
|
||
GibOzelgeSummary,
|
||
GibSearchResult,
|
||
GibDocumentMarkdown,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
if not logger.hasHandlers():
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
|
||
|
||
class GibApiClient:
|
||
"""
|
||
API client for searching and retrieving GİB özelgeler (Turkish Revenue
|
||
Administration tax rulings) via the public gib.gov.tr JSON API.
|
||
|
||
The endpoint is a single POST list endpoint; document retrieval is done
|
||
by filtering the same endpoint with an exact `id`.
|
||
"""
|
||
|
||
BASE_URL = "https://gib.gov.tr/api"
|
||
LIST_PATH = "/gibportal/mevzuat/ozelge/list"
|
||
DOCUMENT_MARKDOWN_CHUNK_SIZE = 5000
|
||
|
||
# Fixed filter values required by the backend
|
||
_REQUIRED_STATUS = 2
|
||
_REQUIRED_DELETED = False
|
||
_REQUIRED_KTYPE = 99 # ktype=99 selects özelge
|
||
_SORT_FIELD = "ozelgeTarih"
|
||
_SORT_TYPE = "DESC"
|
||
|
||
def __init__(self, request_timeout: float = 60.0):
|
||
self.http_client = httpx.AsyncClient(
|
||
base_url=self.BASE_URL,
|
||
headers={
|
||
"Accept": "application/json",
|
||
"Accept-Language": "tr-TR,tr;q=0.9,en;q=0.7",
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "Mozilla/5.0 (compatible; yargi-mcp/1.0; +https://github.com/saidsurucu/yargi-mcp)",
|
||
},
|
||
timeout=request_timeout,
|
||
verify=True,
|
||
follow_redirects=True,
|
||
)
|
||
|
||
@staticmethod
|
||
def _normalize_date(value: str, end_of_day: bool = False) -> Optional[str]:
|
||
"""
|
||
Accept 'YYYY-MM-DD' or full ISO 8601; always return full ISO 8601.
|
||
|
||
GİB backend rejects date-only strings.
|
||
"""
|
||
if not value:
|
||
return None
|
||
v = value.strip()
|
||
if not v:
|
||
return None
|
||
# Already ISO with time component
|
||
if "T" in v:
|
||
return v
|
||
# Simple YYYY-MM-DD - expand to start/end of day
|
||
suffix = "T23:59:59.999Z" if end_of_day else "T00:00:00.000Z"
|
||
return f"{v}{suffix}"
|
||
|
||
def _build_search_body(self, params: GibSearchRequest) -> Dict[str, Any]:
|
||
body: Dict[str, Any] = {
|
||
"status": self._REQUIRED_STATUS,
|
||
"deleted": self._REQUIRED_DELETED,
|
||
"ktype": self._REQUIRED_KTYPE,
|
||
}
|
||
|
||
keywords = params.keywords.strip()
|
||
kanun_no = params.kanunNo.strip()
|
||
# Frontend sets title/kanunNo/description to the SAME value; the backend
|
||
# ORs across them. If the caller supplies both, combine them so kanun_no
|
||
# still biases toward ruling text, while keywords remain primary.
|
||
search_term = keywords or kanun_no
|
||
if keywords and kanun_no and kanun_no not in keywords:
|
||
search_term = f"{keywords} {kanun_no}"
|
||
if search_term:
|
||
body["title"] = search_term
|
||
body["kanunNo"] = search_term
|
||
body["description"] = search_term
|
||
|
||
if params.ozelgeNo.strip():
|
||
body["ozelgeNo"] = params.ozelgeNo.strip()
|
||
|
||
if params.kanunId and params.kanunId > 0:
|
||
body["kanunIds"] = [params.kanunId]
|
||
|
||
start_iso = self._normalize_date(params.ozelgeStartDate, end_of_day=False)
|
||
end_iso = self._normalize_date(params.ozelgeEndDate, end_of_day=True)
|
||
if start_iso:
|
||
body["ozelgeStartDate"] = start_iso
|
||
if end_iso:
|
||
body["ozelgeEndDate"] = end_iso
|
||
|
||
return body
|
||
|
||
def _build_query_params(self, page_1_indexed: int, page_size: int) -> Dict[str, Any]:
|
||
# API expects 0-indexed page
|
||
zero_indexed = max(0, page_1_indexed - 1)
|
||
return {
|
||
"page": zero_indexed,
|
||
"size": page_size,
|
||
"sortFieldName": self._SORT_FIELD,
|
||
"sortType": self._SORT_TYPE,
|
||
}
|
||
|
||
@staticmethod
|
||
def _to_summary(item: Dict[str, Any]) -> Optional[GibOzelgeSummary]:
|
||
if not isinstance(item, dict):
|
||
return None
|
||
raw_id = item.get("id")
|
||
if raw_id is None:
|
||
return None
|
||
try:
|
||
ozelge_id = int(raw_id)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
return GibOzelgeSummary(
|
||
id=ozelge_id,
|
||
ozelgeNo=item.get("ozelgeNo"),
|
||
ozelgeTarih=item.get("ozelgeTarih"),
|
||
title=item.get("title"),
|
||
kanunNo=item.get("kanunNo"),
|
||
kanunTitle=item.get("kanunTitle"),
|
||
siteLink=item.get("siteLink"),
|
||
)
|
||
|
||
async def search_ozelge(self, params: GibSearchRequest) -> GibSearchResult:
|
||
"""Search GİB özelgeler."""
|
||
body = self._build_search_body(params)
|
||
query = self._build_query_params(params.page, params.pageSize)
|
||
logger.info(
|
||
"GibApiClient: search page=%s size=%s body_keys=%s",
|
||
params.page, params.pageSize, sorted(body.keys()),
|
||
)
|
||
|
||
try:
|
||
resp = await self.http_client.post(self.LIST_PATH, params=query, json=body)
|
||
resp.raise_for_status()
|
||
payload = resp.json()
|
||
except httpx.HTTPStatusError as e:
|
||
logger.error("GibApiClient: HTTP %s during search", e.response.status_code)
|
||
return GibSearchResult(
|
||
ozelgeler=[],
|
||
total_results=0,
|
||
total_pages=0,
|
||
current_page=params.page,
|
||
page_size=params.pageSize,
|
||
)
|
||
except Exception as e:
|
||
logger.error("GibApiClient: search request failed: %s", e)
|
||
return GibSearchResult(
|
||
ozelgeler=[],
|
||
total_results=0,
|
||
total_pages=0,
|
||
current_page=params.page,
|
||
page_size=params.pageSize,
|
||
)
|
||
|
||
container = (payload or {}).get("resultContainer") or {}
|
||
raw_items = container.get("content") or []
|
||
|
||
summaries = []
|
||
for raw in raw_items:
|
||
summary = self._to_summary(raw)
|
||
if summary is not None:
|
||
summaries.append(summary)
|
||
|
||
total_results = container.get("totalElements") or 0
|
||
total_pages = container.get("totalPages") or 0
|
||
try:
|
||
total_results = int(total_results)
|
||
except (TypeError, ValueError):
|
||
total_results = 0
|
||
try:
|
||
total_pages = int(total_pages)
|
||
except (TypeError, ValueError):
|
||
total_pages = 0
|
||
|
||
return GibSearchResult(
|
||
ozelgeler=summaries,
|
||
total_results=total_results,
|
||
total_pages=total_pages,
|
||
current_page=params.page,
|
||
page_size=params.pageSize,
|
||
)
|
||
|
||
def _convert_html_to_markdown(self, html_content: str) -> Optional[str]:
|
||
"""Convert HTML content to Markdown using MarkItDown with BytesIO."""
|
||
if not html_content:
|
||
return None
|
||
try:
|
||
html_bytes = html_content.encode("utf-8")
|
||
html_stream = io.BytesIO(html_bytes)
|
||
md_converter = MarkItDown(enable_plugins=False)
|
||
result = md_converter.convert(html_stream)
|
||
return result.text_content
|
||
except Exception as e:
|
||
logger.error("GibApiClient: HTML→Markdown conversion failed: %s", e)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _build_header_block(item: Dict[str, Any]) -> str:
|
||
"""Build a small Markdown header block summarising the ruling metadata."""
|
||
parts = []
|
||
title = item.get("title")
|
||
if title:
|
||
parts.append(f"# {title}")
|
||
meta_lines = []
|
||
if item.get("ozelgeNo"):
|
||
meta_lines.append(f"**Sayı:** {item['ozelgeNo']}")
|
||
if item.get("ozelgeTarih"):
|
||
meta_lines.append(f"**Tarih:** {item['ozelgeTarih']}")
|
||
if item.get("kanunTitle"):
|
||
kanun_no = item.get("kanunNo")
|
||
if kanun_no:
|
||
meta_lines.append(f"**Kanun:** {item['kanunTitle']} ({kanun_no})")
|
||
else:
|
||
meta_lines.append(f"**Kanun:** {item['kanunTitle']}")
|
||
if item.get("siteLink"):
|
||
meta_lines.append(f"**Kaynak:** {item['siteLink']}")
|
||
if meta_lines:
|
||
parts.append("\n".join(meta_lines))
|
||
return "\n\n".join(parts).strip()
|
||
|
||
async def get_ozelge_document(
|
||
self, ozelge_id: int, page_number: int = 1
|
||
) -> GibDocumentMarkdown:
|
||
"""Retrieve a single özelge and return its paginated Markdown form."""
|
||
logger.info(
|
||
"GibApiClient: fetching özelge id=%s page=%s", ozelge_id, page_number
|
||
)
|
||
|
||
if not isinstance(ozelge_id, int) or ozelge_id <= 0:
|
||
return GibDocumentMarkdown(
|
||
ozelge_id=ozelge_id if isinstance(ozelge_id, int) else 0,
|
||
current_page=page_number,
|
||
total_pages=0,
|
||
is_paginated=False,
|
||
error_message="ozelge_id must be a positive integer",
|
||
)
|
||
|
||
body = {
|
||
"status": self._REQUIRED_STATUS,
|
||
"deleted": self._REQUIRED_DELETED,
|
||
"ktype": self._REQUIRED_KTYPE,
|
||
"id": ozelge_id,
|
||
}
|
||
query = {"page": 0, "size": 1}
|
||
|
||
try:
|
||
resp = await self.http_client.post(self.LIST_PATH, params=query, json=body)
|
||
resp.raise_for_status()
|
||
payload = resp.json()
|
||
except httpx.HTTPStatusError as e:
|
||
msg = f"HTTP {e.response.status_code} when fetching özelge {ozelge_id}"
|
||
logger.error("GibApiClient: %s", msg)
|
||
return GibDocumentMarkdown(
|
||
ozelge_id=ozelge_id,
|
||
current_page=page_number,
|
||
total_pages=0,
|
||
is_paginated=False,
|
||
error_message=msg,
|
||
)
|
||
except Exception as e:
|
||
msg = f"Request failed: {e}"
|
||
logger.error("GibApiClient: %s", msg)
|
||
return GibDocumentMarkdown(
|
||
ozelge_id=ozelge_id,
|
||
current_page=page_number,
|
||
total_pages=0,
|
||
is_paginated=False,
|
||
error_message=msg,
|
||
)
|
||
|
||
container = (payload or {}).get("resultContainer") or {}
|
||
content = container.get("content") or []
|
||
if not content:
|
||
return GibDocumentMarkdown(
|
||
ozelge_id=ozelge_id,
|
||
current_page=page_number,
|
||
total_pages=0,
|
||
is_paginated=False,
|
||
error_message=f"Özelge {ozelge_id} not found",
|
||
)
|
||
|
||
item = content[0] if isinstance(content[0], dict) else {}
|
||
description_html = item.get("description") or ""
|
||
markdown_body = self._convert_html_to_markdown(description_html) or ""
|
||
header_block = self._build_header_block(item)
|
||
|
||
if header_block and markdown_body:
|
||
full_markdown = f"{header_block}\n\n---\n\n{markdown_body}"
|
||
else:
|
||
full_markdown = header_block or markdown_body
|
||
|
||
if not full_markdown.strip():
|
||
return GibDocumentMarkdown(
|
||
ozelge_id=ozelge_id,
|
||
ozelge_no=item.get("ozelgeNo"),
|
||
title=item.get("title"),
|
||
ozelge_tarih=item.get("ozelgeTarih"),
|
||
kanun_title=item.get("kanunTitle"),
|
||
kanun_no=item.get("kanunNo"),
|
||
site_link=item.get("siteLink"),
|
||
current_page=page_number,
|
||
total_pages=0,
|
||
is_paginated=False,
|
||
error_message="Document body is empty",
|
||
)
|
||
|
||
total_pages = max(
|
||
1, math.ceil(len(full_markdown) / self.DOCUMENT_MARKDOWN_CHUNK_SIZE)
|
||
)
|
||
current_page_clamped = max(1, min(page_number, total_pages))
|
||
start = (current_page_clamped - 1) * self.DOCUMENT_MARKDOWN_CHUNK_SIZE
|
||
end = start + self.DOCUMENT_MARKDOWN_CHUNK_SIZE
|
||
chunk = full_markdown[start:end]
|
||
|
||
return GibDocumentMarkdown(
|
||
ozelge_id=ozelge_id,
|
||
ozelge_no=item.get("ozelgeNo"),
|
||
title=item.get("title"),
|
||
ozelge_tarih=item.get("ozelgeTarih"),
|
||
kanun_title=item.get("kanunTitle"),
|
||
kanun_no=item.get("kanunNo"),
|
||
site_link=item.get("siteLink"),
|
||
markdown_chunk=chunk,
|
||
current_page=current_page_clamped,
|
||
total_pages=total_pages,
|
||
is_paginated=total_pages > 1,
|
||
error_message=None,
|
||
)
|
||
|
||
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("GibApiClient: HTTP client session closed.")
|