初始化项目：数据同步服务源码与文档

- src: config_loader / source_watcher / excel_parser / sync_writer / runner - 169 Excel 全量同步至 114 SQL Server + Access 生产库 - 新增 .gitignore（忽略 venv、日志、数据、含凭据的 config.yaml 等） - 新增 README.md（用途、安装、配置、运行模式、部署说明）
2026-07-15 14:14:39 +08:00
commit 5ae7acffde
9 changed files with 557 additions and 0 deletions
--- a/.gitignore
+++ 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/
--- a/README.md
+++ 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` 基准。
--- a/requirements.txt
+++ b/requirements.txt
@@ -0,0 +1,3 @@
+PyYAML>=6.0
+openpyxl>=3.1
+pyodbc>=5.0
--- a/src/init.py
+++ b/src/init.py
--- a/src/config_loader.py
+++ 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']};"
--- a/src/excel_parser.py
+++ 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
--- a/src/runner.py
+++ 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()
--- a/src/source_watcher.py
+++ 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)
--- a/src/sync_writer.py
+++ 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("=== 同步完成 ===")