1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
#!/usr/bin/env python3
#
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""notmuch post-new hook: auto-tag incoming mail from the shared rule store.
Runs after every `notmuch new`. Reads ~/.config/mailrules/rules.json and
applies each enabled rule scoped to `tag:new`, in stage order, then consumes
the `tag:new` marker.
Rules may add any tag and remove most, but this hook REFUSES to remove `unread`
or `inbox` unattended and skips any rule that asks: see PROTECTED_REMOVALS
below for why, and for the conditions under which that restriction should be
lifted. It is expected to be relaxed once there is a story for confirming such
a rule before it runs.
Requires `new` in [new] tags= in ~/.notmuch-config. Without it every scoped
query matches nothing and this silently no-ops.
Install: copy to <database.path>/.notmuch/hooks/post-new, with mailrules.py
importable (same directory, or on PYTHONPATH).
"""
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import mailrules
SCOPE = "tag:new"
# Tags this hook refuses to REMOVE, whatever a rule says.
#
# maildir.synchronize_flags is true, so `unread` is not just an index entry:
# removing it rewrites Maildir filenames and propagates to the server on the
# next mbsync. `inbox` is what keeps mail visible at all. Unattended, on every
# sync, either one silently reorganizes a mailbox in a way that is tedious to
# undo and reaches other clients before anyone notices.
#
# Adding these tags is untouched, and so is removing anything else: a rule may
# still strip `promo` or any tag of its own making.
#
# DELIBERATELY CONSERVATIVE, AND EXPECTED TO BE RELAXED. The rules in use today
# only add tags, so this forbids nothing anyone is doing. It exists because the
# hook runs unattended and a mistake here is expensive, not because removing
# `unread` is wrong in principle: an "archive anything in notify/* older than
# 90 days" rule is a reasonable thing to want and would need this list revised.
# When that day comes, the question to answer first is what confirms the rule
# before it runs, not whether the guard is annoying.
PROTECTED_REMOVALS = frozenset({"unread", "inbox"})
def log(message):
print(f"post-new: {message}", file=sys.stderr)
def protected_removals(rule):
"""The protected tags this rule would remove, if any."""
return sorted(PROTECTED_REMOVALS.intersection(rule.remove))
def run_tag(arguments, query):
result = subprocess.run(["notmuch", "tag"] + arguments + ["--", query],
capture_output=True, text=True)
if result.returncode != 0:
log(f"notmuch tag failed: {result.stderr.strip()}")
return False
return True
def main():
store = mailrules.load()
# A file that will not load must NOT reach the consumer below. If the
# marker were cleared while the rules did not run, that mail could never
# be tagged by these rules again: the failure is silent, permanent, and
# invisible until someone notices a gap months later. Leaving tag:new in
# place makes the next successful run catch up instead.
if store.failed:
for warning in store.warnings:
log(warning)
log("rules did not load; leaving tag:new in place")
return 1
if store.missing:
log("no rules file; nothing to do")
return 0
# A dropped rule is not fatal, but it must be visible: this goes to the
# sync log, which is where someone looks when a tag stops appearing.
for warning in store.warnings:
log(warning)
applied = 0
for rule in mailrules.ordered(store.rules):
# Skip the rule, do not abort the run. A single over-reaching rule
# must not cost the tagging every other rule would have done, and
# aborting here would also leave tag:new set forever: the rule would
# be refused again on every subsequent sync and the marker would never
# be consumed.
refused = protected_removals(rule)
if refused:
log(f"rule '{rule.id}' would remove {', '.join(refused)}; "
f"skipped, this hook does not remove those unattended")
continue
query = mailrules.scoped_query(rule, SCOPE)
if not run_tag(mailrules.tag_arguments(rule), query):
log(f"rule '{rule.id}' failed; leaving tag:new in place")
return 1
applied += 1
# Only after every rule succeeded. A failure part way through leaves the
# marker set, so re-running the hook is safe and finishes the work.
if not run_tag(["-new"], SCOPE):
return 1
log(f"applied {applied} rule(s)")
return 0
if __name__ == "__main__":
sys.exit(main())
|
