#!/usr/bin/env python3 """DAILY IMAGE POSTING — cron 2026-10-04T18:00Z (= 5 Okt 2026 01:00 WIB) Item index 64: uky_quote_slow_business -> uky3_output/slow_business_biru.png (brand ukyhandoko, source uky3_output) -> Threads @ukyhandoko. Tema kartu (vision_analyze, square 1080x1080, bg #122B63 teks putih, standar 16 Sep): Judul : "Slow Business Is Good Business" Body : dulu gue pikir bisnis harus serba cepat (cepet dapet klien, cepet ngerjain, cepet closing) -> stres, burnout, kualitas jeblok; sekarang pelan-pelan, ambil klien yang cocok, proses teliti -> hasil lebih baik, klien puas, tidur nyenyak. Footer : @ukyhandoko - Founder StudioSoft | Header @ukyhandoko + 1/1 | pixel(8,8)=(18,43,99) Layout : rata kiri, satu blok tengah vertikal, tanpa separator. OK. Aturan seri: Threads <=500 char (target <=480), 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). Seri uky TIDAK pakai WP. Target akun HANYA Threads @ukyhandoko. Distribusi tone pool (hook yang cocok pool, dihitung ulang dari SELURUH caption_final log, n=24 entri): netral 14 / opini 6 / sanjungan 4 -> 58.3/25.0/16.7. Deviasi ke target 60/25/15: +netral -> 60.0/24.0/16.0 dev 2.00 <-- TERKECIL +opini -> 56.0/28.0/16.0 dev 8.00 +sanjungan -> 56.0/24.0/20.0 dev 10.00 Pilih **netral**. Hook `umum/netral`: "Menarik bukan? Apa tanggapan Anda soal ini?" (FREE, 0 kali dipakai) - hook generik, nyambung dgn tema refleksi bisnis. PITFALL tetap: baris baru caption dibangun dengan chr(10)*2 (NL2), JANGAN escape backslash-n literal. """ 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) NL2 = chr(10) * 2 ITEM_ID = "uky_quote_slow_business" FILENAME = "slow_business_biru.png" SRC = os.path.join(ROOT, "uky3_output", FILENAME) MEDIA_NAME = "uky_quote_slow_business.jpg" TITLE = "Slow Business Is Good Business" TOPIC = ("Slow business: dulu ngejar serba cepat (cepet dapet klien, cepet ngerjain, cepet closing) " "-> stres, burnout, kualitas jeblok; sekarang pelan-pelan, ambil klien yang cocok, " "proses teliti -> hasil lebih baik, klien lebih puas, tidur lebih nyenyak") UKY_T = ACC.UKY_THREADS BODY = ( "Dulu saya pikir bisnis itu harus serba cepat \u2014 cepet dapet klien, cepet ngerjain, " "cepet closing. Hasilnya? Stres, burnout, dan kualitas yang jeblok." + NL2 + "Sekarang saya memilih jalan yang lebih pelan: ambil klien yang cocok, kerjakan dengan " "teliti. Hasilnya lebih baik, klien lebih puas, dan tidur lebih nyenyak." ) # Hook NETRAL dari pool `umum` (FREE, belum dipakai di log mana pun). HOOK_RAW = "Menarik bukan? Apa tanggapan Anda soal ini?" HOOK = HOOK_RAW HOOK_POOL = ("umum", "netral") CTA_RE = re.compile(r"komentar|pendapat|diskusi|setuju|share|pengalaman|menurut|strategi|sepakat|\?", re.I) CJK_RE = re.compile( "[" + chr(0x4E00) + "-" + chr(0x9FFF) + chr(0x3040) + "-" + chr(0x30FF) + chr(0xAC00) + "-" + chr(0xD7AF) + "]" ) 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 - pilih hook lain") assert BODY.count(NL2) == 1, "BODY harus punya tepat 1 pemisah paragraf" return BODY + NL2 + 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("INFO: tidak ada di wp_image_map.json - normal untuk seri uky (tanpa WP):", key) return None print(f"WP MAP: {key} -> id={ent['id']} {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):") print(cap) print("--- repr(newline check):", repr(cap[:120])) assert len(cap) <= 500, f"caption {len(cap)} char > 500" assert len(cap) <= 480, f"caption {len(cap)} char > 480 (hindari elipsis truncate)" assert NL2 in cap, "caption tidak punya baris kosong ganda (newline asli)" assert "\\n" not in cap, "caption memuat backslash-n LITERAL" assert not re.search(r"@\w+", cap), "ada mention!" assert not re.search(r"https?://", cap), "ada link (seri ini tanpa link)" 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)" assert not CJK_RE.search(cap), "ada teks CJK!" 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("STOP: URL publik tidak 200 - 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"SKIP: sudah ada {len(pre)} jadwal utk media ini (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"VERIFIKASI: {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 desc = d.get("description") or "" 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(desc)}) replies={len(d.get('replies') or [])} newline_asli={NL2 in desc} literal={chr(92)+'n' in desc}") if d.get("id") == sid and d.get("status") in ("pending", "success") and NL2 in desc: ok = True if not (ok and len(hits) == 1): print("GAGAL: verifikasi tidak pas (harus 1 jadwal, sid cocok, newline asli)") return 1 slot_utc = hits[0]["scheduleAt"] print("OK TERVERIFIKASI: 1 jadwal, media + caption (newline asli) 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("INFO: item sudah tercatat di log - state tidak di-increment lagi") else: st["index"] = (st["index"] + 1) % st["total"] json.dump(st, open(sp, "w"), indent=2) print("STATE ->", 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": f"{HOOK_POOL[1]} (pool {HOOK_POOL[0]}) - dipilih manual, 0 kali dipakai di {len(log)} entri log", }) json.dump(log, open(lp, "w"), indent=2, ensure_ascii=False) print("LOG ->", lp, f"({len(log)} entri)") return 0 if __name__ == "__main__": sys.exit(main())