python基础封装

.gitignore

@@ -8,3 +8,5 @@ data/
 .vscode/launch.json
 public/electron/
 pnpm-lock.yaml
+__pycache__/*
+*.pyc

python/.env

@@ -0,0 +1,18 @@
+# 应用名称
+app_name="智慧拍-后端应用"
+# 应用host地址
+app_host="0.0.0.0"
+# 应用服务启动名称
+app_run="api:app"
+# 应用版本号
+version="1.0.0"
+# 是否调试
+debug=True
+# 端口号
+port=7074
+# 线程数
+works=1
+# 日志相关
+log_file_name="app.log" #日志名称
+max_bytes=102400 #最大字节数
+backup_counts=3 #备份数量

python/.gitignore

@@ -0,0 +1,2 @@
+.venv/
+venv/

python/api.py

@@ -0,0 +1,20 @@
+from models import *
+from logger import logger
+from serial.tools import list_ports
+@app.get("/")
+async def index():
+    logger.info("Starting")
+    return {"message": "Hello World"}
+
+
+@app.get("/api/hello")
+async def hello():
+    return {"app_name": "FastAPI框架学习", "app_version": "v0.0.1"}
+
+
+@app.get("/scan_serials")
+async def scanSerials():
+    '''扫描串口'''
+    ports = list_ports.comports()
+    print("Scanning",ports)
+    return {"message": "Hello World"}

python/app.log

@@ -0,0 +1,2 @@
+2025-03-11 13:06:36,268 - api - index - line:6 - INFO - Starting
+2025-03-11 13:06:50,590 - api - scanSerials - line:19 - INFO - Starting

python/index.py

@@ -0,0 +1,24 @@
+import sys
+import uvicorn
+import signal
+from api import *
+from sockets import *
+
+def handle_shutdown(signum, frame):
+    """关闭系统应用服务"""
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    signal.signal(signal.SIGINT, handle_shutdown)
+    signal.signal(signal.SIGTERM, handle_shutdown)
+    # uvicorn会多创建一个进程，并且stdio独立于控制台，如果（开发时）出现进程没有关闭，可尝试关闭终端
+    uvicorn.run(
+        APP_RUN,
+        host=APP_HOST,
+        port=int(PORT),
+        reload=IS_DEBUG,
+        workers=int(APP_WORKS),
+    )
+    # 控制台默认关闭输出信息，如果想要查看控制台输出，请单独启动服务 npm run dev-python
+    print("python server is running at port:", PORT)

python/logger.py

@@ -0,0 +1,23 @@
+import logging.handlers
+import logging
+# 导入设置
+from settings import LOG_FILE_NAME, MAX_BYTES, BACKUP_COUNTS
+
+# 日志相关配置
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+ch = logging.StreamHandler()
+fh = logging.handlers.RotatingFileHandler(
+    str(LOG_FILE_NAME),
+    mode="a",
+    maxBytes=int(MAX_BYTES),
+    backupCount=int(BACKUP_COUNTS),
+)
+formatter = logging.Formatter(
+    "%(asctime)s - %(module)s - %(funcName)s - line:%(lineno)d - %(levelname)s - %(message)s"
+)
+ch.setFormatter(formatter)
+fh.setFormatter(formatter)
+logger.addHandler(ch)  # 将日志输出至屏幕
+logger.addHandler(fh)  # 将日志输出至文件
+logger = logging.getLogger(__name__)

python/middleware.py

@@ -0,0 +1,64 @@
+from typing import Union, Optional
+from fastapi import FastAPI, Request, Body, Form, Query
+from fastapi.websockets import WebSocket
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware.cors import CORSMiddleware
+from starlette.exceptions import HTTPException
+from settings import *
+import random
+import os
+from pydantic import BaseModel, validator, conint, constr, Field
+
+# 关闭文档
+app = FastAPI()
+# app.mount("/model", StaticFiles(directory="model"), name="model")
+app.add_middleware(
+    CORSMiddleware,
+    # 允许跨域的源列表，例如 ["http://www.example.org"] 等等，["*"] 表示允许任何源
+    allow_origins=["*"],
+    # 跨域请求是否支持 cookie，默认是 False，如果为 True，allow_origins 必须为具体的源，不可以是 ["*"]
+    allow_credentials=False,
+    # 允许跨域请求的 HTTP 方法列表，默认是 ["GET"]
+    allow_methods=["*"],
+    # 允许跨域请求的 HTTP 请求头列表，默认是 []，可以使用 ["*"] 表示允许所有的请求头
+    # 当然 Accept、Accept-Language、Content-Language 以及 Content-Type 总之被允许的
+    allow_headers=["*"],
+    # 可以被浏览器访问的响应头, 默认是 []，一般很少指定
+    # expose_headers=["*"]
+    # 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600，一般也很少指定
+    # max_age=1000
+)
+
+
+class UnicornException(Exception):
+    def __init__(self, msg: str, code: int = -1):
+        self.msg = msg
+        self.code = code
+
+@app.exception_handler(UnicornException)
+async def error_throw(request: Request, exc: UnicornException):
+
+    return JSONResponse(
+        status_code=200,
+        content={
+            "code": exc.code,
+            "msg": exc.msg,
+        },
+    )
+
+
+@app.exception_handler(HTTPException)
+async def global_exception_handler(request, exc):
+    if exc.status_code == 500:
+        err_msg = "Server Internal Error"
+    else:
+        err_msg = exc.detail
+    return JSONResponse(
+        {"code": exc.status_code, "err_msg": err_msg, "status": "Failed"}
+    )
+@app.on_event("shutdown")
+async def shutdown_event():
+    # 清理操作
+    print("服务关闭完成,done!")

python/models.py

@@ -0,0 +1 @@
+from middleware import *

python/settings.py

@@ -0,0 +1,36 @@
+from dotenv import load_dotenv, find_dotenv
+from pathlib import Path  # Python 3.6+ only
+import os
+
+# 一、自动搜索 .env 文件
+load_dotenv(verbose=True)
+
+
+# 二、与上面方式等价
+load_dotenv(find_dotenv(), verbose=True)
+
+# 三、或者指定 .env 文件位置
+env_path = Path(".") / ".env"
+load_dotenv(dotenv_path=env_path, verbose=True)
+# 应用名称
+APP_NAME = os.getenv("app_name")
+# 应用版本号
+APP_VERSION = os.getenv("version")
+# 是否开启调试模式
+IS_DEBUG = os.getenv("debug")
+# 应用端口号
+PORT = os.getenv("port")
+# 应用线程数
+APP_WORKS = os.getenv("works")
+# 应用host地址
+APP_HOST = os.getenv("app_host")
+# 应用服务启动名称
+APP_RUN = os.getenv("app_run")
+
+# 日志名称
+LOG_FILE_NAME = os.getenv("log_file_name")
+# 最大字节数
+MAX_BYTES = os.getenv("max_bytes")
+# 备份数量
+BACKUP_COUNTS = os.getenv("backup_counts")
+

python/setup.py

@@ -0,0 +1,24 @@
+from cx_Freeze import setup, Executable
+
+# 创建可执行文件的配置
+executableApp = Executable(
+    script="main.py",
+    target_name="pyapp",
+)
+
+# 打包的参数配置
+options = {
+    "build_exe": {
+        "build_exe":"./dist/",
+        "excludes": ["*.txt"],
+        "optimize": 2,
+    }
+}
+
+setup(
+    name="pyapp",
+    version="1.0",
+    description="python app",
+    options=options,
+    executables=[executableApp]
+)

python/sockets.py

@@ -0,0 +1,7 @@
+from models import *
+@app.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    while True:
+        data = await websocket.receive_text()
+        await websocket.send_text(f"Message text was: {data}")