Script Telegram for a quick counting of winning in Fantan

Скрипт для быстрого подчета выйграша в FanTan

Пример

Код

import cv2
import numpy as np
import asyncio
import tempfile
import os
from aiogram import Bot, Dispatcher, types, F
from aiogram.filters import Command
from aiogram.fsm.storage.memory import MemoryStorage
import logging

logging.basicConfig(level=logging.INFO)


BOT_TOKEN = "TOKEN"


bot = Bot(token=BOT_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(storage=storage)

async def cleanup_files(file_path):
if not file_path or not os.path.exists(file_path):
return

for attempt in range(3):
try:
await asyncio.sleep(0.1 * (attempt + 1))
os.unlink(file_path)
return
except PermissionError:
if attempt == 2:
asyncio.create_task(delayed_file_removal(file_path))
continue

async def delayed_file_removal(file_path):
for i in range(10):
await asyncio.sleep(1)
try:
if os.path.exists(file_path):
os.unlink(file_path)
return
except PermissionError:
continue

def count_balls(image_path):
image = cv2.imread(image_path)
if image is None:
return 0, None

output = image.copy()
height, width = image.shape[:2]

top_margin = 120
left_margin = 320
side_margin = 30

gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
blurred = cv2.GaussianBlur(gray, (11, 11), 0)
_, thresh = cv2.threshold(blurred, 210, 255, cv2.THRESH_BINARY)

kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
cleaned = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)

contours, _ = cv2.findContours(cleaned, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

ball_count = 0
for cnt in contours:
area = cv2.contourArea(cnt)
if 40 < area < 400:
M = cv2.moments(cnt)
if M["m00"] == 0:
continue
cX = int(M["m10"] / M["m00"])
cY = int(M["m01"] / M["m00"])

if cX < left_margin:
continue
if cY < top_margin:
continue
if cX < side_margin or cX > width - side_margin:
continue

ball_count += 1
cv2.drawContours(output, [cnt], -1, (0, 255, 0), 2)
cv2.circle(output, (cX, cY), 4, (0, 0, 255), -1)

return ball_count, output

@dp.message(Command("start"))
async def start_command(message: types.Message):
await message.answer("Отправь фото стола с шариками ")

@dp.message(F.photo)
async def handle_photo(message: types.Message):
try:
file = await bot.get_file(message.photo[-1].file_id)

tmp_file = tempfile.NamedTemporaryFile(suffix='.jpg', delete=False)
tmp_file_path = tmp_file.name
tmp_file.close()

await bot.download_file(file.file_path, tmp_file_path)

ball_count, processed_image = count_balls(tmp_file_path)

columns = ball_count // 4
remainder = ball_count % 4

if remainder > 0:
result_text = f" Найдено шариков: {ball_count}\n {columns} столбцов и {remainder} остаток"
else:
result_text = f" Найдено шариков: {ball_count}\n {columns} столбцов"

if processed_image is not None:
tmp_result = tempfile.NamedTemporaryFile(suffix='.jpg', delete=False)
tmp_result_path = tmp_result.name
tmp_result.close()

cv2.imwrite(tmp_result_path, processed_image)

photo = types.FSInputFile(tmp_result_path)
await message.answer_photo(photo=photo, caption=result_text)

await asyncio.sleep(0.1)

try:
os.unlink(tmp_result_path)
except PermissionError:
asyncio.create_task(delayed_file_removal(tmp_result_path))
else:
await message.answer(result_text)

await cleanup_files(tmp_file_path)

except Exception as e:
await message.answer(f"Ошибка: {str(e)}")

@dp.message()
async def handle_text(message: types.Message):
await message.answer("Отправь фото ")

async def main():
await dp.start_polling(bot)

if __name__ == "__main__":
asyncio.run(main())

Python

import cv2

import numpy as np

import asyncio

import tempfile

import os

from aiogram import Bot, Dispatcher, types, F

from aiogram.filters import Command

from aiogram.fsm.storage.memory import MemoryStorage

import logging



logging.basicConfig(level=logging.INFO)





BOT_TOKEN = "TOKEN"





bot = Bot(token=BOT_TOKEN)

storage = MemoryStorage()

dp = Dispatcher(storage=storage)



async def cleanup_files(file_path):

    if not file_path or not os.path.exists(file_path):

        return

  

    for attempt in range(3):

        try:

            await asyncio.sleep(0.1 * (attempt + 1))

            os.unlink(file_path)

            return

        except PermissionError:

            if attempt == 2:

                asyncio.create_task(delayed_file_removal(file_path))

            continue



async def delayed_file_removal(file_path):

    for i in range(10):

        await asyncio.sleep(1)

        try:

            if os.path.exists(file_path):

                os.unlink(file_path)

            return

        except PermissionError:

            continue



def count_balls(image_path):

    image = cv2.imread(image_path)

    if image is None:

        return 0, None

  

    output = image.copy()

    height, width = image.shape[:2]



    top_margin = 120

    left_margin = 320

    side_margin = 30



    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

    blurred = cv2.GaussianBlur(gray, (11, 11), 0)

    _, thresh = cv2.threshold(blurred, 210, 255, cv2.THRESH_BINARY)



    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))

    cleaned = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)



    contours, _ = cv2.findContours(cleaned, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)



    ball_count = 0

    for cnt in contours:

        area = cv2.contourArea(cnt)

        if 40 < area < 400:

            M = cv2.moments(cnt)

            if M["m00"] == 0:

                continue

            cX = int(M["m10"] / M["m00"])

            cY = int(M["m01"] / M["m00"])



            if cX < left_margin:

                continue

            if cY < top_margin:

                continue

            if cX < side_margin or cX > width - side_margin:

                continue



            ball_count += 1

            cv2.drawContours(output, [cnt], -1, (0, 255, 0), 2)

            cv2.circle(output, (cX, cY), 4, (0, 0, 255), -1)



    return ball_count, output



