tingwu_new

2026-03-11 20:31:09 +08:00
parent 44d90e643f
commit b0aa902f89
5 changed files with 156 additions and 2 deletions
--- a/backend/ai_services/management/commands/test_tingwu_local.py
+++ b/backend/ai_services/management/commands/test_tingwu_local.py
@@ -0,0 +1,102 @@
+import os
+import sys
+import django
+import json
+import logging
+from django.conf import settings
+
+# 设置 Django 环境
+# 添加项目根目录到 sys.path
+sys.path.append('/Volumes/data/Quant-Speed/market_page/backend')
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings') # 修正为正确的 settings 模块路径
+django.setup()
+
+from ai_services.services import AliyunTingwuService
+from ai_services.models import TranscriptionTask
+
+# 配置日志
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def test_tingwu_transcription():
+    file_url = "https://tangledup-ai-staging.oss-cn-shanghai.aliyuncs.com/Video/%E6%95%99%E5%AD%A6.mp4"
+    
+    print(f"Testing transcription for: {file_url}")
+    
+    service = AliyunTingwuService()
+    
+    # 1. 创建任务
+    try:
+        print("Creating task...")
+        response = service.create_transcription_task(file_url)
+        print(f"Create task response: {json.dumps(response, indent=2, ensure_ascii=False)}")
+        
+        if 'Data' in response and isinstance(response['Data'], dict):
+            task_id = response['Data'].get('TaskId')
+        else:
+            task_id = response.get('TaskId')
+            
+        if not task_id:
+            print("Failed to get TaskId")
+            return
+            
+        print(f"Task created with ID: {task_id}")
+        
+        # 2. 轮询查询任务状态
+        import time
+        max_retries = 60 # 5 minutes
+        for i in range(max_retries):
+            print(f"Checking status (attempt {i+1}/{max_retries})...")
+            result = service.get_task_info(task_id)
+            
+            # 解析结果
+            if isinstance(result, str):
+                try:
+                    result = json.loads(result)
+                except:
+                    pass
+            
+            if isinstance(result, dict):
+                data_obj = result.get('Data', result)
+            else:
+                data_obj = result
+                
+            task_status = data_obj.get('TaskStatus')
+            if not task_status:
+                task_status = data_obj.get('Status')
+                
+            print(f"Current status: {task_status}")
+            
+            if task_status in ['COMPLETE', 'COMPLETED', 'SUCCEEDED']:
+                print("Task succeeded!")
+                print(f"Full Result: {json.dumps(data_obj, indent=2, ensure_ascii=False)}")
+                
+                # 尝试解析 Transcription
+                task_result = data_obj.get('Result', {})
+                transcription_data = task_result.get('Transcription', {})
+                
+                if isinstance(transcription_data, str) and transcription_data.startswith('http'):
+                    import requests
+                    print(f"Downloading transcription from {transcription_data}")
+                    t_resp = requests.get(transcription_data)
+                    if t_resp.status_code == 200:
+                        content = t_resp.json()
+                        print(f"Downloaded content structure keys: {content.keys()}")
+                        # print(f"Content sample: {json.dumps(content, indent=2, ensure_ascii=False)[:500]}...")
+                    else:
+                        print(f"Failed to download: {t_resp.status_code}")
+                
+                break
+            elif task_status == 'FAILED':
+                print(f"Task failed: {data_obj}")
+                break
+                
+            time.sleep(5)
+            
+    except Exception as e:
+        print(f"Error: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    test_tingwu_transcription()