Project/mytool/db_helper/db_operation.py

import sys

sys.path.append("..")
import mytool.db_helper.mysqlhelper as mdb


# 数据库操作
# 可自行添加操作函数

class DBOperation:
    def __init__(self,db_ip, db_port, db_name, db_user, db_password):
        self._db = mdb.DB(db_ip, db_port, db_name, db_user, db_password)
    # 查询
    def query(self, sql):
        return self._db.query_all_sql(sql)
    # 更新
    def update(self, sql):
        return self._db.execute_sql(sql)
    def query_parking_in_unit_tableid(self, unit,tableid):
        sql = "select * from space WHERE unit=%s and table_id=%s"
        return self._db.query_all_sql(sql, (unit,tableid))

    def query_command_in_unit(self, unit):
        sql = "select * from command_queue WHERE unit=%s"
        return self._db.query_all_sql(sql, unit)

    def query_pick_command_in_unit(self, unit):
        sql = "select * from command_queue where unit=%s and type=2 order by queue_id ASC"
        return self._db.query_all_sql(sql, unit)

    def query_pick_command_in_unit_and_statu(self, unit, statu):
        sql = "select * from command_queue where unit=%s and statu=%s and type=2 order by queue_id ASC"
        return self._db.query_all_sql(sql, (unit, statu))

    def query_sort_pick_command(self, unit):
        sql = "select * from command_queue where unit=%s and (statu=0 or statu=1) and type=2 order by queue_id ASC"
        return self._db.query_all_sql(sql, unit)
    def query_command_all_in_unit_and_sort(self, unit):
        sql = "select * from command_queue where unit=%s order by type,statu DESC,queue_id ASC"
        return self._db.query_all_sql(sql, unit)

    def query_command_all_in_unit_type_and_sort(self, unit, type):
        sql = "select * from command_queue where unit=%s and type=%s order by type,statu DESC,queue_id ASC"
        return self._db.query_all_sql(sql, (unit, type))
    def query_command_all(self):
        sql = "select * from command_queue"
        return self._db.query_all_sql(sql)

    def query_command_in_car_number(self, car_number):
        sql = "select * from command_queue WHERE car_number=%s"
        return self._db.query_all_sql(sql, car_number)

    def query_command_in_primary_key(self, primary_key):
        sql = "select * from command_queue WHERE primary_key=%s"
        return self._db.query_all_sql(sql, primary_key)

    def query_command_in_height_unit_and_statu(self, height, unit):
        sql = "select * from command_queue WHERE height>=%s and unit=%s and statu=0 and type=1"  # type=1 存车
        return self._db.query_all_sql(sql, (height, unit))
    def query_command_in_height_minmax_unit_and_statu(self, min,max, unit):
        sql = "select * from command_queue WHERE height>=%s and height<=%s and unit=%s and statu=0 and type=1"  # type=1 存车
        return self._db.query_all_sql(sql, ( min,max, unit))
    def query_space_in_car_number(self, car_number):
        sql = "select * from space WHERE car_number=%s"
        return self._db.query_all_sql(sql, car_number)
    def query_space_in_primary_key(self, primary_key):
        sql = "select * from space WHERE primary_key=%s"
        return self._db.query_all_sql(sql, primary_key)

    def query_space_in_unit_and_table_id(self, unit, table_id):
        sql = "select * from space WHERE unit=%s and table_id=%s"
        return self._db.query_all_sql(sql, (unit, table_id))
    def query_space_in_height(self, height):
        sql = "select * from space WHERE height>=%s"
        return self._db.query_all_sql(sql, height)

    def query_space_in_space_id(self, space_id):
        sql = "select * from space WHERE id=%s"
        return self._db.query_all_sql(sql, space_id)
    def query_space_in_height_and_empty(self, height):
        sql = "select * from space WHERE height>=%s and statu=0 and car_number IS NULL"
        return self._db.query_all_sql(sql, height)

    def query_space_in_height_unit_and_empty(self, height, unit):
        sql = "select * from space WHERE height>=%s and unit=%s and statu=0 and car_number IS NULL"
        return self._db.query_all_sql(sql, (height, unit))

    def query_space_in_height_unit_and_empty_2(self, min,max, unit):
        sql = "select * from space WHERE height>=%s and height<=%s and unit=%s and statu=0 and car_number IS NULL"
        return self._db.query_all_sql(sql, (min, max, unit))

    def query_space_in_unit(self, unit):
        sql = "select * from space WHERE unit=%s"
        return self._db.query_all_sql(sql, unit)

    def query_space_all(self):
        sql = "select * from space"
        return self._db.query_all_sql(sql)
    def query_vehicle_primary_key(self, car_number):
        sql = "select primary_key from vehicle where car_number=%s"
        return self._db.query_all_sql(sql, car_number)
    def update_space_status_in_space_id(self, space_id, statu):
        sql = "update space set statu=%s where id=%s"
        return self._db.execute_sql(sql, (statu, space_id))

    def update_space_status_car_number_in_space_id(self, space_id,car_number,statu):
        sql = "update space set statu=%s,car_number=%s where id=%s"
        return self._db.execute_sql(sql, (statu, car_number,space_id))

    def update_command_status_in_car_number(self,car_number,statu):
        sql = "update command_queue set statu=%s where car_number=%s"
        return self._db.execute_sql(sql, (statu, car_number))
    def update_space_remark_and_statu_in_space_id(self,space_id,remark,statu):
        sql = "update space set remark=%s,statu=%s where id=%s"
        return self._db.execute_sql(sql, (remark, statu,space_id))
    def clear_space_data(self, space_id):
        sql = "update space set car_number=NULL,primary_key=NULL,statu=0,space_info=NULL,measure_info=NULL,plate_color=NULL,plate_type=NULL,plate_confidence=NULL,recognition_time=NULL,plate_full_image=NULL,plate_clip_image=NULL,remark=NULL where id=%s"
        return self._db.execute_sql(sql, space_id)



    def delete_command(self, car_number):
        sql = "delete from command_queue where car_number=%s"
        return self._db.execute_sql(sql, car_number)

    # 查询一段时间内停车次数
    def query_park_count_in_time(self, in_date, out_date):
        sql = "select * from record where in_time_start >= %s and in_time_start <= %s"
        return self._db.query_all_sql(sql, (in_date,out_date))
    def query_pick_count_out_time(self, in_date, out_date):
        sql = "select * from record where out_time_start >= %s and out_time_start <= %s"
        return self._db.query_all_sql(sql, (in_date, out_date))

    def query_park_count_in_time_and_unit(self, in_date, out_date,unit):
        sql = "select * from record where (in_time_start >= %s and in_time_start <= %s) and unit_id=%s"
        return self._db.query_all_sql(sql, (in_date,out_date,unit))
    def query_pick_count_out_time_and_unit(self, in_date, out_date,unit):
        sql = "select * from record where (out_time_start >= %s and out_time_start <= %s) and unit_id=%s"
        return self._db.query_all_sql(sql, (in_date, out_date,unit))

     # 查询一段时间内停车车数
    # period_type = 1 :当天内
    #               2 :当月内
    #               3 :自定义起止时间
    def query_carQuantity_inPeriod(self, period_type, date):
        sql = "select count(distinct(car_number)) from record where in_time_start like %s"
        if period_type == "day":
            date = date[0:10]
        elif period_type == "month":
            date = date[0:7]
        date = date + "%"  # 模糊查询

        result = self._db.query_all_sql(sql, date)  # [{'count(distinct(car_number))': 4}]

        return result[0]["count(distinct(car_number))"]