Python

Fastapi+Supervisor+Gunicorn+Nginx部署.md

Fastapi+Supervisor+Gunicorn+Nginx部署

创建用户

sudo adduser tonytan

chmod -R 640 /etc/sudoers
sudo vim /etc/sudoers
chmod -R 440 /etc/sudoers

安装宝塔

wget -O install.sh https://download.bt.cn/install/install-ubuntu_6.0.sh && sudo bash install.sh ed8484bec

创建代码仓库密钥

ssh-keygen -t rsa -b 2048 -C "[email protected]"
cat .ssh/id_rsa.pub

下载/安装/配置 conda

wget https://repo.anaconda.com/miniconda/Miniconda3-py310_22.11.1-1-Linux-x86_64.sh
bash Miniconda3-py310_22.11.1-1-Linux-x86_64.sh
source ../.bashrc
conda create -n fastapi python=3.9

pip修改源

pip install -r requestment.txt -i https://pypi.tuna.tsinghua.edu.cn/simple/

fastapi 安装

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 gunicorn

Supervisor 配置

sudo 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

gunicorn.conf

# 并行工作进程数
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'

Nginx配置

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;
    }

docs 不能访问问题

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更新提交

moviepy优化视频.py

# 视频优化器
# 通过使用以下自动化脚本，你不仅可以使用 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")

PDF转图片.py

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)

Pillow做图片优化.py

# 图片优化器
#使用这个很棒的自动化脚本，可以帮助把图像处理的更好，你可以像在 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")

PyMuPDF做PDF转图片.py

# 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")

Pyside2GUI.py

# 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())

urllib3-poolmanager请求api接口.py

# 获取 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)

互联网下载器internetdownloadmanager示例.py

# 互联网下载器
# 你们可能使用下载软件从 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")

拼写修正.py

# 拼写修正
# 这个很棒的脚本将帮助你纠正你的文本单词拼写错误。你可以在下面找到脚本，将告诉你如何修复句子中的单个单词或多个单词。

# 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")

电池指示灯.py

# 电池指示灯
# 这个方便的脚本可以让你设置你想要得到通知的电池百分比，该脚本使用 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)
    

获取世界新闻.py

# 获取世界新闻
# 使用此自动化脚本让你随时了解每日世界新闻，你可以使用任何语言从任何国家/地区获取新闻。这个 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())

语法固定器.py

# 语法固定器
# 厌倦了校对你的长文章或文本，然后，你可以试试这个自动化脚本，它将扫描你的文本并纠正语法错误，这个很棒的脚本使用 Happtransformer 模块，这是一个机器学习模块，经过训练可以修复文本中的语法错误。

# Grammer Fixer
# 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)

阿里云OSS签名后直传.py

# -*- 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