#!/usr/bin/env python3 """ Repliz API Client — Central module untuk semua integrasi Repliz. Single source of truth untuk credential, account IDs, dan API calls. Pemakaian: from repliz_client import ReplizClient, ACC, get_client client = get_client() result = client.create_text(ACC.STUDIOSOFT_THREADS, "caption...", "2026-07-25T...") """ import base64 import json import os import io import re import urllib.error import urllib.request from datetime import datetime, timezone, timedelta from typing import Optional, Union # ── WP Credentials (TIDAK DIPAKAI LAGI untuk upload gambar — pindah ke server OpenClaw) ── WP_URL = "https://studiosoft.id" # non-www: www kena redirect loop 301 WP_AUTH_B64 = base64.b64encode(b"bambangbot:q26r RePo BVy6 pwU2 F3z7 ouyl").decode() # ── Server Gambar OpenClaw (FIXED — 21 Agustus 2026) ── # Semua image disimpan di server OpenClaw (folder openclaw-img), diserve via HTTP (port 8123) # + Tailscale Funnel → URL publik. NO base64, NO WP. IMG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "openclaw-img") IMG_BASE_URL = "https://mantapserver.tail3947bb.ts.net" # FIXED 21/08: Tailscale Funnel aktif IMG_PORT = 8123 WP_HEADERS = {"Authorization": f"Basic {WP_AUTH_B64}", "User-Agent": "Mozilla/5.0"} # ============================================================================= # CREDENTIALS — Satu tempat, ganti di sini aja # ============================================================================= ACCESS_KEY = "4654609866" SECRET_KEY = "soOoH2IRy6BkaIrHSAf1nGly2Rcoiw9Z" BASE_URL = "https://api.repliz.com" # Pre-encoded untuk backward compatibility AUTH_BASE64 = "NDY1NDYwOTg2Njpzb09vSDJJUnk2QmthSXJIU0FmMW5HbHkyUmNvaXc5Wg==" AUTH = AUTH_BASE64 AUTH_HEADER = "Basic " + base64.b64encode(f"{ACCESS_KEY}:{SECRET_KEY}".encode()).decode() HEADERS = { "Authorization": AUTH_HEADER, "Content-Type": "application/json", } BASE = BASE_URL # ============================================================================= # ACCOUNT IDS — Semua akun dalam satu tempat # ============================================================================= # ── Cara lama: dict ── ACCOUNTS = { # @ukyhandoko "uky_threads": "6a334e89c5ff4ce3a3225a48", "uky_twitter": "6a792f0a62caae1e04f57992", "uky_handoko_ig": "6a8bfbe462caae1e04046553", # StudioSoft "studiosoft_threads": "6a335b9ec5ff4ce3a3225c2a", "studiosoft_ig": "6a38ac33c5ff4ce3a3231232", "studiosoft_fb": "6a334952c5ff4ce3a3225941", "studiosoft_linkedin": "6a334c49c5ff4ce3a32259c9", # DMark "dmark_ig": "6a3ba04ac5ff4ce3a329a7f3", "dmark_linkedin": "6a350ecbc5ff4ce3a32299ce", "dmark_fb": "6a3ba0cfc5ff4ce3a329ab22", # LEIT "leit_fb": "6a618e9262caae1e04e99518", "leit_ig": "6a7d765862caae1e04f7ded3", "leit_linkedin": "6a60451662caae1e04e91040", "leit_threads": "6a5f4af462caae1e04e8b9a6", "leit_tiktok": "6a82c69a62caae1e04fbde9c", # TS Law "tslaw_fb": "6a3b41d1c5ff4ce3a32753e8", "tslaw_threads": "6a35f8b7c5ff4ce3a322b2ed", "tslaw_linkedin": "6a60101562caae1e04e901ac", # Apotek Arek Lancor "arelancor_fb": "6a3b41d7c5ff4ce3a3275409", "arelancor_ig": "6a37287bc5ff4ce3a322dab5", # DAS Bengkel Indonesia "das_fb": "6a3b41e1c5ff4ce3a327543e", "das_ig": "6a39e461c5ff4ce3a3233e09", # TSC / DNI Clinic Krian "tsc_threads": "6a4bbc5920e4f4f195e52960", "tsc_ig": "6a86a7e062caae1e04ff8244", "tsc_fb": "6a3b41ddc5ff4ce3a327542c", "tsc_linkedin": "6a4da7f520e4f4f195e796e1", # Soto Semarang Pak Otto (klien baru 20/08/2026) "soto_ig": "6a86d6e962caae1e04ffedaa", "soto_tiktok": "6a86d70362caae1e04ffedef", "soto_threads": "6a86d84b62caae1e04fff12c", # Dojo Ksatria (klien baru 21/08/2026) "dojo_ksatria_ig": "6a88416362caae1e040147d2", # EveryBe (TikTok connect 8 Sep 2026) "everybe_threads": "6a44ed1920e4f4f195d8ecf7", "everybe_fb": "6a3b41dac5ff4ce3a3275418", "everybe_tiktok": "6aa27c1362caae1e041a8d6a", # RSIA Kendangsari "rsia_fb": "6a3b41e8c5ff4ce3a3275470", # Warung Pak Otto (FB) "soto_fb": "6a8f9df162caae1e040754d8", # TSLaw Instagram "tslaw_ig": "6a76a54d62caae1e04f30968", # Uky Yanuhandoko (akun pribadi, lengkap) "uky_ig": "6a83c90262caae1e04fc83b8", "uky_linkedin": "6a33ca10c5ff4ce3a322705d", "uky_tiktok": "6a34fec8c5ff4ce3a322975a", # Yasma Natura "yasma_fb": "6a3b41e3c5ff4ce3a327544b", # Surabaya Skin Centre (SSC) "ssc_fb": "6a3b41e6c5ff4ce3a3275462", } # ── Cara baru: namespace object (autocomplete friendly) ── class _ACC: """Account IDs sebagai attribute — lebih rapi di IDE.""" UKY_THREADS = "6a334e89c5ff4ce3a3225a48" UKY_TWITTER = "6a792f0a62caae1e04f57992" UKY_HANDOKO_IG = "6a8bfbe462caae1e04046553" # IG pribadi @_ukyhandoko (baru 24 Agu 2026) STUDIOSOFT_THREADS = "6a335b9ec5ff4ce3a3225c2a" STUDIOSOFT_IG = "6a38ac33c5ff4ce3a3231232" STUDIOSOFT_FB = "6a334952c5ff4ce3a3225941" STUDIOSOFT_LINKEDIN = "6a334c49c5ff4ce3a32259c9" DMARK_IG = "6a3ba04ac5ff4ce3a329a7f3" DMARK_LINKEDIN = "6a350ecbc5ff4ce3a32299ce" DMARK_FB = "6a3ba0cfc5ff4ce3a329ab22" LEIT_FB = "6a618e9262caae1e04e99518" LEIT_IG = "6a7d765862caae1e04f7ded3" LEIT_LINKEDIN = "6a60451662caae1e04e91040" LEIT_THREADS = "6a5f4af462caae1e04e8b9a6" LEIT_TIKTOK = "6a82c69a62caae1e04fbde9c" TSLAW_FB = "6a3b41d1c5ff4ce3a32753e8" TSLAW_THREADS = "6a35f8b7c5ff4ce3a322b2ed" TSLAW_LINKEDIN = "6a60101562caae1e04e901ac" TSLAW_IG = "6a76a54d62caae1e04f30968" ARELANCOR_FB = "6a3b41d7c5ff4ce3a3275409" ARELANCOR_IG = "6a37287bc5ff4ce3a322dab5" DAS_FB = "6a3b41e1c5ff4ce3a327543e" DAS_IG = "6a39e461c5ff4ce3a3233e09" DAS_THREADS = "6a39e461c5ff4ce3a3233e09" TSC_THREADS = "6a4bbc5920e4f4f195e52960" TSC_IG = "6a86a7e062caae1e04ff8244" SOTO_IG = "6a86d6e962caae1e04ffedaa" SOTO_TIKTOK = "6a86d70362caae1e04ffedef" SOTO_THREADS = "6a86d84b62caae1e04fff12c" DOJO_KSATRIA_IG = "6a88416362caae1e040147d2" # EveryBe EVERYBE_THREADS = "6a44ed1920e4f4f195d8ecf7" EVERYBE_FB = "6a3b41dac5ff4ce3a3275418" EVERYBE_TIKTOK = "6aa27c1362caae1e041a8d6a" RSIA_FB = "6a3b41e8c5ff4ce3a3275470" # Warung Pak Otto (FB) SOTO_FB = "6a8f9df162caae1e040754d8" # TSLaw Instagram TSLAW_IG = "6a76a54d62caae1e04f30968" # Uky Yanuhandoko (akun pribadi) UKY_IG = "6a83c90262caae1e04fc83b8" UKY_LINKEDIN = "6a33ca10c5ff4ce3a322705d" UKY_TIKTOK = "6a34fec8c5ff4ce3a322975a" # Yasma Natura YASMA_FB = "6a3b41e3c5ff4ce3a327544b" # Surabaya Skin Centre (SSC) SSC_FB = "6a3b41e6c5ff4ce3a3275462" ACC = _ACC() # ============================================================================= # CLIENT # ============================================================================= class ReplizError(Exception): """Repliz API error.""" pass # ── GUARD TEKS CJK (27 Sep 2026 — perintah Bos: "jangan sampai ada text china") ── # Karakter China/Jepang/Korea TIDAK BOLEH nyasar ke caption/title/reply. # Pasang di chokepoint (create & update) supaya jadwal bermasalah gagal di sini, # bukan tayang di akun klien. CJK_RE = re.compile( r"[\u3000-\u303f\u3040-\u30ff\u31f0-\u31ff\u3400-\u4dbf\u4e00-\u9fff" r"\uf900-\ufaff\uff00-\uffef]") def _semua_teks(x): """Ambil semua string dari str/list/dict bersarang.""" if isinstance(x, str): yield x elif isinstance(x, dict): for v in x.values(): yield from _semua_teks(v) elif isinstance(x, (list, tuple)): for v in x: yield from _semua_teks(v) def cek_teks_cjk(*nilai, konteks: str = "") -> None: """Lempar ReplizError kalau ada karakter CJK. Dipanggil sebelum kirim ke API.""" for t in _semua_teks(nilai): m = CJK_RE.search(t) if m: i = m.start() raise ReplizError( f"⛔ TEKS CJK DITOLAK [{konteks}] huruf '{m.group()}' di " f"...{t[max(0, i - 60):i + 60]}... — konten wajib Bahasa Indonesia/Inggris murni." ) # ── GUARD TOPIC THREADS (7 Okt 2026) ───────────────────────────────────────── # Repliz memakai field `topic` sebagai topic_tag di Threads. API-nya menolak # topic_tag >50 karakter: "Param topic_tag must be at most 50 characters long." # Jadwalnya TETAP tersimpan tanpa peringatan, tapi gagal saat jam tayang — jadi # gua di sini, bukan di generator. Bukti: 544 jadwal Threads yang sukses punya # topic <=48 char; jadwal 7 Okt 08:00 WIB ("Social Proof Itu Bernilai Lebih dari # Iklan", topic 208 char) gagal persis karena ini. MAKS_TOPIC_THREADS = 50 # accountId akun Threads (dibaca dari objek `account` di daftar jadwal Repliz). # Sumber data: repliz_accounts_threads.json (workspace). THREADS_ACCOUNT_IDS = frozenset({ "6a334e89c5ff4ce3a3225a48", # Uky Yanuhandoko — @ukyhandoko "6a335b9ec5ff4ce3a3225c2a", # StudioSoft — studiosoft.id "6a35f8b7c5ff4ce3a322b2ed", # TITO SUPRIANTO LAWFIRM — tslaw_firm "6a39e461c5ff4ce3a3233e09", # DAS INDONESIA — das_indonesia "6a4bbc5920e4f4f195e52960", # THE SKIN CODE CLINIC — theskincode_clinic "6a44ed1920e4f4f195d8ecf7", # EveryBe Official — theeverybe "6a5f4af462caae1e04e8b9a6", # LEIT Surabaya — (lihat _ACC.LEIT_THREADS) "6a86d84b62caae1e04fff12c", # Warung Pak Otto / Soto — _ACC.SOTO_THREADS }) def pangkas_topic(topic: str, maks: int = MAKS_TOPIC_THREADS) -> str: """Pangkas topic ke batas kata <= maks karakternya (batas API Threads).""" t = (topic or "").strip() if len(t) <= maks: return t potong = t[:maks] if " " in potong: potong = potong[:potong.rfind(" ")] return potong.rstrip(" ,;:-—") def _topic_threads_aman(account_id: str, topic: str) -> str: """Topic untuk akun Threads yang >50 char dipangkas + catat ke stdout.""" if account_id not in THREADS_ACCOUNT_IDS: return topic aman = pangkas_topic(topic) if aman != (topic or ""): print(f" ⚠️ topic Threads dipangkas {len(topic or '')}->{len(aman)} char " f"(batas API {MAKS_TOPIC_THREADS}): {aman!r}") return aman class ReplizClient: """Repliz API client — semua method API di sini.""" def __init__(self): self.base = BASE_URL self.headers = dict(HEADERS) # ── HTTP Layer ── def _request(self, method: str, endpoint: str, data: dict = None, raw: bool = False) -> Union[dict, str]: url = f"{self.base}{endpoint}" body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, headers=self.headers, method=method) try: with urllib.request.urlopen(req, timeout=60) as resp: raw_body = resp.read() if resp.status == 204: return {"success": True} if raw: return raw_body.decode() return json.loads(raw_body) except urllib.error.HTTPError as e: err = e.read().decode() raise ReplizError(f"HTTP {e.code} on {method} {endpoint}: {err}") from e except Exception as e: # FALLBACK 21 Sep 2026 — DPI/TLS-fingerprint memblok TLS urllib & curl ke # api.repliz.com dari host ini (SSL: UNEXPECTED_EOF_WHILE_READING). # Handshake lewat CLI `openssl s_client` masih lolos → pakai shim itu. try: from repliz_openssl_transport import api_request as _ossl_request except Exception: raise ReplizError(f"Request failed {method} {endpoint}: {e}") from e try: return _ossl_request(method, endpoint, data, raw) except Exception as e2: raise ReplizError( f"Request failed {method} {endpoint}: urllib={e}; openssl={e2}") from e2 def get(self, endpoint: str) -> dict: return self._request("GET", endpoint) def post(self, endpoint: str, data: dict) -> dict: return self._request("POST", endpoint, data) def patch(self, endpoint: str, data: dict) -> dict: return self._request("PATCH", endpoint, data) def delete(self, endpoint: str) -> dict: return self._request("DELETE", endpoint) # ── Schedule CRUD ── def create_schedule(self, account_id: str, description: str, schedule_at: str, title: str = "", topic: str = "", type_: str = "text", medias: list = None, replies: list = None, meta: dict = None, additional_info: dict = None) -> dict: """Buat jadwal post baru.""" cek_teks_cjk(description, title, topic, replies or [], konteks="create_schedule") topic = _topic_threads_aman(account_id, topic) payload = { "accountId": account_id, "title": title or "", "description": description or "", "topic": topic or "", "type": type_, "medias": medias or [], "scheduleAt": schedule_at, "replies": replies or [], } if meta: payload["meta"] = meta if additional_info: payload["additionalInfo"] = additional_info res = self.post("/public/schedule", payload) # 27 Sep 2026: API balas {"scheduleId": "..."} — bukan _id/id. Tanpa normalisasi, # semua pipeline yang mengecek `r.get('_id') or r.get('id')` melaporkan gagal # padahal jadwalnya TERBUAT (kasus das-short-daily: 38 hari dilaporkan ❌, # padahal tayang tiap hari). Samakan kuncinya di sini. if isinstance(res, dict) and res.get("scheduleId"): res.setdefault("_id", res["scheduleId"]) res.setdefault("id", res["scheduleId"]) return res def schedule_text(self, account_id: str, description: str, schedule_at: str, title: str = "", topic: str = "") -> dict: """Post teks aja.""" return self.create_schedule(account_id, description, schedule_at, title=title, topic=topic, type_="text") def upload_to_wp(self, filepath: str, filename: str = None) -> str: """Upload image ke SERVER OPENCLAW (folder openclaw-img) → URL publik (JPEG). FIXED 21/08: tidak lagi upload ke WordPress — semua image di server OpenClaw. """ if not IMG_BASE_URL: print(" ⚠️ IMG_BASE_URL belum di-set — upload gambar ke server OpenClaw GAGAL") return "" os.makedirs(IMG_DIR, exist_ok=True) if filename is None: filename = os.path.basename(filepath) # Convert ke JPEG (Bos: format PNG atau JPEG) from PIL import Image img = Image.open(filepath) if img.mode in ('RGBA', 'LA', 'P'): img = img.convert('RGB') base, _ = os.path.splitext(filename) jpg_name = base + '.jpg' out_path = os.path.join(IMG_DIR, jpg_name) img.save(out_path, format='JPEG', quality=90) # URL harus pakai /openclaw-img/ (server serve folder openclaw-img di path itu) url = f"{IMG_BASE_URL.rstrip('/')}/openclaw-img/{jpg_name}" print(f" ✅ Server OpenClaw: {url}") return url def schedule_image(self, account_id: str, description: str, schedule_at: str, image_path: str, filename: str = None, title: str = "", topic: str = "") -> dict: """ Post satu gambar — simpan di server OpenClaw → URL publik (JPEG). NO base64 (aturan FIX 21/08). Kalau upload gagal → jangan buat schedule. """ if filename is None: filename = os.path.basename(image_path) # Simpan ke server OpenClaw biar dapet URL publik img_url = self.upload_to_wp(image_path, filename) if not img_url: return {"ok": False, "error": "upload gambar ke server OpenClaw gagal — skip (no base64)"} return self.create_schedule( account_id, description, schedule_at, title=title, topic=topic, type_="image", medias=[{"url": img_url, "type": "image", "alt": filename}], ) def schedule_album(self, account_id: str, description: str, schedule_at: str, medias: list, title: str = "", meta: dict = None) -> dict: """Post carousel/album.""" return self.create_schedule( account_id, description, schedule_at, title=title, type_="album", medias=medias, meta=meta, ) def get_schedules(self, page: int = 1, limit: int = 50) -> dict: """Lihat daftar jadwal.""" return self.get(f"/public/schedule?page={page}&limit={limit}") def get_schedule(self, schedule_id: str) -> dict: """Lihat satu jadwal.""" return self.get(f"/public/schedule/{schedule_id}") def delete_schedule(self, schedule_id: str) -> dict: """Hapus jadwal.""" return self.delete(f"/public/schedule/{schedule_id}") def update_schedule(self, schedule_id: str, data: dict) -> dict: """Update jadwal. CATATAN API (diverifikasi 4 Okt 2026): PATCH /public/schedule/ TIDAK didukung (balas HTTP 404). Yang bekerja adalah PUT dengan objek jadwal UTUH hasil GET, hanya kolom yang ingin diubah yang diganti. Jangan sertakan _id/createdAt/updatedAt (kalau ikut dikirim, server balas HTTP 500 "Cannot read properties of undefined"). """ asli = self.get_schedule(schedule_id) or {} badan = dict(asli) badan.update(data) for kunci in ("_id", "id", "createdAt", "updatedAt", "__v"): badan.pop(kunci, None) if badan.get("topic"): badan["topic"] = _topic_threads_aman(badan.get("accountId", ""), badan["topic"]) cek_teks_cjk(badan, konteks=f"update_schedule {schedule_id}") return self._request("PUT", f"/public/schedule/{schedule_id}", badan) # ── Utility ── @staticmethod def encode_image(filepath: str) -> tuple: """Encode file gambar ke base64. Return (b64, filename).""" with open(filepath, "rb") as f: b64 = base64.b64encode(f.read()).decode() return b64, os.path.basename(filepath) @staticmethod def make_media(url: str, type_: str = "image", alt: str = "") -> dict: return {"url": url, "type": type_, "alt": alt} @staticmethod def wib_to_utc(wib_str: str) -> str: """'2026-07-25T08:00:00' WIB → ISO UTC.""" dt = datetime.strptime(wib_str, "%Y-%m-%dT%H:%M:%S") return (dt - timedelta(hours=7)).strftime("%Y-%m-%dT%H:%M:%S.000Z") @staticmethod def utc_now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z") @staticmethod def make_schedule(year: int, month: int, day: int, hour: int, minute: int = 0, wib: bool = True) -> str: """Buat string ISO untuk scheduleAt.""" h = hour - 7 if wib else hour if wib: dt = datetime(year, month, day, hour, minute) - timedelta(hours=7) else: dt = datetime(year, month, day, hour, minute) return dt.strftime("%Y-%m-%dT%H:%M:%S.000Z") @staticmethod def next_slot_utc(hour_wib: int, minute_wib: int = 0, base: datetime = None) -> str: """Schedule di jam WIB tertentu hari ini (UTC).""" now = base or datetime.now(timezone.utc) + timedelta(hours=7) target = datetime(now.year, now.month, now.day, hour_wib, minute_wib) target_utc = target - timedelta(hours=7) return target_utc.strftime("%Y-%m-%dT%H:%M:%S.000Z") # ── Singleton ── _client: Optional[ReplizClient] = None def get_client() -> ReplizClient: global _client if _client is None: _client = ReplizClient() return _client # ── Convenience (gampang import) ── def upload_image_to_server(filepath: str, filename: str = None) -> str: """Module-level helper: simpan image ke server OpenClaw (openclaw-img) → URL publik. NO base64, NO WP — FIXED 21/08 untuk SEMUA akun.""" return ReplizClient().upload_to_wp(filepath, filename) def schedule_text(account_id: str, description: str, schedule_at: str, title: str = "", topic: str = "") -> dict: return get_client().schedule_text(account_id, description, schedule_at, title, topic) def schedule_image(account_id: str, description: str, schedule_at: str, image_path: str, filename: str = None) -> dict: return get_client().schedule_image(account_id, description, schedule_at, image_path, filename) def schedule_album(account_id: str, description: str, schedule_at: str, medias: list, title: str = "") -> dict: return get_client().schedule_album(account_id, description, schedule_at, medias, title) # ============================================================================= # TEST # ============================================================================= if __name__ == "__main__": c = get_client() # Test koneksi try: result = c.get_schedules(page=1, limit=1) print(f"✅ Repliz connected! Total schedules: {len(result.get('docs',[]))}") print(f" Accounts available: {len(ACCOUNTS)}") except ReplizError as e: print(f"❌ Connection failed: {e}")