""" Widget Telega/MAX — мультиарендный сервис виджетов Telegram-каналов (MVP). Сущности: - Виджет (widget): { id, channel, config } — один виджет = один канал + свои настройки. - Посты хранятся по каналам (таблица posts, ключ channel+id). - Кабинет защищён паролем (ADMIN_PASSWORD); чтение постов/конфига — публичное (нужно виджету на сайте), управление — только с токеном. Один процесс: 1) BACKFILL — при старте и при создании виджета берёт посты из t.me/s/<канал>. 2) LIVE — бот-администратор ловит новые публикации любого подключённого канала. 3) API — отдаёт посты и конфиг виджету, управляет виджетами из кабинета. + /img — прокси картинок Telegram CDN (обход hotlink 403). Запуск: python app.py (см. README.md) """ import asyncio import os import re import json import time import secrets import sqlite3 import html import urllib.parse from datetime import datetime, timezone import aiohttp from aiohttp import web from bs4 import BeautifulSoup from aiogram import Bot, Dispatcher from aiogram.types import Message # ---------- Конфиг сервиса ---------- BOT_TOKEN = os.getenv("BOT_TOKEN", "") SEED_CHANNEL = os.getenv("CHANNEL", "furgonlab").lstrip("@") # канал первого виджета при первом старте ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin") # пароль входа в кабинет PORT = int(os.getenv("PORT", "8000")) MAX_POSTS = int(os.getenv("MAX_POSTS", "30")) DB_PATH = os.getenv("DB_PATH", "posts.db") WIDGETS_PATH = os.getenv("WIDGETS_PATH", "widgets.json") MEDIA_DIR = os.getenv("MEDIA_DIR", "media") os.makedirs(MEDIA_DIR, exist_ok=True) DEFAULT_CONFIG = { "title": "Наши работы", "description": "", "layout": "grid", "cols": 4, "startRows": 1, "maxRows": 3, "limit": 9, "accent": "#229ED9", "radius": 14, "showText": True, "showDate": True, "badge": True, } # ---------- Хранилище виджетов (JSON) ---------- def load_widgets() -> dict: try: with open(WIDGETS_PATH, encoding="utf-8") as f: return json.load(f) except Exception: return {} def save_widgets(d: dict): with open(WIDGETS_PATH, "w", encoding="utf-8") as f: json.dump(d, f, ensure_ascii=False, indent=2) def new_widget_id() -> str: return "wgt_" + secrets.token_hex(4) def widget_public(w: dict) -> dict: cfg = dict(DEFAULT_CONFIG); cfg.update(w.get("config", {})) return {"id": w["id"], "channel": w["channel"], "config": cfg, "created": w.get("created", "")} def create_widget(channel: str) -> dict: channel = (channel or "").strip().lstrip("@") wid = new_widget_id() w = {"id": wid, "channel": channel, "config": dict(DEFAULT_CONFIG), "created": datetime.now(timezone.utc).isoformat()} d = load_widgets(); d[wid] = w; save_widgets(d) return w def get_widget(wid: str): return load_widgets().get(wid) def first_widget(): d = load_widgets() return next(iter(d.values()), None) if d else None def delete_widget(wid: str): d = load_widgets(); d.pop(wid, None); save_widgets(d) def update_widget(wid: str, channel=None, config_patch=None): d = load_widgets(); w = d.get(wid) if not w: return None if channel is not None: w["channel"] = channel.strip().lstrip("@") if config_patch is not None: cfg = dict(DEFAULT_CONFIG); cfg.update(w.get("config", {})) for k in DEFAULT_CONFIG: if k in config_patch: cfg[k] = config_patch[k] w["config"] = cfg save_widgets(d) return w def all_channels() -> set: return {w["channel"] for w in load_widgets().values() if w.get("channel")} # ---------- Хранилище постов (SQLite, по каналам) ---------- def db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db(): with db() as c: cols = [r["name"] for r in c.execute("PRAGMA table_info(posts)").fetchall()] if cols and "channel" not in cols: # старая одноканальная схема — пересоздаём c.execute("DROP TABLE posts") c.execute("""CREATE TABLE IF NOT EXISTS posts( channel TEXT, id INTEGER, date TEXT, text TEXT, media TEXT, link TEXT, PRIMARY KEY(channel, id))""") def upsert_post(channel: str, p: dict): row = {"channel": channel, "id": p["id"], "date": p["date"], "text": p["text"], "media": json.dumps(p["media"], ensure_ascii=False), "link": p["link"]} with db() as c: c.execute("""INSERT INTO posts(channel,id,date,text,media,link) VALUES(:channel,:id,:date,:text,:media,:link) ON CONFLICT(channel,id) DO UPDATE SET date=excluded.date, text=excluded.text, media=excluded.media, link=excluded.link""", row) c.execute("""DELETE FROM posts WHERE channel=? AND id NOT IN (SELECT id FROM posts WHERE channel=? ORDER BY id DESC LIMIT ?)""", (channel, channel, MAX_POSTS)) def get_posts(channel: str, limit: int): with db() as c: rows = c.execute("SELECT * FROM posts WHERE channel=? ORDER BY id DESC LIMIT ?", (channel, limit)).fetchall() out = [] for r in rows: d = dict(r) try: d["media"] = json.loads(d["media"]) if d["media"] else [] except Exception: d["media"] = [] out.append(d) return out # ---------- Разбор текста/медиа ---------- def clean(txt: str) -> str: txt = html.unescape(txt or "") txt = re.sub(r"[ \t]+\n", "\n", txt) txt = re.sub(r"\n{3,}", "\n\n", txt) return txt.strip() def extract_text(el) -> str: if not el: return "" for br in el.find_all("br"): br.replace_with("\n") return clean(el.get_text("")) def bg_url(style: str) -> str: m = re.search(r"url\(['\"]?(.*?)['\"]?\)", style or "") return m.group(1) if m else "" # ---------- 1) BACKFILL ---------- def _parse_message(channel, m): """Разбирает один блок сообщения t.me/s. Возвращает (mid, dict|None).""" post_attr = m.get("data-post", "") if "/" not in post_attr: return None, None mid = int(post_attr.split("/")[-1]) text = extract_text(m.select_one(".tgme_widget_message_text")) media = [] for ph in m.select("a.tgme_widget_message_photo_wrap"): u = bg_url(ph.get("style", "")) if u: media.append({"type": "photo", "url": u}) for vp in m.select(".tgme_widget_message_video_player"): vid = vp.select_one("video") src = vid.get("src") if vid else "" thumb = vp.select_one(".tgme_widget_message_video_thumb") pst = bg_url(thumb.get("style", "")) if thumb else "" if src: media.append({"type": "video", "url": src, "poster": pst}) elif pst: media.append({"type": "video", "url": "", "poster": pst}) t_el = m.select_one("time") date = t_el["datetime"] if t_el and t_el.has_attr("datetime") \ else datetime.now(timezone.utc).isoformat() return mid, {"id": mid, "date": date, "text": text, "media": media, "link": f"https://t.me/{channel}/{mid}"} async def backfill(channel: str): # t.me/s отдаёт ~страницу постов; альбом = 1 пост, поэтому листаем через ?before=, # пока не наберём MAX_POSTS или страницы не закончатся. seen = set() before = None try: async with aiohttp.ClientSession(headers={"User-Agent": "Mozilla/5.0"}) as s: for _ in range(8): # предохранитель по числу страниц url = f"https://t.me/s/{channel}" + (f"?before={before}" if before else "") async with s.get(url, timeout=20) as r: page = await r.text() blocks = BeautifulSoup(page, "html.parser").select("div.tgme_widget_message") mids = [] for m in blocks: mid, post = _parse_message(channel, m) if mid is None or mid in seen: continue seen.add(mid); mids.append(mid) upsert_post(channel, post) if not mids: # новых постов не пришло — конец break if len(seen) >= MAX_POSTS: break before = min(mids) # следующая страница — старше самого раннего except Exception as e: print(f"[backfill:{channel}] пропущен ({e})") return print(f"[backfill:{channel}] загружено постов: {len(seen)}") async def backfill_all(): for ch in all_channels(): await backfill(ch) # ---------- 2) LIVE ---------- dp = Dispatcher() album_buffer = {} # media_group_id -> {ids, media, text, date, ts, channel} async def _download(bot: Bot, file_id: str, name: str) -> str: try: f = await bot.get_file(file_id) dest = os.path.join(MEDIA_DIR, name) await bot.download_file(f.file_path, destination=dest) return f"/media/{name}" except Exception as e: print(f"[live] не скачал {name}: {e}") return "" async def _extract_media(bot: Bot, message: Message) -> list: out = [] if message.photo: u = await _download(bot, message.photo[-1].file_id, f"{message.message_id}.jpg") if u: out.append({"type": "photo", "url": u}) if message.video: pst = "" if message.video.thumbnail: pst = await _download(bot, message.video.thumbnail.file_id, f"{message.message_id}_p.jpg") vurl = await _download(bot, message.video.file_id, f"{message.message_id}.mp4") out.append({"type": "video", "url": vurl, "poster": pst}) return out @dp.channel_post() async def on_channel_post(message: Message, bot: Bot): ch = (message.chat.username or "").lower() channels = {c.lower() for c in all_channels()} if ch not in channels: return text = clean(message.text or message.caption or "") media = await _extract_media(bot, message) date = (message.date or datetime.now(timezone.utc)).isoformat() gid = message.media_group_id if gid: g = album_buffer.setdefault(gid, {"ids": [], "media": [], "text": "", "date": date, "channel": ch}) g["ids"].append(message.message_id) g["media"].extend(media) if text: g["text"] = text g["ts"] = time.monotonic() return upsert_post(ch, {"id": message.message_id, "date": date, "text": text, "media": media, "link": f"https://t.me/{ch}/{message.message_id}"}) print(f"[live:{ch}] новый пост #{message.message_id}") async def album_flusher(): while True: await asyncio.sleep(1) now = time.monotonic() for gid, g in list(album_buffer.items()): if now - g.get("ts", now) > 2: album_buffer.pop(gid, None) mid = min(g["ids"]); ch = g["channel"] upsert_post(ch, {"id": mid, "date": g["date"], "text": g["text"], "media": g["media"], "link": f"https://t.me/{ch}/{mid}"}) print(f"[live:{ch}] альбом #{mid}: {len(g['media'])} медиа") # ---------- 3) HTTP API ---------- CORS = {"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Headers": "Content-Type, X-Admin-Token", "Access-Control-Allow-Methods": "GET, POST, OPTIONS"} def authed(request) -> bool: return request.headers.get("X-Admin-Token", "") == ADMIN_PASSWORD def resolve_channel(request): wid = request.query.get("wid", "") if wid: w = get_widget(wid) return w["channel"] if w else None ch = request.query.get("channel", "") if ch: return ch.lstrip("@") w = first_widget() # фолбэк для старого embed без wid return w["channel"] if w else None def resolve_config(request): wid = request.query.get("wid", "") w = get_widget(wid) if wid else first_widget() cfg = dict(DEFAULT_CONFIG) if w: cfg.update(w.get("config", {})) return cfg async def api_posts(request): ch = resolve_channel(request) if not ch: return web.json_response({"channel": None, "posts": []}, headers=CORS) limit = min(int(request.query.get("limit", MAX_POSTS)), MAX_POSTS) return web.json_response({"channel": ch, "source": "telegram", "posts": get_posts(ch, limit)}, headers=CORS) async def api_config_get(request): return web.json_response(resolve_config(request), headers=CORS) async def api_config_post(request): if not authed(request): return web.json_response({"error": "unauthorized"}, status=401, headers=CORS) wid = request.query.get("wid", "") try: patch = await request.json() except Exception: return web.json_response({"error": "bad json"}, status=400, headers=CORS) w = update_widget(wid, config_patch=patch or {}) if not w: return web.json_response({"error": "widget not found"}, status=404, headers=CORS) return web.json_response(widget_public(w), headers=CORS) async def api_widgets_get(request): if not authed(request): return web.json_response({"error": "unauthorized"}, status=401, headers=CORS) return web.json_response({"widgets": [widget_public(w) for w in load_widgets().values()]}, headers=CORS) async def api_widgets_create(request): if not authed(request): return web.json_response({"error": "unauthorized"}, status=401, headers=CORS) try: body = await request.json() except Exception: body = {} channel = (body.get("channel") or "").strip().lstrip("@") if not channel: return web.json_response({"error": "channel required"}, status=400, headers=CORS) w = create_widget(channel) asyncio.create_task(backfill(channel)) # сразу наполняем return web.json_response(widget_public(w), headers=CORS) async def api_widgets_update(request): if not authed(request): return web.json_response({"error": "unauthorized"}, status=401, headers=CORS) try: body = await request.json() except Exception: body = {} wid = body.get("wid", "") channel = body.get("channel") w = update_widget(wid, channel=channel) if not w: return web.json_response({"error": "widget not found"}, status=404, headers=CORS) if channel: asyncio.create_task(backfill(channel.strip().lstrip("@"))) return web.json_response(widget_public(w), headers=CORS) async def api_widgets_delete(request): if not authed(request): return web.json_response({"error": "unauthorized"}, status=401, headers=CORS) try: body = await request.json() except Exception: body = {} delete_widget(body.get("wid", "")) return web.json_response({"ok": True}, headers=CORS) async def api_login(request): try: body = await request.json() except Exception: body = {} if (body.get("password") or "") == ADMIN_PASSWORD: return web.json_response({"ok": True, "token": ADMIN_PASSWORD}, headers=CORS) return web.json_response({"ok": False}, status=401, headers=CORS) # Прокси медиа Telegram CDN (обход hotlink 403) ALLOWED_IMG_HOSTS = ("telesco.pe", "cdn-telegram.org", "telegram.org") async def api_img(request): u = request.query.get("u", "") host = (urllib.parse.urlparse(u).hostname or "").lower() if not (u.startswith("https://") and any(host == h or host.endswith("." + h) for h in ALLOWED_IMG_HOSTS)): return web.Response(status=400, text="bad url") try: async with aiohttp.ClientSession(headers={"User-Agent": "Mozilla/5.0"}) as s: async with s.get(u, timeout=30) as r: body = await r.read() ct = r.headers.get("Content-Type", "application/octet-stream").split(";")[0] except Exception as e: print(f"[img] ошибка: {e}") return web.Response(status=502, text="fetch failed") hdrs = {"Cache-Control": "public, max-age=86400"}; hdrs.update(CORS) return web.Response(body=body, content_type=ct, headers=hdrs) async def cors_preflight(request): return web.Response(status=204, headers=CORS) async def start_api(): app = web.Application() app.router.add_get("/api/posts", api_posts) app.router.add_get("/api/config", api_config_get) app.router.add_post("/api/config", api_config_post) app.router.add_get("/api/widgets", api_widgets_get) app.router.add_post("/api/widgets", api_widgets_create) app.router.add_post("/api/widgets/update", api_widgets_update) app.router.add_post("/api/widgets/delete", api_widgets_delete) app.router.add_post("/api/login", api_login) app.router.add_get("/img", api_img) app.router.add_route("OPTIONS", "/api/{tail:.*}", cors_preflight) app.router.add_static("/media/", MEDIA_DIR, show_index=False) app.router.add_static("/", ".", show_index=True) runner = web.AppRunner(app) await runner.setup() await web.TCPSite(runner, "0.0.0.0", PORT).start() print(f"[api] http://localhost:{PORT} (cabinet.html · demo.html)") # ---------- main ---------- async def main(): init_db() if not load_widgets() and SEED_CHANNEL: # первый запуск — заводим виджет из CHANNEL w = create_widget(SEED_CHANNEL) try: # перенос старого одиночного config.json, если был with open("config.json", encoding="utf-8") as f: update_widget(w["id"], config_patch=json.load(f)) print("[seed] перенёс config.json в первый виджет") except Exception: pass print(f"[seed] создан виджет для канала @{SEED_CHANNEL}") if not BOT_TOKEN: print("!! BOT_TOKEN пуст — без live-бота (только backfill + API).") print(f"[auth] пароль кабинета: ADMIN_PASSWORD (по умолчанию 'admin')") await start_api() asyncio.create_task(backfill_all()) if BOT_TOKEN: bot = Bot(BOT_TOKEN) asyncio.create_task(album_flusher()) print("[live] бот запущен, слушаю новые посты каналов…") await dp.start_polling(bot) else: while True: await asyncio.sleep(3600) if __name__ == "__main__": try: asyncio.run(main()) except (KeyboardInterrupt, SystemExit): print("\nОстановлено.")