video-insight

utils/aliyun_security.py

@@ -31,7 +31,6 @@ class Security:
             'content': content
         }
         textModerationPlusRequest = models.TextModerationPlusRequest(
-            # 检测类型
             service='llm_response_moderation_01',
             service_parameters=json.dumps(serviceParameters)
         )
@@ -39,16 +38,28 @@ class Security:
         try:
             response = clt.text_moderation_plus(textModerationPlusRequest)
             if response.status_code == 200:
-                # 调用成功
-                result = response.body
-                logger.info('response success. result:{}'.format(result))
-                if result.code == 200:
-                    if result.Data.RiskLevel == "high" or result.Data.RiskLevel == "medium":
-                        return false,result
-                return true,result 
+                # 将响应体转换为字典格式
+                result_dict = response.body.to_map()
+                logger.info('response success. result:{}'.format(result_dict))
+                
+                if result_dict['Code'] == 200:
+                    risk_level = result_dict['Data']['RiskLevel']
+                    if risk_level in ["high", "medium"]:
+                        return False, result_dict
+                    return True, result_dict
+                else:
+                    logger.error('Business error. code:{}, message:{}'.format(
+                        result_dict['Code'], result_dict['Message']))
+                    return False, result_dict
             else:
-                print('response not success. status:{} ,result:{}'.format(response.status_code, response))
+                error_msg = 'API request failed. status:{}'.format(response.status_code)
+                logger.error(error_msg)
+                return False, {'error': error_msg}
+                
         except Exception as err:
-            print(err)
+            logger.error(f"Error in security check: {str(err)}")
+            return False, {'error': str(err)}
 if __name__ == '__main__':
-    Security.security("测试安全边界")
+    is_safe, result = Security.security("测试安全边界")
+    print(f"安全状态: {is_safe}")
+    print(f"详细结果: {result}")

workers/video_insight_consumption_work.py

@@ -76,7 +76,7 @@ class ConsumptionRecommend(object):
         for entry in data:
 
             hook_desc = entry.get('需求钩子话术', '')
-            result,msg = security(hook_desc)
+            result,msg = Security.security(hook_desc)
             audit_status = 0
             if result :
                 audit_status = 1