Initial commit: SEO rule engine, auditor, validator, exceptions, mirror
Versioning ~/seo-tools as critical SEO infrastructure (was home-dir only). Canonical home of seo_rules.py — ResearchOwl will vendor a copy with a CI diff guard against this repo. No secrets; Ghost access is via the ghst-en wrapper. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+136
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tool A — OG/Twitter auto-mirror write path (theexclusionzone.com, Ghost).
|
||||
|
||||
For each published post, mirror into EMPTY social fields only:
|
||||
og_title <- meta_title (only if og_title empty)
|
||||
twitter_title <- meta_title (only if twitter_title empty)
|
||||
og_description <- meta_description (only if og_description empty)
|
||||
twitter_description<- meta_description (only if twitter_description empty)
|
||||
|
||||
Never overwrites a non-empty field. Touches ONLY those 4 fields (+ updated_at,
|
||||
required by Ghost's optimistic-concurrency check). Idempotent: re-running only
|
||||
fills what is still empty.
|
||||
|
||||
Modes:
|
||||
python3 seo_mirror.py --list # fetch live, show posts needing mirror
|
||||
python3 seo_mirror.py --apply <slug> # apply to ONE post
|
||||
python3 seo_mirror.py --apply-all [--skip a,b] # apply to all needing, minus skips
|
||||
|
||||
Reuses fetch + emptiness logic from the audit engine. READ-ONLY in --list.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
import seo_rules as R
|
||||
from seo_audit import fetch_posts
|
||||
|
||||
SOCIAL = {
|
||||
"og_title": "meta_title",
|
||||
"twitter_title": "meta_title",
|
||||
"og_description": "meta_description",
|
||||
"twitter_description": "meta_description",
|
||||
}
|
||||
|
||||
|
||||
def build_patch(post):
|
||||
"""Return {field: source_value} for each EMPTY social field with a non-empty source."""
|
||||
patch = {}
|
||||
for field, src in SOCIAL.items():
|
||||
if R._empty(post.get(field)):
|
||||
srcval = post.get(src)
|
||||
if not R._empty(srcval):
|
||||
patch[field] = srcval
|
||||
return patch
|
||||
|
||||
|
||||
def needing(posts):
|
||||
out = []
|
||||
for p in posts:
|
||||
patch = build_patch(p)
|
||||
if patch:
|
||||
out.append((p, patch))
|
||||
return out
|
||||
|
||||
|
||||
def apply_one(post, patch):
|
||||
"""PATCH one post via ghst-en with a minimal --from-json body. Returns ghst JSON."""
|
||||
body = {"posts": [dict(patch, updated_at=post["updated_at"])]}
|
||||
fd, jpath = tempfile.mkstemp(suffix=".json", prefix="mirror_patch_")
|
||||
os.close(fd)
|
||||
fd, opath = tempfile.mkstemp(suffix=".json", prefix="mirror_out_")
|
||||
os.close(fd)
|
||||
try:
|
||||
json.dump(body, open(jpath, "w"))
|
||||
cmd = f"ghst-en post update {post['id']} --from-json {jpath} --json > {opath}"
|
||||
r = subprocess.run(cmd, shell=True, stderr=subprocess.PIPE, text=True)
|
||||
if r.returncode != 0:
|
||||
return {"_error": r.stderr.strip()}
|
||||
return json.load(open(opath))
|
||||
finally:
|
||||
os.unlink(jpath)
|
||||
os.unlink(opath)
|
||||
|
||||
|
||||
def fmt_patch(patch):
|
||||
return ", ".join(f"{k}←{('meta_title' if 'title' in k else 'meta_description')}" for k in patch)
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
g = ap.add_mutually_exclusive_group(required=True)
|
||||
g.add_argument("--list", action="store_true")
|
||||
g.add_argument("--apply", metavar="SLUG")
|
||||
g.add_argument("--apply-all", action="store_true")
|
||||
ap.add_argument("--skip", default="", help="comma-separated slugs to skip")
|
||||
args = ap.parse_args()
|
||||
|
||||
posts = fetch_posts()
|
||||
todo = needing(posts)
|
||||
skip = {s.strip() for s in args.skip.split(",") if s.strip()}
|
||||
|
||||
if args.list:
|
||||
print(f"{len(todo)} post(s) need the OG/Twitter mirror "
|
||||
f"(of {len(posts)} published):\n")
|
||||
for p, patch in sorted(todo, key=lambda t: t[0]["slug"]):
|
||||
print(f" {p['slug'][:50]:50} fills: {', '.join(patch.keys())}")
|
||||
return
|
||||
|
||||
if args.apply:
|
||||
match = [(p, pt) for p, pt in todo if p["slug"] == args.apply]
|
||||
if not match:
|
||||
sys.exit(f"'{args.apply}' not found among posts needing mirror "
|
||||
f"(maybe already mirrored, or empty meta source).")
|
||||
p, patch = match[0]
|
||||
print(f"Applying to {p['slug']} ({p['id']}): {fmt_patch(patch)}")
|
||||
res = apply_one(p, patch)
|
||||
if "_error" in res:
|
||||
sys.exit(f"FAILED: {res['_error']}")
|
||||
print(" ghst returned OK.")
|
||||
return
|
||||
|
||||
# --apply-all
|
||||
applied, failed = [], []
|
||||
for p, patch in sorted(todo, key=lambda t: t[0]["slug"]):
|
||||
if p["slug"] in skip:
|
||||
print(f" SKIP {p['slug']}")
|
||||
continue
|
||||
res = apply_one(p, patch)
|
||||
if "_error" in res:
|
||||
failed.append((p["slug"], res["_error"]))
|
||||
print(f" FAIL {p['slug']}: {res['_error'][:80]}")
|
||||
else:
|
||||
applied.append(p["slug"])
|
||||
print(f" OK {p['slug']} ({', '.join(patch.keys())})")
|
||||
print(f"\nApplied: {len(applied)} | Failed: {len(failed)} | Skipped: {len(skip)}")
|
||||
if failed:
|
||||
for s, e in failed:
|
||||
print(f" FAILED {s}: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user