commit 5ae7acffde86ecd38d079972b581e15df50ec122 Author: Misaka_Company Date: Wed Jul 15 14:14:39 2026 +0800 初始化项目:数据同步服务源码与文档 - src: config_loader / source_watcher / excel_parser / sync_writer / runner - 169 Excel 全量同步至 114 SQL Server + Access 生产库 - 新增 .gitignore(忽略 venv、日志、数据、含凭据的 config.yaml 等) - 新增 README.md(用途、安装、配置、运行模式、部署说明) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..91577b7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,38 @@ +# ===== Python 运行时 / 字节码 ===== +.venv/ +venv/ +__pycache__/ +*.py[cod] +*$py.class +*.so + +# ===== 日志与运行输出 ===== +logs/ +*.log + +# ===== 数据与同步产物(运行时生成,非源码) ===== +data/ + +# ===== 配置(可能含数据库凭据,勿提交;如需共享请改用 config.yaml.example) ===== +config/config.yaml + +# ===== 环境变量 / 密钥 ===== +.env +.env.* +*.secret + +# ===== 编辑器 / 系统垃圾 ===== +.DS_Store +Thumbs.db +.idea/ +.vscode/ + +# ===== 本地 Agent 运行时状态(非项目源码) ===== +.claude/ +.workbuddy/ + +# ===== 构建 / 打包产物 ===== +build/ +dist/ +*.egg-info/ +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..94a3c7b --- /dev/null +++ b/README.md @@ -0,0 +1,123 @@ +# ProcurementVisibilityHub_DataSync + +采购执行情况即时数据同步服务。将 169 共享盘上的 Excel(`采购执行情况即时数据.xlsx`)同步写入两个目标库: + +- **SQL Server**(114 / `CompanyDB`,schema `procurementVisibilityHub`,表 `请购执行`) +- **Access 生产库**(`ProcurementVisibilityHub.accdb`,表 `procurementVisibilityHub_请购执行`) + +采用**全量覆盖**策略:SQL Server 端 `TRUNCATE + INSERT`,Access 端 `DELETE + INSERT`,并写入统一的「导入时间」。 + +--- + +## 目录结构 + +| 路径 | 说明 | +|---|---| +| `config/config.yaml` | 运行配置(含 DB 凭据,**已被 .gitignore 忽略**) | +| `src/config_loader.py` | 读取 YAML 配置、构造 ODBC 连接串 | +| `src/source_watcher.py` | 源文件变化检测(mtime/size)+ 下载到 `data/` | +| `src/excel_parser.py` | 解析 xlsx → 行字典,含字段映射与类型转换 | +| `src/sync_writer.py` | 全量写入 SQL Server + Access | +| `src/runner.py` | 主入口:服务循环 / 单次 / 本地联调 / dry-run | +| `data/` | 源文件历史副本(运行时生成,已忽略) | +| `logs/sync.log` | 运行日志(运行时生成,已忽略) | + +--- + +## 环境要求 + +- Python **3.10+** +- 目标主机需安装 ODBC 驱动: + - **ODBC Driver 17 for SQL Server**(连 114 SQL Server) + - **Microsoft Access Driver (*.mdb, *.accdb)**(连 Access 生产库) +- 依赖见 `requirements.txt`:`PyYAML`、`openpyxl`、`pyodbc` + +--- + +## 安装 + +```bash +# 1. 创建并激活虚拟环境(项目内) +python -m venv .venv +.\.venv\Scripts\activate # Windows +# source .venv/bin/activate # Linux/macOS + +# 2. 安装依赖 +pip install -r requirements.txt + +# 3. 准备配置(从模板复制后填入真实凭据) +cp config/config.yaml.example config/config.yaml # 若无模板,直接新建 config/config.yaml +``` + +> ⚠️ `config/config.yaml` 含数据库账号密码,**不纳入版本管理**。如需团队共享,请提交一份 `config/config.yaml.example`(占位凭据),真实配置留本地。 + +--- + +## 配置说明(config/config.yaml 关键字段) + +| 字段 | 说明 | +|---|---| +| `source.unc_path` | 源 Excel 的 UNC 路径(169 共享盘) | +| `source.local_cache_dir` | 本地历史副本目录(默认 `data`) | +| `scan.interval_seconds` | 服务循环扫描间隔(秒) | +| `sql_server.*` | SQL Server 连接信息(server/port/database/user/password…) | +| `access.*` | Access 库路径与表名(DBQ/table) | +| `field_map` | Excel 列名 → 数据库字段名(仅列名不同者) | +| `auto_fields` | 数据库自动生成字段:`id_field`(自增不写)、`import_time_field`(=同步时刻) | +| `sync.strategy` | 同步策略,当前固定 `full` | +| `sync.batch_size` | 批量插入大小(默认 1000) | +| `logging` | 日志级别与目录 | + +--- + +## 使用方式 + +```bash +# 服务循环模式(默认,配合 NSSM 部署为 Windows 服务) +python -m src.runner + +# 单次运行后退出 +python -m src.runner --once + +# 用本地文件作源(跳过 169 下载,便于联调) +python -m src.runner --local X.xlsx + +# 仅解析并打印,不写库 +python -m src.runner --local X.xlsx --dry-run + +# 指定配置文件 +python -m src.runner --config config/config.yaml +``` + +运行逻辑:检测源 Excel 是否变化(mtime/size),无变化则跳过;有变化则下载副本 → 解析 → 全量写入两个目标库。 + +--- + +## 运行流程 + +``` +169 共享 Excel + │ source_watcher 检测变化 + 下载到 data/ + ▼ +excel_parser 解析(字段映射 + 类型转换:float/datetime/int/text) + ▼ +sync_writer 全量写入 + ├─ SQL Server:TRUNCATE + 分批 INSERT(fast_executemany,事务提交) + └─ Access: DELETE + 分批 INSERT(autocommit) + ▼ +logs/sync.log 记录全过程 +``` + +--- + +## 部署(NSSM 服务) + +以 Windows 服务常驻运行(主机 114),由 NSSM 拉起 `python -m src.runner` 进入循环模式。日志统一写入 `logs/sync.log`。 + +--- + +## 常见问题 + +- **连不上 SQL Server**:确认 ODBC Driver 17 已装、114:1433 可达、账号密码正确(`Encrypt`/`TrustServerCertificate` 与服务器端匹配)。 +- **Access 写入失败**:确认目标主机装了 Microsoft Access Driver,且 `access.dbq` 路径存在、有写权限。 +- **源无变化不执行**:属正常行为(基于 mtime/size 的增量跳过);想强制全量可用 `--local` 指定当前文件或删除 `data/.source_meta.json` 基准。 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6d182e0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +PyYAML>=6.0 +openpyxl>=3.1 +pyodbc>=5.0 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/config_loader.py b/src/config_loader.py new file mode 100644 index 0000000..70115bd --- /dev/null +++ b/src/config_loader.py @@ -0,0 +1,29 @@ +"""配置加载:从 YAML 读取全部参数,并构造 ODBC 连接串。""" +from pathlib import Path + +import yaml + + +def load_config(path: str = "config/config.yaml") -> dict: + cfg_path = Path(path) + if not cfg_path.exists(): + raise FileNotFoundError(f"配置文件不存在: {cfg_path}") + with open(cfg_path, "r", encoding="utf-8") as f: + return yaml.safe_load(f) + + +def build_sql_connstr(cfg: dict) -> str: + s = cfg["sql_server"] + return ( + f"DRIVER={{{s['driver']}}};" + f"SERVER={s['server']},{s['port']};" + f"DATABASE={s['database']};" + f"UID={s['user']};PWD={s['password']};" + f"Encrypt={'yes' if s.get('encrypt') else 'no'};" + f"TrustServerCertificate={'yes' if s.get('trust_server_certificate') else 'no'}" + ) + + +def build_access_connstr(cfg: dict) -> str: + a = cfg["access"] + return f"DRIVER={{{a['driver']}}};DBQ={a['dbq']};" diff --git a/src/excel_parser.py b/src/excel_parser.py new file mode 100644 index 0000000..8aace58 --- /dev/null +++ b/src/excel_parser.py @@ -0,0 +1,127 @@ +"""Excel 解析:读取 xlsx → 字段映射 → 类型转换,输出行字典。""" +import logging +from datetime import datetime +from pathlib import Path + +import openpyxl +from openpyxl.utils.datetime import from_excel + +logger = logging.getLogger(__name__) + +# 请购执行表字段类型(基于 SQL Server 结构;未列出者按 text 处理) +FIELD_TYPES = { + # float:数量 / 金额 + "主数量": "float", "到货主数量": "float", "未到货数量": "float", + "订货主数量": "float", "订货单价": "float", "订货金额": "float", + "退货主数量": "float", "入库主数量": "float", "退库主数量": "float", + "补货主数量": "float", "未执行关闭数量": "float", + # datetime:日期 / 时间 + "创建时间": "datetime", "请购日期": "datetime", "到货日期": "datetime", + "需求日期": "datetime", "建议订货日期": "datetime", "计划到货日期": "datetime", + "订单日期": "datetime", "入库日期": "datetime", + # int + "行号": "int", +} + +_DATE_FORMATS = ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%Y/%m/%d %H:%M:%S", "%Y/%m/%d") + + +def _to_float(v): + if v is None or v == "": + return None + try: + return float(str(v).strip()) + except (ValueError, TypeError): + logger.warning("float 转换失败: %r", v) + return None + + +def _to_int(v): + if v is None or v == "": + return None + try: + return int(float(str(v).strip())) + except (ValueError, TypeError): + logger.warning("int 转换失败: %r", v) + return None + + +def _to_datetime(v): + if v is None or v == "": + return None + if isinstance(v, datetime): + return v + if isinstance(v, (int, float)): + try: + return from_excel(v) + except (ValueError, OverflowError): + logger.warning("日期序列号转换失败: %r", v) + return None + s = str(v).strip().lstrip("'") + if not s: + return None + for fmt in _DATE_FORMATS: + try: + return datetime.strptime(s, fmt) + except ValueError: + continue + logger.warning("日期解析失败: %r", v) + return None + + +def _to_text(v): + if v is None: + return None + s = str(v).strip() + return s if s else None + + +CONVERTERS = { + "float": _to_float, + "int": _to_int, + "datetime": _to_datetime, + "text": _to_text, +} + + +def parse_excel(path, field_map=None, field_types=None): + """解析 xlsx,返回 (records, fields)。 + + records: list[dict],键为数据库字段名(经 field_map 映射)。 + fields: list[str],数据库字段名顺序(去空去重)。 + """ + field_map = field_map or {} + field_types = field_types or FIELD_TYPES + path = Path(path) + wb = openpyxl.load_workbook(path, read_only=True, data_only=True) + try: + ws = wb.active + rows = list(ws.iter_rows(values_only=True)) + finally: + wb.close() + if not rows: + return [], [] + + raw_header = [str(c).strip() if c is not None else "" for c in rows[0]] + mapped_header = [field_map.get(h, h) for h in raw_header] + + records = [] + for row in rows[1:]: + if all(c is None or c == "" for c in row): + continue + rec = {} + for idx, field in enumerate(mapped_header): + if not field or idx >= len(row): + continue + raw = row[idx] + rec[field] = CONVERTERS[field_types.get(field, "text")](raw) + records.append(rec) + + fields, seen = [], set() + for f in mapped_header: + if f and f not in seen: + seen.add(f) + fields.append(f) + + logger.info("解析完成: %d 行, %d 字段", len(records), len(fields)) + return records, fields diff --git a/src/runner.py b/src/runner.py new file mode 100644 index 0000000..2e2d70f --- /dev/null +++ b/src/runner.py @@ -0,0 +1,97 @@ +"""同步服务主入口(NSSM 部署)。 + +用法: + python -m src.runner # 服务循环模式(NSSM) + python -m src.runner --once # 单次运行后退出 + python -m src.runner --local X.xlsx # 用本地文件作源(跳过 169 下载,联调用) + python -m src.runner --local X.xlsx --dry-run # 只解析并打印,不写库 +""" +import argparse +import logging +import os +import sys +import time +from pathlib import Path + +from .config_loader import load_config +from .excel_parser import parse_excel +from .sync_writer import sync_all, _build_insert +from . import source_watcher + +logger = logging.getLogger(__name__) + + +def setup_logging(cfg): + log_cfg = cfg.get("logging", {}) + level = getattr(logging, log_cfg.get("level", "INFO").upper(), logging.INFO) + log_dir = Path(log_cfg.get("dir", "logs")) + log_dir.mkdir(parents=True, exist_ok=True) + fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s") + root = logging.getLogger() + root.setLevel(level) + fh = logging.FileHandler(log_dir / "sync.log", encoding="utf-8") + fh.setFormatter(fmt) + root.addHandler(fh) + sh = logging.StreamHandler(sys.stdout) + sh.setFormatter(fmt) + root.addHandler(sh) + + +def run_once(cfg, source_path=None, dry_run=False): + if source_path: + local_file = source_path + logger.info("使用本地源: %s", local_file) + else: + unc = cfg["source"]["unc_path"] + cache = cfg["source"]["local_cache_dir"] + meta_path = os.path.join(cache, ".source_meta.json") + if not source_watcher.has_changed(unc, meta_path): + return + local_file = source_watcher.download(unc, cache) + source_watcher.save_baseline(meta_path, source_watcher.get_source_meta(unc)) + + records, fields = parse_excel(local_file, cfg.get("field_map", {})) + + if dry_run: + import_col = cfg.get("auto_fields", {}).get("import_time_field", "导入时间") + s = cfg["sql_server"] + sql_tbl = f"{s['schema']}.[{s['table']}]" + logger.info("[dry-run] 字段数=%d 行数=%d", len(fields), len(records)) + logger.info("[dry-run] SQL INSERT: %s", _build_insert(sql_tbl, fields, import_col)) + for i, r in enumerate(records[:3]): + logger.info("[dry-run] 样本%d: %s", i + 1, {k: r[k] for k in fields[:6]}) + return + + sync_all(cfg, records, fields) + + +def main(): + parser = argparse.ArgumentParser(description="采购执行情况即时数据 同步服务") + parser.add_argument("--once", action="store_true", help="单次运行后退出") + parser.add_argument("--local", metavar="PATH", help="用本地文件作源(跳过 169 下载)") + parser.add_argument("--dry-run", action="store_true", help="只解析并打印,不写库") + parser.add_argument("--config", default="config/config.yaml") + args = parser.parse_args() + + cfg = load_config(args.config) + setup_logging(cfg) + + if args.local or args.dry_run: + run_once(cfg, source_path=args.local, dry_run=args.dry_run) + return + if args.once: + run_once(cfg) + return + + interval = cfg["scan"]["interval_seconds"] + logger.info("服务循环启动,扫描间隔 %s 秒", interval) + while True: + try: + run_once(cfg) + except Exception as e: + logger.exception("同步周期失败: %s", e) + time.sleep(interval) + + +if __name__ == "__main__": + main() diff --git a/src/source_watcher.py b/src/source_watcher.py new file mode 100644 index 0000000..8e5bca3 --- /dev/null +++ b/src/source_watcher.py @@ -0,0 +1,53 @@ +"""源文件变化检测与下载。""" +import json +import logging +import os +import shutil +from datetime import datetime +from pathlib import Path + +logger = logging.getLogger(__name__) + + +def get_source_meta(path): + st = os.stat(path) + return {"mtime": st.st_mtime, "size": st.st_size} + + +def load_baseline(meta_path): + p = Path(meta_path) + if not p.exists(): + return None + with open(p, "r", encoding="utf-8") as f: + return json.load(f) + + +def save_baseline(meta_path, meta): + Path(meta_path).parent.mkdir(parents=True, exist_ok=True) + with open(meta_path, "w", encoding="utf-8") as f: + json.dump(meta, f) + + +def has_changed(path, meta_path): + try: + cur = get_source_meta(path) + except OSError as e: + logger.error("无法访问源文件 %s: %s", path, e) + raise + base = load_baseline(meta_path) + if base is None: + logger.info("无基准记录,视为有变化") + return True + changed = (cur["mtime"] != base["mtime"]) or (cur["size"] != base["size"]) + logger.info("源文件 %s", "有变化" if changed else "无变化") + return changed + + +def download(path, cache_dir): + cache = Path(cache_dir) + cache.mkdir(parents=True, exist_ok=True) + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + dst = cache / f"{Path(path).stem}_{ts}.xlsx" + shutil.copy2(path, str(dst)) + logger.info("已下载: %s -> %s", path, dst) + return str(dst) diff --git a/src/sync_writer.py b/src/sync_writer.py new file mode 100644 index 0000000..82a04bc --- /dev/null +++ b/src/sync_writer.py @@ -0,0 +1,87 @@ +"""全量同步写入:SQL Server (TRUNCATE+INSERT) + Access (DELETE+INSERT)。""" +import logging +from datetime import datetime + +import pyodbc + +from .config_loader import build_sql_connstr, build_access_connstr + +logger = logging.getLogger(__name__) + + +def _build_insert(table_expr, fields, import_col): + """构造 INSERT SQL。table_expr 为完整表表达式(含 schema/方括号)。""" + cols = list(fields) + [import_col] + col_list = ",".join(f"[{c}]" for c in cols) + placeholders = ",".join(["?"] * len(cols)) + return f"INSERT INTO {table_expr} ({col_list}) VALUES ({placeholders})" + + +def _to_rows(records, fields, import_time): + return [tuple(r.get(c) for c in fields) + (import_time,) for r in records] + + +def sync_sql_server(cfg, records, fields, import_time): + if not records: + logger.warning("SQL Server: 无数据,跳过") + return + s = cfg["sql_server"] + table_expr = f"{s['schema']}.[{s['table']}]" # procurementVisibilityHub.[请购执行] + import_col = cfg.get("auto_fields", {}).get("import_time_field", "导入时间") + insert_sql = _build_insert(table_expr, fields, import_col) + rows = _to_rows(records, fields, import_time) + batch = cfg.get("sync", {}).get("batch_size", 1000) + + cn = pyodbc.connect(build_sql_connstr(cfg), autocommit=False) + try: + cur = cn.cursor() + logger.info("SQL Server: 清空 %s", table_expr) + try: + cur.execute(f"TRUNCATE TABLE {table_expr}") + except pyodbc.Error as e: + logger.warning("TRUNCATE 失败(%s),改用 DELETE", e) + cur.execute(f"DELETE FROM {table_expr}") + cur.fast_executemany = True + for i in range(0, len(rows), batch): + cur.executemany(insert_sql, rows[i:i + batch]) + logger.info("SQL Server: 已插入 %d/%d", min(i + batch, len(rows)), len(rows)) + cn.commit() + logger.info("SQL Server: 完成,共 %d 行", len(rows)) + except Exception: + cn.rollback() + raise + finally: + cn.close() + + +def sync_access(cfg, records, fields, import_time): + if not records: + logger.warning("Access: 无数据,跳过") + return + a = cfg["access"] + table_expr = f"[{a['table']}]" # [procurementVisibilityHub_请购执行] + import_col = cfg.get("auto_fields", {}).get("import_time_field", "导入时间") + insert_sql = _build_insert(table_expr, fields, import_col) + rows = _to_rows(records, fields, import_time) + batch = cfg.get("sync", {}).get("batch_size", 1000) + + cn = pyodbc.connect(build_access_connstr(cfg), autocommit=True) + try: + cur = cn.cursor() + logger.info("Access: 清空 %s", table_expr) + cur.execute(f"DELETE FROM {table_expr}") + total = len(rows) + for i in range(0, total, batch): + cur.executemany(insert_sql, rows[i:i + batch]) + logger.info("Access: 已插入 %d/%d", min(i + batch, total), total) + logger.info("Access: 完成,共 %d 行", total) + finally: + cn.close() + + +def sync_all(cfg, records, fields): + import_time = datetime.now().replace(microsecond=0) + logger.info("=== 开始全量同步 导入时间=%s 数据=%d 行 ===", import_time, len(records)) + sync_sql_server(cfg, records, fields, import_time) + sync_access(cfg, records, fields, import_time) + logger.info("=== 同步完成 ===")