„Разговор с базата“ — от български към SQL

В урок 5 питахте асистента на български, копирахте заявката и я проверявахте на ръка. Днес всичко това го прави програмата: получава въпрос, праща го на модела заедно със схемата на базата, взима върнатия SELECT, проверява го, изпълнява го и показва резултата. Сглобяваме я от парчетата, които вече имате — API от урок 19, база от урок 22.

Моделът не може да познае имената на колоните ви. Затова в системното указание влиза схемата — точно текстът на CREATE TABLE, който SQLite пази в sqlite_master. Оттам няма как да е с грешка.

SHEMA_TEKST = "\n".join(
    red[0] + ";" for red in cur.execute(
        "SELECT sql FROM sqlite_master WHERE type = 'table'"))

SISTEMNO = f"""Превеждаш въпроси на български към SQL заявки.
База: SQLite. Схема:
{SHEMA_TEKST}
Правила:
- връщай само една SELECT заявка, без обяснения и без markdown
- никакви INSERT, UPDATE, DELETE, DROP
- ползвай само таблиците и колоните от схемата
- средни стойности закръгляй с ROUND(..., 2)
- давай на колоните разбираеми имена с AS"""

• Диалект — казваме „SQLite“, защото иначе моделът пише MySQL функции, които тук не съществуват.
• Само SELECT — указанието е първата линия; проверката в кода и правата в базата са втората и третата.
• Без markdown — иначе отговорът идва увит в ```sql и базата се оплаква от първия ред.

Върнатият текст е вход от чужд източник — отнасяме се към него като към данни от уеб формуляр. Три въпроса преди да стигне до базата:

import re

ZABRANENI = re.compile(
    r"\b(insert|update|delete|drop|alter|create|replace|attach|pragma)\b", re.I)

def proveri(sql):
    """Връща None, ако заявката е допустима, иначе — причината за отказ."""
    s = sql.strip().rstrip(";").strip()
    if not s.lower().startswith("select"):
        return "не започва със SELECT"
    if ZABRANENI.search(s):
        return "съдържа забранена дума"
    if ";" in s:
        return "повече от една заявка"
    return None

ro = sqlite3.connect("file:uchilishte.db?mode=ro", uri=True)
ro.execute("DELETE FROM marks")
# sqlite3.OperationalError: attempt to write a readonly database

Най-честите грешки не са опасни, а просто неверни:

За първия и третия вид програмата има прост отговор: съобщението на базата се връща на модела заедно с грешната заявка и се иска поправка. Един път. Ако и вторият опит се провали, програмата казва „не мога“ — не влиза в цикъл, който гори лимита.

def otgovori(vapros):
    sql = kam_sql(vapros)
    for opit in (1, 2):
        problem = proveri(sql)
        if problem:
            return f"Отказана заявка ({problem}): {sql}"
        try:
            df = pd.read_sql(sql, ro)
            print("SQL:", sql)
            return df
        except Exception as greshka:          # базата отказа заявката
            if opit == 2:
                return f"Не успях и след поправка: {greshka}"
            sql = kam_sql(vapros, predishna=sql, greshka=str(greshka))

• Моделът получава схемата (текста на CREATE TABLE), диалекта и правилата — не данните.
• Върнатият SQL е чужд вход: проверка, че е един SELECT без забранени думи, преди изпълнение.
• Изпълнението е през връзка само за четене (mode=ro в SQLite, потребител със SELECT в MySQL).
• Грешка от базата се връща на модела за поправка — веднъж. После програмата се отказва честно.
• Синтактично вярна, но смислово грешна заявка се хваща само от човек — затова заявката винаги се показва.