From 75ed7c775ecea17d1051954514308bac395d2698 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Tue, 21 Jul 2026 17:26:15 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=AF=BC=E5=85=A5=E5=89=8D?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=B8=85=E6=B4=97=EF=BC=9A=E6=8C=89=E5=88=9B?= =?UTF-8?q?=E5=BB=BA=E6=97=B6=E9=97=B4=E7=AD=9B=E9=80=89(=E4=B8=A2?= =?UTF-8?q?=E5=BC=83<2026-01-01,=20=E9=85=8D=E7=BD=AE=E5=8C=96)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 src/data_filter.py：filter_records 按 cfg['filter'] 筛选，字段值 >= start_date 保留；drop_if_missing 时缺失/解析失败的行一并丢弃； 配置缺失或 start_date 非法时安全降级为不筛选（不崩、不误删）。 - runner.run_once：parse_excel 之后、sync_all(及 dry-run) 之前插入 records = data_filter.filter_records(...)，SQL Server 与 Access 同时生效。 - config 新增 filter 段(field/start_date/drop_if_missing)；config.yaml 被 .gitignore 忽略不入库，部署时需并入 114 的 config（仅新增该段）。 - 本机实跑 + 共享 SQL Server 实库验证：35389→12918，22471 行<2026-01-01 已清零，MIN(创建时间)=2026-01-04，PASS。 --- src/data_filter.py | 74 ++++++++++++++++++++++++++++++++++++++++++++++ src/runner.py | 5 +++- 2 files changed, 78 insertions(+), 1 deletion(-) create mode 100644 src/data_filter.py diff --git a/src/data_filter.py b/src/data_filter.py new file mode 100644 index 0000000..f2661db --- /dev/null +++ b/src/data_filter.py @@ -0,0 +1,74 @@ +"""数据清洗：导入前按配置的日期字段筛选行。 + +仅保留 ``field`` 字段值 >= ``start_date`` 的行；缺失 / 解析失败的行 +按 ``drop_if_missing`` 决定保留或丢弃。筛选在 ``sync_all`` 之前执行， +因此对 SQL Server 与 Access 两个目标同时生效。 +""" +import logging +from datetime import datetime + +logger = logging.getLogger(__name__) + +_DATE_FORMATS = ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%Y/%m/%d %H:%M:%S", "%Y/%m/%d") + + +def _coerce_datetime(v): + """把任意值收敛为 datetime；无法解析返回 None。""" + if isinstance(v, datetime): + return v + if v is None: + return None + s = str(v).strip() + if not s: + return None + for fmt in _DATE_FORMATS: + try: + return datetime.strptime(s, fmt) + except ValueError: + continue + return None + + +def filter_records(records, cfg): + """按 ``cfg['filter']`` 筛选 records，返回新列表（不修改入参）。 + + 若 cfg 无 ``filter`` 段，原样返回（不筛选）——配置缺失时安全降级。 + """ + fcfg = (cfg or {}).get("filter") + if not fcfg: + return records + + field = fcfg.get("field") + start_str = fcfg.get("start_date") + drop_if_missing = bool(fcfg.get("drop_if_missing", True)) + if not field or not start_str: + logger.warning("filter 配置不完整（缺 field/start_date），跳过筛选") + return records + + try: + start_dt = datetime.strptime(start_str, "%Y-%m-%d") + except ValueError: + logger.warning("filter.start_date 非法(%r)，跳过筛选", start_str) + return records + + kept, dropped_old, dropped_missing = [], 0, 0 + for rec in records: + v = rec.get(field) + dt = _coerce_datetime(v) + if dt is None: + if drop_if_missing: + dropped_missing += 1 + continue + kept.append(rec) + continue + if dt >= start_dt: + kept.append(rec) + else: + dropped_old += 1 + + total = len(records) + logger.info( + "数据清洗: 输入 %d 行 | 丢弃 %s<%s: %d 行 | 丢弃 缺失/解析失败: %d 行 | 保留 %d 行", + total, field, start_str, dropped_old, dropped_missing, len(kept), + ) + return kept diff --git a/src/runner.py b/src/runner.py index 6459977..65077be 100644 --- a/src/runner.py +++ b/src/runner.py @@ -16,7 +16,7 @@ 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 +from . import source_watcher, data_filter logger = logging.getLogger(__name__) @@ -128,6 +128,9 @@ def run_once(cfg, source_path=None, dry_run=False): records, fields = parse_excel(local_file, cfg.get("field_map", {})) + # 数据清洗：导入前按配置筛选（缺失/早于阈值的一并丢弃） + records = data_filter.filter_records(records, cfg) + if dry_run: import_col = cfg.get("auto_fields", {}).get("import_time_field", "导入时间") s = cfg["sql_server"]