谢乾坤 | Kingname

说到Python装饰器的执行顺序，有很多半吊子张口就来：

靠近函数名的装饰器先执行，远离函数名的装饰器后执行。

这种说法是不准确的。

ChatGPT的模型gpt-3.5-turbo发布当天，OpenAI还开源了一个语音转文本的模型：Whisper。但由于ChatGPT本身太过于耀眼，很多人都忽略了Whisper的存在。

我当时也是这样，我一度以为，Whisper也是一个API，需要发送POST请求到OpenAI的服务器上，然后它传回识别的结果。所以我很长一段时间一直都没有试用过这个模型。

直到前几天，我看到有人在少数派上面发了一篇文章，介绍他刚做的语音识别App，并且说这个App基于Whisper，完全不需要联网。我当时还奇怪，不联网你怎么调Whisper的API？于是我终于去认真了解了一下Whisper，发现它是OpenAI开源的语音转文字的模型，并不是API服务。这个模型只需要有Python就能本地离线运行，不需要联网。

ChatGPT一炮而红，让国内很多公司开始做大语言模型。然后他们很快就遇到了第一个问题，训练数据怎么来。有些公司去买数据，有些公司招聘爬虫工程师。但如果现在才开发爬虫，那恐怕已经来不及了。

即使爬虫工程师非常厉害，可以破解任意反爬虫机制，可以让爬虫跑满网络带宽，可是要训练出GPT-3这种规模的大语言模型，这个数据并不是一天两天就能爬完的。并且，有很多老网站的数据，早就被删除了，爬虫想爬也爬不到。

如果你看了今天这篇文章，那么恭喜你，你即将知道如何快速获取600亿网站的数据。从2008年开始爬取，这些网站数据横跨40多种语言。截止我写这篇文章的时候，最新的数据积累到了2023年2月。只要是Google现在或者曾经搜索得到的网站，你在这里都能找到。唯一制约你的，就是你的硬盘大小——仅仅2023年1月和2月的网页加到一起，就有400TB。而且所有这些数据，完全免费！不需要注册登录！不需要梯子！不需要下载任何额外软件！只需要浏览器或者Linux中的wget命令就能直接下载。

去年我写了一篇文章：一日一技：如何捅穿Cloud Flare的5秒盾 ，这篇文章使用的第三方库『cloudscraper』可以绕过免费版的五秒盾。但遇到付费版就无能为力了。

如果你在Flask中启动过子线程，然后在子线程中读写过g对象或者尝试从request对象中读取url参数，那么，你肯定对下面这个报错不陌生：RuntimeError: Working outside of request context..

例如下面这段Flask代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import threading
from flask import Flask, request

app = Flask(__name__)

def inner_func():
    doc_id = request.args.get('doc_id', '')
    print(f'用户ID为：{doc_id}')


@app.route('/start_thread')
def start_thread():
    thread = threading.Thread(target=inner_func)
    thread.start()
    return {'success': True, 'msg': '获取用户ID成功！'}

请求/start_thread接口就会报错，如下图所示：

如果你在网上搜索flask thread RuntimeError: Working outside of request context. ，那么你可能会看到官方文档或者StackOverFlow上面提供了一个装饰器@copy_current_request_context。如下图所示：

照着它这样写，确实能解决问题，如下图所示：

但无论是官网还是StackOverFlow，它的例子都非常简单。但是我们知道，启动线程有很多种方法，例如：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 方法一，启动简单线程
import threading

job = threading.Thread(target=函数名, args=(参数1, 参数2), kwargs={'参数3': xxx, '参数4': yyy})
job.start()

# 方法2，使用类定义线程
import threading

class Job(threading.Thread):
    def __init__(self, 参数):
        super().__init__()
    
    def run(self):
        print('子线程开始运行')

job = Job(参数)
job.start()

# 方法3，使用线程池
from multiprocessing.dummy import Pool
pool = Pool(5)  # 5个线程的线程池
pool.map(函数名, 参数列表)

网上的方法只能解决第一种写法的问题。如果想使用方法2和方法3启动子线程，代码应该怎么写呢？如果在子线程中又启动子线程，再用一次@copy_current_request_context还行吗？

相信我，你在网上搜索一下午，只有两种结果：一是找不到答案，二是找到的答案是晚于2023年1月14日的，因为是别人看了我这篇文章以后，再写的。

我们知道，在软件工程中，单元测试是保证软件质量的重要手段之一。一个优秀的代码，单元测试的代码量，经常会超过被测试的代码本身。一个理想化的开发团队，可能有三分之二的时间是在写测试，剩下的三分之一时间才是写业务代码。

在以前的文章中，在微信群中，我多次强调，写函数的时候，不要把所有参数放到一个大字典里面作为参数到处传，否则时间久了以后，根本不知道字典里面有哪些数据：

1
2
3
4

def parse(data):
    name = data['name']
    age = data['age']
    xxx = data['xx']

上面这样写，对原作者来说确实简单，但是如果代码还有别人来维护，他就根本不知道这个字典里面有哪些数据。必须要一层一层查找调用链，费时费力。

我们有时候可能会需要把一个字符串转换成对应的类型。例如，把'123'转换为int类型的123；或者把'3.14'转成浮点数3.14。

今天有同学在公众号粉丝群问了这样一个问题：

到今天就满30岁了。