Implement automatic reminder engine with admin and company UI.

Add schema v6, read-only scan rules for unsubmitted flows, gaps and pending reviews, deduplicated delivery with append-only event history, daily scan thread, and full API/frontend integration with 18 new tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
2026-08-28 13:33:06 +00:00
co-authored by Cursor multica-agent
parent 951b353765
commit f17636183d
8 changed files with 1934 additions and 270 deletions
+700
View File
@@ -0,0 +1,700 @@
"""Read-only reminder rule engine and delivery helpers.
Scans existing tables to surface pending items; delivery writes append-only
``reminders`` / ``reminder_events`` rows without touching bank or business data.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
import json
import sqlite3
from typing import Any
import uuid
from bank_importer import auth
from bank_importer.db import utc_now
RULE_UNSUBMITTED = "unsubmitted"
RULE_GAP = "gap"
RULE_PENDING = "pending_review"
RULE_MANUAL = "manual"
SETTING_MONTHLY_START_DAY = "monthly_start_day"
SETTING_GAP_DAYS = "gap_days"
SETTING_SCAN_TIME = "scan_time"
DEFAULT_SETTINGS: dict[str, str] = {
SETTING_MONTHLY_START_DAY: "5",
SETTING_GAP_DAYS: "5",
SETTING_SCAN_TIME: "08:00",
}
RULE_LABELS = {
RULE_UNSUBMITTED: "流水未提交",
RULE_GAP: "流水断档",
RULE_PENDING: "待确认/待审核",
}
ACTION_LINKS = {
RULE_UNSUBMITTED: "upload",
RULE_GAP: "flows",
RULE_PENDING: "reconcile",
}
@dataclass(frozen=True)
class ScanFinding:
company_id: int
company_name: str
rule_key: str
dedupe_key: str
rule_params: dict[str, Any]
title: str
content: str
reason: str
days_open: int
deadline: str | None
action_link: str
existing_reminder_id: int | None
send_count: int
def _today() -> date:
return datetime.now(timezone.utc).date()
def get_settings(connection: sqlite3.Connection) -> dict[str, str]:
rows = connection.execute("SELECT key, value FROM reminder_settings").fetchall()
settings = dict(DEFAULT_SETTINGS)
for row in rows:
settings[row["key"]] = row["value"]
return settings
def update_settings(connection: sqlite3.Connection, updates: dict[str, str]) -> dict[str, str]:
allowed = set(DEFAULT_SETTINGS)
now = utc_now()
with connection:
for key, value in updates.items():
if key not in allowed:
raise ValueError(f"unknown setting: {key}")
connection.execute(
"""
INSERT INTO reminder_settings (key, value, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
""",
(key, str(value), now),
)
return get_settings(connection)
def _setting_int(settings: dict[str, str], key: str) -> int:
return int(settings.get(key, DEFAULT_SETTINGS[key]))
def _month_period(day: date | None = None) -> str:
ref = day or _today()
return f"{ref.year:04d}-{ref.month:02d}"
def _parse_date(value: str | None) -> date | None:
if not value:
return None
try:
return date.fromisoformat(value[:10])
except ValueError:
return None
def _days_between(start: date, end: date) -> int:
return max(0, (end - start).days)
def _duration_pill_class(days: int) -> str:
if days >= 7:
return "pill-danger"
if days >= 3:
return "pill-warn"
return "pill-muted"
def _existing_reminder(connection: sqlite3.Connection, dedupe_key: str) -> sqlite3.Row | None:
return connection.execute(
"""
SELECT id, send_count, status, last_sent_at
FROM reminders
WHERE dedupe_key = ? AND status != 'resolved'
""",
(dedupe_key,),
).fetchone()
def _company_has_month_batch(connection: sqlite3.Connection, company_id: int, period: str) -> bool:
year, month = period.split("-")
prefix = f"{year}-{month}"
row = connection.execute(
"""
SELECT 1
FROM import_batches b
JOIN sheet_batches s ON s.import_batch_id = b.id
WHERE b.company_id = ?
AND (
substr(s.period_start, 1, 7) = ?
OR substr(s.period_end, 1, 7) = ?
OR (s.period_start <= ? || '-31' AND s.period_end >= ? || '-01')
)
LIMIT 1
""",
(company_id, prefix, prefix, prefix, prefix),
).fetchone()
return row is not None
def _latest_batch_end(connection: sqlite3.Connection, company_id: int) -> date | None:
row = connection.execute(
"""
SELECT MAX(s.period_end) AS latest_end
FROM import_batches b
JOIN sheet_batches s ON s.import_batch_id = b.id
WHERE b.company_id = ?
""",
(company_id,),
).fetchone()
return _parse_date(row["latest_end"] if row else None)
def _pending_review_count(connection: sqlite3.Connection, company_id: int) -> int:
match_count = connection.execute(
"""
SELECT COUNT(DISTINCT e.id)
FROM canonical_transfer_events e
JOIN current_transfer_decisions c ON c.event_id = e.id
JOIN transfer_match_decisions d ON d.id = c.decision_id
JOIN transfer_decision_participants payer ON payer.decision_id = d.id AND payer.role = 'payer'
JOIN transfer_decision_participants payee ON payee.decision_id = d.id AND payee.role = 'payee'
WHERE e.lifecycle = 'active'
AND d.classification IN ('unresolved', 'needs_review')
AND (payer.company_id = ? OR payee.company_id = ?)
""",
(company_id, company_id),
).fetchone()[0]
sheet_count = connection.execute(
"""
SELECT COUNT(*)
FROM sheet_reviews r
JOIN import_batches b ON b.id = r.import_batch_id
WHERE b.company_id = ?
AND r.outcome = 'parsed'
AND r.review_status = 'pending'
""",
(company_id,),
).fetchone()[0]
account_count = connection.execute(
"""
SELECT COUNT(*) FROM bank_accounts
WHERE company_id = ? AND status = 'pending'
""",
(company_id,),
).fetchone()[0]
return int(match_count) + int(sheet_count) + int(account_count)
def _active_companies(connection: sqlite3.Connection) -> list[sqlite3.Row]:
return connection.execute(
"""
SELECT id, name FROM companies
WHERE COALESCE(status, 'active') != 'disabled'
ORDER BY name
"""
).fetchall()
def scan_findings(connection: sqlite3.Connection) -> list[ScanFinding]:
settings = get_settings(connection)
today = _today()
period = _month_period(today)
monthly_start = _setting_int(settings, SETTING_MONTHLY_START_DAY)
gap_days = _setting_int(settings, SETTING_GAP_DAYS)
findings: list[ScanFinding] = []
for company in _active_companies(connection):
company_id = int(company["id"])
company_name = company["name"]
if today.day >= monthly_start and not _company_has_month_batch(connection, company_id, period):
dedupe_key = f"{company_id}:{RULE_UNSUBMITTED}:{period}"
days_open = _days_between(date(today.year, today.month, monthly_start), today)
existing = _existing_reminder(connection, dedupe_key)
findings.append(
ScanFinding(
company_id=company_id,
company_name=company_name,
rule_key=RULE_UNSUBMITTED,
dedupe_key=dedupe_key,
rule_params={"period": period},
title=f"{company_name} · {today.month} 月流水未提交",
content=(
f"贵公司 {period} 银行流水尚未提交。"
f"请于截止日前完成全部账户流水上传。"
),
reason=f"应交 {today.month:02d}-{monthly_start:02d} · 已逾期 {days_open}",
days_open=days_open,
deadline=None,
action_link=ACTION_LINKS[RULE_UNSUBMITTED],
existing_reminder_id=int(existing["id"]) if existing else None,
send_count=int(existing["send_count"]) if existing else 0,
)
)
latest_end = _latest_batch_end(connection, company_id)
if latest_end is not None:
gap = _days_between(latest_end, today)
if gap > gap_days:
dedupe_key = f"{company_id}:{RULE_GAP}:{latest_end.isoformat()}"
existing = _existing_reminder(connection, dedupe_key)
findings.append(
ScanFinding(
company_id=company_id,
company_name=company_name,
rule_key=RULE_GAP,
dedupe_key=dedupe_key,
rule_params={"latest_end": latest_end.isoformat(), "gap_days": gap},
title=f"{company_name} · 流水断档 {gap}",
content=(
f"最近流水截止日为 {latest_end.isoformat()}"
f"已连续 {gap} 天无新数据,请补传断档期间银行流水。"
),
reason=f"最近截止 {latest_end.strftime('%m-%d')} · 断档 {gap}",
days_open=gap,
deadline=None,
action_link=ACTION_LINKS[RULE_GAP],
existing_reminder_id=int(existing["id"]) if existing else None,
send_count=int(existing["send_count"]) if existing else 0,
)
)
pending_count = _pending_review_count(connection, company_id)
if pending_count > 0:
dedupe_key = f"{company_id}:{RULE_PENDING}:active"
existing = _existing_reminder(connection, dedupe_key)
first_seen = _parse_date(existing["last_sent_at"][:10] if existing and existing["last_sent_at"] else None)
days_open = _days_between(first_seen, today) if first_seen else 0
findings.append(
ScanFinding(
company_id=company_id,
company_name=company_name,
rule_key=RULE_PENDING,
dedupe_key=dedupe_key,
rule_params={"pending_count": pending_count},
title=f"{company_name} · {pending_count} 项待确认/待审核",
content=(
f"贵公司当前有 {pending_count} 项往来确认、流水审核或账户登记待处理,"
f"请尽快完成确认以免影响结账。"
),
reason=f"待处理 {pending_count}",
days_open=days_open,
deadline=None,
action_link=ACTION_LINKS[RULE_PENDING],
existing_reminder_id=int(existing["id"]) if existing else None,
send_count=int(existing["send_count"]) if existing else 0,
)
)
findings.sort(key=lambda item: item.days_open, reverse=True)
return findings
def run_scan(
connection: sqlite3.Connection,
*,
actor: sqlite3.Row | None = None,
ip: str | None = None,
) -> dict[str, Any]:
findings = scan_findings(connection)
counts = {
RULE_UNSUBMITTED: sum(1 for f in findings if f.rule_key == RULE_UNSUBMITTED),
RULE_GAP: sum(1 for f in findings if f.rule_key == RULE_GAP),
RULE_PENDING: sum(1 for f in findings if f.rule_key == RULE_PENDING),
}
auth.audit(
connection,
"reminder_scan",
actor=actor,
detail=json.dumps({"total": len(findings), "by_rule": counts}, ensure_ascii=False),
ip=ip,
)
return {"findings": findings, "counts": counts}
def finding_to_dict(finding: ScanFinding) -> dict[str, Any]:
return {
"company_id": finding.company_id,
"company_name": finding.company_name,
"rule_key": finding.rule_key,
"rule_label": RULE_LABELS.get(finding.rule_key, finding.rule_key),
"dedupe_key": finding.dedupe_key,
"rule_params": finding.rule_params,
"title": finding.title,
"content": finding.content,
"reason": finding.reason,
"days_open": finding.days_open,
"duration_pill": _duration_pill_class(finding.days_open),
"deadline": finding.deadline,
"action_link": finding.action_link,
"existing_reminder_id": finding.existing_reminder_id,
"send_count": finding.send_count,
}
def _append_event(
connection: sqlite3.Connection,
reminder_id: int,
event_type: str,
actor: str,
detail: str | None = None,
) -> None:
connection.execute(
"""
INSERT INTO reminder_events (reminder_id, event_type, actor, detail, created_at)
VALUES (?, ?, ?, ?, ?)
""",
(reminder_id, event_type, actor, detail, utc_now()),
)
def _actor_label(actor: sqlite3.Row | None) -> str:
if actor is None:
return "system"
return str(actor["id"])
def deliver_finding(
connection: sqlite3.Connection,
dedupe_key: str,
*,
actor: sqlite3.Row | None = None,
ip: str | None = None,
) -> int | None:
findings = {item.dedupe_key: item for item in scan_findings(connection)}
finding = findings.get(dedupe_key)
if finding is None:
return None
return _deliver(connection, finding, actor=actor, ip=ip)
def _deliver(
connection: sqlite3.Connection,
finding: ScanFinding,
*,
actor: sqlite3.Row | None = None,
ip: str | None = None,
) -> int:
now = utc_now()
actor_ref = _actor_label(actor)
existing = connection.execute(
"SELECT id, send_count, status FROM reminders WHERE dedupe_key = ?",
(finding.dedupe_key,),
).fetchone()
with connection:
if existing is not None and existing["status"] != "resolved":
reminder_id = int(existing["id"])
send_count = int(existing["send_count"]) + 1
connection.execute(
"""
UPDATE reminders
SET send_count = ?, last_sent_at = ?, status = 'open'
WHERE id = ?
""",
(send_count, now, reminder_id),
)
_append_event(
connection,
reminder_id,
"escalated" if send_count > 1 else "sent",
actor_ref,
f"{send_count} 次催办",
)
else:
connection.execute(
"""
INSERT INTO reminders (
company_id, rule_key, dedupe_key, rule_params, title, content,
deadline, source, status, send_count, first_sent_at, last_sent_at,
created_by, created_at, action_link
) VALUES (?, ?, ?, ?, ?, ?, ?, 'auto', 'open', 1, ?, ?, ?, ?, ?)
""",
(
finding.company_id,
finding.rule_key,
finding.dedupe_key,
json.dumps(finding.rule_params, ensure_ascii=False),
finding.title,
finding.content,
finding.deadline,
now,
now,
actor_ref,
now,
finding.action_link,
),
)
reminder_id = int(connection.execute("SELECT last_insert_rowid()").fetchone()[0])
_append_event(connection, reminder_id, "sent", actor_ref, finding.reason)
auth.audit(
connection,
"reminder_send",
actor=actor,
target=f"reminder:{reminder_id}",
detail=finding.dedupe_key,
ip=ip,
)
return reminder_id
def deliver_many(
connection: sqlite3.Connection,
dedupe_keys: list[str],
*,
actor: sqlite3.Row | None = None,
ip: str | None = None,
) -> list[int]:
sent: list[int] = []
for key in dedupe_keys:
reminder_id = deliver_finding(connection, key, actor=actor, ip=ip)
if reminder_id is not None:
sent.append(reminder_id)
return sent
def send_manual(
connection: sqlite3.Connection,
*,
company_id: int,
display_type: str,
content: str,
deadline: str | None,
actor: sqlite3.Row,
ip: str | None = None,
) -> int:
company = connection.execute(
"SELECT name FROM companies WHERE id = ?", (company_id,)
).fetchone()
if company is None:
raise ValueError("company not found")
now = utc_now()
dedupe_key = f"{company_id}:{RULE_MANUAL}:{uuid.uuid4().hex}"
title = f"{company['name']} · {display_type}"
with connection:
connection.execute(
"""
INSERT INTO reminders (
company_id, rule_key, dedupe_key, rule_params, title, content,
deadline, source, status, send_count, first_sent_at, last_sent_at,
created_by, created_at, action_link
) VALUES (?, ?, ?, ?, ?, ?, ?, 'manual', 'open', 1, ?, ?, ?, ?, NULL)
""",
(
company_id,
RULE_MANUAL,
dedupe_key,
json.dumps({"display_type": display_type}, ensure_ascii=False),
title,
content,
deadline,
now,
now,
str(actor["id"]),
now,
),
)
reminder_id = int(connection.execute("SELECT last_insert_rowid()").fetchone()[0])
_append_event(connection, reminder_id, "sent", str(actor["id"]), display_type)
auth.audit(
connection,
"reminder_send_manual",
actor=actor,
target=f"reminder:{reminder_id}",
detail=display_type,
ip=ip,
)
return reminder_id
def resend_reminder(
connection: sqlite3.Connection,
reminder_id: int,
*,
actor: sqlite3.Row,
ip: str | None = None,
) -> None:
row = connection.execute(
"SELECT id, send_count, status FROM reminders WHERE id = ?", (reminder_id,)
).fetchone()
if row is None or row["status"] == "resolved":
raise ValueError("reminder not available")
now = utc_now()
send_count = int(row["send_count"]) + 1
with connection:
connection.execute(
"""
UPDATE reminders
SET send_count = ?, last_sent_at = ?, status = 'open'
WHERE id = ?
""",
(send_count, now, reminder_id),
)
_append_event(
connection,
reminder_id,
"escalated",
str(actor["id"]),
f"{send_count} 次催办",
)
auth.audit(
connection,
"reminder_resend",
actor=actor,
target=f"reminder:{reminder_id}",
ip=ip,
)
def _reminder_row_to_dict(row: sqlite3.Row, *, company_name: str | None = None) -> dict[str, Any]:
params = json.loads(row["rule_params"]) if row["rule_params"] else {}
rule_key = row["rule_key"]
display_type = params.get("display_type") if rule_key == RULE_MANUAL else RULE_LABELS.get(rule_key, rule_key)
status = row["status"]
status_ui = {"open": "unread", "acknowledged": "doing", "resolved": "done"}[status]
return {
"id": row["id"],
"company_id": row["company_id"],
"company_name": company_name or row["company_name"],
"rule_key": rule_key,
"display_type": display_type,
"title": row["title"],
"content": row["content"],
"deadline": row["deadline"],
"source": row["source"],
"status": status,
"status_ui": status_ui,
"send_count": row["send_count"],
"first_sent_at": row["first_sent_at"],
"last_sent_at": row["last_sent_at"],
"created_by": row["created_by"],
"created_at": row["created_at"],
"action_link": row["action_link"],
"rule_params": params,
}
def list_admin_reminders(
connection: sqlite3.Connection,
*,
source: str | None = None,
) -> list[dict[str, Any]]:
conditions = ["r.send_count > 0"]
params: list[Any] = []
if source in {"auto", "manual"}:
conditions.append("r.source = ?")
params.append(source)
where = " AND ".join(conditions)
rows = connection.execute(
f"""
SELECT r.*, c.name AS company_name
FROM reminders r
JOIN companies c ON c.id = r.company_id
WHERE {where}
ORDER BY r.last_sent_at DESC, r.id DESC
""",
params,
).fetchall()
return [_reminder_row_to_dict(row) for row in rows]
def list_company_reminders(connection: sqlite3.Connection, company_id: int) -> list[dict[str, Any]]:
rows = connection.execute(
"""
SELECT r.*, c.name AS company_name
FROM reminders r
JOIN companies c ON c.id = r.company_id
WHERE r.company_id = ? AND r.send_count > 0
ORDER BY
CASE r.status WHEN 'open' THEN 0 WHEN 'acknowledged' THEN 1 ELSE 2 END,
r.last_sent_at DESC
""",
(company_id,),
).fetchall()
return [_reminder_row_to_dict(row) for row in rows]
def get_reminder_detail(connection: sqlite3.Connection, reminder_id: int) -> dict[str, Any] | None:
row = connection.execute(
"""
SELECT r.*, c.name AS company_name
FROM reminders r
JOIN companies c ON c.id = r.company_id
WHERE r.id = ?
""",
(reminder_id,),
).fetchone()
if row is None:
return None
events = connection.execute(
"""
SELECT event_type, actor, detail, created_at
FROM reminder_events
WHERE reminder_id = ?
ORDER BY id
""",
(reminder_id,),
).fetchall()
payload = _reminder_row_to_dict(row)
payload["events"] = [dict(item) for item in events]
return payload
def update_reminder_status(
connection: sqlite3.Connection,
reminder_id: int,
new_status: str,
*,
company_id: int | None = None,
actor: sqlite3.Row | None = None,
) -> bool:
if new_status not in {"acknowledged", "resolved"}:
raise ValueError("invalid status")
row = connection.execute(
"SELECT id, company_id, status FROM reminders WHERE id = ?",
(reminder_id,),
).fetchone()
if row is None:
return False
if company_id is not None and int(row["company_id"]) != company_id:
return False
if row["status"] == "resolved":
return False
event_type = "acknowledged" if new_status == "acknowledged" else "resolved"
actor_ref = _actor_label(actor)
with connection:
connection.execute(
"UPDATE reminders SET status = ? WHERE id = ?",
(new_status, reminder_id),
)
_append_event(connection, reminder_id, event_type, actor_ref, None)
return True
def company_unread_count(connection: sqlite3.Connection, company_id: int) -> int:
row = connection.execute(
"""
SELECT COUNT(*) AS n FROM reminders
WHERE company_id = ? AND status = 'open' AND send_count > 0
""",
(company_id,),
).fetchone()
return int(row["n"])