duekit

English | 中文文档

---

English

duekit is a framework-agnostic Python toolkit for due-date monitoring, irregular-table extraction and roster data handling. It bundles eight independently usable, highly configurable submodules behind a single import:

Submodule	Solves
duekit.scheduler	When should a check run — multiple daily clock times and fixed intervals, each entry running its own task; pure scheduling logic plus an optional thread runner.
duekit.due	What state is a date in — parse messy dates and classify them against fully custom levels (you choose the number, names and day thresholds) with optional per-level callbacks.
duekit.table	How to read messy tables — schema-driven extraction from PDF / Excel / CSV, merging records split across page breaks.
duekit.clean	full/half-width unification, invisible-character removal and whitespace folding, with a reusable field-aware pipeline.
duekit.renew	renewal / re-certification date arithmetic (validity periods in years/months/days, month-end snap, multi-step schedule).
duekit.validate	required / unique / range / choice / regex / cross-field checks that collect every issue.
duekit.stats	aggregate level counts and percentages, expired/upcoming lists and average remaining days, optionally grouped.
duekit.export	write records back to a styled Excel workbook (level-colored rows) or CSV.

The core package has zero third-party dependencies (standard library only). PDF/Excel backends are optional extras.

Installation

pip install duekit                 # scheduler + due, zero third-party deps
pip install "duekit[table]"       # adds pdfplumber + openpyxl

Quickstart

import duekit

# --- scheduler: run a task every day at 06:30 and every 4 hours ---
from datetime import timedelta
schedule = duekit.Schedule()
schedule.daily("06:30", task=my_morning_check)
schedule.interval(timedelta(hours=4), task=my_interval_check)
schedule.run_pending()             # execute whatever is due right now

# --- due: classify a date against your own levels ---
scheme = duekit.LevelScheme(levels=[
    duekit.Level.expired(),
    duekit.Level("urgent", 7),
    duekit.Level("normal", None),  # catch-all
])
match = scheme.classify("2026.10.02", today=today)
print(match.name)                  # -> "urgent"

# --- table: extract rows described by a schema ---
records = duekit.extract("roster.pdf", schema=my_schema)

# --- clean: normalize messy text in records ---
cleaner = duekit.Cleaner(id_keys=("emp_id",))
records = cleaner.clean_records(records)

# --- renew: compute the next re-certification date (e.g. 4-year validity) ---
rule = duekit.RenewRule(years=4)
next_expiry = rule.apply(current_expiry)

# --- validate: collect every problem across all records ---
validator = duekit.Validator(
    [duekit.FieldRule("emp_id", [])],
    unique_keys=("emp_id",),
)
issues = validator.validate(records)

# --- stats + export: summarize then write a level-colored Excel file ---
summary = duekit.summarize(records, expiry_key="expiry", scheme=scheme)["all"]
duekit.export_excel("report.xlsx", records, level_key="level")

---

1. Scheduler — duekit.scheduler

Mix two kinds of entries; every entry binds its own task:

from datetime import timedelta
import duekit

schedule = duekit.Schedule(timezone="Asia/Shanghai")

# fixed clock time (optionally only on selected weekdays: Monday=0..Sunday=6)
schedule.daily("06:30", task=morning_job)
schedule.daily("22:30", task=night_job, days=(0, 1, 2, 3, 4))

# fixed interval; minutes number or timedelta; optional anchor for alignment
schedule.interval(timedelta(hours=4), task=often_job)
schedule.interval(30, task=every_30_min)            # 30 minutes

Pure logic vs thread runner

The scheduling math is pure and time-injectable — inspect or drive it with any clock, no sleeping required:

upcoming = schedule.next_run()              # earliest PlannedRun, not claimed
due = schedule.pending()                    # claim runs due since last call
results = schedule.run_pending()            # claim + execute tasks
for r in results:
    print(r.ok, r.value, r.error)           # task errors are isolated

For a long-running process, start the optional daemon-thread runner:

runner = duekit.ThreadRunner(schedule, poll_interval=1.0)
runner.start()
...
runner.stop()

You can instead call schedule.run_pending from cron, APScheduler or any external loop — the library never forces threads on you.

Deterministic tests

Inject a fixed clock so scheduler tests need no real waiting:

s = duekit.Schedule(
    [duekit.DailyTime(at="06:30", task=job)],
    clock=lambda: fixed_datetime,
)

