34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
# @Author: Owl
|
|
# @Date: 2025/10/10 16:17
|
|
# @Description:
|
|
|
|
from app.db_func_base import *
|
|
|
|
|
|
class CrossDbHelper(TableDbHelperBase):
|
|
|
|
def __init__(self, pool):
|
|
self.db_pool = pool
|
|
self.DB_Name = 'ledger'
|
|
|
|
def query_cross_usable_date_sql(self, crossid, nodeid):
|
|
sql = f"""
|
|
select distinct day from traffic_{nodeid}.cross_delay where crossid = '{crossid}'
|
|
"""
|
|
return self.do_select(sql)
|
|
|
|
def query_cross_delay_info(self, crossid, nodeid, date_list, tp_start):
|
|
date_list = ','.join(["'" + str(item) + "'" for item in date_list])
|
|
sql = f"""
|
|
select * from traffic_{nodeid}.cross_delay where crossid = '{crossid}' and day in ({date_list}) and tp_start = '{tp_start}' order by day
|
|
"""
|
|
return self.do_select(sql)
|
|
|
|
def query_cross_delay_whole_day_hours(self, crossid, nodeid, query_date):
|
|
sql = f"""
|
|
select * from traffic_{nodeid}.cross_delay where crossid = '{crossid}' and day = '{query_date}' and tp_start like 'h%' order by CAST(REGEXP_SUBSTR(tp_start, '[0-9]+') AS UNSIGNED)
|
|
"""
|
|
return self.do_select(sql)
|
|
|