from pathlib import Path import datetime import hashlib import json import os import urllib.parse import urllib.request from zoneinfo import ZoneInfo ROOT = Path(__file__).resolve().parents[1] ENV_PATH = Path.home() / ".config/gac-mai-audio/postiz.env" FB = "cmrilrryn0011j7c8kfzgx9zq" YT = "cmriltjyk0013j7c8l7dp3o5i" RECEIPT_PATH = ROOT / "log/postiz-schedule.json" def atomic(value): temporary = Path(str(RECEIPT_PATH) + ".tmp") temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n") os.replace(temporary, RECEIPT_PATH) def load_env(): values = {} for line in ENV_PATH.read_text().splitlines(): if line and not line.lstrip().startswith("#") and "=" in line: key, value = line.split("=", 1) values[key.strip()] = os.environ.get(key.strip(), value.strip().strip('"').strip("'")) return values def request_json(base, key, path, payload=None): body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode() request = urllib.request.Request( base + path, data=body, headers={"Authorization": key, **({"Content-Type": "application/json"} if body else {})}, method="POST" if body else "GET", ) with urllib.request.urlopen(request, timeout=120) as response: return json.load(response) def parse_time(value): if value.endswith("Z"): return datetime.datetime.fromisoformat(value[:-1] + "+00:00") parsed = datetime.datetime.fromisoformat(value) return parsed.replace(tzinfo=datetime.timezone.utc) if parsed.tzinfo is None else parsed def calendar(base, key, start, end): query = urllib.parse.urlencode({ "startDate": start.isoformat().replace("+00:00", "Z"), "endDate": end.isoformat().replace("+00:00", "Z"), }) value = request_json(base, key, "/posts?" + query) return value.get("posts", []) values = load_env() base = values["POSTIZ_BASE_URL"].rstrip("/") key = values["POSTIZ_API_KEY"] assert values["POSTIZ_FACEBOOK_INTEGRATION_ID"] == FB assert values["POSTIZ_YOUTUBE_INTEGRATION_ID"] == YT assert request_json(base, key, "/is-connected").get("connected") is True raw_integrations = request_json(base, key, "/integrations") integrations = raw_integrations.get("integrations", raw_integrations) if isinstance(raw_integrations, dict) else raw_integrations selected = [row for row in integrations if row.get("id") in {FB, YT}] assert len(selected) == 2 assert all(row.get("disabled") is False and row.get("name") == "Gác Mái Audio" for row in selected) media = json.loads((ROOT / "log/postiz-media.json").read_text()) assert media.get("status") == "completed" and media.get("verified") is True assert media["video"].get("id") and media["video"].get("path") assert media["thumbnail"].get("id") and media["thumbnail"].get("path") now_local = datetime.datetime.now(ZoneInfo("Asia/Ho_Chi_Minh")) start = now_local.astimezone(datetime.timezone.utc) end = (now_local + datetime.timedelta(days=14)).astimezone(datetime.timezone.utc) posts = calendar(base, key, start, end) occupied = {FB: set(), YT: set()} other_ignored = 0 for post in posts: integration_id = (post.get("integration") or {}).get("id") if integration_id not in occupied: other_ignored += 1 continue if post.get("state") in {"QUEUE", "PUBLISHED"} and post.get("publishDate"): occupied[integration_id].add(parse_time(post["publishDate"]).astimezone(datetime.timezone.utc).replace(microsecond=0)) slot_local = slot_utc = None for offset in range(15): day = (now_local + datetime.timedelta(days=offset)).date() for hour, minute in ((9, 20), (19, 0)): candidate_local = datetime.datetime.combine(day, datetime.time(hour, minute), tzinfo=ZoneInfo("Asia/Ho_Chi_Minh")) if candidate_local <= now_local: continue candidate_utc = candidate_local.astimezone(datetime.timezone.utc).replace(microsecond=0) if candidate_utc not in occupied[FB] and candidate_utc not in occupied[YT]: slot_local, slot_utc = candidate_local, candidate_utc break if slot_utc: break assert slot_utc is not None info = (ROOT / "output/info.txt").read_text() youtube_title = "Bản Nhạc Gửi Từ Căn Phòng Khóa | Truyện Audio Tâm Lý Gia Đình" facebook_caption = "Cứ đúng hai giờ mười bảy, cây piano trong căn phòng khóa suốt mười hai năm lại tự vang lên. Mai lần theo những nốt sai được sắp đặt như mật mã và phát hiện một cái tên đã bị lấy khỏi bản nhạc. Nhưng trả lại sự thật không có nghĩa là được quyền quyết định thay người bị tổn thương. Mời bạn nghe Bản Nhạc Gửi Từ Căn Phòng Khóa trên Gác Mái Audio." video_media = {"id": media["video"]["id"], "path": media["video"]["path"]} thumbnail_media = {"id": media["thumbnail"]["id"], "path": media["thumbnail"]["path"]} payload = { "type": "schedule", "date": slot_utc.isoformat().replace("+00:00", "Z"), "shortLink": False, "tags": [], "posts": [ { "integration": {"id": FB}, "value": [{"content": facebook_caption, "image": [video_media]}], "settings": {"__type": "facebook"}, }, { "integration": {"id": YT}, "value": [{"content": info, "image": [video_media]}], "settings": { "__type": "youtube", "title": youtube_title, "type": "public", "selfDeclaredMadeForKids": "no", "thumbnail": thumbnail_media, "tags": [], }, }, ], } assert 2 <= len(youtube_title) <= 100 assert len(payload["posts"]) == 2 assert json.loads(json.dumps(payload, ensure_ascii=False)) == payload receipt = { "status": "creating", "verified": False, "slot_local": slot_local.isoformat(), "slot_utc": payload["date"], "integration_ids": [FB, YT], "other_posts_ignored": other_ignored, "payload_sha256": hashlib.sha256(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest(), "media_receipt_path": str(ROOT / "log/postiz-media.json"), "created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } atomic(receipt) response = request_json(base, key, "/posts", payload) receipt["create_response"] = response receipt["status"] = "submitted" atomic(receipt) window_start = slot_utc - datetime.timedelta(minutes=2) window_end = slot_utc + datetime.timedelta(minutes=2) readback = calendar(base, key, window_start, window_end) matched = {} for post in readback: integration_id = (post.get("integration") or {}).get("id") if integration_id not in {FB, YT} or not post.get("publishDate"): continue published = parse_time(post["publishDate"]).astimezone(datetime.timezone.utc).replace(microsecond=0) if published == slot_utc and post.get("state") in {"QUEUE", "PUBLISHED"}: matched[integration_id] = { "id": post.get("id"), "state": post.get("state"), "publishDate": post.get("publishDate"), } receipt["readback"] = matched receipt["verified_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat() if set(matched) != {FB, YT}: receipt["status"] = "partial_failure" if matched else "readback_failed" atomic(receipt) raise RuntimeError("Postiz readback did not find both target integrations") receipt.update(status="completed", verified=True) atomic(receipt) print(json.dumps({"status": "completed", "slot_local": receipt["slot_local"], "slot_utc": receipt["slot_utc"], "posts": matched}, ensure_ascii=False))