---

2. Due levels — duekit.due

Parse messy dates

duekit.parse_date("2026.3.21")          # datetime.date(2026, 3, 21)
duekit.parse_date("2026年3月21日")
duekit.parse_date("Wed Feb 11 2026 08:00:00 GMT+0800 (...)")  # JS toString
duekit.days_remaining("2026.10.06", today)   # integer, negative when past

Fully custom levels

There is no fixed set of buckets — define how many levels exist, their names and thresholds. A level matches when days_remaining <= max_days; max_days=None is the catch-all. Levels are sorted automatically and validated (no duplicate thresholds, at most one catch-all).

scheme = duekit.LevelScheme(levels=[
    duekit.Level.expired(),                 # strictly past
    duekit.Level("urgent", 7),
    duekit.Level("warning", 15),
    duekit.Level("normal", None),
])

Need exactly three buckets? Just declare three. The original five-bucket behavior is one call away via LevelScheme.default().

Per-level action callbacks

Bind an action to a level and fire it when a date lands there:

def alert(match):
    print(match.name, match.days, match.target)

scheme = duekit.LevelScheme(levels=[
    duekit.Level("urgent", 7, action=alert),
    duekit.Level("normal", None),
])
match = scheme.classify(target, today, fire=True)   # runs the action
# or: match = scheme.classify(...); match.fire()

Multiple dates

# earliest expiry drives a single result
m = scheme.classify_many([date_a, date_b], today, how="earliest")

# one result per input
matches = scheme.classify_many([date_a, date_b], today, how="each", fire=True)

An unparseable value falls back to the catch-all; without one it raises, so bad input never silently becomes "normal".

---

3. Table extraction — duekit.table

Readers (all return RawRow)

duekit.read_pdf("roster.pdf")
duekit.read_excel("roster.xlsx", sheet=0)
duekit.read_csv("roster.csv")
duekit.read_any("roster.pdf")        # picks reader by extension

Declarative schema

schema = duekit.TableSchema.create(
    [
        duekit.Field("seq", 0, "int", required=True),
        duekit.Field("name", 1, "str"),
        duekit.Field("expiry", 2, "date"),
    ],
    start_keywords=["序号", "姓名"],     # processing starts after this header
    data_when=duekit.first_column_int,    # what counts as a data row
    skip_when=[                           # extra rows to skip
        duekit.column_has(7, "焊接方法"),
    ],
    continuation=duekit.Continuation(
        detect=duekit.first_column_empty, # split-record rows
        merge=duekit.smart_merge,
    ),
)
records = duekit.extract("roster.pdf", schema)

Field types: str / int / float / date. A record failing a required field is dropped.

Merging records split across rows

Real rosters split one logical record over two physical rows. The working record keeps raw strings, continuation rows merge in, and typed conversion runs only once the record is complete:

• date columns concatenate with no separator — 2027.07.2 + 8 becomes 2027.07.28;
• other columns insert a space only when joining two ASCII words (so "GTAW P FW" + "FM1" → "GTAW P FW FM1"); Chinese values concatenate directly.

Restrict which columns merge with Continuation(columns=(2,)); provide your own merge callable for any custom rule.

---

End-to-end example

import duekit

schedule = duekit.Schedule()
schedule.daily("06:30", task=check_expiries)

scheme = duekit.LevelScheme(levels=[
    duekit.Level.expired(name="expired"),
    duekit.Level("urgent", 7),
    duekit.Level("normal", None),
])

def check_expiries():
    records = duekit.extract("roster.pdf", schema=table_schema)
    today = date.today()
    for r in records:
        scheme.classify(r["expiry"], today, fire=True)   # alerts when urgent

# drive it manually, or: duekit.ThreadRunner(schedule).start()
schedule.run_pending()

Notes & limitations

• duekit performs date classification and table extraction; it does not send emails or push notifications — wire those into your level/task callables.
• Scheduler's default clock reads the local timezone; pass clock/today for deterministic output.
• PDF/Excel readers require the [table] extras.

License

MIT © 兰衣

---

中文文档

duekit 是一个与框架无关的 Python 工具包，用于到期日监控、不规则表格抽取 与台账数据处理。一次导入即可使用八个相互独立、高度可配置的子模块：

