from pot_libs.mysql_util.mysql_util import MysqlUtil from unify_api.constants import SDU_ALARM_LIST import pendulum from unify_api.utils.time_format import CST async def get_user_product_auth(user_id): sql = "SELECT * from user_product_auth where user_id=%s and product=4" async with MysqlUtil() as conn: return await conn.fetchone(sql, args=(user_id,)) async def get_basic_info_by_mtid(mtid, cid): sql = "select s.room_name,s.storey_name,m.meter_no,m.sid,m.latitude," \ "m.longitude from storey_room_map s left join point p " \ "on p.pid=s.point_id LEFT JOIN monitor m on m.mtid=p.mtid " \ "where m.mtid=%s and m.demolished=0 and s.cid =%s" async with MysqlUtil() as conn: return await conn.fetchone(sql, args=(mtid, cid)) async def monitor_point_company(cids): sql = "SELECT c.cid,c.address,c.fullname,p.pid,m.mtid,s.storey_name," \ "s.room_name,m.sid FROM `company` c INNER JOIN monitor m " \ "on c.cid=m.cid INNER JOIN point p on m.mtid=p.mtid " \ "INNER JOIN storey_room_map s on s.point_id=p.pid where c.cid in %s" async with MysqlUtil() as conn: return await conn.fetchall(sql, args=(cids,)) async def load_lg_sdu_events(cid, pg_num, pg_size, importance): cond_lst = [f"cid={cid}", f"event_type in {tuple(SDU_ALARM_LIST)}"] if len(importance) > 0: importance = str(tuple(importance)).replace(',)', ')') cond_lst.append(f"importance in {importance}") time_format = "%Y-%m-%d %H:%M:%S" end_date = str(pendulum.now(tz=CST).strftime(time_format)) cond_lst.append(f"event_datetime < '{end_date}'") cond_str = " AND ".join(cond_lst) async with MysqlUtil() as conn: sql = f"select count(*) cnt from point_1min_event WHERE {cond_str};" total = await conn.fetchone(sql) total_count = total.get("cnt", 0) if total_count <= 0: return 0, [] sql = f"select * from point_1min_event WHERE {cond_str} " \ f"order by event_datetime desc limit %s,%s;" data = await conn.fetchall(sql, args=((pg_num - 1) * pg_size, pg_size)) return total_count, data