conda remove --name <env_name> --all
# or
conda env remove --name <env_name>
conda env list
# 将环境名称存储在一个文本文件中，每行一个名称。
cat env_list.txt | while read env ; do conda remove --name $env --all -y ; done
# or
cat env_list.txt | while read env ; do conda env remove --name $env -y ; done
conda env list | awk -F' ' '{print $1}' | while read env ; do conda remove --name $env --all -y ; done
conda install anaconda-clean
anaconda-clean
rm -rf ~/anaconda3
# 在终端输入删除nano编辑器的命令：
sudo apt-get remove nanosudo adduser tonytan
chmod -R 640 /etc/sudoers
sudo vim /etc/sudoers
chmod -R 440 /etc/sudoers
或者使用
sudo visudo退出并用tonytan登录
wget -O install.sh https://download.bt.cn/install/install-ubuntu_6.0.sh && sudo bash install.sh ed8484becssh-keygen -t rsa -b 2048 -C "admin@example.com"
cat .ssh/id_rsa.pub
ssh-keygen -t rsa -b 2048 -C "U_Recommend@51yjjf.com"
cat .ssh/id_rsa.pubwget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
bash Miniconda3-py310_22.11.1-1-Linux-x86_64.sh
source ../.bashrc
conda create -n fastapi python=3.9pip install -r requestment.txt -i https://pypi.tuna.tsinghua.edu.cn/simple/
cd /www/wwwroot/voucher_shubi_apis/
pip install "fastapi[all]"
pip install gunicorn
pip install pyopenssl
pip install python-jose[cryptography]sudo apt-get install supervisor
sudo apt-get install redis-server
sudo apt-get install gunicornsudo vim /etc/supervisor/conf.d/voucher.conf
[program: voucher_8000]
command=/home/tonytan/miniconda3/envs/fastapi/bin/gunicorn main:app -w 4 --worker-class uvicorn.workers.UvicornWorker -b 0.0.0.0:8000
command=/home/tonytan/miniconda3/envs/voucher/bin/gunicorn -w 1 -b 0.0.0.0:8202 activity_voucher_admin.wsgi
command=/home/tonytan/miniconda3/envs/fastapi/bin/gunicorn main:app -c gunicorn.conf
directory=/www/wwwroot/activity_voucher_apis
user = root
redirect_stderr=true
stdout_logfile = /var/log/supervisor/stdout.log
stderr_logfile = /var/log/supervisor/stderr.log
environment=LANG=en_US.UTF-8,LC_ALL=en_US.UTF-8
[program:voucher_huey]
command=/home/tonytan/miniconda3/envs/voucher/bin/python3 manage.py run_huey -w 1
directory = /www/wwwroot/activity_voucher_admin
user = root
redirect_stderr=true
stdout_logfile = /var/log/supervisor/voucher_stdout.log
stderr_logfile = /var/log/supervisor/voucher_stderr.log
environment=DOMAIN=pro
# 并行工作进程数
workers = 4
# 指定每个工作者的线程数
threads = 2
# 监听内网端口5000
bind = '127.0.0.1:5000'
# 设置守护进程,将进程交给supervisor管理
daemon = 'false'
# 工作模式协程
worker_class = 'gevent'
# 设置最大并发量
worker_connections = 2000
# 设置进程文件目录
pidfile = '/var/run/gunicorn.pid'
# 设置访问日志和错误信息日志路径
accesslog = '/var/log/gunicorn_acess.log'
errorlog = '/var/log/gunicorn_error.log'
# 设置日志记录水平
loglevel = 'warning'
wenzhou.trust-will.com
upstream wenzhou_apis {
least_conn;
server 127.0.0.1:8000;
server 127.0.0.1:8001;
server 127.0.0.1:8002;
server 127.0.0.1:8003;
}
location / {
proxy_pass http://wenzhou_apis;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
sudo vim /home/tonytan/miniconda3/envs/fastapi/lib/python3.9/site-packages/fastapi/openapi/docs.py
注释20·22行
增加：
swagger_js_url: str = "/static/swagger-ui/swagger-ui-bundle.js",
swagger_css_url: str = "/static/swagger-ui/swagger-ui.css",
swagger_favicon_url: str = "/static/swagger-ui/favicon-32x32.png",
注释82-83行
增加：
redoc_js_url: str = "/static/redoc/bundles/redoc.standalone.js",
redoc_favicon_url: str = "/static/redoc/favicon.png",
uvicorn main:app --reload
浏览器打开 127.0.0.1:800/openapi.json
将文件另存到 ./static/openapi.json
git更新提交
CSRF_TRUSTED_ORIGINS = ['https://www.02405.com', 'https://02405.com'], 主要要加http或https前缀# jwt卸载命令
pip uninstall jwt
# 保险起见，将PyJWT一同卸载
pip uninstall PyJWT
# 重新安装PyJWT
pip install PyJWT
| # 视频优化器 | |
| # 通过使用以下自动化脚本，你不仅可以使用 Python 来优化视频，还可以使用它来优化图像。该脚本使用 Moviepy 模块，允许你修剪、添加音频、设置视频速度、添加 VFX 等等。 | |
| # Video Optimizer | |
| # pip install moviepy | |
| import moviepy.editor as pyedit | |
| # Load the Video | |
| video = pyedit.VideoFileClip("vid.mp4") | |
| # Trimming | |
| vid1 = video.subclip(0, 10) | |
| vid2 = video.subclip(20, 40) | |
| final_vid = pyedit.concatenate_videoclips([vid1, vid2]) | |
| # Speed up the video | |
| final_vid = final_vid.speedx(2) | |
| # Adding Audio to the video | |
| aud = pyedit.AudioFileClip("bg.mp3") | |
| final_vid = final_vid.set_audio(aud) | |
| # Reverse the Video | |
| final_vid = final_vid.fx(pyedit.vfx.time_mirror) | |
| # Merge two videos | |
| vid1 = pyedit.VideoFileClip("vid1.mp4") | |
| vid2 = pyedit.VideoFileClip("vid2.mp4") | |
| final_vid = pyedit.concatenate_videoclips([vid1, vid2]) | |
| # Add VFX to Video | |
| vid1 = final_vid.fx(pyedit.vfx.mirror_x) | |
| vid2 = final_vid.fx(pyedit.vfx.invert_colors) | |
| final_vid = pyedit.concatenate_videoclips([vid1, vid2]) | |
| # Add Images to Video | |
| img1 = pyedit.ImageClip("img1.jpg") | |
| img2 = pyedit.ImageClip("img2.jpg") | |
| final_vid = pyedit.concatenate_videoclips([img1, img2]) | |
| # Save the video | |
| final_vid.write_videofile("final.mp4") |
| from pdf2image import convert_from_path | |
| import os | |
| pwd_file = "" | |
| img_file = "" | |
| # file_path = "/Users/tonytan/Nustore Files/我的坚果云/创斯维/发票/20220406" | |
| file_path = "/Users/tonytan/Nustore Files/我的坚果云/Jobs/2020-06 集森/发票/20220406/" | |
| pdf_files = [] | |
| print(os.listdir(file_path)) | |
| for file in os.listdir(file_path): | |
| print(file.split(".")[-1]) | |
| full_file = os.path.join(file_path, file) | |
| print(os.path.isfile(full_file)) | |
| if os.path.isfile(full_file) and file.split(".")[-1].lower() == "pdf": | |
| pdf_files.append(file) | |
| print(pdf_files) | |
| img_file = os.path.join(file_path, "img") | |
| def conv_pdf(file): | |
| pdf_file = os.path.join(file_path, file) | |
| print(pdf_file) | |
| filename = file.split(".")[0] | |
| print(file) | |
| pages = convert_from_path(pdf_file, 500) | |
| for i, page in enumerate(pages): | |
| img_filename = f"{filename}_{i}.jpg" | |
| img_path = os.path.join(img_file, img_filename) | |
| page.save(img_path, "JPEG") | |
| for file in pdf_files: | |
| conv_pdf(file) |
| # 图片优化器 | |
| #使用这个很棒的自动化脚本，可以帮助把图像处理的更好，你可以像在 Photoshop 中一样编辑它们。 | |
| # 该脚本使用流行的是 Pillow 模块 | |
| # Image Optimizing | |
| # pip install Pillow | |
| import PIL | |
| # Croping | |
| im = PIL.Image.open("Image1.jpg") | |
| im = im.crop((34, 23, 100, 100)) | |
| # Resizing | |
| im = PIL.Image.open("Image1.jpg") | |
| im = im.resize((50, 50)) | |
| # Flipping | |
| im = PIL.Image.open("Image1.jpg") | |
| im = im.transpose(PIL.Image.FLIP_LEFT_RIGHT) | |
| # Rotating | |
| im = PIL.Image.open("Image1.jpg") | |
| im = im.rotate(360) | |
| # Compressing | |
| im = PIL.Image.open("Image1.jpg") | |
| im.save("Image1.jpg", optimize=True, quality=90) | |
| # Bluring | |
| im = PIL.Image.open("Image1.jpg") | |
| im = im.filter(PIL.ImageFilter.BLUR) | |
| # Sharpening | |
| im = PIL.Image.open("Image1.jpg") | |
| im = im.filter(PIL.ImageFilter.SHARPEN) | |
| # Set Brightness | |
| im = PIL.Image.open("Image1.jpg") | |
| im = PIL.ImageEnhance.Brightness(im) | |
| im = im.enhance(1.5) | |
| # Set Contrast | |
| im = PIL.Image.open("Image1.jpg") | |
| im = PIL.ImageEnhance.Contrast(im) | |
| im = im.enhance(1.5) | |
| # Adding Filters | |
| im = PIL.Image.open("Image1.jpg") | |
| im = PIL.ImageOps.grayscale(im) | |
| im = PIL.ImageOps.invert(im) | |
| im = PIL.ImageOps.posterize(im, 4) | |
| # Saving | |
| im.save("Image1.jpg") |
| # PDF 转图片 | |
| # 这个小型自动化脚本可以方便地获取整个 PDF 页面并将它们转换为图像。该脚本使用流行的 PyMuPDF 模块，该模块以其 PDF 文本提取而闻名。 | |
| # PDF to Images | |
| # pip install PyMuPDF | |
| import fitz | |
| def pdf_to_images(pdf_file): | |
| doc = fitz.open(pdf_file) | |
| for p in doc: | |
| pix = p.get_pixmap() | |
| output = f"page{p.number}.png" | |
| pix.writePNG(output) | |
| pdf_to_images("test.pdf") |
| # PySide2 GUI | |
| # 这个自动化脚本将帮助你使用 PySide2 Gui 模块创建你的 GUI 应用程序。你可以在下面找到开始开发体面的现代应用程序所需的每种方法。 | |
| # PySide 2 | |
| # pip install PySide2 | |
| from PySide6.QtWidgets import * | |
| from PySide6.QtGui import * | |
| import sys | |
| app = QApplication(sys.argv) | |
| window = QWidget() | |
| # Resize the Window | |
| window.resize(500, 500) | |
| # Set the Window Title | |
| window.setWindowTitle("PySide2 Window") | |
| # Add Buttons | |
| button = QPushButton("Click Me", window) | |
| button.move(200, 200) | |
| # Add Label Text | |
| label = QLabel("Hello Medium", window) | |
| label.move(200, 150) | |
| # Add Input Box | |
| input_box = QLineEdit(window) | |
| input_box.move(200, 250) | |
| print(input_box.text()) | |
| # Add Radio Buttons | |
| radio_button = QRadioButton("Radio Button", window) | |
| radio_button.move(200, 300) | |
| # Add Checkbox | |
| checkbox = QCheckBox("Checkbox", window) | |
| checkbox.move(200, 350) | |
| # Add Slider | |
| slider = QSlider(window) | |
| slider.move(200, 400) | |
| # Add Progress Bar | |
| progress_bar = QProgressBar(window) | |
| progress_bar.move(200, 450) | |
| # Add Image | |
| image = QLabel(window) | |
| image.setPixmap(QPixmap("image.png")) | |
| # Add Message Box | |
| msg = QMessageBox(window) | |
| msg.setText("Message Box") | |
| msg.setStandardButtons(QMessageBox.Ok | QMessageBox.Cancel) | |
| window.show() | |
| sys.exit(app.exec()) |
Python有许多内置函数，共有66个。以下是这些内置函数的详细解释和示例代码：
x = -10
print(abs(x)) # 输出：10
iterable = [True, True, False]
print(all(iterable)) # 输出：False
iterable = [False, False, True]
print(any(iterable)) # 输出：True
x = 10
print(bin(x)) # 输出：0b1010
x = 0
print(bool(x)) # 输出：False
source = b'Hello'
arr = bytearray(source)
print(arr) # 输出：bytearray(b'Hello')
source = 'Hello'
b = bytes(source, encoding='utf-8')
print(b) # 输出：b'Hello'
def func():
pass
print(callable(func)) # 输出：True
i = 65
print(chr(i)) # 输出：A
class MyClass:
attr = 10
@classmethod
def class_method(cls):
print(cls.attr)
MyClass.class_method() # 输出：10
source = "print('Hello, World!')"
code = compile(source, filename="", mode="exec")
exec(code) # 输出：Hello, World!
real = 3
imag = 4
c = complex(real, imag)
print(c) # 输出：(3+4j)
class MyClass:
attr = 10
obj = MyClass()
delattr(obj, 'attr')
print(hasattr(obj, 'attr')) # 输出：False
d = dict(a=1, b=2, c=3)
print(d) # 输出：{'a': 1, 'b': 2, 'c': 3}
class MyClass:
attr = 10
def method(self):
pass
obj = MyClass()
print(dir(obj))
a = 10
b = 3
result = divmod(a, b)
print(result) # 输出：(3, 1)
iterable = ['a', 'b', 'c']
for i, value in enumerate(iterable):
print(i, value)
# 输出：
# 0 a
# 1 b
# 2 c
expression = "2 + 3"
result = eval(expression)
print(result) # 输出：5
code ="""
x = 5
if x > 0:
print("Positive")
else:
print("Non-positive")
"""
exec(code) # 输出：Positive
def is_positive(x):
return x > 0
numbers = [1, -2, 3, -4, 5]
result = filter(is_positive, numbers)
print(list(result)) # 输出：[1, 3, 5]
x = 10
print(float(x)) # 输出：10.0
value = 3.14159
result = format(value, ".2f")
print(result) # 输出：3.14
iterable = [1, 2, 3]
fs = frozenset(iterable)
print(fs) # 输出：frozenset({1, 2, 3})
class MyClass:
attr = 10
obj = MyClass()
print(getattr(obj, 'attr')) # 输出：10
print(globals())
class MyClass:
attr = 10
obj = MyClass()
print(hasattr(obj, 'attr')) # 输出：True
x = 10
print(hash(x))
help(list)
x = 255
print(hex(x)) # 输出：0xff
x = 10
print(id(x))
name = input("请输入您的姓名：")
print("您好，" + name + "！")
x = 3.14
print(int(x)) # 输出：3
class MyClass:
pass
obj = MyClass()
print(isinstance(obj, MyClass)) # 输出：True
class Parent:
pass
class Child(Parent):
pass
print(issubclass(Child, Parent)) # 输出：True'
iterable = [1, 2, 3]
iterator = iter(iterable)
print(next(iterator)) # 输出：1
s = "Hello"
print(len(s)) # 输出：5
iterable = (1, 2, 3)
lst = list(iterable)
print(lst) # 输出：[1, 2, 3]
print(locals())
def square(x):
return x ** 2
numbers = [1, 2, 3, 4, 5]
result = map(square, numbers)
print(list(result)) # 输出：[1, 4, 9,16, 25]
numbers = [3, 1, 4, 2, 5]
print(max(numbers)) # 输出：5
lst = [1, 2, 3, 4, 5]
mv = memoryview(lst)
print(mv[0]) # 输出：1
numbers = [3, 1, 4, 2, 5]
print(min(numbers)) # 输出：1
iterable = [1, 2, 3]
iterator = iter(iterable)
print(next(iterator)) # 输出：1
44.object(): 返回一个新的空对象。
obj = object()
print(obj)
45.oct(x): 将一个整数转换为八进制字符串。
x = 8
print(oct(x)) # 输出：0o10
46.open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None): 打开一个文件，并返回文件对象。
file = open("example.txt", "w")
file.write("Hello, World!")
file.close()
47.ord(c): 返回一个字符的Unicode编码。
c = 'A'
print(ord(c)) # 输出：65
48.pow(x, y[, z]): 返回x的y次幂，如果提供z，则返回x的y次幂对z取模的结果。
x = 2
y = 3
z = 5
result = pow(x, y, z)
print(result) # 输出：3
49.print(objects, sep=' ', end='\n', file=sys.stdout, flush=False): 打印输出到控制台。
name = "Alice"
age = 25
print("My name is", name, "and I am", age, "years old.")
# 输出：My name is Alice and I am 25 years old.
50.property(fget=None, fset=None, fdel=None, doc=None): 创建一个属性。
class MyClass:
def __init__(self):
self._attr = 0
@property
def attr(self):
return self._attr
@attr.setter
def attr(self, value):
self._attr = value
obj = MyClass()
obj.attr = 10
print(obj.attr) # 输出：10
51.range(stop): 返回一个包含从0到stop-1的整数序列的可迭代对象。
for i in range(5):
print(i)
# 输出：
# 0
# 1
# 2
# 3
# 4
52.repr(object): 返回一个对象的字符串表示形式。
s = "Hello"
print(repr(s)) # 输出：'Hello'
**53.reversed(seq): **返回一个反向的迭代器对象。
seq = [1, 2, 3]
rev_seq = reversed(seq)
print(list(rev_seq)) # 输出：[3, 2, 1]
54.round(number[, ndigits]): 对一个数进行四舍五入。
x = 3.14159
print(round(x, 2)) # 输出：3.14
55.set([iterable]): 创建一个集合。
iterable = [1, 2, 3]
s = set(iterable)
print(s) # 输出：{1, 2, 3}
56.setattr(object, name, value): 设置对象的属性值。
class MyClass:
attr = 10
obj = MyClass()
setattr(obj, 'attr', 20)
print(obj.attr) # 输出：20
57.slice(stop): 创建一个切片对象，用于切片操作。
numbers = [1, 2, 3, 4, 5]
s = slice(2)
print(numbers[s]) # 输出：[1, 2]
58.sorted(iterable[, key][, reverse]): 返回一个排序后的列表。
numbers = [3, 1, 4, 2, 5]
sorted_numbers = sorted(numbers)
print(sorted_numbers) # 输出：[1, 2, 3, 4, 5]
59.staticmethod(function): 将一个函数转换为静态方法。
class MyClass:
@staticmethod
def my_method():
print("This is a static method.")
MyClass.my_method() # 输出：This is a static method.
60.str(object=''): 将一个对象转换为字符串。
x = 10
print(str(x)) # 输出：'10'
61.sum(iterable[, start]): 返回可迭代对象中所有元素的总和。
numbers = [1, 2, 3, 4, 5]
print(sum(numbers)) # 输出：15
62.super([type[, object-or-type]]): 返回父类的对象。
class Parent:
def __init__(self):
self.attr = 10
class Child(Parent):
def __init__(self):
super().__init__()
child = Child()
print(child.attr) # 输出：10
63.tuple([iterable]): 创建一个元组。
iterable = [1, 2, 3]
t = tuple(iterable)
print(t) # 输出：(1, 2, 3)
64.type(object): 返回一个对象的类型。
x = 10
print(type(x)) # 输出：<class 'int'>
65.vars([object]): 返回对象的属性和属性值的字典。
class MyClass:
attr = 10
obj = MyClass()
print(vars(obj)) # 输出：{'attr': 10}
*66.zip(iterables): 将多个可迭代对象中对应位置的元素打包成元组，并返回一个由这些元组组成的可迭代对象。
numbers = [1, 2, 3]
letters = ['a', 'b', 'c']
zipped = zip(numbers, letters)
print(list(zipped)) # 输出：[(1, 'a'), (2, 'b'), (3, 'c')]
| # 获取 API 数据 | |
| # 需要从数据库中获取 API 数据或需要向服务器发送 API 请求。那么这个自动化脚本对你来说是一个方便的工具。使用 Urllib3 模块，可让你获取和发布 API 请求。 | |
| # pip install urllib3 | |
| import urllib3 | |
| # Fetch API data | |
| url = "https://api.github.com/users/psf/repos" | |
| http = urllib3.PoolManager() | |
| response = http.request('GET', url) | |
| print(response.status) | |
| print(response.data) | |
| # Post API data | |
| url = "https://httpbin.org/post" | |
| http = urllib3.PoolManager() | |
| response = http.request('POST', url, fields={'hello': 'world'}) | |
| print(response.status) |
| # 互联网下载器 | |
| # 你们可能使用下载软件从 Internet 下载照片或视频，但现在你可以使用 Python IDM 模块创建自己的下载器。 | |
| # Python Downloader | |
| # pip install internetdownloadmanager | |
| import internetdownloadmanager as idm | |
| def Downloader(url, output): | |
| pydownloader = idm.Downloader(worker=20, | |
| part_size=1024*1024*10, | |
| resumable=True,) | |
| pydownloader .download(url, output) | |
| Downloader("Link url", "image.jpg") | |
| Downloader("Link url", "video.mp4") |
| # 拼写修正 | |
| # 这个很棒的脚本将帮助你纠正你的文本单词拼写错误。你可以在下面找到脚本，将告诉你如何修复句子中的单个单词或多个单词。 | |
| # Spell Fixer | |
| # pip install textblob | |
| from textblob import * | |
| # Fixing Paragraph Spells | |
| def fix_paragraph_words(paragraph): | |
| sentence = TextBlob(paragraph) | |
| correction = sentence.correct() | |
| print(correction) | |
| # Fixing Words Spells | |
| def fix_word_spell(word): | |
| word = Word(word) | |
| correction = word.correct() | |
| print(correction) | |
| fix_paragraph_words("This is sammple tet!!") | |
| fix_word_spell("maangoo") |
| # 电池指示灯 | |
| # 这个方便的脚本可以让你设置你想要得到通知的电池百分比，该脚本使用 Pyler 进行通知，使用 Psutil 获取当前的电池百分比。 | |
| # Battery Notifier | |
| # pip instal plyer | |
| from plyer import notification | |
| import psutil | |
| from time import sleep | |
| while True: | |
| battery = psutil.sensors_battery() | |
| life = battery.percent | |
| if life < 50: | |
| notification.notify( | |
| title = "Battery Low", | |
| message = "Please connect to power source", | |
| timeout = 10 | |
| ) | |
| sleep(60) | |
from fn import F
print((F() ** 2) >> (F() + 1))
output = (F() ** 2).then(F() + 1)
assert 16 == output(3)
# 这个管道操作首先对输入3平方得到9，然后加1得到10，最后的结果是10.from pipe import Pipe
@Pipe
def even_filter(nums):
for num in nums:
if num % 2 == 0:
yield num
@Pipe
def multiply_by_three(nums):
for num in nums:
yield num * 3
@Pipe
def convert_to_string(nums):
for num in nums:
yield 'The Number: %s' % num
# 使用
nums = range(10)
print(nums | even_filter | multiply_by_three | convert_to_string)from toolz import pipe
# 定义一系列函数
def add(x, y):
return x + y
def square(x):
return x ** 2
def double(x):
return x * 2
# 创建一个计算管道
calculation_pipeline = pipe(add, square, double)
# 使用计算管道进行计算
result = calculation_pipeline(2, 3)
print(result) # 输出：50，等价于 double(square(add(2, 3)))from toolz import compose
# 定义两个简单的函数
def add(x, y):
return x + y
def square(x):
return x ** 2
# 使用compose将这两个函数组合
composed_function = compose(square, add)
# 调用组合函数
result = composed_function(2, 3)
print(result) # 输出：25，等价于 square(add(2, 3))一个程序运行起来后，一定有一个执行代码的东西，这个东西就是线程； 一般计算(CPU)密集型任务适合多进程，IO密集型任务适合多线程； 一个进程可拥有多个并行的(concurrent)线程，当中每一个线程，共享当前进程的资源
以下是对发现的几种多线程进行的汇总整理，均已测试运行 多线程实现的四种方式分别是： multiprocessing下面有两种：
from multiprocessing.dummy import Pool as ThreadPool # 线程池
from multiprocessing.pool import ThreadPool # 线程池，用法无区别，唯一区别这个是线程池
另外两种：
from concurrent.futures import ThreadPoolExecutor # python原生线程池，这个更主流
import threadpool # 线程池，需要 pip install threadpool，很早之前的
非阻塞方法
multiprocessing.dummy.Pool.apply_async() 和 multiprocessing.dummy.Pool.imap()
线程并发执行
阻塞方法
multiprocessing.dummy.Pool.apply()和 multiprocessing.dummy.Pool.map()
线程顺序执行
from multiprocessing.dummy import Pool as Pool
import time
def func(msg):
print('msg:', msg)
time.sleep(2)
print('end:')
pool = Pool(processes=3)
for i in range(1, 5):
msg = 'hello %d' % (i)
pool.apply_async(func, (msg,)) # 非阻塞，子线程有返回值
# pool.apply(func,(msg,)) # 阻塞,apply()源自内建函数，用于间接的调用函数，并且按位置把元祖或字典作为参数传入。子线程无返回值
# pool.imap(func,[msg,]) # 非阻塞, 注意与apply传的参数的区别 无返回值
# pool.map(func, [msg, ]) # 阻塞 子线程无返回值
print('Mark~~~~~~~~~~~~~~~')
pool.close()
pool.join() # 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
print('sub-process done')
from multiprocessing.pool import ThreadPool # 线程池，用法无区别，唯一区别这个是线程池
from multiprocessing.dummy import Pool as ThreadPool # 线程池
import os
import time
print("hi outside of main()")
def hello(x):
print("inside hello()")
print("Proccess id: %s" %(os.getpid()))
time.sleep(3)
return x*x
if __name__ == "__main__":
p = ThreadPool(5)
pool_output = p.map(hello, range(3))
print(pool_output)
from concurrent.futures import ThreadPoolExecutor
import threading
import time
# 定义一个准备作为线程任务的函数
def action(max):
my_sum = 0
for i in range(max):
print(threading.current_thread().name + ' ' + str(i))
my_sum += i
return my_sum
# 创建一个包含2条线程的线程池
pool = ThreadPoolExecutor(max_workers=2)
# 向线程池提交一个task, 20会作为action()函数的参数
future1 = pool.submit(action, 20)
# 向线程池再提交一个task, 30会作为action()函数的参数
future2 = pool.submit(action, 30)
# 判断future1代表的任务是否结束
print(future1.done())
time.sleep(3)
# 判断future2代表的任务是否结束
print(future2.done())
# 查看future1代表的任务返回的结果
print(future1.result())
# 查看future2代表的任务返回的结果
print(future2.result())
# 关闭线程池
pool.shutdown()
需要 pip install threadpool
import threadpool
def hello(m, n, o):
""""""
print("m = %s, n = %s, o = %s" % (m, n, o))
if __name__ == '__main__':
# 方法1
# lst_vars_1 = ['1', '2', '3']
# lst_vars_2 = ['4', '5', '6']
# func_var = [(lst_vars_1, None), (lst_vars_2, None)]
# 方法2
dict_vars_1 = {'m': '1', 'n': '2', 'o': '3'}
dict_vars_2 = {'m': '4', 'n': '5', 'o': '6'}
func_var = [(None, dict_vars_1), (None, dict_vars_2)]
# 定义了一个线程池，表示最多可以创建poolsize这么多线程
pool = threadpool.ThreadPool(2)
# 调用makeRequests创建了要开启多线程的函数，以及函数相关参数和回调函数，其中回调函数可以不写
requests = threadpool.makeRequests(hello, func_var)
[pool.putRequest(req) for req in requests] # 将所有要运行多线程的请求扔进线程池
pool.wait() # 等待所有线程完成工作后退出
"""
[pool.putRequest(req) for req in requests]等同于
for req in requests:
pool.putRequest(req)
"""
| # 获取世界新闻 | |
| # 使用此自动化脚本让你随时了解每日世界新闻，你可以使用任何语言从任何国家/地区获取新闻。这个 API 让你每天免费获取 50 篇新闻文章。 | |
| # World News Fetcher | |
| # pip install requests | |
| import requests | |
| ApiKey = "YOUR_API_KEY" | |
| url = "https://api.worldnewsapi.com/search-news?text=hurricane&api-key={ApiKey}" | |
| headers = { | |
| 'Accept': 'application/json' | |
| } | |
| response = requests.get(url, headers=headers) | |
| print("News: ", response.json()) |
厌倦了校对你的长文章或文本，然后，你可以试试这个自动化脚本，它将扫描你的文本并纠正语法错误，这个很棒的脚本使用 Happtransformer 模块，这是一个机器学习模块，经过训练可以修复文本中的语法错误。
pip install happytransformer
from happytransformer import HappyTextToText as HappyTTT
from happytransformer import TTSettings
def Grammer_Fixer(Text):
Grammer = HappyTTT("T5","prithivida/grammar_error_correcter_v1")
config = TTSettings(do_sample=True, top_k=10, max_length=100)
corrected = Grammer.generate_text(Text, args=config)
print("Corrected Text: ", corrected.text)
Text = "This is smple tet we how know this"
Grammer_Fixer(Text)* pip install pyopenssl==22.1
* except (ValueError, IndexError):
ERROR: Could not find a version that satisfies the requirement anyjson>=0.3.3 (from kombu) (from versions: 0.1, 0.2.0, 0.2.1, 0.2.2, 0.2.3, 0.2.4, 0.2.5, 0.3, 0.3.1, 0.3.2, 0.3.3)
ERROR: No matching distribution found for anyjson>=0.3.3
* 降低setuptools版本于50一下
pip install setuptools==50.3.2
import OpenSSL
File "/home/ubuntu/miniconda3/envs/version/lib/python3.10/site-packages/OpenSSL/__init__.py", line 8, in <module>
from OpenSSL import crypto, SSL
File "/home/ubuntu/miniconda3/envs/version/lib/python3.10/site-packages/OpenSSL/crypto.py", line 1517, in <module>
class X509StoreFlags(object):
File "/home/ubuntu/miniconda3/envs/version/lib/python3.10/site-packages/OpenSSL/crypto.py", line 1537, in X509StoreFlags
CB_ISSUER_CHECK = _lib.X509_V_FLAG_CB_ISSUER_CHECK
AttributeError: module 'lib' has no attribute 'X509_V_FLAG_CB_ISSUER_CHECK'
pip install pyopenssl --upgrade
from Cryptodome.Signature import pkcs1_15
ModuleNotFoundError: No module named 'Cryptodome'
pip install pycryptodomex
Django提示mysq版本Q过低: django db. utils NotSuppotedError: MySQL 8 or later is required (found 5.7.26). 因为mysql5.7及以下是免费的数据库，8.0之后是收费的。使用免费的数据库更放心,使用上没有什么区别。这个提示只是Django的版本 检测提示，把它注释掉就好了。
全局搜索函数:
在pycharm中 搜索路径 :
django\db\backends\base\base.py "小提示pycharm中搜索快捷键为 Ctrl+shift+n 苹果系统除外"
把如下代码注释了就行
def init_connection_state(self):
Initialize the database connection settings.
global RAN_ DB_ VERSION_ CHECK
if self.alias not in RAN_ DB_ VERSION_ CHECK:
# self. check_ database_ version_ supported( )
RAN_ DB_ VERSION_ CHECK . add(self .alias )
"""就注释一句话就行"""
之后就可以重新运行项目了
| # -*- coding: UTF-8 -*- | |
| import socket | |
| import base64 | |
| import sys | |
| import time | |
| import datetime | |
| import json | |
| import hmac | |
| from hashlib import sha1 as sha | |
| from common.handle.oss import httpserver | |
| # 请填写您的AccessKeyId。 | |
| access_key_id = 'LTAI5tAqg9J6FgZcdyz6unV6' | |
| # 请填写您的AccessKeySecret。 | |
| access_key_secret = 'Zd9CiwWN3liIS6bceMAjtXj4mgIhhl' | |
| # host的格式为 bucketname.endpoint ，请替换为您的真实信息。 | |
| host = 'http://voucher-trust-will.oss-cn-hangzhou.aliyuncs.com' | |
| # callback_url为 上传回调服务器的URL，请将下面的IP和Port配置为您自己的真实信息。 | |
| callback_url = "http://voucher.trust-will.com/car_register_v2/oss_callback" | |
| # 用户上传文件时指定的前缀。 | |
| upload_dir = 'car-register/' | |
| expire_time = 30 | |
| bucket_name = "voucher-trust-will" | |
| def get_iso_8601(expire): | |
| gmt = datetime.datetime.utcfromtimestamp(expire).isoformat() | |
| gmt += 'Z' | |
| return gmt | |
| def get_token(): | |
| now = int(time.time()) | |
| expire_syncpoint = now + expire_time | |
| expire = get_iso_8601(expire_syncpoint) | |
| policy_dict = {} | |
| policy_dict['expiration'] = expire | |
| policy_dict['conditions'] = [['eq', '$bucket', bucket_name], ['content-length-range', 0, 104857600]] | |
| policy = json.dumps(policy_dict).strip() | |
| policy_encode = base64.b64encode(policy.encode('utf-8')) | |
| h = hmac.new(access_key_secret.encode('utf-8'), policy_encode, sha) | |
| policy_encode = policy_encode.decode('utf-8') | |
| sign_result = base64.encodebytes(h.digest()).strip().decode('utf-8') | |
| callback_dict = {} | |
| callback_dict['callbackUrl'] = callback_url | |
| callback_dict['callbackBody'] = 'filename=${object}&size=${size}&mimeType=${mimeType}&height=${imageInfo.height}&width=${imageInfo.width}' | |
| callback_dict['callbackBodyType'] = 'application/x-www-form-urlencoded' | |
| callback_param = json.dumps(callback_dict).strip() | |
| base64_callback_body = base64.b64encode(callback_param.encode('utf-8')).decode('utf-8') | |
| token_dict = {} | |
| token_dict['accessid'] = access_key_id | |
| token_dict['host'] = host | |
| token_dict['policy'] = policy_encode | |
| token_dict['signature'] = sign_result | |
| token_dict['expire'] = expire_syncpoint | |
| token_dict['dir'] = upload_dir | |
| token_dict['callback'] = base64_callback_body | |
| return token_dict |