#!/usr/bin/env python3 """DAILY IMAGE POSTING — cron 2026-09-28T18:00Z (= 29 Sep 2026 01:00 WIB) Item index 58: uky_quote_growth_mindset → uky3_output/growth_mindset_biru.png (brand ukyhandoko, source uky3_output) → Threads @ukyhandoko. Tema kartu (vision_analyze, square 1080x1080, bg #122B63 teks putih, standar 16 Sep): Judul : "Growth Mindset: Kunci Bertahan di Era Digital" Body : yang membedakan bisnis bertahan vs mati bukan modal/relasi, tapi kesediaan buat terus belajar; yang gamau upgrade skill, adaptasi teknologi, denger kritik pasti ketinggalan. "Stay hungry, stay foolish." Footer : @ukyhandoko • Founder StudioSoft | Header @ukyhandoko 1/1 Layout : rata kiri, satu blok tengah vertikal, tanpa separator, tanpa glyph rusak. OK. Aturan seri: Threads <=500 char, gaya personal "saya", pancingan diskusi dari pool diskusi_engine (dipilih manual, belum pernah dipakai di seluruh entri log), 0 hashtag, 0 mention internal, replies=[]. Media PNG -> JPEG di openclaw-img (prefix uky_quote_), URL publik Tailscale funnel, wajib HTTP 200 sebelum schedule. Slot 08:00 WIB bebas pertama (dihitung dari API). Distribusi tone (18 entri log yang memakai pool): netral 11 (61%) / opini 5 (28%) / sanjungan 2 (11%) -> target 60/25/15. Pilih **sanjungan**: deviasi |58-60|+|26-25|+ |16-15| = 4 (paling kecil; netral = 9, opini = 13). Tema "growth mindset / berani belajar teknologi" nyambung langsung dengan hook `teknologi/sanjungan` "Salut untuk Anda yang berani belajar teknologi baru — itu skill paling berharga di era sekarang! 👏" (hook bebas, 0 kali dipakai di log). Hook sanjungan dari pool TIDAK berbentuk pertanyaan -> guard CTA dilonggarkan (CTA_RE.search(cap) or tone == "sanjungan"), alasan didokumentasikan di docstring (preseden item 56 / image_post_20260927_uky26.py). """ import json, os, re, sys, subprocess from datetime import datetime, timezone, timedelta ROOT = "/home/mantapman/.hermes/profiles/atun/workspace" sys.path.insert(0, ROOT) from repliz_client import get_client, ACC, IMG_BASE_URL, IMG_DIR from PIL import Image from diskusi_engine import PANCINGAN import repliz_openssl_transport # 21 Sep 2026: TLS curl/urllib ke api.repliz.com diblok (DPI) -> pakai shim openssl. repliz_openssl_transport.install(verbose=False) ITEM_ID = "uky_quote_growth_mindset" FILENAME = "growth_mindset_biru.png" SRC = os.path.join(ROOT, "uky3_output", FILENAME) MEDIA_NAME = "uky_quote_growth_mindset.jpg" TITLE = "Growth Mindset: Kunci Bertahan di Era Digital" TOPIC = ("Growth mindset = kunci bertahan di era digital: yang membedakan bisnis yang bertahan vs " "yang mati bukan modal atau relasi, tapi kesediaan buat terus belajar — yang gamau upgrade " "skill, adaptasi teknologi, dan denger kritik pasti ketinggalan; 'stay hungry, stay foolish'") UKY_T = ACC.UKY_THREADS BODY = ( "Banyak yang mengira bisnis bertahan karena modal atau relasi. Menurut saya bukan itu.\n\n" "Yang membedakan bisnis yang bertahan dan yang mati adalah kesediaan buat terus belajar. " "Saya sendiri masih belajar hal baru tiap bulan \u2014 dari AI, otomasi, sampai cara ngobrol " "sama klien.\n\n" "Yang gamau upgrade skill, gamau adaptasi teknologi, dan gamau denger kritik, pasti " "ketinggalan. Stay hungry, stay foolish." ) # Hook dipilih manual dari pool: tone sanjungan (target distribusi 60/25/15 -> 16% sanjungan). HOOK_RAW = "Salut untuk Anda yang berani belajar teknologi baru \u2014 itu skill paling berharga di era sekarang! \U0001f44f" HOOK = HOOK_RAW HOOK_POOL = ("teknologi", "sanjungan") CTA_RE = re.compile(r"komentar|pendapat|diskusi|setuju|share|pengalaman|menurut|strategi|\?", re.I) def build_caption(): assert HOOK_RAW in PANCINGAN[HOOK_POOL[0]][HOOK_POOL[1]], "hook tidak ada di pool" log = json.load(open(os.path.join(ROOT, "image_posting_log.json"))) for e in log: if HOOK_RAW in (e.get("caption_final") or ""): raise RuntimeError("hook sudah dipakai di log \u2014 pilih hook lain") return BODY + "\n\n" + HOOK _UKY_CACHE = [] def fetch_uky(client, refresh=False): if _UKY_CACHE and not refresh: return _UKY_CACHE[0] r = client.get_schedules(page=1, limit=100) pages = (r["totalDocs"] + 99) // 100 docs = list(r["docs"]) for p in range(2, pages + 1): docs.extend(client.get_schedules(page=p, limit=100)["docs"]) out = [] for d in docs: acc = d.get("account") or {} if (d.get("accountId") or acc.get("id") or acc.get("_id")) != UKY_T: continue out.append(d) if refresh: _UKY_CACHE[:] = [out] else: _UKY_CACHE.append(out) return out def free_slot(client): """08:00 WIB (01:00Z) bebas pertama utk Threads uky, mulai besok WIB.""" besok = (datetime.now(timezone.utc) + timedelta(hours=7)).date() + timedelta(days=1) ocupado = set() for d in fetch_uky(client): at = d.get("scheduleAt") if not at: continue dt = datetime.fromisoformat(at.replace("Z", "+00:00")) + timedelta(hours=7) if dt.hour == 8 and dt.minute == 0: ocupado.add(dt.date()) hari = besok for _ in range(400): if hari not in ocupado: return f"{hari.isoformat()}T01:00:00.000Z", hari hari += timedelta(days=1) raise RuntimeError("tidak ada slot bebas 400 hari") def upload(): """PNG -> JPEG di openclaw-img, return URL publik + path lokal.""" os.makedirs(IMG_DIR, exist_ok=True) img = Image.open(SRC) if img.mode in ("RGBA", "LA", "P"): img = img.convert("RGB") out = os.path.join(IMG_DIR, MEDIA_NAME) img.save(out, format="JPEG", quality=90) url = f"{IMG_BASE_URL.rstrip('/')}/openclaw-img/{os.path.basename(out)}" return url, out def check_public(url): """Simulasi jalur INTERNET (resolve ke ingress Tailscale) — bukan tailnet-direct.""" codes = {} for ip in ("103.84.155.153", "103.84.155.217"): r = subprocess.run( ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "30", "--resolve", f"mantapserver.tail3947bb.ts.net:443:{ip}", url], capture_output=True, text=True) codes[ip] = r.stdout.strip() return codes def wp_media_check(): """Langkah 3 job: pakai wp_image_map.json — seri uky tidak pakai WP (info saja).""" m = json.load(open(os.path.join(ROOT, "wp_image_map.json"))) key = f"uky3_output/{FILENAME}" ent = m.get(key) if not ent: print("\u2139\ufe0f tidak ada di wp_image_map.json \u2014 normal untuk seri uky (tanpa WP):", key) return None r = subprocess.run(["curl", "-4", "-s", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "30", ent["url"]], capture_output=True, text=True) print(f"WP MAP: {key} \u2192 id={ent['id']} http={r.stdout.strip()} {ent['url']}") return ent def main(): assert os.path.exists(SRC), f"file gambar tidak ada: {SRC}" wp_media_check() cap = build_caption() print(f"CAPTION ({len(cap)} chars):\n{cap}\n") assert len(cap) <= 500, f"caption {len(cap)} char > 500" assert not re.search(r"@\w+", cap), "ada mention!" assert not re.search(r"https?://", cap), "ada link (seri ini tanpa link)" # guard CTA: hook sanjungan dari pool memang berbentuk apresiasi (bukan pertanyaan) assert CTA_RE.search(cap) or HOOK_POOL[1] == "sanjungan", "tidak ada pancingan diskusi" assert "#" not in cap, "ada hashtag (aturan: 0 hashtag untuk seri ini)" url, out = upload() print("LOCAL JPEG:", out, os.path.getsize(out), "bytes") print("PUBLIC URL:", url) codes = check_public(url) print("INTERNET-PATH CODES:", codes) if not any(c == "200" for c in codes.values()): print("\u274c URL publik tidak 200 \u2014 STOP, jangan schedule (aturan media server).") return 1 c = get_client() # Audit anti-dobel: EXACT basename (preseden 15 Sep 2026 — substring = false positive). def list_media(refresh=False): hits = [] for d in fetch_uky(c, refresh=refresh): for m in (d.get("medias") or []): u = (m.get("url") or "").split("?")[0] base = u.rsplit("/", 1)[-1] if base == MEDIA_NAME: hits.append(d) break return hits # ── IDEMPOTENSI (create response pakai 'scheduleId', bukan '_id') ── pre = list_media(refresh=True) if pre: sid = pre[0].get("id") print(f"\u23ed\ufe0f sudah ada {len(pre)} jadwal utk media ini \u2192 SKIP create (id={sid})") else: slot_utc, hari = free_slot(c) print("SLOT:", slot_utc, f"(08:00 WIB {hari})") res = c.create_schedule( UKY_T, cap, slot_utc, title=TITLE, topic=TOPIC, type_="image", medias=[{"url": url, "type": "image", "alt": FILENAME}], replies=[], ) sid = res.get("scheduleId") or res.get("_id") or res.get("id") print("CREATE RESPONSE:", json.dumps(res)[:400]) print("SID:", sid) # ── VERIFIKASI via re-list (fetch ULANG, jangan pakai snapshot lama) ── hits = list_media(refresh=True) print(f"\nVERIFIKASI: {len(hits)} jadwal dgn media {MEDIA_NAME} di Threads uky") ok = False slot_dipakai = None for d in hits: wib = datetime.fromisoformat(d["scheduleAt"].replace("Z", "+00:00")) + timedelta(hours=7) slot_dipakai = wib print(f" id={d.get('id')} status={d.get('status')} type={d.get('type')} at={wib.strftime('%Y-%m-%d %H:%M')} WIB") print(f" media={d['medias'][0]['url']}") print(f" desc({len(d.get('description') or '')}) replies={len(d.get('replies') or [])}") if d.get("id") == sid and d.get("status") in ("pending", "success"): ok = True if not (ok and len(hits) == 1): print("\u26a0\ufe0f verifikasi tidak pas (harus 1 jadwal dgn sid cocok)") return 1 slot_utc = hits[0]["scheduleAt"] print("\n\u2705 TERVERIFIKASI: 1 jadwal, media + caption terpasang") # ── UPDATE STATE + LOG (idempoten) ── lp = os.path.join(ROOT, "image_posting_log.json") log = json.load(open(lp)) sudah = any(e.get("item_id") == ITEM_ID for e in log) sp = os.path.join(ROOT, "image_posting_state.json") st = json.load(open(sp)) if sudah: print("\u2139\ufe0f item sudah tercatat di log \u2192 state tidak di-increment lagi") else: st["index"] = (st["index"] + 1) % st["total"] json.dump(st, open(sp, "w"), indent=2) print("STATE \u2192", st) if not sudah: n_pakai_pool = sum(1 for e in log if any( h in (e.get("caption_final") or "") for ind in PANCINGAN for tn in PANCINGAN[ind] for h in PANCINGAN[ind][tn])) + 1 log.append({ "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z"), "item_id": ITEM_ID, "brand": "ukyhandoko", "source": "uky3_output", "filename": FILENAME, "topic": TOPIC, "slot_note": f"slot 08:00 WIB bebas pertama dari cek API = {slot_dipakai.strftime('%Y-%m-%d %H:%M')} WIB", "schedule_times": {"threads_uky": slot_utc + f" ({slot_dipakai.strftime('%Y-%m-%d')} 08:00 WIB) - image 1"}, "repliz_ids": {"threads_uky": sid}, "caption_chars": len(cap), "media_url": url, "caption_final": cap, "next_index": st["index"], "hook_note": f"{HOOK_POOL[1]} (pool {HOOK_POOL[0]}) \u2014 dipilih manual, 0 kali dipakai di {len(log)} entri log " f"(distribusi pool sekarang: {n_pakai_pool} entri netral/opini/sanjungan)", }) json.dump(log, open(lp, "w"), indent=2, ensure_ascii=False) print("LOG \u2192", lp, f"({len(log)} entri)") return 0 if __name__ == "__main__": sys.exit(main())