fix longgang data 2023-7-19

unify_api/modules/shidianu/service/open_data_service.py

@@ -16,6 +16,7 @@ from pot_libs.aiohttp_util.aiohttp_utils import AioHttpUtils
 from pot_libs.settings import SETTING
 from pot_libs.utils.pendulum_wrapper import my_pendulum
 from pot_libs.aredis_util.aredis_utils import RedisUtils
+from unify_api.utils.taos_new import get_td_engine_data, parse_td_columns
 from unify_api.utils.time_format import (
     CST, convert_dt_to_timestr, convert_to_dt
 )
@@ -76,25 +77,15 @@ async def stb_data_longgang_service(user_id, d_type):
     if not await get_power(user_id, cids):
         return success_res(code=4001, msg="您没有权限访问")
 
-    td_token = get_token()
     db_name = topic2db[d_type]
     stb_url = f"{SETTING.stb_url}{db_name}?tz=Asia/Shanghai"
-    sql = f"""
-                select last_row(*) from electric_stb
-                where cpyid = {cids[0]}
-                group by tbname
-            """
-    resp_str, status = await AioHttpUtils().post_data(
-        stb_url, data=sql, timeout=50,
-        headers={"Authorization": f"Basic {td_token}"}
-    )
-    if not resp_str or status != 200:
+    sql = f"select last_row(*) from {db_name}.{d_type}_stb " \
+          f"where cpyid={cids[0]} group by tbname;"
+    is_succ, results = await get_td_engine_data(stb_url, sql)
+    if not is_succ:
         return success_res(code=4003, msg="未查找到数据")
 
-    results = json.loads(resp_str)
-    head = [re.findall(r'last_row\((.*)\)', i)[0] if "(" in i else i for i in
-            results["head"]]
-
+    head = parse_td_columns(results)
     datas = [dict(zip(head, r)) for r in results["data"]]
     if d_type == "electric":
         [data.pop("ts_received") for data in datas]
@@ -164,27 +155,20 @@ async def supplement_data_service(user_id, cid, start, end, d_type):
     except Exception as e:
         success_res(code=4004, msg="开始时间或者结束时间错误")
 
-    token = get_token()
     db_name = topic2db[d_type]
     stb_url = f"{SETTING.stb_url}{db_name}?tz=Asia/Shanghai"
-
     if d_type == "electric":
         sql = f"select * from {d_type}_stb where cpyid={cid} and " \
               f"ts >= '{start}' and ts <= '{end}'"
     else:
         sql = f"select * from {d_type}_stb where cpyid={cid} and " \
               f"ts_origin >= '{start}' and ts_origin <= '{end}'"
-    resp_str, status = await AioHttpUtils().post_data(
-        stb_url, data=sql, timeout=50,
-        headers={"Authorization": f"Basic {token}"}
-    )
-    if not resp_str or status != 200:
+    is_succ, results = await get_td_engine_data(stb_url, sql)
+    if not is_succ:
         return success_res(code=4003, msg="未查找到数据")
 
-    results = json.loads(resp_str)
-    datas = []
-    for res in results["data"]:
-        datas.append(dict(zip(results["head"], res)))
+    head = parse_td_columns(results)
+    datas = [dict(zip(head, r)) for r in results["data"]]
     if d_type == "electric":
         datas = [data.pop("ts_received") for data in datas]
     else:

unify_api/utils/taos_new.py

@@ -20,7 +20,6 @@ async def get_td_engine_data(url, sql):
         headers={"Authorization": f"Basic {token}"}
     )
     results = json.loads(resp_str)
-    log.info(f"resp_str:{resp_str},status:{status}")
     if results["code"] != 0:
         return False, None
     return True, results