新增s3接口基于minio

database/s3.py

@@ -12,6 +12,13 @@ from database.config import minio_block,minio_client
 from minio import Minio
 from pathlib import Path
 from sqlmodel import Field, SQLModel,Relationship,Column,Session,select,func,UniqueConstraint,PickleType,text
+import s3fs
+def get_client():
+    s3 = s3fs.S3FileSystem(
+      key='miniokey...',
+      secret='asecretkey...',
+      endpoint_url='https://...'
+   )
 
 class S3Object(SQLModel,table=False):
     type:Optional[str]= Field(default=None)
@@ -54,7 +61,7 @@ class S3:
     # s3client is minio python SDK client
     def __init__(self, bucket='swl', client=None) -> None:
         self.bucket = bucket
-        self.minio_client = client
+        self.minio_client:Minio = client
     # 根据本地时间自动获取对象存储前缀，如 log/2024-04-25/xxx
     def get_object_prefix(self, dir='log'):
         # 获取时分秒，并且符合路径的格式
@@ -64,7 +71,7 @@ class S3:
         # 获取时分秒毫秒，并且符合路径的格式，如 023213_123.json
         now = datetime.datetime.now()
         # 格式化时间：小时、分钟、秒、毫秒
-        formatted_time = now.strftime("%H%M%S_%f")  # %f 提供了微秒，所以我们取前三个数字作为毫秒
+        formatted_time = now.strftime("%y%m%d-%H%M%S_%f")  # %f 提供了微秒，所以我们取前三个数字作为毫秒
         return formatted_time
 
     # def put_dict(self, obj, name_by_time=True):
@@ -100,9 +107,8 @@ class S3:
         # 根据文件后缀自动识别 content_type
         if content_type == '':
             content_type = mimetypes.guess_type(file_path)[0] or 'application/octet-stream'
-        object_name = self.get_object_prefix() + '/' + object_name
         # 使用 put_object 上传数据  
-        self.minio_client.fput_object(  
+        return self.minio_client.fput_object(  
             bucket_name=self.bucket,   
             file_path=file_path,   
             object_name=object_name,   

douyin/flow/check_msg.py

@@ -5,7 +5,7 @@ import time
 import os
 import sys
 sys.path.append(os.path.dirname(os.path.dirname(__file__)))
-
+sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
 from dp.page import page
 from douyin import base
 from conf.config import logger,OUTPUT
@@ -60,3 +60,29 @@ def check_has_im_msg_flow(home_url:str):
     ele_im = get_im_icon_ele()
     ele_has_msg = get_im_icon_red_pot_ele(ele_im)
     return ele_im, ele_has_msg
+
+def main():
+    from database.s3 import S3Object,S3
+    from database.config import minio_client
+    from pathlib import Path
+    # res = minio_client.fput_object(bucket_name='public', object_name=base_path+file_name, file_path=r"I:\code\ai-yunying\live-online-people\output\page\\"+file_name,content_type= 'text/html')
+    
+    file_name = '点击陌生人对话框.html'
+    page_dir = Path(r'I:\code\ai-yunying\live-online-people\output\page\\')
+    file_path = page_dir/file_name
+    f = open(file_path, 'w')
+    f.write(tab.html)
+    s3minio = S3(bucket='public',client=minio_client)
+    base_path = '/md/ai-yunying/'
+    obj_name = base_path+ s3minio.get_object_name_by_time() + '-'+ file_name
+    res = s3minio.fput(r'I:\code\ai-yunying\live-online-people\output\page\\'+file_name, obj_name)
+    logger.info(f"{res.bucket_name } {res.object_name} {res._http_headers}")
+    # ele_dlg = tab.ele('xpath://div[@data-e2e="listDlg-container"]')
+    # logger.info(f"{ele_dlg.text}")
+    # check_has_im_msg_flow(home_url="https://www.douyin.com/user/self")
+    # S3Object(path='runtime/check_msg-stranger-msg.html').put(tab.html)
+    # f = open(r'I:\code\ai-yunying\live-online-people\output\page\点击陌生人消息.html', 'w')
+    # f.write(tab.html)
+
+if __name__ == "__main__":
+    main()

dp/page_test.py

@@ -0,0 +1,34 @@
+import os
+import sys
+sys.path.append(os.path.dirname(os.path.dirname(__file__)))
+from dp.page import page
+from conf.config import logger,OUTPUT
+from database.config import  minio_block
+
+tab=page.tab
+
+def main():
+    from database.s3 import S3Object,S3
+    from database.config import minio_client
+    from pathlib import Path
+    # res = minio_client.fput_object(bucket_name='public', object_name=base_path+file_name, file_path=r"I:\code\ai-yunying\live-online-people\output\page\\"+file_name,content_type= 'text/html')
+    
+    file_name = '点击陌生人对话框.html'
+    page_dir = Path(r'I:\code\ai-yunying\live-online-people\output\page\\')
+    file_path = page_dir/file_name
+    f = open(file_path, 'w')
+    f.write(tab.html)
+    s3minio = S3(bucket='public',client=minio_client)
+    base_path = '/md/ai-yunying/'
+    obj_name = base_path+ s3minio.get_object_name_by_time() + '-'+ file_name
+    res = s3minio.fput(r'I:\code\ai-yunying\live-online-people\output\page\\'+file_name, obj_name)
+    logger.info(f"{res.bucket_name } {res.object_name} {res._http_headers}")
+    # ele_dlg = tab.ele('xpath://div[@data-e2e="listDlg-container"]')
+    # logger.info(f"{ele_dlg.text}")
+    # check_has_im_msg_flow(home_url="https://www.douyin.com/user/self")
+    # S3Object(path='runtime/check_msg-stranger-msg.html').put(tab.html)
+    # f = open(r'I:\code\ai-yunying\live-online-people\output\page\点击陌生人消息.html', 'w')
+    # f.write(tab.html)
+
+if __name__ == "__main__":
+    main()