5e3c5d5b31
根因：源 Excel 常被打开/保存，download 在拷贝瞬间抓到 0 字节或 残缺文件，openpyxl 解析即报 “File is not a zip file”；且 runner 在 download 之后、parse 之前就无条件 save_baseline，把 size=0 的瞬态 源记成 “已处理”，污染基线。 修复： - source_watcher.download：下载后校验（非空 + ZIP 头 PK），不合法 则删除并重试（默认 3 次，间隔 3s 等待源落盘）。 - source_watcher.has_changed：源不可达时优雅跳过本轮（WARNING）， 不再每轮刷 traceback（169 间歇不可达为已知问题）。 - runner.run_once：save_baseline 移到整轮（解析+写库）成功之后， 杜绝把损坏/瞬态源记为已处理。
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
"""同步服务主入口（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):
|
||
unc = cfg["source"]["unc_path"]
|
||
cache = cfg["source"]["local_cache_dir"]
|
||
meta_path = os.path.join(cache, ".source_meta.json")
|
||
|
||
if source_path:
|
||
local_file = source_path
|
||
logger.info("使用本地源: %s", local_file)
|
||
else:
|
||
if not source_watcher.has_changed(unc, meta_path):
|
||
return
|
||
local_file = source_watcher.download(unc, cache)
|
||
|
||
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)
|
||
|
||
# 仅整轮（解析 + 写库）成功后才落基线，
|
||
# 避免把瞬态/损坏源记为“已处理”（曾导致基线被记成 size=0）。
|
||
if not source_path:
|
||
try:
|
||
source_watcher.save_baseline(meta_path, source_watcher.get_source_meta(unc))
|
||
except OSError as e:
|
||
logger.warning("保存基线失败(不影响本次同步): %s", e)
|
||
|
||
|
||
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()
|