测试 oss

test.py

@@ -1,7 +1,35 @@
-from common.scheduling_db import MysqlHelper
+import oss2
+import requests
+import urllib.parse
 
-sql = "select user_name from changsha_user_accounts where piaoquan_account_id = 69062464;"
-result = MysqlHelper.get_values(
-    log_type="author", crawler="test", env="prod", sql=sql
-)
-print(result)
+OSS_ACCESS_KEY_ID = "LTAIP6x1l3DXfSxm"
+OSS_ACCESS_KEY_SECRET = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
+# OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou-internal.aliyuncs.com"  # 内网地址
+
+
+OSS_BUCKET_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com"  # 外网地址
+
+
+def upload_to_oss(video_id):
+    """
+    Uploads video file to OSS
+    :param video_id:
+    :return:
+    """
+    # print("start download video...")
+    # await download_video(video_url, "temp.mp4")
+    # print("video download successfully done")
+    oss_object_key = f'single_video/{video_id}'
+    # print(key)
+    auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
+    bucket = oss2.Bucket(auth, OSS_BUCKET_ENDPOINT, "art-pubbucket")
+    response = bucket.put_object_from_file(oss_object_key, "temp.mp4")
+    print("Upload completed successfully.")
+    if 'Content-Length' in response.headers:
+        return {
+            'status': response.status,
+            'oss_object_key': oss_object_key}
+    raise AssertionError(f'OSS上传失败，请求ID: \n{response.headers["x-oss-request-id"]}')
+
+
+upload_to_oss(video_id="123456789")