chemavx-seo-tools/seo_mirror.py

#!/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()