Bot de support en Python : code source

1"""SQLite-backed store data: products, delivery fees, orders and support tickets."""23from __future__ import annotations45import json6import re7import sqlite38import threading9import unicodedata10from collections.abc import Iterator11from contextlib import contextmanager12from datetime import datetime, timezone13from pathlib import Path14from typing import Any1516SCHEMA = """17CREATE TABLE IF NOT EXISTS store_info (18    key   TEXT PRIMARY KEY,19    value TEXT NOT NULL20);21CREATE TABLE IF NOT EXISTS products (22    sku       TEXT PRIMARY KEY,23    name      TEXT NOT NULL,24    category  TEXT NOT NULL,25    price_dzd INTEGER NOT NULL,26    colors    TEXT NOT NULL,27    stock     TEXT NOT NULL28);29CREATE TABLE IF NOT EXISTS delivery_zones (30    code          INTEGER PRIMARY KEY,31    name          TEXT NOT NULL,32    home_dzd      INTEGER NOT NULL,33    stop_desk_dzd INTEGER NOT NULL,34    days          TEXT NOT NULL35);36CREATE TABLE IF NOT EXISTS delivery_aliases (37    alias TEXT PRIMARY KEY,38    code  INTEGER NOT NULL REFERENCES delivery_zones(code)39);40CREATE TABLE IF NOT EXISTS orders (41    id                  TEXT PRIMARY KEY,42    customer_first_name TEXT NOT NULL,43    phone_last4         TEXT NOT NULL,44    status              TEXT NOT NULL,45    items               TEXT NOT NULL,46    total_dzd           INTEGER NOT NULL,47    wilaya              TEXT NOT NULL,48    delivery_type       TEXT NOT NULL,49    created_at          TEXT NOT NULL,50    shipped_at          TEXT,51    expected_delivery   TEXT,52    tracking_code       TEXT53);54CREATE TABLE IF NOT EXISTS support_tickets (55    id         INTEGER PRIMARY KEY AUTOINCREMENT,56    chat_id    INTEGER NOT NULL,57    username   TEXT,58    reason     TEXT NOT NULL,59    created_at TEXT NOT NULL,60    status     TEXT NOT NULL DEFAULT 'open'61);62"""6364ORDER_ID_RE = re.compile(r"^[A-Z]{2}-\d{5}$")65_ARABIC_FOLD = str.maketrans("أإآى", "اااي")66_ARABIC_DIACRITICS = re.compile(r"[ً-ْٰـ]")676869def normalize(text: str) -> str:70    """Lowercase, strip Latin accents and Arabic diacritics, fold hamza and alef variants.7172    Folding is applied to both the stored aliases and the query, so "الجزائر" and the73    common spelling "الجزاير" match the same wilaya.74    """75    text = unicodedata.normalize("NFKC", text).strip().lower()76    text = _ARABIC_DIACRITICS.sub("", text)77    text = text.translate(_ARABIC_FOLD)78    decomposed = unicodedata.normalize("NFD", text)79    text = "".join(ch for ch in decomposed if unicodedata.category(ch) != "Mn")80    text = re.sub(r"^(wilaya\s+(de\s+|d['’])?|ولاية\s+)", "", text)81    text = re.sub(r"[^\w\s-]", " ", text)82    return re.sub(r"\s+", " ", text).strip()838485class Store:86    """Thread-safe access to the store database."""8788    def __init__(self, db_path: Path | str) -> None:89        self._db_path = str(db_path)90        self._lock = threading.Lock()91        self._conn = sqlite3.connect(self._db_path, check_same_thread=False)92        self._conn.row_factory = sqlite3.Row93        self._conn.execute("PRAGMA foreign_keys = ON")94        self._conn.executescript(SCHEMA)9596    def close(self) -> None:97        with self._lock:98            self._conn.close()99100    @contextmanager101    def _tx(self) -> Iterator[sqlite3.Connection]:102        with self._lock:103            try:104                yield self._conn105                self._conn.commit()106            except Exception:107                self._conn.rollback()108                raise109110    # ------------------------------------------------------------------ seeding111112    def is_empty(self) -> bool:113        with self._lock:114            row = self._conn.execute("SELECT COUNT(*) AS n FROM products").fetchone()115        return row["n"] == 0116117    def load_catalog(self, catalog_path: Path | str, *, replace: bool = False) -> None:118        data = json.loads(Path(catalog_path).read_text(encoding="utf-8"))119        with self._tx() as conn:120            if replace:121                for table in ("delivery_aliases", "delivery_zones", "products", "orders", "store_info"):122                    conn.execute(f"DELETE FROM {table}")123            for key, value in data.get("store", {}).items():124                conn.execute("INSERT OR REPLACE INTO store_info(key, value) VALUES (?, ?)", (key, str(value)))125            for p in data.get("products", []):126                conn.execute(127                    "INSERT OR REPLACE INTO products(sku, name, category, price_dzd, colors, stock) VALUES (?, ?, ?, ?, ?, ?)",128                    (p["sku"], p["name"], p["category"], int(p["price_dzd"]), json.dumps(p["colors"]), json.dumps(p["stock"])),129                )130            for z in data.get("delivery", []):131                conn.execute(132                    "INSERT OR REPLACE INTO delivery_zones(code, name, home_dzd, stop_desk_dzd, days) VALUES (?, ?, ?, ?, ?)",133                    (int(z["code"]), z["name"], int(z["home_dzd"]), int(z["stop_desk_dzd"]), z["days"]),134                )135                for alias in {z["name"], *z.get("aliases", [])}:136                    conn.execute(137                        "INSERT OR REPLACE INTO delivery_aliases(alias, code) VALUES (?, ?)",138                        (normalize(alias), int(z["code"])),139                    )140            for o in data.get("orders", []):141                conn.execute(142                    """INSERT OR REPLACE INTO orders(id, customer_first_name, phone_last4, status, items, total_dzd,143                       wilaya, delivery_type, created_at, shipped_at, expected_delivery, tracking_code)144                       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",145                    (146                        o["id"].upper(),147                        o["customer_first_name"],148                        o["phone_last4"],149                        o["status"],150                        json.dumps(o["items"]),151                        int(o["total_dzd"]),152                        o["wilaya"],153                        o["delivery_type"],154                        o["created_at"],155                        o.get("shipped_at"),156                        o.get("expected_delivery"),157                        o.get("tracking_code"),158                    ),159                )160161    # ------------------------------------------------------------------ queries162163    def store_info(self) -> dict[str, str]:164        with self._lock:165            rows = self._conn.execute("SELECT key, value FROM store_info").fetchall()166        return {r["key"]: r["value"] for r in rows}167168    def search_products(self, query: str = "", size: str | None = None, limit: int = 5) -> list[dict[str, Any]]:169        limit = max(1, min(limit, 10))170        with self._lock:171            rows = self._conn.execute("SELECT * FROM products ORDER BY category, price_dzd").fetchall()172173        terms = [t for t in normalize(query).split() if len(t) > 1]174        size_key = size.strip().upper() if size else None175        scored: list[tuple[int, dict[str, Any]]] = []176        for row in rows:177            colors: list[str] = json.loads(row["colors"])178            haystack = normalize(f"{row['sku']} {row['name']} {row['category']} {' '.join(colors)}")179            score = sum(1 for t in terms if t in haystack) if terms else 1180            if score == 0:181                continue182            stock: dict[str, int] = json.loads(row["stock"])183            item: dict[str, Any] = {184                "sku": row["sku"],185                "name": row["name"],186                "category": row["category"],187                "price_dzd": row["price_dzd"],188                "colors": colors,189                "sizes_in_stock": [s for s, qty in stock.items() if qty > 0],190                "sizes_out_of_stock": [s for s, qty in stock.items() if qty <= 0],191            }192            if size_key is not None:193                item["requested_size"] = size_key194                item["requested_size_available"] = stock.get(size_key, 0) > 0195            scored.append((score, item))196        scored.sort(key=lambda pair: pair[0], reverse=True)197        return [item for _, item in scored[:limit]]198199    def delivery_fee(self, wilaya: str) -> dict[str, Any] | None:200        key = wilaya.strip()201        with self._lock:202            if key.isdigit():203                zone = self._conn.execute("SELECT * FROM delivery_zones WHERE code = ?", (int(key),)).fetchone()204            else:205                zone = self._conn.execute(206                    """SELECT z.* FROM delivery_aliases a JOIN delivery_zones z ON z.code = a.code207                       WHERE a.alias = ?""",208                    (normalize(key),),209                ).fetchone()210        if zone is None:211            return None212        return {213            "wilaya": zone["name"],214            "code": zone["code"],215            "home_delivery_dzd": zone["home_dzd"],216            "stop_desk_dzd": zone["stop_desk_dzd"],217            "estimated_days": zone["days"],218        }219220    def covered_wilayas(self) -> list[str]:221        with self._lock:222            rows = self._conn.execute("SELECT name FROM delivery_zones ORDER BY code").fetchall()223        return [r["name"] for r in rows]224225    def order_status(self, order_id: str, phone_last4: str) -> dict[str, Any] | None:226        """Return the order only when the phone digits match, so order numbers cannot be enumerated."""227        oid = order_id.strip().upper().replace(" ", "")228        digits = re.sub(r"\D", "", phone_last4)[-4:]229        if not ORDER_ID_RE.match(oid) or len(digits) != 4:230            return None231        with self._lock:232            row = self._conn.execute("SELECT * FROM orders WHERE id = ?", (oid,)).fetchone()233        if row is None or row["phone_last4"] != digits:234            return None235236        items = json.loads(row["items"])237        names = self._product_names([i["sku"] for i in items])238        return {239            "order_id": row["id"],240            "customer_first_name": row["customer_first_name"],241            "status": row["status"],242            "items": [{"name": names.get(i["sku"], i["sku"]), "size": i["size"], "color": i["color"], "qty": i["qty"]} for i in items],243            "total_dzd": row["total_dzd"],244            "wilaya": row["wilaya"],245            "delivery_type": row["delivery_type"],246            "created_at": row["created_at"],247            "shipped_at": row["shipped_at"],248            "expected_delivery": row["expected_delivery"],249            "tracking_code": row["tracking_code"],250        }251252    def _product_names(self, skus: list[str]) -> dict[str, str]:253        if not skus:254            return {}255        placeholders = ",".join("?" for _ in skus)256        with self._lock:257            rows = self._conn.execute(f"SELECT sku, name FROM products WHERE sku IN ({placeholders})", skus).fetchall()258        return {r["sku"]: r["name"] for r in rows}259260    def create_ticket(self, chat_id: int, username: str | None, reason: str) -> int:261        created_at = datetime.now(timezone.utc).isoformat(timespec="seconds")262        with self._tx() as conn:263            cur = conn.execute(264                "INSERT INTO support_tickets(chat_id, username, reason, created_at) VALUES (?, ?, ?, ?)",265                (chat_id, username, reason[:500], created_at),266            )267            if cur.lastrowid is None:268                raise RuntimeError("ticket insert returned no id")269            return cur.lastrowid270271272def open_store(db_path: Path, catalog_path: Path) -> Store:273    """Open the database and load the demo catalog the first time."""274    db_path.parent.mkdir(parents=True, exist_ok=True)275    store = Store(db_path)276    if store.is_empty():277        store.load_catalog(catalog_path)278    return store