Spaces:
Sleeping
Sleeping
| import re | |
| import logging | |
| from datetime import date | |
| from enum import Enum | |
| from typing import Dict, List, Optional, Any, Tuple | |
| from .db_adapter import get_database_adapter | |
| from config import DEALER_ACCESS_CONFIG | |
| logger = logging.getLogger(__name__) | |
| class IdentifierType(Enum): | |
| CREDIT_FILE_ID = "credit_file_id" | |
| LEAD_ID = "lead_id" | |
| PLATE_NUMBER = "plate_number" | |
| PROPOSAL_NO = "proposal_no" | |
| _LEAD_ID_RE = re.compile(r"\b(L-P\d{6,})\b", re.IGNORECASE) | |
| _PLATE_RE = re.compile(r"\b(\d{1,2})[\s\-]?([A-Za-zÇçĞğİıÖöŞşÜü]{1,3})[\s\-]?(\d{2,4})\b") | |
| _PROPOSAL_NO_RE = re.compile(r"\b(\d{10,})\b") | |
| _CREDIT_FILE_ID_RE = re.compile(r"\b(\d{5,9})\b") | |
| _LEAD_CREDIT_KW = ["kredi numarası", "kredi numarasi", "teklif numarası", "teklif numarasi", "teklif no", "kredi no", "teklif id", "kredi id"] | |
| _LEAD_CUSTOMER_KW = ["müşteri", "musteri", "isim", "ad"] | |
| _LEAD_STATUS_KW = ["durum", "statü", "statu", "status"] | |
| _LEAD_DATE_KW = ["tarih", "ne zaman", "oluşturulma", "olusturulma"] | |
| _LIEN_STATUS_KW = ["rehin var mı", "rehin durumu", "rehni var mı", "rehin var mi", "rehni var mi"] | |
| _LIEN_CREDIT_KW = ["teklif", "kredi numarası", "kredi numarasi", "kredi no"] | |
| def _query_has(query, keywords): | |
| q = query.lower() | |
| return any(kw in q for kw in keywords) | |
| def _format_lead(row, prefix, query=""): | |
| cid = row.get("CreditFileId", "") | |
| name = row.get("CustomerName", "") | |
| lead_created = row.get("LeadCreatedAt", "") | |
| cf_created = row.get("CreditFileCreatedAt", "") | |
| status = row.get("LeadStatus", "") | |
| if _query_has(query, _LEAD_CREDIT_KW): | |
| return f"{prefix} lead numarali basvurunun teklif numarasi {cid}'dir." | |
| if _query_has(query, _LEAD_CUSTOMER_KW): | |
| return f"{prefix} lead numarali basvurunun musterisi {name}'dir." if name else f"{prefix} lead numarali basvurunun musteri bilgisi bulunamadi." | |
| if _query_has(query, _LEAD_STATUS_KW): | |
| return f"{prefix} lead numarali basvurunun durumu: {status}." if status else f"{prefix} lead numarali basvurunun durum bilgisi bulunamadi." | |
| if _query_has(query, _LEAD_DATE_KW): | |
| parts = [] | |
| if lead_created: | |
| parts.append(f"{prefix} lead numarali basvuru {lead_created} tarihinde olusturulmus.") | |
| if cf_created: | |
| parts.append(f"Teklif {cf_created} tarihinde olusturulmus.") | |
| return " ".join(parts) if parts else f"{prefix} lead numarali basvurunun tarih bilgisi bulunamadi." | |
| parts = [f"{prefix} lead numarali basvurunun teklif numarasi {cid}'dir."] | |
| if name: | |
| parts.append(f"Musteri: {name}.") | |
| if lead_created: | |
| parts.append(f"Lead olusturulma tarihi: {lead_created}.") | |
| if cf_created: | |
| parts.append(f"Teklif olusturulma tarihi: {cf_created}.") | |
| if status: | |
| parts.append(f"Lead durumu: {status}.") | |
| return " ".join(parts) | |
| def _format_lead_reverse(row, prefix, query=""): | |
| lead_id = row.get("LeadId", "") | |
| name = row.get("CustomerName", "") | |
| lead_created = row.get("LeadCreatedAt", "") | |
| cf_created = row.get("CreditFileCreatedAt", "") | |
| status = row.get("LeadStatus", "") | |
| if _query_has(query, ["lead id", "lead numarası", "lead numarasi", "lead no"]): | |
| return f"{prefix} numarali teklifin lead ID'si {lead_id}'dir." | |
| if _query_has(query, _LEAD_CUSTOMER_KW): | |
| return f"{prefix} numarali teklifin musterisi {name}'dir." if name else f"{prefix} numarali teklifin musteri bilgisi bulunamadi." | |
| if _query_has(query, _LEAD_DATE_KW): | |
| parts = [] | |
| if lead_created: | |
| parts.append(f"Lead {lead_created} tarihinde olusturulmus.") | |
| if cf_created: | |
| parts.append(f"Teklif {cf_created} tarihinde olusturulmus.") | |
| return " ".join(parts) if parts else f"{prefix} numarali teklifin tarih bilgisi bulunamadi." | |
| parts = [f"{prefix} numarali teklifin lead ID'si {lead_id}'dir."] | |
| if name: | |
| parts.append(f"Musteri: {name}.") | |
| if lead_created: | |
| parts.append(f"Lead olusturulma tarihi: {lead_created}.") | |
| if cf_created: | |
| parts.append(f"Teklif olusturulma tarihi: {cf_created}.") | |
| if status: | |
| parts.append(f"Lead durumu: {status}.") | |
| return " ".join(parts) | |
| def _format_lien(row, prefix, query=""): | |
| plate = row.get("PlateNumber", prefix) | |
| brand = row.get("VehicleBrand", "") | |
| model = row.get("VehicleModel", "") | |
| year = row.get("VehicleYear", "") | |
| vehicle_desc = " ".join(str(p) for p in [year, brand, model] if p).strip() | |
| is_lien = row.get("IsLien") | |
| cid = row.get("CreditFileId", "") | |
| if _query_has(query, _LIEN_CREDIT_KW) and not _query_has(query, _LIEN_STATUS_KW): | |
| if cid: | |
| return f"{plate} plakali aracin ilgili teklif numarasi: {cid}." | |
| return f"{plate} plakali aracin teklif numarasi bulunamadi." | |
| if str(is_lien) in ("1", "True", "true"): | |
| parts = [f"{plate} plakali {vehicle_desc} uzerinde aktif bir rehin bulunmaktadir."] | |
| if cid: | |
| parts.append(f"Ilgili teklif numarasi: {cid}.") | |
| return " ".join(parts) | |
| return f"{plate} plakali {vehicle_desc} uzerinde aktif bir rehin bulunmamaktadir." | |
| _INSURANCE_STATUS_KW = [ | |
| "durum", "aktif mi", "aktif", "pasif", "bekleniyor", "iptal", | |
| "gecerli mi", "ge\u00e7erli mi", "var mi", "var m\u0131", | |
| "ne durumda", "durumu ne", "durumu nedir", | |
| ] | |
| _INSURANCE_DATE_KW = [ | |
| "ne zaman", "tarih", "ba\u015flang\u0131\u00e7", "baslangic", | |
| "biti\u015f", "bitis", "bitiyor", "bitecek", | |
| "ba\u015fl\u0131yor", "basliyor", "s\u00fcre", "sure", | |
| "ka\u00e7 g\u00fcn", "kac gun", "sona eriyor", "ge\u00e7erlilik", | |
| ] | |
| _INSURANCE_AMOUNT_KW = [ | |
| "tutar", "miktar", "tl", "fiyat", "\u00fccret", "ucret", | |
| "bedel", "ne kadar", "ka\u00e7 tl", "kac tl", "para", | |
| ] | |
| _INSURANCE_EXISTENCE_KW = [ | |
| "var mi", "var m\u0131", "mevcut mu", "aktif mi", | |
| "hala ge\u00e7erli mi", "devam ediyor mu", | |
| ] | |
| _INSURANCE_START_KW = ["ba\u015flang\u0131\u00e7", "baslangic", "ba\u015fl\u0131yor", "basliyor"] | |
| _INSURANCE_END_KW = ["biti\u015f", "bitis", "bitiyor", "bitecek", "sona eriyor"] | |
| _TURKISH_MONTH_MAP = { | |
| "ocak": "01", "subat": "02", "şubat": "02", | |
| "mart": "03", "nisan": "04", "mayıs": "05", "mayis": "05", | |
| "haziran": "06", "temmuz": "07", "ağustos": "08", "agustos": "08", | |
| "eylül": "09", "eylul": "09", "ekim": "10", | |
| "kasım": "11", "kasim": "11", "aralık": "12", "aralik": "12", | |
| } | |
| _SINGLE_MONTH_KW = [ | |
| "ayında", "ayinda", "ayı", "ayi", | |
| "ayında kaç", "ayinda kac", "ayındaki", "ayindaki", | |
| "kaç aktif", "kac aktif", "kaç poliçe", "kac police", | |
| "aktif poliçe", "aktif police", "sigorta adedi", | |
| "sigorta sayısı", "sigorta sayisi", | |
| ] | |
| _INSURANCE_MONTHLY_COUNT_KW = [ | |
| "ay bazında", "ay bazinda", "aylık", "aylik", | |
| "aylara göre", "aylara gore", "her ay kaç", "her ay kac", | |
| "ayda kaç", "ayda kac", "ay ay", "aylık dağılım", | |
| "aylik dagilim", "sigorta sayısı", "sigorta sayisi", | |
| "kaç adet sigorta", "kac adet sigorta", "aylık sigorta", | |
| "aylik sigorta", "sigorta adedi", "sigorta adetleri", | |
| ] | |
| def _extract_month_from_query(query: str) -> Optional[str]: | |
| q = query.lower() | |
| for name, num in _TURKISH_MONTH_MAP.items(): | |
| if name in q: | |
| return num | |
| return None | |
| _MONTH_NUMBER_TO_TURKISH = {} | |
| for _name, _num in _TURKISH_MONTH_MAP.items(): | |
| if _num not in _MONTH_NUMBER_TO_TURKISH: | |
| _MONTH_NUMBER_TO_TURKISH[_num] = _name.capitalize() | |
| def _month_number_to_turkish(month_num: str) -> str: | |
| return _MONTH_NUMBER_TO_TURKISH.get(month_num, month_num) | |
| def _is_single_month_query(query: str) -> bool: | |
| q = query.lower() | |
| if _extract_month_from_query(q) is None: | |
| return False | |
| return any(kw in q for kw in _SINGLE_MONTH_KW) | |
| def _is_monthly_aggregate_query(query: str) -> bool: | |
| q = query.lower() | |
| if any(kw in q for kw in _INSURANCE_MONTHLY_COUNT_KW): | |
| return True | |
| return _is_single_month_query(q) | |
| def _insurance_intent(query): | |
| q = query.lower() | |
| has_status = any(kw in q for kw in _INSURANCE_STATUS_KW) | |
| has_date = any(kw in q for kw in _INSURANCE_DATE_KW) | |
| has_amount = any(kw in q for kw in _INSURANCE_AMOUNT_KW) | |
| count = sum([has_status, has_date, has_amount]) | |
| if count != 1: | |
| return "full", q | |
| if has_status: | |
| return "status", q | |
| if has_date: | |
| return "date", q | |
| return "amount", q | |
| def _compute_remaining_days(end_date_str: str) -> Optional[int]: | |
| if not end_date_str or not str(end_date_str).strip(): | |
| return None | |
| try: | |
| date_part = str(end_date_str).strip().split(" ")[0].split("T")[0] | |
| parsed = date.fromisoformat(date_part) | |
| return (parsed - date.today()).days | |
| except (ValueError, IndexError): | |
| return None | |
| def _build_remaining_sentence(remaining: Optional[int]) -> str: | |
| if remaining is None: | |
| return "" | |
| if remaining > 0: | |
| return f" Poli\u00e7enin bitmesine {remaining} g\u00fcn kalm\u0131\u015ft\u0131r." | |
| if remaining == 0: | |
| return " Poli\u00e7e bug\u00fcn sona ermektedir." | |
| return f" Poli\u00e7enin s\u00fcresi {abs(remaining)} g\u00fcn \u00f6nce dolmu\u015ftur." | |
| def _format_insurance_monthly_count(rows, query=""): | |
| if not rows: | |
| month_num = _extract_month_from_query(query) if query else None | |
| if month_num: | |
| month_name = _month_number_to_turkish(month_num) | |
| return f"{month_name} ay\u0131na ait sigorta verisi bulunamad\u0131." | |
| return "Veritaban\u0131nda ayl\u0131k sigorta verisi bulunamad\u0131." | |
| month_num = _extract_month_from_query(query) if query else None | |
| if month_num: | |
| filtered = [r for r in rows if str(r.get("PolicyMonth", "")).endswith(f"-{month_num}")] | |
| if not filtered: | |
| month_name = _month_number_to_turkish(month_num) | |
| return f"{month_name} ay\u0131na ait sigorta verisi bulunamad\u0131." | |
| lines = [] | |
| total = 0 | |
| for row in filtered: | |
| month_label = row.get("PolicyMonth", "-") | |
| count = int(row.get("InsuranceCount", 0)) | |
| total += count | |
| lines.append(f"- {month_label}: {count} poli\u00e7e") | |
| month_name = _month_number_to_turkish(month_num) | |
| if len(filtered) == 1: | |
| return f"{month_name} ay\u0131nda {total} aktif poli\u00e7e bulunmaktad\u0131r." | |
| header = f"{month_name} ay\u0131na ait sigorta da\u011f\u0131l\u0131m\u0131:" | |
| lines.insert(0, header) | |
| lines.append(f"Toplam: {total} poli\u00e7e") | |
| return "\n".join(lines) | |
| lines = ["Aylara g\u00f6re sigorta da\u011f\u0131l\u0131m\u0131 a\u015fa\u011f\u0131daki gibidir:"] | |
| total = 0 | |
| for row in rows: | |
| month = row.get("PolicyMonth", "-") | |
| count = int(row.get("InsuranceCount", 0)) | |
| total += count | |
| lines.append(f"- {month}: {count} poli\u00e7e") | |
| lines.append(f"Toplam: {total} poli\u00e7e") | |
| return "\n".join(lines) | |
| def _format_insurance_policy(row, prefix, query=""): | |
| if not row.get("PolicyStatusName"): | |
| return f"{prefix} numaral\u0131 poli\u00e7eye ait sigorta bilgisi bulunamad\u0131." | |
| intent, q = _insurance_intent(query) | |
| if intent == "status": | |
| end_date = row.get("PolicyEndDate", "") | |
| remaining = _compute_remaining_days(end_date) if end_date else None | |
| base = f"{prefix} numaral\u0131 poli\u00e7enin durumu: {row['PolicyStatusName']}." | |
| if end_date: | |
| base = f"{prefix} numaral\u0131 poli\u00e7enin durumu: {row['PolicyStatusName']}. Biti\u015f tarihi: {end_date}.{_build_remaining_sentence(remaining)}" | |
| return base | |
| if intent == "date": | |
| has_start = any(kw in q for kw in _INSURANCE_START_KW) | |
| has_end = any(kw in q for kw in _INSURANCE_END_KW) | |
| if has_end and not has_start: | |
| return f"{prefix} numaral\u0131 poli\u00e7enin sigorta biti\u015f tarihi: {row.get('PolicyEndDate', '-')}." | |
| if has_start and not has_end: | |
| return f"{prefix} numaral\u0131 poli\u00e7enin sigorta ba\u015flang\u0131\u00e7 tarihi: {row.get('PolicyStartDate', '-')}." | |
| return ( | |
| f"{prefix} numaral\u0131 poli\u00e7enin sigorta ba\u015flang\u0131\u00e7 tarihi: {row.get('PolicyStartDate', '-')}, " | |
| f"biti\u015f tarihi: {row.get('PolicyEndDate', '-')}." | |
| ) | |
| if intent == "amount": | |
| return f"{prefix} numaral\u0131 poli\u00e7enin sigorta tutar\u0131: {row.get('InsuranceAmount', '-')} TL." | |
| return ( | |
| f"{prefix} numaral\u0131 poli\u00e7enin sigorta bilgisi: " | |
| f"Poli\u00e7e Durumu: {row.get('PolicyStatusName', 'Bilinmiyor')}, " | |
| f"Ba\u015flang\u0131\u00e7: {row.get('PolicyStartDate', '-')}, " | |
| f"Biti\u015f: {row.get('PolicyEndDate', '-')}, " | |
| f"Tutar: {row.get('InsuranceAmount', '-')} TL." | |
| ) | |
| def _format_active_insurance(row, prefix, query=""): | |
| if not row.get("PolicyStatusName"): | |
| return f"{prefix} numaral\u0131 poli\u00e7eye ait aktif sigorta bulunamad\u0131." | |
| q = query.lower() | |
| is_existence = any(kw in q for kw in _INSURANCE_EXISTENCE_KW) | |
| intent, q = _insurance_intent(query) | |
| if is_existence and intent in ("status", "full"): | |
| return f"{prefix} numaral\u0131 poli\u00e7eye ait aktif bir sigorta bulunmaktad\u0131r." | |
| if intent == "status": | |
| end_date = row.get("PolicyEndDate", "") | |
| remaining = _compute_remaining_days(end_date) if end_date else None | |
| base = f"{prefix} numaral\u0131 poli\u00e7enin aktif durumu: {row['PolicyStatusName']}." | |
| if end_date: | |
| base = f"{prefix} numaral\u0131 poli\u00e7enin aktif durumu: {row['PolicyStatusName']}. Biti\u015f tarihi: {end_date}.{_build_remaining_sentence(remaining)}" | |
| return base | |
| if intent == "date": | |
| has_start = any(kw in q for kw in _INSURANCE_START_KW) | |
| has_end = any(kw in q for kw in _INSURANCE_END_KW) | |
| if has_end and not has_start: | |
| return f"{prefix} numaral\u0131 poli\u00e7enin aktif sigorta biti\u015f tarihi: {row.get('PolicyEndDate', '-')}." | |
| if has_start and not has_end: | |
| return f"{prefix} numaral\u0131 poli\u00e7enin aktif sigorta ba\u015flang\u0131\u00e7 tarihi: {row.get('PolicyStartDate', '-')}." | |
| return ( | |
| f"{prefix} numaral\u0131 poli\u00e7enin aktif sigorta ba\u015flang\u0131\u00e7 tarihi: {row.get('PolicyStartDate', '-')}, " | |
| f"biti\u015f tarihi: {row.get('PolicyEndDate', '-')}." | |
| ) | |
| if intent == "amount": | |
| return f"{prefix} numaral\u0131 poli\u00e7enin aktif sigorta tutar\u0131: {row.get('InsuranceAmount', '-')} TL." | |
| return ( | |
| f"{prefix} numaral\u0131 poli\u00e7eye ait aktif bir sigorta bulunmaktad\u0131r. " | |
| f"Ba\u015flang\u0131\u00e7: {row.get('PolicyStartDate', '-')}, " | |
| f"Biti\u015f: {row.get('PolicyEndDate', '-')}, " | |
| f"Tutar: {row.get('InsuranceAmount', '-')} TL." | |
| ) | |
| VIEW_REGISTRY: Dict[str, Dict[str, Any]] = { | |
| "v_CurrentCreditFileStatus": { | |
| "keywords": ["statu", "statü", "status", "teklif durum"], | |
| "identifier_type": IdentifierType.CREDIT_FILE_ID, | |
| "identifier_column": "CreditFileId", | |
| "display_field": "StatusName", | |
| "label": "statusu", | |
| "format": lambda row, prefix, query="": ( | |
| f"{prefix} teklifinin statusu: {row['StatusName']}." | |
| if row.get("StatusName") and str(row["StatusName"]).strip() | |
| else f"{prefix} teklifinin statu bilgisi bulunamadi." | |
| ), | |
| }, | |
| "v_CurrentDecisionResult": { | |
| "keywords": [ | |
| "karar", "decision", "sonucu", "sonuc", | |
| "gerekçe", "gerekcesi", "gerekce", "gerekce nedir", | |
| "neden", "neden reddedildi", "neden reddedilmis", | |
| "neden şartlı", "neden onay", "neden onaylanmadi", | |
| "neden kabul edilmedi", "neden sartli", | |
| "karar nedeni", "karar sebebi", | |
| "reddedildi", "reddedilmis", "reddedilen", "red", | |
| "red sebebi", "red nedeni", "neden red", | |
| "kabul edilmedi", "onaylanmadi", "onaylanmamis", | |
| "hangi sebeple", "hangi nedenle", "hangi gerekce", | |
| "sartli onay neden", "sartli neden", | |
| ], | |
| "identifier_type": IdentifierType.CREDIT_FILE_ID, | |
| "identifier_column": "CreditFileId", | |
| "display_field": "DecisionResultName", | |
| "label": "karar sonucu", | |
| "format": lambda row, prefix, query="": ( | |
| ( | |
| f"{prefix} teklifinin karar sonucu: {row['DecisionResultName']}." | |
| + ( | |
| f" Karar gerekçesi: {row['DecisionReason']}." | |
| if row.get("DecisionReason") and str(row["DecisionReason"]).strip() | |
| else " Karar gerekçesi belirtilmemiş." | |
| ) | |
| ) | |
| if row.get("DecisionResultName") and str(row["DecisionResultName"]).strip() | |
| else f"{prefix} teklifinin karar sonucu bilgisi bulunamadi." | |
| ), | |
| }, | |
| "v_CurrentPaymentApprove": { | |
| "keywords": ["odeme onay", "ödeme onay", "odeme", "ödeme", "payment", "onay verildi", "onay durumu"], | |
| "identifier_type": IdentifierType.CREDIT_FILE_ID, | |
| "identifier_column": "CreditFileId", | |
| "display_field": "PaymentApprove", | |
| "label": "odeme onayi", | |
| "format": lambda row, prefix, query="": ( | |
| f"{prefix} teklifinin odeme onayi sonucu: " | |
| + ("Odeme Onayi Verildi" if str(row.get("PaymentApprove")) == "1" else "Odeme Onayi Henuz Verilmedi") | |
| + "." | |
| if row.get("PaymentApprove") is not None | |
| else f"{prefix} teklifinin odeme onayi bilgisi bulunamadi." | |
| ), | |
| }, | |
| "v_CurrentDownPaymentStatus": { | |
| "keywords": ["pesinat", "peşinat", "downpayment", "down payment"], | |
| "identifier_type": IdentifierType.CREDIT_FILE_ID, | |
| "identifier_column": "CreditFileId", | |
| "display_field": "PesinatDurumu", | |
| "label": "pesinat durumu", | |
| "format": lambda row, prefix, query="": ( | |
| f"{prefix} teklifinin pesinat durumu: {row['PesinatDurumu']}." | |
| if row.get("PesinatDurumu") and str(row["PesinatDurumu"]).strip() | |
| else ( | |
| f"{prefix} teklifinin pesinat durumu: " | |
| + ("Pesinat Tamamlandi" if str(row.get("IsReceivedSwtDownpayment")) == "1" else "Pesinat Henuz Tamamlanmadi") | |
| + "." | |
| if row.get("IsReceivedSwtDownpayment") is not None | |
| else f"{prefix} teklifinin pesinat bilgisi bulunamadi." | |
| ) | |
| ), | |
| }, | |
| "v_LeadCreditFileMapping": { | |
| "keywords": ["teklif numarası", "teklif numarasi", "teklif no", "basvuru numarası", "basvuru numarasi", "kredi numarası", "kredi numarasi"], | |
| "identifier_type": IdentifierType.LEAD_ID, | |
| "identifier_column": "LeadId", | |
| "display_field": "CreditFileId", | |
| "label": "teklif numarasi", | |
| "format": _format_lead, | |
| "no_result_format": lambda prefix: f"{prefix} lead numarasina ait bir teklif bulunamadi.", | |
| }, | |
| "v_LeadCreditFileMappingReverse": { | |
| "view_name": "v_LeadCreditFileMapping", | |
| "keywords": ["lead id", "lead numarası", "lead numarasi", "lead no", "lead kimlik", "teklifin lead", "basvurunun lead"], | |
| "identifier_type": IdentifierType.CREDIT_FILE_ID, | |
| "identifier_column": "CreditFileId", | |
| "display_field": "LeadId", | |
| "label": "lead ID", | |
| "format": _format_lead_reverse, | |
| "no_result_format": lambda prefix: f"{prefix} numarali teklif icin lead kaydi bulunamadi.", | |
| }, | |
| "v_VehicleLienStatus": { | |
| "keywords": ["rehin", "rehni", "rehin durumu", "arac rehni", "plaka", "plakali", "aracta rehin"], | |
| "identifier_type": IdentifierType.PLATE_NUMBER, | |
| "identifier_column": "PlateNumber", | |
| "display_field": "IsLien", | |
| "label": "rehin durumu", | |
| "format": _format_lien, | |
| "no_result_format": lambda prefix: f"{prefix} plakali arac sistemde bulunamadi.", | |
| }, | |
| "v_ActiveInsuranceSummary": { | |
| "keywords": ["aktif sigorta", "aktif poli\u00e7e", "ge\u00e7erli sigorta", "gecerli sigorta", "sigorta aktif mi", "sigorta devam ediyor mu", "sigorta devam ediyor", "sigorta hala ge\u00e7erli mi"], | |
| "identifier_type": IdentifierType.PROPOSAL_NO, | |
| "identifier_column": "ProposalNo", | |
| "display_field": "PolicyStatusName", | |
| "label": "aktif sigorta durumu", | |
| "format": _format_active_insurance, | |
| "no_result_format": lambda prefix: f"{prefix} numaral\u0131 poli\u00e7eye ait aktif sigorta kayd\u0131 bulunamad\u0131.", | |
| }, | |
| "v_InsurancePolicyStatus": { | |
| "keywords": ["sigorta", "poli\u00e7e", "police", "sigorta durumu", "sigorta bilgisi", "sigorta ba\u015flang\u0131\u00e7", "sigorta biti\u015f", "sigorta tutar\u0131", "sigorta miktar\u0131", "poli\u00e7e durumu", "sigortas\u0131", "sigortasi", "sigorta var m\u0131", "sigorta var mi", "sigorta s\u00fcresi", "poli\u00e7e bilgisi"], | |
| "identifier_type": IdentifierType.PROPOSAL_NO, | |
| "identifier_column": "ProposalNo", | |
| "display_field": "PolicyStatusName", | |
| "label": "sigorta durumu", | |
| "format": _format_insurance_policy, | |
| "no_result_format": lambda prefix: f"{prefix} numaral\u0131 poli\u00e7eye ait sigorta kayd\u0131 bulunamad\u0131.", | |
| }, | |
| "v_InsuranceCountByMonth": { | |
| "keywords": _INSURANCE_MONTHLY_COUNT_KW, | |
| "identifier_type": None, | |
| "identifier_column": None, | |
| "display_field": "InsuranceCount", | |
| "label": "ayl\u0131k sigorta say\u0131s\u0131", | |
| "format": _format_insurance_monthly_count, | |
| "aggregate": True, | |
| }, | |
| } | |
| GENERAL_DB_KEYWORDS = ["teklif", "dosya", "kredi"] | |
| _DEALER_ACCESS_DENIED_MSG = ( | |
| "Bu teklif, yetkili oldu\u011funuz bayi kapsam\u0131nda de\u011fildir. " | |
| "Eri\u015fim izniniz bulunmamaktad\u0131r. " | |
| "L\u00fctfen kendi bayinize ait teklif numaralar\u0131n\u0131 sorgulay\u0131n." | |
| ) | |
| def check_dealer_access(credit_file_id: str) -> bool: | |
| if not DEALER_ACCESS_CONFIG.get("enabled", False): | |
| return True | |
| current_dealer_id = DEALER_ACCESS_CONFIG.get("current_dealer_id") | |
| if current_dealer_id is None: | |
| return False | |
| adapter = get_database_adapter() | |
| if adapter is None or not adapter.is_connected: | |
| return False | |
| try: | |
| sql = ( | |
| "SELECT DealerId FROM CreditFileVersion " | |
| "WHERE CreditFileId = :cid " | |
| "AND IsCurrentVersion = 1 AND IsPassive = 0 " | |
| "LIMIT 1" | |
| ) | |
| rows = adapter.execute_query(sql, {"cid": credit_file_id}) | |
| if not rows: | |
| return True | |
| dealer_id = rows[0].get("DealerId") | |
| return int(dealer_id) == int(current_dealer_id) | |
| except Exception as e: | |
| logger.warning("Dealer access check failed for CreditFileId=%s: %s", credit_file_id, e) | |
| return False | |
| def extract_lead_id(text: str) -> Optional[str]: | |
| m = _LEAD_ID_RE.search(text) | |
| return m.group(1).upper() if m else None | |
| def extract_plate_number(text: str) -> Optional[str]: | |
| m = _PLATE_RE.search(text) | |
| if not m: | |
| return None | |
| return (m.group(1) + m.group(2) + m.group(3)).upper() | |
| def extract_proposal_no(text: str) -> Optional[str]: | |
| m = _PROPOSAL_NO_RE.search(text) | |
| return m.group(1) if m else None | |
| def extract_credit_file_id(text: str) -> Optional[str]: | |
| m = _CREDIT_FILE_ID_RE.search(text) | |
| return m.group(1) if m else None | |
| def extract_identifier(text: str) -> Tuple[Optional[IdentifierType], Optional[str]]: | |
| lead = extract_lead_id(text) | |
| if lead: | |
| return IdentifierType.LEAD_ID, lead | |
| plate = extract_plate_number(text) | |
| if plate and _has_plate_context(text): | |
| return IdentifierType.PLATE_NUMBER, plate | |
| proposal = extract_proposal_no(text) | |
| if proposal: | |
| return IdentifierType.PROPOSAL_NO, proposal | |
| cid = extract_credit_file_id(text) | |
| if cid: | |
| return IdentifierType.CREDIT_FILE_ID, cid | |
| return None, None | |
| def _has_plate_context(text: str) -> bool: | |
| lower = text.lower() | |
| plate_keywords = ["plaka", "plakali", "plakalı", "rehin", "rehni", "arac", "araç"] | |
| return any(kw in lower for kw in plate_keywords) | |
| def match_view(query: str, identifier_type: Optional[IdentifierType] = None) -> Optional[str]: | |
| query_lower = query.lower() | |
| for view_name, meta in VIEW_REGISTRY.items(): | |
| if identifier_type is not None and meta.get("identifier_type") is not None and meta["identifier_type"] != identifier_type: | |
| continue | |
| if any(kw in query_lower for kw in meta["keywords"]): | |
| return view_name | |
| return None | |
| def match_all_views(query: str, identifier_type: Optional[IdentifierType] = None) -> List[str]: | |
| query_lower = query.lower() | |
| matched = [] | |
| for view_name, meta in VIEW_REGISTRY.items(): | |
| if identifier_type is not None and (meta.get("identifier_type") is None or meta["identifier_type"] != identifier_type): | |
| continue | |
| if any(kw in query_lower for kw in meta["keywords"]): | |
| matched.append(view_name) | |
| return matched | |
| def views_for_identifier_type(id_type: IdentifierType) -> List[str]: | |
| return [name for name, meta in VIEW_REGISTRY.items() if meta["identifier_type"] == id_type] | |
| _ALL_DB_KEYWORDS: Optional[List[str]] = None | |
| def _get_all_db_keywords() -> List[str]: | |
| global _ALL_DB_KEYWORDS | |
| if _ALL_DB_KEYWORDS is None: | |
| kws = list(GENERAL_DB_KEYWORDS) | |
| for meta in VIEW_REGISTRY.values(): | |
| kws.extend(meta["keywords"]) | |
| _ALL_DB_KEYWORDS = kws | |
| return _ALL_DB_KEYWORDS | |
| def has_database_signals(query: str, id_type: Optional[IdentifierType] = None, id_value: Optional[str] = None) -> bool: | |
| query_lower = query.lower() | |
| if _is_monthly_aggregate_query(query_lower): | |
| return True | |
| if id_type is None or id_value is None: | |
| id_type, id_value = extract_identifier(query) | |
| if not id_value: | |
| return False | |
| return any(kw in query_lower for kw in _get_all_db_keywords()) | |
| def has_view_specific_keywords(query: str) -> bool: | |
| query_lower = query.lower() | |
| for meta in VIEW_REGISTRY.values(): | |
| if any(kw in query_lower for kw in meta["keywords"]): | |
| return True | |
| return False | |
| def _resolve_physical_view(view_name: str) -> str: | |
| meta = VIEW_REGISTRY.get(view_name, {}) | |
| return meta.get("view_name", view_name) | |
| def build_query(view_name: str, identifier_value: str, identifier_column: Optional[str] = None) -> Tuple[str, Dict[str, str]]: | |
| physical = _resolve_physical_view(view_name) | |
| col = identifier_column or VIEW_REGISTRY.get(view_name, {}).get("identifier_column", "CreditFileId") | |
| sql = f"SELECT * FROM {physical} WHERE {col} = :id_val" | |
| return sql, {"id_val": identifier_value} | |
| def execute_view_query(view_name: str, identifier_value: str, identifier_column: Optional[str] = None) -> Dict[str, Any]: | |
| sql, params = build_query(view_name, identifier_value, identifier_column) | |
| adapter = get_database_adapter() | |
| if adapter is None or not adapter.is_connected: | |
| return {"success": False, "sql": sql, "rows": [], "error": "Database not connected"} | |
| try: | |
| rows = adapter.execute_query(sql, params) | |
| return {"success": True, "sql": sql, "rows": rows, "error": None} | |
| except Exception as e: | |
| logger.error(f"View query failed: {e}") | |
| return {"success": False, "sql": sql, "rows": [], "error": str(e)} | |
| def execute_view_query_raw(sql: str) -> Dict[str, Any]: | |
| stripped = sql.strip() | |
| if not stripped.upper().startswith("SELECT"): | |
| logger.warning("execute_view_query_raw rejected non-SELECT query: %s", stripped[:80]) | |
| return {"success": False, "sql": sql, "rows": [], "error": "Only SELECT queries are allowed"} | |
| adapter = get_database_adapter() | |
| if adapter is None or not adapter.is_connected: | |
| return {"success": False, "sql": sql, "rows": [], "error": "Database not connected"} | |
| try: | |
| rows = adapter.execute_query(sql) | |
| return {"success": True, "sql": sql, "rows": rows, "error": None} | |
| except Exception as e: | |
| logger.error(f"Raw view query failed: {e}") | |
| return {"success": False, "sql": sql, "rows": [], "error": str(e)} | |
| def format_response(view_name: str, rows: List[Dict[str, Any]], identifier_value: str, query: str = "") -> str: | |
| prefix = identifier_value if identifier_value else "Ilgili teklif" | |
| meta = VIEW_REGISTRY.get(view_name) | |
| if not rows: | |
| if meta and "no_result_format" in meta: | |
| return meta["no_result_format"](prefix) | |
| if identifier_value: | |
| return f"{identifier_value} numarali teklif icin veritabaninda kayit bulunamadi." | |
| return "Veritabaninda ilgili kayit bulunamadi." | |
| row = rows[0] | |
| if meta and "format" in meta: | |
| return meta["format"](row, prefix, query) | |
| row_str = ", ".join(f"{k}: {v}" for k, v in row.items() if v is not None) | |
| return f"{prefix} teklifi icin sonuc: {row_str}." if row_str else "Veritabani sonucu alindi ancak beklenen alan bulunamadi." | |
| def query_and_format(query: str, credit_file_id: Optional[str] = None, identifier_type: Optional[IdentifierType] = None, identifier_value: Optional[str] = None) -> Dict[str, Any]: | |
| id_type = identifier_type | |
| id_val = identifier_value | |
| if not id_val and credit_file_id: | |
| id_type = IdentifierType.CREDIT_FILE_ID | |
| id_val = credit_file_id | |
| if not id_val: | |
| id_type, id_val = extract_identifier(query) | |
| query_lower = query.lower() | |
| if not id_val and _is_monthly_aggregate_query(query_lower): | |
| for vname, vmeta in VIEW_REGISTRY.items(): | |
| if vmeta.get("aggregate"): | |
| physical = vmeta.get("view_name", vname) | |
| sql = f"SELECT * FROM {physical}" | |
| result = execute_view_query_raw(sql) | |
| rows = result.get("rows", []) | |
| response = vmeta["format"](rows, query) | |
| return {"success": result["success"], "response": response, "sql": sql, "rows": rows, "view_name": vname, "is_aggregate": True} | |
| if not id_val: | |
| return {"success": False, "response": "Sorgudan tanimlayici bilgi cikarilmadi.", "sql": None, "rows": None} | |
| view_name = match_view(query, identifier_type=id_type) | |
| if not view_name: | |
| candidate_views = views_for_identifier_type(id_type) if id_type else [ | |
| n for n, m in VIEW_REGISTRY.items() if not m.get("aggregate") | |
| ] | |
| results_all = [] | |
| sql_all = [] | |
| for v in candidate_views: | |
| col = VIEW_REGISTRY[v]["identifier_column"] | |
| r = execute_view_query(v, id_val, identifier_column=col) | |
| sql_all.append(r["sql"]) | |
| if r["success"] and r["rows"]: | |
| results_all.append((v, r["rows"])) | |
| if results_all: | |
| parts = [format_response(v, rows, id_val, query) for v, rows in results_all] | |
| return {"success": True, "response": " ".join(parts), "sql": "; ".join(sql_all), "rows": results_all[0][1]} | |
| no_result_msg = f"{id_val} icin veritabaninda kayit bulunamadi." | |
| return {"success": True, "response": no_result_msg, "sql": "; ".join(sql_all), "rows": []} | |
| col = VIEW_REGISTRY[view_name]["identifier_column"] | |
| result = execute_view_query(view_name, id_val, identifier_column=col) | |
| response = format_response(view_name, result["rows"], id_val, query) | |
| return {"success": result["success"], "response": response, "sql": result["sql"], "rows": result["rows"], "error": result.get("error")} | |