Python flask 模块，send_from_directory() 实例源码

def serve_autogen_hook(path):
    pid = request.args.get("pid", None)
    static = request.args.get("static", "false") == "true"

    tid = api.user.get_team()["tid"]

    if pid not in api.problem.get_unlocked_pids(tid):
        return WebError("You have not unlocked this problem!")

    instance_number = api.autogen.get_instance_number(pid, tid)

    if static:
        instance_path = api.autogen.get_static_instance_path(pid, public=True)
    else:
        instance_path = api.autogen.get_instance_path(pid, instance_number, public=True)

    mime = guess_mimetype(path)
    if mime == 'text/html':
        return send_from_directory(instance_path, path, mimetype=None, as_attachment=False, attachment_filename=None)
    else:
        return send_from_directory(instance_path, path, mimetype=mime)

def download(fname):
    """Download SWEET-Cat table in different formats and clean afterwards"""
    if fname.startswith('sweet-cat'):
        print(fname)
        fmt = fname.split('.')[-1]
        if fmt in ['csv', 'hdf']:
            table_convert(fmt=fmt)

            @after_this_request
            def remove_file(response):
                try:
                    os.remove('data/{}'.format(fname))
                except OSError:  # pragma: no cover
                    pass  # pragma: no cover
                return response
        return send_from_directory('data', fname)
    else:
        return send_from_directory('data', fname)

def _session_history():
    try:
        data = request.args
        sid = data.get('session')
        sess = session_manager.get_session(sid)
        fname = sess.dump_file
        if fname is not None and os.path.isfile(fname):
            return send_from_directory(
                os.path.dirname(fname),
                os.path.basename(fname),
                mimetype='text/plain')
        else:
            return '', 404
    except Exception as ex:
        logger.error("Internal error {}".format(ex))
        return '', 500

def sitemap():    
    conn,curr = sphinx_conn()
    querysql='SELECT info_hash,create_time FROM film order by create_time desc limit 100'
    curr.execute(querysql)
    rows=curr.fetchall()
    sphinx_close(curr,conn)
    sitemaplist=[]
    for row in rows:
        info_hash = row['info_hash']
        mtime = datetime.datetime.fromtimestamp(int(row['create_time'])).strftime('%Y-%m-%d')
        url = domain+'hash/{}.html'.format(info_hash)
        url_xml = '<url><loc>{}</loc><lastmod>{}</lastmod><changefreq>daily</changefreq><priority>0.8</priority></url>'.format(url, mtime)
        sitemaplist.append(url_xml)
    xml_content = '<?xml version="1.0" encoding="UTF-8"?><urlset>{}</urlset>'.format("".join(x for x in sitemaplist))
    with open('static/sitemap.xml', 'wb') as f:
        f.write(xml_content)
        f.close()
    return send_from_directory(app.static_folder, request.path[1:])

def _serve_webui(self, file_name='index.html'):  # pylint: disable=redefined-builtin
        try:
            assert file_name
            web3 = self.flask_app.config.get('WEB3_ENDPOINT')
            if web3 and 'config.' in file_name and file_name.endswith('.json'):
                host = request.headers.get('Host')
                if any(h in web3 for h in ('localhost', '127.0.0.1')) and host:
                    _, _port = split_endpoint(web3)
                    _host, _ = split_endpoint(host)
                    web3 = 'http://{}:{}'.format(_host, _port)
                response = jsonify({'raiden': self._api_prefix, 'web3': web3})
            else:
                response = send_from_directory(self.flask_app.config['WEBUI_PATH'], file_name)
        except (NotFound, AssertionError):
            response = send_from_directory(self.flask_app.config['WEBUI_PATH'], 'index.html')
        return response

def downloadFiles(report_name, script):
  """ Download a specific file in a report project """
  #TODO add a check on the class variable DOWNLOAD to check if the module is downloadable (by default it is the case)
  requestParams = getHttpParams(request)
  if '.' not in script:
    # We assume it is a python script
    script = "%s.py" % script
  if report_name.startswith("_"):
    pathDirectory = os.path.join(current_app.config['ROOT_PATH'], config.ARES_FOLDER, 'reports')
  else:
    pathDirectory = config.ARES_USERS_LOCATION
  if '&' in script:
    splitScriptPath = script.strip("\\").split("&")
    userDirectory = os.path.join(pathDirectory, report_name, *splitScriptPath[:-1])
  else:
    splitScriptPath = [script]
    userDirectory = os.path.join(pathDirectory, report_name)
  return send_from_directory(userDirectory, splitScriptPath[-1], as_attachment=True)

