Compare commits
18 Commits
4721634b81
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
f6580b4994 | ||
|
|
f9b2d3b4da | ||
|
|
3db6bae4e3 | ||
|
|
dd27cbaf97 | ||
|
|
04779c5aea | ||
|
|
4f9343ce69 | ||
|
|
8639fedf74 | ||
|
|
bc461c12ce | ||
|
|
17b6b246c7 | ||
|
|
1ae901a040 | ||
|
|
3791966446 | ||
|
|
11d284a6bf | ||
|
|
086c59cc03 | ||
|
|
2a2ec22d1c | ||
|
|
3bcc5aafbe | ||
|
|
9192f15610 | ||
|
|
a500f45a29 | ||
|
|
d4db287940 |
@@ -1,14 +0,0 @@
|
||||
{
|
||||
"permissions": {
|
||||
"allow": [
|
||||
"Bash(python etl_manager.py --help)",
|
||||
"Bash(python run_incremental_sync.py)",
|
||||
"Bash(python -c \"import run_incremental_sync; print\\(''run_incremental_sync.py: import OK''\\)\")",
|
||||
"Bash(python -c \"import etl_manager; print\\(''etl_manager.py: import OK''\\)\")",
|
||||
"Bash(python -c \"import init_full_sync; print\\(''init_full_sync.py: import OK''\\)\")",
|
||||
"Bash(python -c \"import migration; print\\(''migration.py: import OK''\\)\")",
|
||||
"Bash(python migration.py)",
|
||||
"Bash(python -c \"import log_utils; from log_utils import LoggerManager, log_start, log_complete; print\\(''log_utils: OK''\\)\")"
|
||||
]
|
||||
}
|
||||
}
|
||||
14
.env.example
Normal file
14
.env.example
Normal file
@@ -0,0 +1,14 @@
|
||||
# SQL Server
|
||||
DB_DRIVER=ODBC Driver 18 for SQL Server
|
||||
DB_SERVER=
|
||||
DB_DATABASE=
|
||||
DB_USERNAME=
|
||||
DB_PASSWORD=
|
||||
|
||||
# ntfy
|
||||
NTFY_SERVER_URL=
|
||||
NTFY_TOPIC=
|
||||
NTFY_TOKEN=
|
||||
|
||||
# Uptime Kuma
|
||||
UPTIME_KUMA_PUSH_URL=
|
||||
10
.gitignore
vendored
10
.gitignore
vendored
@@ -4,4 +4,12 @@ build
|
||||
dist
|
||||
log
|
||||
*.spec
|
||||
temp
|
||||
temp
|
||||
|
||||
# 环境变量
|
||||
.env
|
||||
|
||||
# Claude 临时文件
|
||||
.claude/
|
||||
.agents/
|
||||
tmpclaude-*
|
||||
125
CLAUDE.md
125
CLAUDE.md
@@ -6,7 +6,6 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
|
||||
|
||||
This is a data synchronization system (BLD_sync) that extracts data from:
|
||||
- **Microsoft Access databases** (.accdb) located on network shares
|
||||
- **Excel files** (.xlsm) containing production execution cards and contract data
|
||||
|
||||
And synchronizes it to:
|
||||
- **Microsoft SQL Server** (CompanyDB) with multiple schemas organized by data type
|
||||
@@ -21,15 +20,11 @@ The system supports both full initialization sync and incremental sync driven by
|
||||
┌─────────────────────────────────────────────────────────────────────┐
|
||||
│ Network File Sources │
|
||||
│ Access DBs (\\192.168.110.114\生产进度表\) │
|
||||
│ Excel Files (\\192.168.110.113\生产执行卡\) │
|
||||
└─────────────────────────────────────────────────────────────────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────────────┐
|
||||
│ Sync Scripts │
|
||||
│ • etl_manager.py - Main ETL orchestrator (primary) │
|
||||
│ • sync_excel_to_sql.py - Excel to SQL sync (legacy/alternative) │
|
||||
│ • migration.py - Legacy customer product type migration │
|
||||
│ • run_incremental_sync.py - Change log driven incremental sync │
|
||||
│ • init_full_sync.py - Full table truncation/reload │
|
||||
└─────────────────────────────────────────────────────────────────────┘
|
||||
@@ -43,36 +38,18 @@ The system supports both full initialization sync and incremental sync driven by
|
||||
└─────────────────────────────────────────────────────────────────────┘
|
||||
```
|
||||
|
||||
> Note: The system previously also synced Excel `.xlsm` production-execution-card files into
|
||||
> the `warehouseOutbound` schema (`executionCardData`, `contractData`, `customerProductType`).
|
||||
> That Excel pipeline has been removed; the existing tables and their data remain in CompanyDB
|
||||
> but are no longer updated.
|
||||
|
||||
### Key Components
|
||||
|
||||
**etl_manager.py** - Primary ETL orchestrator
|
||||
- `DataSynchronizer` class manages Excel to SQL sync
|
||||
- Supports incremental sync via file modification time comparison
|
||||
- Caches Excel files locally in `excel_cache/` directory
|
||||
- Uses SQLAlchemy with `fast_executemany=True` for bulk operations
|
||||
- Generates `contractData` table via MERGE statement from `executionCardData`
|
||||
|
||||
**config.py** - Central configuration
|
||||
- `SYNC_MAPPING`: Nested dict mapping Access files → tables → SQL targets
|
||||
- `EXCEL_CONFIGS`: List of Excel file configs with sheet names and field mappings
|
||||
- `TABLE_SCHEMA`: Column type definitions for data cleaning
|
||||
- `NTFY_CONFIG`: Push notification settings
|
||||
- `LOG_TABLE_CONFIG`: Change log table schema for incremental sync
|
||||
|
||||
**db_utils.py** - Database connection utilities
|
||||
- `get_sql_conn()`: SQL Server connection via pyodbc
|
||||
- `get_access_conn()`: Access database connection
|
||||
- `fmt_table()`: Safe table name formatting `[schema].[table]`
|
||||
- `generate_insert_sql()`: Dynamic INSERT statement generation
|
||||
|
||||
**ntfy_utils.py** - Push notifications
|
||||
- Sends alerts to ntfy server on errors/completion
|
||||
- Uses Bearer token authentication
|
||||
|
||||
**run_incremental_sync.py** - Log-driven incremental sync
|
||||
- Polls `TableChangeLog` table for unsynced records (`Synced=0`)
|
||||
- Queries by `TableAddress` to match configured file paths
|
||||
- Processes deletions and insertions in batches
|
||||
- Queries by `TableAddress` to match configured Access file paths
|
||||
- Connects to the source Access DB, reads latest rows by primary key, then DELETE + INSERT
|
||||
- Per-batch verification (delete/insert checks by primary key) before marking `Synced=1`
|
||||
- Long-running service with configurable polling interval
|
||||
|
||||
**init_full_sync.py** - Full table reload
|
||||
@@ -80,6 +57,33 @@ The system supports both full initialization sync and incremental sync driven by
|
||||
- Handles IDENTITY_INSERT ON/OFF for tables with identity columns
|
||||
- Progress logging with row counts and throughput metrics
|
||||
|
||||
**config/** - Configuration package
|
||||
- `database.py`: SQL Server / Access driver settings (read from env via `python-dotenv`)
|
||||
- `file_sources.py`: `SYNC_MAPPING` — nested dict mapping Access files → tables → SQL targets
|
||||
- `app_settings.py`: `LOG_TABLE_CONFIG`, `NTFY_CONFIG`, `UPTIME_KUMA_CONFIG`, `POLL_INTERVAL`, `BATCH_SIZE`
|
||||
- Symbols are re-exported from `config/__init__.py` (e.g. `from config import SYNC_MAPPING`)
|
||||
|
||||
**db_utils.py** - Database connection utilities
|
||||
- `get_sql_conn()`: SQL Server connection via pyodbc
|
||||
- `get_access_conn()`: Access database connection
|
||||
- `fmt_table()`: Safe table name formatting `[schema].[table]`
|
||||
- `generate_insert_sql()`: Dynamic INSERT statement generation
|
||||
- `create_table_from_access()`: Auto-create SQL Server table from Access `cursor.description`
|
||||
|
||||
**ntfy_utils.py** - Push notifications
|
||||
- Sends alerts to ntfy server on errors/completion
|
||||
- Uses Bearer token authentication
|
||||
|
||||
**uptime_kuma_utils.py** - Heartbeat monitoring
|
||||
- `UptimeKumaMonitor` pushes heartbeats to Uptime Kuma; used by the incremental sync service
|
||||
|
||||
**log_utils.py** - Unified logging
|
||||
- `LoggerManager` creates a timestamped log file under `log/` and archives old logs by year-month
|
||||
|
||||
**vbareplace.py** / **vba.txt** - Access VBA helper tool
|
||||
- Standalone drag-and-drop tool that injects the change-log VBA macro into Access forms and
|
||||
refreshes linked tables (packaged via PyInstaller; unrelated to the Python sync scripts)
|
||||
|
||||
## Common Development Tasks
|
||||
|
||||
### Install Dependencies
|
||||
@@ -94,60 +98,35 @@ pip install -r requirements.txt
|
||||
python init_full_sync.py
|
||||
```
|
||||
|
||||
### Run Excel to SQL Sync (Standard)
|
||||
|
||||
```bash
|
||||
python etl_manager.py
|
||||
```
|
||||
|
||||
Force sync all files (ignore modification times):
|
||||
```bash
|
||||
python etl_manager.py --force
|
||||
```
|
||||
|
||||
### Run Incremental Sync Service (Change Log Driven)
|
||||
|
||||
```bash
|
||||
python run_incremental_sync.py
|
||||
```
|
||||
|
||||
### Run Legacy Migration Script
|
||||
|
||||
```bash
|
||||
python migration.py
|
||||
```
|
||||
|
||||
Edit `FORCE_UPDATE = True` in `migration.py` to force full refresh.
|
||||
|
||||
## Configuration Management
|
||||
|
||||
**Main Config File**: `config.py`
|
||||
- Contains all database credentials, file paths, and mapping configurations
|
||||
- Modify `SYNC_MAPPING` to add new Access files/tables
|
||||
- Modify `EXCEL_CONFIGS` for new Excel sources
|
||||
|
||||
**Alternative Config**: `update_config.py`
|
||||
- Used by `sync_excel_to_sql.py`
|
||||
- Similar structure but different variable names
|
||||
- Contains `EXECUTION_CARD_FIELDS`, `CONTRACT_DATA_MAPPING`
|
||||
Configuration lives in the **`config/` package** (re-exported via `config/__init__.py`):
|
||||
- Database credentials and driver strings come from environment variables (see `.env.example`),
|
||||
loaded by `python-dotenv`.
|
||||
- Modify `SYNC_MAPPING` in `config/file_sources.py` to add new Access files/tables.
|
||||
|
||||
## Important Implementation Details
|
||||
|
||||
### SQL Server Connection
|
||||
- Uses ODBC Driver 18 for SQL Server
|
||||
- Requires `TrustServerCertificate=yes` due to self-signed cert
|
||||
- SQLAlchemy URL: `mssql+pyodbc:///?odbc_connect=...`
|
||||
- Always enable `fast_executemany=True` for bulk operations
|
||||
- Direct pyodbc connections (no SQLAlchemy); `fast_executemany=True` for bulk inserts
|
||||
|
||||
### Access Database Connection
|
||||
- Driver: `{Microsoft Access Driver (*.mdb, *.accdb)}`
|
||||
- Direct file path connection via pyodbc
|
||||
|
||||
### Data Cleaning
|
||||
- Integer fields: `pd.to_numeric().round().astype('Int64')`
|
||||
- String fields: Truncate to max length, replace empty with None
|
||||
- Date fields: Convert to `date()` objects, None for NaT
|
||||
- Duplicate removal: Based on primary key (usually `ID` or `总排号`)
|
||||
### Type Mapping / Auto Table Creation
|
||||
- When a SQL Server target table does not exist, it is created from the Access `cursor.description`.
|
||||
- Access types map to SQL Server types: `int`→`INT` (PK → `IDENTITY(1,1) PRIMARY KEY`),
|
||||
`float`→`FLOAT`, `bool`→`BIT`, `datetime`→`DATETIME`, `Decimal`→`DECIMAL(p,s)`,
|
||||
`str`→`NVARCHAR(size)` (size > 4000 → `NVARCHAR(MAX)`).
|
||||
|
||||
### Identity Column Handling
|
||||
Tables with identity columns require:
|
||||
@@ -157,7 +136,7 @@ SET IDENTITY_INSERT [schema].[table] ON
|
||||
SET IDENTITY_INSERT [schema].[table] OFF
|
||||
```
|
||||
|
||||
See `has_identity_column()` in `init_full_sync.py` for detection logic.
|
||||
See `has_identity_column()` in `init_full_sync.py` / `run_incremental_sync.py` for detection logic.
|
||||
|
||||
### File Path Matching in Incremental Sync
|
||||
The change log table stores paths in VBA format:
|
||||
@@ -169,7 +148,6 @@ The code constructs multiple match patterns for robust matching.
|
||||
## Database Schema Organization
|
||||
|
||||
SQL Server schemas by function:
|
||||
- `warehouseOutbound` - Execution card data, contract data, customer product types
|
||||
- `productionContractData` - Contract data by year (25年/26年压力表/温度计)
|
||||
- `productWarehousing` - Finished product inspection/warehousing records
|
||||
- `workshopOne/Two/Three` - Workshop production records
|
||||
@@ -181,16 +159,21 @@ SQL Server schemas by function:
|
||||
- `solderingData` - Soldering operation records
|
||||
- `TIGWelding` - TIG welding records
|
||||
- `executionCardIssuanceRecord` - Execution card issuance records
|
||||
- `tubeBending` - Tube bending records
|
||||
- `warehouseOutbound` - (Legacy) Excel-sourced tables: `executionCardData`, `contractData`,
|
||||
`customerProductType` — no longer updated
|
||||
|
||||
## Notifications
|
||||
|
||||
The system uses [ntfy](https://ntfy.sh/) for push notifications:
|
||||
- Configured in `NTFY_CONFIG` within `config.py`
|
||||
- Configured in `NTFY_CONFIG` within `config/app_settings.py`
|
||||
- Sends on: errors, critical failures, task completion
|
||||
- Authenticated via Bearer token
|
||||
|
||||
## Logging
|
||||
|
||||
- File logs: `log/` directory with timestamp rotation
|
||||
- File logs: `log/` directory, one timestamped file per run (`<prefix>_<YYYYMMDD>_<HHMMSS>.log`)
|
||||
- Console output: With emoji prefixes for status (✅ ❌ ⚠️ 🔄)
|
||||
- Incremental sync: Uses `TimedRotatingFileHandler` for daily log files
|
||||
- Old logs are archived into `log/Archive/YYYY-MM/` automatically
|
||||
(see `LoggerManager.archive_old_logs` in `log_utils.py`; `archive_existing_logs.py` is a
|
||||
one-time helper for historical logs)
|
||||
|
||||
92
archive_existing_logs.py
Normal file
92
archive_existing_logs.py
Normal file
@@ -0,0 +1,92 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
一次性脚本：将 log/ 根目录下的所有历史日志按年月移动到 Archive/ 目录
|
||||
|
||||
使用方法：
|
||||
python archive_existing_logs.py
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def extract_year_month_from_filename(filename: str) -> Optional[str]:
|
||||
"""从日志文件名中提取年月信息
|
||||
|
||||
支持的格式：
|
||||
- prefix_YYYYMMDD_HHMMSS.log
|
||||
- incremental_YYYYMMDD_HHMMSS.log
|
||||
- full_sync_YYYYMMDD_HHMMSS.log
|
||||
- excel_sync_YYYYMMDD_HHMMSS.log
|
||||
"""
|
||||
match = re.search(r'(\d{4})(\d{2})\d{2}_\d{6}', filename)
|
||||
if match:
|
||||
year = match.group(1)
|
||||
month = match.group(2)
|
||||
return f"{year}-{month}"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def archive_existing_logs():
|
||||
"""归档现有的所有日志文件到 Archive/YYYY-MM/ 目录"""
|
||||
log_dir = os.path.join(os.getcwd(), "log")
|
||||
archive_base = os.path.join(log_dir, "Archive")
|
||||
|
||||
if not os.path.exists(log_dir):
|
||||
print(f"日志目录不存在: {log_dir}")
|
||||
return
|
||||
|
||||
os.makedirs(archive_base, exist_ok=True)
|
||||
|
||||
moved_count = 0
|
||||
skipped_count = 0
|
||||
|
||||
for filename in os.listdir(log_dir):
|
||||
if not filename.endswith('.log'):
|
||||
continue
|
||||
|
||||
src = os.path.join(log_dir, filename)
|
||||
|
||||
# 跳过目录
|
||||
if os.path.isdir(src):
|
||||
continue
|
||||
|
||||
# 提取年月信息
|
||||
year_month = extract_year_month_from_filename(filename)
|
||||
|
||||
# 如果无法从文件名提取，使用文件修改时间
|
||||
if not year_month:
|
||||
try:
|
||||
stat = os.stat(src)
|
||||
mtime = datetime.fromtimestamp(stat.st_mtime)
|
||||
year_month = mtime.strftime("%Y-%m")
|
||||
except:
|
||||
year_month = "unknown"
|
||||
|
||||
# 创建年月子目录
|
||||
month_dir = os.path.join(archive_base, year_month)
|
||||
os.makedirs(month_dir, exist_ok=True)
|
||||
|
||||
# 移动文件
|
||||
dst = os.path.join(month_dir, filename)
|
||||
|
||||
try:
|
||||
shutil.move(src, dst)
|
||||
moved_count += 1
|
||||
print(f"[OK] {filename} -> Archive/{year_month}/")
|
||||
except Exception as e:
|
||||
print(f"[FAIL] 移动失败 {filename}: {e}")
|
||||
skipped_count += 1
|
||||
|
||||
print(f"\n完成！")
|
||||
print(f" 已归档: {moved_count} 个文件")
|
||||
print(f" 失败: {skipped_count} 个文件")
|
||||
print(f" 归档路径: {archive_base}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
archive_existing_logs()
|
||||
@@ -1,20 +1,19 @@
|
||||
# config/__init__.py
|
||||
# 统一配置导出接口
|
||||
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
|
||||
# 数据库配置
|
||||
from .database import SQL_SERVER_CONFIG, SQL_SERVER_CONN, DB_CONFIG, ACCESS_DRIVER
|
||||
|
||||
# 文件路径配置
|
||||
from .file_sources import SYNC_MAPPING, EXCEL_CONFIGS, MIGRATION_TASKS
|
||||
|
||||
# 字段映射配置
|
||||
from .field_mappings import TABLE_SCHEMA, CONTRACT_MAPPING
|
||||
# 文件路径配置（Access 同步映射）
|
||||
from .file_sources import SYNC_MAPPING
|
||||
|
||||
# 应用设置
|
||||
from .app_settings import (
|
||||
LOG_TABLE_CONFIG, NTFY_CONFIG,
|
||||
POLL_INTERVAL, BATCH_SIZE, CACHE_DIR, TEMP_DIR,
|
||||
EXECUTION_CARD_FIELDS, CONTRACT_DATA_FIELDS, CONTRACT_DATA_MAPPING
|
||||
LOG_TABLE_CONFIG, NTFY_CONFIG, UPTIME_KUMA_CONFIG,
|
||||
POLL_INTERVAL, BATCH_SIZE,
|
||||
)
|
||||
|
||||
# 统一导出列表
|
||||
@@ -22,11 +21,8 @@ __all__ = [
|
||||
# Database
|
||||
'SQL_SERVER_CONFIG', 'SQL_SERVER_CONN', 'DB_CONFIG', 'ACCESS_DRIVER',
|
||||
# File Sources
|
||||
'SYNC_MAPPING', 'EXCEL_CONFIGS', 'MIGRATION_TASKS',
|
||||
# Field Mappings
|
||||
'TABLE_SCHEMA', 'CONTRACT_MAPPING',
|
||||
'SYNC_MAPPING',
|
||||
# App Settings
|
||||
'LOG_TABLE_CONFIG', 'NTFY_CONFIG',
|
||||
'POLL_INTERVAL', 'BATCH_SIZE', 'CACHE_DIR', 'TEMP_DIR',
|
||||
'EXECUTION_CARD_FIELDS', 'CONTRACT_DATA_FIELDS', 'CONTRACT_DATA_MAPPING'
|
||||
'LOG_TABLE_CONFIG', 'NTFY_CONFIG', 'UPTIME_KUMA_CONFIG',
|
||||
'POLL_INTERVAL', 'BATCH_SIZE',
|
||||
]
|
||||
|
||||
@@ -19,97 +19,23 @@ LOG_TABLE_CONFIG = {
|
||||
# 从原 config.py 抽取
|
||||
NTFY_CONFIG = {
|
||||
'enabled': True,
|
||||
'server_url': 'https://ntfy.server10086.icu',
|
||||
'topic': 'bld',
|
||||
'token': 'tk_eop5fs66acxtwxf6vlkiojhdvkgb0',
|
||||
'server_url': os.environ.get('NTFY_SERVER_URL', ''),
|
||||
'topic': os.environ.get('NTFY_TOPIC', ''),
|
||||
'token': os.environ.get('NTFY_TOKEN', ''),
|
||||
'priority': {
|
||||
'error': 'high',
|
||||
'critical': 'urgent'
|
||||
}
|
||||
}
|
||||
|
||||
# ================= Uptime Kuma 心跳配置 =================
|
||||
# 增量同步服务心跳
|
||||
UPTIME_KUMA_CONFIG = {
|
||||
'enabled': True,
|
||||
'push_url': os.environ.get('UPTIME_KUMA_PUSH_URL', ''),
|
||||
'heartbeat_interval': 59 # 心跳间隔(秒)，需要与 Uptime Kuma 设置一致
|
||||
}
|
||||
|
||||
# ================= 运行参数 =================
|
||||
# 合并原 config.py 和 update_config.py 的配置
|
||||
POLL_INTERVAL = 5 # 轮询间隔(秒)
|
||||
POLL_INTERVAL = 30 # 轮询间隔(秒)
|
||||
BATCH_SIZE = 10000 # 批量处理大小
|
||||
CACHE_DIR = os.path.join(os.getcwd(), "temp") # Excel 缓存目录
|
||||
TEMP_DIR = os.path.join(os.getcwd(), "temp") # 临时目录（兼容 migration.py）
|
||||
|
||||
# ================= 字段配置 =================
|
||||
# 从原 update_config.py (sync_excel_to_sql.py) 抽取
|
||||
EXECUTION_CARD_FIELDS = {
|
||||
"合同年份": ("str", 10),
|
||||
"总排号": ("str", 50),
|
||||
"序号": ("int", None),
|
||||
"订单号": ("str", 150),
|
||||
"车间号": ("str", 20),
|
||||
"销售内部号": ("str", 50),
|
||||
"经办人": ("str", 20),
|
||||
"签订日期": ("date", None),
|
||||
"交货日期": ("date", None),
|
||||
"客户名称": ("str", 200),
|
||||
"产品名称": ("str", 200),
|
||||
"客户型号": ("str", 200),
|
||||
"选型型号": ("str", 200),
|
||||
"量程": ("str", 150),
|
||||
"数量": ("int", None),
|
||||
"备注2": ("str", 500),
|
||||
"备注1": ("str", 500),
|
||||
"位号": ("str", 500),
|
||||
"技术参数": ("str", 1000),
|
||||
"车间": ("str", 200),
|
||||
"工令号": ("str", 200),
|
||||
"接单日期": ("date", None),
|
||||
"新参数": ("str", 1000),
|
||||
"基本型号": ("str", 200),
|
||||
"公称外径": ("str", 200),
|
||||
"安装代码": ("str", 200),
|
||||
"设计形式": ("str", 200),
|
||||
"技术安装代码": ("str", 200),
|
||||
"隔膜类型": ("str", 200),
|
||||
"标准": ("str", 100),
|
||||
"隔膜大小": ("str", 200),
|
||||
"隔膜材质": ("str", 200),
|
||||
"膜片": ("str", 200),
|
||||
"膜片材质": ("str", 200),
|
||||
"CRM明细ID号": ("str", 200),
|
||||
"成品物料编码": ("str", 200),
|
||||
"工单号": ("str", 200),
|
||||
"盘号": ("str", 200),
|
||||
"编码": ("str", 200),
|
||||
"型批名称": ("str", 200),
|
||||
"型批型号": ("str", 200),
|
||||
"开票名称": ("str", 200),
|
||||
"开票型号": ("str", 200),
|
||||
"上数据时间": ("date", None),
|
||||
"生产代码1": ("str", 50),
|
||||
"生产代码2": ("str", 50)
|
||||
}
|
||||
|
||||
CONTRACT_DATA_FIELDS = {
|
||||
"合同年份": ("str", 10),
|
||||
"车间号": ("str", 20),
|
||||
"工令号": ("str", 200),
|
||||
"订单号": ("str", 150),
|
||||
"客户名称": ("str", 200),
|
||||
"产品型号": ("str", 200),
|
||||
"量程": ("str", 150),
|
||||
"数量": ("int", None),
|
||||
"单价": ("int", None),
|
||||
"ID": ("int", None),
|
||||
"位号": ("str", 500)
|
||||
}
|
||||
|
||||
CONTRACT_DATA_MAPPING = {
|
||||
"合同年份": "合同年份",
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"订单号": "订单号",
|
||||
"客户名称": "客户名称",
|
||||
"产品型号": "选型型号",
|
||||
"量程": "量程",
|
||||
"数量": "数量",
|
||||
"单价": None,
|
||||
"ID": None,
|
||||
"位号": "位号"
|
||||
}
|
||||
|
||||
@@ -1,13 +1,15 @@
|
||||
# config/database.py
|
||||
# 统一的数据库配置
|
||||
|
||||
import os
|
||||
|
||||
# ================= SQL Server 配置 =================
|
||||
SQL_SERVER_CONFIG = {
|
||||
'driver': 'ODBC Driver 18 for SQL Server',
|
||||
'server': '192.168.110.114',
|
||||
'database': 'CompanyDB',
|
||||
'username': 'peng',
|
||||
'password': 'Cqbld123456.',
|
||||
'driver': os.environ.get('DB_DRIVER', 'ODBC Driver 18 for SQL Server'),
|
||||
'server': os.environ.get('DB_SERVER', ''),
|
||||
'database': os.environ.get('DB_DATABASE', ''),
|
||||
'username': os.environ.get('DB_USERNAME', ''),
|
||||
'password': os.environ.get('DB_PASSWORD', ''),
|
||||
'TrustServerCertificate': 'yes'
|
||||
}
|
||||
|
||||
|
||||
@@ -1,73 +0,0 @@
|
||||
# config/field_mappings.py
|
||||
# 字段映射和清洗规则配置
|
||||
|
||||
import os
|
||||
from config.app_settings import EXECUTION_CARD_FIELDS
|
||||
|
||||
# ================= 字段清洗规则 =================
|
||||
# 从原 update_config.py 抽取
|
||||
# 基于 NVARCHAR (按字符数计算长度)
|
||||
TABLE_SCHEMA = {
|
||||
"合同年份": {"type": "str", "max_len": 10},
|
||||
"总排号": {"type": "str", "max_len": 50},
|
||||
"序号": {"type": "int"},
|
||||
"订单号": {"type": "str", "max_len": 150},
|
||||
"车间号": {"type": "str", "max_len": 20},
|
||||
"销售内部号": {"type": "str", "max_len": 50},
|
||||
"经办人": {"type": "str", "max_len": 20},
|
||||
"签订日期": {"type": "date"},
|
||||
"交货日期": {"type": "date"},
|
||||
"客户名称": {"type": "str", "max_len": 200},
|
||||
"产品名称": {"type": "str", "max_len": 200},
|
||||
"客户型号": {"type": "str", "max_len": 200},
|
||||
"选型型号": {"type": "str", "max_len": 200},
|
||||
"量程": {"type": "str", "max_len": 150},
|
||||
"数量": {"type": "int"},
|
||||
"备注2": {"type": "str", "max_len": 500},
|
||||
"备注1": {"type": "str", "max_len": 500},
|
||||
"位号": {"type": "str", "max_len": 500},
|
||||
"技术参数": {"type": "str", "max_len": 1000},
|
||||
"车间": {"type": "str", "max_len": 200},
|
||||
"工令号": {"type": "str", "max_len": 200},
|
||||
"接单日期": {"type": "date"},
|
||||
"新参数": {"type": "str", "max_len": 1000},
|
||||
"基本型号": {"type": "str", "max_len": 200},
|
||||
"公称外径": {"type": "str", "max_len": 200},
|
||||
"安装代码": {"type": "str", "max_len": 200},
|
||||
"设计形式": {"type": "str", "max_len": 200},
|
||||
"技术安装代码": {"type": "str", "max_len": 200},
|
||||
"隔膜类型": {"type": "str", "max_len": 200},
|
||||
"标准": {"type": "str", "max_len": 100},
|
||||
"隔膜大小": {"type": "str", "max_len": 200},
|
||||
"隔膜材质": {"type": "str", "max_len": 200},
|
||||
"膜片": {"type": "str", "max_len": 200},
|
||||
"膜片材质": {"type": "str", "max_len": 200},
|
||||
"CRM明细ID号": {"type": "str", "max_len": 200},
|
||||
"成品物料编码": {"type": "str", "max_len": 200},
|
||||
"工单号": {"type": "str", "max_len": 200},
|
||||
"盘号": {"type": "str", "max_len": 200},
|
||||
"编码": {"type": "str", "max_len": 200},
|
||||
"型批名称": {"type": "str", "max_len": 200},
|
||||
"型批型号": {"type": "str", "max_len": 200},
|
||||
"开票名称": {"type": "str", "max_len": 200},
|
||||
"开票型号": {"type": "str", "max_len": 200},
|
||||
"上数据时间": {"type": "date"},
|
||||
"生产代码1": {"type": "str", "max_len": 50},
|
||||
"生产代码2": {"type": "str", "max_len": 50}
|
||||
}
|
||||
|
||||
# ================= 合同数据映射 =================
|
||||
# 从原 update_config.py 抽取
|
||||
CONTRACT_MAPPING = {
|
||||
"合同年份": "合同年份",
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"订单号": "订单号",
|
||||
"客户名称": "客户名称",
|
||||
"产品型号": "选型型号",
|
||||
"量程": "量程",
|
||||
"数量": "数量",
|
||||
"单价": None,
|
||||
"ID": None,
|
||||
"位号": "位号"
|
||||
}
|
||||
@@ -26,6 +26,16 @@ SYNC_MAPPING = {
|
||||
"target_schema": "productionContractData",
|
||||
"target_table": "25年温度计合同数据",
|
||||
"pk_col": "ID"
|
||||
},
|
||||
"25年变送器合同数据": {
|
||||
"target_schema": "productionContractData",
|
||||
"target_table": "25年变送器合同数据",
|
||||
"pk_col": "ID"
|
||||
},
|
||||
"26年变送器合同数据": {
|
||||
"target_schema": "productionContractData",
|
||||
"target_table": "26年变送器合同数据",
|
||||
"pk_col": "ID"
|
||||
}
|
||||
},
|
||||
r"\\192.168.110.114\生产进度表\2026年数据\成品入库.accdb": {
|
||||
@@ -86,6 +96,11 @@ SYNC_MAPPING = {
|
||||
"target_schema": "machining",
|
||||
"target_table": "喷涂寄出_YEAR2026",
|
||||
"pk_col": "ID"
|
||||
},
|
||||
"车波纹": {
|
||||
"target_schema": "machining",
|
||||
"target_table": "车波纹_YEAR2026",
|
||||
"pk_col": "ID"
|
||||
}
|
||||
},
|
||||
r"\\192.168.110.114\生产进度表\2026年数据\计划.accdb": {
|
||||
@@ -296,155 +311,12 @@ SYNC_MAPPING = {
|
||||
"target_table": "执行卡下发记录_YEAR2026",
|
||||
"pk_col": "ID"
|
||||
}
|
||||
},
|
||||
r"\\192.168.110.114\生产进度表\2026年数据\弯管车间.accdb": {
|
||||
"烘洗": {
|
||||
"target_schema": "tubeBending",
|
||||
"target_table": "烘洗_YEAR2026",
|
||||
"pk_col": "ID"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
# ================= Excel 文件配置 =================
|
||||
# 从原 update_config.py 抽取
|
||||
EXCEL_CONFIGS = [
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2022.xlsm",
|
||||
"sheet_names": ["Sheet1"],
|
||||
"contract_year": "2022",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2023（1-5月）.xlsm",
|
||||
"sheet_names": ["Sheet1"],
|
||||
"contract_year": "2023",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2023(6月-.xlsm",
|
||||
"sheet_names": ["Sheet1"],
|
||||
"contract_year": "2023",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2024 6月.xlsm",
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"contract_year": "2024",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2024.xlsm",
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"contract_year": "2024",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\生产执行卡2025年.xlsm",
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"contract_year": "2025",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\生产执行卡2026年.xlsm",
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"contract_year": "2026",
|
||||
"field_mapping": {
|
||||
"产品型号": "选型型号",
|
||||
"备注": "备注1",
|
||||
"下单日期": "接单日期"
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
# ================= 迁移任务配置 =================
|
||||
# 从原 migration.py 抽取
|
||||
MIGRATION_TASKS = [
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2022.xlsm",
|
||||
"year": 2022,
|
||||
"sheet_names": ["Sheet1"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2023（1-5月）.xlsm",
|
||||
"year": 2023,
|
||||
"sheet_names": ["Sheet1"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2023(6月-.xlsm",
|
||||
"year": 2023,
|
||||
"sheet_names": ["Sheet1"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2024 6月.xlsm",
|
||||
"year": 2024,
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2024.xlsm",
|
||||
"year": 2024,
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\生产执行卡2025年.xlsm",
|
||||
"year": 2025,
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
},
|
||||
{
|
||||
"file_path": r"\\192.168.110.113\生产执行卡\生产执行卡2026年.xlsm",
|
||||
"year": 2026,
|
||||
"sheet_names": ["重庆数据","北京数据"],
|
||||
"mapping": {
|
||||
"车间号": "车间号",
|
||||
"工令号": "工令号",
|
||||
"客户型号": "客户型号"
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
75
db_utils.py
75
db_utils.py
@@ -1,7 +1,82 @@
|
||||
# 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+
|
||||
|
||||
215
docs/full-sync.md
Normal file
215
docs/full-sync.md
Normal file
@@ -0,0 +1,215 @@
|
||||
# 全量同步机制 (init_full_sync.py)
|
||||
|
||||
## 概述
|
||||
|
||||
全量同步用于初始化或重建 SQL Server 目标表，将 Access 数据源中的**全部数据**一次性加载到 SQL Server。通常在系统初始化、数据修复或新增表映射时执行。
|
||||
|
||||
- **入口脚本**: `init_full_sync.py`
|
||||
- **计划任务**: `AutoRun-init_full_sync`（用户登录时自动启动）
|
||||
- **批量大小**: 10,000 行（`BATCH_SIZE`）
|
||||
|
||||
## 整体架构
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
subgraph 数据源
|
||||
A1["Access 文件 1<br/>成品入库.accdb"]
|
||||
A2["Access 文件 2<br/>25年压力表合同数据.accdb"]
|
||||
A3["Access 文件 N<br/>..."]
|
||||
end
|
||||
|
||||
subgraph 全量同步
|
||||
S[init_full_sync.py]
|
||||
end
|
||||
|
||||
subgraph SQL Server - CompanyDB
|
||||
T1["[schema1].[table1]"]
|
||||
T2["[schema2].[table2]"]
|
||||
T3["[schemaN].[tableN]"]
|
||||
end
|
||||
|
||||
subgraph 通知
|
||||
N[ntfy 推送通知]
|
||||
end
|
||||
|
||||
A1 & A2 & A3 -->|SELECT *| S
|
||||
S -->|TRUNCATE + INSERT| T1 & T2 & T3
|
||||
S -->|汇总通知| N
|
||||
```
|
||||
|
||||
## 同步流程
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
START([开始全量同步]) --> SQL_CONN[连接 SQL Server]
|
||||
SQL_CONN --> FILE_LOOP{{遍历 SYNC_MAPPING<br/>中的每个 Access 文件}}
|
||||
|
||||
FILE_LOOP --> FILE_CHECK{文件是否存在?}
|
||||
FILE_CHECK -->|不存在| SKIP[跳过该文件]
|
||||
FILE_CHECK -->|存在| ACC_CONN[连接 Access 数据库]
|
||||
|
||||
ACC_CONN --> TABLE_LOOP{{遍历该文件下的<br/>每个表映射}}
|
||||
|
||||
TABLE_LOOP --> ACC_READ[读取 Access 表结构<br/>SELECT TOP 1 * FROM table]
|
||||
|
||||
ACC_READ --> TARGET_CHECK{目标表是否存在?}
|
||||
|
||||
TARGET_CHECK -->|不存在| AUTO_CREATE[自动建表<br/>根据 Access schema 创建]
|
||||
TARGET_CHECK -->|存在| TRUNCATE[TRUNCATE 目标表]
|
||||
AUTO_CREATE --> DDL_COMMIT[DDL 单独提交]
|
||||
DDL_COMMIT --> IDENT_CHECK
|
||||
TRUNCATE --> IDENT_CHECK
|
||||
|
||||
IDENT_CHECK{含 IDENTITY 列?}
|
||||
IDENT_CHECK -->|是| ID_ON[SET IDENTITY_INSERT ON]
|
||||
IDENT_CHECK -->|否| DATA_TRANSFER
|
||||
ID_ON --> DATA_TRANSFER
|
||||
|
||||
DATA_TRANSFER[批量数据传输<br/>fetchmany BATCH_SIZE]
|
||||
DATA_TRANSFER --> HAS_MORE{{还有数据?}}
|
||||
|
||||
HAS_MORE -->|有| BATCH[executemany 插入一个批次<br/>记录进度日志]
|
||||
BATCH --> HAS_MORE
|
||||
|
||||
HAS_MORE -->|无| ID_OFF_CHECK{IDENTITY_INSERT<br/>是否已开启?}
|
||||
ID_OFF_CHECK -->|是| ID_OFF[SET IDENTITY_INSERT OFF]
|
||||
ID_OFF_CHECK -->|否| COMMIT
|
||||
ID_OFF --> COMMIT[提交事务]
|
||||
|
||||
COMMIT --> STATS[记录表级统计<br/>行数 / 速率 / 用时]
|
||||
|
||||
STATS --> NEXT_TABLE{{下一张表?}}
|
||||
NEXT_TABLE -->|是| TABLE_LOOP
|
||||
NEXT_TABLE -->|否| CLOSE_ACC[关闭 Access 连接]
|
||||
CLOSE_ACC --> FILE_SUMMARY[输出文件级汇总]
|
||||
|
||||
SKIP --> FILE_LOOP
|
||||
FILE_SUMMARY --> FILE_LOOP
|
||||
|
||||
FILE_LOOP -->|全部文件处理完| FINAL[输出全局汇总]
|
||||
FINAL --> NTFY[发送 ntfy 推送通知]
|
||||
NTFY --> END([结束])
|
||||
|
||||
TABLE_LOOP -->|异常| ERR_HANDLE[记录失败日志<br/>回滚事务<br/>清理 IDENTITY_INSERT]
|
||||
ERR_HANDLE --> NEXT_TABLE
|
||||
|
||||
style AUTO_CREATE fill:#bbf,stroke:#333
|
||||
style DATA_TRANSFER fill:#bfb,stroke:#333
|
||||
```
|
||||
|
||||
## 数据传输细节
|
||||
|
||||
### 批量读取与插入
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant ACC as Access 数据库
|
||||
participant PY as Python 脚本
|
||||
participant SQL as SQL Server
|
||||
|
||||
PY->>ACC: SELECT * FROM [table]
|
||||
loop 每 BATCH_SIZE 行
|
||||
ACC-->>PY: fetchmany(10000)
|
||||
PY->>SQL: executemany(INSERT, rows)
|
||||
Note over PY: 每 5 秒或每 10000 行<br/>记录一次进度
|
||||
end
|
||||
PY->>SQL: COMMIT
|
||||
```
|
||||
|
||||
### 进度日志
|
||||
|
||||
传输过程中按时间和行数双条件输出进度：
|
||||
|
||||
```
|
||||
表 [成品入库] → [成品入库记录]
|
||||
检测到 25 个列
|
||||
已清空目标表
|
||||
开始数据传输...
|
||||
进度: 10,000 行 | 速率: 45,000 行/秒
|
||||
进度: 20,000 行 | 速率: 43,500 行/秒
|
||||
表 [成品入库记录] 完成: 23,456 行 | 速率: 44,200 行/秒 | 用时: 0.5秒
|
||||
```
|
||||
|
||||
## 自动建表
|
||||
|
||||
当目标表在 SQL Server 中不存在时，系统自动创建：
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
A[Access cursor.description] --> B[_access_col_to_sql<br/>类型映射]
|
||||
B --> C["CREATE TABLE [schema].[table] (<br/> [col1] INT IDENTITY(1,1) PRIMARY KEY,<br/> [col2] NVARCHAR(100),<br/> ...<br/>)"]
|
||||
C --> D[DDL 单独提交<br/>确保后续参数绑定正常]
|
||||
```
|
||||
|
||||
DDL 建表后**必须单独 `commit()`**，否则 pyodbc 的参数绑定无法获取正确的列元数据。
|
||||
|
||||
### 类型映射
|
||||
|
||||
| Access 类型 | SQL Server 类型 | 备注 |
|
||||
|-------------|----------------|------|
|
||||
| `int` (主键) | `INT IDENTITY(1,1) PRIMARY KEY` | 自增主键 |
|
||||
| `int` (非主键) | `INT` | |
|
||||
| `float` | `FLOAT` | |
|
||||
| `bool` | `BIT` | |
|
||||
| `datetime.datetime` | `DATETIME` | |
|
||||
| `decimal.Decimal` | `DECIMAL(p, s)` | 保留精度 |
|
||||
| `str` (size ≤ 4000) | `NVARCHAR(size)` | |
|
||||
| `str` (size > 4000) | `NVARCHAR(MAX)` | Memo 字段 |
|
||||
|
||||
## IDENTITY_INSERT 处理
|
||||
|
||||
SQL Server 中含标识列（自增列）的表在插入显式 ID 值时，必须开启 `IDENTITY_INSERT`：
|
||||
|
||||
```mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> Off
|
||||
Off --> On: SET IDENTITY_INSERT ON
|
||||
On --> Inserting: INSERT with explicit ID values
|
||||
Inserting --> On: executemany 完成
|
||||
On --> Off: SET IDENTITY_INSERT OFF
|
||||
Off --> [*]: COMMIT
|
||||
|
||||
note right of On: 会话级设置<br/>不随事务回滚
|
||||
```
|
||||
|
||||
关键点：
|
||||
- `IDENTITY_INSERT` 是**会话级设置**，每个连接同一时刻只能对一张表开启
|
||||
- 异常时必须在 `except` 块中显式关闭，否则后续同表同步会报错
|
||||
- 不随事务 `ROLLBACK` 回滚，必须手动关闭
|
||||
|
||||
## 通知机制
|
||||
|
||||
全量同步完成后通过 **ntfy** 发送汇总通知：
|
||||
|
||||
```
|
||||
🎯 全量同步完成
|
||||
|
||||
✅ 成功 15 张表:
|
||||
• 成品入库记录: 23,456 行, 0.5秒
|
||||
• 25年压力表合同数据: 12,345 行, 0.3秒
|
||||
...
|
||||
|
||||
❌ 失败 1 张表:
|
||||
• 某表名: 错误信息摘要...
|
||||
|
||||
总计: 35,801 行 | 用时: 2.3分钟
|
||||
```
|
||||
|
||||
- 全部成功：普通优先级
|
||||
- 存在失败：高优先级 + warning 标签
|
||||
|
||||
## 与增量同步的对比
|
||||
|
||||
| 维度 | 全量同步 | 增量同步 |
|
||||
|------|---------|---------|
|
||||
| **触发方式** | 用户登录 / 手动执行 | 持续轮询服务 |
|
||||
| **数据范围** | 全部数据 | 仅变更记录 |
|
||||
| **目标表处理** | TRUNCATE + 全量 INSERT | DELETE 旧 + INSERT 新（按主键） |
|
||||
| **校验机制** | 无逐条校验 | 逐主键校验 + 提交后复核 |
|
||||
| **适用场景** | 初始化、数据重建 | 日常实时同步 |
|
||||
| **运行时长** | 一次性执行完毕 | 常驻后台运行 |
|
||||
| **数据完整性** | 依赖 TRUNCATE 原子性 | 事务 + 校验双重保障 |
|
||||
|
||||
## 配置参考
|
||||
|
||||
全量同步的行为由 `config.py` 中的 `SYNC_MAPPING` 驱动，新增表映射后首次运行全量同步即可自动建表并填充数据。
|
||||
210
docs/incremental-sync.md
Normal file
210
docs/incremental-sync.md
Normal file
@@ -0,0 +1,210 @@
|
||||
# 增量同步机制 (run_incremental_sync.py)
|
||||
|
||||
## 概述
|
||||
|
||||
增量同步是一个**长轮询服务**，持续监控 SQL Server 中的 `TableChangeLog` 变更日志表，将 Access 数据源的变更实时同步到 SQL Server 目标表。
|
||||
|
||||
- **入口脚本**: `run_incremental_sync.py`
|
||||
- **计划任务**: `AutoRun-run_incremental_sync`（用户登录时自动启动）
|
||||
- **轮询间隔**: 30 秒（`POLL_INTERVAL`）
|
||||
|
||||
## 整体架构
|
||||
|
||||
```mermaid
|
||||
graph LR
|
||||
subgraph 数据源侧
|
||||
A[Access .accdb 文件] -->|VBA 宏写入日志| B[TableChangeLog]
|
||||
end
|
||||
|
||||
subgraph 增量同步服务
|
||||
B -->|轮询 Synced=0| C[run_incremental_sync.py]
|
||||
C -->|读最新数据| A
|
||||
C -->|DELETE + INSERT| D[SQL Server 目标表]
|
||||
C -->|标记 Synced=1| B
|
||||
end
|
||||
|
||||
subgraph 监控
|
||||
C -->|心跳| E[Uptime Kuma]
|
||||
end
|
||||
```
|
||||
|
||||
## 变更日志驱动
|
||||
|
||||
Access 端的 VBA 宏在数据变更时，向 `TableChangeLog` 表写入一条记录：
|
||||
|
||||
| 字段 | 说明 |
|
||||
|------|------|
|
||||
| `LogID` | 自增主键 |
|
||||
| `TableAddress` | Access 文件路径（多种格式） |
|
||||
| `TableName` | 发生变更的 Access 表名 |
|
||||
| `RecordID` | 变更记录的主键值 |
|
||||
| `Synced` | 同步标记，0=未同步，1=已同步 |
|
||||
|
||||
### 路径匹配策略
|
||||
|
||||
VBA 端写入的 `TableAddress` 有多种格式，系统构造 4 种候选值进行匹配：
|
||||
|
||||
```
|
||||
;DATABASE=\\192.168.110.114\生产进度表\成品入库.accdb ← 网络路径前缀
|
||||
LOCAL=\\server\share\file.accdb ← 本地等号前缀
|
||||
LOCAL:\\server\share\file.accdb ← 本地冒号前缀
|
||||
\\192.168.110.114\生产进度表\成品入库.accdb ← 裸路径
|
||||
```
|
||||
|
||||
## 核心同步流程
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
START([服务启动]) --> POLL[轮询 TableChangeLog<br/>WHERE Synced=0]
|
||||
|
||||
POLL -->|无记录| SLEEP[休眠 POLL_INTERVAL 秒]
|
||||
SLEEP --> POLL
|
||||
|
||||
POLL -->|发现未同步记录| GROUP[按 TableName 分组<br/>合并同一表的 record_ids]
|
||||
|
||||
GROUP --> CONNECT[连接 Access 数据库]
|
||||
|
||||
CONNECT --> FOR_EACH{{遍历每个表}}
|
||||
|
||||
FOR_EACH --> CHECK_TABLE{目标表是否存在?}
|
||||
|
||||
CHECK_TABLE -->|不存在| AUTO_CREATE[自动建表<br/>ensure_schema + create_table_from_access]
|
||||
CHECK_TABLE -->|存在| READ_ACCESS
|
||||
AUTO_CREATE --> READ_ACCESS
|
||||
|
||||
READ_ACCESS[A. 从 Access 读取最新数据<br/>SELECT WHERE PK IN ...]
|
||||
|
||||
READ_ACCESS --> EXTRACT_PK[提取 inserted_pk_set<br/>Access 实际返回的主键集合]
|
||||
|
||||
EXTRACT_PK --> DELETE[B. 删除目标表旧记录<br/>DELETE WHERE PK IN ...]
|
||||
|
||||
DELETE --> HAS_NEW{{有新数据?}}
|
||||
|
||||
HAS_NEW -->|有| IDENTITY_CHECK{含 IDENTITY 列?}
|
||||
IDENTITY_CHECK -->|是| ID_ON[SET IDENTITY_INSERT ON]
|
||||
IDENTITY_CHECK -->|否| INSERT
|
||||
ID_ON --> INSERT[C. 批量插入新记录<br/>executemany]
|
||||
INSERT --> ID_OFF[SET IDENTITY_INSERT OFF]
|
||||
ID_OFF --> VERIFY
|
||||
|
||||
HAS_NEW -->|无| VERIFY
|
||||
|
||||
VERIFY[D. 提交前逐主键校验] --> VERIFY_OK{校验通过?}
|
||||
|
||||
VERIFY_OK -->|通过| MARK_SYNCED[E. 标记 Synced=1]
|
||||
MARK_SYNCED --> COMMIT[F. 提交事务]
|
||||
COMMIT --> POST_VERIFY
|
||||
|
||||
POST_VERIFY{提交后复核<br/>ENABLE_POST_COMMIT_VERIFY} -->|关闭| SUCCESS
|
||||
POST_VERIFY -->|开启| POST_OK{复核通过?}
|
||||
POST_OK -->|通过| SUCCESS[记录同步成功日志]
|
||||
POST_OK -->|失败| REVERT[回滚 Synced=0<br/>下一轮重试]
|
||||
|
||||
VERIFY_OK -->|失败| ROLLBACK[回滚事务<br/>保持 Synced=0]
|
||||
ROLLBACK --> NEXT_TABLE
|
||||
|
||||
SUCCESS --> NEXT_TABLE
|
||||
REVERT --> NEXT_TABLE
|
||||
|
||||
NEXT_TABLE{{下一张表?}} -->|是| FOR_EACH
|
||||
NEXT_TABLE -->|否| CLOSE_ACC[关闭 Access 连接]
|
||||
CLOSE_ACC --> SUMMARY[输出文件级汇总]
|
||||
SUMMARY --> POLL
|
||||
|
||||
style VERIFY fill:#f9f,stroke:#333
|
||||
style POST_VERIFY fill:#f9f,stroke:#333
|
||||
style AUTO_CREATE fill:#bbf,stroke:#333
|
||||
```
|
||||
|
||||
## 逐主键校验机制
|
||||
|
||||
传统的"数量对比"方法存在缺陷：少插和漏删的错误可能互相抵消。本系统采用**逐主键确认**方式：
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
subgraph 输入
|
||||
A[record_ids<br/>日志中的主键列表]
|
||||
B[inserted_pk_set<br/>Access 实际读到的主键]
|
||||
end
|
||||
|
||||
subgraph 删除校验
|
||||
C[差集: record_ids - inserted_pk_set<br/>= 应删除的主键]
|
||||
D[查询目标表<br/>这些主键是否还存在?]
|
||||
C --> D
|
||||
D -->|还存在任一条| E[❌ 校验失败]
|
||||
D -->|全部不存在| F[✅ 删除校验通过]
|
||||
end
|
||||
|
||||
subgraph 插入校验
|
||||
G[查询目标表<br/>inserted_pk_set 是否都在?]
|
||||
G -->|有任一条查不到| E
|
||||
G -->|全部存在| H[✅ 插入校验通过]
|
||||
end
|
||||
|
||||
A --> C
|
||||
B --> C
|
||||
B --> G
|
||||
```
|
||||
|
||||
### 校验函数说明
|
||||
|
||||
| 函数 | 作用 |
|
||||
|------|------|
|
||||
| `fetch_existing_pks()` | 分批查询目标表，返回实际存在的主键集合（IN 子句每批不超过 900 个参数） |
|
||||
| `verify_sync_result()` | 执行删除校验 + 插入校验，失败时抛出 `SyncVerificationError` |
|
||||
| `_norm_key()` | 主键归一化为字符串，规避 Access/SQL Server 类型差异 |
|
||||
| `_chunked()` | 将列表分批，避免超出 SQL Server 参数上限 |
|
||||
|
||||
## 故障恢复
|
||||
|
||||
```mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> Pending: VBA 写入日志 Synced=0
|
||||
Pending --> Syncing: 同步服务读取
|
||||
Syncing --> Verified: 校验通过 + 提交
|
||||
Verified --> Committed: Synced=1 标记生效
|
||||
Committed --> [*]: 同步完成
|
||||
|
||||
Syncing --> Rollback: 校验失败 / 异常
|
||||
Rollback --> Pending: 事务回滚 Synced=0<br/>下一轮自动重试
|
||||
|
||||
Committed --> Pending: 提交后复核失败<br/>Synced 撤回为 0
|
||||
```
|
||||
|
||||
### 三层保障
|
||||
|
||||
1. **提交前校验** — 删/插完成后、标记 Synced=1 前，逐主键确认结果正确
|
||||
2. **事务回滚** — 校验失败或异常时回滚事务，保持 `Synced=0`，下一轮自动重试
|
||||
3. **提交后复核** — `commit()` 后再查一次数据库确认数据已持久化（可通过 `ENABLE_POST_COMMIT_VERIFY` 开关控制）
|
||||
|
||||
## 轮询策略
|
||||
|
||||
```
|
||||
while True:
|
||||
has_work = process_sync_task()
|
||||
|
||||
if has_work:
|
||||
sleep(0.1) # 有积压，快速重试
|
||||
else:
|
||||
sleep(30) # 无工作，标准间隔
|
||||
```
|
||||
|
||||
有未处理数据时以 0.1 秒间隔快速处理积压；无数据时按 `POLL_INTERVAL` 休眠。
|
||||
|
||||
## 自动建表
|
||||
|
||||
当目标表在 SQL Server 中不存在时，系统根据 Access 表的列定义自动建表：
|
||||
|
||||
```
|
||||
Access cursor.description → 类型映射 → CREATE TABLE 语句
|
||||
```
|
||||
|
||||
| Access 类型 | SQL Server 类型 |
|
||||
|-------------|----------------|
|
||||
| `int` | `INT`（主键时追加 `IDENTITY(1,1) PRIMARY KEY`） |
|
||||
| `float` | `FLOAT` |
|
||||
| `bool` | `BIT` |
|
||||
| `datetime.datetime` | `DATETIME` |
|
||||
| `decimal.Decimal` | `DECIMAL(p, s)` |
|
||||
| `str` (size ≤ 4000) | `NVARCHAR(size)` |
|
||||
| `str` (size > 4000) | `NVARCHAR(MAX)` |
|
||||
@@ -0,0 +1,168 @@
|
||||
# 移除计划：移除 Excel 迁移相关内容
|
||||
|
||||
- **日期**: 2026-06-17
|
||||
- **状态**: 待审核（通过后执行）
|
||||
- **范围**: 仅代码仓库 `D:\python\BLD_sync`（不含 SQL Server 表数据）
|
||||
|
||||
---
|
||||
|
||||
## 1. 背景与目标
|
||||
|
||||
项目不再处理 Excel 文档（`.xlsm` 生产执行卡）的迁移工作。后续仅保留 **Access → SQL Server** 的同步链路（全量 `init_full_sync.py` + 增量 `run_incremental_sync.py`）。
|
||||
|
||||
目标：把所有"Excel 迁移"相关代码、配置、依赖与文档清理干净，使仓库成为一个干净、自洽的 **Access-only** 同步系统，不残留无用导入、死代码或过时说明。
|
||||
|
||||
### 已确认的范围决策
|
||||
|
||||
| 决策点 | 选择 |
|
||||
|---|---|
|
||||
| 清理彻底程度 | **彻底（Thorough）**：代码+配置+依赖+缓存目录+`CLAUDE.md`/文档修正 |
|
||||
| Excel 写入的 3 张 SQL 表 | **保留不动**（`warehouseOutbound.executionCardData` / `contractData` / `customerProductType`，含现有数据） |
|
||||
|
||||
---
|
||||
|
||||
## 2. 方案对比（已选定 Thorough）
|
||||
|
||||
| 方案 | 内容 | 取舍 |
|
||||
|---|---|---|
|
||||
| A. 彻底清理 ✅ | 删脚本+配置+依赖+缓存，更新 CLAUDE.md/.env.example | 仓库自洽，无残留；改动面较大 |
|
||||
| B. 仅核心代码+配置 | 删脚本+配置+依赖+.env.example，留 CLAUDE.md/缓存/文档 | 改动小；但 CLAUDE.md 过时、缓存占盘 |
|
||||
| C. 仅代码，保留依赖 | 仅删脚本+配置符号，pandas/sqlalchemy/openpyxl 留作他用 | 最保守；依赖膨胀 |
|
||||
|
||||
**选定 A（彻底）**：与"项目不再处理 Excel"的意图一致，避免后续误导。
|
||||
|
||||
---
|
||||
|
||||
## 3. 受影响资产清单（已核实）
|
||||
|
||||
经全仓库（排除 `.venv`）grep 核实，以下符号/依赖**仅被 Excel 脚本使用**：
|
||||
|
||||
- `pandas`、`sqlalchemy`、`openpyxl` → 仅 `excel_sync_to_sql.py`、`migration.py` 使用
|
||||
- `EXCEL_CONFIGS`、`MIGRATION_TASKS`、`TABLE_SCHEMA`、`CONTRACT_MAPPING`、`EXECUTION_CARD_FIELDS`、`CONTRACT_DATA_FIELDS`、`CONTRACT_DATA_MAPPING`、`CACHE_DIR`、`TEMP_DIR`、`EXCEL_SYNC_INTERVAL`、`EXCEL_SYNC_UPTIME_KUMA_CONFIG` → 仅被 Excel 脚本或其配置链使用
|
||||
- 保留脚本（`init_full_sync.py` / `run_incremental_sync.py` / `db_utils.py` / `ntfy_utils.py` / `log_utils.py` / `uptime_kuma_utils.py` / `check_drivers.py` / `archive_existing_logs.py` / `vbareplace.py` / `vba.txt`）**不依赖**上述任何 Excel 符号 → 删除后无悬空引用。
|
||||
|
||||
---
|
||||
|
||||
## 4. 详细变更清单
|
||||
|
||||
### 4.1 整体删除（DELETE）
|
||||
|
||||
| 路径 | 说明 |
|
||||
|---|---|
|
||||
| `excel_sync_to_sql.py` | Excel→SQL 主同步器（`DataSynchronizer`），写 `executionCardData`、生成 `contractData` |
|
||||
| `migration.py` | Excel→`warehouseOutbound.customerProductType` 迁移 |
|
||||
| `config/field_mappings.py` | 仅含 `TABLE_SCHEMA`、`CONTRACT_MAPPING`（均 Excel 专用）→ 整文件删除 |
|
||||
| `temp/`（未纳入 git，~180MB） | Excel 本地缓存（`.xlsm` + `sync_log.txt`）→ 删除目录 |
|
||||
|
||||
> `tmp/` 为空且未被代码引用，保留不动。
|
||||
|
||||
### 4.2 配置编辑（EDIT）
|
||||
|
||||
#### `config/__init__.py`
|
||||
- 删除 `from .field_mappings import TABLE_SCHEMA, CONTRACT_MAPPING`（整行）
|
||||
- `from .file_sources import ...` 改为仅 `SYNC_MAPPING`
|
||||
- `from .app_settings import ...` 改为仅 `LOG_TABLE_CONFIG, NTFY_CONFIG, UPTIME_KUMA_CONFIG, POLL_INTERVAL, BATCH_SIZE`
|
||||
- `__all__` 移除：`EXCEL_CONFIGS`、`MIGRATION_TASKS`、`TABLE_SCHEMA`、`CONTRACT_MAPPING`、`EXCEL_SYNC_UPTIME_KUMA_CONFIG`、`EXCEL_SYNC_INTERVAL`、`CACHE_DIR`、`TEMP_DIR`、`EXECUTION_CARD_FIELDS`、`CONTRACT_DATA_FIELDS`、`CONTRACT_DATA_MAPPING`
|
||||
|
||||
最终 `__init__.py` 导出：`SQL_SERVER_CONFIG, SQL_SERVER_CONN, DB_CONFIG, ACCESS_DRIVER, SYNC_MAPPING, LOG_TABLE_CONFIG, NTFY_CONFIG, UPTIME_KUMA_CONFIG, POLL_INTERVAL, BATCH_SIZE`
|
||||
|
||||
#### `config/file_sources.py`
|
||||
- 删除 `EXCEL_CONFIGS`（含其上方注释 `# ================= Excel 文件配置 =================`）
|
||||
- 删除 `MIGRATION_TASKS`（含其上方注释 `# ================= 迁移任务配置 =================`）
|
||||
- 保留 `SYNC_MAPPING`（Access 映射，不动）
|
||||
|
||||
#### `config/app_settings.py`
|
||||
- 删除 `EXCEL_SYNC_UPTIME_KUMA_CONFIG`
|
||||
- 删除 `EXCEL_SYNC_INTERVAL`
|
||||
- 删除 `CACHE_DIR`、`TEMP_DIR`（"运行参数"段仅留 `POLL_INTERVAL`、`BATCH_SIZE`）
|
||||
- 删除 `EXECUTION_CARD_FIELDS`、`CONTRACT_DATA_FIELDS`、`CONTRACT_DATA_MAPPING`（"字段配置"整段）
|
||||
- 保留 `LOG_TABLE_CONFIG`、`NTFY_CONFIG`、`UPTIME_KUMA_CONFIG`、`POLL_INTERVAL`、`BATCH_SIZE`
|
||||
- `import os` 保留（`NTFY_CONFIG`/`UPTIME_KUMA_CONFIG` 仍用 `os.environ.get`）
|
||||
|
||||
### 4.3 依赖与环境（EDIT）
|
||||
|
||||
#### `requirements.txt`
|
||||
移除：
|
||||
```
|
||||
pandas>=1.5.0
|
||||
sqlalchemy>=2.0.0
|
||||
openpyxl>=3.0.0
|
||||
```
|
||||
保留：`pyodbc`、`python-dotenv`、`requests`（仍被 ntfy/uptime/config 使用）
|
||||
|
||||
#### `.env.example`
|
||||
移除行：`EXCEL_SYNC_UPTIME_KUMA_PUSH_URL=`
|
||||
|
||||
### 4.4 文档修正（EDIT）
|
||||
|
||||
#### `CLAUDE.md`
|
||||
当前 CLAUDE.md 已过时（引用了不存在的 `etl_manager.py`、`sync_excel_to_sql.py`、`update_config.py`、`config.py`）。借此一并修正为真实结构（`config/` 包 + 仅 Access 脚本）：
|
||||
- **Project Overview**：删除 "Excel files (.xlsm)..." 条目
|
||||
- **Data Flow / 架构图**：删除 Excel 源、删除 `etl_manager.py`/`sync_excel_to_sql.py`/`migration.py` 脚本框
|
||||
- **Key Components**：删除 `etl_manager.py`、`sync_excel_to_sql.py`；`config` 描述去掉 `EXCEL_CONFIGS`/`TABLE_SCHEMA`
|
||||
- **Common Tasks**：删除 "Run Excel to SQL Sync" 与 `migration.py` 小节
|
||||
- **Configuration Management**：删除 `update_config.py`/`EXCEL_CONFIGS`，改为描述 `config/` 包结构（`database.py`/`file_sources.py`/`app_settings.py`）
|
||||
- **Database Schema**：`warehouseOutbound` 描述更新（其表为历史 Excel 写入，现不再更新）
|
||||
- 删除所有 Excel 相关实现细节（MERGE 生成 contractData 等）
|
||||
|
||||
#### 轻量文档串修正（彻底清理）
|
||||
- `uptime_kuma_utils.py` 顶部 docstring：删除"excel_sync_to_sql.py 等仍在使用"字样（向后兼容接口保留为通用工具方法，不删）
|
||||
- `log_utils.py` / `archive_existing_logs.py` docstring 中 `excel_sync_...` 文件名示例：可选移除（文件名解析是通用正则，功能不受影响）
|
||||
|
||||
### 4.5 不在仓库内、需手动处理（仅提示，本计划不自动执行）
|
||||
|
||||
| 项目 | 动作 |
|
||||
|---|---|
|
||||
| Windows 任务计划程序 | 若存在 `AutoRun-excel_sync` 之类计划任务，手动禁用/删除（保留 `AutoRun-init_full_sync`、`AutoRun-run_incremental_sync`） |
|
||||
| Uptime Kuma | 禁用/删除 Excel 同步心跳监控项（对应 `EXCEL_SYNC_UPTIME_KUMA_PUSH_URL`） |
|
||||
| `.env`（已 gitignore） | 手动删除其中的 `EXCEL_SYNC_UPTIME_KUMA_PUSH_URL=...` 行 |
|
||||
| SQL Server 表 | **按决策保留不动**，无需任何操作 |
|
||||
|
||||
---
|
||||
|
||||
## 5. 执行顺序
|
||||
|
||||
1. **删除文件**：`excel_sync_to_sql.py`、`migration.py`、`config/field_mappings.py`、`temp/`
|
||||
2. **编辑配置**：`config/__init__.py` → `config/file_sources.py` → `config/app_settings.py`
|
||||
3. **编辑依赖/环境**：`requirements.txt`、`.env.example`
|
||||
4. **编辑文档**：`CLAUDE.md` + 上述轻量 docstring
|
||||
5. **验证**（见第 6 节）
|
||||
6. **提交**：单条 commit，英文信息（如 `refactor: remove Excel migration pipeline`），含删除/修改；提交后按全局规则推送到远端
|
||||
|
||||
> 全程在 `.venv` 内执行（遵循全局 CLAUDE.md 协议）。
|
||||
|
||||
---
|
||||
|
||||
## 6. 验证计划
|
||||
|
||||
执行后用仓库 `.venv` 的 Python 逐项核验：
|
||||
|
||||
1. **配置包导入自洽**：
|
||||
```bash
|
||||
.venv/Scripts/python -c "import config; print(config.SYNC_MAPPING is not None)"
|
||||
```
|
||||
2. **所有保留脚本语法编译通过**：
|
||||
```bash
|
||||
.venv/Scripts/python -m py_compile init_full_sync.py run_incremental_sync.py db_utils.py ntfy_utils.py log_utils.py uptime_kuma_utils.py check_drivers.py archive_existing_logs.py vbareplace.py config/*.py
|
||||
```
|
||||
3. **无悬空引用**：grep 确认仓库（排除 `.venv`）不再出现已删符号：
|
||||
`EXCEL_CONFIGS|MIGRATION_TASKS|TABLE_SCHEMA|CONTRACT_MAPPING|EXECUTION_CARD_FIELDS|CONTRACT_DATA_FIELDS|CONTRACT_DATA_MAPPING|CACHE_DIR|TEMP_DIR|EXCEL_SYNC_INTERVAL|EXCEL_SYNC_UPTIME_KUMA_CONFIG|read_excel|openpyxl`
|
||||
4. **增量/全量脚本可正常进入主流程**（可选冒烟）：分别运行 `run_incremental_sync.py`、`init_full_sync.py` 数秒后中断，确认无 ImportError、能连接 SQL Server。
|
||||
|
||||
---
|
||||
|
||||
## 7. 回滚
|
||||
|
||||
全部变更均在 git 跟踪范围内（`temp/` 除外，但其为可再生缓存）。如需回滚：
|
||||
```bash
|
||||
git revert <commit-sha>
|
||||
```
|
||||
`temp/` 缓存可由历史 Excel 脚本重新生成（已无意义）。
|
||||
|
||||
---
|
||||
|
||||
## 8. 风险与说明
|
||||
|
||||
- **无数据风险**：不动 SQL Server 任何表与数据。
|
||||
- **无运行中服务风险**：`run_incremental_sync.py`（Access 增量）代码路径不变，导入符号均保留。
|
||||
- **CLAUDE.md 改动较大**：因原文已与实际代码脱节，顺带修正为真实结构；如只希望"最小改动"可告知，仅删 Excel 段落、不补真实结构。
|
||||
315
etl_manager.py
315
etl_manager.py
@@ -1,315 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import shutil
|
||||
import logging
|
||||
import argparse
|
||||
import datetime
|
||||
import urllib.parse
|
||||
import warnings
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.engine import URL
|
||||
from sqlalchemy.types import NVARCHAR, Integer, Date
|
||||
|
||||
# 导入配置
|
||||
from log_utils import (log_error, log_warning, log_info, log_processing, log_file, log_sync,
|
||||
log_start, log_complete, LoggerManager)
|
||||
from config import DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA
|
||||
|
||||
# ================= 抑制 openpyxl 的数据验证警告 =================
|
||||
warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl')
|
||||
|
||||
class DataSynchronizer:
|
||||
def __init__(self, force_sync=False):
|
||||
self.force_sync = force_sync
|
||||
self.engine = self._get_db_connection()
|
||||
self.cache_dir = CACHE_DIR
|
||||
|
||||
if not os.path.exists(self.cache_dir):
|
||||
os.makedirs(self.cache_dir)
|
||||
|
||||
def _get_db_connection(self):
|
||||
connection_string = (
|
||||
f"DRIVER={{{DB_CONFIG['driver']}}};"
|
||||
f"SERVER={DB_CONFIG['server']};"
|
||||
f"DATABASE={DB_CONFIG['database']};"
|
||||
f"UID={DB_CONFIG['username']};"
|
||||
f"PWD={DB_CONFIG['password']};"
|
||||
f"TrustServerCertificate={DB_CONFIG.get('TrustServerCertificate', 'yes')};"
|
||||
)
|
||||
connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})
|
||||
return create_engine(connection_url, fast_executemany=True)
|
||||
|
||||
def _should_process_file(self, remote_path, local_path):
|
||||
if self.force_sync:
|
||||
return True, "强制同步"
|
||||
|
||||
if not os.path.exists(local_path):
|
||||
return True, "缓存不存在"
|
||||
|
||||
try:
|
||||
remote_mtime = os.path.getmtime(remote_path)
|
||||
local_mtime = os.path.getmtime(local_path)
|
||||
if remote_mtime > local_mtime + 1:
|
||||
return True, f"源文件更新"
|
||||
except OSError as e:
|
||||
log_error(f"无法访问源文件: {remote_path}, Error: {e}")
|
||||
return False, "源文件无法访问"
|
||||
|
||||
return False, "文件未变更"
|
||||
|
||||
def _clean_dataframe(self, df, contract_year):
|
||||
"""主表数据清洗与验证"""
|
||||
# 1. 设置合同年份
|
||||
df['合同年份'] = contract_year
|
||||
|
||||
# 2. 移除总排号为空的行
|
||||
if '总排号' in df.columns:
|
||||
df = df.dropna(subset=['总排号'])
|
||||
df = df[df['总排号'].astype(str).str.strip() != '']
|
||||
else:
|
||||
log_error("数据源中找不到映射后的[总排号]列，跳过此 sheet")
|
||||
return None, None
|
||||
|
||||
# ★ 新增：去除重复的总排号（保留第一条）
|
||||
if '总排号' in df.columns:
|
||||
df['总排号'] = df['总排号'].astype(str).str.strip()
|
||||
duplicates = df[df.duplicated(subset=['总排号'], keep='first')]
|
||||
if not duplicates.empty:
|
||||
log_warning(f"发现 {len(duplicates)} 条重复的总排号，已自动去重。重复的总排号: {duplicates['总排号'].tolist()[:10]}")
|
||||
df = df.drop_duplicates(subset=['总排号'], keep='first')
|
||||
|
||||
# 3. 补全列
|
||||
for col in TABLE_SCHEMA.keys():
|
||||
if col not in df.columns:
|
||||
df[col] = None
|
||||
|
||||
# 用于存储每一列的 SQL 类型
|
||||
dtype_dict = {}
|
||||
|
||||
# 4. 字段清洗
|
||||
for col, rules in TABLE_SCHEMA.items():
|
||||
if col not in df.columns:
|
||||
continue
|
||||
|
||||
if rules['type'] == 'int':
|
||||
# ★ 修改：先转换为数值，然后四舍五入到整数
|
||||
df[col] = pd.to_numeric(df[col], errors='coerce')
|
||||
# 将浮点数四舍五入为整数（处理如 123.5 这样的值）
|
||||
df[col] = df[col].round(0)
|
||||
# 转换为可空整数类型
|
||||
df[col] = df[col].astype('Int64')
|
||||
# 将 NaN 替换为 None
|
||||
df[col] = df[col].replace({pd.NA: None})
|
||||
dtype_dict[col] = Integer()
|
||||
|
||||
elif rules['type'] == 'date':
|
||||
df[col] = pd.to_datetime(df[col], errors='coerce')
|
||||
df[col] = df[col].apply(lambda x: x.date() if pd.notnull(x) else None)
|
||||
dtype_dict[col] = Date()
|
||||
|
||||
elif rules['type'] == 'str':
|
||||
# 先转换为字符串
|
||||
df[col] = df[col].fillna('').astype(str)
|
||||
# 替换各种空值表示
|
||||
df[col] = df[col].replace({'nan': '', 'None': '', '<NA>': ''})
|
||||
|
||||
# 强制截断
|
||||
max_len = rules.get('max_len', 255)
|
||||
df[col] = df[col].str.slice(0, max_len)
|
||||
# 将空字符串转为 None
|
||||
df[col] = df[col].replace('', None)
|
||||
|
||||
dtype_dict[col] = NVARCHAR(max_len)
|
||||
|
||||
final_cols = list(TABLE_SCHEMA.keys())
|
||||
|
||||
return df[final_cols], dtype_dict
|
||||
|
||||
def _sync_to_db(self, df, dtype_dict):
|
||||
"""同步主表数据 - 使用更稳健的方法"""
|
||||
if df is None or df.empty:
|
||||
return
|
||||
|
||||
target_table = "[warehouseOutbound].[executionCardData]"
|
||||
|
||||
with self.engine.connect() as conn:
|
||||
existing_ids = pd.read_sql(f"SELECT [总排号] FROM {target_table}", conn)
|
||||
|
||||
existing_id_set = set(existing_ids['总排号'].astype(str))
|
||||
df['总排号'] = df['总排号'].astype(str).str.strip()
|
||||
|
||||
df_update = df[df['总排号'].isin(existing_id_set)].copy()
|
||||
df_insert = df[~df['总排号'].isin(existing_id_set)].copy()
|
||||
|
||||
log_info(f"分析结果: 需插入 {len(df_insert)} 条, 需更新 {len(df_update)} 条")
|
||||
|
||||
# 1. 插入新数据
|
||||
if not df_insert.empty:
|
||||
log_info("正在执行批量插入...")
|
||||
df_insert.to_sql('executionCardData', self.engine, schema='warehouseOutbound',
|
||||
if_exists='append', index=False, chunksize=BATCH_SIZE,
|
||||
dtype=dtype_dict)
|
||||
log_info("批量插入完成。")
|
||||
|
||||
# 2. 更新现有数据 - 改用逐条或小批量 UPDATE
|
||||
if not df_update.empty:
|
||||
log_info("正在执行批量更新...")
|
||||
|
||||
cols = [c for c in df.columns if c != '总排号']
|
||||
set_clause = ", ".join([f"[{c}] = :{c}" for c in cols])
|
||||
update_sql = f"""
|
||||
UPDATE [warehouseOutbound].[executionCardData]
|
||||
SET {set_clause}
|
||||
WHERE [总排号] = :总排号
|
||||
"""
|
||||
|
||||
with self.engine.begin() as conn:
|
||||
batch_size = 1000
|
||||
total_rows = len(df_update)
|
||||
update_count = 0
|
||||
|
||||
for i in range(0, total_rows, batch_size):
|
||||
batch = df_update.iloc[i:i+batch_size]
|
||||
records = batch.to_dict('records')
|
||||
|
||||
result = conn.execute(text(update_sql), records)
|
||||
update_count += result.rowcount
|
||||
|
||||
if (i + batch_size) % 5000 == 0:
|
||||
log_info(f"已更新 {i + batch_size}/{total_rows} 条记录...")
|
||||
|
||||
# 修复 SQL Server executemany 返回负数 rowcount 的问题
|
||||
affected_rows = abs(update_count) if update_count < 0 else total_rows
|
||||
log_info(f"批量更新完成，共影响 {affected_rows} 行。")
|
||||
|
||||
def process_excel_files(self):
|
||||
for cfg in EXCEL_CONFIGS:
|
||||
remote_path = cfg['file_path']
|
||||
filename = os.path.basename(remote_path)
|
||||
local_path = os.path.join(self.cache_dir, filename)
|
||||
|
||||
should_sync, reason = self._should_process_file(remote_path, local_path)
|
||||
|
||||
if should_sync:
|
||||
log_info(f"开始处理文件: {filename} ({reason})")
|
||||
try:
|
||||
# 复制文件到本地缓存（只复制一次）
|
||||
if os.path.exists(remote_path):
|
||||
shutil.copy2(remote_path, local_path)
|
||||
|
||||
# 遍历该文件的所有指定 sheet
|
||||
for sheet_name in cfg['sheet_names']:
|
||||
log_info(f" → 处理工作表: {sheet_name} (合同年份: {cfg['contract_year']})")
|
||||
try:
|
||||
df = pd.read_excel(local_path, sheet_name=sheet_name, header=0, engine='openpyxl')
|
||||
df.columns = [str(c).strip() for c in df.columns]
|
||||
df.rename(columns=cfg['field_mapping'], inplace=True)
|
||||
|
||||
cleaned_df, dtype_mapping = self._clean_dataframe(df, cfg['contract_year'])
|
||||
|
||||
if cleaned_df is not None:
|
||||
self._sync_to_db(cleaned_df, dtype_mapping)
|
||||
log_info(f" 工作表 {sheet_name} 同步成功。")
|
||||
else:
|
||||
log_warning(f" 工作表 {sheet_name} 清洗失败，跳过。")
|
||||
except Exception as e:
|
||||
log_error(f" 处理工作表 {sheet_name} 时发生错误: {str(e)}", exc_info=True)
|
||||
|
||||
log_info(f"文件 {filename} 所有工作表处理完成。")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"处理文件 {filename} 时发生错误: {str(e)}", exc_info=True)
|
||||
else:
|
||||
log_info(f"跳过文件: {filename} ({reason})")
|
||||
|
||||
def generate_contract_data(self):
|
||||
log_info("开始生成/更新 contractData 表...")
|
||||
|
||||
merge_sql = """
|
||||
WITH SourceData AS (
|
||||
SELECT
|
||||
CAST(ISNULL([合同年份], '') AS NVARCHAR(10)) AS [合同年份],
|
||||
CAST(ISNULL([车间号], '') AS NVARCHAR(20)) AS [车间号],
|
||||
CAST(ISNULL([工令号], '') AS NVARCHAR(200)) AS [工令号],
|
||||
CAST([订单号] AS NVARCHAR(150)) AS [订单号],
|
||||
CAST([客户名称] AS NVARCHAR(200)) AS [客户名称],
|
||||
CAST([产品名称] AS NVARCHAR(200)) AS [产品型号],
|
||||
CAST([量程] AS NVARCHAR(150)) AS [量程],
|
||||
TRY_CAST([数量] AS INT) AS [数量],
|
||||
CAST(NULL AS INT) AS [单价],
|
||||
TRY_CAST([序号] AS INT) AS [ID],
|
||||
CAST([位号] AS NVARCHAR(500)) AS [位号],
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY [合同年份], [车间号], [工令号]
|
||||
ORDER BY [总排号] DESC
|
||||
) as rn
|
||||
FROM [warehouseOutbound].[executionCardData]
|
||||
WHERE
|
||||
[车间号] IS NOT NULL AND [车间号] <> ''
|
||||
AND [工令号] IS NOT NULL AND [工令号] <> ''
|
||||
)
|
||||
|
||||
MERGE INTO [warehouseOutbound].[contractData] AS Target
|
||||
USING (SELECT * FROM SourceData WHERE rn = 1) AS Source
|
||||
ON (
|
||||
Target.[合同年份] = Source.[合同年份]
|
||||
AND Target.[车间号] = Source.[车间号]
|
||||
AND Target.[工令号] = Source.[工令号]
|
||||
)
|
||||
|
||||
WHEN MATCHED THEN
|
||||
UPDATE SET
|
||||
Target.[订单号] = Source.[订单号],
|
||||
Target.[客户名称] = Source.[客户名称],
|
||||
Target.[产品型号] = Source.[产品型号],
|
||||
Target.[量程] = Source.[量程],
|
||||
Target.[数量] = Source.[数量],
|
||||
Target.[ID] = Source.[ID],
|
||||
Target.[位号] = Source.[位号]
|
||||
|
||||
WHEN NOT MATCHED BY TARGET THEN
|
||||
INSERT (
|
||||
[合同年份], [车间号], [工令号],
|
||||
[订单号], [客户名称], [产品型号],
|
||||
[量程], [数量], [单价], [ID], [位号]
|
||||
)
|
||||
VALUES (
|
||||
Source.[合同年份], Source.[车间号], Source.[工令号],
|
||||
Source.[订单号], Source.[客户名称], Source.[产品型号],
|
||||
Source.[量程], Source.[数量], Source.[单价], Source.[ID], Source.[位号]
|
||||
)
|
||||
;
|
||||
"""
|
||||
|
||||
try:
|
||||
with self.engine.begin() as conn:
|
||||
result = conn.execute(text(merge_sql))
|
||||
log_info(f"ContractData 表同步完成 (SQL Server 内部处理)。rowcount: {result.rowcount}")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"生成 ContractData 失败: {e}", exc_info=True)
|
||||
|
||||
def main():
|
||||
# 初始化日志管理器
|
||||
LoggerManager("etl_manager", log_prefix="sync")
|
||||
|
||||
parser = argparse.ArgumentParser(description="Excel数据同步至SQL Server")
|
||||
parser.add_argument('--force', action='store_true', help='强制同步所有文件')
|
||||
args = parser.parse_args()
|
||||
|
||||
syncer = DataSynchronizer(force_sync=args.force)
|
||||
|
||||
if args.force:
|
||||
log_start("Excel 同步任务 (强制模式)")
|
||||
else:
|
||||
log_start("Excel 同步任务 (增量模式)")
|
||||
|
||||
syncer.process_excel_files()
|
||||
syncer.generate_contract_data()
|
||||
|
||||
log_complete("Excel 同步任务已完成")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -3,6 +3,7 @@ from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, BATCH_SIZE
|
||||
import db_utils
|
||||
import time
|
||||
import os
|
||||
import ntfy_utils
|
||||
from log_utils import (LoggerManager, log_success, log_error, log_warning, log_info, log_processing,
|
||||
log_skip, log_start, log_file, log_database, log_sync, log_complete)
|
||||
|
||||
@@ -21,7 +22,7 @@ def format_duration(seconds):
|
||||
def has_identity_column(sql_cursor, schema, table):
|
||||
"""检查表是否包含标识列"""
|
||||
query = """
|
||||
SELECT COUNT(*)
|
||||
SELECT COUNT(*)
|
||||
FROM sys.columns c
|
||||
JOIN sys.tables t ON c.object_id = t.object_id
|
||||
JOIN sys.schemas s ON t.schema_id = s.schema_id
|
||||
@@ -30,18 +31,69 @@ def has_identity_column(sql_cursor, schema, table):
|
||||
sql_cursor.execute(query, (schema, table))
|
||||
return sql_cursor.fetchone()[0] > 0
|
||||
|
||||
def _send_summary_notification(sync_results, success_count, failed_count, total_rows, total_time):
|
||||
"""发送汇总 ntfy 通知"""
|
||||
if not sync_results:
|
||||
return
|
||||
|
||||
# 分类成功和失败的表
|
||||
success_tables = [r for r in sync_results if r['status'] == 'success']
|
||||
failed_tables = [r for r in sync_results if r['status'] == 'failed']
|
||||
|
||||
# 构建消息
|
||||
lines = []
|
||||
lines.append("🎯 全量同步完成\n")
|
||||
|
||||
# 成功的表
|
||||
if success_tables:
|
||||
lines.append(f"✅ 成功 {len(success_tables)} 张表:")
|
||||
for r in success_tables:
|
||||
lines.append(f" • {r['table']}: {r['rows']:,} 行, {format_duration(r['duration'])}")
|
||||
lines.append("")
|
||||
|
||||
# 失败的表
|
||||
if failed_tables:
|
||||
lines.append(f"❌ 失败 {failed_count} 张表:")
|
||||
for r in failed_tables:
|
||||
error_brief = r['error'][:50] + "..." if len(r['error']) > 50 else r['error']
|
||||
lines.append(f" • {r['table']}: {error_brief}")
|
||||
lines.append("")
|
||||
|
||||
# 总计
|
||||
lines.append(f"总计: {total_rows:,} 行 | 用时: {format_duration(total_time)}")
|
||||
|
||||
message = "\n".join(lines)
|
||||
|
||||
# 根据成功/失败情况决定优先级和标签
|
||||
if failed_count > 0:
|
||||
priority = "high"
|
||||
tags = ["warning", "database"]
|
||||
else:
|
||||
priority = "default"
|
||||
tags = ["white_check_mark", "database"]
|
||||
|
||||
ntfy_utils.send_ntfy(
|
||||
message=message,
|
||||
title="全量同步完成",
|
||||
priority=priority,
|
||||
tags=tags
|
||||
)
|
||||
|
||||
def run_full_sync():
|
||||
log_info("=" * 70)
|
||||
log_start( "全量初始化同步")
|
||||
log_info( f"配置文件数: {len(SYNC_MAPPING)} 个")
|
||||
log_info( f"批次大小: {BATCH_SIZE} 行")
|
||||
log_info("=" * 70)
|
||||
|
||||
|
||||
sync_start_time = time.time()
|
||||
total_tables = 0
|
||||
success_tables = 0
|
||||
failed_tables = 0
|
||||
total_rows = 0
|
||||
|
||||
# 收集每张表的同步结果，用于最终汇总通知
|
||||
sync_results = [] # 格式: {'table': str, 'status': 'success'|'failed', 'rows': int, 'duration': float, 'error': str}
|
||||
|
||||
try:
|
||||
sql_conn = db_utils.get_sql_conn()
|
||||
@@ -88,9 +140,18 @@ def run_full_sync():
|
||||
log_info( f" 检测到 {len(columns)} 个列")
|
||||
insert_sql = db_utils.generate_insert_sql(target_schema, target_table, columns)
|
||||
|
||||
# 3. TRUNCATE 目标表
|
||||
sql_cursor.execute(f"TRUNCATE TABLE {full_target_name}")
|
||||
log_info( f" 已清空目标表")
|
||||
# 3. 检查目标表是否存在，不存在则自动创建
|
||||
if not db_utils.table_exists(sql_cursor, target_schema, target_table):
|
||||
db_utils.ensure_schema(sql_cursor, target_schema)
|
||||
db_utils.create_table_from_access(
|
||||
sql_cursor, target_schema, target_table,
|
||||
acc_cursor.description, target_config['pk_col']
|
||||
)
|
||||
sql_conn.commit() # DDL 必须单独提交，否则 pyodbc 参数绑定无法获取正确的列元数据
|
||||
log_info( f" 目标表不存在，已自动创建")
|
||||
else:
|
||||
sql_cursor.execute(f"TRUNCATE TABLE {full_target_name}")
|
||||
log_info( f" 已清空目标表")
|
||||
|
||||
# 4. 检查并启用 IDENTITY_INSERT
|
||||
has_identity = has_identity_column(sql_cursor, target_schema, target_table)
|
||||
@@ -137,12 +198,21 @@ def run_full_sync():
|
||||
log_warning( f" 关闭 IDENTITY_INSERT 时警告: {id_err}")
|
||||
|
||||
sql_conn.commit()
|
||||
|
||||
|
||||
# 显示最终统计
|
||||
table_time = time.time() - table_start_time
|
||||
final_rate = table_rows / table_time if table_time > 0 else 0
|
||||
log_success( f"表 [{target_table}] 完成: {table_rows:,} 行 | 速率: {final_rate:,.0f} 行/秒 | 用时: {format_duration(table_time)}")
|
||||
|
||||
log_info( f"表 [{target_table}] 完成: {table_rows:,} 行 | 速率: {final_rate:,.0f} 行/秒 | 用时: {format_duration(table_time)}")
|
||||
|
||||
# 收集成功结果
|
||||
sync_results.append({
|
||||
'table': target_table,
|
||||
'status': 'success',
|
||||
'rows': table_rows,
|
||||
'duration': table_time,
|
||||
'error': None
|
||||
})
|
||||
|
||||
success_tables += 1
|
||||
file_success += 1
|
||||
total_rows += table_rows
|
||||
@@ -151,9 +221,20 @@ def run_full_sync():
|
||||
except Exception as tbl_err:
|
||||
failed_tables += 1
|
||||
file_failed += 1
|
||||
log_error( f"表 [{acc_table}] 同步失败: {tbl_err}", exc_info=True)
|
||||
error_msg = str(tbl_err)
|
||||
log_info( f"表 [{acc_table}] 同步失败: {error_msg}")
|
||||
|
||||
# 收集失败结果
|
||||
sync_results.append({
|
||||
'table': target_table,
|
||||
'status': 'failed',
|
||||
'rows': 0,
|
||||
'duration': time.time() - table_start_time,
|
||||
'error': error_msg
|
||||
})
|
||||
|
||||
sql_conn.rollback()
|
||||
|
||||
|
||||
# 确保清理 IDENTITY_INSERT 状态
|
||||
try:
|
||||
sql_cursor.execute(f"SET IDENTITY_INSERT {full_target_name} OFF")
|
||||
@@ -175,11 +256,11 @@ def run_full_sync():
|
||||
log_error( f"文件 [{os.path.basename(acc_path)}] 处理失败: {file_err}", exc_info=True)
|
||||
|
||||
sql_conn.close()
|
||||
|
||||
|
||||
# 总结统计
|
||||
sync_time = time.time() - sync_start_time
|
||||
log_info("\n" + "=" * 70)
|
||||
log_complete( "全量同步任务结束")
|
||||
log_info( "全量同步任务结束")
|
||||
log_info("-" * 70)
|
||||
log_info( f"总表数: {total_tables} 张 | 成功: {success_tables} 张 | 失败: {failed_tables} 张")
|
||||
log_info( f"总行数: {total_rows:,} 行")
|
||||
@@ -188,5 +269,8 @@ def run_full_sync():
|
||||
log_info( f"整体平均速率: {total_rows/sync_time:,.0f} 行/秒")
|
||||
log_info("=" * 70)
|
||||
|
||||
# 发送汇总 ntfy 通知
|
||||
_send_summary_notification(sync_results, success_tables, failed_tables, total_rows, sync_time)
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_full_sync()
|
||||
79
log_utils.py
79
log_utils.py
@@ -4,7 +4,10 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import shutil
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
import ntfy_utils
|
||||
|
||||
# ================= 全局 logger 实例 =================
|
||||
@@ -48,7 +51,8 @@ class LoggerManager:
|
||||
file_handler.setFormatter(file_formatter)
|
||||
_logger.addHandler(file_handler)
|
||||
|
||||
# 控制台处理器
|
||||
# 控制台处理器（强制 UTF-8 避免 GBK 编码错误）
|
||||
#sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_formatter = logging.Formatter(LOG_FORMAT, DATE_FORMAT)
|
||||
console_handler.setFormatter(console_formatter)
|
||||
@@ -56,6 +60,79 @@ class LoggerManager:
|
||||
|
||||
_logger.info(f"日志文件: {log_file}")
|
||||
|
||||
# 归档旧日志
|
||||
self.archive_old_logs(log_path, log_file)
|
||||
|
||||
def archive_old_logs(self, log_dir: str, current_log_file: str):
|
||||
"""将 log 目录下的旧日志移动到 Archive/YYYY-MM/ 子目录
|
||||
|
||||
Args:
|
||||
log_dir: 日志目录路径
|
||||
current_log_file: 当前正在使用的日志文件路径（不会被移动）
|
||||
"""
|
||||
archive_base = os.path.join(log_dir, "Archive")
|
||||
os.makedirs(archive_base, exist_ok=True)
|
||||
|
||||
# 遍历 log 根目录下的所有 .log 文件
|
||||
for filename in os.listdir(log_dir):
|
||||
if not filename.endswith('.log'):
|
||||
continue
|
||||
|
||||
file_path = os.path.join(log_dir, filename)
|
||||
|
||||
# 跳过当前正在使用的日志文件
|
||||
if file_path == current_log_file:
|
||||
continue
|
||||
|
||||
# 跳过 Archive 目录本身
|
||||
if os.path.isdir(file_path):
|
||||
continue
|
||||
|
||||
# 从文件名提取日期信息（格式：prefix_YYYYMMDD_HHMMSS.log）
|
||||
year_month = self._extract_year_month_from_filename(filename)
|
||||
|
||||
# 如果无法从文件名提取日期，使用文件修改时间
|
||||
if not year_month:
|
||||
try:
|
||||
stat = os.stat(file_path)
|
||||
mtime = datetime.fromtimestamp(stat.st_mtime)
|
||||
year_month = mtime.strftime("%Y-%m")
|
||||
except:
|
||||
year_month = "unknown"
|
||||
|
||||
# 创建年月子目录
|
||||
month_dir = os.path.join(archive_base, year_month)
|
||||
os.makedirs(month_dir, exist_ok=True)
|
||||
|
||||
# 移动文件到对应的年月目录
|
||||
dest_path = os.path.join(month_dir, filename)
|
||||
try:
|
||||
shutil.move(file_path, dest_path)
|
||||
_logger.info(f"已归档: {filename} -> Archive/{year_month}/")
|
||||
except Exception as e:
|
||||
_logger.warning(f"归档失败 {filename}: {e}")
|
||||
|
||||
def _extract_year_month_from_filename(self, filename: str) -> Optional[str]:
|
||||
"""从日志文件名中提取年月信息
|
||||
|
||||
支持的格式：
|
||||
- prefix_YYYYMMDD_HHMMSS.log
|
||||
- incremental_YYYYMMDD_HHMMSS.log
|
||||
- full_sync_YYYYMMDD_HHMMSS.log
|
||||
- excel_sync_YYYYMMDD_HHMMSS.log
|
||||
|
||||
Returns:
|
||||
年月字符串 (格式: YYYY-MM) 或 None
|
||||
"""
|
||||
# 匹配 YYYYMMDD 模式
|
||||
match = re.search(r'(\d{4})(\d{2})\d{2}_\d{6}', filename)
|
||||
if match:
|
||||
year = match.group(1)
|
||||
month = match.group(2)
|
||||
return f"{year}-{month}"
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def get_logger():
|
||||
"""获取全局 logger 实例"""
|
||||
|
||||
171
migration.py
171
migration.py
@@ -1,171 +0,0 @@
|
||||
import pandas as pd
|
||||
import os
|
||||
import shutil
|
||||
import urllib
|
||||
from sqlalchemy import create_engine, text
|
||||
from config import DB_CONFIG, MIGRATION_TASKS, TEMP_DIR
|
||||
from log_utils import LoggerManager, log_start, log_skip, log_processing, log_success, log_warning, log_error, log_complete
|
||||
|
||||
# 初始化日志管理器
|
||||
LoggerManager("migration", log_prefix="migration")
|
||||
|
||||
# ==========================================
|
||||
# 1. 脚本配置 (Configuration)
|
||||
# ==========================================
|
||||
|
||||
# 目标表配置
|
||||
TARGET_DB_SCHEMA = "warehouseOutbound"
|
||||
TARGET_TABLE_NAME = "customerProductType"
|
||||
SQL_SOURCE_FILE_COL = "SourceFile" # 你在SQL中新增的字段名
|
||||
|
||||
# 字段映射常量
|
||||
SQL_COL_YEAR = "合同年份"
|
||||
SQL_COL_WORKSHOP = "车间号"
|
||||
SQL_COL_ORDER = "工令号"
|
||||
SQL_COL_MODEL = "客户型号"
|
||||
|
||||
# 运行参数
|
||||
FORCE_UPDATE = False # 如果设为 True，则无视时间对比，强制更新所有文件
|
||||
|
||||
# ==========================================
|
||||
# 2. 核心辅助函数
|
||||
# ==========================================
|
||||
|
||||
def get_db_engine():
|
||||
params = urllib.parse.quote_plus(
|
||||
f"DRIVER={{{DB_CONFIG['driver']}}};"
|
||||
f"SERVER={DB_CONFIG['server']};"
|
||||
f"DATABASE={DB_CONFIG['database']};"
|
||||
f"UID={DB_CONFIG['username']};"
|
||||
f"PWD={DB_CONFIG['password']};"
|
||||
f"TrustServerCertificate=yes;"
|
||||
)
|
||||
# fast_executemany 极大提高写入速度
|
||||
return create_engine(f"mssql+pyodbc:///?odbc_connect={params}", fast_executemany=True)
|
||||
|
||||
def get_file_mtime(path):
|
||||
"""获取文件最后修改时间戳"""
|
||||
try:
|
||||
return os.path.getmtime(path)
|
||||
except OSError:
|
||||
return 0
|
||||
|
||||
def delete_old_data(engine, filename):
|
||||
"""根据 SourceFile 字段精确删除旧数据"""
|
||||
full_table = f"[{TARGET_DB_SCHEMA}].[{TARGET_TABLE_NAME}]"
|
||||
sql = text(f"DELETE FROM {full_table} WHERE [{SQL_SOURCE_FILE_COL}] = :fname")
|
||||
with engine.begin() as conn:
|
||||
conn.execute(sql, {"fname": filename})
|
||||
|
||||
# ==========================================
|
||||
# 3. 迁移主逻辑
|
||||
# ==========================================
|
||||
|
||||
def run_migration():
|
||||
# 初始化环境
|
||||
if not os.path.exists(TEMP_DIR):
|
||||
os.makedirs(TEMP_DIR)
|
||||
|
||||
engine = get_db_engine()
|
||||
sync_count = 0
|
||||
error_count = 0
|
||||
|
||||
log_start(f"增量同步任务 (强制更新={FORCE_UPDATE})")
|
||||
|
||||
for task in MIGRATION_TASKS:
|
||||
remote_path = task['file_path']
|
||||
filename = os.path.basename(remote_path)
|
||||
local_path = os.path.join(TEMP_DIR, filename)
|
||||
|
||||
# 1. 检查源文件
|
||||
if not os.path.exists(remote_path):
|
||||
msg = f"远程文件未找到: {remote_path}"
|
||||
log_error(msg)
|
||||
continue
|
||||
|
||||
# 2. 增量判定
|
||||
remote_mtime = get_file_mtime(remote_path)
|
||||
local_mtime = get_file_mtime(local_path)
|
||||
|
||||
if not FORCE_UPDATE and os.path.exists(local_path) and remote_mtime <= local_mtime:
|
||||
log_skip(f"{filename} (文件未变更)")
|
||||
continue
|
||||
|
||||
log_processing(f"正在处理: {filename} ...")
|
||||
|
||||
try:
|
||||
# 3. 复制文件到本地 temp
|
||||
shutil.copy2(remote_path, local_path)
|
||||
|
||||
# 4. 读取 Excel
|
||||
xls_dict = pd.read_excel(local_path, sheet_name=task['sheet_names'])
|
||||
if not isinstance(xls_dict, dict):
|
||||
xls_dict = {task['sheet_names'][0]: xls_dict}
|
||||
|
||||
# 准备存放该文件所有 Sheet 的合并数据
|
||||
df_all_sheets = []
|
||||
|
||||
for sheet_name, df in xls_dict.items():
|
||||
if df.empty: continue
|
||||
|
||||
# 清洗与过滤
|
||||
df.columns = df.columns.astype(str).str.strip()
|
||||
source_cols = list(task['mapping'].keys())
|
||||
|
||||
missing = [c for c in source_cols if c not in df.columns]
|
||||
if missing:
|
||||
log_warning(f"Sheet[{sheet_name}] 缺失列: {missing}")
|
||||
continue
|
||||
|
||||
# 提取并重命名
|
||||
df_subset = df[source_cols].copy()
|
||||
df_subset.rename(columns=task['mapping'], inplace=True)
|
||||
|
||||
# 注入年份和来源文件名
|
||||
df_subset[SQL_COL_YEAR] = task['year']
|
||||
df_subset[SQL_SOURCE_FILE_COL] = filename # 存入文件名，用于下次精准删除
|
||||
|
||||
# 数据清洗
|
||||
subset_keys = [SQL_COL_YEAR, SQL_COL_WORKSHOP, SQL_COL_ORDER]
|
||||
df_subset.dropna(subset=subset_keys, inplace=True)
|
||||
df_subset.drop_duplicates(subset=subset_keys, keep='first', inplace=True)
|
||||
|
||||
if not df_subset.empty:
|
||||
df_all_sheets.append(df_subset)
|
||||
|
||||
# 5. 写入数据库
|
||||
if df_all_sheets:
|
||||
final_df = pd.concat(df_all_sheets, ignore_index=True)
|
||||
|
||||
# 执行删除并插入 (事务)
|
||||
with engine.begin() as conn:
|
||||
# A. 删除旧记录
|
||||
delete_sql = text(f"DELETE FROM [{TARGET_DB_SCHEMA}].[{TARGET_TABLE_NAME}] WHERE [{SQL_SOURCE_FILE_COL}] = :fname")
|
||||
conn.execute(delete_sql, {"fname": filename})
|
||||
|
||||
# B. 插入新记录
|
||||
final_df.to_sql(
|
||||
name=TARGET_TABLE_NAME,
|
||||
schema=TARGET_DB_SCHEMA,
|
||||
con=conn,
|
||||
if_exists='append',
|
||||
index=False,
|
||||
chunksize=1000
|
||||
)
|
||||
|
||||
log_success(f"成功同步: {len(final_df)} 行记录")
|
||||
sync_count += 1
|
||||
else:
|
||||
log_warning("文件内容为空或格式不符")
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"文件 [{filename}] 处理失败: {str(e)}"
|
||||
log_error(error_msg)
|
||||
error_count += 1
|
||||
|
||||
# 结束汇总
|
||||
summary = f"同步完成: 成功 {sync_count} 个文件, 失败 {error_count} 个文件。"
|
||||
log_complete(f"同步完成: 成功 {sync_count} 个文件, 失败 {error_count} 个文件")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_migration()
|
||||
@@ -1,4 +1,3 @@
|
||||
pyodbc>=5.0.0
|
||||
pandas>=1.5.0
|
||||
sqlalchemy>=2.0.0
|
||||
openpyxl>=3.0.0
|
||||
python-dotenv>=1.0.0
|
||||
requests>=2.28.0
|
||||
@@ -2,58 +2,163 @@ import time
|
||||
import os
|
||||
import sys
|
||||
import pyodbc
|
||||
from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, POLL_INTERVAL, LOG_TABLE_CONFIG
|
||||
from config import SQL_SERVER_CONN, ACCESS_DRIVER, SYNC_MAPPING, POLL_INTERVAL, LOG_TABLE_CONFIG, UPTIME_KUMA_CONFIG
|
||||
import db_utils
|
||||
from log_utils import (LoggerManager, log_success, log_error, log_warning, log_info, log_processing,
|
||||
log_skip, log_critical, log_start, log_stop, log_file, log_database, log_sync)
|
||||
from log_utils import (LoggerManager, log_error, log_warning, log_info, log_processing,
|
||||
log_skip, log_critical, log_stop, log_file, log_database, log_sync)
|
||||
from uptime_kuma_utils import UptimeKumaMonitor
|
||||
|
||||
# 初始化日志管理器
|
||||
LoggerManager("run_incremental_sync", log_prefix="incremental")
|
||||
|
||||
# 初始化 Uptime Kuma 监控器
|
||||
uptime_monitor = UptimeKumaMonitor(UPTIME_KUMA_CONFIG)
|
||||
uptime_monitor.set_logger(log_warning)
|
||||
|
||||
# 提交后是否再做一次独立复核 (额外保险, 略增开销; 若不需要可设为 False)
|
||||
ENABLE_POST_COMMIT_VERIFY = True
|
||||
|
||||
# IN 子句单批最大参数数 (SQL Server 上限约 2100, 留足余量)
|
||||
IN_CLAUSE_BATCH = 900
|
||||
|
||||
|
||||
class SyncVerificationError(Exception):
|
||||
"""数据落库校验未通过时抛出, 触发事务回滚并保留 Synced=0 以便下一轮重试"""
|
||||
pass
|
||||
|
||||
|
||||
# ================= 辅助函数 =================
|
||||
|
||||
def has_identity_column(sql_cursor, schema, table):
|
||||
"""检查表是否包含标识列"""
|
||||
query = """
|
||||
SELECT COUNT(*)
|
||||
FROM sys.columns c
|
||||
JOIN sys.tables t ON c.object_id = t.object_id
|
||||
JOIN sys.schemas s ON t.schema_id = s.schema_id
|
||||
WHERE s.name = ? AND t.name = ? AND c.is_identity = 1
|
||||
"""
|
||||
sql_cursor.execute(query, (schema, table))
|
||||
return sql_cursor.fetchone()[0] > 0
|
||||
|
||||
|
||||
def _chunked(values, size=IN_CLAUSE_BATCH):
|
||||
"""将集合/列表分批, 避免 IN 子句参数数量超过 SQL Server 上限"""
|
||||
items = list(values)
|
||||
for i in range(0, len(items), size):
|
||||
yield items[i:i + size]
|
||||
|
||||
|
||||
def _norm_key(v):
|
||||
"""主键归一化, 用于跨来源(日志侧 / Access 侧 / SQL 侧)的集合比较, 规避类型差异"""
|
||||
return str(v).strip() if v is not None else None
|
||||
|
||||
|
||||
def fetch_existing_pks(sql_cursor, full_name, pk_col, pk_values):
|
||||
"""
|
||||
拿一批主键去目标表查询, 返回其中【实际存在】的主键集合 (分批查询)。
|
||||
这是逐主键校验的基础: 用 "查到了哪些" 来精确判断存在 / 不存在,
|
||||
而不是用数量相减 (数量法会被 "一边少插一边漏删" 互相抵消而误判)。
|
||||
"""
|
||||
found = set()
|
||||
for chunk in _chunked(pk_values):
|
||||
if not chunk:
|
||||
continue
|
||||
placeholders = ','.join(['?'] * len(chunk))
|
||||
sql = f"SELECT [{pk_col}] FROM {full_name} WHERE [{pk_col}] IN ({placeholders})"
|
||||
sql_cursor.execute(sql, list(chunk))
|
||||
for row in sql_cursor.fetchall():
|
||||
found.add(row[0])
|
||||
return found
|
||||
|
||||
|
||||
def verify_sync_result(sql_cursor, full_name, pk_col, record_ids, inserted_pk_set):
|
||||
"""
|
||||
逐主键校验本批同步结果 (不依赖数量比较):
|
||||
|
||||
- 删除校验: 源端已读不到的记录 (record_ids 中不在 inserted_pk_set 的部分),
|
||||
删除后必须【全部不存在】于目标表; 只要还查得到任意一条即判失败。
|
||||
- 插入校验: 从 Access 实际读到并重新插入的主键 (inserted_pk_set),
|
||||
必须【全部存在】于目标表; 只要有一条查不到即判失败。
|
||||
|
||||
校验不通过抛出 SyncVerificationError, 并给出具体出问题的主键样例。
|
||||
返回值: 本批 "随源删除" 的记录条数 (供日志展示)。
|
||||
"""
|
||||
inserted_keys = {_norm_key(pk) for pk in inserted_pk_set}
|
||||
|
||||
# 1) 删除校验: 应删除的记录, 删除后必须不在目标表
|
||||
to_delete = [rid for rid in record_ids if _norm_key(rid) not in inserted_keys]
|
||||
if to_delete:
|
||||
leftover = fetch_existing_pks(sql_cursor, full_name, pk_col, to_delete)
|
||||
if leftover:
|
||||
sample = list(leftover)[:5]
|
||||
raise SyncVerificationError(
|
||||
f"删除校验失败: {len(leftover)} 条记录删除后仍存在于目标表, 例如 {sample}")
|
||||
|
||||
# 2) 插入校验: 应插入的记录, 插入后必须存在于目标表
|
||||
if inserted_pk_set:
|
||||
present_keys = {_norm_key(p) for p in fetch_existing_pks(sql_cursor, full_name, pk_col, inserted_pk_set)}
|
||||
missing = [pk for pk in inserted_pk_set if _norm_key(pk) not in present_keys]
|
||||
if missing:
|
||||
raise SyncVerificationError(
|
||||
f"插入校验失败: {len(missing)} 条记录插入后在目标表查不到, 例如 {missing[:5]}")
|
||||
|
||||
return len(to_delete)
|
||||
|
||||
|
||||
def find_pk_index(acc_cols, pk_col):
|
||||
"""在 Access 结果列中定位主键列下标 (大小写不敏感兜底)"""
|
||||
if pk_col in acc_cols:
|
||||
return acc_cols.index(pk_col)
|
||||
lower_map = {c.lower(): i for i, c in enumerate(acc_cols)}
|
||||
return lower_map.get(pk_col.lower())
|
||||
|
||||
|
||||
# ================= 主逻辑 =================
|
||||
|
||||
def process_sync_task():
|
||||
"""
|
||||
逻辑重构：
|
||||
逻辑重构:
|
||||
遍历 SYNC_MAPPING 中的每一个文件 -> 去日志表查询该文件下特定表的未同步记录。
|
||||
|
||||
校验机制 (逐主键确认):
|
||||
删旧插新后, 在提交前对本批数据做删除校验 + 插入校验 ——
|
||||
应删除的主键确认已不在目标表、应插入的主键确认已在目标表, 二者都通过才标记
|
||||
Synced=1 并提交; 否则回滚整批, 保持 Synced=0, 下一轮自动重试。
|
||||
"""
|
||||
sql_conn = db_utils.get_sql_conn()
|
||||
sql_cursor = sql_conn.cursor()
|
||||
sql_cursor.fast_executemany = True # 高性能开关
|
||||
|
||||
sql_cursor.fast_executemany = True # 高性能开关
|
||||
|
||||
# 标记是否有工作被处理（用于控制轮询休眠时间）
|
||||
work_done = False
|
||||
|
||||
|
||||
cols = LOG_TABLE_CONFIG
|
||||
log_full_name = db_utils.fmt_table(cols['schema'], cols['table_name'])
|
||||
|
||||
try:
|
||||
# === 核心循环：以配置文件为驱动 ===
|
||||
for clean_path, tables_map in SYNC_MAPPING.items():
|
||||
|
||||
|
||||
# 1. 准备查询条件
|
||||
# 获取该文件下所有需要同步的表名列表
|
||||
target_tables = list(tables_map.keys())
|
||||
if not target_tables:
|
||||
continue
|
||||
|
||||
# 构造 TableAddress 的精确匹配条件
|
||||
# VBA 逻辑：网络路径带 ";DATABASE=", 本地路径带 "LOCAL:"
|
||||
# 我们直接构造这两个字符串，让 SQL Server 做精确匹配，效率极高
|
||||
# 注意：将路径转为 Windows 标准反斜杠
|
||||
win_path = os.path.normpath(clean_path)
|
||||
win_path = os.path.normpath(clean_path)
|
||||
addr_candidates = [
|
||||
f";DATABASE={win_path}", # 情况1
|
||||
f"LOCAL={win_path}", # 情况2 (注意 VBA 代码里可能是 LOCAL: 或 LOCAL=，请核对)
|
||||
f"LOCAL:{win_path}", # 情况3
|
||||
win_path # 情况4 (兼容没有前缀的情况)
|
||||
f";DATABASE={win_path}", # 情况1
|
||||
f"LOCAL={win_path}", # 情况2
|
||||
f"LOCAL:{win_path}", # 情况3
|
||||
win_path # 情况4 (兼容没有前缀的情况)
|
||||
]
|
||||
|
||||
|
||||
# 2. 构造动态 SQL 查询
|
||||
# WHERE Synced=0 AND Address IN (...) AND TableName IN (...)
|
||||
placeholders_addr = ','.join(['?'] * len(addr_candidates))
|
||||
placeholders_tbl = ','.join(['?'] * len(target_tables))
|
||||
|
||||
|
||||
query_log = f"""
|
||||
SELECT TOP 1000
|
||||
{cols['col_log_id']},
|
||||
@@ -66,21 +171,21 @@ def process_sync_task():
|
||||
AND {cols['col_table_name']} IN ({placeholders_tbl})
|
||||
ORDER BY {cols['col_log_id']} ASC
|
||||
"""
|
||||
|
||||
|
||||
# 参数列表：先放地址，再放表名
|
||||
params = addr_candidates + target_tables
|
||||
|
||||
|
||||
sql_cursor.execute(query_log, params)
|
||||
logs = sql_cursor.fetchall()
|
||||
|
||||
if not logs:
|
||||
continue # 这个文件没有需要同步的记录，检查下一个文件
|
||||
|
||||
work_done = True # 标记有工作
|
||||
if not logs:
|
||||
continue # 这个文件没有需要同步的记录，检查下一个文件
|
||||
|
||||
work_done = True # 标记有工作
|
||||
log_file(f"{os.path.basename(clean_path)} 发现 {len(logs)} 条待同步变更")
|
||||
|
||||
# 3. 本地分组 (按表名)
|
||||
# 结构: table_tasks[TableName] = { ids: {}, log_ids: [] }
|
||||
# 结构: table_tasks[TableName] = { record_ids: set(), log_ids: [] }
|
||||
table_tasks = {}
|
||||
for row in logs:
|
||||
log_id, acc_table, record_id = row
|
||||
@@ -94,6 +199,7 @@ def process_sync_task():
|
||||
log_error(f"无法访问文件: {clean_path}")
|
||||
continue
|
||||
|
||||
acc_conn = None
|
||||
try:
|
||||
acc_conn = db_utils.get_access_conn(clean_path)
|
||||
acc_cursor = acc_conn.cursor()
|
||||
@@ -107,65 +213,137 @@ def process_sync_task():
|
||||
file_error_count = 0
|
||||
file_total_records = 0
|
||||
|
||||
for acc_table, data in table_tasks.items():
|
||||
record_ids = list(data['record_ids'])
|
||||
log_ids = data['log_ids']
|
||||
|
||||
# 读取目标配置
|
||||
target_conf = tables_map[acc_table]
|
||||
target_schema = target_conf['target_schema']
|
||||
target_table = target_conf['target_table']
|
||||
pk_col = target_conf['pk_col']
|
||||
target_full_name = db_utils.fmt_table(target_schema, target_table)
|
||||
try:
|
||||
for acc_table, data in table_tasks.items():
|
||||
record_ids = list(data['record_ids'])
|
||||
log_ids = data['log_ids']
|
||||
|
||||
log_processing(f"正在同步表 [{acc_table}] → [{target_table}] ({len(record_ids)} 条记录)")
|
||||
# 读取目标配置
|
||||
target_conf = tables_map[acc_table]
|
||||
target_schema = target_conf['target_schema']
|
||||
target_table = target_conf['target_table']
|
||||
pk_col = target_conf['pk_col']
|
||||
target_full_name = db_utils.fmt_table(target_schema, target_table)
|
||||
|
||||
# 检查目标表是否存在，不存在则自动创建
|
||||
if not db_utils.table_exists(sql_cursor, target_schema, target_table):
|
||||
db_utils.ensure_schema(sql_cursor, target_schema)
|
||||
acc_cursor.execute(f"SELECT TOP 1 * FROM [{acc_table}]")
|
||||
db_utils.create_table_from_access(
|
||||
sql_cursor, target_schema, target_table,
|
||||
acc_cursor.description, pk_col
|
||||
)
|
||||
log_info(f"目标表 [{target_table}] 不存在，已自动创建")
|
||||
|
||||
log_processing(f"正在同步表 [{acc_table}] → [{target_table}] ({len(record_ids)} 条记录)")
|
||||
|
||||
identity_enabled = False
|
||||
try:
|
||||
# --- A. 从 Access 读取最新数据 ---
|
||||
ids_placeholders = ','.join(['?'] * len(record_ids))
|
||||
acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})"
|
||||
acc_cursor.execute(acc_sql, record_ids)
|
||||
new_rows = acc_cursor.fetchall()
|
||||
acc_cols = [col[0] for col in acc_cursor.description]
|
||||
|
||||
# 定位主键列, 取出 Access 实际读到的主键集合 (校验基准)
|
||||
pk_index = find_pk_index(acc_cols, pk_col)
|
||||
if new_rows and pk_index is None:
|
||||
raise SyncVerificationError(
|
||||
f"在 Access 表 [{acc_table}] 中找不到主键列 [{pk_col}]")
|
||||
inserted_pk_set = (set(row[pk_index] for row in new_rows)
|
||||
if pk_index is not None else set())
|
||||
|
||||
# --- B. SQL Server 删除旧记录 ---
|
||||
del_sql = f"DELETE FROM {target_full_name} WHERE [{pk_col}] IN ({ids_placeholders})"
|
||||
sql_cursor.execute(del_sql, record_ids)
|
||||
|
||||
# --- C. 插入新记录 ---
|
||||
if new_rows:
|
||||
insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols)
|
||||
|
||||
# 检查并启用 IDENTITY_INSERT
|
||||
if has_identity_column(sql_cursor, target_schema, target_table):
|
||||
try:
|
||||
sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON")
|
||||
identity_enabled = True
|
||||
except Exception as id_err:
|
||||
log_error(f"无法启用 IDENTITY_INSERT: {id_err}")
|
||||
raise
|
||||
|
||||
sql_cursor.executemany(insert_sql, new_rows)
|
||||
|
||||
# 立即关闭 IDENTITY_INSERT (会话级设置, 不随事务回滚)
|
||||
if identity_enabled:
|
||||
sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF")
|
||||
identity_enabled = False
|
||||
|
||||
# --- D. 提交前校验 (逐主键确认: 应删的已不在 / 应插的已在) ---
|
||||
removed_count = verify_sync_result(
|
||||
sql_cursor, target_full_name, pk_col, record_ids, inserted_pk_set)
|
||||
|
||||
# --- E. 校验通过 -> 标记日志 Synced = 1 ---
|
||||
log_placeholders = ','.join(['?'] * len(log_ids))
|
||||
update_log_sql = f"""
|
||||
UPDATE {log_full_name}
|
||||
SET {cols['col_synced']} = 1
|
||||
WHERE {cols['col_log_id']} IN ({log_placeholders})
|
||||
"""
|
||||
sql_cursor.execute(update_log_sql, log_ids)
|
||||
|
||||
# --- F. 提交事务 (删/插/标记 原子生效) ---
|
||||
sql_conn.commit()
|
||||
|
||||
# --- G. 提交后独立复核 (可选, 防 "提交成功但未持久化" 等极端情况) ---
|
||||
if ENABLE_POST_COMMIT_VERIFY:
|
||||
try:
|
||||
verify_sync_result(
|
||||
sql_cursor, target_full_name, pk_col, record_ids, inserted_pk_set)
|
||||
except SyncVerificationError as post_err:
|
||||
log_critical(
|
||||
f"严重: 表 [{target_table}] 提交后复核失败! {post_err}; "
|
||||
f"撤销同步标记以便重试")
|
||||
revert_sql = f"""
|
||||
UPDATE {log_full_name}
|
||||
SET {cols['col_synced']} = 0
|
||||
WHERE {cols['col_log_id']} IN ({log_placeholders})
|
||||
"""
|
||||
sql_cursor.execute(revert_sql, log_ids)
|
||||
sql_conn.commit()
|
||||
file_error_count += 1
|
||||
continue
|
||||
|
||||
# 同步成功 (仅本地日志记录, 不推送 ntfy; 服务存活状态由 Uptime Kuma 心跳负责)
|
||||
msg = f"表 [{target_table}] 同步并校验通过: {len(inserted_pk_set)} 条入库"
|
||||
if removed_count > 0:
|
||||
msg += f", {removed_count} 条随源删除"
|
||||
log_info(msg)
|
||||
|
||||
file_success_count += 1
|
||||
file_total_records += len(record_ids)
|
||||
|
||||
except Exception as tbl_err:
|
||||
# 清理可能残留的 IDENTITY_INSERT 会话状态
|
||||
if identity_enabled:
|
||||
try:
|
||||
sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF")
|
||||
except:
|
||||
pass
|
||||
# 回滚本表的 删/插/标记, 保持 Synced=0, 下一轮自动重试
|
||||
try:
|
||||
sql_conn.rollback()
|
||||
except:
|
||||
pass
|
||||
log_error(f"表 [{acc_table}] 同步失败 (已回滚, 将重试): {tbl_err}")
|
||||
file_error_count += 1
|
||||
finally:
|
||||
# 确保 Access 连接关闭
|
||||
try:
|
||||
# --- A. Access 查新数据 ---
|
||||
ids_placeholders = ','.join(['?'] * len(record_ids))
|
||||
acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})"
|
||||
acc_cursor.execute(acc_sql, record_ids)
|
||||
new_rows = acc_cursor.fetchall()
|
||||
acc_cols = [col[0] for col in acc_cursor.description]
|
||||
|
||||
# --- B. SQL Server 删旧插新 (事务) ---
|
||||
# 1. 删除
|
||||
del_sql = f"DELETE FROM {target_full_name} WHERE [{pk_col}] IN ({ids_placeholders})"
|
||||
sql_cursor.execute(del_sql, record_ids)
|
||||
|
||||
# 2. 插入
|
||||
if new_rows:
|
||||
insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols)
|
||||
try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON")
|
||||
except: pass
|
||||
|
||||
sql_cursor.executemany(insert_sql, new_rows)
|
||||
|
||||
try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF")
|
||||
except: pass
|
||||
|
||||
# 3. 标记日志 Synced = 1
|
||||
log_placeholders = ','.join(['?'] * len(log_ids))
|
||||
update_log_sql = f"""
|
||||
UPDATE {log_full_name}
|
||||
SET {cols['col_synced']} = 1
|
||||
WHERE {cols['col_log_id']} IN ({log_placeholders})
|
||||
"""
|
||||
sql_cursor.execute(update_log_sql, log_ids)
|
||||
|
||||
sql_conn.commit()
|
||||
log_success(f"表 [{target_table}] 同步完成: {len(record_ids)} 条记录")
|
||||
|
||||
file_success_count += 1
|
||||
file_total_records += len(record_ids)
|
||||
|
||||
except Exception as tbl_err:
|
||||
log_error(f"表 [{acc_table}] 同步失败: {tbl_err}")
|
||||
sql_conn.rollback()
|
||||
file_error_count += 1
|
||||
|
||||
acc_conn.close() # 关闭 Access 连接
|
||||
|
||||
if acc_conn:
|
||||
acc_conn.close()
|
||||
log_info(f"已关闭 Access 连接: {os.path.basename(clean_path)}")
|
||||
except Exception as close_err:
|
||||
log_warning(f"关闭 Access 连接时出错: {close_err}")
|
||||
# 输出文件级别的汇总
|
||||
if file_success_count > 0 or file_error_count > 0:
|
||||
summary = f"文件 [{os.path.basename(clean_path)}] 同步汇总: "
|
||||
@@ -180,25 +358,46 @@ def process_sync_task():
|
||||
log_critical(f"全局异常: {e}")
|
||||
return False
|
||||
finally:
|
||||
try: sql_conn.close()
|
||||
except: pass
|
||||
try:
|
||||
sql_conn.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
# ================= Uptime Kuma 心跳 =================
|
||||
# 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现
|
||||
|
||||
if __name__ == "__main__":
|
||||
log_start("增量同步服务已启动 (配置驱动模式)")
|
||||
# 仅本地日志记录, 不推送 ntfy 启动消息 (服务存活状态由 Uptime Kuma 心跳负责, 避免重复通知)
|
||||
log_info("增量同步服务已启动 (配置驱动模式 + 逐主键落库校验)")
|
||||
log_info(f"轮询间隔: {POLL_INTERVAL} 秒")
|
||||
log_info(f"监控配置: {len(SYNC_MAPPING)} 个文件")
|
||||
log_info(f"提交后复核: {'开启' if ENABLE_POST_COMMIT_VERIFY else '关闭'}")
|
||||
if UPTIME_KUMA_CONFIG.get('enabled', False):
|
||||
log_info(f"心跳间隔: {UPTIME_KUMA_CONFIG['heartbeat_interval']} 秒")
|
||||
log_info("=" * 70)
|
||||
|
||||
while True:
|
||||
try:
|
||||
has_work = process_sync_task()
|
||||
# 如果有工作，说明可能还有积压，休息短一点(0.1s)
|
||||
# 如果没工作，休息标准间隔(5s)
|
||||
time.sleep(0.1 if has_work else POLL_INTERVAL)
|
||||
except KeyboardInterrupt:
|
||||
log_info("=" * 70)
|
||||
log_stop("收到停止信号，服务正在关闭...")
|
||||
break
|
||||
except Exception as e:
|
||||
log_critical(f"主循环崩溃: {e}")
|
||||
time.sleep(5)
|
||||
|
||||
# 启动后台心跳线程: 立即发送首跳, 之后按间隔周期发送 (与同步主循环解耦)
|
||||
uptime_monitor.start()
|
||||
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
has_work = process_sync_task()
|
||||
|
||||
# 心跳由后台线程独立周期发送, 此处无需再调用
|
||||
|
||||
# 如果有工作，说明可能还有积压，休息短一点(0.1s)
|
||||
# 如果没工作，休息标准间隔
|
||||
time.sleep(0.1 if has_work else POLL_INTERVAL)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
log_info("=" * 70)
|
||||
log_stop("收到停止信号，服务正在关闭...")
|
||||
break
|
||||
except Exception as e:
|
||||
log_critical(f"主循环崩溃: {e}")
|
||||
time.sleep(5)
|
||||
finally:
|
||||
# 停止后台心跳线程, 并向 Uptime Kuma 发送 down 信号
|
||||
uptime_monitor.stop()
|
||||
168
uptime_kuma_utils.py
Normal file
168
uptime_kuma_utils.py
Normal file
@@ -0,0 +1,168 @@
|
||||
"""
|
||||
Uptime Kuma 心跳监控工具
|
||||
|
||||
用于向 Uptime Kuma 服务发送心跳信号，监控服务运行状态。
|
||||
|
||||
设计要点:
|
||||
- 推荐通过 start() 启动后台心跳线程, 与业务主循环解耦 ——
|
||||
心跳的网络耗时 / 重试不会阻塞业务逻辑。
|
||||
- 单次心跳自带重试 (MAX_RETRIES), 容忍 Uptime Kuma 服务偶发的慢响应 / 超时 / 4xx。
|
||||
- send_heartbeat() / check_and_send_heartbeat() / send_stop_signal() 保留为
|
||||
向后兼容的同步接口。
|
||||
"""
|
||||
|
||||
import time
|
||||
import threading
|
||||
import requests
|
||||
|
||||
|
||||
# 单次请求超时(秒) —— 适当放宽, 容忍 Uptime Kuma 偶发的慢响应
|
||||
REQUEST_TIMEOUT = 10
|
||||
# 单次心跳的最大尝试次数 (含首次)
|
||||
MAX_RETRIES = 3
|
||||
# 重试间隔(秒)
|
||||
RETRY_BACKOFF = 2
|
||||
# stop() 时等待后台线程退出的最长时间(秒); 线程为 daemon, 超时也不会阻塞进程退出
|
||||
STOP_JOIN_TIMEOUT = 40
|
||||
|
||||
|
||||
class UptimeKumaMonitor:
|
||||
"""
|
||||
Uptime Kuma 心跳监控器
|
||||
|
||||
推荐 (后台线程模式, 与业务循环解耦)::
|
||||
|
||||
monitor = UptimeKumaMonitor({
|
||||
'enabled': True,
|
||||
'push_url': 'https://uptimekuma.example.com/api/push/xxx',
|
||||
'heartbeat_interval': 59
|
||||
})
|
||||
monitor.start() # 启动后台线程: 立即发首跳, 之后按间隔周期发送
|
||||
try:
|
||||
... # 业务主循环
|
||||
finally:
|
||||
monitor.stop() # 停止线程并发送一次 down 信号
|
||||
|
||||
向后兼容 (同步模式, 不推荐新代码使用)::
|
||||
|
||||
monitor.send_heartbeat() # 同步发一次 (带重试)
|
||||
monitor.check_and_send_heartbeat() # 按间隔节流后同步发送
|
||||
monitor.send_stop_signal()
|
||||
"""
|
||||
|
||||
def __init__(self, config):
|
||||
"""
|
||||
Args:
|
||||
config (dict): 配置字典, 包含:
|
||||
- enabled (bool): 是否启用心跳
|
||||
- push_url (str): Uptime Kuma 推送 URL
|
||||
- heartbeat_interval (int): 心跳间隔(秒)
|
||||
"""
|
||||
self.config = config or {}
|
||||
self.enabled = self.config.get('enabled', False)
|
||||
self.push_url = self.config.get('push_url')
|
||||
self.heartbeat_interval = self.config.get('heartbeat_interval', 60)
|
||||
self._last_heartbeat_time = 0
|
||||
self._logger = None
|
||||
# 后台线程相关
|
||||
self._thread = None
|
||||
self._stop_event = threading.Event()
|
||||
|
||||
def set_logger(self, logger_func):
|
||||
"""设置日志记录函数 (如 log_warning)。"""
|
||||
self._logger = logger_func
|
||||
|
||||
def _log(self, message):
|
||||
if self._logger:
|
||||
self._logger(message)
|
||||
|
||||
# ================= 单次请求 + 重试 =================
|
||||
|
||||
def _do_request(self, status, msg):
|
||||
"""发送一次心跳请求, 返回是否成功 (HTTP 2xx)。失败时记录日志。"""
|
||||
try:
|
||||
params = {'status': status, 'msg': msg, 'ping': ''}
|
||||
response = requests.get(self.push_url, params=params, timeout=REQUEST_TIMEOUT)
|
||||
response.raise_for_status()
|
||||
return True
|
||||
except Exception as e:
|
||||
self._log(f"心跳发送失败 (status={status}): {e}")
|
||||
return False
|
||||
|
||||
def _send_with_retry(self, status='up', msg='OK'):
|
||||
"""
|
||||
带重试的心跳发送: 任一尝试成功即视为成功。
|
||||
成功发送 'up' 时更新最近心跳时间。返回是否最终成功。
|
||||
"""
|
||||
if not self.enabled or not self.push_url:
|
||||
return False
|
||||
for attempt in range(1, MAX_RETRIES + 1):
|
||||
if self._do_request(status, msg):
|
||||
if status == 'up':
|
||||
self._last_heartbeat_time = time.time()
|
||||
return True
|
||||
if attempt < MAX_RETRIES:
|
||||
time.sleep(RETRY_BACKOFF)
|
||||
return False
|
||||
|
||||
# ================= 后台线程模式 (推荐) =================
|
||||
|
||||
def start(self):
|
||||
"""
|
||||
启动后台心跳线程: 立即发送一次, 之后按 heartbeat_interval 周期发送。
|
||||
与业务主循环完全解耦, 心跳的网络耗时 / 重试不会阻塞业务。
|
||||
重复调用安全 (已在运行则直接返回)。
|
||||
"""
|
||||
if not self.enabled or not self.push_url:
|
||||
return
|
||||
if self._thread is not None and self._thread.is_alive():
|
||||
return
|
||||
self._stop_event.clear()
|
||||
self._thread = threading.Thread(
|
||||
target=self._heartbeat_loop, daemon=True, name='uptime-kuma-heartbeat')
|
||||
self._thread.start()
|
||||
|
||||
def _heartbeat_loop(self):
|
||||
# 启动立即发一次
|
||||
self._send_with_retry()
|
||||
# 周期发送, 直到 stop() 触发 _stop_event
|
||||
# Event.wait(interval) 在超时返回 False (继续发), 被置位时返回 True (退出)
|
||||
while not self._stop_event.wait(self.heartbeat_interval):
|
||||
self._send_with_retry()
|
||||
|
||||
def stop(self):
|
||||
"""停止后台心跳线程, 并向 Uptime Kuma 发送一次 down 信号。"""
|
||||
self._stop_event.set()
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=STOP_JOIN_TIMEOUT)
|
||||
self._thread = None
|
||||
self.send_stop_signal()
|
||||
|
||||
# ================= 向后兼容的同步接口 =================
|
||||
|
||||
def send_heartbeat(self):
|
||||
"""同步发送一次心跳 (自带重试)。向后兼容用法。"""
|
||||
return self._send_with_retry()
|
||||
|
||||
def send_stop_signal(self):
|
||||
"""发送停止(down)信号。失败不影响主逻辑。"""
|
||||
if not self.enabled or not self.push_url:
|
||||
return False
|
||||
return self._do_request('down', 'Service stopped')
|
||||
|
||||
def check_and_send_heartbeat(self):
|
||||
"""按间隔节流后同步发送心跳。向后兼容用法。"""
|
||||
if not self.enabled:
|
||||
return False
|
||||
if time.time() - self._last_heartbeat_time >= self.heartbeat_interval:
|
||||
return self.send_heartbeat()
|
||||
return False
|
||||
|
||||
def get_time_since_last_heartbeat(self):
|
||||
"""获取距离上次心跳的时间(秒)。"""
|
||||
return time.time() - self._last_heartbeat_time
|
||||
|
||||
@property
|
||||
def last_heartbeat_time(self):
|
||||
"""获取上次心跳时间戳。"""
|
||||
return self._last_heartbeat_time
|
||||
Reference in New Issue
Block a user