diff --git a/.env.example b/.env.example index 6d68e09..6b8d173 100644 --- a/.env.example +++ b/.env.example @@ -12,4 +12,3 @@ NTFY_TOKEN= # Uptime Kuma UPTIME_KUMA_PUSH_URL= -EXCEL_SYNC_UPTIME_KUMA_PUSH_URL= diff --git a/CLAUDE.md b/CLAUDE.md index 6f1b2ae..073e420 100644 --- a/CLAUDE.md +++ b/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 (`__.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) diff --git a/config/__init__.py b/config/__init__.py index f4e146e..422f1d7 100644 --- a/config/__init__.py +++ b/config/__init__.py @@ -7,17 +7,13 @@ 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, UPTIME_KUMA_CONFIG, EXCEL_SYNC_UPTIME_KUMA_CONFIG, - POLL_INTERVAL, EXCEL_SYNC_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, ) # 统一导出列表 @@ -25,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', 'UPTIME_KUMA_CONFIG', 'EXCEL_SYNC_UPTIME_KUMA_CONFIG', - 'POLL_INTERVAL', 'EXCEL_SYNC_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', ] diff --git a/config/app_settings.py b/config/app_settings.py index a04201d..2a1164c 100644 --- a/config/app_settings.py +++ b/config/app_settings.py @@ -36,96 +36,6 @@ UPTIME_KUMA_CONFIG = { 'heartbeat_interval': 59 # 心跳间隔(秒),需要与 Uptime Kuma 设置一致 } -# Excel 同步服务心跳 -EXCEL_SYNC_UPTIME_KUMA_CONFIG = { - 'enabled': True, - 'push_url': os.environ.get('EXCEL_SYNC_UPTIME_KUMA_PUSH_URL', ''), - 'heartbeat_interval': 59 # 心跳间隔(秒),需要与 Uptime Kuma 设置一致 -} - # ================= 运行参数 ================= -# 合并原 config.py 和 update_config.py 的配置 POLL_INTERVAL = 30 # 轮询间隔(秒) -EXCEL_SYNC_INTERVAL = 600 # Excel 同步周期(秒),默认 10 分钟 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, - "位号": "位号" -} diff --git a/config/field_mappings.py b/config/field_mappings.py deleted file mode 100644 index ee20a64..0000000 --- a/config/field_mappings.py +++ /dev/null @@ -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, - "位号": "位号" -} diff --git a/config/file_sources.py b/config/file_sources.py index 34199d2..cad17e3 100644 --- a/config/file_sources.py +++ b/config/file_sources.py @@ -313,160 +313,10 @@ SYNC_MAPPING = { } }, 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": { - "车间号": "车间号", - "工令号": "工令号", - "客户型号": "客户型号" - } - } -] diff --git a/docs/superpowers/specs/2026-06-17-remove-excel-migration-design.md b/docs/superpowers/specs/2026-06-17-remove-excel-migration-design.md new file mode 100644 index 0000000..0c0e64e --- /dev/null +++ b/docs/superpowers/specs/2026-06-17-remove-excel-migration-design.md @@ -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 +``` +`temp/` 缓存可由历史 Excel 脚本重新生成(已无意义)。 + +--- + +## 8. 风险与说明 + +- **无数据风险**:不动 SQL Server 任何表与数据。 +- **无运行中服务风险**:`run_incremental_sync.py`(Access 增量)代码路径不变,导入符号均保留。 +- **CLAUDE.md 改动较大**:因原文已与实际代码脱节,顺带修正为真实结构;如只希望"最小改动"可告知,仅删 Excel 段落、不补真实结构。 diff --git a/excel_sync_to_sql.py b/excel_sync_to_sql.py deleted file mode 100644 index 33eb661..0000000 --- a/excel_sync_to_sql.py +++ /dev/null @@ -1,369 +0,0 @@ -import os -import sys -import shutil -import logging -import argparse -import datetime -import urllib.parse -import warnings -import time -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, log_stop, LoggerManager) -from config import (DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA, - EXCEL_SYNC_INTERVAL, EXCEL_SYNC_UPTIME_KUMA_CONFIG) -from uptime_kuma_utils import UptimeKumaMonitor - -# 初始化 Uptime Kuma 监控器 -excel_uptime_monitor = UptimeKumaMonitor(EXCEL_SYNC_UPTIME_KUMA_CONFIG) -excel_uptime_monitor.set_logger(log_warning) - -# ================= 抑制 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': '', '': ''}) - - # 强制截断 - 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) - -# ================= Uptime Kuma 心跳 ================= -# 使用 uptime_kuma_utils.UptimeKumaMonitor 替代原有实现 - -def main(): - # 初始化日志管理器 - LoggerManager("excel_sync", log_prefix="excel_sync") - - # 解析参数 - parser = argparse.ArgumentParser(description="Excel数据同步至SQL Server") - parser.add_argument('--force', action='store_true', help='强制同步所有文件') - parser.add_argument('--once', action='store_true', help='只运行一次后退出') - args = parser.parse_args() - - syncer = DataSynchronizer(force_sync=args.force) - - # 启动信息 - mode = "强制模式" if args.force else "增量模式" - if args.once: - log_start(f"Excel 同步任务 ({mode}, 单次运行)") - syncer.process_excel_files() - syncer.generate_contract_data() - log_complete("Excel 同步任务已完成") - return - - # 周期性运行模式 - log_start(f"Excel 同步服务已启动 ({mode})") - log_info(f"同步周期: {EXCEL_SYNC_INTERVAL} 秒 ({EXCEL_SYNC_INTERVAL//60} 分钟)") - if EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False): - log_info(f"心跳间隔: {EXCEL_SYNC_UPTIME_KUMA_CONFIG['heartbeat_interval']} 秒") - log_info("=" * 70) - - # 启动时发送第一次心跳 - excel_uptime_monitor.send_heartbeat() - - try: - while True: - try: - # 执行同步任务 - log_info(f"开始执行周期性同步检查...") - syncer.process_excel_files() - syncer.generate_contract_data() - log_info(f"周期性同步检查完成") - - # 下次同步时间 - next_sync_time = time.time() + EXCEL_SYNC_INTERVAL - log_info(f"下次同步将在 {EXCEL_SYNC_INTERVAL//60} 分钟后进行") - - # 等待下次同步,期间持续发送心跳 - while time.time() < next_sync_time: - # 检查是否需要发送心跳 - excel_uptime_monitor.check_and_send_heartbeat() - - # 短暂休眠 - time.sleep(1) - - except KeyboardInterrupt: - log_info("=" * 70) - log_stop("收到停止信号,服务正在关闭...") - break - except Exception as e: - log_error(f"同步任务异常: {e}", exc_info=True) - log_info(f"将在 {EXCEL_SYNC_INTERVAL//60} 分钟后重试...") - time.sleep(EXCEL_SYNC_INTERVAL) - finally: - # 停止时发送心跳停止信号 - excel_uptime_monitor.send_stop_signal() - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/migration.py b/migration.py deleted file mode 100644 index 51f92fd..0000000 --- a/migration.py +++ /dev/null @@ -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() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 357c790..a4a5f13 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +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 \ No newline at end of file diff --git a/uptime_kuma_utils.py b/uptime_kuma_utils.py index 9f79c78..c506b6c 100644 --- a/uptime_kuma_utils.py +++ b/uptime_kuma_utils.py @@ -8,7 +8,7 @@ Uptime Kuma 心跳监控工具 心跳的网络耗时 / 重试不会阻塞业务逻辑。 - 单次心跳自带重试 (MAX_RETRIES), 容忍 Uptime Kuma 服务偶发的慢响应 / 超时 / 4xx。 - send_heartbeat() / check_and_send_heartbeat() / send_stop_signal() 保留为 - 向后兼容的同步接口 (excel_sync_to_sql.py 等仍在使用)。 + 向后兼容的同步接口。 """ import time