新增导入前数据清洗:按创建时间筛选(丢弃<2026-01-01, 配置化)

- 新增 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。
This commit is contained in:
Misaka_Company
2026-07-21 17:26:15 +08:00
parent 0750f4c41d
commit 75ed7c775e
2 changed files with 78 additions and 1 deletions

74
src/data_filter.py Normal file
View File

@@ -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

View File

@@ -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"]