ixuuux

from playwright.sync_api import sync_playwright


def save_mhtml(path: str, text: str):
    with open(path, mode='w', encoding='UTF-8', newline='\n') as file:
        file.write(text)

def save_page(url: str, path: str):
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(headless=False)

ixuuux

class Singleton(object):
    """ 单例类
    
    > @Singleton
    > class Demo:
    >     def __init__(self):
    >         pass
    >
    > assert Demo() == Demo()

ixuuux

import sys
import atexit
from typing import Dict

# noinspection PyProtectedMember
from loguru._logger import Core as _loguru_Core
# noinspection PyProtectedMember
from loguru._logger import Logger as _loguru_Logger
# noinspection PyProtectedMember
from loguru import _defaults

ixuuux

# 解决因类型标注导致的循环导入报错

# file1.py
from typing import TYPE_CHECKING  # https://docs.python.org/3/library/typing.html#typing.TYPE_CHECKING

if TYPE_CHECKING:  # 关键
    from file2 import C2

class C1:

ixuuux

from functools import wraps
from typing import Callable


# 为函数增加结果回调功能
def add_done_callback(call_func: Callable, *call_args, **call_kwargs):
    """ 为函数增加结果回调功能的装饰器（主要为获取子线程运行结果）

    def done_call(result, token):
        print(result, token)

ixuuux

import re

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# ------------------ 错误捕捉处理 ------------------
@app.exception_handler(RequestValidationError)

ixuuux

# -*- coding: utf-8 -*-
"""
File Name   : send_mail.py
Create Date : 2020/10/24 15:26
"""
import threading
import smtplib
from email.mime.text import MIMEText

class SendMail:

ixuuux

# -*- coding: utf-8 -*-
import os


def remove_any_dir_file(dir_or_file_path: str) -> None:
    """ 删除任意文件/文件夹。非空文件夹依然可以正确删除。

    例如：
        >>> import os
        >>> dir_path = os.path.join(os.getcwd(), 'test')

ixuuux

import multiprocessing

# example:
# gunicorn app:app -b localhost:10021 -w 3 -D -k uvicorn.workers.UvicornWorker --error-logfile ./logs/gunicorn_error.log --log-level warning

# 并行工作进程数
workers = multiprocessing.cpu_count()*2+1  # -w 3

# 监听内网端口
bind = '0.0.0.0:8000'  # -b 0.0.0.0:8000

ixuuux

# -*- coding: utf-8 -*-
import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor

thread_pool = ThreadPoolExecutor(max_workers=5)

def task(a: int, b: int) -> int:
    time.sleep(1)
    return a * b