modify

trans2es/management/commands/trans2es_data2es_parallel.py

@@ -15,6 +15,11 @@ from trans2es.utils import topic_transfer
 from libs.table_scan import TableSlicer,TableSlicerChunk
 from trans2es.type_info import get_type_info_map,TypeInfo
 
+from libs.cache import redis_client
+from trans2es.models.face_user_contrast_similar import FaceUserContrastSimilar
+import json
+
+
 class Job(object):
     __es = None
 
@@ -42,6 +47,42 @@ class Job(object):
             es=self.get_es(),
         )
 
+
+class SyncDataToRedis(object):
+
+    @classmethod
+    def sync_face_similar_data_to_redis(cls):
+        try:
+            result_items = FaceUserContrastSimilar.objects.filter(is_online=True,
+                                                                  is_deleted=False).distinct().values(
+                "participant_user_id").values_list("participant_user_id", flat=True)
+
+            logging.info("duan add,begin sync_face_similar_data_to_redis!")
+
+            redis_key_prefix = "physical:user_similar:participant_user_id:"
+            for participant_user_id in result_items:
+
+                redis_key = redis_key_prefix + str(participant_user_id)
+                similar_result_items = FaceUserContrastSimilar.objects.filter(is_online=True, is_deleted=False,
+                                                                              participant_user_id=participant_user_id,
+                                                                              similarity__gt=0.4).order_by(
+                    "-similarity").limit(100)
+
+                item_list = list()
+                for item in similar_result_items:
+                    item_list.append(
+                        {
+                            "contrast_user_id": item.contrast_user_id,
+                            "similarity": item.similarity
+                        }
+                    )
+                redis_client.set(redis_key, json.dumps(item_list))
+
+                logging.info("duan add,participant_user_id:%d set data done!" % participant_user_id)
+        except:
+            logging.error("catch exception,err_msg:%s" % traceback.format_exc())
+
+
 class Command(BaseCommand):
 
     args = ''
@@ -56,6 +97,7 @@ class Command(BaseCommand):
         make_option('-s', '--pks', dest='pks', help='specify sync pks, comma separated', metavar='PKS', default=''),
         make_option('--streaming-slicing', dest='streaming_slicing', action='store_true', default=True),
         make_option('--no-streaming-slicing', dest='streaming_slicing', action='store_false', default=True),
+        make_option('-S', '--sync_type',dest='sync_type', help='sync data to es',metavar='TYPE',default='')
     )
 
     def __sync_data_by_type(self, type_name):
@@ -84,5 +126,9 @@ class Command(BaseCommand):
                         logging.info("begin sync [%s] data to es!" % type_name)
                         self.__sync_data_by_type(type_name)
 
+
+            if len(options["sync_type"]) and options["sync_type"]=="sync_data_to_es":
+                SyncDataToRedis.sync_face_similar_data_to_redis()
+
         except:
             logging.error("catch exception,err_msg:%s" % traceback.format_exc())