109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
# backend/app.py
|
||
import logging
|
||
from logging.handlers import RotatingFileHandler
|
||
from flask import Flask, jsonify, request
|
||
from flask_cors import CORS
|
||
import sqlite3
|
||
import os
|
||
|
||
# 获取当前文件所在目录
|
||
basedir = os.path.abspath(os.path.dirname(__file__))
|
||
db_path = os.path.join(basedir, '..', 'data', 'cars.db')
|
||
|
||
# 创建 logs 目录
|
||
logs_dir = os.path.join(basedir, '..', 'logs')
|
||
os.makedirs(logs_dir, exist_ok=True)
|
||
|
||
# 配置日志
|
||
log_formatter = logging.Formatter(
|
||
'%(asctime)s %(levelname)s %(name)s %(message)s'
|
||
)
|
||
log_file = os.path.join(logs_dir, 'app.log')
|
||
|
||
# 创建 rotating file handler (最大5MB，保留5个备份)
|
||
file_handler = RotatingFileHandler(
|
||
log_file,
|
||
maxBytes=5*1024*1024, # 5MB
|
||
backupCount=5
|
||
)
|
||
file_handler.setFormatter(log_formatter)
|
||
file_handler.setLevel(logging.INFO)
|
||
|
||
# 创建 logger
|
||
app = Flask(__name__)
|
||
app.logger.addHandler(file_handler)
|
||
app.logger.setLevel(logging.INFO)
|
||
|
||
# 添加 CORS 支持
|
||
CORS(app)
|
||
|
||
# 数据库连接函数
|
||
def get_db_connection():
|
||
conn = sqlite3.connect(db_path)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
@app.route('/api/cars', methods=['GET'])
|
||
def get_all_cars():
|
||
app.logger.info('获取所有车辆信息')
|
||
try:
|
||
conn = get_db_connection()
|
||
cars = conn.execute('SELECT * FROM cars').fetchall()
|
||
conn.close()
|
||
|
||
cars_list = [dict(car) for car in cars]
|
||
app.logger.info(f'成功获取 {len(cars_list)} 条车辆记录')
|
||
return jsonify(cars_list)
|
||
except Exception as e:
|
||
app.logger.error(f'获取所有车辆信息时出错: {str(e)}')
|
||
return jsonify({'error': '服务器内部错误'}), 500
|
||
|
||
@app.route('/api/cars/search', methods=['GET'])
|
||
def search_car():
|
||
plate = request.args.get('plate', '')
|
||
app.logger.info(f'搜索车牌号码: {plate}')
|
||
|
||
if not plate:
|
||
app.logger.warning('用户未提供车牌号码')
|
||
return jsonify({'error': '请输入车牌号码'}), 400
|
||
|
||
try:
|
||
conn = get_db_connection()
|
||
cars = conn.execute('SELECT * FROM cars WHERE plate LIKE ?', (f'%{plate}%',)).fetchall()
|
||
conn.close()
|
||
|
||
cars_list = [dict(car) for car in cars]
|
||
app.logger.info(f'搜索 "{plate}" 返回 {len(cars_list)} 条结果')
|
||
return jsonify(cars_list)
|
||
except Exception as e:
|
||
app.logger.error(f'搜索车牌号码时出错: {str(e)}')
|
||
return jsonify({'error': '服务器内部错误'}), 500
|
||
|
||
@app.route('/api/cars/<plate>', methods=['GET'])
|
||
def get_car_by_plate(plate):
|
||
app.logger.info(f'查询特定车牌号码: {plate}')
|
||
|
||
try:
|
||
conn = get_db_connection()
|
||
car = conn.execute('SELECT * FROM cars WHERE plate = ?', (plate,)).fetchone()
|
||
conn.close()
|
||
|
||
if car is None:
|
||
app.logger.warning(f'未找到车牌号码: {plate}')
|
||
return jsonify({'error': '未找到该车牌号码对应的车主信息'}), 404
|
||
|
||
app.logger.info(f'成功找到车牌号码: {plate}')
|
||
return jsonify(dict(car))
|
||
except Exception as e:
|
||
app.logger.error(f'查询特定车牌时出错: {str(e)}')
|
||
return jsonify({'error': '服务器内部错误'}), 500
|
||
|
||
# 添加一个根路径用于健康检查
|
||
@app.route('/', methods=['GET'])
|
||
def health_check():
|
||
app.logger.info('健康检查请求')
|
||
return jsonify({'status': 'ok', 'message': '服务正常运行'})
|
||
|
||
if __name__ == '__main__':
|
||
app.logger.info('启动 Flask 应用')
|
||
app.run(debug=True, host='0.0.0.0', port=5000) |