1ae901a040
- Add new Access table mappings for transmitter contracts (25/26年变送器) and tube bending workshop (弯管车间烘洗) - Add auto-create target table from Access schema when table does not exist (db_utils + init_full_sync) - Implement per-primary-key verification in incremental sync: verify deletes are gone and inserts are present before marking Synced=1 - Add post-commit re-verification with Synced rollback on failure for automatic retry - Batch IN clause parameters to stay under SQL Server limit - Adjust poll interval from 5s to 30s - Improve IDENTITY_INSERT cleanup in finally blocks Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
# db_utils.py
|
|
import pyodbc
|
|
import datetime
|
|
import decimal
|
|
from config import SQL_SERVER_CONN, ACCESS_DRIVER
|
|
|
|
|
|
# ================= 表存在性检查 & 自动建表 =================
|
|
|
|
def table_exists(cursor, schema, table):
|
|
"""检查 SQL Server 表是否存在"""
|
|
query = """
|
|
SELECT COUNT(*)
|
|
FROM sys.tables t
|
|
JOIN sys.schemas s ON t.schema_id = s.schema_id
|
|
WHERE s.name = ? AND t.name = ?
|
|
"""
|
|
cursor.execute(query, (schema, table))
|
|
return cursor.fetchone()[0] > 0
|
|
|
|
|
|
def ensure_schema(cursor, schema):
|
|
"""确保 schema 存在，不存在则创建"""
|
|
cursor.execute("SELECT SCHEMA_ID(?)", (schema,))
|
|
if cursor.fetchone()[0] is None:
|
|
cursor.execute(f"CREATE SCHEMA [{schema}]")
|
|
|
|
|
|
def _access_col_to_sql(col_info, pk_col):
|
|
"""将 Access 列描述 (cursor.description 元组) 转为 SQL Server 列定义
|
|
|
|
Access ODBC 驱动的 type_code 是 Python 类型对象:
|
|
int → INT
|
|
str → NVARCHAR(size) (size > 4000 时为 Memo 字段 → NVARCHAR(MAX))
|
|
datetime.datetime → DATETIME
|
|
float → FLOAT
|
|
decimal.Decimal → DECIMAL(p,s)
|
|
bool → BIT
|
|
"""
|
|
col_name, type_code, _, size, precision, scale, nullable = col_info
|
|
is_pk = (col_name == pk_col)
|
|
|
|
if type_code is int:
|
|
sql_type = "INT"
|
|
if is_pk:
|
|
sql_type += " IDENTITY(1,1) PRIMARY KEY"
|
|
elif type_code is float:
|
|
sql_type = "FLOAT"
|
|
elif type_code is bool:
|
|
sql_type = "BIT"
|
|
elif type_code is datetime.datetime:
|
|
sql_type = "DATETIME"
|
|
elif type_code is decimal.Decimal:
|
|
sql_type = f"DECIMAL({precision or 18}, {scale or 0})"
|
|
elif type_code is str:
|
|
# size > 4000 → Access Memo 字段，用 NVARCHAR(MAX)
|
|
if not size or size <= 0 or size > 4000:
|
|
sql_type = "NVARCHAR(MAX)"
|
|
else:
|
|
sql_type = f"NVARCHAR({size})"
|
|
if is_pk:
|
|
sql_type += " PRIMARY KEY"
|
|
else:
|
|
sql_type = "NVARCHAR(255)"
|
|
if is_pk:
|
|
sql_type += " PRIMARY KEY"
|
|
|
|
return f"[{col_name}] {sql_type}"
|
|
|
|
|
|
def create_table_from_access(sql_cursor, target_schema, target_table,
|
|
acc_description, pk_col):
|
|
"""根据 Access cursor.description 在 SQL Server 自动建表"""
|
|
col_defs = [_access_col_to_sql(col, pk_col) for col in acc_description]
|
|
full_name = fmt_table(target_schema, target_table)
|
|
col_str = ",\n ".join(col_defs)
|
|
create_sql = f"CREATE TABLE {full_name} (\n {col_str}\n)"
|
|
sql_cursor.execute(create_sql)
|
|
|
|
def get_sql_conn():
|
|
"""获取 SQL Server 连接"""
|
|
# 显式添加 TrustServerCertificate=yes 以兼容 ODBC Driver 18+
|
|
conn_str = (
|
|
f"DRIVER={SQL_SERVER_CONN['driver']};SERVER={SQL_SERVER_CONN['server']};"
|
|
f"DATABASE={SQL_SERVER_CONN['database']};UID={SQL_SERVER_CONN['uid']};PWD={SQL_SERVER_CONN['pwd']};"
|
|
"Encrypt=yes;TrustServerCertificate=yes;"
|
|
)
|
|
return pyodbc.connect(conn_str)
|
|
|
|
def get_access_conn(file_path):
|
|
"""获取 Access 连接"""
|
|
conn_str = f"DRIVER={ACCESS_DRIVER};DBQ={file_path};"
|
|
return pyodbc.connect(conn_str)
|
|
|
|
def fmt_table(schema, table):
|
|
"""格式化 SQL Server 表名 [schema].[table]"""
|
|
return f"[{schema}].[{table}]"
|
|
|
|
def get_columns(cursor, table_name):
|
|
"""获取 Access 表的列名"""
|
|
# Access 查询表名加 []
|
|
cursor.execute(f"SELECT TOP 1 * FROM [{table_name}]")
|
|
return [column[0] for column in cursor.description]
|
|
|
|
def generate_insert_sql(target_schema, target_table, columns):
|
|
"""生成带架构的 INSERT 语句"""
|
|
full_table_name = fmt_table(target_schema, target_table)
|
|
col_str = ",".join([f"[{col}]" for col in columns])
|
|
placeholders = ",".join(["?"] * len(columns))
|
|
return f"INSERT INTO {full_table_name} ({col_str}) VALUES ({placeholders})" |