#!/usr/bin/env python3 """DAILY IMAGE POSTING — cron 2026-09-26T18:00Z (= 27 Sep 2026 01:00 WIB) Item index 56: uky_quote_feedback_adalah_emas → uky3_output/feedback_adalah_emas_biru.png (brand ukyhandoko, source uky3_output) → Threads @ukyhandoko. Tema kartu (vision, square 1080x1080, bg #122B63 teks putih, standar 16 Sep): Judul : "Feedback Adalah Hadiah Paling Mahal" Body : klien yang ngasih kritik itu bukan musuh, mereka aset. Kritik pedas bikin kami benahi SOP, sistem, komunikasi. Bisnis yang tak pernah dikritik = bisnis yang tak pernah tumbuh. Footer : @ukyhandoko • Founder StudioSoft | Header @ukyhandoko 1/1 Layout : rata kiri, satu blok tengah, tanpa separator, tanpa glyph rusak. OK. Aturan seri: Threads ≤500 char, gaya personal "saya", pancingan diskusi dari pool diskusi_engine (dipilih manual, belum dipakai di 53 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 (16 entri log yang pakai pool): netral 10 (62%) / opini 5 (31%) / sanjungan 1 (6%) → target 60/25/15. Pilih **sanjungan** (pool umum): hasil jadi 59/29/12 = paling dekat target (kalau netral → 65/29/6, kalau opini → 59/35/6). Tema "kritik klien = hadiah" nyambung dengan apresiasi ke audiens yang sudah mau mendengar kritik & benahi diri. Hook sanjungan TIDAK berbentuk pertanyaan (sesuai spesifikasi Bos), jadi guard CTA diperiksa sebagai: hook verbatim dari pool `umum/sanjungan` DAN caption ≤500 char, 0 hashtag, 0 mention (preseden 25 Sep 2026). """ 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_feedback_adalah_emas" FILENAME = "feedback_adalah_emas_biru.png" SRC = os.path.join(ROOT, "uky3_output", FILENAME) MEDIA_NAME = "uky_quote_feedback_adalah_emas.jpg" TITLE = "Feedback Adalah Hadiah Paling Mahal" TOPIC = ("Kritik klien = hadiah paling mahal: klien yang berani komplain itu aset, bukan musuh — " "dari kritik pedas lahir SOP, sistem & komunikasi yang lebih rapi; bisnis yang tak pernah " "dikritik adalah bisnis yang tak pernah tumbuh") UKY_T = ACC.UKY_THREADS BODY = ( "Kritik dari klien itu bukan serangan. Itu hadiah paling mahal untuk bisnis.\n\n" "Dulu saya sempat sakit hati waktu klien ngomong pedas. Tapi dari situ justru lahir SOP, " "sistem, dan cara komunikasi yang lebih rapi. Klien yang diam biasanya pergi tanpa kabar " "\u2014 yang berani mengkritik justru menyelamatkan bisnis kita.\n\n" "Jadi lain kali ada yang komplain, dengarkan dulu sampai habis." ) # Hook dipilih manual dari pool: tone sanjungan (target distribusi 60/25/15 → 12% sanjungan). HOOK_RAW = "Buat yang sudah mempraktikkan ini, Anda pasti merasakan bedanya. Bangga sama Anda!" HOOK = HOOK_RAW + " \U0001F44F" HOOK_POOL = ("umum", "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 — upload hanya kalau path belum ada.""" 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("\u26a0\ufe0f belum ada di wp_image_map.json \u2014 perlu upload WP manual:", 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]}) — dipilih manual, 0 kali dipakai di 53 entri log " f"(distribusi pool sekarang: netral/opini/sanjungan dari {n_pakai_pool} entri)", }) 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())