Python

PDM介绍 一个现代的 Python 包管理器，支持 PEP 582 跟pipenv与poetry有着相同功能 抛弃传统包管理器所需的虚拟环境 - PEP 582 比poetry更快的依赖解析 类似pnpm的中心化安装缓存 插件系统 PEP 621 元数据格式 兼容 PEP 517 的构建后端，用于构建发布包(源码格式与 wheel 格式) ...

基本的上下文管理器 需要定义： class ContextM: def __enter__(): dosomethings() def __exit__(): dosometings() class AsyncContextM: async def __aenter__(): await dosometings() async def __aexit__(): await dosometings() 调用时使用with ContextM()或async with AsyncContextM ContextManager Python内置的一个很方便的生成上下文管理器的装饰器 ...

日志重写 import logging import os import sys from pprint import pformat from loguru import logger from config import settings class InterceptHandler(logging.Handler): def emit(self, record): # Get corresponding Loguru level if it exists try: level = logger.level(record.levelname).name except ValueError: level = record.levelno # Find caller from where originated the logged message frame, depth = logging.currentframe(), 2 while frame.f_code.co_filename == logging.__file__: frame = frame.f_back depth += 1 logger.opt(depth=depth, exception=record.exc_info).log( level, record.getMessage() ) def format_record(record: dict) -> str: format_string = settings.config.LOG_FORMAT if record["extra"].get("payload") is not None: record["extra"]["payload"] = pformat( record["extra"]["payload"], indent=4, compact=True, width=88 ) format_string += "\n<level>{extra[payload]}</level>" format_string += "{exception}\n" return format_string # it requires app init or won't be work def logger_init(): log_path_all = os.path.join(settings.config.log_path, settings.config.LOG_FILE) log_path_error = os.path.join(settings.config.log_error_path, settings.config.LOG_FILE) logging.getLogger().handlers = [InterceptHandler()] logger.configure( handlers=[{"sink": sys.stdout, "level": logging.DEBUG, "format": format_record}]) # logger.add(config.LOG_DIR, encoding='utf-8', rotation="9:46") logger.add(log_path_all, rotation="00:00", retention="2 week", enqueue=True, diagnose=True, level="DEBUG", format=format_record ) logger.add(log_path_error, rotation="00:00", retention="3 week", enqueue=True, backtrace=True, diagnose=True, level="ERROR", format=format_record ) logger.debug('日志系统已加载\n\n\n') logging.getLogger("uvicorn.access").handlers = [InterceptHandler()] 在app中调用 app = FastAPI( title=config.settings.title, debug=config.settings.config.DEBUG ) # app create frist, then init logger logger_init() origins = [ "http://127.0.0.1", f"http://127.0.0.1:{config.settings.config.port}", "http://localhost", f"http://localhost:{config.settings.config.port}", # "*" ] app.include_router(scraper.router) app.include_router(works.router) app.include_router(users.router) app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) if __name__ == '__main__': uvicorn.run( app='main:app', host='0.0.0.0', port=config.settings.config.port, reload=True, # should be False at production access_log=True, ws_ping_interval=300, ws_ping_timeout=300 )

以Depends的方式定义jwt from jose import JWTError, jwt from passlib.context import CryptContext from fastapi.security import OAuth2PasswordBearer from fastapi import Depends, HTTPException, status from db import Curd from typing import Optional from datetime import datetime from config import settings from utils import logger from .db import get_db from datetime import timedelta pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/login") def hash_password(plain_password: str) -> str: return pwd_context.hash(plain_password) def verity_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) async def jwt_authenticate_user(username: str, password: str): db = get_db() user = await Curd.get_user_by_name(await db.__anext__(), username) if not user: return False if not verity_password(password, user.password): return False return user def created_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.config.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(claims=to_encode, key=settings.default.SECRET_KEY, algorithm="HS256") return encoded_jwt async def jwt_verity(token: str = Depends(oauth2_schema)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: playload = jwt.decode(token=token, key=settings.default.SECRET_KEY, algorithms=["HS256"]) except JWTError as e: logger.error("验证出现错误:\n", e) raise credentials_exception else: return playload async def jwt_get_current_user(playload: dict = Depends(jwt_verity)): db = get_db() user_id = int(playload.get("sub")) user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id) return user async def jwt_verity_admin(playload: dict = Depends(jwt_verity)): db = get_db() user_id = int(playload.get("sub")) is_admin = playload.get("is_admin") if is_admin: user = await Curd.get_user_by_id(await db.__anext__(), user_id=user_id) if user.is_admin: return raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) 需要分发token的url router = APIRouter( prefix="/users", tags=["user"], ) # it defind at dependencies.jwt # oauth2_schema = OAuth2PasswordBearer(tokenUrl="/users/token") @router.post("/register", response_model=UserToken) async def user_register(user: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)): if await Curd.is_user_name_exists(db, user.username): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="username already exists") hashed_password = hash_password(user.password) new_user = UserCreate(id=worker.get_id(), name=user.username, password=hashed_password) await Curd.register_user(db, new_user) @router.post("/login", response_model=UserToken) async def user_login(user: OAuth2PasswordRequestForm = Depends()): user = await jwt_authenticate_user(username=user.username, password=user.password) if not user: raise HTTPException( status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token = created_access_token(data={"sub": str(user.id)}) user_token = UserToken(username=user.name, access_token=access_token, token_type="bearer") return user_token 需要token进行访问限制的url router = APIRouter( prefix="/works", tags=["works"], ) @router.get("/get", response_model=Works) async def test( order: int, nsfw: bool, page: int, sort: int, user: User = Depends(jwt_get_current_user), db: AsyncSession = Depends(get_db), ): works = await Curd.get_works(db, user.id, order, nsfw, page, sort) return works

