# -*- coding: utf-8 -*- """ ربات ساده‌ی ساخت پست برای کانال. ادمین محتوا (متن/عکس/ویدیو) را می‌فرستد، اختیاری یک دکمه (متن + لینک + رنگ + آیکون پریمیوم) برایش می‌سازد، پیش‌نمایش می‌بیند و تایید می‌کند؛ با copyMessage به کانال می‌رود (بدون برچسب «Forwarded from»، چون از forwardMessage استفاده نشده). ایموجی‌های پریمیومِ داخل متن/کپشن اصلی هم چون خودِ پیام کپی می‌شود، دست‌نخورده و سالم منتقل می‌شوند. """ import hmac import logging from flask import Flask, request, jsonify import config import database as db import telegram_api as tg logging.basicConfig(level=logging.INFO) logger = logging.getLogger("channel_poster_bot") app = Flask(__name__) db.init_db() def is_admin(user_id): return user_id in config.ADMIN_IDS # --------------------------------------------------------------------------- # کیبوردها # --------------------------------------------------------------------------- def button_decision_keyboard(): return {"inline_keyboard": [[ {"text": "➕ بله، دکمه اضافه کن", "callback_data": "btn_yes"}, {"text": "بدون دکمه", "callback_data": "btn_no"}, ]]} def style_keyboard(): rows = [] row = [] for i, (name, _style) in enumerate(config.BUTTON_STYLES): row.append({"text": name, "callback_data": f"style:{i}"}) if len(row) == 3: rows.append(row) row = [] if row: rows.append(row) return {"inline_keyboard": rows} def confirm_keyboard(): return {"inline_keyboard": [ [{"text": "✅ ارسال به همین کانال", "callback_data": "confirm"}], [{"text": "🔀 عوض‌کردن کانال", "callback_data": "change_channel"}], [{"text": "✏️ شروع دوباره", "callback_data": "restart"}, {"text": "❌ لغو", "callback_data": "cancel"}], ]} def channel_choice_keyboard(channels): rows = [] for ch in channels: label = ch["title"] or ch["username"] or str(ch["chat_id"]) rows.append([{"text": f"📣 {label}", "callback_data": f"pick_channel:{ch['chat_id']}"}]) rows.append([{"text": "❌ لغو", "callback_data": "cancel"}]) return {"inline_keyboard": rows} def build_post_reply_markup(draft, include_icon=True): if not draft.get("button_url"): return None button = {"text": draft["button_text"], "url": draft["button_url"]} if draft.get("button_style"): button["style"] = draft["button_style"] if include_icon and draft.get("button_icon_emoji_id"): button["icon_custom_emoji_id"] = draft["button_icon_emoji_id"] return {"inline_keyboard": [[button]]} # --------------------------------------------------------------------------- # کمکی: استخراج آیکون پریمیوم از ابتدای متنی که ادمین برای دکمه فرستاده # --------------------------------------------------------------------------- def _utf16_len(s): return len(s.encode("utf-16-le")) // 2 def extract_leading_custom_emoji(text, entities): """اگر اولین کاراکتر پیام یک ایموجی پریمیوم (custom_emoji) بود، آیدی‌اش را برمی‌گرداند و همان کاراکتر را از ابتدای متن حذف می‌کند (برای این‌که فقط به‌عنوان آیکون دکمه استفاده شود، نه بخشی از متن دکمه، چون متن دکمه‌ی تلگرام خودش نمی‌تواند ایموجی پریمیوم داشته باشد).""" for e in entities or []: if e.get("type") == "custom_emoji" and e.get("offset") == 0: end_units = e["length"] * 2 utf16 = text.encode("utf-16-le") prefix = utf16[:end_units].decode("utf-16-le", errors="ignore") remaining = text[len(prefix):].strip() return e["custom_emoji_id"], remaining return None, text.strip() CONTENT_KIND_FIELDS = ["photo", "video", "animation", "document", "audio", "voice"] def detect_content_kind(message): for field in CONTENT_KIND_FIELDS: if field in message: return field if "text" in message: return "text" return None # --------------------------------------------------------------------------- # مراحل ساخت پست # --------------------------------------------------------------------------- def start_new_post(chat_id, admin_id): db.reset_draft(admin_id, step="awaiting_content") tg.send_message( chat_id, "📝 محتوای پست را بفرست: متن، عکس یا ویدیو (می‌تونی کپشن هم داشته باشه).\n" "همون چیزی که بفرستی عیناً (با همون ایموجی‌های پریمیوم داخلش) توی کانال منتشر می‌شه.\n\n" "هر موقع خواستی لغو کنی، /cancel رو بفرست.", ) def handle_content_received(chat_id, admin_id, message): kind = detect_content_kind(message) if not kind: tg.send_message(chat_id, "این نوع پیام پشتیبانی نمی‌شود. متن، عکس یا ویدیو بفرست.") return db.update_draft( admin_id, content_chat_id=chat_id, content_message_id=message["message_id"], content_kind=kind, step="awaiting_button_decision", ) tg.send_message(chat_id, "محتوا دریافت شد ✅\n\nمی‌خوای زیرش یک دکمه هم اضافه کنی؟", reply_markup=button_decision_keyboard()) def handle_button_text_received(chat_id, admin_id, message): text = message.get("text") if not text: tg.send_message(chat_id, "متن دکمه را به‌صورت پیام متنی بفرست.") return icon_id, label = extract_leading_custom_emoji(text, message.get("entities")) if not label: tg.send_message(chat_id, "متن دکمه نمی‌تواند خالی باشد (فقط ایموجی کافی نیست). دوباره بفرست.") return db.update_draft(admin_id, button_text=label, button_icon_emoji_id=icon_id, step="awaiting_button_url") note = " (آیکون پریمیومش هم ثبت شد ✅)" if icon_id else "" tg.send_message(chat_id, f"متن دکمه ثبت شد{note}.\n\nحالا لینک دکمه را بفرست (باید با http:// یا https:// شروع شود):") def handle_button_url_received(chat_id, admin_id, message): text = (message.get("text") or "").strip() if not (text.startswith("http://") or text.startswith("https://") or text.startswith("tg://")): tg.send_message(chat_id, "لینک نامعتبر است؛ باید با http:// یا https:// شروع شود. دوباره بفرست.") return db.update_draft(admin_id, button_url=text, step="awaiting_button_style") tg.send_message( chat_id, "لینک ثبت شد ✅\n\nرنگ دکمه رو انتخاب کن:", reply_markup=style_keyboard(), ) def start_channel_selection(chat_id, admin_id): channels = db.list_active_channels() if not channels: db.update_draft(admin_id, step="awaiting_channel") tg.send_message( chat_id, "📣 هنوز هیچ کانالی برای این ربات ثبت نشده.\n\n" "برای ثبت یک کانال:\n" "1) ربات را در آن کانال ادمین کن (اگر تازه ادمینش کردی، خودکار ثبت می‌شود).\n" "2) اگر قبل از راه‌اندازی ربات ادمینش کرده بودی، یک پیام از همان کانال را همین‌جا برای من فوروارد کن.", ) return db.update_draft(admin_id, step="awaiting_channel") tg.send_message(chat_id, "📣 این پست را در کدام کانال منتشر کنم؟", reply_markup=channel_choice_keyboard(channels)) def show_preview(chat_id, admin_id): draft = db.get_draft(admin_id) reply_markup = build_post_reply_markup(draft, include_icon=True) resp = tg.copy_message(chat_id, draft["content_chat_id"], draft["content_message_id"], reply_markup=reply_markup) if not resp or not resp.get("ok"): tg.send_message(chat_id, "⚠️ ساخت پیش‌نمایش با خطا مواجه شد. دوباره تلاش کن یا /newpost بزن.") logger.error("copy_message preview failed: %s", resp) return db.update_draft(admin_id, step="preview") note = "" if draft.get("button_icon_emoji_id"): note = ( "\n\n⚠️ توجه: آیکون پریمیومِ دکمه طبق قوانین تلگرام فقط توی چت خصوصی/گروه تضمین‌شده است. " "توی این پیش‌نمایش (که به همین چت خصوصی میاد) آیکون رو می‌بینی، ولی وقتی پست واقعاً توی " "کانال منتشر بشه ممکنه تلگرام آیکون رو نپذیرد؛ اگر این‌طور شد، پست بدون آیکون (ولی با همون " "رنگ و متن) منتشر می‌شود و به‌جای گیرکردن، خودم به‌صورت خودکار انجامش می‌دم." ) channel = db.get_channel(draft["channel_id"]) if draft.get("channel_id") else None channel_label = (channel["title"] or channel["username"]) if channel else "؟" tg.send_message( chat_id, f"☝️ این پیش‌نمایش دقیق چیزیه که (به‌جز نکته‌ی زیر، اگه بود) توی کانال «{channel_label}» منتشر می‌شه. " "اگه اوکیه بزن «ارسال»:" + note, reply_markup=confirm_keyboard(), ) def publish_to_channel(chat_id, admin_id): draft = db.get_draft(admin_id) if not draft or draft.get("step") != "preview" or not draft.get("channel_id"): tg.send_message(chat_id, "پست فعالی برای ارسال پیدا نشد. /newpost بزن.") return target = draft["channel_id"] reply_markup = build_post_reply_markup(draft, include_icon=True) resp = tg.copy_message(target, draft["content_chat_id"], draft["content_message_id"], reply_markup=reply_markup) icon_dropped = False if (not resp or not resp.get("ok")) and draft.get("button_icon_emoji_id"): # تلگرام آیکون پریمیوم را برای پست کانال نپذیرفت (طبق مستندات، این آیکون فقط توی چت # خصوصی/گروه یا با یوزرنیم خریداری‌شده از Fragment تضمین‌شده)؛ بدون آیکون دوباره تلاش می‌کنیم. icon_dropped = True reply_markup = build_post_reply_markup(draft, include_icon=False) resp = tg.copy_message(target, draft["content_chat_id"], draft["content_message_id"], reply_markup=reply_markup) if resp and resp.get("ok"): msg = "✅ پست با موفقیت منتشر شد." if icon_dropped: msg += "\n(⚠️ آیکون پریمیومِ دکمه توی کانال پذیرفته نشد و بدون آیکون منتشر شد؛ رنگ و متن دکمه سرجاشونن.)" tg.send_message(chat_id, msg) db.reset_draft(admin_id, step="idle") else: tg.send_message( chat_id, "❌ ارسال به کانال ناموفق بود. مطمئن شو ربات هنوز توی اون کانال ادمین (با دسترسی ارسال پیام) است.\n" f"جزئیات خطا: {resp}", ) logger.error("publish_to_channel failed: %s", resp) def process_my_chat_member(update): """وقتی وضعیت عضویت ربات در یک کانال عوض می‌شود (ادمین شد/برداشته شد)، تلگرام این آپدیت را می‌فرستد؛ از همینجا لیست کانال‌ها را خودکار به‌روز نگه می‌داریم.""" chat = update.get("chat") or {} if chat.get("type") != "channel": return new_status = (update.get("new_chat_member") or {}).get("status") if new_status == "administrator": db.upsert_channel(chat["id"], chat.get("title"), chat.get("username")) elif new_status in ("left", "kicked", "member", "restricted"): db.deactivate_channel(chat["id"]) # --------------------------------------------------------------------------- # دیسپچر # --------------------------------------------------------------------------- def process_message(message): from_user = message.get("from") or {} chat_id = message["chat"]["id"] admin_id = from_user.get("id") if admin_id is None: return if not is_admin(admin_id): return # کاملاً بی‌صدا؛ غیرادمین حتی نمی‌فهمد ربات جواب می‌دهد یا نه text = message.get("text") if text and text.startswith("/start"): tg.send_message( chat_id, "سلام 👋 با این ربات می‌تونی پست (متن/عکس/ویدیو + دکمه اختیاری) رو مستقیم توی کانالت منتشر کنی.\n\n" "برای شروع /newpost رو بفرست.", ) return if text and text.startswith("/newpost"): start_new_post(chat_id, admin_id) return if text and text.startswith("/cancel"): db.reset_draft(admin_id, step="idle") tg.send_message(chat_id, "لغو شد. برای شروع دوباره /newpost بزن.") return if text and text.startswith("/channels"): channels = db.list_active_channels() if not channels: tg.send_message( chat_id, "هنوز هیچ کانالی ثبت نشده. ربات را در یک کانال ادمین کن، یا اگر از قبل ادمین بوده، " "یک پیام از همان کانال را همین‌جا فوروارد کن.", ) else: lines = [f"• {c['title'] or c['username'] or c['chat_id']}" for c in channels] tg.send_message(chat_id, "📣 کانال‌های ثبت‌شده:\n\n" + "\n".join(lines)) return # ثبت کانال با فوروارد یک پیام از آن کانال (برای کانال‌هایی که قبل از راه‌اندازی ربات ادمین شده بودند) forward_chat = message.get("forward_from_chat") if forward_chat and forward_chat.get("type") == "channel": db.upsert_channel(forward_chat["id"], forward_chat.get("title"), forward_chat.get("username")) tg.send_message(chat_id, f"✅ کانال «{forward_chat.get('title')}» ثبت شد؛ حالا موقع ساخت پست توی لیست میاد.") return draft = db.get_draft(admin_id) step = draft["step"] if draft else "idle" if step == "awaiting_content": handle_content_received(chat_id, admin_id, message) return if step == "awaiting_button_text": handle_button_text_received(chat_id, admin_id, message) return if step == "awaiting_button_url": handle_button_url_received(chat_id, admin_id, message) return tg.send_message(chat_id, "برای ساخت پست جدید /newpost رو بفرست.") def process_callback_query(callback_query): from_user = callback_query["from"] admin_id = from_user["id"] chat_id = callback_query["message"]["chat"]["id"] data = callback_query.get("data", "") if not is_admin(admin_id): tg.answer_callback_query(callback_query["id"]) # فقط اسپینر را می‌بندد، چیزی فاش نمی‌کند return draft = db.get_draft(admin_id) step = draft["step"] if draft else "idle" if data == "btn_yes" and step == "awaiting_button_decision": db.update_draft(admin_id, step="awaiting_button_text") tg.answer_callback_query(callback_query["id"]) tg.send_message( chat_id, "متن دکمه رو بفرست. اگه می‌خوای آیکون پریمیوم هم داشته باشه، همون ایموجی پریمیوم رو " "اول متن بچسبون (مثلاً: «💎 خرید کن»).\n\n" "⚠️ نکته: طبق قوانین تلگرام، آیکون پریمیومِ دکمه توی چت خصوصی/گروه همیشه کار می‌کنه، ولی " "توی پستِ کانال فقط اگه ربات یوزرنیم خریداری‌شده از Fragment داشته باشه تضمین‌شده‌ست؛ اگه " "نداشته باشه، ممکنه موقع انتشار در کانال خودکار حذف بشه (متن و رنگ دکمه بدون تغییر می‌مونن).", ) return if data == "btn_no" and step == "awaiting_button_decision": db.update_draft(admin_id, button_text=None, button_url=None, button_icon_emoji_id=None, button_style=None) tg.answer_callback_query(callback_query["id"]) start_channel_selection(chat_id, admin_id) return if data.startswith("style:") and step == "awaiting_button_style": idx = int(data.split(":", 1)[1]) name, style = config.BUTTON_STYLES[idx] db.update_draft(admin_id, button_style=style) tg.answer_callback_query(callback_query["id"], f"رنگ: {name}") start_channel_selection(chat_id, admin_id) return if data.startswith("pick_channel:") and step == "awaiting_channel": target = int(data.split(":", 1)[1]) channel = db.get_channel(target) if not channel or not channel["active"]: tg.answer_callback_query(callback_query["id"], "این کانال دیگر در دسترس نیست.", show_alert=True) return db.update_draft(admin_id, channel_id=target) tg.answer_callback_query(callback_query["id"], f"کانال: {channel['title'] or channel['username']}") show_preview(chat_id, admin_id) return if data == "change_channel" and step == "preview": tg.answer_callback_query(callback_query["id"]) start_channel_selection(chat_id, admin_id) return if data == "confirm" and step == "preview": tg.answer_callback_query(callback_query["id"]) publish_to_channel(chat_id, admin_id) return if data == "restart": tg.answer_callback_query(callback_query["id"]) start_new_post(chat_id, admin_id) return if data == "cancel": db.reset_draft(admin_id, step="idle") tg.answer_callback_query(callback_query["id"], "لغو شد.") tg.send_message(chat_id, "لغو شد. برای شروع دوباره /newpost بزن.") return tg.answer_callback_query(callback_query["id"]) # --------------------------------------------------------------------------- # مسیرهای Flask # --------------------------------------------------------------------------- def _setup_key_ok(): return hmac.compare_digest(request.args.get("key", ""), config.SETUP_KEY) @app.route(config.WEBHOOK_PATH, methods=["POST"]) def webhook(): incoming_secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") if not hmac.compare_digest(incoming_secret, config.WEBHOOK_SECRET): return jsonify({"ok": False, "error": "forbidden"}), 403 update = request.get_json(silent=True) or {} try: if "message" in update: process_message(update["message"]) elif "callback_query" in update: process_callback_query(update["callback_query"]) elif "my_chat_member" in update: process_my_chat_member(update["my_chat_member"]) except Exception: logger.exception("خطا در پردازش آپدیت") return jsonify({"ok": True}) @app.route("/set_webhook", methods=["GET"]) def route_set_webhook(): if not _setup_key_ok(): return jsonify({"ok": False, "error": "forbidden"}), 403 url = config.BASE_URL.rstrip("/") + config.WEBHOOK_PATH result = tg.set_webhook(url, secret_token=config.WEBHOOK_SECRET) return jsonify(result) @app.route("/set_commands", methods=["GET"]) def route_set_commands(): if not _setup_key_ok(): return jsonify({"ok": False, "error": "forbidden"}), 403 commands = [ {"command": "start", "description": "شروع مجدد"}, {"command": "newpost", "description": "ساخت پست جدید"}, {"command": "channels", "description": "لیست کانال‌های ثبت‌شده"}, {"command": "cancel", "description": "لغو پست در حال ساخت"}, ] results = {} for admin_id in config.ADMIN_IDS: results[str(admin_id)] = tg.set_my_commands(commands, chat_id=admin_id) return jsonify({"admin": results}) @app.route("/", methods=["GET"]) def index(): return jsonify({"ok": True, "service": "channel-poster-bot"}) if __name__ == "__main__": app.run(port=5000, debug=True)