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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
|
"""Local review page for marketer drafts, for every project in projects.toml.
Serves each project's drafts (DATA/<slug>/posts/, one JSON file per batch,
YYYY-MM-DD.json) at http://127.0.0.1:8787/?project=<slug>. Joel comments and
marks drafts posted or skipped from the page; every change is written back
into the batch file, and each action Joel takes is printed as one EVENT line
on stdout so a Claude session can watch the log and respond live.
Stdlib only. Usage: python3 server.py [--config DIR] [--data DIR] [--port N]
"""
import argparse
import html
import json
import os
import re
import sys
import threading
import time
import tomllib
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse, urlunparse
HERE = Path(__file__).resolve().parent
LOCK = threading.Lock()
STATIC = {}
for font in (HERE / "fonts").glob("*.woff2"):
STATIC[f"/fonts/{font.name}"] = (f"fonts/{font.name}", "font/woff2")
CONFIG_DIR = None
DATA_DIR = None
def projects():
"""projects.toml, read on every call so a new project shows up without a restart."""
with (CONFIG_DIR / "projects.toml").open("rb") as f:
return tomllib.load(f)
def project(slug):
"""The named project, or the only one when no name is given. None if unknown."""
all_ = projects()
if not slug and len(all_) == 1:
slug = next(iter(all_))
if not slug or "/" in slug or slug not in all_:
return None
return {**all_[slug], "slug": slug}
def posts_dir(proj):
d = DATA_DIR / proj["slug"] / "posts"
d.mkdir(parents=True, exist_ok=True)
return d
def card_hosts(proj):
host = urlparse(proj.get("site", "")).hostname or ""
host = host.removeprefix("www.")
return {host, "www." + host} if host else set()
CARDS = {}
META_RE = re.compile(r'<meta\s+property="og:(title|description|image)"\s+content="([^"]*)"', re.I)
def link_card(proj, target):
u = urlparse(target)
if u.scheme != "https" or u.hostname not in card_hosts(proj):
return {"error": f"previews only fetch {proj.get('site', 'the project site')} links"}
clean = urlunparse(u._replace(query="", fragment=""))
if clean not in CARDS:
try:
req = urllib.request.Request(clean, headers={"User-Agent": "marketer-post-queue"})
with urllib.request.urlopen(req, timeout=5) as r:
page = r.read(200_000).decode("utf-8", "replace")
CARDS[clean] = {k.lower(): html.unescape(v) for k, v in META_RE.findall(page)}
CARDS[clean]["domain"] = u.hostname
except OSError as e:
return {"error": str(e)}
return CARDS[clean]
IMAGES = {}
def card_image(proj, target):
"""Fetch a card image from the project's site once and serve it from here,
so the preview doesn't depend on the browser reaching the CDN directly."""
u = urlparse(target)
if u.scheme != "https" or u.hostname not in card_hosts(proj):
return None
if target not in IMAGES:
try:
req = urllib.request.Request(target, headers={"User-Agent": "marketer-post-queue"})
with urllib.request.urlopen(req, timeout=5) as r:
IMAGES[target] = (r.headers.get_content_type(), r.read(5_000_000))
except OSError:
return None
return IMAGES[target]
def batch_files(posts_dir):
return sorted(posts_dir.glob("*.json"), reverse=True)
def load(path):
with path.open() as f:
return json.load(f)
def save(path, data):
tmp = path.with_suffix(".json.tmp")
with tmp.open("w") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
f.write("\n")
os.replace(tmp, path)
def event(kind, slug, batch, draft_id, text=""):
line = " ".join(text.split())
print(f"EVENT {kind} project={slug} batch={batch} draft={draft_id} {line}".rstrip(), flush=True)
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def send_json(self, status, payload):
body = json.dumps(payload).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def send_file(self, path, ctype, cache="max-age=86400"):
body = path.read_bytes()
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Cache-Control", cache)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
url = urlparse(self.path)
query = parse_qs(url.query)
if url.path in ("/", "/index.html"):
body = (HERE / "index.html").read_bytes()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
if url.path == "/favicon.ico":
self.send_response(204)
self.end_headers()
return
static = STATIC.get(url.path)
if static:
name, ctype = static
self.send_file(HERE / name, ctype)
return
if url.path == "/avatar.jpg":
avatar = CONFIG_DIR / "avatar.jpg"
if avatar.is_file():
self.send_file(avatar, "image/jpeg", "max-age=3600")
else:
self.send_json(404, {"error": f"no {avatar}"})
return
if url.path == "/api/me":
try:
self.send_json(200, load(CONFIG_DIR / "me.json"))
except (OSError, json.JSONDecodeError) as e:
self.send_json(200, {"error": str(e)})
return
if url.path == "/api/projects":
try:
rows = [{"slug": slug, "name": p.get("name", slug), "headline": p.get("headline", ""),
"site": p.get("site", ""), "mark": bool(p.get("mark"))}
for slug, p in projects().items()]
except (OSError, tomllib.TOMLDecodeError) as e:
self.send_json(500, {"error": f"projects.toml: {e}"})
return
self.send_json(200, {"projects": rows})
return
proj = project(query.get("project", [""])[0])
if proj is None:
self.send_json(404, {"error": "unknown project; pass ?project=<slug> from projects.toml"})
return
if url.path == "/mark.svg":
mark = Path(proj["repo"]).expanduser() / proj.get("mark", "")
if proj.get("mark") and mark.is_file():
self.send_file(mark, "image/svg+xml", "max-age=3600")
else:
self.send_json(404, {"error": "no mark"})
return
if url.path == "/api/og":
self.send_json(200, link_card(proj, query.get("url", [""])[0]))
return
if url.path == "/api/ogimg":
img = card_image(proj, query.get("url", [""])[0])
if img is None:
self.send_json(404, {"error": "no image"})
return
self.send_response(200)
self.send_header("Content-Type", img[0])
self.send_header("Cache-Control", "max-age=3600")
self.send_header("Content-Length", str(len(img[1])))
self.end_headers()
self.wfile.write(img[1])
return
if url.path == "/api/drafts":
with LOCK:
batches = []
for p in batch_files(posts_dir(proj)):
try:
data = load(p)
except (OSError, json.JSONDecodeError) as e:
data = {"error": str(e), "drafts": []}
data["batch"] = p.stem
batches.append(data)
self.send_json(200, {"project": proj["slug"], "batches": batches})
return
self.send_json(404, {"error": "not found"})
def do_POST(self):
try:
length = int(self.headers.get("Content-Length", 0))
req = json.loads(self.rfile.read(length) or b"{}")
batch, draft_id = req["batch"], req["id"]
except (ValueError, KeyError):
self.send_json(400, {"error": "need batch and id"})
return
proj = project(str(req.get("project", "")))
if proj is None:
self.send_json(404, {"error": "unknown project"})
return
slug = proj["slug"]
path = posts_dir(proj) / f"{batch}.json"
if not path.is_file() or "/" in batch:
self.send_json(404, {"error": "no such batch"})
return
with LOCK:
data = load(path)
draft = next((d for d in data.get("drafts", []) if d.get("id") == draft_id), None)
if draft is None:
self.send_json(404, {"error": "no such draft"})
return
who = req.get("who", "joel")
now = time.strftime("%Y-%m-%d %H:%M")
if self.path == "/api/comment":
text = str(req.get("text", "")).strip()
if not text:
self.send_json(400, {"error": "empty comment"})
return
draft.setdefault("comments", []).append({"who": who, "text": text, "at": now})
draft["needs_claude"] = who == "joel"
save(path, data)
if who == "joel":
event("comment", slug, batch, draft_id, text)
elif self.path == "/api/edit":
text = str(req.get("text", "")).rstrip()
if not text.strip():
self.send_json(400, {"error": "empty text"})
return
if text == draft.get("text"):
self.send_json(200, {"ok": True})
return
draft.setdefault("claude_text", draft.get("text", ""))
draft["text"] = text
draft["edited_at"] = now
if who == "joel":
draft["needs_claude"] = True
save(path, data)
if who == "joel":
event("edit", slug, batch, draft_id, text)
elif self.path == "/api/status":
status = req.get("status")
if status not in ("draft", "posted", "skipped"):
self.send_json(400, {"error": "status must be draft, posted or skipped"})
return
draft["status"] = status
draft["status_at"] = now
note = str(req.get("reason") or req.get("url") or "").strip()
if status == "skipped":
draft["skip_reason"] = note
if note and who == "joel":
draft["needs_claude"] = True
if status == "posted" and "url" in req:
draft["post_url"] = note
save(path, data)
if who == "joel":
event(status, slug, batch, draft_id, note)
else:
self.send_json(404, {"error": "not found"})
return
self.send_json(200, {"ok": True})
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="~/.config/marketer", help="holds projects.toml, me.json, avatar.jpg")
ap.add_argument("--data", default="~/Documents/marketing", help="drafts live in DATA/<slug>/posts/")
ap.add_argument("--port", type=int, default=8787)
args = ap.parse_args()
global CONFIG_DIR, DATA_DIR
CONFIG_DIR = Path(args.config).expanduser()
DATA_DIR = Path(args.data).expanduser()
names = ", ".join(projects())
srv = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
print(f"READY http://127.0.0.1:{args.port} data={DATA_DIR} projects={names}", flush=True)
try:
srv.serve_forever()
except KeyboardInterrupt:
sys.exit(0)
if __name__ == "__main__":
main()
|