103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""数据模型：使用 sqlite3 管理车辆数据"""
|
|
|
|
import sqlite3
|
|
import os
|
|
|
|
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'vehicles.db')
|
|
|
|
|
|
def get_conn():
|
|
"""获取数据库连接，返回 Row 类型便于转字典"""
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_db():
|
|
"""初始化数据库表"""
|
|
conn = get_conn()
|
|
conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS vehicles (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
plate TEXT NOT NULL UNIQUE,
|
|
brand TEXT NOT NULL,
|
|
model TEXT NOT NULL,
|
|
year INTEGER,
|
|
color TEXT,
|
|
status TEXT DEFAULT 'available',
|
|
owner TEXT
|
|
)
|
|
''')
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def list_vehicles(keyword=None):
|
|
"""列出车辆，可按关键字搜索"""
|
|
conn = get_conn()
|
|
if keyword:
|
|
like = f'%{keyword}%'
|
|
rows = conn.execute(
|
|
'SELECT * FROM vehicles WHERE plate LIKE ? OR brand LIKE ? OR model LIKE ? ORDER BY id DESC',
|
|
(like, like, like)
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute('SELECT * FROM vehicles ORDER BY id DESC').fetchall()
|
|
conn.close()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
def get_vehicle(vid):
|
|
"""获取单条车辆"""
|
|
conn = get_conn()
|
|
row = conn.execute('SELECT * FROM vehicles WHERE id = ?', (vid,)).fetchone()
|
|
conn.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def add_vehicle(data):
|
|
"""新增车辆"""
|
|
conn = get_conn()
|
|
cur = conn.execute(
|
|
'INSERT INTO vehicles (plate, brand, model, year, color, status, owner) VALUES (?, ?, ?, ?, ?, ?, ?)',
|
|
(data.get('plate'), data.get('brand'), data.get('model'),
|
|
data.get('year'), data.get('color'),
|
|
data.get('status', 'available'), data.get('owner'))
|
|
)
|
|
conn.commit()
|
|
vid = cur.lastrowid
|
|
conn.close()
|
|
return vid
|
|
|
|
|
|
def update_vehicle(vid, data):
|
|
"""更新车辆"""
|
|
conn = get_conn()
|
|
conn.execute(
|
|
'UPDATE vehicles SET plate=?, brand=?, model=?, year=?, color=?, status=?, owner=? WHERE id=?',
|
|
(data.get('plate'), data.get('brand'), data.get('model'),
|
|
data.get('year'), data.get('color'), data.get('status'),
|
|
data.get('owner'), vid)
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def delete_vehicle(vid):
|
|
"""删除车辆"""
|
|
conn = get_conn()
|
|
conn.execute('DELETE FROM vehicles WHERE id = ?', (vid,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def stats():
|
|
"""统计信息"""
|
|
conn = get_conn()
|
|
total = conn.execute('SELECT COUNT(*) AS c FROM vehicles').fetchone()['c']
|
|
rows = conn.execute('SELECT status, COUNT(*) AS c FROM vehicles GROUP BY status').fetchall()
|
|
conn.close()
|
|
by_status = {r['status']: r['c'] for r in rows}
|
|
return {'total': total, 'by_status': by_status}
|