BrandCultivation/utils/file_stream.py

from config import load_service_config
from io import BytesIO
import os
import pandas as pd
import requests


class FileStreamUtils:
    cfgs = load_service_config()
    upload_url = cfgs["aliyun"]["upload_url"]
    download_url = cfgs["aliyun"]["download_url"]
    # cookies = cfgs["aliyun"]['cookies']
     # 设置请求头
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        "Accept": "*/*",
    }
    
    @staticmethod
    def upload_files(reports_dir, files):
        files_id = {}
        for filename in files:
            file_path = os.path.join(reports_dir, f'{filename}.xlsx')
            try:
                with open(file_path, 'rb') as f:
                    files = {'file': (os.path.basename(file_path), f)}

                    response = requests.post(
                        FileStreamUtils.upload_url,
                        headers=FileStreamUtils.headers,
                        files=files,
                        # cookies=FileStreamUtils.cookies,
                        verify=True
                    )
                    
                    if response.json()["success"]:
                        file_id = response.json()["data"]["file_info"]["fileid"]
                        files_id[filename] = file_id
            except requests.exceptions.RequestException as e:
                print("请求出错:", e)
                return None
            except Exception as e:
                return None
                
        return files_id
    
    @staticmethod
    def download_file(file_id, file_type='xlsx'):
        """通过file_id从阿里云文件数据库下载文件"""
        try:
            params = {
                'fileid': file_id,
                'action': 'download'
            }
            
            response = requests.get(
                FileStreamUtils.download_url,
                headers=FileStreamUtils.headers,
                # cookies=FileStreamUtils.cookies,
                params=params,
                verify=True
            )
            
            if response.status_code == 200:
                file_content = BytesIO(response.content)
                print('请求成功了')
                if file_type == 'xlsx':
                    data = pd.read_excel(file_content)
                elif file_type == 'csv':
                    data = pd.read_csv(file_content)
                else:
                    raise ValueError(f"不支持的文件类型：{file_type}" )
                
                return data
            else:
                print(f'请求失败了{response.status_code}')
                return None
        except requests.exceptions.RequestException as e:
            print("Request Error: ", e)
            return None
        except Exception as e:
            print("File download Error: ", e)
            return None
    
if __name__ == '__main__':
    # print(FileStreamUtils.cfgs["aliyun"]["cookies"])
    file_id = '11C1AC088863421C9BC32A5E722F5147'
    
    data = FileStreamUtils.download_file(file_id)
    data.to_excel('./recommend_list.xlsx', index=False)