| import unicodedata |
| from typing import Any |
|
|
| from data.database import get_connection, initialize_database |
|
|
|
|
| def normalize_text(value: str) -> str: |
| """ |
| Türkçe karakter ve büyük-küçük harf farklılıklarını azaltarak |
| arama işlemlerini daha dayanıklı hâle getirir. |
| |
| Örnek: |
| "KÖRLÜK" -> "korluk" |
| "Suç ve Ceza" -> "suc ve ceza" |
| """ |
|
|
| value = value.strip().casefold() |
|
|
| replacements = { |
| "ı": "i", |
| "ğ": "g", |
| "ü": "u", |
| "ş": "s", |
| "ö": "o", |
| "ç": "c", |
| } |
|
|
| for source, target in replacements.items(): |
| value = value.replace(source, target) |
|
|
| return "".join( |
| character |
| for character in unicodedata.normalize("NFKD", value) |
| if not unicodedata.combining(character) |
| ) |
|
|
|
|
| def serialize_book(book: Any) -> dict[str, Any]: |
| """SQLite kitap satırını JSON uyumlu sözlüğe dönüştürür.""" |
|
|
| return { |
| "book_id": book["id"], |
| "title": book["title"], |
| "author": book["author"], |
| "category": book["category"], |
| "price": round(float(book["price"]), 2), |
| "stock": int(book["stock"]), |
| "available": int(book["stock"]) > 0, |
| } |
|
|
|
|
| def search_books( |
| query: str | None = None, |
| author: str | None = None, |
| category: str | None = None, |
| in_stock_only: bool = False, |
| limit: int = 10, |
| ) -> dict[str, Any]: |
| """ |
| Kitapları başlık, yazar veya kategoriye göre arar. |
| |
| Model, kitap bilgilerini kendisi üretmek yerine bu fonksiyondan |
| dönen gerçek verileri kullanmalıdır. |
| """ |
|
|
| initialize_database() |
|
|
| if limit < 1: |
| return { |
| "success": False, |
| "error": "limit değeri en az 1 olmalıdır.", |
| } |
|
|
| limit = min(limit, 20) |
|
|
| normalized_query = normalize_text(query) if query else None |
| normalized_author = normalize_text(author) if author else None |
| normalized_category = normalize_text(category) if category else None |
|
|
| with get_connection() as connection: |
| rows = connection.execute( |
| """ |
| SELECT |
| id, |
| title, |
| author, |
| category, |
| price, |
| stock |
| FROM books |
| ORDER BY title |
| """ |
| ).fetchall() |
|
|
| matched_books: list[dict[str, Any]] = [] |
|
|
| for row in rows: |
| normalized_title_value = normalize_text(row["title"]) |
| normalized_author_value = normalize_text(row["author"]) |
| normalized_category_value = normalize_text(row["category"]) |
|
|
| if normalized_query: |
| query_matches = ( |
| normalized_query in normalized_title_value |
| or normalized_query in normalized_author_value |
| or normalized_query in normalized_category_value |
| ) |
|
|
| if not query_matches: |
| continue |
|
|
| if ( |
| normalized_author |
| and normalized_author not in normalized_author_value |
| ): |
| continue |
|
|
| if ( |
| normalized_category |
| and normalized_category not in normalized_category_value |
| ): |
| continue |
|
|
| if in_stock_only and row["stock"] <= 0: |
| continue |
|
|
| matched_books.append(serialize_book(row)) |
|
|
| if len(matched_books) >= limit: |
| break |
|
|
| return { |
| "success": True, |
| "count": len(matched_books), |
| "filters": { |
| "query": query, |
| "author": author, |
| "category": category, |
| "in_stock_only": in_stock_only, |
| "limit": limit, |
| }, |
| "books": matched_books, |
| "message": ( |
| f"{len(matched_books)} kitap bulundu." |
| if matched_books |
| else "Arama kriterlerine uygun kitap bulunamadı." |
| ), |
| } |
|
|
|
|
| def create_order( |
| book_id: int, |
| quantity: int, |
| customer_name: str, |
| ) -> dict[str, Any]: |
| """ |
| Sipariş oluşturur ve kitap stoğunu düşürür. |
| |
| Sipariş kaydı ile stok güncellemesi aynı transaction içinde yapılır. |
| Böylece işlemlerden biri başarısız olursa veritabanı yarım kalmaz. |
| """ |
|
|
| initialize_database() |
|
|
| customer_name = customer_name.strip() |
|
|
| if not customer_name: |
| return { |
| "success": False, |
| "error": "Müşteri adı boş bırakılamaz.", |
| } |
|
|
| if not isinstance(book_id, int) or isinstance(book_id, bool): |
| return { |
| "success": False, |
| "error": "book_id tam sayı olmalıdır.", |
| } |
|
|
| if not isinstance(quantity, int) or isinstance(quantity, bool): |
| return { |
| "success": False, |
| "error": "quantity tam sayı olmalıdır.", |
| } |
|
|
| if quantity < 1: |
| return { |
| "success": False, |
| "error": "Sipariş miktarı en az 1 olmalıdır.", |
| } |
|
|
| connection = get_connection() |
|
|
| try: |
| |
| |
| connection.execute("BEGIN IMMEDIATE") |
|
|
| book = connection.execute( |
| """ |
| SELECT |
| id, |
| title, |
| author, |
| category, |
| price, |
| stock |
| FROM books |
| WHERE id = ? |
| """, |
| (book_id,), |
| ).fetchone() |
|
|
| if book is None: |
| connection.rollback() |
|
|
| return { |
| "success": False, |
| "error": "Belirtilen kimliğe sahip kitap bulunamadı.", |
| "book_id": book_id, |
| } |
|
|
| current_stock = int(book["stock"]) |
|
|
| if current_stock < quantity: |
| connection.rollback() |
|
|
| return { |
| "success": False, |
| "error": "Yeterli stok bulunmuyor.", |
| "book_id": book_id, |
| "title": book["title"], |
| "requested_quantity": quantity, |
| "available_stock": current_stock, |
| } |
|
|
| unit_price = round(float(book["price"]), 2) |
| total_price = round(unit_price * quantity, 2) |
| new_stock = current_stock - quantity |
|
|
| cursor = connection.execute( |
| """ |
| INSERT INTO orders ( |
| customer_name, |
| book_id, |
| quantity, |
| unit_price, |
| total_price, |
| status |
| ) |
| VALUES (?, ?, ?, ?, ?, ?) |
| """, |
| ( |
| customer_name, |
| book_id, |
| quantity, |
| unit_price, |
| total_price, |
| "Hazırlanıyor", |
| ), |
| ) |
|
|
| order_id = cursor.lastrowid |
|
|
| connection.execute( |
| """ |
| UPDATE books |
| SET stock = ? |
| WHERE id = ? |
| """, |
| ( |
| new_stock, |
| book_id, |
| ), |
| ) |
|
|
| connection.commit() |
|
|
| return { |
| "success": True, |
| "message": "Sipariş başarıyla oluşturuldu.", |
| "order": { |
| "order_id": order_id, |
| "customer_name": customer_name, |
| "book_id": book_id, |
| "title": book["title"], |
| "author": book["author"], |
| "quantity": quantity, |
| "unit_price": unit_price, |
| "total_price": total_price, |
| "status": "Hazırlanıyor", |
| }, |
| "stock_update": { |
| "previous_stock": current_stock, |
| "new_stock": new_stock, |
| }, |
| } |
|
|
| except Exception as error: |
| connection.rollback() |
|
|
| return { |
| "success": False, |
| "error": "Sipariş oluşturulurken veritabanı hatası oluştu.", |
| "detail": str(error), |
| } |
|
|
| finally: |
| connection.close() |
|
|
|
|
| def get_order_status(order_id: int) -> dict[str, Any]: |
| """Sipariş numarasına göre sipariş bilgilerini getirir.""" |
|
|
| initialize_database() |
|
|
| if not isinstance(order_id, int) or isinstance(order_id, bool): |
| return { |
| "success": False, |
| "error": "order_id tam sayı olmalıdır.", |
| } |
|
|
| with get_connection() as connection: |
| order = connection.execute( |
| """ |
| SELECT |
| orders.id AS order_id, |
| orders.customer_name, |
| orders.quantity, |
| orders.unit_price, |
| orders.total_price, |
| orders.status, |
| orders.created_at, |
| books.id AS book_id, |
| books.title, |
| books.author |
| FROM orders |
| INNER JOIN books |
| ON books.id = orders.book_id |
| WHERE orders.id = ? |
| """, |
| (order_id,), |
| ).fetchone() |
|
|
| if order is None: |
| return { |
| "success": False, |
| "error": "Sipariş bulunamadı.", |
| "order_id": order_id, |
| } |
|
|
| return { |
| "success": True, |
| "order": { |
| "order_id": order["order_id"], |
| "customer_name": order["customer_name"], |
| "book_id": order["book_id"], |
| "title": order["title"], |
| "author": order["author"], |
| "quantity": order["quantity"], |
| "unit_price": round(float(order["unit_price"]), 2), |
| "total_price": round(float(order["total_price"]), 2), |
| "status": order["status"], |
| "created_at": order["created_at"], |
| }, |
| } |
|
|
|
|
| TOOL_FUNCTIONS = { |
| "search_books": search_books, |
| "create_order": create_order, |
| "get_order_status": get_order_status, |
| } |
|
|
|
|
| def execute_tool( |
| tool_name: str, |
| arguments: dict[str, Any], |
| ) -> dict[str, Any]: |
| """ |
| Model tarafından seçilen tool adını ilgili Python fonksiyonuna yönlendirir. |
| """ |
|
|
| tool_function = TOOL_FUNCTIONS.get(tool_name) |
|
|
| if tool_function is None: |
| return { |
| "success": False, |
| "error": f"Desteklenmeyen tool: {tool_name}", |
| } |
|
|
| try: |
| return tool_function(**arguments) |
|
|
| except TypeError as error: |
| return { |
| "success": False, |
| "error": "Tool parametreleri geçersiz veya eksik.", |
| "detail": str(error), |
| } |
|
|
| except Exception as error: |
| return { |
| "success": False, |
| "error": "Tool çalıştırılırken beklenmeyen hata oluştu.", |
| "detail": str(error), |
| } |
