opt sid iccid map 2025-6-3 10:06

unify_api/modules/zhiwei_u/service/sid_iccid_map.py

 # -*- coding:utf-8 -*-
 """
 DATE：2024/12/11 08:56
+# 查询装置配置信息，导出SIM卡编号和运营商
 
 """
-# -*- coding:utf-8 -*-
-"""
-DATE：2024/12/9 16:03
-
-"""
-# 查询装置配置信息，导出SIM卡编号和运营商
 
+import time
 from uuid import uuid4
 import pendulum
 import asyncio
+from pot_libs.logger import log
 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
+    iccid, sim_opt = None, None
     try:
         req_id = str(uuid4())
         data = {
@@ -34,15 +31,19 @@ async def iccid_from_get_status(sid, method="get-status"):
                 iccid = rlt["data"]["iccid"]
             elif "hardware" in rlt["data"]:
                 iccid = rlt["data"]["hardware"]["iccid"]
+
+        if "SIM_Operator" in rlt["data"]:
+            sim_opt = rlt["data"]["SIM_Operator"]
+        elif "SIM_Operator" in rlt["data"]["hardware"]:
+            sim_opt = rlt["data"]["hardware"]["SIM_Operator"]
     except Exception as e:
         pass
-        # print(f"sid:{sid} 在get-status中未查询到数据")
-    return iccid
+    return iccid, sim_opt
 
 
 async def iccid_from_get_configv1(sid, method="get"):
     """v1.0查询配置信息"""
-    rlt = None
+    iccid, sim_opt = None, None
     try:
         req_id = str(uuid4())
         data = {
@@ -54,15 +55,23 @@ async def iccid_from_get_configv1(sid, method="get"):
         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"]
+
+            if "SIM_Operator" in rlt["data"]:
+                sim_opt = rlt["data"]["SIM_Operator"]
+            elif "SIM_Operator" in rlt["data"]["hardware"]:
+                sim_opt = rlt["data"]["hardware"]["SIM_Operator"]
     except Exception as e:
-        print(f"sid:{sid} 在1.0协议get中未查询到数据")
-        # 4.2 根据2.0协议查询数据
-    return rlt
+        pass
+    return iccid, sim_opt
 
 
-async def iccid_from_get_configv2(sid, method="get-config", key="ratio_ids"):
+async def iccid_from_get_configv2(sid, method="get-config", key="hardware"):
     """v1.0查询配置信息"""
-    rlt = None
+    iccid, sim_opt = None, None
     try:
         req_id = str(uuid4())
         data = {
@@ -75,10 +84,18 @@ async def iccid_from_get_configv2(sid, method="get-config", key="ratio_ids"):
         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"]
+
+            if "SIM_Operator" in rlt["data"]:
+                sim_opt = rlt["data"]["SIM_Operator"]
+            elif "SIM_Operator" in rlt["data"]["hardware"]:
+                sim_opt = rlt["data"]["hardware"]["SIM_Operator"]
     except Exception as e:
-        print(f"sid:{sid} 在2.0协议未查询到数据")
-        # 4.2 根据2.0协议查询数据
-    return rlt
+        pass
+    return iccid, sim_opt
 
 
 async def load_compy_sids(cid, db="power_iot"):
@@ -88,9 +105,28 @@ async def load_compy_sids(cid, db="power_iot"):
         return [r["sid"] for r in await conn.fetchall(sql)]
 
 
+async def load_all_sids():
+    sid_lst = []
+    dbs = ["power_iot", "bromake", "electric_ops", "micro_grid", "sic", "co2e"]
+    for db in dbs:
+        sql = f"SELECT DISTINCT(sid) " \
+              f"FROM {db}.point INNER JOIN {db}.monitor " \
+              f"ON {db}.point.mtid={db}.monitor.mtid " \
+              f"WHERE demolished=0;"
+        async with MysqlUtil() as conn:
+            sid_lst.extend([r["sid"] for r in await conn.fetchall(sql)])
+
+    # add some special sid
+    sid_lst.extend(["A2004000373"])
+    print(len(sid_lst))
+    return list(set(sid_lst))
+
+
 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);"
+    sql = f"INSERT INTO devops.sid_sim(sid, iccid, sim_opt) " \
+          f"VALUES (%s, %s, %s) " \
+          f"ON DUPLICATE KEY UPDATE " \
+          f"iccid=VALUES(iccid), sim_opt=VALUES(sim_opt);"
     async with MysqlUtil() as conn:
         await conn.insert_many(sql, seq_param_lst)
 
@@ -131,34 +167,34 @@ async def main():
     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)
+async def load_iccid(sid):
+    iccid, sim_opt = await iccid_from_get_status(sid)
     if iccid:
-            return sid, iccid
+        return sid, iccid, sim_opt
 
-    # rlt = await iccid_from_get_configv1(sid)
-    # if rlt:
-    #     print("get_configv1", sid, rlt)
-    #     return sid, None
+    iccid, sim_opt = await iccid_from_get_configv2(sid)
+    if iccid:
+        return sid, iccid, sim_opt
 
-    return sid, None
+    iccid, sim_opt = await iccid_from_get_configv1(sid)
+    if iccid:
+        return sid, iccid, sim_opt
 
+    return sid, None, None
 
-async def main2():
-    cid = 223
-    sid_iccids, fail_sid = [], []
-    tmp = []
-    sid_total = await load_compy_sids(cid)
+
+async def update_all():
+    sid_iccids, fail_sid, tmp = [], [], []
+    sid_total = await load_all_sids()
     for i, sid in enumerate(sid_total, start=1):
         tmp.append(sid)
-        if i % 300 and i < len(sid_total):
+        if i % 200 and i < len(sid_total):
             continue
 
         task_lst = [load_iccid(sid) for sid in tmp]
-        for _sid, iccid in await asyncio.gather(*task_lst):
+        for _sid, iccid, sim_opt in await asyncio.gather(*task_lst):
             if iccid:
-                sid_iccids.append((_sid, iccid))
+                sid_iccids.append((_sid, iccid, sim_opt))
             else:
                 fail_sid.append(_sid)
 
@@ -167,28 +203,39 @@ async def main2():
     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}")
+    log.info(f"finish task, success：{len(sid_iccids)}, fail：{len(fail_sid)}")
+    log.info(f"fail sids:{fail_sid}")
+
     if fail_sid:
-        sid_iccids, fail_two = [], []
+        sid_iccids, fail_two, tmp = [], [], []
         for i, sid in enumerate(fail_sid, start=1):
             tmp.append(sid)
-            if i % 300 and i < len(fail_sid):
+            if i % 200 and i < len(fail_sid):
                 continue
 
             task_lst = [load_iccid(sid) for sid in tmp]
-            for _sid, iccid in await asyncio.gather(*task_lst):
+            for _sid, iccid, sim_opt in await asyncio.gather(*task_lst):
                 if iccid:
-                    sid_iccids.append((_sid, iccid))
+                    sid_iccids.append((_sid, iccid, sim_opt))
                 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}")
+        log.info(f"twice, sucess：{len(sid_iccids)} fail：{len(fail_two)}")
+        log.info(f"twice fail sids:{fail_two}")
+
+
+async def main2():
+    while True:
+        try:
+            await update_all()
+        except Exception as e:
+            log.error(f"main2 error:{e}")
+
+        time.sleep(3600)
 
 
 if __name__ == '__main__':
-    # asyncio.run(main2())
-    asyncio.run(update())
+    asyncio.run(main2())