'commit'

2025-08-21 10:47:52 +08:00
parent 58cd2cbf7d
commit 3f2eb61c09
3 changed files with 63 additions and 0 deletions
--- a/dsLightRag/Test/TestObsUtilStream.py
+++ b/dsLightRag/Test/TestObsUtilStream.py
@@ -0,0 +1,60 @@
+import sys
+import os
+import uuid
+import requests
+# 添加项目根目录到Python路径
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+from Util.ObsUtil import ObsUploader
+
+def main():
+    # 测试视频URL - 使用用户提供的可播放视频URL
+    test_video_url = "https://v3-vvecloud.yangyi08.com/3e4dec4994152917baa2f14302927595/68a69142/video/tos/cn/tos-cn-v-242bcc/oEiAEhzf0xyEw7FaNgU0QAiGSAZBOCQgKm1tmA/?a=7073&ch=0&cr=0&dr=0&er=0&lr=default&cd=0%7C0%7C0%7C0&br=4542&bt=4542&cs=0&ds=4&ft=xztlUQhhe6BMyqjL_jkJD12Nzj&mime_type=video_mp4&qs=13&rc=M2w1Nm85cjk7NTgzNGczM0BpM2w1Nm85cjk7NTgzNGczM0A2YWo1MmRzM2dhLS1kXjBzYSM2YWo1MmRzM2dhLS1kXjBzcw%3D%3D&btag=80000e00008000&dy_q=1755743021&l=202508211023414B2120DB0DDDEF74FB7F"
+    
+    # 使用UUID生成唯一文件名
+    unique_id = uuid.uuid4()
+    test_object_key = f"HuangHai/JiMeng/{unique_id}.mp4"
+    
+    # 临时文件保存路径
+    temp_file_path = os.path.join(os.path.dirname(__file__), f"{unique_id}.mp4")
+    
+    try:
+        # 1. 下载视频到本地临时文件
+        print(f"开始下载视频: {test_video_url}")
+        with requests.get(test_video_url, stream=True, timeout=3600) as r:
+            r.raise_for_status()
+            with open(temp_file_path, 'wb') as f:
+                for chunk in r.iter_content(chunk_size=8192):
+                    f.write(chunk)
+        print(f"视频下载完成，保存至: {temp_file_path}")
+        
+        # 2. 实例化ObsUploader
+        obs_uploader = ObsUploader()
+        
+        print(f"开始上传文件到OBS: {temp_file_path}\n对象键: {test_object_key}")
+        
+        # 3. 调用文件上传方法
+        result = obs_uploader.upload_file(
+            file_path=temp_file_path,
+            object_key=test_object_key
+        )
+        
+        # 4. 打印结果
+        if hasattr(result, 'get') and result.get('success'):
+            print(f"上传成功! 视频地址: {result.get('url')}")
+        else:
+            print(f"上传失败: {str(result)}")
+            
+    except Exception as e:
+        print(f"测试过程中发生异常: {str(e)}")
+    finally:
+        # 5. 清理临时文件
+        if os.path.exists(temp_file_path):
+            os.remove(temp_file_path)
+            print(f"临时文件已删除: {temp_file_path}")
+
+def test_upload_file():
+    # 单元测试函数，可用于集成到测试框架
+    main()
+
+if __name__ == "__main__":
+    main()