資料分析與智慧問數

import os
import re
import sqlite3
from openai import OpenAI

client = OpenAI(base_url="https://www.tutujin.com/v1", api_key=os.environ["TUTUJIN_API_KEY"])

# Only the tables relevant to the question (with many tables, find them by embedding search with text-embedding-3-small).
SCHEMA = """
CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT, region TEXT);  -- region: east, north, south, west
CREATE TABLE orders (id INTEGER PRIMARY KEY, store_id INTEGER, paid_at TEXT, amount REAL, status TEXT);
-- status: paid or refunded. Sales = sum(amount) of paid orders. paid_at looks like 2026-10-31 18:05:00.
"""
ALLOWED = {"stores", "orders"}
WRITES = re.compile(r"\b(insert|update|delete|drop|alter|create|truncate|attach|pragma|grant)\b", re.I)


def write_sql(question, problem=None):
    prompt = "Question: " + question
    if problem:
        prompt += "\nYour previous query was rejected (" + problem + "). Write it again."
    resp = client.chat.completions.create(
        model="deepseek-v4.1-flash",
        messages=[
            {"role": "system", "content": "Write one SQLite SELECT query for the question, using only these tables:\n"
             + SCHEMA + "\nAlways add a LIMIT. Reply with the SQL only."},
            {"role": "user", "content": prompt},
        ],
    )
    sql = resp.choices[0].message.content.strip()
    return re.sub(r"^\x60{3}(?:sql)?\s*|\s*\x60{3}$", "", sql).strip().rstrip(";")  # drop a code fence


def check(sql):
    """Why the query is unsafe, or None. A SQL parser is sturdier than regexes; keep the read-only user anyway."""
    if ";" in sql:
        return "more than one statement"
    if not re.match(r"(select|with)\b", sql, re.I):
        return "not a SELECT query"
    if WRITES.search(sql):
        return "contains a write keyword"
    ctes = set(re.findall(r"\b(\w+)\s+as\s*\(", sql, re.I))
    tables = set(re.findall(r"\b(?:from|join)\s+(\w+)", sql, re.I)) - ctes
    if not tables <= ALLOWED:
        return "unknown table " + ", ".join(sorted(tables - ALLOWED))
    if not re.search(r"\blimit\s+\d+", sql, re.I):
        return "no LIMIT"
    return None


def ask(question):
    sql = write_sql(question)
    problem = check(sql)
    if problem:  # one rewrite, then give up
        sql = write_sql(question, problem)
        problem = check(sql)
        if problem:
            raise ValueError("no safe query: " + problem)
    # read-only connection; in production, a SELECT-only database user on a replica, with a query timeout
    db = sqlite3.connect("file:shop.db?mode=ro", uri=True)
    rows = db.execute(sql).fetchmany(50)
    db.close()
    resp = client.chat.completions.create(
        model="qwen3.5-plus",
        messages=[
            {"role": "system", "content": "Explain the result in two or three sentences for a business reader. Use only the numbers given."},
            {"role": "user", "content": f"Question: {question}\nSQL: {sql}\nRows: {rows}"},
        ],
    )
    return sql, rows, resp.choices[0].message.content


sql, rows, summary = ask("Which 5 stores in the east region had the highest sales last month?")
print(sql, rows, summary, sep="\n\n")