62 lines
1.5 KiB
Python
62 lines
1.5 KiB
Python
"""反馈模块数据库连接
|
||
自动检测：优先走 SSH 隧道（127.0.0.1:3307→生产库）
|
||
隧道不可用时回退到本地 3306
|
||
"""
|
||
import socket
|
||
import pymysql
|
||
from dbutils.pooled_db import PooledDB
|
||
|
||
_pool = None
|
||
|
||
|
||
def _detect_port():
|
||
"""检测 3307（隧道）是否可用，否则用本地 3306"""
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(2)
|
||
try:
|
||
result = sock.connect_ex(('127.0.0.1', 3307))
|
||
return 3307 if result == 0 else 3306
|
||
finally:
|
||
sock.close()
|
||
|
||
|
||
def _make_conn():
|
||
"""创建数据库连接（修复PooledDB + caching_sha2_password兼容问题）"""
|
||
port = _detect_port()
|
||
return pymysql.connect(
|
||
host="127.0.0.1",
|
||
port=port,
|
||
user="dy_app",
|
||
password="dy_data_2026",
|
||
database="dy_data",
|
||
charset="utf8mb4",
|
||
cursorclass=pymysql.cursors.DictCursor,
|
||
)
|
||
|
||
|
||
def get_prod_pool():
|
||
global _pool
|
||
if _pool is None:
|
||
_pool = PooledDB(
|
||
creator=_make_conn,
|
||
maxconnections=5,
|
||
mincached=1,
|
||
blocking=True,
|
||
)
|
||
return _pool
|
||
|
||
|
||
def prod_query(sql, params=None):
|
||
"""对统一数据库执行SQL"""
|
||
pool = get_prod_pool()
|
||
conn = pool.connection()
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, params or ())
|
||
if sql.strip().upper().startswith("SELECT"):
|
||
return cur.fetchall()
|
||
conn.commit()
|
||
return cur.lastrowid
|
||
finally:
|
||
conn.close()
|