from decorator.singleton import singleton
|
from db.db_conn import db_conn
|
from db.models.tables import SiteLatestTime
|
from utils.date_utils import DateUtils
|
from datetime import datetime
|
from enumeration.enum_site_status import EnumSiteStatus
|
|
|
@singleton
|
class SiteLatestTimeRepository(object):
|
"""站点数据最新时刻相关数据库操作"""
|
|
def __init__(self) -> None:
|
pass
|
|
def query_all(self) -> list:
|
"""查询全部"""
|
return db_conn.query_sql(lambda session: session.query(SiteLatestTime).all(), new_session=True)
|
|
def refresh_site_status(self):
|
"""根据当前数据表中的数据最新时刻刷新运行状态"""
|
|
def f(session):
|
datas = session.query(SiteLatestTime).all()
|
for d in datas:
|
self.update_site_status(d, d.latest_time)
|
session.commit()
|
return db_conn.query_sql(f, new_session=True)
|
|
def update_site_status(self, obj: SiteLatestTime, time) -> SiteLatestTime:
|
"""
|
更新站点最新时刻及状态
|
状态：上线(0)、停运(1)和下线(2)
|
"""
|
diff_days = DateUtils.time_distance(time, datetime.now())
|
if diff_days < 0:
|
return obj
|
|
if diff_days <= 3:
|
status = EnumSiteStatus.Online.value
|
elif diff_days <= 7:
|
status = EnumSiteStatus.Stop.value
|
else:
|
status = EnumSiteStatus.Offline.value
|
obj.latest_time = DateUtils.time_check(time)
|
obj.device_status = status
|
return obj
|
|
def query_by_site_status(self, status: int) -> list:
|
"""根据站点状态获取最新时刻表"""
|
|
def f(session):
|
return session.query(SiteLatestTime).filter_by(device_status=status).all()
|
return db_conn.query_sql(f, new_session=True)
|
