mac
89810942cd
- backend/db.py：db() 的 DB_PASSWORD 改用 _require_env 强制读取，删除弱默认兜底 - backend/db.py：新增 _convert_placeholders 状态机替换 ? 占位符（跳过字符串字面量），并校验占位符/参数数量，替换原 sql.replace 全局替换脆弱点 - backend/flask_app.py：secret_key 改用 _require_env 强制读取，删除可预测弱默认，防止 session 伪造 - .gitea/workflows/deploy.yml：新增 2.5 步，发布时若线上 shared/.env 缺 DB_PASSWORD/SECRET_KEY 则从 CI secrets 自动补齐（已存在则保留），确保 P0 改动上线后可正常启动
121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
# db.py — 基础层：配置常量 + 数据库连接 + SQL 工具 + logger
|
|
# 被 helpers.py, routes.py, migrations/*, flask_app.py 共同依赖
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
import mysql.connector
|
|
|
|
# 确保 backend 目录在 sys.path 中（兼容 gunicorn --preload 模式）
|
|
_backend_dir = os.path.dirname(os.path.abspath(__file__))
|
|
if _backend_dir not in sys.path:
|
|
sys.path.insert(0, _backend_dir)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
|
|
|
# ---------- 路径常量 ----------
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DATA_DIR = ROOT / "data"
|
|
UPLOAD_DIR = DATA_DIR / "uploads"
|
|
DB_PATH = DATA_DIR / "opc.sqlite"
|
|
|
|
# ---------- 环境变量 ----------
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv(ROOT / ".env")
|
|
except ImportError:
|
|
pass
|
|
|
|
WEIXIN_BASE = Path(os.environ.get("WEIXIN_BASE", "/Users/mac/天机阁/地阁/慰心斋"))
|
|
|
|
# 建目录
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# ---------- 数据库连接 ----------
|
|
def _require_env(name):
|
|
"""强制读取必需环境变量，缺失即抛出清晰错误（禁止弱默认降级）"""
|
|
val = os.environ.get(name)
|
|
if not val:
|
|
raise RuntimeError(
|
|
f"缺少必需的环境变量 {name}，请在 .env 中显式配置（关键凭据禁止硬编码兜底）"
|
|
)
|
|
return val
|
|
|
|
|
|
def db():
|
|
return mysql.connector.connect(
|
|
host=os.environ.get("DB_HOST", "127.0.0.1"),
|
|
port=int(os.environ.get("DB_PORT", "3306")),
|
|
user=os.environ.get("DB_USER", "opc"),
|
|
password=_require_env("DB_PASSWORD"),
|
|
database=os.environ.get("DB_NAME", "opc"),
|
|
charset="utf8mb4",
|
|
collation="utf8mb4_unicode_ci",
|
|
)
|
|
|
|
|
|
def now():
|
|
return datetime.utcnow().isoformat()
|
|
|
|
|
|
def _convert_placeholders(sql):
|
|
"""将 ? 占位符转为 MySQL 的 %s，仅替换字符串字面量之外的 ?。
|
|
|
|
避免 `sql.replace('?', '%s')` 的全局替换缺陷：当 SQL 字面量或 LIKE 模式
|
|
中恰好含 `?` 时会误替换、导致参数错位。这里用状态机跳过单/双引号内的内容。
|
|
"""
|
|
out = []
|
|
in_single = in_double = False
|
|
i, n = 0, len(sql)
|
|
while i < n:
|
|
ch = sql[i]
|
|
if ch == "\\" and i + 1 < n: # 跳过转义序列
|
|
out.append(ch)
|
|
out.append(sql[i + 1])
|
|
i += 2
|
|
continue
|
|
if ch == "'" and not in_double:
|
|
in_single = not in_single
|
|
elif ch == '"' and not in_single:
|
|
in_double = not in_double
|
|
out.append("%s" if (ch == "?" and not in_single and not in_double) else ch)
|
|
i += 1
|
|
return "".join(out)
|
|
|
|
|
|
def _exec(conn, sql, args=()):
|
|
"""执行 SQL，将 ? 占位符安全地转为 MySQL 的 %s。
|
|
|
|
占位符数量与参数数量必须一致，否则立即报错，防止静默错位。
|
|
"""
|
|
converted = _convert_placeholders(sql)
|
|
n_ph = converted.count("%s")
|
|
if n_ph != len(args):
|
|
raise ValueError(
|
|
f"SQL 占位符数量({n_ph})与参数数量({len(args)})不匹配，请检查 SQL 与传参: {sql}"
|
|
)
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute(converted, args)
|
|
return cur
|
|
|
|
|
|
def rows(conn, sql, args=()):
|
|
cur = _exec(conn, sql, args)
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
return rows
|
|
|
|
|
|
def one(conn, sql, args=()):
|
|
cur = _exec(conn, sql, args)
|
|
row = cur.fetchone()
|
|
cur.close()
|
|
return row
|