用途：BOS需要一个简单的文件查看页面，用于查看和下载
在百度云CFC函数计算中创建新的“空白函数”，运行时选择“Python 3.6“，超时时间选择 30s。
触发器选择“HTTP触发器”，URL路径为/{path+}，HTTP方法为GET，不开启验证
创建完成后，将bos_fileserver.py的代码贴入代码框，调整其中的
BOS_HOST = "bj.bcebos.com"
BOS_BUCKET = "bos-fileserver"
即可
然后去左边栏的“触发器”中查看 https endpoint，类似于https://xcvsf323fds.cfc-execute.bj.baidubce.com
目前需要BOS开启公共读，后续可以看CFC是否支持301重定向，就可以提供私有链接
| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| # 为了实现生成文件列表的功能，您需要一个 BOS bucket | |
| import os | |
| import datetime | |
| import sys | |
| import logging | |
| from baidubce.services.bos.bos_client import BosClient | |
| from baidubce.bce_client_configuration import BceClientConfiguration | |
| from baidubce.auth.bce_credentials import BceCredentials | |
| logging.basicConfig(level=logging.INFO, stream=sys.stdout) | |
| logger = logging.getLogger() | |
| BOS_HOST = "bj.bcebos.com" | |
| # 如果开启CDN或者其他情况，下载的HOST和API的HOST不同，此处可以配置 | |
| BOS_DOWNLOAD_HOST = BOS_HOST | |
| BOS_BUCKET = "bos-fileserver" | |
| def handler(event, context): | |
| # event 格式参见 https://cloud.baidu.com/doc/CFC/s/kjzmgugty#event%E7%9A%84%E6%A0%BC%E5%BC%8F | |
| path = event["pathParameters"]["path"] | |
| bos_client = init_bos_client(BOS_HOST) | |
| logger.info("going to list file [%s] from [%s]" %(path, BOS_BUCKET)) | |
| # list file from BOS | |
| try: | |
| objects = bos_client.list_objects(BOS_BUCKET, prefix=path, delimiter = '/') | |
| except Exception as e: | |
| print("list file [%s] failed" % path) | |
| print(e) | |
| raise e | |
| logger.info("list file [%s] success" % path) | |
| logger.info("generate index function start") | |
| starttime = datetime.datetime.now() | |
| try: | |
| body = generate_index(objects, path) | |
| except Exception as e: | |
| print(e) | |
| raise e | |
| endtime = datetime.datetime.now() | |
| logger.info("generate index take " + str((endtime-starttime).microseconds/1000) + "ms") | |
| resp = { | |
| "isBase64Encoded": False, | |
| "statusCode": 200, | |
| "headers": {}, | |
| "body": body | |
| } | |
| return resp | |
| def init_bos_client(host): | |
| ak = os.environ['BCE_ACCESS_KEY_ID'] | |
| sk = os.environ['BCE_ACCESS_KEY_SECRET'] | |
| token = os.environ['BCE_SESSION_TOKEN'] | |
| config = BceClientConfiguration(credentials=BceCredentials(ak, sk), endpoint = host, security_token=token) | |
| return BosClient(config) | |
| def generate_bos_file_path(path): | |
| """生成BOS文件的下载地址""" | |
| return f"http://{BOS_DOWNLOAD_HOST}/{BOS_BUCKET}/{path}" | |
| def generate_index(objects, path): | |
| files = [] | |
| # INFO:root:{last_modified:u'2020-12-21T06:19:15Z',key:u'files/index.html',size:38} | |
| for obj in objects.contents: | |
| name = obj.key.split('/')[-1] | |
| if not name: | |
| # 过滤掉本目录 | |
| continue | |
| files.append(dict(path=generate_bos_file_path(obj.key), name=name, type='file', last_modified_iso=obj.last_modified, size_bytes=obj.size)) | |
| for obj in objects.common_prefixes: | |
| dir_name = obj.prefix.split('/')[-2] + '/' | |
| files.append(dict(path='/' + obj.prefix, name=dir_name, type='folder', last_modified_iso='-', size_bytes=-1)) | |
| sorted(files, key=lambda x: x["name"]) | |
| logger.info(files) | |
| body = """<!DOCTYPE html><html> | |
| <head> | |
| <title>Index of {path} </title> | |
| <meta charset="UTF-8"> | |
| </head> | |
| <body> | |
| <h1>Index of {path}</h1> | |
| <pre>由于浏览器限制，文件下载请在链接上右键在新标签页打开</pre> | |
| <hr> | |
| <pre><a href="../">../</a> | |
| """.format(path=path) | |
| for one in files: | |
| body += '<a href="{path}">{name}</a> {last_modified_iso} {size_bytes}\n'.format(**one) | |
| body += """</pre><hr></body></html>""" | |
| logger.info("generate index success") | |
| return body |