Python mimetypes 模块，MimeTypes() 实例源码

def send_email(subject, recipients, template, **kwargs):
    if not isinstance(recipients, list):
        recipients = list(recipients)
    msg = Message(subject, reply_to=current_app.config['MAIL_DEFAULT_SENDER'],
                  recipients=recipients)
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)

    attachments = kwargs.get('attachments', [])
    mimes = MimeTypes()
    for file in attachments:
        path_ = os.path.join(current_app.config['APP_UPLOADS'], file)
        with current_app.open_resource(path_) as fp:
            mime = mimes.guess_type(fp.name)
            msg.attach(path_, mime[0], fp.read())

    app = current_app._get_current_object()
    t = Thread(target=send_async_email, args=[app, msg])
    t.start()
    return t

def video_tag(self, attrs=None):
        if attrs is None:
            attrs = {}
        else:
            attrs = attrs.copy()
        if self.thumbnail:
            attrs['poster'] = self.thumbnail.url

        transcodes = self.transcodes.exclude(processing=True).filter(error_message__exact='')
        sources = []
        for transcode in transcodes:
            sources.append("<source src='{0}' type='video/{1}' >".format(transcode.url, transcode.media_format.name))

        mime = mimetypes.MimeTypes()
        sources.append("<source src='{0}' type='{1}'>"
                       .format(self.url, mime.guess_type(self.url)[0]))

        sources.append("<p>Sorry, your browser doesn't support playback for this video</p>")
        return mark_safe(
            "<video {0}>\n{1}\n</video>".format(flatatt(attrs), "\n".join(sources)))

def upload(session: Session, token: str, file_path: str) -> Response:
    """Upload logo.

    :param session: Supplier session object
    :param token: CSRF token required to upload the file
    :param file_path: absolute path to the uploaded file
    :return: response object
    """
    headers = {"Referer": get_absolute_url("ui-buyer:upload-logo")}
    url = get_absolute_url("ui-buyer:upload-logo")
    data = {
        "csrfmiddlewaretoken": token,
        "company_profile_logo_edit_view-current_step": "logo",
    }
    with open(file_path, "rb") as f:
        picture = f.read()
    mime = mimetypes.MimeTypes().guess_type(file_path)[0]
    files = {"logo-logo": (os.path.basename(file_path), picture, mime)}
    response = make_request(
        Method.POST, url, session=session, headers=headers, data=data,
        files=files)
    return response

