Add GİB özelge (tax rulings) MCP module
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>
This commit is contained in:
saidsurucu
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
2cec4dccd6
commit
4c06a5926b
@@ -0,0 +1,354 @@
|
||||
# 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.")
|
||||
Reference in New Issue
Block a user