#!/usr/bin/env python3 """ Webapp server: serves Mini App static files + provides /api/create-invoice endpoint. Replaces: python3 -m http.server 8080 Run: python3 /opt/lottie-bot/webapp/server.py """ from __future__ import annotations import asyncio import hashlib import hmac import json import logging import os import sqlite3 import sys import tempfile import time import uuid from pathlib import Path from urllib.parse import parse_qs import aiohttp from aiohttp import web from dotenv import load_dotenv # Make app package importable (project root is one level up) _PROJECT_ROOT = Path(__file__).parent.parent if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) # Load .env from project root (one level up from webapp/) _env_path = Path(__file__).parent.parent / ".env" load_dotenv(_env_path) logging.basicConfig( format="%(asctime)s %(name)s %(levelname)s %(message)s", level=logging.INFO, ) logger = logging.getLogger("webapp_server") STATIC_DIR = Path(__file__).parent BOT_TOKEN = os.environ.get("BOT_TOKEN", "") DATABASE_PATH = os.environ.get("DATABASE_PATH", "/opt/lottie-bot/storage/bot.db") PORT = int(os.environ.get("WEBAPP_PORT", "8080")) logger.info("Loaded BOT_TOKEN: %s", "YES" if BOT_TOKEN else "NO (empty!)") CORS_HEADERS = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "POST, OPTIONS", "Access-Control-Allow-Headers": "Content-Type", } # ────────────────────────────────────────────────────────────── # Database helpers # ────────────────────────────────────────────────────────────── def _get_db() -> sqlite3.Connection: conn = sqlite3.connect(DATABASE_PATH) conn.row_factory = sqlite3.Row return conn def _init_orders_table() -> None: with _get_db() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS miniapp_orders ( order_id TEXT PRIMARY KEY, tg_user_id INTEGER NOT NULL, stars INTEGER NOT NULL, fill_color TEXT, recolor INTEGER DEFAULT 1, keep_colors INTEGER DEFAULT 1, build_data TEXT, status TEXT NOT NULL DEFAULT 'pending', created_at INTEGER NOT NULL ) """) conn.commit() def _save_order(order_id: str, tg_user_id: int, stars: int, fill_color: str, recolor: bool, keep_colors: bool, build_data: dict) -> None: with _get_db() as conn: conn.execute( """INSERT OR REPLACE INTO miniapp_orders (order_id, tg_user_id, stars, fill_color, recolor, keep_colors, build_data, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?)""", (order_id, tg_user_id, stars, fill_color, 1 if recolor else 0, 1 if keep_colors else 0, json.dumps(build_data), int(time.time())), ) conn.commit() # ────────────────────────────────────────────────────────────── # Telegram initData validation # ────────────────────────────────────────────────────────────── def _validate_init_data(init_data: str, bot_token: str) -> dict | None: """Return parsed user dict if initData is valid, else None.""" try: params = parse_qs(init_data, keep_blank_values=True) hash_values = params.pop("hash", [None]) hash_val = hash_values[0] if hash_values else None if not hash_val: return None check_pairs = sorted((k, v[0]) for k, v in params.items()) check_string = "\n".join(f"{k}={v}" for k, v in check_pairs) secret_key = hmac.new(b"WebAppData", bot_token.encode(), hashlib.sha256).digest() expected = hmac.new(secret_key, check_string.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, hash_val): return None # Re-parse original (before pop) to get user field orig = parse_qs(init_data, keep_blank_values=True) user_str = orig.get("user", [None])[0] return json.loads(user_str) if user_str else None except Exception: logger.exception("initData validation error") return None # ────────────────────────────────────────────────────────────── # API handler # ────────────────────────────────────────────────────────────── async def handle_create_invoice(request: web.Request) -> web.Response: if request.method == "OPTIONS": return web.Response(status=204, headers=CORS_HEADERS) try: body = await request.json() except Exception: return web.json_response({"error": "invalid JSON"}, status=400, headers=CORS_HEADERS) stars = int(body.get("stars", 0)) if stars <= 0: return web.json_response({"error": "تعداد Stars نامعتبر است"}, status=400, headers=CORS_HEADERS) # Determine user ID from initData or fallback field tg_user_id: int | None = None init_data = body.get("initData", "") if init_data and BOT_TOKEN: user = _validate_init_data(init_data, BOT_TOKEN) if user: tg_user_id = int(user["id"]) if not tg_user_id: tg_user_id = int(body.get("user_id", 0)) or None if not tg_user_id: return web.json_response({"error": "احراز هویت ناموفق"}, status=401, headers=CORS_HEADERS) # Create order record in DB order_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}" _save_order( order_id=order_id, tg_user_id=tg_user_id, stars=stars, fill_color=body.get("fill_color", "#2ea6ff"), recolor=bool(body.get("recolor_stroke", True)), keep_colors=bool(body.get("keep_logo_colors", True)), build_data=body.get("build_data", {}), ) # Create invoice link via Telegram Bot API invoice_payload = json.dumps( {"type": "miniapp-order", "uid": tg_user_id, "oid": order_id}, separators=(",", ":"), ) title = body.get("title", "مجموعه استیکر انیمیشن") description = body.get("description", f"{stars} Stars – خرید مجموعه استیکر") async with aiohttp.ClientSession() as session: url = f"https://api.telegram.org/bot{BOT_TOKEN}/createInvoiceLink" tg_body = { "title": title, "description": description, "payload": invoice_payload, "currency": "XTR", "prices": [{"label": "مجموعه استیکر", "amount": stars}], } async with session.post(url, json=tg_body) as resp: data = await resp.json() if not data.get("ok"): err = data.get("description", "خطا در ایجاد لینک پرداخت") logger.error("createInvoiceLink failed: %s", err) return web.json_response({"error": err}, status=500, headers=CORS_HEADERS) invoice_url = data["result"] logger.info("invoice created: order=%s user=%s stars=%s", order_id, tg_user_id, stars) return web.json_response({"invoice_url": invoice_url}, headers=CORS_HEADERS) # ────────────────────────────────────────────────────────────── # /api/preview-lottie – server-side logo embedding for preview # ────────────────────────────────────────────────────────────── async def handle_preview_lottie(request: web.Request) -> web.Response: if request.method == "OPTIONS": return web.Response(status=204, headers=CORS_HEADERS) try: body = await request.json() except Exception: return web.json_response({"error": "invalid JSON"}, status=400, headers=CORS_HEADERS) lottie_rel = body.get("lottie_path", "") svg_content = body.get("svg_content", "") fill_color = body.get("fill_color", "#2ea6ff") recolor_stroke = bool(body.get("recolor_stroke", False)) keep_logo_colors = bool(body.get("keep_logo_colors", True)) if not lottie_rel or not svg_content: return web.json_response({"error": "lottie_path and svg_content required"}, status=400, headers=CORS_HEADERS) # Resolve lottie path safely (must stay inside STATIC_DIR) try: lottie_path = (STATIC_DIR / lottie_rel.lstrip("/")).resolve() if not str(lottie_path).startswith(str(STATIC_DIR.resolve())): raise ValueError("path traversal") if not lottie_path.exists(): raise FileNotFoundError(str(lottie_path)) except Exception as exc: return web.json_response({"error": f"lottie file error: {exc}"}, status=400, headers=CORS_HEADERS) tmp_svg = None tmp_out_dir = None try: from app.services.logo_automation import LogoAutomationService, LogoAutomationError # Write SVG to temp file tmp_svg = tempfile.NamedTemporaryFile(suffix=".svg", delete=False, mode="w", encoding="utf-8") tmp_svg.write(svg_content) tmp_svg.close() tmp_out_dir = Path(tempfile.mkdtemp()) svc = LogoAutomationService(output_dir=tmp_out_dir) result = await asyncio.get_event_loop().run_in_executor( None, lambda: svc.process( lottie_path=lottie_path, svg_path=Path(tmp_svg.name), fill_hex=fill_color, recolor_stroke=recolor_stroke, keep_logo_colors=keep_logo_colors, ), ) lottie_json = result.output_json.read_text(encoding="utf-8") logger.info("preview-lottie: processed %s", lottie_path.name) return web.Response( text=lottie_json, content_type="application/json", headers=CORS_HEADERS, ) except Exception as exc: logger.warning("preview-lottie error: %s", exc) return web.json_response({"error": str(exc)}, status=500, headers=CORS_HEADERS) finally: if tmp_svg: try: Path(tmp_svg.name).unlink(missing_ok=True) except Exception: pass if tmp_out_dir: try: import shutil shutil.rmtree(tmp_out_dir, ignore_errors=True) except Exception: pass # ────────────────────────────────────────────────────────────── # App setup # ────────────────────────────────────────────────────────────── async def main() -> None: _init_orders_table() app = web.Application(client_max_size=4 * 1024 * 1024) # 4 MB for SVG uploads app.router.add_options("/api/create-invoice", handle_create_invoice) app.router.add_post("/api/create-invoice", handle_create_invoice) app.router.add_options("/api/preview-lottie", handle_preview_lottie) app.router.add_post("/api/preview-lottie", handle_preview_lottie) app.router.add_static("/", STATIC_DIR, show_index=True) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, "0.0.0.0", PORT) await site.start() logger.info("Webapp server running on http://0.0.0.0:%d", PORT) await asyncio.sleep(float("inf")) if __name__ == "__main__": asyncio.run(main())