191 lines
10 KiB
Python
191 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""Публикация карусели (slides/*.png) в соцсети Alex.
|
||
|
||
python3 video/carousels/publish_carousel.py video/carousels/<slug> --platforms telegram,x,linkedin,facebook,instagram
|
||
|
||
Тексты — <slug>/post.json: {"text": "...", "x": ["твит 1", "твит 2"], "alt": ["alt слайда 1", ...]}
|
||
Креды — те же, что у /root/coop/social/publish.py. Картинки для Buffer должны быть по публичному URL:
|
||
скрипт кладёт JPG в /srv/raw.cil.by/simbiot/carousels/<slug>/.
|
||
Результаты пишутся в <slug>/published.json; уже опубликованные площадки повторно не постятся.
|
||
"""
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
|
||
sys.path.insert(0, "/root/coop/social")
|
||
import publish # noqa: E402 load_env, LI_VERSION
|
||
|
||
PUBLIC_DIR, PUBLIC_URL = "/srv/raw.cil.by/simbiot/carousels", "https://raw.cil.by/simbiot/carousels"
|
||
|
||
|
||
def jpgs(d, slug):
|
||
out_dir = os.path.join(PUBLIC_DIR, slug)
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
files = []
|
||
for f in sorted(os.listdir(os.path.join(d, "slides"))):
|
||
if f.startswith("slide-") and f.endswith(".png"):
|
||
dst = os.path.join(out_dir, f.replace(".png", ".jpg"))
|
||
subprocess.run(["ffmpeg", "-nostdin", "-loglevel", "error", "-y", "-i", os.path.join(d, "slides", f), "-q:v", "2", dst], check=True)
|
||
files.append(dst)
|
||
return files
|
||
|
||
|
||
def multipart(fields, files):
|
||
b = uuid.uuid4().hex
|
||
parts = [f"--{b}\r\nContent-Disposition: form-data; name=\"{k}\"\r\n\r\n{v}\r\n".encode() for k, v in fields.items()]
|
||
for name, (fname, data, ctype) in files.items():
|
||
parts.append(f"--{b}\r\nContent-Disposition: form-data; name=\"{name}\"; filename=\"{fname}\"\r\nContent-Type: {ctype}\r\n\r\n".encode() + data + b"\r\n")
|
||
parts.append(f"--{b}--\r\n".encode())
|
||
return b"".join(parts), f"multipart/form-data; boundary={b}"
|
||
|
||
|
||
# ── X: тред, по 4 картинки на твит ──
|
||
def x_thread(env, tweets, images):
|
||
q = lambda s: urllib.parse.quote(str(s), safe="")
|
||
|
||
def auth(method, url):
|
||
p = {"oauth_consumer_key": env["X_CONSUMER_KEY"], "oauth_nonce": uuid.uuid4().hex, "oauth_signature_method": "HMAC-SHA1",
|
||
"oauth_timestamp": str(int(time.time())), "oauth_token": env["X_ACCESS_TOKEN"], "oauth_version": "1.0"}
|
||
base = "&".join(f"{q(k)}={q(v)}" for k, v in sorted(p.items()))
|
||
key = f"{q(env['X_CONSUMER_SECRET'])}&{q(env['X_ACCESS_TOKEN_SECRET'])}"
|
||
p["oauth_signature"] = base64.b64encode(hmac.new(key.encode(), f"{method}&{q(url)}&{q(base)}".encode(), hashlib.sha1).digest()).decode()
|
||
return "OAuth " + ", ".join(f'{k}="{q(v)}"' for k, v in p.items())
|
||
|
||
UP = "https://upload.twitter.com/1.1/media/upload.json"
|
||
ids, prev = [], None
|
||
for k, text in enumerate(tweets):
|
||
media = []
|
||
for img in images[4 * k:4 * k + 4]:
|
||
body, ctype = multipart({}, {"media": (os.path.basename(img), open(img, "rb").read(), "image/jpeg")})
|
||
r = urllib.request.urlopen(urllib.request.Request(UP, data=body, method="POST",
|
||
headers={"Authorization": auth("POST", UP), "Content-Type": ctype}), timeout=120)
|
||
media.append(json.loads(r.read())["media_id_string"])
|
||
payload = {"text": text[:280]}
|
||
if media:
|
||
payload["media"] = {"media_ids": media}
|
||
if prev:
|
||
payload["reply"] = {"in_reply_to_tweet_id": prev}
|
||
url = "https://api.twitter.com/2/tweets"
|
||
r = urllib.request.urlopen(urllib.request.Request(url, data=json.dumps(payload).encode(), method="POST",
|
||
headers={"Authorization": auth("POST", url), "Content-Type": "application/json"}), timeout=60)
|
||
prev = json.loads(r.read())["data"]["id"]
|
||
ids.append(prev)
|
||
return f"https://x.com/alexshkor/status/{ids[0]}"
|
||
|
||
|
||
# ── Telegram: альбом (sendMediaGroup, подпись у первого фото) ──
|
||
def telegram_album(env, text, images):
|
||
api = f"https://api.telegram.org/bot{env['TG_PUBLISH_BOT_TOKEN']}"
|
||
media, files = [], {}
|
||
for k, img in enumerate(images[:10]):
|
||
item = {"type": "photo", "media": f"attach://p{k}"}
|
||
if k == 0:
|
||
item["caption"] = text[:1024]
|
||
media.append(item)
|
||
files[f"p{k}"] = (f"p{k}.jpg", open(img, "rb").read(), "image/jpeg")
|
||
body, ctype = multipart({"chat_id": env["TG_PUBLISH_CHANNEL"], "media": json.dumps(media)}, files)
|
||
resp = json.loads(urllib.request.urlopen(urllib.request.Request(f"{api}/sendMediaGroup", data=body, method="POST",
|
||
headers={"Content-Type": ctype}), timeout=180).read())
|
||
if not resp.get("ok"):
|
||
raise RuntimeError(resp)
|
||
ch = env["TG_PUBLISH_CHANNEL"].lstrip("@")
|
||
return f"https://t.me/{ch}/{resp['result'][0]['message_id']}"
|
||
|
||
|
||
# ── LinkedIn: multiImage ──
|
||
def linkedin_multi(env, text, images, alts):
|
||
tok, urn = env["LINKEDIN_ACCESS_TOKEN"], env["LINKEDIN_PERSON_URN"]
|
||
H = {"Authorization": f"Bearer {tok}", "X-Restli-Protocol-Version": "2.0.0", "LinkedIn-Version": publish.LI_VERSION,
|
||
"Content-Type": "application/json"}
|
||
urns = []
|
||
for img in images[:20]:
|
||
init = json.loads(urllib.request.urlopen(urllib.request.Request(
|
||
"https://api.linkedin.com/rest/images?action=initializeUpload",
|
||
data=json.dumps({"initializeUploadRequest": {"owner": urn}}).encode(), method="POST", headers=H), timeout=30).read())["value"]
|
||
urllib.request.urlopen(urllib.request.Request(init["uploadUrl"], data=open(img, "rb").read(), method="PUT",
|
||
headers={"Authorization": f"Bearer {tok}", "Content-Type": "image/jpeg"}), timeout=120)
|
||
urns.append(init["image"])
|
||
time.sleep(5)
|
||
payload = {"author": urn, "commentary": text, "visibility": "PUBLIC",
|
||
"distribution": {"feedDistribution": "MAIN_FEED", "targetEntities": [], "thirdPartyDistributionChannels": []},
|
||
"content": {"multiImage": {"images": [{"id": u, "altText": (alts[k] if k < len(alts) else "")[:120]} for k, u in enumerate(urns)]}},
|
||
"lifecycleState": "PUBLISHED", "isReshareDisabledByAuthor": False}
|
||
r = urllib.request.urlopen(urllib.request.Request("https://api.linkedin.com/rest/posts", data=json.dumps(payload).encode(),
|
||
method="POST", headers=H), timeout=60)
|
||
pid = r.headers.get("x-restli-id", "")
|
||
return f"https://www.linkedin.com/feed/update/{pid}/" if pid else "posted"
|
||
|
||
|
||
# ── Facebook / Instagram через Buffer ──
|
||
def buffer_images(env, service, text, urls, alts):
|
||
key, org = env["BUFFER_ACCESS_TOKEN"], env.get("BUFFER_ORG_ID", "6151d52eea195d0f412ff493")
|
||
|
||
def gql(qs, v=None):
|
||
req = urllib.request.Request("https://api.buffer.com", data=json.dumps({"query": qs, "variables": v or {}}).encode(),
|
||
headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"})
|
||
return json.loads(urllib.request.urlopen(req, timeout=90).read())
|
||
|
||
chans = gql('query($o: OrganizationId!){ channels(input:{organizationId:$o}){ id service isDisconnected isLocked } }', {"o": org})["data"]["channels"]
|
||
ch = next(c for c in chans if c["service"] == service and not c["isDisconnected"] and not c["isLocked"])
|
||
meta = {"facebook": {"facebook": {"type": "post"}}, "instagram": {"instagram": {"type": "post", "shouldShareToFeed": True}}}[service]
|
||
assets = [{"image": {"url": u, "metadata": {"altText": (alts[k] if k < len(alts) else "slide")[:120]}}} for k, u in enumerate(urls[:10])]
|
||
inp = {"channelId": ch["id"], "text": text, "assets": assets, "mode": "shareNow", "schedulingType": "automatic",
|
||
"needsApproval": False, "metadata": meta, "source": "simbiot-carousel"}
|
||
r = gql("mutation($input: CreatePostInput!){ createPost(input:$input){ __typename ... on PostActionSuccess { post { id } } ... on MutationError { message } } }",
|
||
{"input": inp})["data"]["createPost"]
|
||
if r["__typename"] != "PostActionSuccess":
|
||
raise RuntimeError(r.get("message"))
|
||
pid = r["post"]["id"]
|
||
for _ in range(60):
|
||
p = gql("query($id: PostId!){ post(input:{id:$id}){ status error { message } externalLink } }", {"id": pid})["data"]["post"]
|
||
if p["status"] == "sent":
|
||
return p.get("externalLink") or f"sent {pid}"
|
||
if p.get("error"):
|
||
raise RuntimeError(p["error"])
|
||
time.sleep(10)
|
||
return f"pending {pid}"
|
||
|
||
|
||
def main():
|
||
d = os.path.abspath(sys.argv[1])
|
||
slug = os.path.basename(d)
|
||
plats = sys.argv[sys.argv.index("--platforms") + 1].split(",") if "--platforms" in sys.argv else \
|
||
["telegram", "x", "linkedin", "facebook", "instagram"]
|
||
post = json.load(open(os.path.join(d, "post.json")))
|
||
env = publish.load_env()
|
||
images = jpgs(d, slug)
|
||
urls = [f"{PUBLIC_URL}/{slug}/{os.path.basename(p)}" for p in images]
|
||
done_path = os.path.join(d, "published.json")
|
||
done = json.load(open(done_path)) if os.path.exists(done_path) else {}
|
||
alts = post.get("alt", [])
|
||
for p in plats:
|
||
if str(done.get(p, "")).startswith("http"):
|
||
print(p, "already:", done[p])
|
||
continue
|
||
try:
|
||
if p == "telegram":
|
||
done[p] = telegram_album(env, post["text"], images)
|
||
elif p == "x":
|
||
done[p] = x_thread(env, post["x"], images)
|
||
elif p == "linkedin":
|
||
done[p] = linkedin_multi(env, post["text"], images, alts)
|
||
elif p in ("facebook", "instagram"):
|
||
done[p] = buffer_images(env, p, post["text"], urls, alts)
|
||
except urllib.error.HTTPError as e:
|
||
done[p] = f"ERROR {e.code}: {e.read()[:300]!r}"
|
||
except Exception as e:
|
||
done[p] = f"ERROR: {str(e)[:300]}"
|
||
print(p, "→", done[p], flush=True)
|
||
json.dump(done, open(done_path, "w"), ensure_ascii=False, indent=1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|