def backup(username:str, timeframe:str, backup_date:str):
    """
    Route: /backup/username/timeframe/backup_date
        This route returns the requested backup.

    :param username the server username of the user needing their backup.
    :param timeframe the timeframe of the requested backup. Can be either
        "weekly", or "monthly".
    :param backup_date the backup-date of the requested backup. Must be in the
        form YYYY-MM-DD.
    """
    if flask.request.method != "GET":
        return flask.abort(400)

    # make sure the arguments are valid
    if not re.match(r"^[a-z]+$", username) or \
            not re.match(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}", backup_date) or \
            timeframe not in ["weekly", "monthly"]:
        app.logger.debug("backups(%s, %s, %s): invalid arguments"%(
                username, timeframe, backup_date))
        return flask.abort(400)

    backups_base_dir = os.path.join(b.BACKUPS_DIR, username, timeframe)
    return flask.send_from_directory(backups_base_dir, backup_date+".tgz")

def returnTracks(data_id, ext):

    data_id = data_id + '.' + ext

    mimetype =  'application/zip, application/octet-stream' if ext=='zip' else 'audio/mpeg'

    print 'MIME TYPE ========='
    print mimetype

    print 'Globals.DOWNLOAD_PATH => ' + Globals.DOWNLOAD_PATH

    return send_from_directory(Globals.DOWNLOAD_PATH,
                                data_id,
                                as_attachment=True,
                                attachment_filename=data_id,
                                mimetype=mimetype
                            )

def configure_static_route(app):
    # Note (dmsimard)
    # /static/ is provided from in-tree bundled files and libraries.
    # /static/packaged/ is routed to serve packaged (i.e, XStatic) libraries.
    #
    # The reason why this isn't defined as a proper view by itself is due to
    # a limitation in flask-frozen. Blueprint'd views methods are like so:
    # "<view>.<method>. The URL generator of flask-frozen is a method decorator
    # that expects the method name as the function and, obviously, you can't
    # really have dots in functions.
    # By having the route configured at the root of the application, there's no
    # dots and we can decorate "serve_static_packaged" instead of, say,
    # "static.serve_packaged".

    @app.route('/static/packaged/<module>/<path:filename>')
    def serve_static_packaged(module, filename):
        xstatic = current_app.config['XSTATIC']

        if module in xstatic:
            return send_from_directory(xstatic[module], filename)
        else:
            abort(404)

def serve_logs(args):
    print("Starting flask")
    import flask
    flask_app = flask.Flask(__name__)

    @flask_app.route('/log/<path:filename>')
    def serve_logs(filename):  # noqa
        log = os.path.expanduser(conf.get('core', 'BASE_LOG_FOLDER'))
        return flask.send_from_directory(
            log,
            filename,
            mimetype="application/json",
            as_attachment=False)

    WORKER_LOG_SERVER_PORT = \
        int(conf.get('celery', 'WORKER_LOG_SERVER_PORT'))
    flask_app.run(
        host='0.0.0.0', port=WORKER_LOG_SERVER_PORT)

def apidocs():
    url = urlparse(request.url)
    if ":" in url.netloc:
        host, port = url.netloc.split(":")
    else:
        host = url.netloc
        port = "80"
    base_path = url.path.replace('/apidocs','') if url.path != "/apidocs" else "/"
    schemes = [url.scheme]
    other_scheme = "https" if url.scheme is "http" else "http"
    try:
        if request.get(other_scheme+"://"+url.netloc+url.path.replace('/apidocs','')+"/scheme").status_code is 200:
            schemes += [other_scheme]
    except:
        pass
    r = make_response(swagger.json(schemes, host, port, base_path))
    r.mimetype = 'application/json'
    return r
    # return send_from_directory("www","swagger.json")

