#!/usr/bin/env python3 """DAILY IMAGE POSTING — cron 2026-09-22T18:00Z (= 23 Sep 2026 01:00 WIB) Item index 52: uky_quote_belajar_gagal → belajar_gagal_biru.png (brand ukyhandoko, source uky3_output) → Threads @ukyhandoko. Tema kartu (vision, square 1080x1080, bg #122B63 teks putih, standar 16 Sep): Judul : "Gagal Bukan Akhir, Tapi Awal yang Baru" Body : setiap kegagalan bisnis = pelajaran paling berharga, bukan dari buku / seminar tapi dari jatuh-bangun sendiri; jangan ulangi kesalahan yang sama. Footer : @ukyhandoko • Founder StudioSoft | Header @ukyhandoko 1/1 Layout : rata kiri, satu blok tengah, tanpa separator. OK. Aturan seri: Threads ≤500 char, gaya personal "saya", pancingan diskusi (dipilih manual dari pool diskusi_engine agar TIDAK berulang), 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. """ 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_belajar_gagal" FILENAME = "belajar_gagal_biru.png" SRC = os.path.join(ROOT, "uky3_output", FILENAME) MEDIA_NAME = "uky_quote_belajar_gagal.jpg" TITLE = "Gagal Bukan Akhir, Tapi Awal yang Baru" TOPIC = ("Belajar dari kegagalan: setiap kegagalan bisnis = pelajaran paling berharga, " "bukan dari buku/seminar tapi dari jatuh-bangun sendiri — kuncinya jangan ulangi kesalahan yang sama") UKY_T = ACC.UKY_THREADS BODY = ( "Saya pernah gagal sampai harus membangun ulang dari nol. Waktu itu rasanya pengin berhenti aja. \U0001F4A1\n\n" "Ternyata kegagalan bukan lawan dari pertumbuhan \u2014 dia bagian dari prosesnya. " "Yang membedakan bukan yang tidak pernah gagal, tapi yang berhenti mengulang kesalahan yang sama.\n\n" "Gagal itu data, bukan vonis." ) # Hook dipilih manual: netral (60% bucket), belum dipakai item 17-20 & 51. HOOK = "Berapa lama Anda bertahan sebelum melihat hasil dari digital marketing? Share pengalamannya." CTA_RE = re.compile(r"komentar|pendapat|diskusi|setuju|share|pengalaman|menurut", re.I) def build_caption(): assert HOOK in PANCINGAN["digital_marketing"]["netral"], "hook tidak ada di pool" cap = BODY + "\n\n" + HOOK return cap _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) _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(150): if hari not in ocupado: return f"{hari.isoformat()}T01:00:00.000Z", hari hari += timedelta(days=1) raise RuntimeError("tidak ada slot bebas 150 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 main(): 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 CTA_RE.search(cap), "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() 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 (jangan percaya response create) ── 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") == "pending": ok = True if not (ok and len(hits) == 1): print("\u26a0\ufe0f verifikasi tidak pas (harus 1 pending dgn sid cocok)") return 1 slot_utc = hits[0]["scheduleAt"] print("\n\u2705 TERVERIFIKASI: 1 jadwal pending, 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: 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": "netral (pool digital_marketing) — dipilih manual, belum dipakai item 17-20 & 51", }) 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())