The OC Guy
1c1492cfc2
Oliver asked for our self-hosted Plausible snippet on oc-guy.com. The template edit matters most: the daily sync only writes pages for new posts, so a page-level-only change would be lost on the next generated post. Both existing blog pages were updated directly to match the template exactly, so the next sync produces no head diff. The snippet is async and renders nothing: no visual change, no touch to CSS, layout or copy.
429 lines
16 KiB
Python
429 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Pull the day's posts from the n8n webhook and publish them into the site.
|
|
|
|
Runs once a day from cron. Everything it writes is derived from the webhook
|
|
payload: post pages under blog/<slug>.html, images under assets/blog/, and the
|
|
manifest at data/posts.json which the homepage reads.
|
|
|
|
Rules (from MANIFESTO.md / STYLEGUIDE.md):
|
|
- only real posts are published; attachments that are not images are ignored
|
|
- every published item keeps its date and category, no undated entries
|
|
- duplicates (same id, same link, same content hash) are never published twice
|
|
|
|
Usage:
|
|
sync_posts.py fetch, publish, update manifest
|
|
sync_posts.py --dry-run fetch and report, write nothing
|
|
sync_posts.py --fixture f.json read a saved payload instead of the webhook
|
|
sync_posts.py --check report manifest state and exit
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.request
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
MANIFEST = ROOT / "data" / "posts.json"
|
|
POSTS_DIR = ROOT / "blog"
|
|
ASSETS_DIR = ROOT / "assets" / "blog"
|
|
LOCK = ROOT / "data" / ".sync.lock"
|
|
DEFAULT_URL = "https://n8n.odoo4projects.com/webhook/oc-guy/posts"
|
|
UA = "oc-guy-sync/1.0 (+https://oc-guy.com)"
|
|
MAX_IMAGE_BYTES = 8 * 1024 * 1024
|
|
|
|
IMAGE_KEYS = ("image", "images", "cover", "banner", "picture", "photo", "featured_image", "file")
|
|
FILE_KEYS = ("file", "files", "attachment", "attachments", "document", "documents", "file_url")
|
|
TITLE_KEYS = ("header", "headline", "title", "name", "subject")
|
|
BODY_KEYS = ("text", "body", "content", "html", "markdown", "description", "story")
|
|
CATEGORY_KEYS = ("category", "area", "tags", "tag", "section", "type")
|
|
DATE_KEYS = ("scheduled", "date", "published", "pubDate", "created", "timestamp")
|
|
ID_KEYS = ("id", "slug", "uuid", "_id")
|
|
|
|
KNOWN_CATEGORIES = {"tips": "tips", "tip": "tips", "tricks": "tips",
|
|
"news": "news", "community": "news", "oca": "news",
|
|
"blog": "blog", "article": "blog", "post": "blog"}
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%SZ')}] {msg}", flush=True)
|
|
|
|
|
|
def fetch(url: str, timeout: int = 30) -> object:
|
|
"""GET the webhook, fall back to POST — n8n registers one method, not both."""
|
|
last = None
|
|
for method in ("GET", "POST"):
|
|
req = urllib.request.Request(url, method=method, headers={
|
|
"User-Agent": UA, "Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
}, data=b"{}" if method == "POST" else None)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return json.loads(r.read().decode("utf-8", "replace"))
|
|
except urllib.error.HTTPError as e:
|
|
last = f"{method} {url} -> HTTP {e.code} {e.reason}"
|
|
except Exception as e: # noqa: BLE001 - network, DNS, JSON
|
|
last = f"{method} {url} -> {type(e).__name__}: {e}"
|
|
raise RuntimeError(last or "webhook unreachable")
|
|
|
|
|
|
def unwrap(payload: object) -> list[dict]:
|
|
"""n8n hands back a list, or {data:[...]}, or {posts:[...]} — accept all three."""
|
|
if isinstance(payload, list):
|
|
return [p for p in payload if isinstance(p, dict)]
|
|
if isinstance(payload, dict):
|
|
for key in ("posts", "data", "items", "records", "results"):
|
|
v = payload.get(key)
|
|
if isinstance(v, list):
|
|
return [p for p in v if isinstance(p, dict)]
|
|
if any(k in payload for k in TITLE_KEYS):
|
|
return [payload]
|
|
return []
|
|
|
|
|
|
def pick(item: dict, keys: tuple[str, ...]) -> object:
|
|
for k in keys:
|
|
if k in item and item[k] not in (None, "", []):
|
|
return item[k]
|
|
return None
|
|
|
|
|
|
def slugify(text: str, fallback: str) -> str:
|
|
s = re.sub(r"[^a-z0-9]+", "-", str(text).lower()).strip("-")
|
|
return s[:80] or fallback
|
|
|
|
|
|
def normalise_date(value: object) -> str:
|
|
if isinstance(value, (int, float)):
|
|
return datetime.fromtimestamp(value, timezone.utc).date().isoformat()
|
|
if isinstance(value, str):
|
|
m = re.match(r"(\d{4}-\d{2}-\d{2})", value)
|
|
if m:
|
|
return m.group(1)
|
|
for fmt in ("%d.%m.%Y", "%d/%m/%Y", "%Y/%m/%d"):
|
|
try:
|
|
return datetime.strptime(value, fmt).date().isoformat()
|
|
except ValueError:
|
|
pass
|
|
return date.today().isoformat()
|
|
|
|
|
|
def normalise_category(value: object) -> str:
|
|
"""Map whatever tag comes in onto the three site sections."""
|
|
raw: list[str] = []
|
|
if isinstance(value, str):
|
|
raw = [v.strip().lower() for v in re.split(r"[,\s]+", value) if v.strip()]
|
|
elif isinstance(value, list):
|
|
raw = [str(v).strip().lower() for v in value if str(v).strip()]
|
|
for tag in raw:
|
|
if tag in KNOWN_CATEGORIES:
|
|
return KNOWN_CATEGORIES[tag]
|
|
return "blog"
|
|
|
|
|
|
def image_urls(value: object) -> list[str]:
|
|
out: list[str] = []
|
|
if isinstance(value, str):
|
|
if re.search(r"\.(jpe?g|png|webp|avif|gif)(\?|$)", value, re.I) or value.startswith("data:image/"):
|
|
out.append(value)
|
|
elif isinstance(value, list):
|
|
for v in value:
|
|
out.extend(image_urls(v))
|
|
elif isinstance(value, dict):
|
|
for key in ("url", "src", "path", "link"):
|
|
if isinstance(value.get(key), str):
|
|
out.extend(image_urls(value[key]))
|
|
return out
|
|
|
|
|
|
def strip_tags(raw: str) -> str:
|
|
text = re.sub(r"(?is)<(script|style).*?</\1>", " ", raw)
|
|
text = re.sub(r"(?s)<[^>]+>", " ", text)
|
|
return re.sub(r"\s+", " ", html.unescape(text)).strip()
|
|
|
|
|
|
def safe_html(raw: str) -> str:
|
|
"""Keep the post's formatting, drop anything executable or remote-loading."""
|
|
out = re.sub(r"(?is)<(script|style|iframe|object|embed|form)[^>]*>.*?</\1>", "", raw)
|
|
out = re.sub(r"(?i)\son\w+\s*=\s*(\"[^\"]*\"|'[^']*'|[^\s>]+)", "", out)
|
|
out = re.sub(r"(?i)(href|src)\s*=\s*(\"|')javascript:[^\"']*(\"|')", r'\1="#"', out)
|
|
return out.strip()
|
|
|
|
|
|
def download_image(src: str, dest_stem: Path) -> str | None:
|
|
"""Save one image next to the post. Anything that is not an image is dropped."""
|
|
try:
|
|
if src.startswith("data:image/"):
|
|
header, _, b64 = src.partition(",")
|
|
import base64
|
|
blob = base64.b64decode(b64)
|
|
ext = (re.match(r"data:image/(\w+)", header) or [None, "png"])[1]
|
|
ext = "jpg" if ext == "jpeg" else ext
|
|
data, ctype = blob, f"image/{ext}"
|
|
else:
|
|
req = urllib.request.Request(src, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=45) as r:
|
|
ctype = (r.headers.get("Content-Type") or "").split(";")[0].strip().lower()
|
|
data = r.read(MAX_IMAGE_BYTES + 1)
|
|
except Exception as e: # noqa: BLE001
|
|
log(f" image failed ({type(e).__name__}): {src[:90]}")
|
|
return None
|
|
|
|
if not ctype.startswith("image/"):
|
|
log(f" skipped non-image file: {src[:90]} ({ctype or 'unknown type'})")
|
|
return None
|
|
if len(data) > MAX_IMAGE_BYTES:
|
|
log(f" skipped oversized image: {src[:90]}")
|
|
return None
|
|
if len(data) < 64:
|
|
log(f" skipped empty image: {src[:90]}")
|
|
return None
|
|
|
|
ext = {"image/jpeg": "jpg", "image/png": "png", "image/webp": "webp",
|
|
"image/avif": "avif", "image/gif": "gif"}.get(ctype, "img")
|
|
dest = dest_stem.with_suffix("." + ext)
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_bytes(data)
|
|
return str(dest.relative_to(ROOT))
|
|
|
|
|
|
POST_PAGE = """<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
|
<title>{title} — The OC Guy</title>
|
|
<meta name="description" content="{description}">
|
|
<meta name="color-scheme" content="light">
|
|
<meta name="theme-color" content="#F2E9D6">
|
|
<link rel="preconnect" href="https://fonts.googleapis.com">
|
|
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
|
|
<link href="https://fonts.googleapis.com/css2?family=Fraunces:ital,opsz,wght@0,9..144,300..700;1,9..144,300..600&family=Inter:wght@400;500;600&display=swap" rel="stylesheet">
|
|
<!-- Privacy-friendly analytics by Plausible -->
|
|
<script async src="https://plausible.office.odoo4projects.com/js/pa-rN8268mNMAUdwSbE2e1wZ.js"></script>
|
|
<script>
|
|
window.plausible=window.plausible||function(){{(plausible.q=plausible.q||[]).push(arguments)}},plausible.init=plausible.init||function(i){{plausible.o=i||{{}}}};
|
|
plausible.init()
|
|
</script>
|
|
<link rel="stylesheet" href="../css/site.css">
|
|
</head>
|
|
<body>
|
|
<header class="masthead">
|
|
<a class="wordmark" href="../index.html">The OC Guy</a>
|
|
<nav class="nav" aria-label="Sections">
|
|
<a href="../index.html#tips">Tips</a>
|
|
<a href="../index.html#blog">Blog</a>
|
|
<a href="../index.html#community">Community</a>
|
|
<a href="../index.html#ask">About</a>
|
|
</nav>
|
|
</header>
|
|
<main class="section post">
|
|
<p class="post-meta"><span class="card-cat">{category}</span> <time datetime="{date}">{date_human}</time></p>
|
|
<h1 class="post-title">{title}</h1>
|
|
{image}
|
|
<div class="post-body">{body}</div>
|
|
<p class="post-note">Published {date_human} · Odoo Community {category} by The OC Guy.
|
|
Conditions under which this advice fails are stated in the text; if they are missing, the post is not finished.</p>
|
|
</main>
|
|
<footer class="foot">
|
|
<p class="foot-name" aria-hidden="true">The OC Guy</p>
|
|
<div class="foot-cols">
|
|
<p class="foot-note">Odoo Community advocate. Written by Oliver Arnold, 安雷威 — no partner programme, no licences sold.</p>
|
|
<nav class="foot-links" aria-label="Elsewhere">
|
|
<a href="../index.html">Home</a>
|
|
<a href="https://github.com/OCA" target="_blank" rel="noopener">OCA modules</a>
|
|
<a href="mailto:oliver@oc-guy.com">Email</a>
|
|
</nav>
|
|
</div>
|
|
</footer>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
|
|
|
|
|
|
def human_date(iso: str) -> str:
|
|
try:
|
|
d = datetime.strptime(iso, "%Y-%m-%d")
|
|
return f"{d.day} {MONTHS[d.month - 1]} {d.year}"
|
|
except ValueError:
|
|
return iso
|
|
|
|
|
|
def build_entry(item: dict, seen_ids: set[str], seen_hashes: set[str], dry: bool) -> dict | None:
|
|
title = str(pick(item, TITLE_KEYS) or "").strip()
|
|
body_raw = str(pick(item, BODY_KEYS) or "").strip()
|
|
if not title and not body_raw:
|
|
return None
|
|
|
|
iso = normalise_date(pick(item, DATE_KEYS))
|
|
category = normalise_category(pick(item, CATEGORY_KEYS))
|
|
raw_id = str(pick(item, ID_KEYS) or "").strip()
|
|
if not title:
|
|
title = strip_tags(body_raw)[:80]
|
|
# a URL wants the headline, not a database id: "17.html" tells a reader nothing
|
|
base = slugify(title, slugify(raw_id, f"post-{iso}"))
|
|
sid = raw_id or base
|
|
|
|
body_html = safe_html(body_raw)
|
|
body_text = strip_tags(body_raw)
|
|
digest = hashlib.sha256((title + "\u0000" + body_text).encode("utf-8")).hexdigest()[:16]
|
|
|
|
if sid in seen_ids or base in seen_ids:
|
|
log(f" duplicate id, skipped: {sid}")
|
|
return None
|
|
if digest in seen_hashes:
|
|
log(f" duplicate content, skipped: {title[:60]}")
|
|
return None
|
|
|
|
teaser = body_text[:200].rsplit(" ", 1)[0] + "…" if len(body_text) > 200 else body_text
|
|
image_rel = None
|
|
sources = image_urls(pick(item, IMAGE_KEYS))
|
|
if not sources:
|
|
# an item carrying only a non-image attachment (pdf, docx, zip) is not a post
|
|
for key in FILE_KEYS:
|
|
value = pick(item, (key,))
|
|
for ref in image_urls(value) or ([value] if isinstance(value, str) else []):
|
|
if isinstance(ref, str) and re.search(r"\.\w{2,5}(\?|$)", ref) and not re.search(
|
|
r"\.(jpe?g|png|webp|avif|gif)(\?|$)", ref, re.I):
|
|
log(f" skipped non-post attachment: {base} ({ref[:80]})")
|
|
return None
|
|
if sources:
|
|
if dry:
|
|
image_rel = f"assets/blog/{base}.<ext>"
|
|
log(f" [dry] would download {len(sources)} image(s), first: {sources[0][:80]}")
|
|
else:
|
|
image_rel = download_image(sources[0], ASSETS_DIR / base)
|
|
else:
|
|
log(f" no image in payload: {base}")
|
|
|
|
if not dry:
|
|
POSTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
page = POST_PAGE.format(
|
|
title=html.escape(title),
|
|
description=html.escape(teaser)[:180],
|
|
category=html.escape(category),
|
|
date=iso,
|
|
date_human=html.escape(human_date(iso)),
|
|
image=f'<img class="post-image" src="../{image_rel}" alt="" width="1200" height="800">' if image_rel else "",
|
|
body=body_html or f"<p>{html.escape(body_text)}</p>",
|
|
)
|
|
(POSTS_DIR / f"{base}.html").write_text(page, encoding="utf-8")
|
|
|
|
return {
|
|
"id": base,
|
|
"source_id": sid,
|
|
"date": iso,
|
|
"category": category,
|
|
"headline": title,
|
|
"teaser": teaser,
|
|
"image": image_rel,
|
|
"link": f"blog/{base}.html",
|
|
"hash": digest,
|
|
"published": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
}
|
|
|
|
|
|
def load_manifest() -> dict:
|
|
if MANIFEST.exists():
|
|
try:
|
|
return json.loads(MANIFEST.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
log("manifest unreadable, starting a new one")
|
|
return {"generated": None, "source": DEFAULT_URL, "posts": []}
|
|
|
|
|
|
def write_manifest(manifest: dict) -> None:
|
|
MANIFEST.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest["posts"].sort(key=lambda p: (p.get("date", ""), p.get("published", "")), reverse=True)
|
|
fd, tmp = tempfile.mkstemp(dir=str(MANIFEST.parent), suffix=".json")
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
json.dump(manifest, fh, ensure_ascii=False, indent=2)
|
|
fh.write("\n")
|
|
os.replace(tmp, MANIFEST)
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--url", default=os.environ.get("OC_GUY_POSTS_WEBHOOK", DEFAULT_URL))
|
|
ap.add_argument("--fixture", help="read a saved JSON payload instead of the webhook")
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
ap.add_argument("--check", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
manifest = load_manifest()
|
|
posts = manifest.get("posts", [])
|
|
|
|
if args.check:
|
|
log(f"manifest: {len(posts)} post(s)")
|
|
for p in posts[:10]:
|
|
log(f" {p['date']} {p['category']:<5} {p['id']}")
|
|
return 0
|
|
|
|
if LOCK.exists():
|
|
log("another sync is running, exiting")
|
|
return 0
|
|
LOCK.parent.mkdir(parents=True, exist_ok=True)
|
|
LOCK.write_text(str(os.getpid()), encoding="utf-8")
|
|
|
|
try:
|
|
if args.fixture:
|
|
payload = json.loads(Path(args.fixture).read_text(encoding="utf-8"))
|
|
else:
|
|
try:
|
|
payload = fetch(args.url)
|
|
except Exception as e: # noqa: BLE001
|
|
log(f"ERROR webhook unavailable: {e}")
|
|
log("nothing published, manifest untouched")
|
|
return 2
|
|
|
|
items = unwrap(payload)
|
|
log(f"payload: {len(items)} item(s)")
|
|
|
|
seen_ids = {p.get("id") for p in posts} | {p.get("source_id") for p in posts}
|
|
seen_hashes = {p.get("hash") for p in posts}
|
|
new_entries = []
|
|
|
|
for item in items:
|
|
entry = build_entry(item, seen_ids, seen_hashes, args.dry_run)
|
|
if not entry:
|
|
continue
|
|
if args.dry_run:
|
|
log(f" [dry] would publish: {entry['category']:<5} {entry['date']} {entry['headline'][:70]}")
|
|
continue
|
|
seen_ids.add(entry["id"])
|
|
seen_ids.add(entry["source_id"])
|
|
seen_hashes.add(entry["hash"])
|
|
new_entries.append(entry)
|
|
log(f" published: {entry['category']:<5} {entry['date']} {entry['id']}")
|
|
|
|
if args.dry_run:
|
|
log("dry run: nothing written")
|
|
return 0
|
|
|
|
if new_entries:
|
|
manifest["posts"] = new_entries + posts
|
|
manifest["generated"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
manifest["source"] = args.url
|
|
write_manifest(manifest)
|
|
log(f"manifest updated: {len(new_entries)} new, {len(manifest['posts'])} total")
|
|
else:
|
|
log("no new posts")
|
|
return 0
|
|
finally:
|
|
LOCK.unlink(missing_ok=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|