def graphs(date):

    if request.method == 'POST':
        if date == "today":
            date = datetime.datetime.now().strftime("%Y-%m-%d")
        return send_from_directory(rootdir + "/logs",date+'.csv',as_attachment=True,attachment_filename=date+'.csv')

    if date != "today":
        if logger.logExist(date):
            dates,f1,f2,f3 = logger.getLogDate(date)
        else:
            return redirect("/graphs/today")
    else:
        dates,f1,f2,f3 = logger.getLogDate("today")

    return render_template('graphs.html',date=date,dates=dates,f1=f1,f2=f2,f3=f3)

def downloadPic(source,id,fileName):
    file, ext = os.path.splitext(fileName)
    result =None

    article = Article.query.get(id)
    article.download_num = article.download_num+1
    view = ArticleDownload()
    view.article_id=id
    view.ip = request.remote_addr
    if current_user.is_authenticated:
        view.user_id = current_user.get_id()

    db.session.add(view)
    db.session.commit()
    result = send_from_directory('pics/'+source, file+ext)
    return result

def send_static_file(self, filename):
        """
        Send static files from the static folder in the current selected theme prior to the global static folder.

        :param filename: static filename
        :return: response object
        """
        if self.config['MODE'] == 'api-only':
            abort(404)  # if 'api-only' mode is set, we should not send static files

        theme_static_folder = getattr(self, 'theme_static_folder', None)
        if theme_static_folder:
            try:
                return send_from_directory(theme_static_folder, filename)
            except NotFound:
                pass
        return super(CustomFlask, self).send_static_file(filename)

def get(self, instance_id):
        if not self.is_exist(instance_id):
            abort(404)

        if not self.is_allowed(instance_id):
            abort(403)

        folder_path = thumbnail_utils.get_preview_folder_name(
            "originals",
            instance_id
        )
        file_name = "%s.mp4" % instance_id

        return send_from_directory(
            directory=folder_path,
            filename=file_name
        )

def get(self, instance_id):
        if not self.is_exist(instance_id):
            abort(404)

        if not self.is_allowed(instance_id):
            abort(403)

        folder_path = thumbnail_utils.get_preview_folder_name(
            self.subfolder,
            instance_id
        )
        file_name = thumbnail_utils.get_file_name(instance_id)

        # Use legacy folder name if the file cannot be found.
        if not os.path.exists(os.path.join(folder_path, file_name)):
            folder_path = thumbnail_utils.get_folder_name("preview-files")

        return send_from_directory(
            directory=folder_path,
            filename=file_name
        )

def configure_static_route(app):
    # Note (dmsimard)
    # /static/ is provided from in-tree bundled files and libraries.
    # /static/packaged/ is routed to serve packaged (i.e, XStatic) libraries.
    #
    # The reason why this isn't defined as a proper view by itself is due to
    # a limitation in flask-frozen. Blueprint'd views methods are like so:
    # "<view>.<method>. The URL generator of flask-frozen is a method decorator
    # that expects the method name as the function and, obviously, you can't
    # really have dots in functions.
    # By having the route configured at the root of the application, there's no
    # dots and we can decorate "serve_static_packaged" instead of, say,
    # "static.serve_packaged".

    @app.route('/static/packaged/<module>/<path:filename>')
    def serve_static_packaged(module, filename):
        xstatic = current_app.config['XSTATIC']

        if module in xstatic:
            return send_from_directory(xstatic[module], filename)
        else:
            abort(404)

def download(filename):
    if "email" not in session:
        return redirect(url_for("homepage"))

    member = Member.query.filter_by(email=session["email"]).first()
    homework = Homework.query.filter_by(filename=filename).first()

    if not homework.is_public:
        if member.points <= 0:
            return redirect(url_for("homepage"))
        member.points -= 1
        db.session.commit()

    current_path = os.path.dirname(os.path.realpath(__file__))
    uploads = os.path.join(current_path, app.config["UPLOAD_FOLDER"])
    return send_from_directory(uploads, filename)

def storage(path):
    return send_from_directory('storage', path)