@dp.message(Command("start"))

async def start_command(message: types.Message):

    await message.answer("Отправь фото стола с шариками ")



@dp.message(F.photo)

async def handle_photo(message: types.Message):

    try:

        file = await bot.get_file(message.photo[-1].file_id)

      

        tmp_file = tempfile.NamedTemporaryFile(suffix='.jpg', delete=False)

        tmp_file_path = tmp_file.name

        tmp_file.close()

      

        await bot.download_file(file.file_path, tmp_file_path)

      

        ball_count, processed_image = count_balls(tmp_file_path)

      

        columns = ball_count // 4

        remainder = ball_count % 4

      

        if remainder > 0:

            result_text = f" Найдено шариков: {ball_count}\n {columns} столбцов и {remainder} остаток"

        else:

            result_text = f" Найдено шариков: {ball_count}\n {columns} столбцов"

      

        if processed_image is not None:

            tmp_result = tempfile.NamedTemporaryFile(suffix='.jpg', delete=False)

            tmp_result_path = tmp_result.name

            tmp_result.close()

          

            cv2.imwrite(tmp_result_path, processed_image)

          

            photo = types.FSInputFile(tmp_result_path)

            await message.answer_photo(photo=photo, caption=result_text)

          

            await asyncio.sleep(0.1)

          

            try:

                os.unlink(tmp_result_path)

            except PermissionError:

                asyncio.create_task(delayed_file_removal(tmp_result_path))

        else:

            await message.answer(result_text)

      

        await cleanup_files(tmp_file_path)

      

    except Exception as e:

        await message.answer(f"Ошибка: {str(e)}")



@dp.message()

async def handle_text(message: types.Message):

    await message.answer("Отправь фото ")



async def main():

    await dp.start_polling(bot)



if __name__ == "__main__":

    asyncio.run(main())

Спросите зачем?
Ответ: