python实现兼容阿里云oss协议的文件服务端代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python实现兼容阿里云oss协议的文件服务端代码

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
mini_oss_server.py —— 单文件版阿里云 OSS 兼容文件存储服务 (MiniOSS)
====================================================================

特性
----
1. 完整兼容阿里云 OSS V1 签名协议, 官方 oss2 SDK / 各语言 OSS SDK 直接使用, 零改造切换
2. Bucket: 创建 / 删除 / 列举 (PutBucket / DeleteBucket / ListBuckets / HeadBucket)
3. 对象: 上传 / 下载 / 追加 / 拷贝 / 删除 / 批量删除 / 分片上传 / Range 断点下载
4. 图片处理: x-oss-process=image/resize,w_100 等 (resize / crop / rotate / blur /
   sharpen / quality / format / roundcorner / info), 基于 Pillow
5. 本地文件系统存储: DATA_DIR/<bucket>/<key>, 元数据存于 DATA_DIR/.meta/
6. 单文件、零配置, 仅依赖 Python 标准库; 图片处理需可选安装 Pillow

依赖
----
- 运行: Python 3.7+ (标准库即可)
- 图片缩放: pip install pillow
- 客户端: pip install oss2

启动
----
    python mini_oss_server.py                          # 默认 0.0.0.0:8366, 数据目录 ./oss_data
    python mini_oss_server.py --port 9000 --data D:/oss-data --ak myak --sk mysk

环境变量 (均可被命令行参数覆盖):
    OSS_PORT / OSS_HOST / OSS_DATA_DIR
    OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET
    OSS_AUTO_CREATE_BUCKET  默认 1 (PUT 时 bucket 不存在自动创建)
    OSS_ALLOW_ANONYMOUS     默认 0 (强制签名校验, 与真实 OSS 行为一致)

oss2 客户端切换示例 (把 endpoint 从阿里云域名换成本机地址即可):
----------------------------------------------------------------
    import oss2
    auth = oss2.Auth('accesskeyid', 'accesskeysecret')      # 与服务端 --ak/--sk 保持一致
    bucket = oss2.Bucket(auth, 'http://127.0.0.1:8366', 'my-bucket')

    bucket.put_object('a.jpg', b'hello')                       # 上传 (bucket 不存在会自动创建)
    bucket.get_object('a.jpg').read()                          # 下载
    bucket.get_object('a.jpg', process='image/resize,w_100')   # 图片缩放
    bucket.delete_object('a.jpg')                              # 删除
----------------------------------------------------------------
"""

import argparse
import base64
import hashlib
import hmac
import io
import json
import mimetypes
import os
import re
import shutil
import sys
import tempfile
import threading
import time
import uuid
import xml.etree.ElementTree as ET
from email.utils import formatdate
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qsl, quote, unquote, urlsplit
from xml.sax.saxutils import escape as xml_escape

try:
    from PIL import Image, ImageDraw, ImageFilter, ImageOps
    PIL_AVAILABLE = True
except ImportError:
    PIL_AVAILABLE = False

# ============================ 配置 ============================

HOST = os.getenv('OSS_HOST', '0.0.0.0')
PORT = int(os.getenv('OSS_PORT', '8366'))
ACCESS_KEY_ID = os.getenv('OSS_ACCESS_KEY_ID', 'accesskeyid')
ACCESS_KEY_SECRET = os.getenv('OSS_ACCESS_KEY_SECRET', 'accesskeysecret')
DATA_DIR = os.path.abspath(os.getenv('OSS_DATA_DIR', './oss_data'))
AUTO_CREATE_BUCKET = os.getenv('OSS_AUTO_CREATE_BUCKET', '1') == '1'
ALLOW_ANONYMOUS = os.getenv('OSS_ALLOW_ANONYMOUS', '0') == '1'
REGION = os.getenv('OSS_REGION', 'oss-cn-hangzhou')
OWNER_ID = 'minioss'
OWNER_NAME = 'minioss'

# 与 oss2 SDK (auth.py _subresource_key_set) 完全一致的子资源列表, 参与签名计算
SUBRESOURCES = frozenset([
    'response-content-type', 'response-content-language', 'response-cache-control', 'logging',
    'response-content-encoding', 'acl', 'uploadId', 'uploads', 'partNumber', 'group', 'link',
    'delete', 'website', 'location', 'objectInfo', 'objectMeta', 'response-expires',
    'response-content-disposition', 'cors', 'lifecycle', 'restore', 'qos', 'referer', 'stat',
    'bucketInfo', 'append', 'position', 'security-token', 'live', 'comp', 'status', 'vod',
    'startTime', 'endTime', 'x-oss-process', 'symlink', 'callback', 'callback-var', 'tagging',
    'encryption', 'versions', 'versioning', 'versionId', 'policy', 'requestPayment',
    'x-oss-traffic-limit', 'qosInfo', 'asyncFetch', 'x-oss-request-payer', 'sequential',
    'inventory', 'inventoryId', 'continuation-token', 'worm', 'wormId', 'wormExtend',
    'replication', 'replicationLocation', 'replicationProgress', 'transferAcceleration', 'cname',
    'metaQuery', 'x-oss-ac-source-ip', 'x-oss-ac-subnet-mask', 'x-oss-ac-vpc-id',
    'x-oss-ac-forward-allow', 'resourceGroup', 'style', 'styleName', 'x-oss-async-process',
    'regionList', 'x-oss-write-get-object-response', 'bucketArchiveDirectRead', 'httpsConfig',
    'redundancyTransition', 'x-oss-target-redundancy-type', 'x-oss-redundancy-transition-taskid',
    'accessPoint', 'accessPointPolicy', 'publicAccessBlock', 'x-oss-access-point-name',
    'requesterQosInfo', 'qosRequester', 'resourcePoolInfo', 'resourcePool', 'resourcePoolBuckets',
])

BUCKET_RE = re.compile(r'^[a-z0-9][a-z0-9-]{1,61}[a-z0-9]$')
XML_DECL = '<?xml version="1.0" encoding="UTF-8"?>\n'
IO_LOCK = threading.Lock()

# GET 请求 response-* 查询参数覆盖响应头
RESPONSE_OVERRIDES = {
    'response-content-type': 'Content-Type',
    'response-content-language': 'Content-Language',
    'response-expires': 'Expires',
    'response-cache-control': 'Cache-Control',
    'response-content-disposition': 'Content-Disposition',
    'response-content-encoding': 'Content-Encoding',
}

# ============================ 异常 ============================


class OSSError(Exception):
    """带 OSS 错误码的异常, 会被转换成标准 <Error> XML 响应"""

    def __init__(self, status, code, message, headers=None):
        super(OSSError, self).__init__(message)
        self.status = status
        self.code = code
        self.message = message
        self.headers = headers or {}


# ============================ 通用工具 ============================


def new_request_id():
    return uuid.uuid4().hex[:24].upper()


def http_date(ts=None):
    """RFC 7231 GMT 时间, 如: Tue, 15 Nov 1994 12:45:26 GMT"""
    return formatdate(ts, usegmt=True)


def iso_time(ts):
    """OSS XML 中使用的 ISO8601 时间, 如: 2012-02-24T06:07:48.000Z"""
    return time.strftime('%Y-%m-%dT%H:%M:%S.000Z', time.gmtime(ts))


def md5_of_file(path, chunk=1 << 20):
    h = hashlib.md5()
    with open(path, 'rb') as f:
        for blk in iter(lambda: f.read(chunk), b''):
            h.update(blk)
    return h.hexdig.........完整代码请登录后点击上方下载按钮下载查看