439 lines
17 KiB
Python
439 lines
17 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")
|
|
|
|
# The three categories are the three areas of the homepage. Nothing else exists:
|
|
# tips -> /#tips news -> /#news modules -> /#modules
|
|
KNOWN_CATEGORIES = {"tips": "tips", "tip": "tips", "tricks": "tips",
|
|
"news": "news", "community": "news", "blog": "news",
|
|
"article": "news", "post": "news", "release": "news",
|
|
"modules": "modules", "module": "modules", "oca": "modules",
|
|
"featured": "modules", "integration": "modules"}
|
|
|
|
|
|
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 "news"
|
|
|
|
|
|
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?v=20261024b">
|
|
</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#news">News</a>
|
|
<a href="../index.html#modules">Modules</a>
|
|
<a href="../index.html#ask">About</a>
|
|
</nav>
|
|
</header>
|
|
|
|
<aside class="sponsor" aria-label="Sponsor">
|
|
<p>Our Sponsor: <a href="https://ODOO4projects.com/?utm_source=oc-guy.com&utm_medium=website&utm_campaign={sponsor_campaign}" target="_blank" rel="noopener sponsored">Awesome Odoo Community Hosting</a></p>
|
|
</aside>
|
|
<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>
|
|
<p class="sponsor-foot">Our Sponsor: <a href="https://ODOO4projects.com/?utm_source=oc-guy.com&utm_medium=website&utm_campaign={sponsor_campaign}" target="_blank" rel="noopener sponsored">Awesome Odoo Community Hosting</a></p>
|
|
</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)),
|
|
sponsor_campaign=base,
|
|
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())
|