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