def data(command):
    def handle_file():
        f_name = request.args.get('file_name')
        path = app.config['UPLOAD_FOLDER']
        if not f_name:
            path, f_name = os.path.split(app._cr.csv_file)
        return path, f_name

    def _set_data_file(path, f_name):
        _file = os.path.join(path, f_name)
        app._cr.csv_file = _file
        app._ar.csv_file = _file

    def allowed_file(filename):
        return '.' in filename and filename.rsplit('.', 1)[1] in ALLOWED_EXTENSIONS

    if request.method == 'GET':
        if command == 'set':
            path, f_name = handle_file()
            _set_data_file(path, f_name)
            return 'data file set to %s\n' % f_name
        elif command == 'download':
            path, f_name = handle_file()
            return send_from_directory(path, f_name, as_attachment=True)
        elif command == 'upload':
            return render_template('upload_file.html')
        elif command == 'list':
            files = os.listdir(app.config['UPLOAD_FOLDER'])
            files = [f for f in files if allowed_file(f)]
            return render_template('file_list.html', file_list=files)

    if request.method == 'POST':
        file = request.files['data_file']
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
            return "File Saved!\n"

def RunFileServer(fileServerDir, fileServerPort):
    """
    Run a Flask file server on the given port.

    Explicitly specify instance_path, because Flask's
    auto_find_instance_path can fail when run in a frozen app.
    """
    app = Flask(__name__, instance_path=fileServerDir)

    @app.route('/fileserver-is-ready', methods=['GET'])
    def FileserverIsReady():  # pylint: disable=unused-variable
        """
        Used to test if file server has started.
        """
        return 'Fileserver is ready!'

    @app.route('/<path:filename>', methods=['GET'])
    def ServeFile(filename):  # pylint: disable=unused-variable
        """
        Serves up a file from PYUPDATER_FILESERVER_DIR.
        """
        return send_from_directory(fileServerDir, filename.strip('/'))

    def ShutDownServer():
        """
        Shut down the file server.
        """
        func = request.environ.get('werkzeug.server.shutdown')
        if func is None:
            raise RuntimeError('Not running with the Werkzeug Server')
        func()

    @app.route('/shutdown', methods=['POST'])
    def ShutDown():  # pylint: disable=unused-variable
        """
        Respond to a POSTed request to shut down the file server.
        """
        ShutDownServer()
        return 'Server shutting down...'

    app.run(host=LOCALHOST, port=fileServerPort)

def jupyter_control():
    return redirect('/dashboard/')

# for download basefs.tar.bz
# remove, not the function of docklet
# should download it from a http server
#@app.route('/download/basefs', methods=['GET'])
#def download():
    #fsdir = env.getenv("FS_PREFIX")
    #return send_from_directory(fsdir+'/local', 'basefs.tar.bz', as_attachment=True)

# jupyter auth APIs

def browse_default():
  try:
    return send_from_directory('ui', 'index.html')
  except Exception as e:
    return e.message

def staticx(path):
   return send_from_directory('ui', path)

def arm_file(filename):
    return send_from_directory(app.config['UPLOAD_FOLDER'],
                               filename)

def create_epub():
    toots, _, _ = get_toots()
    accounts, _, _ = get_accounts()
    book = epub.EpubBook()

    # add metadata
    book.set_identifier('mercredi-fiction')
    book.set_title('#MercrediFiction')
    book.set_language('fr')

    for account in accounts:
        book.add_author(account.username)

    chapter = epub.EpubHtml(title='Toutes les histoires',
                            file_name='index.xhtml',
                            lang='fr')
    chapter.content = render_template('epub.html', toots=toots)
    book.add_item(chapter)

    book.toc = (epub.Link('index.xhtml', 'Toutes les histoires', 'histoires'),)

    book.add_item(epub.EpubNcx())
    book.add_item(epub.EpubNav())

    book.spine = ['nav', chapter]

    clean_epub_directory()
    epub_name = str(time()) + '.epub'
    epub_path = path.join(config.EPUB_DIRECTORY, epub_name)
    epub.write_epub(epub_path, book)
    response = send_from_directory(config.EPUB_DIRECTORY, epub_name)
    response.headers['Content-Disposition'] = 'attachment;filename="mercredifiction.epub"'
    return response

def files(path):
    print(path)
    return send_from_directory('static',path)

def index():
    return send_from_directory('static','index.html')

def download():
    filepath = sys.path[0] + '/download/' + request.args.get('path')
    filename = request.args.get('name')
    if os.path.isfile(filepath + filename):
        print(filepath + filename)
        return send_from_directory(filepath, filename, as_attachment=True)
    else:
        abort(404)

def send_js(path):
    return send_from_directory('E:\\imgsave\\', path)

