chore: 基础设施 — 阿里云源、北京时间、HEX计数器、打印机配置

This commit is contained in:
2026-08-05 14:00:13 +08:00
parent 8458b99782
commit 82f474f71f
6 changed files with 91 additions and 2 deletions

4
.gitignore vendored
View File

@ -35,6 +35,10 @@ frontend/src-tauri/target/
# ===== Docker =====
docker-compose.override.yml
# ===== Runtime 目录 =====
backend/uploads/
backend/data/
# ===== Misc =====
*.log
*.tmp

View File

@ -5,6 +5,10 @@ FROM python:3.13-slim
WORKDIR /app
# 换阿里云源(国内网络加速)
RUN sed -i 's|http://deb.debian.org/debian|https://mirrors.aliyun.com/debian|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
sed -i 's|http://deb.debian.org/debian|https://mirrors.aliyun.com/debian|g' /etc/apt/sources.list 2>/dev/null || true
# 系统依赖Pillow / cryptography 编译所需)
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \

View File

@ -1,8 +1,9 @@
"""安全模块 — JWT Token 生成与验证"""
from datetime import datetime, timedelta, timezone
from datetime import timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
from app.core.time_utils import get_beijing_time
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
@ -12,7 +13,7 @@ ALGORITHM = "HS256"
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
"""生成 JWT Access Token"""
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (
expire = get_beijing_time() + (
expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
)
to_encode.update({"exp": expire})

View File

@ -0,0 +1,10 @@
"""全局北京时间 (UTC+8)"""
from datetime import datetime
from zoneinfo import ZoneInfo
BEIJING_TZ = ZoneInfo("Asia/Shanghai")
def get_beijing_time() -> datetime:
"""返回当前北京时间"""
return datetime.now(BEIJING_TZ)

View File

@ -0,0 +1,26 @@
"""16进制自增计数器 — 基于 PostgreSQL Sequence生成 16 位 HEX 唯一 ID"""
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
SEQUENCE_NAME = "product_hex_counter"
async def ensure_sequence(db: AsyncSession) -> None:
"""确保 counter sequence 存在(幂等)"""
await db.execute(
text(f"CREATE SEQUENCE IF NOT EXISTS {SEQUENCE_NAME} START 1;")
)
async def next_hex_id(db: AsyncSession, length: int = 16) -> str:
"""
生成下一个 hex ID。
示例: 1 → "0000000000000001"
15 → "000000000000000F"
16 → "0000000000000010"
255 → "00000000000000FF"
"""
result = await db.execute(text(f"SELECT nextval('{SEQUENCE_NAME}');"))
counter: int = result.scalar_one()
return format(counter, f"0{length}X")

View File

@ -0,0 +1,44 @@
"""打印机配置管理 — JSON 文件持久化"""
import json
import os
from pathlib import Path
CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "data"
CONFIG_FILE = CONFIG_DIR / "printer_config.json"
DEFAULT_CONFIG = {
"label_printer": {
"ip": "192.168.9.221",
"port": 9100,
"enabled": False,
},
}
class PrintConfigManager:
"""打印机 IP/端口 配置读写"""
@staticmethod
def _ensure_file() -> None:
if not CONFIG_DIR.exists():
CONFIG_DIR.mkdir(parents=True)
if not CONFIG_FILE.exists():
PrintConfigManager.save_config(DEFAULT_CONFIG)
@staticmethod
def get_config() -> dict:
PrintConfigManager._ensure_file()
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
return json.load(f)
@staticmethod
def save_config(config: dict) -> None:
if not CONFIG_DIR.exists():
CONFIG_DIR.mkdir(parents=True)
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2, ensure_ascii=False)
@staticmethod
def get_printer(name: str = "label_printer") -> dict:
config = PrintConfigManager.get_config()
return config.get(name, DEFAULT_CONFIG.get("label_printer", {}))