前言之前我整理收藏的图片时，都是简单粗暴地直接全部重命名为hash值，然后配合mongodb管理网络链接（参见python+mongodb+php个人管理图片方案），后来也加上了机器学习给不少图片打上了标签（参见用DeepDanbooru为二次元图片打上标签并存入数据库中），并且搞出了一个随机图API（参见用fastapi搭建随机图api）。但是最近参观了几个随机图API，发现有的返回值更加丰富，不仅有图床链接，而且还有作者ID、标题等。突然觉得有必要也整一个，要不然和转载别人文章不标明出处的行为有什么区别。所以就有了这一篇文章。首先介绍一下思路，人工找图片来源的方法一般是借助搜索引擎（比如谷歌识图、百度识图），还有些专门针对二次元图片搜索的网站（比如ascii2d、saucenao）。那么只要我借助python批量上传检索，不就可以自动化找到来源了吗？图片压缩之所以要进行图片压缩，是为了减少图片上传的耗时，也避免给搜索引擎带来压力导致ip被封。我一开始想的就是直接用python库压缩，但是找了一下，貌似没有比较满意的解决方案。不是压缩效果不理想（Pillow模块），就是需要上传到云端压缩（TinyPNG），或者是压缩时间较长（squoosh）。所以专业的事情交给专业的软件去做，直接请出大杀器Caesium（大家可以网上找找破解版），它压缩速度快而且效果明显（大小减少很多且清晰度没有明显变化）。7000张图片，大概等了十几分钟就压缩完毕，原本10个G的文件夹一下子就变成了2个G。这里需要注意的是，压缩版的图片只是为了搜索，所以质量稍微损失一点是没关系的，一定注意不要覆盖掉原始文件夹，最好统一输出到另外一个文件夹。图片搜索引擎既然我需要找的都是二次元图片，那么ascii2d、saucenao显然要比谷歌百度要好，不仅是因为它们更精确，而且还会输出原始链接、标题、作者，其中saucenao甚至还会输出p站id。不用考虑，就它们了。接下来的难题就是如何使用python上传了，我对爬虫这块不太熟，所以在网上找了一下看有没有别人写好的代码。没想到直接搜到了一个完整的轮子，PicImageSearch，它封装了几个搜索引擎，提供了统一风格的API。安装使用也比较简单，参见说明文档。安装：pip install PicImageSearch -i https://pypi.tuna.tsinghua.edu.cn/simple看了说明文档我才知道原来saucenao是有API的，不过免费账户有限制，30秒最多搜索5次，24小时最多搜索200次，免费是足够了，但对我7000+的图片来说，有点不够看。原本想的是几个搜索引擎轮番调用，然后多申请几个saucenao账号。但是想了想这又加大了工作量，而且saucenao确实也好用（上传速度快，搜索结果精确，而且返回值更多），所以干脆花钱买省心吧。直接在这里升级，遗憾的是它只支持信用卡。我尝试给管理员发邮件提出paypal支付，回复说是可以的，于是我就有了个VIP账户。看了一下30秒内可以搜索25次，每天可以搜索5000次，完全满足需求。数据保存数据保存方案就选择sqlite了，反正都是结构化的数据。不过这里还需要再配合数据写一个判断条件，当前搜索的图片若已存在于数据库中，就跳过，防止重复搜索。另外说一下我的图片文件结构。项目根目录 ----压缩后 ----搜索完毕 ----搜索失败 ----缩略图这里的缩略图也是从saucenao下的，这样方便我们人工核对一遍。代码from pathlib import Path from PicImageSearch import SauceNAO import sqlite3 import shutil import time program_dir = Path(r'D:\素材') pic_dir = program_dir / '压缩后' # 待搜索的图片文件夹 suo_dir = program_dir / '缩略图' # 搜索完毕后缩略图存放地 SauceNAO_api = '这里填写你的API' pic_path_list = pic_dir.iterdir() db_path = program_dir / 'pic.db' con = sqlite3.connect(db_path) cur = con.cursor() # 新建表acg，若已有该表则忽略， cur.execute("CREATE TABLE IF NOT EXISTS acg(hash TEXT, title TEXT, author TEXT, url TEXT,pixiv_id INT,member_id INT)") exists_list = cur.execute("SELECT hash FROM acg").fetchall() exists_list = [x[0] for x in exists_list] # 已存在的hash表 def insert_db(res): cur.execute("INSERT INTO acg(hash,title,author,url,pixiv_id,member_id) VALUES(?,?,?,?,?,?)",res) con.commit() def move_pic(pic_path, target_dir): target_path = pic_path.parent.parent / target_dir / pic_path.name shutil.move(pic_path, target_path) def search_img(pic_path, source= 'saucenao'): flag = True pic_hash = pic_path.stem if pic_hash in exists_list: print(pic_hash,'已存在，跳过') return suo_path = suo_dir / pic_path.name # 缩略图 if source == 'saucenao': # 每30秒25个，每天5000个（普通账户5/200） saucenao = SauceNAO(api_key=SauceNAO_api,numres = 1,minsim = 80,output_type = 2) res = saucenao.search(str(pic_path)) if res.results_requested > 0: title = res.raw[0].title author = res.raw[0].author url = res.raw[0].url pixiv_id = res.raw[0].pixiv_id member_id = res.raw[0].member_id res.raw[0].download_thumbnail(filename=suo_path) result = (pic_hash, title, author, url, pixiv_id, member_id) insert_db(result) print('{}搜索成功，结果为:'.format(pic_hash), result) else: print('{}搜索失败'.format(source)) flag = False print('30秒内剩余次数:{}，24小时内剩余次数:{}'.format(res.short_remaining, res.long_remaining)) if res.short_remaining == 0: print('一小时内无法使用saucenao') if res.long_remaining == 0: print('二十四小时内无法使用saucenao') if flag: move_pic(pic_path, '搜索完毕') else: move_pic(pic_path, '搜索失败') for pic_path in pic_path_list: try: search_img(pic_path, source= 'saucenao') except: time.sleep(10)效果： 改善计划打算五一的时候写一个GUI界面，左边展示原图（压缩后），右边展示缩略图，确认搜索结果是正确的点个按钮就到下一页，结果不正确的图片就点个按钮，移到另外文件夹内，最后人工搜一遍。另外很多推特图title是时间，作者没有。可以写个爬虫从推特上找到作者，title就命名为当时推文的内容好了。
前言最近遇到了个需求，需要接收数据并且可视化展示出来，数据格式为DataFame。这个用Dash很容易实现，但实时更新就难搞了。Dash官方有定时刷新的组件，读取倒是不难，难点在于如何推送数据。经过这两天摸索，我找出了三个解决方案。存储到本地磁盘Dash官方有dash_core_components.Interval组件，这是一个计时器，可以设定时间间隔循环刷新页面。那么很自然的想法，我把生成的数据存到本地，然后网页不断刷新读取就行了。根据前一篇文章的经验，我们知道feather格式最适合小规模的dataframe的写入与读取，有了这一点，很容易就可以写出代码。import dash import dash_core_components as dcc import dash_html_components as html import dash_table # 渲染表格的插件 from dash.dependencies import Input, Output external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div( html.Div([ html.H4('实时更新数据'), html.Div(id='live-update-text'), dash_table.DataTable(id='live-update-table'), dcc.Interval( id='interval', interval=5*1000, # 5秒钟间隔 n_intervals=0 ) ]) ) @app.callback([Output("live-update-table", "data")], [Input("interval", "n_intervals")]) def make_table(n): df = pd.read_feather('log.feather') return df.to_dict('records')当然，还需要另外开一个进程用于存储df到feather。测试了一段时间，发现feather读写有bug：当写入的同时读取，就会报错，从而中断计时器的运行，必须手动刷新网页才能重新开始更新。这就不行了，于是我的同事想出了另外一种解决方案：通过ZMQ推送到内存中通过ZMQ推送到内存中首先需要开一个端口用于发送数据import zmq import time import random import numpy as np import pandas as pd context = zmq.Context() socket = context.socket(zmq.PUB) socket.bind("tcp://*:5411") while True: df = pd.DataFrame(data=np.random.normal(0, 1, [2, 3])) socket.send_pyobj(df) time.sleep(1)然后上述代码中的make_table就可以直接读取内存中的数据，这样速度更快，而且不会产生读写冲突bug。# 其他代码省略 def create_zmq_socket(zmq_port="5411"): """连接ZMQ服务""" context = zmq.Context() zmq_socket = context.socket(zmq.SUB) zmq_socket.connect("tcp://localhost:%s" % zmq_port) zmq_socket.setsockopt_string(zmq.SUBSCRIBE,'') # 消息过滤 return zmq_socket def recv_zmq(listen_port="5411"): """从ZMQ中接受数据""" with create_zmq_socket(listen_port) as socket: msg = socket.recv_pyobj() return msg @app.callback([Output("live-update-table", "data")], [Input("interval", "n_intervals")]) def make_table(n): df = recv_zmq() return df.to_dict('records')这个方案其实比较完美了，而且也符合我对前后端分离的粗浅认知。但还是有些不足，最重要的一点就是，诸如PYQT这种库，都有相应的事件回调函数，只有有数据推送时才会更新，不需要实时刷新监听。我们这种方案，时间间隔太短可能会消耗更多资源，时间间隔长的话对低延时的业务又非常不利。如果能实现类似于PYQT这种回调触发更新的逻辑就好了。通过调用API更新数据由于计算机基础和网络通信等知识的欠缺，我一开始就走向了死胡同，找了很多个方法，都是基于“数据推送过来，触发Dash的回调”进行的。我甚至在考虑要不要把数据推送也写到网页代码里面，然后用while True循环。很明显这样是不行的，While循环会阻塞Dash的线程，结果就是网页根本启动不了。查了很多资料，最后发现了dash_devices库，它对Dash进行了扩展，其中一个新功能就是可以直接在代码内部改变组件值，而无需通过函数回调。看到这个功能，我突然灵光一现，是不是可以在网页代码里写一个API，数据推送模块一旦要更新数据，就调用一下这个API，然后改变页面上某个隐藏控件的属性，并链式回调触发make_table函数，提醒它从ZQM中获取数据。（当然理论上Post直接更新也可以，但是我对Post传数据不太了解，所以目前是通过这种方式间接实现的）因为Dash是基于Flask实现的，所以我们可以很容易绑定一个路由。完整代码如下：import dash_devices from dash_devices.dependencies import Input, Output, State import dash_core_components as dcc import dash_html_components as html import dash_table import zmq import time import random import numpy as np import pandas as pd external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash_devices.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div( html.Div([ html.H4('实时更新数据'), html.Div(id='live-update-text'), dash_table.DataTable(id='live-update-table'), html.Div(id='flag',hidden=True), ]) ) @app.callback([Output("live-update-table", "data")], Input("flag", "children")) def make_table(n): df = recv_zmq() return df.to_dict('records') def create_zmq_socket(zmq_port="5411"): """连接ZMQ服务""" context = zmq.Context() zmq_socket = context.socket(zmq.SUB) zmq_socket.connect("tcp://localhost:%s" % zmq_port) zmq_socket.setsockopt_string(zmq.SUBSCRIBE,'') # 消息过滤 return zmq_socket def recv_zmq(listen_port="5411"): """ 接受数据 """ with create_zmq_socket(listen_port) as socket: msg = socket.recv_pyobj() return msg def get_count(): global count count += 1 if count >= 10: count = 0 return count @app.server.route('/update_data',methods = ['get','post']) def update_data(): app.push_mods({'flag': {'children': str(get_count())}}) return '请求成功'前半部分代码和上两章差不多，只不过用dash_devices替换掉了dash，核心在于最后的绑定路由。这样的话，在数据推送模块，每次推送新数据后，都可以get一下API，提醒网页更新数据。import requests requests.get('127.0.0.1:5000/update_data')
前言最近在写的策略框架需要对数据进行频繁读取，传统的sqlite肯定不考虑，因为数据量比较大，读取速度太慢，所以考虑使用其他格式存储数据。常见的df存储格式有csv，hdf5，feather，Parquet，本文将对这几种格式进行测试，看看哪一种格式最适合我。测试代码参考了以下代码：https://github.com/stefan-jansen/machine-learning-for-trading/blob/master/02_market_and_fundamental_data/05_storage_benchmark/storage_benchmark.ipynbfrom pathlib import Path import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import random import string results = {} def generate_test_data(nrows=1000, numerical_cols=200, text_cols=0, text_length=10): s = "".join([random.choice(string.ascii_letters) for _ in range(text_length)]) data = pd.concat([pd.DataFrame(np.random.random(size=(nrows, numerical_cols))), pd.DataFrame(np.full(shape=(nrows, text_cols), fill_value=s))], axis=1, ignore_index=True) data.columns = [str(i) for i in data.columns] return data data_type = 'Numeric' df = generate_test_data(numerical_cols=200, text_cols=10) # 数据生成 df.info() parquet_file = Path('test.parquet') # 以parquet为例 df.to_parquet(parquet_file) size = parquet_file.stat().st_size %%timeit -o df = pd.read_parquet(parquet_file) # 读取 read = _ parquet_file.unlink() %%timeit -o df.to_parquet(parquet_file) # 写入 parquet_file.unlink() write = _ results['Parquet'] = {'read': np.mean(read.all_runs), 'write': np.mean(write.all_runs), 'size': size} results['Parquet'] # 测试结果保存pandas读取写入这些格式的代码：df = pd.read_parquet(parquet_file) # parquet读取 df.to_parquet(parquet_file) # parquet写入 with pd.HDFStore(test_store) as store: store.get(key) # HDF Fixed、HDF Table、HDF Select读取 with pd.HDFStore(test_store) as store: store.put(key, df) # HDF Fixed写入 store.append('file', df, format='t') # HDF Table写入 store.append('file', df, format='t', data_columns=['company', 'form']) # HDF Select写入 pd.read_csv(test_csv) # csv读取 df.to_csv(test_csv) # csv写入 pd.read_feather(test_feather) # feather读取 df.to_feather(test_feather) # feather写入测试结果共测试了三个不同大小的数据集，其中表头为数据集占用内存大小，因为电脑性能原因，我没办法测试更大的数据了（实际上HDF Table和Select就因为内存爆了所以没有测试结果）。测试结果如下。 1.6+MB152.6+ MB1.5+ GBReadParquet22.5 ms697 ms5.14 s HDF Fixed9.9 ms3.21 s81 s HDF Table16.1 ms7.41 s HDF Select15.6 ms7.36 s csv47.5 ms3.92 s41.9 s feather3.79 ms390 ms4.14 sWriteParquet34.4 ms3.71 s33.7 s HDF Fixed11.9 ms3.23 s51 s HDF Table17.9 ms8.49 s HDF Select17.8 ms8.49 s csv301 ms18.3 s191 s feather13 ms1.98 s17.2 sSizeParquet1.88MB94.12MB966.96MB HDF Fixed2.67MB201.41MB2004.86MB HDF Table1.76MB172MB HDF Select1.76MB172MB csv3.79MB288.74MB2887.42MB feather1.67MB209.95MB2098.23MB可以看到，三种不同大小的数据集中，feather读写速度都一枝独秀，大小占用中规中矩。Parquet在小数据集上表现较差，但随着数据量的增加，其读写速度相比与其他格式就有了很大优势，在大数据集上，Parquet的读取速度甚至能和feather一较高下，可以想象数据量突破2G后，Parquet的读取速度可能就是最快的了。但是在写方面，Parquet一直没有表现出超越feather的势头。Parquet另外一个优势就是压缩率高，占用空间相比于其他格式一直是比较小的。hdf比较尴尬，在小数据集上打不过feather，在大数据集上的读取速度甚至比不上csv。至于csv就不用说了，各方面拉跨。值得注意的是，feather虽然快，但是它不支持dataframe的index，也就是说你传入带index的df它会报错……看了测试结果，相信你心中已经有了各个格式的应用场景。尴尬的是，我的单个数据集刚好只有40M，这个场景下用Parquet感觉没啥必要……
【python】实现一个简易的文本相似度检测程序
前言我手头存了不少小说txt文档，一直都想整理一番，但是手工整理太累了所以也没动手。这半年来感觉我的python水平有了很大提升，于是打算写个程序帮我整理。首先是去重，因为txt文档来源于不同地方，标题作者名什么的有可能不一致，内容上排版稍微差一点hash值也就变了，所以无论是靠文件名还是hash值检测重复文档都不可靠，干脆直接检测文本的相似度吧。网上搜了一下别人写的代码，整合了一下实现了一个最简单的文本相似度检测。因为之前没接触过自然语言学习，所以本文的理解和代码可能有误，欢迎评论区指正。github原理可以参考这两篇文章：https://blog.csdn.net/qq_43350003/article/details/105392702https://www.cnblogs.com/airnew/p/9563703.html首先介绍一下什么是词袋模型。例句:Jane wants to go to Shenzhen.Bob wants to go to Shanghai.将所有词语装进一个袋子里，不考虑其词法和语序的问题 ，即每个词语都是独立的。例如上面 2 个例句，就可以构成一个词袋，袋子里包括 Jane、wants、to、go、Shenzhen、Bob、Shanghai。假设建立一个数组（或词典）用于映射匹配[Jane, wants, to, go, Shenzhen, Bob, Shanghai]那么上面两个例句就可以用以下两个向量表示，对应的下标与映射数组的下标相匹配，其值为该词语出现的次数[1,1,2,1,1,0,0][0,1,2,1,0,1,1]这两个词频向量就是词袋模型。我们知道，一篇文章除了词频外，语序也是很重要的，词向量模型就是考虑词语位置关系的一种模型。因为这次任务并不是专业的查抄袭，所以词袋模型就足够了。有了词频向量后，还需要一种指标来衡量两端文本间的相似度。本文采用余弦相似度算法，所谓余弦相似度算法，就是用一个向量空间中两个向量夹角间的余弦值作为衡量两个个体之间差异的大小，余弦值接近 1，夹角趋于 0，表明两个向量越相似，余弦值接近于 0，夹角趋于 90 度，表明两个向量越不相似。公式如下：$$ \cos (\theta)=\frac{\sum_{i=1}^{n}\left(x_{i} \times y_{i}\right)}{\sqrt{\sum_{i=1}^{n}\left(x_{i}\right)^{2}} \times \sqrt{\sum_{i=1}^{n}\left(y_{i}\right)^{2}}} $$代码主要参考的就是这篇文章：https://www.cnblogs.com/airnew/p/9563703.html我主要是将其numpy化，加快了运行速度。另外，考虑到txt文档的编码方式可能不一样，所以写了一个函数判断所打开文档的编码。代码实现思路和上一节类似，首先用jieba分词将文档进行分词，然后去掉标点符号。为了防止介词状语等影响相似度的准确性，所以分词后要将这类词语去掉。本文使用了百度停用词，存放在./stopwords文件夹下，里面还有中文停用词表、哈工大停用词表、四川大学机器智能实验室停用词库，数据来源于https://github.com/goto456/stopwords，可自行修改。接着就是生成两个文档的one-hot向量，计算出余弦相似度。代码如下：import jieba import re import os import numpy as np import codecs import chardet class SimilarTest(object): def __init__(self, a_path, b_path): s1_code = self.get_code(a_path) s2_code = self.get_code(b_path) chars_pattern = re.compile( r'[\s·’!"\#$%&\'\(\)＃！（）*\+,-./:;<=>\?\@，：?￥★＊、…．＞【】［］《》？“”‘’\[\]\^_`{|}~。—]+') s1 = open(a_path, 'r', encoding=s1_code, errors='ignore').read() s2 = open(b_path, 'r', encoding=s2_code, errors='ignore').read() # 去除标点符号 self.s1 = chars_pattern.sub("", s1) self.s2 = chars_pattern.sub("", s2) # 设置停用词 self.stopwords = open('./stopwords/baidu_stopwords.txt', 'r', encoding='utf-8-sig').read().splitlines() def test(self): s1_split = np.array(jieba.lcut(self.s1)) s2_split = np.array(jieba.lcut(self.s2)) s1_cut = s1_split[~np.isin(s1_split, self.stopwords)] s2_cut = s2_split[~np.isin(s2_split, self.stopwords)] word_set = np.union1d(s1_cut, s2_cut) # 用字典保存两篇文章中出现的所有词并编上号 word_dict = dict() for i, word in enumerate(word_set): word_dict[word] = i # 根据词袋模型统计词在每篇文档中出现的次数，形成向量 s1_cut_code = np.zeros(len(word_dict)) s2_cut_code = np.zeros(len(word_dict)) for word in s1_cut: s1_cut_code[word_dict[word]] += 1 for word in s2_cut: s2_cut_code[word_dict[word]] += 1 # 计算余弦相似度 try: cos_arr = np.sum(s1_cut_code * s2_cut_code) sq1 = np.power(s1_cut_code, 2).sum() sq2 = np.power(s2_cut_code, 2).sum() result = np.around(cos_arr / (np.sqrt(sq1) * np.sqrt(sq2)), 3) except ZeroDivisionError: result = 0.0 print("文本相似度为：{}".format(result)) @staticmethod def get_code(path): with open(path, 'rb') as file: data = file.read(200) dicts = chardet.detect(data) return dicts["encoding"] if __name__ == '__main__': a_path = 'atext.txt' b_path = 'btext.txt' textST = SimilarTest(a_path, b_path) textST.test()效果：
前言之前工作的时候，需要写一个日内交易信号监控软件，预计的功能是实时更新数据，出现信号时有提示功能（比如颜色变化和声音提醒）。如果可能的话最好也有可视化功能。我第一时间就想到了plotly同家出品的Dash，它是一个python Web可视化框架库，之前就有兴趣学，但是因为懒所以一直没有入手。需求来了才开始看文档，边看边翻译、实践，花了几天就写出了一个简单的网站。我个人还是挺满意的，所以也打算出一期教程介绍一下Dash。先把之前翻译的文档放上来把，之后有时间就讲一些高级用法。下文均翻译自https://dash.plotly.com/，算是一些比较基础的用法，有些地方可能是机翻的，最好还是去官网看。如果你有html和plotly基础，那入门应该会相当快，没有也没关系，我就不懂html，仿照着官网例子也能写。安装pip install dash布局基本用法首先，使用以下代码创建一个名为 app.py 的文件。import dash import dash_core_components as dcc import dash_html_components as html import plotly.express as px import pandas as pd external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) # assume you have a "long-form" data frame # see https://plotly.com/python/px-arguments/ for more options df = pd.DataFrame({ "Fruit": ["Apples", "Oranges", "Bananas", "Apples", "Oranges", "Bananas"], "Amount": [4, 1, 2, 2, 4, 5], "City": ["SF", "SF", "SF", "Montreal", "Montreal", "Montreal"] }) fig = px.bar(df, x="Fruit", y="Amount", color="City", barmode="group") app.layout = html.Div(children=[ html.H1(children='Hello Dash'), html.Div(children=''' Dash: A web application framework for Python. '''), dcc.Graph( id='example-graph', figure=fig ) ]) if __name__ == '__main__': app.run_server(debug=True)然后运行程序$ python app.py访问 http://127.0.0.1:8050/效果如下： 注意：布局由一个“组件”树组成，比如 html.Div 和 dcc.Graph。对于每个 HTML 标记，dash_html_components 库都有一个组件。html.H1(children='Hello Dash')组件在应用程序中生成一个 Hello Dash HTML 元素。并非所有组件都是纯 HTML。dash_core_components 是可交互式高级组件，通过 React.js 库用 JavaScript、HTML 和 CSS 生成。每个组件都是通过关键字属性来描述的。dash 是声明式语法：主要通过这些属性描述应用程序。children 属性是特殊的。按照惯例，它总是第一个属性，这意味着您可以忽略它：html.H1(children='Hello Dash') 等价于 html.H1('Hello Dash')。它还可以包含字符串、数字、单个组件或组件列表。应用程序中的字体看起来与此处显示的字体略有不同。此应用程序使用自定义 CSS 样式表来修改元素的默认样式。您可以在 https://dash.plotly.com/external-resources 中了解更多信息，但现在您可以使用external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets)获得和示例相同的样式。做出第一项改变dash 包括 热加载 功能，该功能默认激活，用 app.run_server(debug=True) 启动程序后，当你修改代码时，浏览器会自动刷新你的修改。试一试：修改标题 Hello DashDon't like hot-reloading? You can turn this off with app.run_server(dev_tools_hot_reload=False)更多关于 HTML 的知识dash_html_components 库对每个 HTML 标记都有对应的组件，同样对于所有 HTML 参数都有对应的关键字参数。# -*- coding: utf-8 -*- # Run this app with `python app.py` and # visit http://127.0.0.1:8050/ in your web browser. import dash import dash_core_components as dcc import dash_html_components as html import plotly.express as px import pandas as pd external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) colors = { 'background': '#111111', 'text': '#7FDBFF' } # assume you have a "long-form" data frame # see https://plotly.com/python/px-arguments/ for more options df = pd.DataFrame({ "Fruit": ["Apples", "Oranges", "Bananas", "Apples", "Oranges", "Bananas"], "Amount": [4, 1, 2, 2, 4, 5], "City": ["SF", "SF", "SF", "Montreal", "Montreal", "Montreal"] }) fig = px.bar(df, x="Fruit", y="Amount", color="City", barmode="group") fig.update_layout( plot_bgcolor=colors['background'], paper_bgcolor=colors['background'], font_color=colors['text'] ) app.layout = html.Div(style={'backgroundColor': colors['background']}, children=[ html.H1( children='Hello Dash', style={ 'textAlign': 'center', 'color': colors['text'] } ), html.Div(children='Dash: A web application framework for Python.', style={ 'textAlign': 'center', 'color': colors['text'] }), dcc.Graph( id='example-graph-2', figure=fig ) ]) if __name__ == '__main__': app.run_server(debug=True)在这个例子中，我们通过 style 属性修改了 html.Div 和 html.H1 的内联样式。html.H1('Hello Dash', style={'textAlign': 'center', 'color': '#7FDBFF'}) 转换为 HTML 代码即 Hello Dash dash_html_components 和 HTML 属性之间几乎没有重要区别：HTML 中的 style 属性是以分号分隔的字符串。在 Dash 中，你可以只提供字典。style 字典中的键为驼峰命名，所以使用 textAlign 代替 text-alignHTML class 属性在 dash 中为·className·。HTML 标记的子级是通过 children 关键字参数指定的。按照惯例，这总是第一个参数，所以它经常被省略。除此之外，所有可用的 HTML 属性和标记都可以在 Python context 中使用。可复用组件通过用 Python 编写我们的标记，我们可以创建复杂的可重用组件，如表，而无需切换上下文或语言。# Run this app with `python app.py` and # visit http://127.0.0.1:8050/ in your web browser. import dash import dash_core_components as dcc import dash_html_components as html import pandas as pd df = pd.read_csv('https://gist.githubusercontent.com/chriddyp/c78bf172206ce24f77d6363a2d754b59/raw/c353e8ef842413cae56ae3920b8fd78468aa4cb2/usa-agricultural-exports-2011.csv') def generate_table(dataframe, max_rows=10): return html.Table([ html.Thead( html.Tr([html.Th(col) for col in dataframe.columns]) ), html.Tbody([ html.Tr([ html.Td(dataframe.iloc[i][col]) for col in dataframe.columns ]) for i in range(min(len(dataframe), max_rows)) ]) ]) external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div(children=[ html.H4(children='US Agriculture Exports (2011)'), generate_table(df) ]) if __name__ == '__main__': app.run_server(debug=True)更多关于可视化的知识dash_core_components 库包括一个名为 Graph 的组件Graph 使用开源的 plotly.js JavaScript 图形库渲染可交互式数据。plotly.js 支持超过 35 种图表类型，并以矢量质量 SVG 和高性能 WebGL 呈现图表。dash_core_components.Graph 中的 figure 参数和 plotly.py 中的 figure 参数相同。# Run this app with `python app.py` and # visit http://127.0.0.1:8050/ in your web browser. import dash import dash_core_components as dcc import dash_html_components as html import plotly.express as px import pandas as pd external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) df = pd.read_csv('https://gist.githubusercontent.com/chriddyp/5d1ea79569ed194d432e56108a04d188/raw/a9f9e8076b837d541398e999dcbac2b2826a81f8/gdp-life-exp-2007.csv') fig = px.scatter(df, x="gdp per capita", y="life expectancy", size="population", color="continent", hover_name="country", log_x=True, size_max=60) app.layout = html.Div([ dcc.Graph( id='life-exp-vs-gdp', figure=fig ) ]) if __name__ == '__main__': app.run_server(debug=True)效果： Markdown我们也可以直接使用 Markdown 语法来实现部分 HTML 代码# Run this app with `python app.py` and # visit http://127.0.0.1:8050/ in your web browser. import dash import dash_core_components as dcc import dash_html_components as html external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) markdown_text = ''' ### Dash and Markdown Dash apps can be written in Markdown. Dash uses the [CommonMark](http://commonmark.org/) specification of Markdown. Check out their [60 Second Markdown Tutorial](http://commonmark.org/help/) if this is your first introduction to Markdown! ''' app.layout = html.Div([ dcc.Markdown(children=markdown_text) ]) if __name__ == '__main__': app.run_server(debug=True)Core Componentsdash_core_components 包括了大量的高级组件，如下拉菜单、复选框等等，详见 https://dash.plotly.com/dash-core-components# -*- coding: utf-8 -*- # Run this app with `python app.py` and # visit http://127.0.0.1:8050/ in your web browser. import dash import dash_core_components as dcc import dash_html_components as html external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div([ html.Label('Dropdown'), dcc.Dropdown( options=[ {'label': 'New York City', 'value': 'NYC'}, {'label': u'Montréal', 'value': 'MTL'}, {'label': 'San Francisco', 'value': 'SF'} ], value='MTL' ), html.Label('Multi-Select Dropdown'), dcc.Dropdown( options=[ {'label': 'New York City', 'value': 'NYC'}, {'label': u'Montréal', 'value': 'MTL'}, {'label': 'San Francisco', 'value': 'SF'} ], value=['MTL', 'SF'], multi=True ), html.Label('Radio Items'), dcc.RadioItems( options=[ {'label': 'New York City', 'value': 'NYC'}, {'label': u'Montréal', 'value': 'MTL'}, {'label': 'San Francisco', 'value': 'SF'} ], value='MTL' ), html.Label('Checkboxes'), dcc.Checklist( options=[ {'label': 'New York City', 'value': 'NYC'}, {'label': u'Montréal', 'value': 'MTL'}, {'label': 'San Francisco', 'value': 'SF'} ], value=['MTL', 'SF'] ), html.Label('Text Input'), dcc.Input(value='MTL', type='text'), html.Label('Slider'), dcc.Slider( min=0, max=9, marks={i: 'Label {}'.format(i) if i == 1 else str(i) for i in range(1, 6)}, value=5, ), ], style={'columnCount': 2}) if __name__ == '__main__': app.run_server(debug=True)value 是默认值Dash 回调本章介绍如何使用回调函数来制作 Dash 应用程序：当输入组件的属性更改时，Dash 自动调用 python 函数。简单的交互式 Dash 应用import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div([ html.H6("Change the value in the text box to see callbacks in action!"), html.Div(["Input: ", dcc.Input(id='my-input', value='initial value', type='text')]), html.Br(), html.Div(id='my-output'), ]) @app.callback( Output(component_id='my-output', component_property='children'), [Input(component_id='my-input', component_property='value')] ) def update_output_div(input_value): return 'Output: {}'.format(input_value) if __name__ == '__main__': app.run_server(debug=True)让我们拆解这个例子：应用程序接口的“输入”和“输出”被声明为装饰器 @app.callback 的参数通过编写这个 decorator，我们告诉 Dash 只要“输入组件”（文本框）的值发生变化，就调用该函数，以更新页面上“输出”组件（HTML div）的子组件装饰器下的函数名可以任意装饰器中的 id 必须与 dash 中 layout 的 id 相同在 Dash 中，我们应用程序的输入和输出只是一个特定组件的属性。在这个例子中，我们的输入是 ID 为 "my-input "的组件的 "value "属性。我们的输出是 ID 为 "my-output "的组件的 "children "属性。每当一个输入属性发生变化时，回调装饰器封装的函数就会自动被调用。Dash 向函数提供输入属性的新值作为输入参数，Dash 则使用函数返回的任何内容更新输出组件的属性。component_id 和 component_property 关键字是可选的（每个对象只有两个参数）。为了清楚起见，在这个例子中包含了这两个关键字，但为了简洁和可读性，在文档的其余部分将省略这两个关键字。不要混淆 dash.dependencies.Input 对象和 dash_core_components.Input 对象。前者只是在这些回调中使用，后者是一个实际的组件。请注意，我们没有在布局中为 my-output 组件的 children 属性设置一个值。当 Dash 应用程序启动时，它会自动调用所有具有输入组件初始值的回调，以便填充输出组件的初始状态。在此示例中，如果您指定了类似 html.Div(id='my-output', children='Hello world') 这样的内容，当应用程序启动时，它将被覆盖。Dash App Layout With Figure and Slider让我们看看另一个例子，在这个例子中，dcc.Slider 更新了一个 dcc.Graph。import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output import plotly.express as px import pandas as pd df = pd.read_csv('https://cdn.jsdelivr.net/gh/plotly/datasets/gapminderDataFiveYear.csv') external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div([ dcc.Graph(id='graph-with-slider'), dcc.Slider( id='year-slider', min=df['year'].min(), max=df['year'].max(), value=df['year'].min(), marks={str(year): str(year) for year in df['year'].unique()}, step=None ) ]) @app.callback( Output('graph-with-slider', 'figure'), [Input('year-slider', 'value')]) def update_figure(selected_year): filtered_df = df[df.year == selected_year] fig = px.scatter(filtered_df, x="gdpPercap", y="lifeExp", size="pop", color="continent", hover_name="country", log_x=True, size_max=55) fig.update_layout(transition_duration=500) return fig if __name__ == '__main__': app.run_server(debug=True)效果图： 在这个例子中，Slider 的 "value "属性是应用程序的输入，应用程序的输出是 Graph 的 "figure "属性。每当 Slider 的值发生变化时，Dash 就会使用新的值调用回调函数 update_figure。该函数使用该新值过滤数据框，构造一个 figure 对象，并将其返回给 Dash 应用程序。在这个例子中，有一些不错的思路。我们使用 Pandas 库来导入和过滤内存中的数据集。我们在应用开始时加载我们的 df：df = pd.read_csv('...')。这个 df 在应用的全局状态下，可以在回调函数里面读取。将数据加载到内存中是很昂贵的。通过在应用开始时加载查询数据，而不是在回调函数里面加载查询数据，我们可以确保这个操作只在应用服务器启动时完成。当用户访问应用程序或与应用程序交互时，该数据（df）已经在内存中。如果可能的话，昂贵的初始化（比如下载或查询数据）应该在应用的全局范围内完成，而不是在回调函数内完成。回调不会修改原始数据，它只是通过 pandas 过滤器过滤来创建数据框架的副本。这一点很重要：你的回调永远不应该突变其范围外的变量。如果你的回调修改了全局状态，那么一个用户的会话可能会影响下一个用户的会话，当应用部署在多个进程或线程上时，这些修改不会在不同的会话之间共享。我们正在使用 layout.transition 开启过渡，以便让人了解数据集如何随着时间的推移而演变：过渡允许图表从一个状态平滑地更新到下一个状态，就像它是动画一样。其 transition_duration=500 参数表示过渡到下一个图表所需时间。多输入import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output import plotly.express as px import pandas as pd external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) df = pd.read_csv('https://plotly.github.io/datasets/country_indicators.csv') available_indicators = df['Indicator Name'].unique() app.layout = html.Div([ html.Div([ html.Div([ dcc.Dropdown( id='xaxis-column', options=[{'label': i, 'value': i} for i in available_indicators], value='Fertility rate, total (births per woman)' ), dcc.RadioItems( id='xaxis-type', options=[{'label': i, 'value': i} for i in ['Linear', 'Log']], value='Linear', labelStyle={'display': 'inline-block'} ) ], style={'width': '48%', 'display': 'inline-block'}), html.Div([ dcc.Dropdown( id='yaxis-column', options=[{'label': i, 'value': i} for i in available_indicators], value='Life expectancy at birth, total (years)' ), dcc.RadioItems( id='yaxis-type', options=[{'label': i, 'value': i} for i in ['Linear', 'Log']], value='Linear', labelStyle={'display': 'inline-block'} ) ],style={'width': '48%', 'float': 'right', 'display': 'inline-block'}) ]), dcc.Graph(id='indicator-graphic'), dcc.Slider( id='year--slider', min=df['Year'].min(), max=df['Year'].max(), value=df['Year'].max(), marks={str(year): str(year) for year in df['Year'].unique()}, step=None ) ]) @app.callback( Output('indicator-graphic', 'figure'), [Input('xaxis-column', 'value'), Input('yaxis-column', 'value'), Input('xaxis-type', 'value'), Input('yaxis-type', 'value'), Input('year--slider', 'value')]) def update_graph(xaxis_column_name, yaxis_column_name, xaxis_type, yaxis_type, year_value): dff = df[df['Year'] == year_value] fig = px.scatter(x=dff[dff['Indicator Name'] == xaxis_column_name]['Value'], y=dff[dff['Indicator Name'] == yaxis_column_name]['Value'], hover_name=dff[dff['Indicator Name'] == yaxis_column_name]['Country Name']) fig.update_layout(margin={'l': 40, 'b': 40, 't': 10, 'r': 0}, hovermode='closest') fig.update_xaxes(title=xaxis_column_name, type='linear' if xaxis_type == 'Linear' else 'log') fig.update_yaxes(title=yaxis_column_name, type='linear' if yaxis_type == 'Linear' else 'log') return fig if __name__ == '__main__': app.run_server(debug=True)在这个例子中，每当 Dropdown、Slider 或 RadioItems 组件的值属性改变时， update_graph 函数就会被调用。update_graph 函数的输入参数是每个 Input 属性的新值或当前值，按照它们被指定的顺序。即使一次只改变一个单一的输入（用户在给定的时刻只能改变一个 Dropdown 的值），Dash 也会收集所有指定的输入属性的当前状态，并为您将它们传递到您的函数中。你的回调函数总是被保证传递给应用程序的代表状态。让我们扩展我们的例子以包括多个输出。多输出到目前为止，我们写的所有回调都只更新一个输出属性。我们也可以同时更新几个属性：把所有要更新的属性作为一个列表放在装饰器中，然后从回调中返回这么多项目。如果两个输出依赖于同一个计算密集的中间结果，比如一个缓慢的数据库查询，就会表现得特别好。import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div([ dcc.Input( id='num-multi', type='number', value=5 ), html.Table([ html.Tr([html.Td(['x', html.Sup(2)]), html.Td(id='square')]), html.Tr([html.Td(['x', html.Sup(3)]), html.Td(id='cube')]), html.Tr([html.Td([2, html.Sup('x')]), html.Td(id='twos')]), html.Tr([html.Td([3, html.Sup('x')]), html.Td(id='threes')]), html.Tr([html.Td(['x', html.Sup('x')]), html.Td(id='x^x')]), ]), ]) @app.callback( [Output('square', 'children'), Output('cube', 'children'), Output('twos', 'children'), Output('threes', 'children'), Output('x^x', 'children')], [Input('num-multi', 'value')]) def callback_a(x): return x**2, x**3, 2**x, 3**x, x**x if __name__ == '__main__': app.run_server(debug=True)需要提醒的是：即使可以合并输出，也不一定是个好主意。如果输出依赖于一些而不是所有相同的 Inputs，将它们分开可以避免不必要的更新。如果它们有相同的 Inputs，但用这些 Inputs 做独立的计算，保持回调分离可以让它们并行运行。链式回调你也可以将输出和输入链在一起：一个回调函数的输出可以是另一个回调函数的输入。这种模式可以用来创建动态用户界面，其中一个输入组件更新下一个输入组件的可用选项。下面是一个简单的例子。# -*- coding: utf-8 -*- import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) all_options = { 'America': ['New York City', 'San Francisco', 'Cincinnati'], 'Canada': [u'Montréal', 'Toronto', 'Ottawa'] } app.layout = html.Div([ dcc.RadioItems( id='countries-radio', options=[{'label': k, 'value': k} for k in all_options.keys()], value='America' ), html.Hr(), dcc.RadioItems(id='cities-radio'), html.Hr(), # 横线 html.Div(id='display-selected-values') ]) @app.callback( Output('cities-radio', 'options'), [Input('countries-radio', 'value')]) def set_cities_options(selected_country): return [{'label': i, 'value': i} for i in all_options[selected_country]] @app.callback( Output('cities-radio', 'value'), [Input('cities-radio', 'options')]) def set_cities_value(available_options): return available_options[0]['value'] @app.callback( Output('display-selected-values', 'children'), [Input('countries-radio', 'value'), Input('cities-radio', 'value')]) def set_display_children(selected_country, selected_city): return u'{} is a city in {}'.format( selected_city, selected_country, ) if __name__ == '__main__': app.run_server(debug=True)第一个回调根据第一个 RadioItems 组件中的选择值更新第二个 RadioItems 组件中的可用选项。第二个回调在选项属性改变时设置一个初始值：它将其设置为该选项数组中的第一个值。最后一个回调显示每个组件的选定值。如果您更改了 countries RadioItems 组件的值，Dash 将等待直到 cities 组件的值更新后再调用最终回调。这可以防止你的回调被调用时状态不一致，比如 "America "和 "Montréal"。状态在某些情况下，你可能会在你的应用程序中使用一个 "表单 "类型的模式。在这种情况下，你可能想读取输入组件的值，但只有当用户完成了他或她在表单中的所有信息输入时才会读取。直接给输入值附加一个回调可以实现这一点。# -*- coding: utf-8 -*- import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output external_stylesheets = ["https://codepen.io/chriddyp/pen/bWLwgP.css"] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div( [ dcc.Input(id="input-1", type="text", value="Montréal"), dcc.Input(id="input-2", type="text", value="Canada"), html.Div(id="number-output"), ] ) @app.callback( Output("number-output", "children"), [Input("input-1", "value"), Input("input-2", "value")], ) def update_output(input1, input2): return u'Input 1 is "{}" and Input 2 is "{}"'.format(input1, input2) if __name__ == "__main__": app.run_server(debug=True) 在这个例子中，每当 dash.dependencies.Input 描述的任何属性发生变化时，回调函数就会被触发。你可以在上面的输入中输入数据来试试。dash.dependencies.State 允许你在不触发回调的情况下传递额外的值。这里的例子和上面的一样，但是把 dcc.Input 作为 dash.dependencies.State，把按钮作为 dash.dependencies.Input。# -*- coding: utf-8 -*- import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output, State external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div([ dcc.Input(id='input-1-state', type='text', value='Montréal'), dcc.Input(id='input-2-state', type='text', value='Canada'), html.Button(id='submit-button-state', n_clicks=0, children='Submit'), html.Div(id='output-state') ]) @app.callback(Output('output-state', 'children'), [Input('submit-button-state', 'n_clicks')], [State('input-1-state', 'value'), State('input-2-state', 'value')]) def update_output(n_clicks, input1, input2): return u''' The Button has been pressed {} times, Input 1 is "{}", and Input 2 is "{}" '''.format(n_clicks, input1, input2) if __name__ == '__main__': app.run_server(debug=True) 在这个例子中，改变 dcc.Input 框中的文本不会触发回调，但点击按钮会。尽管 dcc.Input 值的当前值没有触发回调函数本身，但它们仍然会被传递到回调中。请注意，我们是通过监听 html.Button 组件的 n_clicks 属性来触发回调的。n_clicks 是一个每次组件被单击时都会递增的属性。dash_html_components 库中的每个组件都有这个属性。可交互式可视化可交互式作图dash_core_components 库中包含一个名为 Graph 的组件。Graph 使用开源的 plotly.js JavaScript 图形库来渲染交互式数据可视化。Plotly.js 支持超过 35 种图表类型，并以矢量质量的 SVG 和高性能的 WebGL 渲染图表。dash_core_components.Graph 组件中的 figure 参数与 Plotly 的开源 Python 图表库 plotly.py 使用的 figure 参数相同。查看 plotly.py 文档和图库以了解更多。Dash 组件由一组属性声明性地描述。所有这些属性都可以通过回调函数更新，但只有一个子集的属性是通过用户交互更新的，比如当你点击 dcc.Dropdown 组件中的一个选项时，该组件的值属性就会改变。dcc.Graph 组件有四个属性可以通过用户交互来改变：hoverData、clickData、selectedData、 relayoutData。当你将鼠标悬停在点上，点击点，或者选择图形中的点的区域时，这些属性就会更新。import json import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output import plotly.express as px import pandas as pd external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) styles = { 'pre': { 'border': 'thin lightgrey solid', 'overflowX': 'scroll' } } df = pd.DataFrame({ "x": [1,2,1,2], "y": [1,2,3,4], "customdata": [1,2,3,4], "fruit": ["apple", "apple", "orange", "orange"] }) fig = px.scatter(df, x="x", y="y", color="fruit", custom_data=["customdata"]) fig.update_layout(clickmode='event+select') fig.update_traces(marker_size=20) app.layout = html.Div([ dcc.Graph( id='basic-interactions', figure=fig ), html.Div(className='row', children=[ html.Div([ dcc.Markdown(""" **Hover Data** Mouse over values in the graph. """), html.Pre(id='hover-data', style=styles['pre']) ], className='three columns'), html.Div([ dcc.Markdown(""" **Click Data** Click on points in the graph. """), html.Pre(id='click-data', style=styles['pre']), ], className='three columns'), html.Div([ dcc.Markdown(""" **Selection Data** Choose the lasso or rectangle tool in the graph's menu bar and then select points in the graph. Note that if `layout.clickmode = 'event+select'`, selection data also accumulates (or un-accumulates) selected data if you hold down the shift button while clicking. """), html.Pre(id='selected-data', style=styles['pre']), ], className='three columns'), html.Div([ dcc.Markdown(""" **Zoom and Relayout Data** Click and drag on the graph to zoom or click on the zoom buttons in the graph's menu bar. Clicking on legend items will also fire this event. """), html.Pre(id='relayout-data', style=styles['pre']), ], className='three columns') ]) ]) @app.callback( Output('hover-data', 'children'), [Input('basic-interactions', 'hoverData')]) def display_hover_data(hoverData): return json.dumps(hoverData, indent=2) @app.callback( Output('click-data', 'children'), [Input('basic-interactions', 'clickData')]) def display_click_data(clickData): return json.dumps(clickData, indent=2) @app.callback( Output('selected-data', 'children'), [Input('basic-interactions', 'selectedData')]) def display_selected_data(selectedData): return json.dumps(selectedData, indent=2) @app.callback( Output('relayout-data', 'children'), [Input('basic-interactions', 'relayoutData')]) def display_relayout_data(relayoutData): return json.dumps(relayoutData, indent=2) if __name__ == '__main__': app.run_server(debug=True)Update Graphs on Hoverimport dash import dash_core_components as dcc import dash_html_components as html import pandas as pd import plotly.express as px external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) df = pd.read_csv('https://cdn.jsdelivr.net/gh/plotly/datasets/country_indicators.csv') available_indicators = df['Indicator Name'].unique() app.layout = html.Div([ html.Div([ html.Div([ dcc.Dropdown( id='crossfilter-xaxis-column', options=[{'label': i, 'value': i} for i in available_indicators], value='Fertility rate, total (births per woman)' ), dcc.RadioItems( id='crossfilter-xaxis-type', options=[{'label': i, 'value': i} for i in ['Linear', 'Log']], value='Linear', labelStyle={'display': 'inline-block'} ) ], style={'width': '49%', 'display': 'inline-block'}), html.Div([ dcc.Dropdown( id='crossfilter-yaxis-column', options=[{'label': i, 'value': i} for i in available_indicators], value='Life expectancy at birth, total (years)' ), dcc.RadioItems( id='crossfilter-yaxis-type', options=[{'label': i, 'value': i} for i in ['Linear', 'Log']], value='Linear', labelStyle={'display': 'inline-block'} ) ], style={'width': '49%', 'float': 'right', 'display': 'inline-block'}) ], style={ 'borderBottom': 'thin lightgrey solid', 'backgroundColor': 'rgb(250, 250, 250)', 'padding': '10px 5px' }), html.Div([ dcc.Graph( id='crossfilter-indicator-scatter', hoverData={'points': [{'customdata': 'Japan'}]} ) ], style={'width': '49%', 'display': 'inline-block', 'padding': '0 20'}), html.Div([ dcc.Graph(id='x-time-series'), dcc.Graph(id='y-time-series'), ], style={'display': 'inline-block', 'width': '49%'}), html.Div(dcc.Slider( id='crossfilter-year--slider', min=df['Year'].min(), max=df['Year'].max(), value=df['Year'].max(), marks={str(year): str(year) for year in df['Year'].unique()}, step=None ), style={'width': '49%', 'padding': '0px 20px 20px 20px'}) ]) @app.callback( dash.dependencies.Output('crossfilter-indicator-scatter', 'figure'), [dash.dependencies.Input('crossfilter-xaxis-column', 'value'), dash.dependencies.Input('crossfilter-yaxis-column', 'value'), dash.dependencies.Input('crossfilter-xaxis-type', 'value'), dash.dependencies.Input('crossfilter-yaxis-type', 'value'), dash.dependencies.Input('crossfilter-year--slider', 'value')]) def update_graph(xaxis_column_name, yaxis_column_name, xaxis_type, yaxis_type, year_value): dff = df[df['Year'] == year_value] fig = px.scatter(x=dff[dff['Indicator Name'] == xaxis_column_name]['Value'], y=dff[dff['Indicator Name'] == yaxis_column_name]['Value'], hover_name=dff[dff['Indicator Name'] == yaxis_column_name]['Country Name'] ) fig.update_traces(customdata=dff[dff['Indicator Name'] == yaxis_column_name]['Country Name']) fig.update_xaxes(title=xaxis_column_name, type='linear' if xaxis_type == 'Linear' else 'log') fig.update_yaxes(title=yaxis_column_name, type='linear' if yaxis_type == 'Linear' else 'log') fig.update_layout(margin={'l': 40, 'b': 40, 't': 10, 'r': 0}, hovermode='closest') return fig def create_time_series(dff, axis_type, title): fig = px.scatter(dff, x='Year', y='Value') fig.update_traces(mode='lines+markers') fig.update_xaxes(showgrid=False) fig.update_yaxes(type='linear' if axis_type == 'Linear' else 'log') fig.add_annotation(x=0, y=0.85, xanchor='left', yanchor='bottom', xref='paper', yref='paper', showarrow=False, align='left', bgcolor='rgba(255, 255, 255, 0.5)', text=title) fig.update_layout(height=225, margin={'l': 20, 'b': 30, 'r': 10, 't': 10}) return fig @app.callback( dash.dependencies.Output('x-time-series', 'figure'), [dash.dependencies.Input('crossfilter-indicator-scatter', 'hoverData'), dash.dependencies.Input('crossfilter-xaxis-column', 'value'), dash.dependencies.Input('crossfilter-xaxis-type', 'value')]) def update_y_timeseries(hoverData, xaxis_column_name, axis_type): country_name = hoverData['points'][0]['customdata'] dff = df[df['Country Name'] == country_name] dff = dff[dff['Indicator Name'] == xaxis_column_name] title = '{} {}'.format(country_name, xaxis_column_name) return create_time_series(dff, axis_type, title) @app.callback( dash.dependencies.Output('y-time-series', 'figure'), [dash.dependencies.Input('crossfilter-indicator-scatter', 'hoverData'), dash.dependencies.Input('crossfilter-yaxis-column', 'value'), dash.dependencies.Input('crossfilter-yaxis-type', 'value')]) def update_x_timeseries(hoverData, yaxis_column_name, axis_type): dff = df[df['Country Name'] == hoverData['points'][0]['customdata']] dff = dff[dff['Indicator Name'] == yaxis_column_name] return create_time_series(dff, axis_type, yaxis_column_name) if __name__ == '__main__': app.run_server(debug=True) 试着将鼠标放在左边散点图中的点上。注意右边的线图是如何根据你悬停的点进行更新的。交叉过滤import dash import dash_core_components as dcc import dash_html_components as html import numpy as np import pandas as pd from dash.dependencies import Input, Output import plotly.express as px external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) # make a sample data frame with 6 columns np.random.seed(0) df = pd.DataFrame({"Col " + str(i+1): np.random.rand(30) for i in range(6)}) app.layout = html.Div([ html.Div( dcc.Graph(id='g1', config={'displayModeBar': False}), className='four columns' ), html.Div( dcc.Graph(id='g2', config={'displayModeBar': False}), className='four columns' ), html.Div( dcc.Graph(id='g3', config={'displayModeBar': False}), className='four columns' ) ], className='row') def get_figure(df, x_col, y_col, selectedpoints, selectedpoints_local): if selectedpoints_local and selectedpoints_local['range']: ranges = selectedpoints_local['range'] selection_bounds = {'x0': ranges['x'][0], 'x1': ranges['x'][1], 'y0': ranges['y'][0], 'y1': ranges['y'][1]} else: selection_bounds = {'x0': np.min(df[x_col]), 'x1': np.max(df[x_col]), 'y0': np.min(df[y_col]), 'y1': np.max(df[y_col])} # set which points are selected with the `selectedpoints` property # and style those points with the `selected` and `unselected` # attribute. see # https://medium.com/@plotlygraphs/notes-from-the-latest-plotly-js-release-b035a5b43e21 # for an explanation fig = px.scatter(df, x=df[x_col], y=df[y_col], text=df.index) fig.update_traces(selectedpoints=selectedpoints, customdata=df.index, mode='markers+text', marker={ 'color': 'rgba(0, 116, 217, 0.7)', 'size': 20 }, unselected={'marker': { 'opacity': 0.3 }, 'textfont': { 'color': 'rgba(0, 0, 0, 0)' }}) fig.update_layout(margin={'l': 20, 'r': 0, 'b': 15, 't': 5}, dragmode='select', hovermode=False) fig.add_shape(dict({'type': 'rect', 'line': { 'width': 1, 'dash': 'dot', 'color': 'darkgrey' }}, **selection_bounds)) return fig # this callback defines 3 figures # as a function of the intersection of their 3 selections @app.callback( [Output('g1', 'figure'), Output('g2', 'figure'), Output('g3', 'figure')], [Input('g1', 'selectedData'), Input('g2', 'selectedData'), Input('g3', 'selectedData')] ) def callback(selection1, selection2, selection3): selectedpoints = df.index for selected_data in [selection1, selection2, selection3]: if selected_data and selected_data['points']: selectedpoints = np.intersect1d(selectedpoints, [p['customdata'] for p in selected_data['points']]) return [get_figure(df, "Col 1", "Col 2", selectedpoints, selection1), get_figure(df, "Col 3", "Col 4", selectedpoints, selection2), get_figure(df, "Col 5", "Col 6", selectedpoints, selection3)] if __name__ == '__main__': app.run_server(debug=True) 试着在任何一个图中点击并拖动以过滤不同的区域。在每次选择时，三个图形回调都会与每个图的最新选定区域一起启动。pandas 数据框会根据所选点进行过滤，并重新绘制图形，高亮显示所选点，并将所选区域画成虚线矩形。回调间共享数据在回调入门指南中提到，Dash 核心原则之一是 Dash 回调决不能修改其作用域之外的变量。修改任何全局变量都是不安全的。本章解释了为什么，并提供了一些在回调之间共享状态的替代模式。为什么分享状态？在某些应用程序中，您可能有多个回调，这些回调依赖于昂贵的数据处理任务，如进行 SQL 查询、运行模拟或下载数据。与其让每个回调运行同样昂贵的任务，不如让一个回调运行任务，然后将结果分享给其他回调。现在这种需求已经得到了一定的改善，你可以为一个回调设置多个输出。这样一来，那个昂贵的任务就可以做一次，并立即用于所有的输出。但是在某些情况下，这仍然不是很理想，比如说如果有简单的后续任务修改结果，比如单位转换。我们不应该为了把结果从华氏度改成摄氏度而重复一个大型数据库查询！为什么全局变量会打断你的 app？Dash 被设计为在多用户环境中工作，在这种环境中，多人可能会同时查看应用程序，并且会有独立的会话。如果您的应用程序使用修改过的全局变量，那么一个用户的会话可以将变量设置为一个值，这将影响下一个用户的会话。Dash 还被设计成能够与多个 python Worker 一起运行，这样回调就可以并行执行。这通常是通过使用 gunicorn 来实现的，使用的语法包括$ gunicorn --workers 4 app:server(app refers to a file named app.py and server refers to a variable in that file named server: server = app.server).当 Dash 应用程序在多个 Worker 之间运行时，它们的内存是不共享的。这意味着，如果您在一个回调中修改了全局变量，则该修改将不会应用于其他 Worker。这里是一个应用程序的草图，它有一个回调，可以修改其范围外的数据。由于上述原因，这种类型的模式*不会可靠地工作df = pd.DataFrame({ 'a': [1, 2, 3], 'b': [4, 1, 4], 'c': ['x', 'y', 'z'], }) app.layout = html.Div([ dcc.Dropdown( id='dropdown', options=[{'label': i, 'value': i} for i in df['c'].unique()], value='a' ), html.Div(id='output'), ]) @app.callback(Output('output', 'children'), [Input('dropdown', 'value')]) def update_output_1(value): # Here, `df` is an example of a variable that is # "outside the scope of this function". # *It is not safe to modify or reassign this variable # inside this callback.* global df = df[df['c'] == value] # do not do this, this is not safe! return len(df)要修复这个例子，只需将过滤器重新分配到回调内的一个新变量，或者遵循本指南下一部分中概述的策略之一。df = pd.DataFrame({ 'a': [1, 2, 3], 'b': [4, 1, 4], 'c': ['x', 'y', 'z'], }) app.layout = html.Div([ dcc.Dropdown( id='dropdown', options=[{'label': i, 'value': i} for i in df['c'].unique()], value='a' ), html.Div(id='output'), ]) @app.callback(Output('output', 'children'), [Input('dropdown', 'value')]) def update_output_1(value): # Safely reassign the filter to a new variable filtered_df = df[df['c'] == value] return len(filtered_df)回调间共享数据为了在多个 python 进程之间安全地共享数据，我们需要将数据存储在每个进程都能访问的地方。有三个主要的地方可以存储这些数据。1 - 在用户的浏览器会话中2 - 在磁盘上（例如，在一个文件或新的数据库上）。3 - 在共享内存空间中，比如 Redis。以下三个例子说明了这些方法。案例 1：在浏览器中使用隐藏的 Div 存储数据。通过在 https://community.plotly.com/t/sharing-a-dataframe-between-plots/6173 中解释的方法将数据保存为 Dash 前端存储的一部分来实现。数据必须转换为 JSON 这样的字符串进行存储和传输。以这种方式缓存的数据将只在用户当前会话中可用。如果你打开一个新的浏览器，应用程序的回调将始终计算数据。数据只在会话内的回调之间进行缓存和传输。因此，与缓存不同，这种方法不会增加应用程序的内存占用。在网络传输方面可能会有成本。如果你在回调之间共享 10MB 的数据，那么这些数据将在每个回调之间通过网络传输。如果网络成本太高，那就在前期计算聚合并传输这些。你的应用很可能不会显示 10MB 的数据，它只是显示其中的一个子集或聚合。这个例子概述了如何在一个回调中执行一个昂贵的数据处理步骤，将输出序列化为 JSON，并将其作为输入提供给其他回调。这个例子使用标准的 Dash 回调，并将 JSON 化的数据存储在应用程序中的一个隐藏的 div 中。global_df = pd.read_csv('...') app.layout = html.Div([ dcc.Graph(id='graph'), html.Table(id='table'), dcc.Dropdown(id='dropdown'), # Hidden div inside the app that stores the intermediate value html.Div(id='intermediate-value', style={'display': 'none'}) ]) @app.callback(Output('intermediate-value', 'children'), [Input('dropdown', 'value')]) def clean_data(value): # some expensive clean data step cleaned_df = your_expensive_clean_or_compute_step(value) # more generally, this line would be # json.dumps(cleaned_df) return cleaned_df.to_json(date_format='iso', orient='split') @app.callback(Output('graph', 'figure'), [Input('intermediate-value', 'children')]) def update_graph(jsonified_cleaned_data): # more generally, this line would be # json.loads(jsonified_cleaned_data) dff = pd.read_json(jsonified_cleaned_data, orient='split') figure = create_figure(dff) return figure @app.callback(Output('table', 'children'), [Input('intermediate-value', 'children')]) def update_table(jsonified_cleaned_data): dff = pd.read_json(jsonified_cleaned_data, orient='split') table = create_table(dff) return table 案例 2：预先计算如果数据很大，通过网络发送计算数据可能会很昂贵。在某些情况下，将这些数据和 JSON 序列化也会很昂贵。在许多情况下，您的应用程序将只显示计算或过滤数据的一个子集或聚合。在这些情况下，你可以在你的数据处理回调中预先计算你的聚合，并将这些聚合传输到其余回调中。下面是一个简单的例子，说明如何将过滤的或聚合的数据传输到多个回调。@app.callback( Output('intermediate-value', 'children'), [Input('dropdown', 'value')]) def clean_data(value): # an expensive query step cleaned_df = your_expensive_clean_or_compute_step(value) # a few filter steps that compute the data # as it's needed in the future callbacks df_1 = cleaned_df[cleaned_df['fruit'] == 'apples'] df_2 = cleaned_df[cleaned_df['fruit'] == 'oranges'] df_3 = cleaned_df[cleaned_df['fruit'] == 'figs'] datasets = { 'df_1': df_1.to_json(orient='split', date_format='iso'), 'df_2': df_2.to_json(orient='split', date_format='iso'), 'df_3': df_3.to_json(orient='split', date_format='iso'), } return json.dumps(datasets) @app.callback( Output('graph', 'figure'), [Input('intermediate-value', 'children')]) def update_graph_1(jsonified_cleaned_data): datasets = json.loads(jsonified_cleaned_data) dff = pd.read_json(datasets['df_1'], orient='split') figure = create_figure_1(dff) return figure @app.callback( Output('graph', 'figure'), [Input('intermediate-value', 'children')]) def update_graph_2(jsonified_cleaned_data): datasets = json.loads(jsonified_cleaned_data) dff = pd.read_json(datasets['df_2'], orient='split') figure = create_figure_2(dff) return figure @app.callback( Output('graph', 'figure'), [Input('intermediate-value', 'children')]) def update_graph_3(jsonified_cleaned_data): datasets = json.loads(jsonified_cleaned_data) dff = pd.read_json(datasets['df_3'], orient='split') figure = create_figure_3(dff) return figure案例 3：缓存和信号传递这个例子：通过 Flask -Cache 使用 Redis 来存储 "全局变量"。这些数据通过一个函数进行访问，函数的输出被缓存，并通过其输入参数进行键控。当昂贵的计算完成后，使用隐藏的 div 方案向其他回调发送信号。注意，你也可以不使用 Redis，而是将其保存到文件系统中。更多细节请参见 https://flask-caching.readthedocs.io/en/latest/。这种 "信号传递 "很酷，因为它允许昂贵的计算只占用一个进程。如果没有这种类型的信令，每个回调最终可能会并行计算昂贵的计算，锁定四个进程而不是一个进程。这种方法的优势还在于，未来的会话可以使用预先计算的值。这对于那些输入数量较少的应用来说，会有很好的效果。下面是一个示例。一些需要注意的事情。我通过使用 time.sleep(5)模拟了一个昂贵的过程。当应用程序加载时，渲染所有四张图需要 5 秒。初始计算只阻止了一个进程。一旦计算完成，就会发送信号，并并行执行四个回调来渲染图形。每一个回调都会从 "全局存储"：Redis 或文件系统缓存中检索数据。我在 app.run_server 中设置了 processes=6，这样就可以并行执行多个回调。在生产中，可以使用类似 $ gunicorn --workers 6 --threads 2 app:server 这样的方法来完成。如果过去已经选择了一个值，那么在下拉菜单中选择该值的时间将少于 5 秒。这是因为该值是从缓存中提取的。同样，重新加载页面或在新窗口中打开应用程序也很快，因为初始状态和初始昂贵的计算已经被计算过了。import os import copy import time import datetime import dash import dash_core_components as dcc import dash_html_components as html import numpy as np import pandas as pd from dash.dependencies import Input, Output from flask_caching import Cache external_stylesheets = [ # Dash CSS 'https://codepen.io/chriddyp/pen/bWLwgP.css', # Loading screen CSS 'https://codepen.io/chriddyp/pen/brPBPO.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) CACHE_CONFIG = { # try 'filesystem' if you don't want to setup redis 'CACHE_TYPE': 'redis', 'CACHE_REDIS_URL': os.environ.get('REDIS_URL', 'redis://localhost:6379') } cache = Cache() cache.init_app(app.server, config=CACHE_CONFIG) N = 100 df = pd.DataFrame({ 'category': ( (['apples'] * 5 * N) + (['oranges'] * 10 * N) + (['figs'] * 20 * N) + (['pineapples'] * 15 * N) ) }) df['x'] = np.random.randn(len(df['category'])) df['y'] = np.random.randn(len(df['category'])) app.layout = html.Div([ dcc.Dropdown( id='dropdown', options=[{'label': i, 'value': i} for i in df['category'].unique()], value='apples' ), html.Div([ html.Div(dcc.Graph(id='graph-1'), className="six columns"), html.Div(dcc.Graph(id='graph-2'), className="six columns"), ], className="row"), html.Div([ html.Div(dcc.Graph(id='graph-3'), className="six columns"), html.Div(dcc.Graph(id='graph-4'), className="six columns"), ], className="row"), # hidden signal value html.Div(id='signal', style={'display': 'none'}) ]) # perform expensive computations in this "global store" # these computations are cached in a globally available # redis memory store which is available across processes # and for all time. @cache.memoize() def global_store(value): # simulate expensive query print('Computing value with {}'.format(value)) time.sleep(5) return df[df['category'] == value] def generate_figure(value, figure): fig = copy.deepcopy(figure) filtered_dataframe = global_store(value) fig['data'][0]['x'] = filtered_dataframe['x'] fig['data'][0]['y'] = filtered_dataframe['y'] fig['layout'] = {'margin': {'l': 20, 'r': 10, 'b': 20, 't': 10}} return fig @app.callback(Output('signal', 'children'), [Input('dropdown', 'value')]) def compute_value(value): # compute value and send a signal when done global_store(value) return value @app.callback(Output('graph-1', 'figure'), [Input('signal', 'children')]) def update_graph_1(value): # generate_figure gets data from `global_store`. # the data in `global_store` has already been computed # by the `compute_value` callback and the result is stored # in the global redis cached return generate_figure(value, { 'data': [{ 'type': 'scatter', 'mode': 'markers', 'marker': { 'opacity': 0.5, 'size': 14, 'line': {'border': 'thin darkgrey solid'} } }] }) @app.callback(Output('graph-2', 'figure'), [Input('signal', 'children')]) def update_graph_2(value): return generate_figure(value, { 'data': [{ 'type': 'scatter', 'mode': 'lines', 'line': {'shape': 'spline', 'width': 0.5}, }] }) @app.callback(Output('graph-3', 'figure'), [Input('signal', 'children')]) def update_graph_3(value): return generate_figure(value, { 'data': [{ 'type': 'histogram2d', }] }) @app.callback(Output('graph-4', 'figure'), [Input('signal', 'children')]) def update_graph_4(value): return generate_figure(value, { 'data': [{ 'type': 'histogram2dcontour', }] }) if __name__ == '__main__': app.run_server(debug=True, processes=6) 案例 4：服务器上基于用户的会话数据实时更新数据dash_core_components.Interval component 组件Dash 中的组件通常通过用户交互来更新：选择下拉、拖动滑块、悬停在点上。如果您正在构建一个用于监控的应用程序，您可能希望每隔几秒或几分钟更新一次应用程序中的组件。dash_core_components.Interval 元素允许你在预定义的时间间隔上更新组件。n_intervals 属性是一个整数，每次间隔的毫秒数过后都会自动递增。您可以在应用程序的回调中监听这个变量，以便在预定义的时间间隔上启动回调。这个例子从实时卫星馈送中提取数据，并每秒钟更新一次图表和文本。import datetime import dash import dash_core_components as dcc import dash_html_components as html import plotly from dash.dependencies import Input, Output # pip install pyorbital from pyorbital.orbital import Orbital satellite = Orbital('TERRA') external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) app.layout = html.Div( html.Div([ html.H4('TERRA Satellite Live Feed'), html.Div(id='live-update-text'), dcc.Graph(id='live-update-graph'), dcc.Interval( id='interval-component', interval=5*1000, # 5秒钟间隔 n_intervals=0 ) ]) ) @app.callback(Output('live-update-text', 'children'), [Input('interval-component', 'n_intervals')]) def update_metrics(n): lon, lat, alt = satellite.get_lonlatalt(datetime.datetime.now()) style = {'padding': '5px', 'fontSize': '16px'} return [ html.Span('Longitude: {0:.2f}'.format(lon), style=style), html.Span('Latitude: {0:.2f}'.format(lat), style=style), html.Span('Altitude: {0:0.2f}'.format(alt), style=style) ] # Multiple components can update everytime interval gets fired. @app.callback(Output('live-update-graph', 'figure'), [Input('interval-component', 'n_intervals')]) def update_graph_live(n): satellite = Orbital('TERRA') data = { 'time': [], 'Latitude': [], 'Longitude': [], 'Altitude': [] } # Collect some data for i in range(180): time = datetime.datetime.now() - datetime.timedelta(seconds=i*20) lon, lat, alt = satellite.get_lonlatalt( time ) data['Longitude'].append(lon) data['Latitude'].append(lat) data['Altitude'].append(alt) data['time'].append(time) # Create the graph with subplots fig = plotly.tools.make_subplots(rows=2, cols=1, vertical_spacing=0.2) fig['layout']['margin'] = { 'l': 30, 'r': 10, 'b': 30, 't': 10 } fig['layout']['legend'] = {'x': 0, 'y': 1, 'xanchor': 'left'} fig.append_trace({ 'x': data['time'], 'y': data['Altitude'], 'name': 'Altitude', 'mode': 'lines+markers', 'type': 'scatter' }, 1, 1) fig.append_trace({ 'x': data['Longitude'], 'y': data['Latitude'], 'text': data['time'], 'name': 'Longitude vs Latitude', 'mode': 'lines+markers', 'type': 'scatter' }, 2, 1) return fig if __name__ == '__main__': app.run_server(debug=True)上面说过，我们不能修改全局变量，但当全局变量是个字典时，你可以用它来存储新增的数据，以此实现依赖上一个数据计算的情况。比如：data = { 'time': [], 'Latitude': [], 'Longitude': [], 'Altitude': [] } # Multiple components can update everytime interval gets fired. @app.callback(Output('live-update-graph', 'figure'), [Input('interval-component', 'n_intervals')]) def update_graph_live(n): satellite = Orbital('TERRA') # Collect some data time = datetime.datetime.now() - datetime.timedelta(seconds=i*20) lon, lat, alt = satellite.get_lonlatalt( time ) data['Longitude'].append(lon) data['Latitude'].append(lat) data['Altitude'].append(alt) data['time'].append(time) # Create the graph with subplots fig = plotly.tools.make_subplots(rows=2, cols=1, vertical_spacing=0.2) fig['layout']['margin'] = { 'l': 30, 'r': 10, 'b': 30, 't': 10 } fig['layout']['legend'] = {'x': 0, 'y': 1, 'xanchor': 'left'} fig.append_trace({ 'x': data['time'], 'y': data['Altitude'], 'name': 'Altitude', 'mode': 'lines+markers', 'type': 'scatter' }, 1, 1) fig.append_trace({ 'x': data['Longitude'], 'y': data['Latitude'], 'text': data['time'], 'name': 'Longitude vs Latitude', 'mode': 'lines+markers', 'type': 'scatter' }, 2, 1) return fig if __name__ == '__main__': app.run_server(debug=True)页面加载的更新默认情况下，Dash 应用程序会将 app.layout 存储在内存中。这可确保布局只在应用程序启动时计算一次。如果您将 app.layout 设置为函数，那么您可以在每次页面加载时提供动态布局。例如，如果您的 app.layout 看起来像这样。import datetime import dash import dash_html_components as html app.layout = html.H1('The time is: ' + str(datetime.datetime.now())) if __name__ == '__main__': app.run_server(debug=True)那么你的应用程序将显示应用程序启动时的时间。如果你把它改为函数，那么每次刷新页面时都会计算出一个新的日期时间。试试吧。提高性能缓存由于 Dash 的回调在本质上是函数式的（它们不包含任何状态），所以很容易添加 memoization 缓存。Memoization 会在函数被调用后存储它的结果，并且如果用相同的参数调用该函数，则会重新使用该结果。Dash 应用程序经常跨多个进程或线程部署。在这些情况下，每个进程或线程都包含自己的内存，它不会跨实例共享内存。这意味着，如果我们要使用 lru_cache，我们的缓存结果可能不会在各个会话之间共享。取而代之的是，我们可以使用 Flask-Caching 库，它可以将结果保存在共享内存数据库中，比如 Redis，或者作为文件系统中的文件。Flask-Caching 还有其他不错的功能，比如基于时间的过期。如果你想每小时或每天更新数据（清除缓存），基于时间的过期是很有用的。下面是一个使用 Redis 的 Flask-Caching 的例子。import datetime import os import dash import dash_core_components as dcc import dash_html_components as html from dash.dependencies import Input, Output from flask_caching import Cache external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) cache = Cache(app.server, config={ # try 'filesystem' if you don't want to setup redis 'CACHE_TYPE': 'redis', 'CACHE_REDIS_URL': os.environ.get('REDIS_URL', '') }) app.config.suppress_callback_exceptions = True timeout = 20 app.layout = html.Div([ html.Div(id='flask-cache-memoized-children'), dcc.RadioItems( id='flask-cache-memoized-dropdown', options=[ {'label': 'Option {}'.format(i), 'value': 'Option {}'.format(i)} for i in range(1, 4) ], value='Option 1' ), html.Div('Results are cached for {} seconds'.format(timeout)) ]) @app.callback( Output('flask-cache-memoized-children', 'children'), [Input('flask-cache-memoized-dropdown', 'value')]) @cache.memoize(timeout=timeout) # in seconds def render(value): return 'Selected "{}" at "{}"'.format( value, datetime.datetime.now().strftime('%H:%M:%S') ) if __name__ == '__main__': app.run_server(debug=True)这里是一个缓存数据集而不是回调的例子。它使用 FileSystem 缓存，将缓存的结果保存到文件系统中。如果有一个数据集被用来更新几个回调，这种方法很好用。import datetime as dt import os import time import dash import dash_core_components as dcc import dash_html_components as html import numpy as np import pandas as pd from dash.dependencies import Input, Output from flask_caching import Cache external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css'] app = dash.Dash(__name__, external_stylesheets=external_stylesheets) cache = Cache(app.server, config={ 'CACHE_TYPE': 'filesystem', 'CACHE_DIR': 'cache-directory' }) TIMEOUT = 60 @cache.memoize(timeout=TIMEOUT) def query_data(): # This could be an expensive data querying step df = pd.DataFrame( np.random.randint(0,100,size=(100, 4)), columns=list('ABCD') ) now = dt.datetime.now() df['time'] = [now - dt.timedelta(seconds=5*i) for i in range(100)] return df.to_json(date_format='iso', orient='split') def dataframe(): return pd.read_json(query_data(), orient='split') app.layout = html.Div([ html.Div('Data was updated within the last {} seconds'.format(TIMEOUT)), dcc.Dropdown( id='live-dropdown', value='A', options=[{'label': i, 'value': i} for i in dataframe().columns] ), dcc.Graph(id='live-graph') ]) @app.callback(Output('live-graph', 'figure'), [Input('live-dropdown', 'value')]) def update_live_graph(value): df = dataframe() now = dt.datetime.now() return { 'data': [{ 'x': df['time'], 'y': df[value], 'line': { 'width': 1, 'color': '#0074D9', 'shape': 'spline' } }], 'layout': { # display the current position of now # this line will be between 0 and 60 seconds # away from the last datapoint 'shapes': [{ 'type': 'line', 'xref': 'x', 'x0': now, 'x1': now, 'yref': 'paper', 'y0': 0, 'y1': 1, 'line': {'color': 'darkgrey', 'width': 1} }], 'annotations': [{ 'showarrow': False, 'xref': 'x', 'x': now, 'xanchor': 'right', 'yref': 'paper', 'y': 0.95, 'yanchor': 'top', 'text': 'Current time ({}:{}:{})'.format( now.hour, now.minute, now.second), 'bgcolor': 'rgba(255, 255, 255, 0.8)' }], # aesthetic options 'margin': {'l': 40, 'b': 40, 'r': 20, 't': 10}, 'xaxis': {'showgrid': False, 'zeroline': False}, 'yaxis': {'showgrid': False, 'zeroline': False} } } if __name__ == '__main__': app.run_server(debug=True)
前言之所以想做这个 API，主要是我博客上有百度收录检测，但是速度有点慢，后来想想可能是因为服务器在国外，所以就有了把 API 搬到国内的想法。但是国内的 API 已经用 python 搭建了一个随机图服务，不好再加一个 PHP 吧？算了，干脆我重写一个 API 得了，于是就有了这个项目。代码已经放到 GitHub 上了，欢迎大家交流讨论。[scode type="yellow"]查询次数过于频繁有大概率会查询失败，目前只是半成品，只适合自用[/scode][scode type="lblue"]源码里有其他 API，大家直接看 get_baidu 即可[/scode]Github 地址Demo：https://api.yanshu.work/baidu/?u=https://www.sitstars.com思路原作者检测的方法貌似是直接查询网址，然后匹配有多少搜索结果几条，如果搜索结果大于 0 就认为已收录。但是这种方法有个问题，百度有时候会收录一些与目标网址无关的页面，这样检测会不准确。网上搜了搜，其他一些文章是查询百度跳转链接，然后一个个获取真实链接，再与查询的网址进行匹配，如果能匹配上就说明已收录。嗯想法很好，理论上准确性非常高，就是有点耗时间……不管怎么样总归是个方案，先试试。很遗憾，技术有限，我并不能获取百度跳转后的真实链接，Stack Overflow 上给出的解决方法也统统不起作用，这个方案宣告失败。PS：记录下获取跳转后真实链接的几种方法，说不定以后会用上。# 方法一 r = requests.get('http://techtv.mit.edu/videos/1585-music-session-02/download.source') print(r.url) # http://d1baxxa0joomi3.cloudfront.net/2515a9db659b0ab26d869b4ff2dadca9/original.mov # 方法二：这种适用于多次跳转 response = requests.get(someurl) if response.history: print("Request was redirected") for resp in response.history: print(resp.status_code, resp.url) print("Final destination:") print(response.status_code, response.url) else: print("Request was not redirected") # 方法三 r = requests.get('http://github.com/', allow_redirects=False) r.status_code # 302 r.url # http://github.com, not https. r.headers['Location'] # https://github.com/ -- the redirect destination # 方法四 import urllib.request res = urllib.request.urlopen(starturl) finalurl = res.geturl() print(finalurl) 参考资料：https://stackoverflow.com/questions/20475552/python-requests-library-redirect-new-url分析只能一步步分析了，首先看个已收录的文章： 再看看没有收录的文章： 突然有了想法，我直接解析 url 不就行了？如果能匹配到就说明已收录，匹配不到就没有收录。顺着这个思路，很快写好了第一版代码。def get_baidu(url): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36' } baidu = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&tn=baidu&wd=' search_url = baidu + url response = requests.get(search_url,headers=headers) text = response.text if 'http' in url: url = url[url.index('//')+2:] pattern=re.compile(r'{}'.format(url)) result_list = pattern.findall(text) if result_list: return 200 else: return 403随便测试了几篇文章，确实有效，可以收工了……且慢，再找找其他人的文章测试一下吧。输入 http://www.ruanyifeng.com/blog/2017/12/blockchain-tutorial.html，嗯？怎么返回 403 了？用百度搜了一下，没错啊，已经收录了，而且搜索出来的结果也是加粗的。 仔细看了一下，哦……原来百度把太长的网址隐藏了一部分啊，所以匹配不到。这就难办了，诚然我可以截取网址的一部分进行匹配，但是这样就又不精确了。我又仔细观察了一下，发现如果百度确实收录了，那么这个网址会在搜索结果的最上方，而且第一条和第二条搜索结果之间会有“以下是网页中包含……”的提示信息。于是我有了个新的想法，首先截取网页文本从开头到“以下是网页中……”的内容，然后用正则匹配百度跳转链接，如果能匹配得到说明百度第一条有搜索到的网页结果，那这篇文章肯定收录了。于是我写出了第二版代码。@app.get("/baidu/") async def get_baidu(u: str = Query(..., min_length=3)): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36', 'Referer': 'https://www.baidu.com' } # 网址后缀处理 if '.' in u[-5:]: pass elif u[-1] != '/': u += '/' # 网址协议头处理 if ('http' in u) and not ('www' in u): u = 'www.' + u[u.index('//')+2:] elif ('http' not in u) and not ('www' in u): u = 'www.' + u baidu = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&tn=baidu&wd=' search_url = baidu + u try: response = requests.get(search_url, headers=headers) text = response.text text = text[:text.index('以下是网页中包含')] pattern = re.compile(r'"http://www.baidu.com/link\?url=.+?"') result_list = pattern.findall(text) if result_list: return {'code': 200, 'msg': '该网址已被百度收录！'} else: return {'code': 403, 'msg': '该网址暂未被百度收录！'} except: return {'code': 404, 'msg': '百度收录查询失败！'}嗯，很好，这次是真的可以收工了……（Flag 预定）。PS：可以看到代码里有个网址的预处理，这是因为网址没有最后一个 / 时会查不出来，但是 .html 这种格式的又不需要加。另外，网址开头可以不带 http/https，但是必须带 www，要不然同样搜不出来。跨域问题测试没问题，那就可以上线了，部署到服务器上，按照 https://www.sitstars.com/archives/65/中的教程修改 API 地址，看下情况。问题又来了……API 请求地址出现了一个鲜红的红色，提示 strict-origin-when-cross-origin 错误，看样子是跨域错误？试着用错误提示 +fastapi 在 Google 上搜了一下，果然 fastapi 早已经考虑到了这个问题，并且给出了解决方案。参考资料：https://fastapi.tiangolo.com/tutorial/cors/简单来说，加入以下代码就可以了：from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware app = FastAPI() origins = [ "http://localhost.tiangolo.com", "https://localhost.tiangolo.com", "http://localhost", "http://localhost:8080", "*" ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )origins 是一个所有允许的请求域列表，加入 * 代表允许所有请求，当然也可以加入具体的网址和端口，比如上面几个例子。加好后再测试一下，成功显示！而且速度还挺快。百度自身问题正当我开心地刷新网站看效果时，悲剧了，网页上突然提示我百度收录查询失败！这是为何？手动试了一下 API，果然输出不了结果，但是本地使用又完全没有问题。莫非是我搜得过于频繁导致 ip 被百度拉黑了？于是我找了很多种解决方案，比如 headers 添加更多参数，更换代理 ip，完全没用！没办法我又看了原作者的博客，注意到 PHP 版本 API 的代码里有个 CURLOPT_FOLLOWLOCATION，百度一下发现它的功能是跟随网址的重定向。嗯……会不会是服务器搜索太快导致百度重定向没反应过来？于是我借助第二节的代码，print()出最终跳转后的网址，看看它到底给我定向到哪了。修改好代码后再次部署，没想到这次 API 又有用了，看来是抽风性质的问题。刷新了几次又不能用了，而此时服务器的控制台上也刷出了重定向后的网址，和我拼接的百度查询网址一样！这又是什么情况？我自身对网络方面知之甚少，所以最终也没有解决方案，只是知道了判断百度查询失败的另一个方法：只要有重定向那就一定失败了。后记这玩意搞了我一个晚上，最终还是放弃了，留了一个 GitHub 上的半成品，本来想部署好后开放给大家用的，但是现在只能先自用了。大家有兴趣的话可以自行部署，或者留言我把你的域名添加到跨域列表里。接下来我的计划是加入缓存，请求成功后存储到本地，一段时间内只从本地 读取结果，防止每次打开文章都要请求一次百度。
最近一时兴起想好好搞一下我的 API 项目，用的自然是 python 的 fastapi 库。没做几个 API 就想到个问题：我怎么知道我的 API 有没有人用？解决方案当然就是统计一下 API 调用次数啦。网上搜了一下貌似没有 fastapi 简单可行的统计方案，没办法只能自己写了。那么首先我需要知道访客的 ip，这样相当于有个 key，也可以大致了解调用情况。去 fastapi 文档上查了一下，有相关代码！from fastapi import FastAPI, Request app = FastAPI() @app.get("/items/{item_id}") def read_root(item_id: str, request: Request): client_host = request.client.host return {"client_host": client_host, "item_id": item_id}赶紧复制到 python 里试试效果，没想到，它给我返回 127.0.0.1？我想让它给我返回用户的真实 ip 而不是 本地 ip 啊。想必部署到服务器上显示的也是服务端的 ip 吧，试了一下果然如此。那没办法了，万事不决求助 Google。很快在 Stack Overflow 上找到了类似的提问，整合了一下几个回答，终于摸索出了解决方案。具体来说分为三步：首先使用 nginx 进行反代，因为我这个项目本身就是用的 nginx 所以这一步跳过。然后修改 nginx 的配置文件如下：server { # the port your site will be served on listen 80; # the domain name it will serve for server_name <your_host_name>; # substitute your machine's IP address or FQDN # add_header Access-Control-Allow-Origin *; # add_header Access-Control-Allow-Credentials: true; add_header Access-Control-Allow-Headers Content-Type,XFILENAME,XFILECATEGORY,XFILESIZE; add_header access-control-allow-headers authorization; # Finally, send all non-media requests to the Django server. location / { proxy_pass http://127.0.0.1:8000/; # the uvicorn server address proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } }本人小白，只会用宝塔，同样用宝塔的同学可以在反向代理中找到配置文件，不过宝塔的反代貌似不用做任何修改。反正我是这一步也跳过了。最后，也是最重要的一步，使用 uvicorn 启动服务，但是要加上 --proxy-headers --forwarded-allow-ips='*'，也即：uvicorn main:app --host 0.0.0.0 --port 8000 --proxy-headers --forwarded-allow-ips='*'这样就可以成功获取访客 ip 啦。但是测试下来，我另外一个网站调用的这个 API，打开网站时日志输出的是我电脑的 ip，而非网站服务器的 ip，目前仍然不知道解决方案……
今天讲讲我在实习中学到的一点 python 知识，核心内容是多进程，也即我们常说的并行计算。map首先提个问题，给出一个列表，对列表中的每个元素都平方，代码怎么写？最简单直观的方法自然就是 for 循环。alist = [1,2,3,4,5,6,7,8] def power_value(num): return num**2 result_list = [] for num in alist: result_list.append(power_value(num)) print(result_list) >>> [1, 4, 9, 16, 25, 36, 49, 64]注：当然你也可以不定义函数直接循环，这里主要是为了方便开展下文。学过列表推导式的同学可以写出更加简洁的方案：result_list = [power_value(x) for x in alist] print(result_list) >>> [1, 4, 9, 16, 25, 36, 49, 64]但其实，除了上面两种方法外，python 还内置了一个函数 map，专门解决这种不断往同一个函数传不同参数的问题。它的语法也很简单：map(function, iterable, ...)第一个参数为函数名，第二个参数和之后的参数均为序列（列表、元组、range()等等）。第一个参数 function 以参数序列中的每一个元素调用 function 函数，返回包含每次 function 函数返回值的新列表。有了这个函数，上面的代码就可以改写为：result_list = list(map(power_value,alist)) print(result_list) >>> [1, 4, 9, 16, 25, 36, 49, 64]这里为什么要加一个 list() 呢？因为 map 返回的是一个迭代器，一般情况下只能循环遍历读取，所以加个 list 把它转换为列表。上面的函数只有一个参数，那假如它有两个参数呢？我们分为两种情况，一是这两个参数是成对出现的，比如说，两个列表相加，就是列表对应的元素相加。这种情况我们也可以直接应用 map。alist = [1,2,3,4,5,6,7,8] blist = [2,3,4,5,6,7,8,9] def add_value(a,b): return a + b result_list = list(map(add_value,alist,blist)) print(result_list) >>> [3, 5, 7, 9, 11, 13, 15, 17]第二种情况，这两个参数并不对应，常见情况是需要固定一个参数，比如上面的加法函数，我需要固定 b=5，这时候，可以借助 python 提供的另外一个函数 partial，这是 python 自带包 functools 里的函数，它的作用就是固定某些参数，从而构造出一个新函数。其语法为：partial(func, param, ……)这里有个容易忽略的点，如果不指定参数位置的话，那么默认从第一个参数开始固定，所以使用它的时候最好指定参数。但是被指定的参数必须在函数的最后部分，比如 func01(a,b,c)，我们可以固定 b 和 c，可以固定 c，但是绝对不能只固定 b！否则会报错。from functools import partial alist = [1,2,3,4,5,6,7,8] def add_value(a,b): return (a + b) * a func = partial(add_value,b = 5) result_list = list(map(func,alist)) print(result_list) >>> [6, 14, 24, 36, 50, 66, 84, 104]多进程常被和进程一起提起的是线程，不过我也没搞懂这两的具体原理。总之明白两个区别就好：进程之前数据不共享，线程之间数据共享进程之间不会相互影响，一个进程挂掉另外一个进程照常工作。而一个线程挂掉则所有线程都会挂掉。我在工作中常用到的是多进程，所以主要讲讲它。python 自带了一个多进程库 multiprocessing ，利用上面学到的 map 知识，我们可以很容易实现并行运算，有了它，上面的代码可以改写如下：import multiprocessing from functools import partial alist = [1,2,3,4,5,6,7,8] def add_value(a,b): return (a + b) * a func = partial(add_value,b = 5) if __name__ == '__main__': num_processes = multiprocessing.cpu_count()-2 # 使用核心数 pool = multiprocessing.Pool(processes=num_processes) # 实例化进程池 func = partial(add_value,b = 5) print(pool.map(func,alist))这样就实现了一个最简单的多进程，可以发现，和 map 相比就多了一个实例化进程池的过程。需要注意的是，这里开启多进程，肯定是要比单进程速度慢的，因为系统在进程间分配任务也是需要时间的，所以我们也不能无脑开多进程，大型任务才可能需要。另外，pool.map 返回的是一个列表，如果想像 map 那样返回迭代器的话，可以使用 pool.imap，这两个函数返回都是有序的，如果有特殊需求想返回无须结果，可以使用 imap_unordered 最重要的一点，多进程必须在 if __name__ == '__main__' 下写！否则会报错，当然了，你在函数里使用多进程，然后在 if __name__ == '__main__' 下调用这个函数也是允许的。这是 Windows 上多进程的实现问题。在 Windows 上，子进程会自动 import 启动它的这个文件，而在 import 的时候是会执行这些语句的。如果你这么写的话就会无限递归创建子进程报错。所以必须把创建子进程的部分用那个 if 判断保护起来， import 的时候 name 不是 main，就不会递归运行了。@cholerae进度条在执行大型任务的时候，我一般是输出一个中间结果就 print 一条提示语句，这样可以让我知道程序仍然在正常运行，而且也能大致知道处理到哪了。不过，如果只是上面两点需求的话，其实可以使用进度条，这样更加方便直观。最常见的进度条库是 tqdm，pip 安装好后就可以使用了，语法也很简单。from tqdm import tqdm,trange for i in trange(100): time.sleep(0.1) for i in tqdm(range(100)): time.sleep(0.1)不过问题在于，多进程下如何使用进度条？自然，可以为每个进程生成一个进度条，但是我想看总的任务进度，而不是展示 5、6 根进度，怎么做？针对这个需求，我借鉴 Lei Mao，重写了 run_imap_mp 函数，用法和 map 基本一样，其他参数注释里也有说明，这里就不细说了。def run_imap_mp(func, argument_list, num_processes='', is_tqdm=True): ''' 多进程与进度条结合 param: ------ func:function 函数 argument_list:list 参数列表 num_processes:int 进程数，不填默认为总核心-3 is_tqdm:bool 是否展示进度条，默认展示 ''' result_list_tqdm = [] try: import multiprocessing if num_processes == '': num_processes = multiprocessing.cpu_count()-3 pool = multiprocessing.Pool(processes=num_processes) if is_tqdm: from tqdm import tqdm for result in tqdm(pool.imap(func=func, iterable=argument_list), total=len(argument_list)): result_list_tqdm.append(result) else: for result in pool.imap(func=func, iterable=argument_list): result_list_tqdm.append(result) pool.close() except: result_list_tqdm = list(map(func,argument_list)) return result_list_tqdm
这篇文章来自：https://uwpce-pythoncert.github.io/SystemDevelopment/profiling.html，译者进行了调整与补充。前言：性能分析与优化什么是性能？"性能 "可以是以下任何一项的衡量标准。资源使用（CPU、内存）调用函数的频率与时间耗费优化步骤编写可维护/可读性好的代码测试收集性能统计数据如果够快，就 ok 了。不然就根据数据优化最耗费的部分。优化的重点在于程序核心部分，对于非核心部分，如果考虑到调试和维护的问题，对这部分提高效率的尝试实际上会产生强烈的负面影响。过早的优化是万恶之源。-Donald Knuth优化方法高效算法(大 O 等...)合适的 Python 数据类型等适当的 Python 规范专用包(numpy, scipy)调用外部包使用 C/C++/Fortran/Cython 等。什么是大 O？一个算法的效率通常用 "O"符号来描述。之所以使用字母 O，是因为函数的增长速度也被称为函数的阶数，它描述了一个算法所使用的资源与输入数据量的函数关系。O(1) - (性能恒定) 无论提供多少数据，执行时间都保持不变。如：字典新增一个键值对。O(n) - 执行时间随输入数据的增加而线性上升。如：遍历列表。$O(n^2)$ - 执行时间随输入数据的增加而呈二次方上升。如：最糟糕情况下的冒泡排序。O(log(n)) - 随着输入数据的对数增加而增加。如：二等分搜索。 统计函数执行时间最简单的方法当然是使用 python 内置的计时器，当一个单位的代码（如函数）开始时开始计时，当代码返回时停止计时。像大多数计时基准一样，获得的数据只对特定的测试环境（机器/操作系统/python 版本）有效。例如，在网络快、磁盘慢的机器上运行，在网络慢、磁盘快的系统上运行，结果可能大不相同。time.clock() / time.time()将时间模块用作装饰器。time.time() 返回 unix 系统时间（挂钟时间）。time.clock()返回当前进程的 CPU 时间。很简单，但是可以直观地反映情况。例子：import time def timer(func): def timer(*args, **kwargs): """a decorator which prints execution time of the decorated function""" t1 = time.time() result = func(*args, **kwargs) t2 = time.time() print("-- executed %s in %.4f seconds" % (func.func_name, (t2 - t1))) return result return timer @timer def expensive_function(): time.sleep(1) @timer def less_expensive_function(): time.sleep(.02) expensive_function() less_expensive_function()timeit用于测试小块代码多次运行给定的语句并计算平均执行时间。可以从命令行运行。从命令行运行：python -m timeit '"-".join(str(n) for n in range(100))'参考文档：https://docs.python.org/3.5/library/timeit.html可选参数：-nN：在一个循环中执行给定语句 N 次。如果没有给出这个值，则选择一个合适的值。-rR：重复循环迭代 R 次，取最佳结果。默认值：3-t：使用 time.time 来测量时间，这是 Unix 上的默认值。该函数测量时钟时间。-c：使用 time.clock 来测量时间，这是 Windows 上的默认值，测量时钟时间。在 Unix 上，使用 resource.getrusage 代替，返回 CPU 用户时间。-pP：使用精度为 P 的数字来显示计时结果。默认值：3如：$ python -m timeit -n 1000 -t "len([x**2 for x in range(1000)])"timeit 同样可以作为一个模块引入：参考文档：http://docs.python.org/3/library/timeit.html#timeit.timeitimport timeit statement = "char in text" setup_code = """'food'.find('f') >= 0""" timeit.timeit(statement, setup=setup_code, number=1000)Setup 参数即要测试的代码语句。timeit 同样包括在 ipython 的魔法命令中。%timeit pass u = None %timeit u is None %timeit -r 4 u == None import time %timeit -n1 time.sleep(2) %timeit -n 10000 "f" in "food"参考文档 http://ipython.readthedocs.io/en/stable/interactive/magics.html?#magic-timeit获取更加详细的信息只有当你知道你的代码慢在哪里的时候，上一节的代码才是有用的。如果你只知道你的代码很慢，却不知道到底哪里出问题了，怎么办？一部分一部分试吗？当然不，这里我们就要用到 profiler（分析器）了。profiler 会对运行时的性能进行测量，并将结果汇总成分析报告。报告的指标可包括：随着时间推移使用的内存每个函数分配的内存函数调用频率函数调用时间子函数调用的累计时间python 内置的分析器Python 自带了几个 profiler 模块profile - 纯 Python 编写。如果你需要扩展 profiler，可以使用它。否则不要用，因为它比较慢。cProfile - 与 profile 的 API 相同，但用 C 语言编写，开销较小。可以直接在命令行中使用：python -m cProfile [-o output_file] [-s sort_order] read_bna.py # 调用命令 11111128 function calls in 8.283 seconds Ordered by: standard name ncalls tottime percall cumtime percall filename:lineno(function) 1 0.000 0.000 0.000 0.000 integrate.py:1() 11111110 2.879 0.000 2.879 0.000 integrate.py:1(f) [....]ncalls：调用次数tottime：调用该函数中花费的总时间，不包括调用子函数的时间。percall：tottime / ncalls，每次调用的耗时cumtime：在函数中花费的总时间，包括在子函数中的时间。percall： cumtime / ncallsfilename:lineno - 函数的位置。分析统计用 -o < 文件名 > 参数输出为二进制文件。在进行性能分析时，请保存您的分析文件，以便和优化后的程序进行比较。分析文件可以通过 pstats 打开python -m pstats详细用法：python -m cProfile -o prof_dump ./read_bna.py python -m pstats % read prof_dump # show stats: prof_dump% stats # only the top 5 results: prof_dump% stats 5 # sort by cumulative time: prof_dump% sort cumulative # shorten long filenames for display: prof_dump% strip # show results again: prof_dump% stats 5pstats 当然也可以作为方法调用：import pstats p = pstats.Stats('prof_dump') p.sort_stats('calls', 'cumulative') p.print_stats() # Output can be restricted via arguments to print_stats(). # Each restriction is either an integer (to select a count of lines), # a decimal fraction between 0.0 and 1.0 inclusive (to select a percentage of lines), # or a regular expression (to pattern match the standard name that is printed. # If several restrictions are provided, then they are applied sequentially.装饰器和上一节一样，我们同样可以将 cProfile 编写为装饰器。下面代码参考了：https://juejin.cn/post/6844903760414654478def func_cprofile(f): """ 内建分析器 """ @wraps(f) def wrapper(*args, **kwargs): profile = cProfile.Profile() try: profile.enable() result = f(*args, **kwargs) profile.disable() return result finally: profile.print_stats(sort='time') return wrapper使用：@func_cprofile def test(): for x in range(10000000): print(x)其他工具SNAKEVIZ一个分析文件的可视化展示工具。https://jiffyclub.github.io/snakeviz/安装：pip install snakeviz使用：snakeviz program.prof line profiler上面介绍的方法大多是以函数为单位进行分析，line_profiler 则是一个用于对函数进行逐行剖析的模块。使用 -l 启用逐行分析。用 @profile 来装饰你要分析的函数，然后运行。# the -v option will display the profile data immediately, instead # of just writing it to <filename.py>.lprof $ kernprof -l -v integrate_main.py # load the output with $ python -m line_profiler integrate_main.py.lprofhttps://github.com/rkern/line_profiler使用内存分析一个选择是 heapy，它自带 Guppy，一个用于 Python 环境下内存分析的库 。from guppy import hpy; hp=hpy() hp.doc.heap hp.heap() %run define.py Robot hp.heap()其他：https://pypi.python.org/pypi/memory_profilerhttp://mg.pov.lt/objgraph/https://launchpad.net/meliaehttp://pythonhosted.org/Pympler/muppy.htmlhttp://jmdana.github.io/memprof/提升 python 性能有一些方法可以更好地构建你的 Python 代码以提高性能。几个关键方法避免重复调用函数，尽可能用内置函数替代自己写的函数。import time x = 0 def doit1(i): global x x = x + i list = range(100000) t = time.time() for i in list: doit1(i) print("%.3f" % (time.time()-t))VSimport time x = 0 def doit2(list): global x for i in list: x = x + i list = range(100000) t = time.time() doit2(list) print("%.3f" % (time.time()-t))第二种写法要比第一种快。字符串处理：使用 "".join(list_of_strings) 而不是连续调用 +=使用列表推导式、列表生成器或 map() 代替 for 循环可以更快利用 C 扩展库，例如 Numpy 来进行快速数组操作。将耗时较长的代码重写为 C 模块。使用 ctypes、Cython、SWIG...等。内存管理有时候，推送内存的时间可能比进行计算的时间要长。所以对于大数据集要牢记：使用正确的数据结构使用高效算法使用生成器和迭代器，而不是列表。使用迭代器从数据库、sockets、文件、......中提取你需要的数据。
前言很久之前学过一点点爬虫，前几天需要爬一点资料所以又捡起来了。算是小需求所以也没用太复杂的框架，就 Requests+正则 就 ok 了。遗憾的是，需要爬的目标站是那种登录才可以见全文的那种网站，所以我得想办法获取 cookie。网上搜了一下，比较简单的方法有两种，一种是直接查看登录后网站存储的 cookie，然后解析使用。另一种是构造一个登录函数，登录后再获取 cookie。前者最为方便，但由于 cookie 会失效，所以需要及时获取新的 cookie，不适合自动化工作。后者不用担心时效性，但是构造一个合适的请求并成功登录还是有点难度的（至少我从来没有成功过），而且面对手机号 + 验证码登录的网站就抓瞎了。本文在将会介绍一个 python 库 browser_cookie3，只要在 Chrome 或者 Firefox 上成功登录，这个库就会自动抓取已经存储到本地的 cookie，尝试了一下还是挺好用的。当然，我们之前还介绍过 webscraper 这个 Chrome 插件，同样不用考虑 cookie 的问题，但是因为不太灵活，所以我也很少再用了。在文章开始前，我要提醒一下读者，本人对爬虫相关知识并不是非常了解，可能对 cookie 的解释有错误之处，仅供参考，如有不同意见可以在评论区交流。方法一：直接解析 cookie首先登录目标网站，然后按下 F12 打开开发者工具，点开 Network，找到目标网站的链接，点开之后往下拉，就可以看到 cookie 了。 然后进行解析，格式化为字典形式，代码如下。cookies = '复制过来的cookie' cookie = {i.split("=")[0]:i.split("=")[1] for i in cookies.split(";")} 然后就可以愉快 Requests 了headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36' } requests.get(url,headers=headers, cookies=cookie)方法二：模拟登录获取 cookie参考文章：https://blog.csdn.net/qlzy_5418/article/details/101056488https://zhuanlan.zhihu.com/p/27867925最简单的方案如下：def get_cookie(): url = 'https://www.xxx.com/signin' #登录网址 data = {'username': 'xxx', 'password': 'xxx'} headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36', } session = requests.session() cookie_jar = session.get(url=url, data=data, headers=headers).cookies cookie_t = requests.utils.dict_from_cookiejar(cookie_jar) return cookie_t但是一般而言是成功不了的，你还要考虑 data 的字段，post 还是 get 请求，以及动态变换的参数。可能使用 selenium 进行模拟登录还更简单一些。方法三：browser_cookie3GitHub 网站：https://github.com/borisbabic/browser_cookie3有 python2 版本，库名为 browser_cookie使用方法很简单，首先安装：pip install browser-cookie3然后登录目标网站最后获取 cookie：import browser_cookie3 import requests cj = browser_cookie3.chrome() # firefox可以替换为browser_cookie3.firefox() r = requests.get(url, cookies=cj)上面的代码是获取特定浏览器的 cookie，也可以获取所有浏览器的 cookie：cj = browser_cookie3.load()或者是获取指定网站的 cookie：browser_cookie3.chrome(domain_name='www.bitbucket.com')