| # from fastapi import FastAPI | |
| # app = FastAPI() | |
| # @app.get("/") | |
| # def greet_json(): | |
| # return {"Hello": "World!"} | |
| import logging | |
| from aiogram.client.default import DefaultBotProperties | |
| from aiogram import Bot, Dispatcher, F | |
| from aiogram.types import Message, Update | |
| from aiogram.filters import CommandStart | |
| from aiogram.enums import ParseMode | |
| from fastapi import FastAPI | |
| from fastapi.responses import HTMLResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.templating import Jinja2Templates | |
| from fastapi.requests import Request | |
| import uvicorn | |
| from contextlib import asynccontextmanager | |
| import os | |
| # from aiohttp_socks import SocksConnector | |
| # # Define your SOCKS proxy | |
| # socks_proxy = "socks5://127.0.0.1:9050" | |
| # # Create a connector with SOCKS proxy and rdns=True for DNS resolution | |
| # connector = SocksConnector.from_url(socks_proxy, rdns=True) | |
| # from aiogram.client.session.aiohttp import AiohttpSession | |
| #session = AiohttpSession(proxy='http://proxy.server:3128') # в proxy указан прокси сервер pythonanywhere, он нужен для подключения | |
| # session = AiohttpSession(proxy='http://w.10.mtproto.ru') # в proxy указан прокси сервер pythonanywhere, он нужен для подключения | |
| # from aiogram.client.telegram import TelegramAPIServer | |
| # from aiogram.client.session.aiohttp import AiohttpSession | |
| token=os.getenv('BOT_TOKEN') | |
| # test_servers = AiohttpSession( | |
| # api=TelegramAPIServer( | |
| # base="https://api.telegram.org/bot{token}/test/{method}", | |
| # file="https://api.telegram.org/file/bot{token}/test/{path}", | |
| # ) | |
| # ) | |
| # import ssl | |
| # ssl_context = ssl.create_default_context() | |
| # ssl_context.check_hostname = False | |
| # ssl_context.verify_mode = ssl.CERT_NONE | |
| # bot = Bot(token=os.getenv('BOT_TOKEN'), | |
| # default=DefaultBotProperties(parse_mode=ParseMode.HTML) | |
| # ,session=test_servers, | |
| # ssl_context=ssl_context) | |
| # session = AiohttpSession(proxy="http://172.28.1.5:3128/") | |
| # session = AiohttpSession(proxy="http://188.114.98.233:80") | |
| #bot = Bot(token=token, session=session) | |
| # bot = Bot(token=token) | |
| # Create a session with the custom connector | |
| # session = bot.session | |
| # session.connector = connector | |
| # dp = Dispatcher() | |
| # @asynccontextmanager | |
| # async def lifespan(app: FastAPI): | |
| # await bot.set_webhook(url='https://jameshns-btf.hf.space/webhook', | |
| # allowed_updates=dp.resolve_used_update_types(), | |
| # drop_pending_updates=True) | |
| # yield | |
| # await bot.delete_webhook() | |
| app = FastAPI() | |
| # # app.mount("/static", StaticFiles(directory="static"), name="static") | |
| # # templates = Jinja2Templates(directory="templates") | |
| # @dp.message(CommandStart()) | |
| # async def start(message: Message) -> None: | |
| # await message.answer('Привет!') | |
| # # @app.get("/", response_class=HTMLResponse) | |
| # # async def read_root(request: Request): | |
| # # return templates.TemplateResponse("index.html", {"request": request}) | |
| # @app.get("/") | |
| # def greet_json(): | |
| # return {"Hello": "World!"} | |
| # # @app.post("/webhook") | |
| # # async def webhook(request: Request) -> None: | |
| # # update = Update.model_validate(await request.json(), context={"bot": bot}) | |
| # # await dp.feed_update(bot, update) | |
| # if __name__ == "__main__": | |
| # logging.basicConfig( | |
| # level=logging.INFO, | |
| # format=u'%(filename)s:%(lineno)d #%(levelname)-8s [%(asctime)s] - %(name)s - %(message)s', | |
| # ) | |
| # uvicorn.run(app, host="0.0.0.0", port=5000) | |
| import os | |
| import asyncio | |
| from aiogram import Bot, Dispatcher | |
| from aiogram import Router | |
| from aiogram.fsm.storage.memory import MemoryStorage | |
| from aiogram.types import Message | |
| from aiogram.filters.command import Command | |
| from dotenv import load_dotenv | |
| from aiogram.fsm.context import FSMContext | |
| from aiogram.fsm.state import State, StatesGroup | |
| import random | |
| load_dotenv() | |
| bot = Bot(token=token) | |
| dp = Dispatcher(storage=MemoryStorage()) | |
| bot_router = Router() | |
| class Step(StatesGroup): | |
| guesses = State() | |
| finish = State() | |
| async def cmd_start(message: Message, state: FSMContext): | |
| await message.answer("Начнём игру. Я загадал число от 1 до 100 (1 тоже может быть, и 100 тоже может быть). Число целое. Попробуй угадать какое за наименьшее количество попыток.") | |
| await state.set_state(Step.guesses) | |
| the_number = random.randint(1, 100) | |
| await state.update_data(the_number=the_number) | |
| await state.update_data(attempt=0) | |
| async def ask(message: Message, state: FSMContext): | |
| storage_data = await state.get_data() | |
| attempt = int(storage_data.get('attempt', 0)) | |
| the_number = int(storage_data.get('the_number')) | |
| attempt = attempt + 1 | |
| await state.update_data(attempt=attempt) | |
| try: | |
| a_number = int(message.text) | |
| if a_number == the_number: | |
| await message.answer('Молодец, ты отгадал!') | |
| await message.answer(f'За {attempt} попыток') | |
| else: | |
| # не отгадал. дадим подсказку | |
| if the_number > a_number: | |
| await message.answer('Загаданное число больше') | |
| else: | |
| await message.answer('Загаданное число меньше') | |
| except Exception as e: | |
| await message.answer('Это не число') | |
| async def main(): | |
| await dp.start_polling(bot) | |
| def greet_json(): | |
| return {"Hello2": "World!2"} | |
| if __name__ == "__main__": | |
| asyncio.run(main()) |