78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import sqlite3
|
|
import os
|
|
from datetime import datetime
|
|
|
|
DB_PATH = "./data/tickets.db"
|
|
|
|
|
|
def _conn() -> sqlite3.Connection:
|
|
os.makedirs("./data", exist_ok=True)
|
|
return sqlite3.connect(DB_PATH)
|
|
|
|
|
|
def init_db() -> None:
|
|
with _conn() as db:
|
|
db.execute("""
|
|
CREATE TABLE IF NOT EXISTS tickets (
|
|
id TEXT PRIMARY KEY,
|
|
user_id INTEGER,
|
|
username TEXT,
|
|
chat_id INTEGER,
|
|
chantier TEXT,
|
|
description TEXT,
|
|
produit TEXT,
|
|
status TEXT DEFAULT 'ouvert',
|
|
created_at TEXT,
|
|
resolved_at TEXT,
|
|
cause TEXT,
|
|
solution TEXT,
|
|
duree_min INTEGER
|
|
)
|
|
""")
|
|
|
|
|
|
def _next_id() -> str:
|
|
year = datetime.now().year
|
|
with _conn() as db:
|
|
row = db.execute(
|
|
"SELECT COUNT(*) FROM tickets WHERE id LIKE ?", (f"{year}-%",)
|
|
).fetchone()
|
|
n = (row[0] if row else 0) + 1
|
|
return f"{year}-{n:03d}"
|
|
|
|
|
|
def creer_ticket(user_id: int, username: str, chat_id: int, chantier: str, description: str, produit: str = "") -> str:
|
|
ticket_id = _next_id()
|
|
with _conn() as db:
|
|
db.execute(
|
|
"INSERT INTO tickets (id, user_id, username, chat_id, chantier, description, produit, created_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(ticket_id, user_id, username, chat_id, chantier, description, produit, datetime.now().isoformat()),
|
|
)
|
|
return ticket_id
|
|
|
|
|
|
def get_ticket(ticket_id: str) -> dict | None:
|
|
with _conn() as db:
|
|
db.row_factory = sqlite3.Row
|
|
row = db.execute("SELECT * FROM tickets WHERE id = ?", (ticket_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def resoudre_ticket(ticket_id: str, cause: str, solution: str, duree_min: int) -> bool:
|
|
with _conn() as db:
|
|
r = db.execute(
|
|
"UPDATE tickets SET status='resolu', resolved_at=?, cause=?, solution=?, duree_min=? WHERE id=?",
|
|
(datetime.now().isoformat(), cause, solution, duree_min, ticket_id),
|
|
)
|
|
return r.rowcount > 0
|
|
|
|
|
|
def tickets_ouverts() -> list[dict]:
|
|
with _conn() as db:
|
|
db.row_factory = sqlite3.Row
|
|
rows = db.execute(
|
|
"SELECT * FROM tickets WHERE status='ouvert' ORDER BY created_at DESC"
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|