Shu's Garden - Python
type: channel
link: https://www.sitstars.com/category/py/
description: 人生苦短，我用python
items: fastapi解决默认文档打不开的方案
items: 【python】给收藏的二次元图片都找到来源
items: Dash实现数据推送与实时更新
items: 数据存储格式选择：Parquet、HDF还是Feather？
items: 【python】实现一个简易的文本相似度检测程序
items: python web可视化框架Dash介绍与使用
items: 【python】制作一个百度收录检测API
items: fastapi获取访客真实ip
items: 【python教程】map、多进程与进度条
items: 【译】python性能分析与优化指南

fastapi解决默认文档打不开的方案
type: item
date: 2023-11-23T15:10:00+08:00
description: 前言今天久违地打开了了github仓库，发现shu-s-project居然有十五个star！又引起我写代码的兴趣了，于是clone下来准备先加几个API。调试过程中突然发现fastapi自带的docs文档打不开，控制台看了一下原来是静态资源请求失败，由于众所周知的原因，jsdelivr​已经被屏蔽了，自然导致docs打开一片空白。先在网上搜了一下解决方案，有的说直接把静态资源放到本地，有的说改fastapi源码，把CDN地址切换到其他可以访问的地址。这两种方案我都不喜欢，前者意味着可能每个项目都要搞一套静态资源，而且失去了CDN的意义（当然本地访问无所谓），后者更是下策，稍微升个级可能又失效了。Github上也有相关讨论，参见It is suggested to change the CDN of Swagger and Redoc、a parameter of FastAPI to alter base url of cdn.jsdelivr.net。经过考虑，我决定采用Beipy的方案，即直接在主程序中写一个函数，主动替换CDN地址，这样不用修改源码。代码如下：from fastapi import applications from fastapi.openapi.docs import get_swagger_ui_html def swagger_monkey_patch(*args, **kwargs): """ Wrap the function which is generating the HTML for the /docs endpoint and overwrite the default values for the swagger js and css. """ return get_swagger_ui_html( *args, **kwargs, swagger_js_url="https://unpkg.zhimg.com/swagger-ui-dist@3.29/swagger-ui-bundle.js", swagger_css_url="https://unpkg.zhimg.com/swagger-ui-dist@3.29/swagger-ui.css") # Actual monkey patch applications.get_swagger_ui_html = swagger_monkey_patch原理见https://fastapi.tiangolo.com/reference/openapi/docs/，docs和redoc都有专门的函数生成，其参数中有引用的CDN地址，那我们只需要把函数的参数改掉就行了。新的问题想必大家也注意到了，替换的js和css地址是有版本号的，事实上可能不同的fastapi版本用的是不同版本的静态资源，如果我们替换的版本号和fastapi实际使用的版本不一样会怎么样？试了一下，果然报错：​ 这就麻烦了，难道我每次都要去源码查看CDN版本，再手动替换？这样的话还不如用本地的方案呢。不过办法总比困难多，既然原始的get_swagger_ui_html函数给出了引用的CDN地址，那我们直接获取这个参数值，然后替换前缀不就行了（jsdelivr​和unpkg​基本上可以一一对应，所以我们只要replace即可，这里使用unpkg​，因为要比知乎的CDN全），顺着这个思路，很快代码就写好了。最终解决完整代码如下：import inspect def get_function_default_args(func): '''获取函数默认参数''' sign = inspect.signature(func) return { k: v.default for k, v in sign.parameters.items() if v.default is not inspect.Parameter.empty } def swagger_monkey_patch(*args, **kwargs): """ Wrap the function which is generating the HTML for the /docs endpoint and overwrite the default values for the swagger js and css. """ param_dict = get_function_default_args(get_swagger_ui_html) swagger_js_url = param_dict['swagger_js_url'].replace('https://cdn.jsdelivr.net/npm/', '…

【python】给收藏的二次元图片都找到来源
type: item
date: 2021-04-24T22:55:00+08:00
description: 前言之前我整理收藏的图片时，都是简单粗暴地直接全部重命名为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("S…

Dash实现数据推送与实时更新
type: item
date: 2021-04-02T00:43:00+08:00
description: 前言最近遇到了个需求，需要接收数据并且可视化展示出来，数据格式为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')这个方案其实比较完美了，而且也符合我对前后端分离的粗浅认知。但还是有些不足，…

数据存储格式选择：Parquet、HDF还是Feather？
type: item
date: 2021-03-23T00:18:00+08:00
description: 前言最近在写的策略框架需要对数据进行频繁读取，传统的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.…

【python】实现一个简易的文本相似度检测程序
type: item
date: 2021-01-28T19:02:46+08:00
description: 前言我手头存了不少小说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('./stopw…

python web可视化框架Dash介绍与使用
type: item
date: 2021-01-19T22:31:32+08:00
description: 前言之前工作的时候，需要写一个日内交易信号监控软件，预计的功能是实时更新数据，出现信号时有提示功能（比如颜色变化和声音提醒）。如果可能的话最好也有可视化功能。我第一时间就想到了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)获得和示例相同的样式。…

【python】制作一个百度收录检测API
type: item
date: 2021-01-16T17:39:00+08:00
description: 前言之所以想做这个 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=r…

fastapi获取访客真实ip
type: item
date: 2021-01-15T21:58:00+08:00
description: 最近一时兴起想好好搞一下我的 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、多进程与进度条
type: item
date: 2021-01-13T01:30:00+08:00
description: 今天讲讲我在实习中学到的一点 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_proc…

【译】python性能分析与优化指南
type: item
date: 2021-01-08T23:38:53+08:00
description: 这篇文章来自：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, nu…