zhiweiu add iccid rt query 2024-12-11 14:17

unify_api/modules/zhiwei_u/service/equipment_operations_service.py

@@ -23,6 +23,9 @@ from unify_api.modules.zhiwei_u.procedures.dev_replace import dev_replace
 from unify_api.modules.zhiwei_u.procedures.equipment_operations_pds import \
     alarm_setting_pds, device_get_v1, device_get_config
 from unify_api.utils.response_code import RET
+from unify_api.modules.zhiwei_u.service.sid_iccid_map import (
+    iccid_from_get_status
+)
 
 
 async def equipment_operations_service(userid, prod_id, cid, sid, pid,
@@ -115,12 +118,15 @@ async def equipment_details_mqtt_service(sid, mtid):
                     sid).get("threshold").get(
                     "scopeEnable") == 1 else "关闭"
                 software_version = res_scope.get("software_version")
+
+    iccid = await iccid_from_get_status(sid)
     return EdResp(
         software_version=software_version,
         device_status=status_,
         scope_status=scope_status,
         upload_freq=upload_freq,
-        harm_count=harm_count
+        harm_count=harm_count,
+        iccid=iccid
     )
 
 

unify_api/modules/zhiwei_u/service/sid_iccid_map.py

+# -*- coding:utf-8 -*-
+"""
+DATE：2024/12/11 08:56
+
+"""
+# -*- coding:utf-8 -*-
+"""
+DATE：2024/12/9 16:03
+
+"""
+# 查询装置配置信息，导出SIM卡编号和运营商
+
+from uuid import uuid4
+import pendulum
+import asyncio
+from pot_libs.aiomqtt_util.hbmqtt_utils import MqttUtil
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+
+
+async def iccid_from_get_status(sid, method="get-status"):
+    iccid = None
+    try:
+        req_id = str(uuid4())
+        data = {
+            "sid": sid,
+            "method": method,
+            "request_id": req_id,
+            "time": pendulum.now().to_datetime_string()
+        }
+        async with MqttUtil() as emq:
+            rlt = await emq.device_response(sid=sid, request_id=req_id,
+                                            data=data)
+            if "iccid" in rlt["data"]:
+                iccid = rlt["data"]["iccid"]
+            elif "hardware" in rlt["data"]:
+                iccid = rlt["data"]["hardware"]["iccid"]
+    except Exception as e:
+        pass
+        # print(f"sid:{sid} 在get-status中未查询到数据")
+    return iccid
+
+
+async def iccid_from_get_configv1(sid, method="get"):
+    """v1.0查询配置信息"""
+    rlt = None
+    try:
+        req_id = str(uuid4())
+        data = {
+            "sid": sid,
+            "method": method,
+            "request_id": req_id,
+            "time": pendulum.now().to_datetime_string()
+        }
+        async with MqttUtil() as emq:
+            rlt = await emq.device_response(sid=sid, request_id=req_id,
+                                            data=data)
+    except Exception as e:
+        print(f"sid:{sid} 在1.0协议get中未查询到数据")
+        # 4.2 根据2.0协议查询数据
+    return rlt
+
+
+async def iccid_from_get_configv2(sid, method="get-config", key="ratio_ids"):
+    """v1.0查询配置信息"""
+    rlt = None
+    try:
+        req_id = str(uuid4())
+        data = {
+            "sid": sid,
+            "method": method,
+            "key": key,
+            "request_id": req_id,
+            "time": pendulum.now().to_datetime_string()
+        }
+        async with MqttUtil() as emq:
+            rlt = await emq.device_response(sid=sid, request_id=req_id,
+                                            data=data)
+    except Exception as e:
+        print(f"sid:{sid} 在2.0协议未查询到数据")
+        # 4.2 根据2.0协议查询数据
+    return rlt
+
+
+async def load_compy_sids(cid, db="power_iot"):
+    sql = f"SELECT DISTINCT(sid) FROM {db}.monitor " \
+          f"WHERE cid={cid} AND demolished=0;"
+    async with MysqlUtil() as conn:
+        return [r["sid"] for r in await conn.fetchall(sql)]
+
+
+async def batch_insert(seq_param_lst):
+    sql = f"INSERT INTO devops.sid_sim2(sid, iccid) VALUES (%s, %s) " \
+          f"ON DUPLICATE KEY UPDATE iccid=VALUES(iccid);"
+    async with MysqlUtil() as conn:
+        await conn.insert_many(sql, seq_param_lst)
+
+
+async def update():
+    async with MysqlUtil() as conn:
+        sql = f"SELECT DISTINCT(sid) FROM devops.sid_sim2;"
+        sids = [r["sid"] for r in await conn.fetchall(sql)]
+
+        sid_iccids = []
+        sql = f"SELECT sid, iccid FROM devops.sid_sim;"
+        for r in await conn.fetchall(sql):
+            sid, iccid = r["sid"], r["iccid"]
+            if sid not in sids:
+                sid_iccids.append((sid, iccid))
+
+        print(len(sid_iccids), sid_iccids)
+        if sid_iccids:
+            await batch_insert(sid_iccids)
+
+
+async def main():
+    cid = 223
+    sid_iccids = []
+    for sid in await load_compy_sids(cid):
+        iccid = await iccid_from_get_status(sid)
+        if iccid:
+            sid_iccids.append((sid, iccid))
+            continue
+
+        rlt = await iccid_from_get_configv1(sid)
+        if not rlt:
+            rlt = await iccid_from_get_configv2(sid)
+
+    if sid_iccids:
+        await batch_insert(sid_iccids)
+
+    print(f"finish all sid iccid, {len(sid_iccids)}")
+
+
+async def load_iccid(sid, trys=5):
+    for i in range(trys):
+        iccid = await iccid_from_get_status(sid)
+        if iccid:
+            return sid, iccid
+
+    # rlt = await iccid_from_get_configv1(sid)
+    # if rlt:
+    #     print("get_configv1", sid, rlt)
+    #     return sid, None
+
+    return sid, None
+
+
+async def main2():
+    cid = 223
+    sid_iccids, fail_sid = [], []
+    tmp = []
+    sid_total = await load_compy_sids(cid)
+    for i, sid in enumerate(sid_total, start=1):
+        tmp.append(sid)
+        if i % 300 and i < len(sid_total):
+            continue
+
+        task_lst = [load_iccid(sid) for sid in tmp]
+        for _sid, iccid in await asyncio.gather(*task_lst):
+            if iccid:
+                sid_iccids.append((_sid, iccid))
+            else:
+                fail_sid.append(_sid)
+
+        tmp = []
+
+    if sid_iccids:
+        await batch_insert(sid_iccids)
+
+    print(f"finish all task, success {len(sid_iccids)}, fail {len(fail_sid)}")
+    print(f"fail sids:{fail_sid}")
+    if fail_sid:
+        sid_iccids, fail_two = [], []
+        for i, sid in enumerate(fail_sid, start=1):
+            tmp.append(sid)
+            if i % 300 and i < len(fail_sid):
+                continue
+
+            task_lst = [load_iccid(sid) for sid in tmp]
+            for _sid, iccid in await asyncio.gather(*task_lst):
+                if iccid:
+                    sid_iccids.append((_sid, iccid))
+                else:
+                    fail_two.append(_sid)
+        if sid_iccids:
+            await batch_insert(sid_iccids)
+
+        print(f"fail request, sucess {len(sid_iccids)} fail:{len(fail_two)}")
+        print(f"twice fail sids:{fail_two}")
+
+
+if __name__ == '__main__':
+    # asyncio.run(main2())
+    asyncio.run(update())