def main(argv=None):
    if argv is None:
        argv = sys.argv
    p = argparse.ArgumentParser(
        description='SAVAPI Client',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    p.add_argument('--savapi-host', dest='host', help='SAVAPI service host',
                   default='127.0.0.1')
    p.add_argument('--savapi-port', dest='port', help='SAVAPI service port',
                   default=9999, type=int)
    p.add_argument('file', help='Absolute path of file')

    try:
        args = p.parse_args(argv[1:])
    except SystemExit:
        return 2

    log = logging.getLogger()
    log.addHandler(logging.StreamHandler())
    log.setLevel(logging.DEBUG)

    archive_mimes = ('application/zip', 'application/x-tar', 'application/rar')
    if magic:
        mime = magic.Magic(mime=True)
        mimetype = mime.from_file(args.file)
    else:
        mimes = MimeTypes()
        mimetype, _ = mimes.guess_type(args.file)

    with SAVAPIClient(args.host, args.port) as savapi:
        r = savapi.command('SET', 'PRODUCT', '10776')
        log.info(r)
        if mimetype in archive_mimes:
            r = savapi.command('SET', 'ARCHIVE_SCAN', '1')
            log.info(r)
        for r in savapi.scan(args.file):
            print('{} {} <<< {}'.format(r.code, r.definition, r.data))
            if int(r.code) in [319, 350]:
                savapi.command('QUIT')

def send_email(sender, recipients, subject, text_body, html_body=None,
               attach=None):
    msg = Message(subject, reply_to=sender, recipients=recipients)
    msg.body = text_body
    msg.html = html_body
    if attach:
        mimes = MimeTypes()
        for file in attach:
            path_ = os.path.join(current_app.config['APP_UPLOADS'], file)
            with current_app.open_resource(path_) as fp:
                mime = mimes.guess_type(fp.name)
                msg.attach(file, mime[0], fp.read())
    mail.send(msg)

def __init__(self, allowed_domains, depth_limit=1):
        super().__init__()
        self.allowed_domains = allowed_domains
        # allow local links
        self.allowed_domains.append('')
        self.mime = MimeTypes()
        # Queue implemented as dict as we need to store depth data alongside url
        self.queue = dict()  # { url: depth, .. }
        # TODO: make crawled a class
        self.crawled = dict()  # { url: {type: type, outlinks: {link: count, ..}}, .. }
        self.depth_limit = depth_limit

def builtin_mimetypes(func):

    @wraps(func)
    async def decorated(*args, **kwargs):
        with patch.object(utils, 'magic') as magic:
            magic.__bool__.return_value = False
            with patch.object(utils, 'mime') as mime:
                mime.guess_type.side_effect = mimetypes.MimeTypes().guess_type

                await func(*args, **kwargs)

    return decorated

def item_enclosure_mime_type(self, item):
        mime = MimeTypes()
        try:
            if item.direction == 1:
                filename = join(self.folder, '{1}=={2}=={3}=={4}==longs.png'.format(\
                    item.broker.slug, item.symbol.symbol, item.period.period, item.system.title))
            elif item.direction == 2:
                filename = join(self.folder, '{1}=={2}=={3}=={4}==shorts.png'.format(\
                    item.broker.slug, item.symbol.symbol, item.period.period, item.system.title))

            mime_type = mime.guess_type(filename)[0]
            return mime_type
        except:
            return None

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_encoding(self):
        getpreferredencoding = locale.getpreferredencoding
        self.addCleanup(setattr, locale, 'getpreferredencoding',
                                 getpreferredencoding)
        locale.getpreferredencoding = lambda: 'ascii'

        filename = support.findfile("mime.types")
        mimes = mimetypes.MimeTypes([filename])
        exts = mimes.guess_all_extensions('application/vnd.geocube+xml',
                                          strict=True)
        self.assertEqual(exts, ['.g3', '.g\xb3'])

def setUp(self):
        # ensure all entries actually come from the Windows registry
        self.original_types_map = mimetypes.types_map.copy()
        mimetypes.types_map.clear()
        mimetypes.init()
        self.db = mimetypes.MimeTypes()

def item_enclosure_mime_type(self, item):
        mime = MimeTypes()
        try:
            mime_type = mime.guess_type("{0}{1}".format(self.folder, item.image))[0]
            return mime_type
        except:
            return None

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_registry_parsing(self):
        # the original, minimum contents of the MIME database in the
        # Windows registry is undocumented AFAIK.
        # Use file types that should *always* exist:
        eq = self.assertEqual
        mimetypes.init()
        db = mimetypes.MimeTypes()
        eq(db.guess_type("foo.txt"), ("text/plain", None))
        eq(db.guess_type("image.jpg"), ("image/jpeg", None))
        eq(db.guess_type("image.png"), ("image/png", None))

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_registry_parsing(self):
        # the original, minimum contents of the MIME database in the
        # Windows registry is undocumented AFAIK.
        # Use file types that should *always* exist:
        eq = self.assertEqual
        mimetypes.init()
        db = mimetypes.MimeTypes()
        eq(db.guess_type("foo.txt"), ("text/plain", None))
        eq(db.guess_type("image.jpg"), ("image/jpeg", None))
        eq(db.guess_type("image.png"), ("image/png", None))

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_encoding(self):
        getpreferredencoding = locale.getpreferredencoding
        self.addCleanup(setattr, locale, 'getpreferredencoding',
                                 getpreferredencoding)
        locale.getpreferredencoding = lambda: 'ascii'

        filename = support.findfile("mime.types")
        mimes = mimetypes.MimeTypes([filename])
        exts = mimes.guess_all_extensions('application/vnd.geocube+xml',
                                          strict=True)
        self.assertEqual(exts, ['.g3', '.g\xb3'])

def setUp(self):
        # ensure all entries actually come from the Windows registry
        self.original_types_map = mimetypes.types_map.copy()
        mimetypes.types_map.clear()
        mimetypes.init()
        self.db = mimetypes.MimeTypes()

def validate_video_file(upload):
    """
    Validates uploaded file name by using mimetypes, part of Python standard
    library, to guess the mime type from the file extension.

    The file extension is not a reliable way to determine mime type. A better
    alternative is using the python-magic library, but that requires the
    libmagic library which is extra overhead, so not worth.
    """
    mime_type = mimetypes.MimeTypes().guess_type(upload.name)[0]
    if mime_type not in VIDEO_MIME_TYPES:
        raise ValidationError('File type not supported. MP4, Quicktime, or WebM recommended.')

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_registry_parsing(self):
        # the original, minimum contents of the MIME database in the
        # Windows registry is undocumented AFAIK.
        # Use file types that should *always* exist:
        eq = self.assertEqual
        mimetypes.init()
        db = mimetypes.MimeTypes()
        eq(db.guess_type("foo.txt"), ("text/plain", None))
        eq(db.guess_type("image.jpg"), ("image/jpeg", None))
        eq(db.guess_type("image.png"), ("image/png", None))

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_encoding(self):
        getpreferredencoding = locale.getpreferredencoding
        self.addCleanup(setattr, locale, 'getpreferredencoding',
                                 getpreferredencoding)
        locale.getpreferredencoding = lambda: 'ascii'

        filename = support.findfile("mime.types")
        mimes = mimetypes.MimeTypes([filename])
        exts = mimes.guess_all_extensions('application/vnd.geocube+xml',
                                          strict=True)
        self.assertEqual(exts, ['.g3', '.g\xb3'])

def setUp(self):
        # ensure all entries actually come from the Windows registry
        self.original_types_map = mimetypes.types_map.copy()
        mimetypes.types_map.clear()
        mimetypes.init()
        self.db = mimetypes.MimeTypes()

def setUp(self):
        self.db = mimetypes.MimeTypes()

def setUp(self):
        # ensure all entries actually come from the Windows registry
        self.original_types_map = mimetypes.types_map.copy()
        mimetypes.types_map.clear()
        mimetypes.init()
        self.db = mimetypes.MimeTypes()

def setUp(self):
        self.db = mimetypes.MimeTypes()

def test_encoding(self):
        getpreferredencoding = locale.getpreferredencoding
        self.addCleanup(setattr, locale, 'getpreferredencoding',
                                 getpreferredencoding)
        locale.getpreferredencoding = lambda: 'ascii'

        filename = support.findfile("mime.types")
        mimes = mimetypes.MimeTypes([filename])
        exts = mimes.guess_all_extensions('application/vnd.geocube+xml',
                                          strict=True)
        self.assertEqual(exts, ['.g3', '.g\xb3'])

def setUp(self):
        # ensure all entries actually come from the Windows registry
        self.original_types_map = mimetypes.types_map.copy()
        mimetypes.types_map.clear()
        mimetypes.init()
        self.db = mimetypes.MimeTypes()

def guess_mime_type(audio_file_path, forced_mime_type=None):
    mime = MimeTypes()
    mime_type = forced_mime_type or mime.guess_type(audio_file_path)[0]
    if mime_type not in SUPPORTED_CONTENT_TYPES:
        raise ValueError('MimeType {} not recognized or supported for '
                         'audio file {}'.format(mime_type, audio_file_path))
    return mime_type

def view(self, request, **kwargs):
        '''
        allow a file to be viewed as opposed to download. This is particularly needed when a video file is stored
        in the fileservice and user wants to be able to use a view the video as opposed to having to download it
        first. It passes the serving of the file to nginx/apache which will return all the proper headers allowing,
        say, html5's video viewer's 'seek' indicator/knob to work. Otherwise the video is only played sequentially

        Note that nginx/apache need to be configured accordingly. nginx for example:
        location /var/lib/geoserver_data/file-service-store/ {
           # forces requests to be authorized
           internal;
           alias   /var/lib/geoserver_data/file-service-store/;
        }

        for apache, need to install xsendfile module, enable it, set the path and then
        XSendFile on
        XSendFilePath /var/lib/geoserver_data/file-service-store

        example use:
        /fileservice/view/med.mp4
        or
        /fileservice/med.mp4/view

        Note that media players tend to require the route to end with the filename like /fileservice/view/med.mp4
        '''

        # method check to avoid bad requests
        self.method_check(request, allowed=['get'])
        # Must be done otherwise endpoint will be wide open
        self.is_authenticated(request)

        file_item_name = kwargs.get('name', None)
        url = urllib.pathname2url(file_item_name)
        mime = MimeTypes()
        mime_type = mime.guess_type(url)
        response = HttpResponse(content_type=mime_type[0])
        file_with_route = smart_str('{}{}'.format(helpers.get_fileservice_dir(), file_item_name))
        # apache header
        response['X-Sendfile'] = file_with_route
        # nginx header
        response['X-Accel-Redirect'] = file_with_route

        return response

def prepare_form_data(token: str, case_study: CaseStudy) -> (dict, dict):
    """Prepare form data based on the flags and custom values provided.

    :param token: CSRF middleware token required to submit the form
    :param case_study: a CaseStudy namedtuple with random data
    :return: a tuple consisting of form data and files to upload
    """
    data = {
        "csrfmiddlewaretoken": token,
        "supplier_case_study_wizard_view-current_step": "rich-media",
        "rich-media-image_one_caption": case_study.caption_1,
        "rich-media-image_two_caption": case_study.caption_2,
        "rich-media-image_three_caption": case_study.caption_3,
        "rich-media-testimonial": case_study.testimonial,
        "rich-media-testimonial_name": case_study.source_name,
        "rich-media-testimonial_job_title": case_study.source_job,
        "rich-media-testimonial_company": case_study.source_company
    }

    paths = [case_study.image_1, case_study.image_2, case_study.image_3]

    def get_basename(path):
        return basename(path) if path is not None else ""

    def get_mimetype(path):
        return MimeTypes().guess_type(path)[0] if path is not None else ""

    def read_image(path: str):
        res = ""
        if path is not None:
            with open(path, "rb") as f:
                res = f.read()
        return res

    name_1, name_2, name_3 = (get_basename(path) for path in paths)
    mime_1, mime_2, mime_3 = (get_mimetype(path) for path in paths)
    file_1, file_2, file_3 = (read_image(path) for path in paths)
    files = {
        "rich-media-image_one": (name_1, file_1, mime_1),
        "rich-media-image_two": (name_2, file_2, mime_2),
        "rich-media-image_three": (name_3, file_3, mime_3),
    }

    return data, files