子模块	解决的问题
duekit.scheduler	何时执行检查——支持多个每日固定时刻与固定间隔，每个条目运行各自的任务；纯调度逻辑 + 可选线程运行器。
duekit.due	日期处于什么状态——解析各种脏日期，按完全自定义的级别（级别数量、名称、天数阈值都由你定）分级，并支持级别回调。
duekit.table	如何读取脏表格——按声明式方案从 PDF / Excel / CSV 抽取，并合并跨页被拆分的记录。
duekit.clean	全/半角统一、去除不可见字符、折叠空白，提供可复用的按字段清洗管道。
duekit.renew	续期/换证日期推算（年/月/日有效期、月末对齐、连续多次排期）。
duekit.validate	必填/唯一/值域/枚举/正则/跨字段校验，一次性收集全部问题。
duekit.stats	各级数量与占比、过期/将到期清单、平均剩余天数，支持分组。
duekit.export	写回带样式（按级别着色）的 Excel 或 CSV。

核心包零第三方依赖（仅用标准库）；PDF/Excel 后端为可选扩展。

安装

pip install duekit                  # 调度 + 分级，零第三方依赖
pip install "duekit[table]"         # 额外安装 pdfplumber + openpyxl

快速上手

import duekit

# --- 调度：每天 06:30 跑一次，并每 4 小时跑一次 ---
from datetime import timedelta
schedule = duekit.Schedule()
schedule.daily("06:30", task=my_morning_check)
schedule.interval(timedelta(hours=4), task=my_interval_check)
schedule.run_pending()              # 立刻执行当前已到期的任务

# --- 分级：按你自己定义的级别判断日期 ---
scheme = duekit.LevelScheme(levels=[
    duekit.Level.expired(),
    duekit.Level("urgent", 7),
    duekit.Level("normal", None),   # 兜底级别
])
match = scheme.classify("2026.10.02", today=today)
print(match.name)                   # -> "urgent"

# --- 表格：按 schema 抽取记录 ---
records = duekit.extract("清单.pdf", schema=my_schema)

# --- 清洗：规范化记录里的脏文本 ---
cleaner = duekit.Cleaner(id_keys=("emp_id",))
records = cleaner.clean_records(records)

# --- 续期：计算下次换证日期（如证书 4 年有效）---
rule = duekit.RenewRule(years=4)
next_expiry = rule.apply(current_expiry)

# --- 校验：一次性收集所有记录的问题 ---
validator = duekit.Validator(
    [duekit.FieldRule("emp_id", [])],
    unique_keys=("emp_id",),
)
issues = validator.validate(records)

# --- 统计 + 导出：汇总后写出按级别着色的 Excel ---
summary = duekit.summarize(records, expiry_key="expiry", scheme=scheme)["all"]
duekit.export_excel("报表.xlsx", records, level_key="level")

---

1. 调度 —— duekit.scheduler

两类条目可自由混用，每个条目绑定自己的任务函数：

from datetime import timedelta
import duekit

schedule = duekit.Schedule(timezone="Asia/Shanghai")

# 固定钟点（可选 days 限定星期：周一=0 … 周日=6）
schedule.daily("06:30", task=morning_job)
schedule.daily("22:30", task=night_job, days=(0, 1, 2, 3, 4))

# 固定间隔；可传分钟数或 timedelta；anchor 可指定对齐基准
schedule.interval(timedelta(hours=4), task=often_job)
schedule.interval(30, task=every_30_min)            # 每 30 分钟

纯逻辑 与 线程运行器

调度计算是纯函数、可注入时间，无需真正等待即可检查或驱动：

upcoming = schedule.next_run()              # 最近一次 PlannedRun（不认领）
due = schedule.pending()                   # 认领上次以来到期的运行
results = schedule.run_pending()           # 认领并执行任务
for r in results:
    print(r.ok, r.value, r.error)          # 任务异常被隔离，不影响其他

在长驻进程中，可启动可选的守护线程运行器：

runner = duekit.ThreadRunner(schedule, poll_interval=1.0)
runner.start()
...
runner.stop()

你也可以用 cron、APScheduler 或任何外部循环来调用 schedule.run_pending——库不会强制你使用线程。

确定性测试

注入固定时钟，调度测试无需真实等待：