#Web Service

def pdf(project, branch='master'):
    build_path = os.path.abspath(join('repos', project.name, branch.name,
                                      'build/latex'))
    build_latex(project.name, branch.name)
    command = '(cd ' + build_path + '; pdflatex -interaction nonstopmode linux.tex > /tmp/222 || true)'
    os.system(command)
    return flask.send_from_directory(build_path, 'linux.pdf')

def tex(project, branch='master'):
    build_path = os.path.abspath(join('repos', project.name, branch.name,
                                      'build/latex'))
    build_latex(project.name, branch.name)
    return flask.send_from_directory(build_path, 'linux.tex')

def source(project, branch, filename):
    source_path = os.path.abspath(join('repos', project.name, branch.name, 'source'))
    return flask.send_from_directory(source_path, filename)

def get_tikz(project, branch, action, filename):
    images_path = join('repos', project.name, branch.name,
                       'build/html/_images')
    response = send_from_directory(os.path.abspath(images_path), filename)
    if '/_images/tikz-' in request.url:
        response.cache_control.max_age = 5256000
    return response

def resources(project, filename):
    return send_from_directory(
        os.path.abspath(join(project.get_folder(), '_resources/original')),
        filename)

def send_js(path):
    return send_from_directory('web_content/static', path)

def results(filename):
    if not os.path.isfile(os.path.join('results', 'finished_'+filename[:filename.rindex('.')]+'.txt')):
        abort(404)
        return
    return send_from_directory('results',filename)

def service_worker():
    return send_from_directory('static/', 'service-worker.js')

def uploaded_file(filename):  # pragma: no cover
    """Return uploaded file."""
    return send_from_directory(uploader.upload_folder, filename)

def download_avatar(fname):
    if platform.system() == "Windows":
        path = UPLOAD_AVATAR_FOLDER + "/" + fname   # windows??????????????????????
    else:
        path = os.path.join(UPLOAD_AVATAR_FOLDER, fname)  # ?????
    if os.path.isfile(path):
        return send_from_directory(UPLOAD_AVATAR_FOLDER, fname, as_attachment=True)
    abort(404)

def download_music(fname):
    if platform.system() == "Windows":
        path = UPLOAD_FOLDER + "/" + fname   # windows??????????????????????
    else:
        path = os.path.join(UPLOAD_FOLDER, fname)  # ?????
    if os.path.isfile(path):
        return send_from_directory(UPLOAD_FOLDER, fname, as_attachment=True)
    abort(404)

def get_image(img_name):
    if platform.system() == "Windows":
        path = UPLOAD_IMAGE_FOLDER + "/" + img_name   # windows??????????????????????
    else:
        path = os.path.join(UPLOAD_IMAGE_FOLDER, img_name)  # ?????
    if os.path.isfile(path):
        return send_from_directory(UPLOAD_IMAGE_FOLDER, img_name, as_attachment=True)
    abort(404)

def uploaded_file(filename):
    return send_from_directory(FULL_UPLOAD_PATH, filename)

def send_static(path):
    return send_from_directory('static', path)

def get_visualization(filename):
    return send_from_directory(VIZ_FOLDER, filename)

def get_mbtiles(id):
    return send_from_directory(
        IMAGERY_PATH,
        os.path.join(id, 'index.mbtiles'),
        as_attachment=True,
        attachment_filename='{}.mbtiles'.format(id),
        conditional=True
    )


# TODO allow bounding boxes + zoom ranges to be provided

def api_pheno(phenocode):
    return send_from_directory(common_filepaths['manhattan'](''), phenocode)

def api_pheno_qq(phenocode):
    return send_from_directory(common_filepaths['qq'](''), phenocode)

def web_remote_file():
    remote_path = request.args.get('path')
    if not remote_path:
        return "No remote path given."
    local_path, name = get_remote_file(remote_path)
    if not local_path:
        return "Remote file '{}' not found.".format(remote_path)
    res = send_from_directory(local_path, name, as_attachment=True)
    shutil.rmtree(local_path)
    return res

# ########################################################################### #
#                                API Part                                     #
# ########################################################################### #

def template_files(filename):
    template_path = os.path.join(current_app.root_path,
                                 current_app.template_folder)
    return send_from_directory(template_path, filename)