ORM ORM：对象关系映射（Object Relational Mapping，简称ORM，或O/RM，或O/R mapping)，用于实现面向对象编程语言里不同类型系统的数据之间的转换 ...

思路 C++实在是弄不懂 代码 from typing import List, Optional class Node: """ 结点 """ def __init__(self): self.name: Optional[str] = None self.weight: Optional[int] = None self.left_child: Optional[Node] = None self.right_child: Optional[Node] = None self.parent: Optional[Node] = None class NodeCode: """ 结点的编码 """ def __init__(self, name: str, code: str): self.name = name self.code = code def __str__(self): return f"{self.name}:{self.code}" class HaffumanTreeCode: """ 哈夫曼树的编码 """ def __init__(self): self.code_list: List[NodeCode] = [] def append(self, code: NodeCode): self.code_list.append(code) def code(self, massage: str): """ 针对编码表编码 :param massage: 需要编码的信息 :return: """ code = "" for i in massage: for n in self.code_list: if n.name == i: code += n.code break # raise ValueError(f"'{i}'不存在") return code def decode(self, code: str): """ 针对编码表解码 :param code:需要解码的哈夫曼码 :return: """ out = "" i = 0 while code: code_need_match = code[0:i+1] for n in self.code_list: if n.code == code_need_match: out += n.name code = code[i+1:] i = 0 break i += 1 return out def __str__(self): """ 打印 :return: """ out = "" for i in self.code_list: out += i.__str__() + "\n" return out.strip() class HaffumanTree: """ 哈夫曼树 """ def __init__(self): self.root = Node() self.nodes: List[Node] = [] self.nodes_raw: List[Node] = [] self.codes: HaffumanTreeCode = HaffumanTreeCode() def inser(self, newnode: Node): for i in range(len(self.nodes)): if self.nodes[i].weight < newnode.weight: self.nodes.insert(i, newnode) return self.nodes.append(newnode) def build_tree(self, name_list: List[str], weight_list: List[int]): """ 创建哈夫曼树 :param name_list: 字符列表 :param weight_list: 权重列表 :return: """ # 队每个字符生成结点 while len(name_list) > 0: max_weight = max(weight_list) max_index = weight_list.index(max_weight) max_name = name_list[max_index] newnode = Node() newnode.name = max_name newnode.weight = max_weight self.nodes.append(newnode) name_list.remove(max_name) weight_list.remove(max_weight) # 保存源结点 self.nodes_raw = self.nodes[:] # 创建新结点并生成树 while len(self.nodes) > 2: left = self.nodes.pop() right = self.nodes.pop() newnode = Node() newnode.weight = left.weight + right.weight newnode.left_child = left newnode.right_child = right left.parent = newnode right.parent = newnode # 根据权重插入 self.inser(newnode) left = self.nodes.pop() right = self.nodes.pop() self.root.weight = left.weight + right.weight self.root.left_child = left self.root.right_child = right left.parent = self.root right.parent = self.root def make_code(self): """ 生成编码表 :return: """ while len(self.nodes_raw): node = self.nodes_raw.pop() name = node.name code = "" while node.parent != self.root: parent = node.parent if parent.left_child == node: code = "0" + code elif parent.right_child == node: code = "1" + code node = parent if self.root.left_child == node: code = "0" + code elif self.root.right_child == node: code = "1" + code self.codes.append(NodeCode(name, code)) def decode(self, code: str): return self.codes.decode(code) def code(self, massage: str): return self.codes.code(massage) if __name__ == '__main__': h = HaffumanTree() name_list = ["a", "b", "c", "d", "e"] wight_list = [5, 11, 6, 21, 19] h.build_tree(name_list, wight_list) h.make_code() print(f"编码表：\n{h.codes}") print(f'bcacce编码为：{h.code("bcacce")}') print(f'01001000001001解码为： {h.decode("0100100000100110")}')

安装 pip install PyMySQL 连接 db = pymysql.connect(host="192.168.1.1", port=3306, user="root", password="password", database="database_name") cursor = db.cursor() 查/增 sql_insert = """ INSERT INTO 表名(id, name, path) VALUES (%s, %s, %s) """ # 无论数据库数据类型为int或date，全部使用py的字符串形式插入，所以使用%s cursor.execute(sql_insert, (user_id, name, path)) # 这种可以有效避免因敏感字符而导致出现语法错误 db.commit() # 进行提交，否则无效 # 当数据源为http相关时，注意转义 import html from pymysql.converters import escape_string safe_data = escape_string(html.unescape(unsafe_data)) # unescape转义html字符，escape转义为可直接插入数据库的安全字符串 # 多数据一次插入 cursor.executemany(sql, info) # info可为list，循环按顺序进行插入 db.commit() 以及 因为图库已经快20g了，为了管理方便，顺便就上了mysql做数据管理，但有个问题，pymysql非异步，可能会造成bot的io堵塞，现在的数据量查询速度还算快，但要是时间久了就不一定； ...

轻量web服务 介绍 FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 3.6+ 并基于标准的 Python 类型提示。 FastAPI 框架，高性能，易于学习，高效编码，生产可用 ...

一般的多条件判断 if 判断1 and 判断2 and 判断3: run() 当语句非常长的时候，就显得难懂且不美观 all()方法多条件判断 if all([条件1, 条件2, 条件3]): run() all()传入一个可迭代对象，当所有条件为True时，返回True

这个bot写了有段小时间了，一直没有做个总结之类的，正好最近使用了saya, 代码算是可以看了点，于是就准备做个小总结。 Bot是什么 输入 - 处理 - 结果 大致上就是如此，发送一句话，bot根据这句话进行处理，然后得到预期结果。很明显，bot就是IO密集型程序，使用异步的方式是最便捷也是最高效的，这里就要说下开发框架 ...