s = duekit.Schedule(
    [duekit.DailyTime(at="06:30", task=job)],
    clock=lambda: fixed_datetime,
)

---

2. 到期分级 —— duekit.due

解析各种脏日期

duekit.parse_date("2026.3.21")          # datetime.date(2026, 3, 21)
duekit.parse_date("2026年3月21日")
duekit.parse_date("Wed Feb 11 2026 08:00:00 GMT+0800 (...)")  # JS toString
duekit.days_remaining("2026.10.06", today)   # 整数，过去为负

级别完全自定义

没有任何固定档位——级别数量、名称、阈值全部由你定义。当 days_remaining <= max_days 时命中该级；max_days=None 为兜底。 级别会自动排序并校验（阈值不可重复、最多一个兜底级）。

scheme = duekit.LevelScheme(levels=[
    duekit.Level.expired(),                 # 严格已过期
    duekit.Level("urgent", 7),
    duekit.Level("warning", 15),
    duekit.Level("normal", None),
])

只想要三档？声明三个即可。原五档行为可用 LevelScheme.default() 一键获得。

级别回调动作

为级别绑定动作，日期命中时触发：

def alert(match):
    print(match.name, match.days, match.target)

scheme = duekit.LevelScheme(levels=[
    duekit.Level("urgent", 7, action=alert),
    duekit.Level("normal", None),
])
match = scheme.classify(target, today, fire=True)   # 立即执行动作
# 也可：match = scheme.classify(...); match.fire()

多个日期

# 取最早到期日，返回一个结果
m = scheme.classify_many([date_a, date_b], today, how="earliest")

# 每个日期分别返回结果
matches = scheme.classify_many([date_a, date_b], today, how="each", fire=True)

无法解析的值会落入兜底级；若没有兜底级则报错——坏输入绝不会被静默当作 normal。

---

3. 表格抽取 —— duekit.table

Reader（均返回 RawRow）

duekit.read_pdf("清单.pdf")
duekit.read_excel("清单.xlsx", sheet=0)
duekit.read_csv("清单.csv")
duekit.read_any("清单.pdf")        # 按扩展名选择

声明式 schema

schema = duekit.TableSchema.create(
    [
        duekit.Field("seq", 0, "int", required=True),
        duekit.Field("name", 1, "str"),
        duekit.Field("expiry", 2, "date"),
    ],
    start_keywords=["序号", "姓名"],     # 此行之后开始处理
    data_when=duekit.first_column_int,   # 什么算数据行
    skip_when=[                          # 额外要跳过的行
        duekit.column_has(7, "焊接方法"),
    ],
    continuation=duekit.Continuation(
        detect=duekit.first_column_empty, # 被拆分的延续行
        merge=duekit.smart_merge,
    ),
)
records = duekit.extract("清单.pdf", schema)

字段类型：str / int / float / date。未通过 required 校验的记录会被丢弃。

合并跨行拆分的记录

真实清单常把一条逻辑记录拆成两个物理行。工作记录先保留原始字符串、由 延续行合并，整条完成后才做一次类型转换：

• 日期列无分隔直接拼接——2027.07.2 + 8 成为 2027.07.28；
• 其他列仅在拼接两个 ASCII 词时补空格（"GTAW P FW" + "FM1" → "GTAW P FW FM1"）；中文取值直接拼接。

可用 Continuation(columns=(2,)) 限定参与合并的列；需要自定义规则时 传入自己的 merge 可调用对象。

---

端到端示例

import duekit

schedule = duekit.Schedule()
schedule.daily("06:30", task=check_expiries)

scheme = duekit.LevelScheme(levels=[
    duekit.Level.expired(name="expired"),
    duekit.Level("urgent", 7),
    duekit.Level("normal", None),
])

def check_expiries():
    records = duekit.extract("清单.pdf", schema=table_schema)
    today = date.today()
    for r in records:
        scheme.classify(r["expiry"], today, fire=True)   # 紧急则触发动作

# 手动驱动，或：duekit.ThreadRunner(schedule).start()
schedule.run_pending()

注意与限制

• duekit 只做日期分级与表格抽取，不负责发送邮件或手机推送——请把 这些写进你的级别/任务回调。
• 调度默认读取本地时区；传入 clock/today 可得到确定性结果。
• PDF/Excel Reader 需要安装 [table] 扩展。

许可证

MIT © 兰衣