获取alexa对象检测或快照图片上传链接

AnsjerPush/urls.py

@@ -2,7 +2,7 @@ from django.urls import path, re_path
 
 from Controller import DetectController, ShadowController, DetectControllerV2, AiController, gatewayController, \
     PowerWarningController, InitController, CustomizedPushController, TransparentTransmissionPushController, \
-    DeviceReportController, ReportController
+    DeviceReportController, ReportController, AlexaController
 from Controller.ComboCron import ComboCronPushController
 from Controller.Cron import CronTaskController
 
@@ -27,4 +27,7 @@ urlpatterns = [
     re_path('transparent-transmission/(?P<operation>.*)',
             TransparentTransmissionPushController.TransparentTransmissionPushView.as_view()),
     re_path('cron/create/(?P<operation>.*)', CronTaskController.CronTaskView.as_view()),
+
+    re_path('alexa/(?P<operation>.*)', AlexaController.AlexaView.as_view()),
+
 ]

Controller/AlexaController.py

@@ -0,0 +1,68 @@
+# @Author    : Rocky
+# @File      : AlexaController.py
+# @Time      : 2025/7/22 17:20
+import threading
+import time
+
+from django.views import View
+
+from Model.models import AlexaPush
+from Object.ResponseObject import ResponseObject
+from AnsjerPush.config import CONFIG_INFO, CONFIG_EUR
+from Service.DevicePushService import DevicePushService
+
+
+class AlexaView(View):
+
+    def get(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        return self.validation(request.GET, request, operation)
+
+    def post(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        return self.validation(request.POST, request, operation)
+
+    def validation(self, request_dict, request, operation):
+        response = ResponseObject()
+        if operation == 'ObjectDetectionOrSnapshot':
+            return self.object_detection_or_snapshot(request_dict, response)
+        else:
+            return response.json(414)
+
+    @classmethod
+    def object_detection_or_snapshot(cls, request_dict, response):
+        uid = request_dict.get('uid', None)
+        channel = request_dict.get('channel', '1')
+        event_type = request_dict.get('event_type', None)
+        event_time = request_dict.get('event_time', None)
+
+        if not all([uid, event_type, event_time]):
+            return response.json(444)
+
+        # 欧洲: OCI伦敦, 其他: OCI凤凰城
+        storage_location: int = 4 if CONFIG_INFO == CONFIG_EUR else 3
+
+        kwargs = {
+            'uid': uid,
+            'channel': channel,
+            'event_type': int(event_type),
+            'event_time': event_time,
+            'storage_location': storage_location,
+        }
+        # 异步推送消息和保存数据
+        threading.Thread(
+            target=cls.save_data_and_send_alexa_event,
+            kwargs=kwargs).start()
+
+        kwargs.pop('event_type')
+        res_data: str = DevicePushService.get_oci_req_url(**kwargs)
+
+        return response.json(0, res_data)
+
+    @staticmethod
+    def save_data_and_send_alexa_event(**kwargs):
+        kwargs['add_time'] = int(time.time())
+        AlexaPush.objects.create(**kwargs)
+        # 请求Alexa服务器发送 ObjectDetection 或 Snapshot 事件

Model/models.py

@@ -3561,6 +3561,23 @@ class AbnormalEvent(models.Model):
         app_label = 'db2'
 
 
+class AlexaPush(models.Model):
+    id = models.AutoField(primary_key=True, verbose_name='自增标记ID')
+    uid = models.CharField(default='', max_length=20, db_index=True, verbose_name='uid')
+    channel = models.PositiveSmallIntegerField(default=1, verbose_name='设备通道')
+    # 事件类型, 0: 快照, 1: 人形检测
+    event_type = models.PositiveSmallIntegerField(default=0, verbose_name='事件类型')
+    # 1: 阿里云, 2: AWS, 3: oci美国凤凰城, 4: oci英国伦敦, 5: 华为云
+    storage_location = models.PositiveSmallIntegerField(default=3, verbose_name='存储桶位置')
+    event_time = models.IntegerField(default=0, verbose_name='上报时间')
+    add_time = models.IntegerField(default=0, verbose_name='添加时间')
+
+    class Meta:
+        db_table = 'alexa_push'
+        verbose_name = 'Alexa推送'
+        app_label = 'db2'
+
+
 class AlbumTitle(models.Model):
     id = models.AutoField(primary_key=True)
     album_title = models.TextField(blank=True, default='', verbose_name='相册标题')

Service/DevicePushService.py

@@ -1124,6 +1124,18 @@ class DevicePushService:
             res_data['msg'] = 'success 3'
         return res_data
 
+    @classmethod
+    def get_oci_req_url(cls, uid: str, channel: str, event_time: int, storage_location: int) -> str:
+        """
+        获取oci预认证请求url
+        """
+        region: str = 'eur' if storage_location == 4 else 'us'
+        oci_client: OCIObjectStorage = OCIObjectStorage(region)
+        key_name: str = '{}/{}/{}.jpeg'.format(uid, channel, event_time)
+        img_url: str = cls.create_oci_req_url(
+            storage_location=storage_location, bucket='foreignpush', obj_name=key_name, oci=oci_client)
+        return img_url
+
     @staticmethod
     def generate_s3_url(aws_s3_client, params):
         """