"""TEST AMAÇLI sahte LLM sunucusu — gerçek bir dil modeli DEĞİLDİR. Anthropic (/v1/models, /v1/messages), OpenAI (/v1/models, /v1/chat/completions) ve Ollama (/api/tags, /api/pull, /api/chat, /api/version) uç noktalarının istek/yanıt biçimlerini taklit eder. Böylece model senkronizasyonu, sağlayıcı bağlantıları ve tüm AI özellikleri API anahtarı olmadan uçtan uca sınanabilir. SQL görevlerinde yanıtı projenin kural tabanlı üreticisiyle üretir. Kendini düzeltme döngüsünü göstermek için, soruda gg.aa.yyyy biçimli bir tarih varsa ilk denemede bilerek Türkçe tarih literali yazar. Çalıştırma: python tests/mock_llm_server.py [port] (vars. 8765) Claude : ANTHROPIC_BASE_URL=http://localhost:8765 ANTHROPIC_API_KEY=test OpenAI : Base URL http://localhost:8765/v1 anahtar: test Ollama : OLLAMA_HOST=http://localhost:8765 """ from __future__ import annotations import json import os import re import sys import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) from t2s.rulebased import RuleBasedTextToSQL # noqa: E402 from t2s.schema import connect # noqa: E402 from t2s.trnum import slugify # noqa: E402 LAST = {} # son Ollama isteğinin seçenekleri (testler /mock/last ile okur) DB = os.getenv("MOCK_DB", str(ROOT / "outputs" / "db" / "arayuz.db")) OLLAMA_MODELS = ["qwen2.5-coder:7b", "llama3.1:8b"] # MOCK_EXTRA=1: otomasyon sınaması için bozuk ('gpt-mock-bozuk') ve sınamayı geçip işte çöken # ('claude-mock-dengesiz') modelleri de listele EXTRA = os.getenv("MOCK_EXTRA") == "1" EN_DEMO = { # İngilizce demo soruları (sahte sunucu çeviri yapamaz) "what was izmir's electricity consumption in 2024?": "İzmir'in 2024 elektrik tüketimi ne kadar?", "which province consumed the most electricity in 2024?": "2024'te en çok elektrik tüketen il hangisi?", "how much were personnel expenses in 2024?": "2024 yılında personel giderleri için ne kadar harcama gerçekleşti?", } def _sql_reply(messages: list[dict]) -> str: first = messages[0]["content"] m = re.search(r"### Soru\n(.+?)\n", first, re.S) q = (m.group(1) if m else "").strip() q_tr = EN_DEMO.get(q.lower(), q) rb = RuleBasedTextToSQL(connect(DB, readonly=True)) r = rb.generate(q_tr) feedback_turns = sum(1 for x in messages if x["role"] == "user") - 1 if not r.candidates: return json.dumps({"dusunce": "Tabloda karşılık yok.", "cevaplanabilir": False, "sql": "", "gerekce": r.reason}, ensure_ascii=False) sql = r.candidates[0] dm = re.search(r"\b(\d{2})\.(\d{2})\.(\d{4})\b", q_tr) if dm and feedback_turns == 0: # bilinçli hata: kendini düzeltme gösterimi iso = f"{dm.group(3)}-{dm.group(2)}-{dm.group(1)}" sql = sql.replace(f"'{iso}'", f"'{dm.group(0)}'") return json.dumps({"dusunce": "Şemadaki ilgili tablo ve sütunlar kullanıldı.", "cevaplanabilir": True, "sql": sql}, ensure_ascii=False) def _lang(system: str) -> str: return "en" if "Yanıt dili: English" in system else "tr" def reply(system: str, messages: list[dict]) -> str: task = (re.search(r"\[GÖREV:([A-Z_]+)\]", system) or [None, "DIGER"])[1] user = messages[-1]["content"] if messages else "" en = _lang(system) == "en" if task == "SQL": return _sql_reply(messages) if task == "PING": return "ok" if task == "SINAMA": # otomasyon modül sınaması return "SELECT SUM(tutar) FROM satis WHERE yil = 2024;" if task == "TABLO_ZENGINLESTIR": d = json.loads(user) cap = re.sub(r"^\s*Tablo\s*\d+\s*:\s*", "", d.get("belgedeki_baslik", "")) cols = {} warns = [] for c in d["sutunlar"]: desc = (f"Values of the '{c['orijinal']}' column" if en else f"'{c['orijinal']}' başlıklı sütunun değerleri") cols[c["ad"]] = {"aciklama": desc, "birim": c.get("birim") or ""} if c.get("ayristirilamayan"): warns.append((f"{c['ad']}: {c['ayristirilamayan']} cells could not be parsed as numbers." if en else f"{c['ad']}: {c['ayristirilamayan']} hücre sayı olarak ayrıştırılamadı.")) if c["tip"] == "TEXT" and re.search(r"(19|20)\d{2}", c["ad"]): warns.append(f"{c['ad']}: " + ("year-named column stored as text." if en else "yıl adlı sütun metin olarak saklanmış.")) name = slugify(cap, 40) summ = (f"[MOCK] '{cap}' table with {len(cols)} columns." if en else f"[MOCK] '{cap}' tablosu; {len(cols)} sütun içerir.") return json.dumps({"tablo_adi": name, "ozet": summ, "sutunlar": cols, "uyarilar": warns}, ensure_ascii=False) if task == "SORU_ONER": qs = (["What was İzmir's electricity consumption in 2024?", "Which province consumed the most electricity in 2024?", "How much were personnel expenses in 2024?", "İzmir'in 2024 elektrik tüketimi ne kadar?", "Yürürlük tarihi 15.03.2025 olan hizmetlerin ortalama ücreti nedir?", "Mersin'in 2024 elektrik tüketimi ne kadar?"] if en else ["Faiz giderlerinin bütçesi 2023'ten 2024'e yüzde kaç arttı?", "Marmara bölgesindeki illerin 2024 toplam tüketimi ne kadar?", "Yürürlük tarihi 15.03.2025 olan hizmetlerin ortalama ücreti nedir?", "En fazla su tüketen mahalle hangisi?", "2024 yılında hasılat ne kadardı?", "Mersin'in 2024 elektrik tüketimi ne kadar?"]) return json.dumps(qs, ensure_ascii=False) if task == "CEVAP": m = re.search(r"Biçimlenmiş kısa cevap: (.+)", user) short = m.group(1).strip() if m else "" short = re.sub(r"^(Cevap|Answer): ", "", short) return (f"[MOCK] Based on the tables, the answer is {short}." if en else f"[MOCK] Tablolara göre sorunuzun cevabı: {short}.") if task == "HATA_ACIKLA": d = json.loads(user) if en: return (f"**[MOCK] What happened?** A `{d.get('tip')}` error at the {d.get('asama')} stage.\n\n" f"**Why?** {d.get('mesaj')}\n\n**How to fix?** {'; '.join(d.get('oneriler') or []) or '-'}") return (f"**[MOCK] Ne oldu?** {d.get('asama')} aşamasında `{d.get('tip')}` hatası.\n\n" f"**Neden?** {d.get('mesaj')}\n\n**Nasıl düzeltilir?** {'; '.join(d.get('oneriler') or []) or '-'}") if task == "DEGERLENDIRME_OZETI": rows = json.loads(user) cnt = {} for r in rows: cnt.setdefault(r.get("hata_tipi") or "?", []).append(r["id"]) head = "**[MOCK] Failure themes**" if en else "**[MOCK] Hata temaları**" return head + "\n\n" + "\n".join(f"- **{k}** — {len(v)} ({', '.join(v[:6])})" for k, v in cnt.items()) if task == "TABLO_SEC": names = re.findall(r"^- ([a-z0-9_]+):", user, re.M) q = (re.search(r"^Soru: (.+)$", user, re.M) or [None, ""])[1] try: # kural tabanlı tablo puanlarıyla seç (sahte sunucu dil modeli değildir) r = RuleBasedTextToSQL(connect(DB, readonly=True)).generate(q) ranked = [n for _, n in r.trace.get("tablo_puanlari", []) if n in names][:3] except Exception: # noqa: BLE001 ranked = [] return json.dumps(ranked or names[:3]) if task == "TABLO_OZETI": return "[MOCK] This table summarises the values in the document." if en else \ "[MOCK] Bu tablo belgedeki değerleri özetler." return "ok" class H(BaseHTTPRequestHandler): def log_message(self, *a): # sessiz pass def _json(self, code: int, obj, ctype="application/json"): body = json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): # noqa: N802 p = self.path.split("?")[0] if p == "/v1/models": if self.headers.get("x-api-key") is not None: if self.headers.get("x-api-key") != "test": return self._json(401, {"type": "error", "error": {"type": "authentication_error", "message": "invalid x-api-key"}}) return self._json(200, {"data": [ {"type": "model", "id": "claude-mock-large", "display_name": "Claude Mock Large", "created_at": "2026-06-01T00:00:00Z"}, {"type": "model", "id": "claude-mock-small", "display_name": "Claude Mock Small", "created_at": "2025-10-01T00:00:00Z"}] + ([{"type": "model", "id": "claude-mock-dengesiz", "display_name": "Claude Mock Dengesiz", "created_at": "2025-01-01T00:00:00Z"}] if EXTRA else []), "has_more": False, "first_id": "claude-mock-large", "last_id": "claude-mock-small"}) return self._json(200, {"object": "list", "data": [ {"id": "gpt-mock-mini", "object": "model", "created": 1750000000, "owned_by": "mock"}, {"id": "gpt-mock", "object": "model", "created": 1760000000, "owned_by": "mock"}] + ([{"id": "gpt-mock-bozuk", "object": "model", "created": 1700000000, "owned_by": "mock"}] if EXTRA else [])}) if p == "/api/version": return self._json(200, {"version": "0.0-mock"}) if p == "/mock/last": return self._json(200, LAST) if p == "/api/tags": return self._json(200, {"models": [{"name": n, "model": n, "size": 4_700_000_000, "details": {"parameter_size": "7B", "quantization_level": "Q4_K_M"}} for n in OLLAMA_MODELS]}) return self._json(404, {"error": "not found"}) def do_POST(self): # noqa: N802 n = int(self.headers.get("Content-Length") or 0) d = json.loads(self.rfile.read(n) or b"{}") p = self.path.split("?")[0] if p == "/v1/messages": if self.headers.get("x-api-key") != "test": return self._json(401, {"type": "error", "error": {"type": "authentication_error", "message": "invalid x-api-key"}}) if d.get("model") == "claude-mock-dengesiz" and "[GÖREV:SQL]" in d.get("system", ""): # sınamayı geçen ama gerçek işte çöken model (otomasyonun yedek model davranışını sınar) return self._json(529, {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}) txt = reply(d.get("system", ""), d.get("messages", [])) return self._json(200, {"id": "msg_mock", "type": "message", "role": "assistant", "model": d.get("model"), "content": [{"type": "text", "text": txt}], "stop_reason": "end_turn", "usage": {"input_tokens": len(json.dumps(d)) // 4, "output_tokens": len(txt) // 4}}) if p == "/v1/chat/completions": if d.get("model") == "gpt-mock-bozuk": # sınamada elenmesi gereken bozuk model return self._json(400, {"error": {"message": "The model `gpt-mock-bozuk` does not exist", "type": "invalid_request_error"}}) msgs = d.get("messages", []) system = msgs[0]["content"] if msgs and msgs[0]["role"] == "system" else "" txt = reply(system, [m for m in msgs if m["role"] != "system"]) return self._json(200, {"id": "chatcmpl-mock", "object": "chat.completion", "model": d.get("model"), "choices": [{"index": 0, "message": {"role": "assistant", "content": txt}, "finish_reason": "stop"}], "usage": {"prompt_tokens": len(json.dumps(d)) // 4, "completion_tokens": len(txt) // 4}}) if p == "/api/chat": msgs = d.get("messages", []) LAST.clear() LAST.update({"model": d.get("model"), "options": d.get("options", {})}) if d.get("model") not in OLLAMA_MODELS: return self._json(404, {"error": f"model '{d.get('model')}' not found"}) system = msgs[0]["content"] if msgs and msgs[0]["role"] == "system" else "" txt = reply(system, [m for m in msgs if m["role"] != "system"]) return self._json(200, {"model": d.get("model"), "message": {"role": "assistant", "content": txt}, "done": True, "prompt_eval_count": len(json.dumps(d)) // 4, "eval_count": len(txt) // 4}) if p == "/api/pull": name = d.get("model") or d.get("name") self.send_response(200) self.send_header("Content-Type", "application/x-ndjson") self.end_headers() for line in ({"status": "pulling manifest"}, {"status": "downloading", "completed": 50, "total": 100}, {"status": "downloading", "completed": 100, "total": 100}, {"status": "success"}): self.wfile.write((json.dumps(line) + "\n").encode()) self.wfile.flush() time.sleep(0.05) if name and name not in OLLAMA_MODELS: OLLAMA_MODELS.append(name) return None return self._json(404, {"error": "not found"}) def serve(port: int = 8765): srv = ThreadingHTTPServer(("127.0.0.1", port), H) srv.serve_forever() if __name__ == "__main__": serve(int(sys.argv[1]) if len(sys.argv) > 1 else 8765)