Python-сообщество

 import aioodbc
import datetime
import time
import asyncio
import logging
from logging.handlers import TimedRotatingFileHandler
import re
import aiogram
from aiogram import Bot, Dispatcher, types
from aiogram.utils import executor
import traceback
import smtplib
from email.message import EmailMessage
from email.mime.text import MIMEText
from email_validator import validate_email, EmailNotValidError
TOKEN = '...'
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)
# Queues
telegram_queue = asyncio.Queue()
email_queue = asyncio.Queue()
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Create loggers for different log types
logger_telegram = logging.getLogger('telegram')
logger_system = logging.getLogger('system')
logger_email = logging.getLogger('email')
# Configure TimedRotatingFileHandler for each log type
file_handler_telegram = logging.handlers.TimedRotatingFileHandler('logs/telegram.txt', when='midnight', interval=1)
file_handler_system = logging.handlers.TimedRotatingFileHandler('logs/system.txt', when='midnight', interval=1)
file_handler_email = logging.handlers.TimedRotatingFileHandler('logs/email.txt', when='midnight', interval=1)
# Set the log format
log_format = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# Configure log levels and handlers for each logger
logger_telegram.setLevel(logging.DEBUG)
logger_telegram.addHandler(file_handler_telegram)
file_handler_telegram.setFormatter(log_format)
logger_system.setLevel(logging.DEBUG)
logger_system.addHandler(file_handler_system)
file_handler_system.setFormatter(log_format)
logger_email.setLevel(logging.DEBUG)
logger_email.addHandler(file_handler_system)
file_handler_email.setFormatter(log_format)
# Example usage
logger_telegram.info('Start Telegram log message')
logger_system.info('Start System log message')
logger_email.info('Start Email log message')
# Connecting to database
dsn = 'DRIVER={ODBC Driver 17 for SQL Server};' \
          'SERVER=192.168.1.150;' \
          'DATABASE=davp;' \
          'UID=...;' \
          'PWD=...'
@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
    user_id = message.from_user.id
    user_firstname = message.from_user.first_name
    user_lastname = message.from_user.last_name
    user_login = message.from_user.username
    print(f"Command '/start' received from Name: {user_firstname}, Surname: {user_lastname}, Login: {user_login}, ID: {user_id}")
    await message.answer(f"User ID: {user_id}")
@dp.message_handler(commands=['chatid'])
async def send_chatid(message: types.Message):
    current_chat_id = message.chat.id
    await message.answer(f'Current chat ID: {current_chat_id}')
    
# Telegram Sending function
async def send_telegram(result):
    new_dt = result[4].strftime("%d.%m.%Y %H:%M:%S")
    g_nr = result[2]
    receivers = result[8]
    # Add Emoji to reactions
    reaction_type = str(result[11])
    match reaction_type:
        case "1": # Alarm
            emoji_text = "🚨"
        case "2": #Panic
            emoji_text = "‼"
        case "3": # Fire
            emoji_text = "🧯"
        case "4": # Restore
            emoji_text = "✅"
        case "5": # Close
            emoji_text = "🔐"
        case "6": # Open
            emoji_text = "🔓"
        case "7": # Low Power
            emoji_text = "⚠️"
        case "9": # Power Fault
            emoji_text = "⚠️"
        case "17": # Late open
            emoji_text = "🔓"
        case "18": # Trouble
            emoji_text = "⚠️"
        case "20": # Alarm
            emoji_text = "🚨"
        case _: # None
            emoji_text = ""
    tgID = receivers.replace("t", "")
    message = f'[{new_dt}] [Nr]: {result[9]} [Event]: {emoji_text} {result[0]} [Rajons]: {g_nr}'
    try:
        #await bot.send_message(tgID, message)
        #print('Telegram Success')
        logger_telegram.info(f'[INFO]: Telegram sent: User: {result[6]}, ID: {receivers}, Nr.:{result[9]} {result[0]}, Rajons: {g_nr}, Laiks: {new_dt}')
    except Exception as e:
        logger_telegram.error(f'Telegram Error {e}')
# E-Mail Sending function
async def send_email(result):
    new_dt = result[4].strftime("%d.%m.%Y %H:%M:%S")
    g_nr = result[2]
    receivers = result[8]
    try:
        valid_email = validate_email(receivers)
        # Constructing email body
        subject = f"User: {receivers} Nr.:{result[9]} {result[0]}, Rajons: {g_nr}, Laiks: {new_dt}"
        message_body = "Message Sender"
        
        msg = MIMEText(message_body, 'plain', 'utf-8')
        msg['From'] = 'TEST <test@mymail.com>'
        msg['To'] = 'testinbox@mymail.com'
        msg['Subject'] = subject
        
        smtpObj = smtplib.SMTP('smtp-relay.gmail.com', 25)
        smtpObj.sendmail('test@mymail.com', ['testinbox@mymail.com'], msg.as_string())
        smtpObj.quit()
        print('E-Mail Success!')
        print(f"[INFO]: Mail sent: {receivers}, Nr.:{result[9]} {result[0]}, Rajons: {g_nr}, Laiks: {new_dt}")
        
    except EmailNotValidError as e:
        print(f'[ERROR]: Can not send message due to an error [Nr]: {result[3]}, data: {receivers}, converted {tgID}')
        print(str(e))
    except Exception as e:
        print(f'[ERROR]: E-Mail - {e}')
async def process_telegram_queue():
    while True:
        result = await telegram_queue.get()
        if result:
            await send_telegram(result)
        telegram_queue.task_done()
        await asyncio.sleep(2)
async def process_email_queue():
    while True:
        results = []
        # Collect multiple results from the queue
        for _ in range(email_queue.qsize()):
            result = await email_queue.get()
            if result:
                results.append(result)
            email_queue.task_done()
        
        # Process each result
        for result in results:
            await send_email(result)
        await asyncio.sleep(2)
async def get_events():
    last_execution_time = datetime.datetime.now()
    last_event_id = 0
    #receivers = result[8]
    #result = None  # Initialize result outside the loop
    #receivers = None  # Initialize receivers outside the loop
    
    while True:
        try:
            async with aioodbc.connect(dsn=dsn) as connection:
                async with connection.cursor() as cursor:
                    query = """
                    SELECT DISTINCT
                            e.eventname,
                            e.unifier_id,
                            e.groupnr,
                            e.objectdbid,
                            e.receivedtime,
                            d.SendMail,
                            cp.name,
                            cp.objectid,
                            cp.email,
                            ou.unified_id,
                            e.id,
                            e.reactionid
                        FROM
                            (SELECT * FROM davp.dbo.events
                            UNION ALL
                            SELECT * FROM davp_archive.dbo.events) e
                        JOIN davp.dbo.eventsdescription d
                            ON e.eventdescid = d.Id
                        JOIN davp.dbo.object_unifiers ou
                            ON ou.id = e.unifier_id
                        JOIN davp.dbo.corresppersons cp
                            ON e.objectdbid = cp.objectid
                        WHERE d.SendMail = 'Yes' AND cp.SendMail = 'Yes' AND e.repeated = 'No'
                            AND e.receivedtime > ? AND e.id > ?
                            ORDER BY CAST(e.receivedtime AS DATETIME) DESC"""
                    await cursor.execute(query, last_execution_time, last_event_id)
                    results = await cursor.fetchall()
                    # Sort Results by receivedtime
                    results.sort(key=lambda x: x[4], reverse=True)
                    # Check if any results are found
                    for result in results:
                        if result[4] is not None and result is not None:
                            receivers = result[8]
                            last_event_id = result[10]
                            #print(f'Found - {result}')
                            # Check and Split to queues
                            if re.match(r"t\d+$", receivers):
                                await telegram_queue.put(result)
                                #print(f'TG Queue {result}')
                            else:
                                await email_queue.put(result)
                                #print(f'EM Queue {result}')
                        else:
                            print(f'[ERROR]: result[4] value - {result[4]}')
            await asyncio.sleep(10)
        except Exception as e:
            logger_system.error(f'Exception: {type(e).__name__}, Message: {str(e)}')
            logger_system.error(f'Traceback: {traceback.format_exc()}')
            logger_system.error(f'Result[4] value - {result[4]}, Event ID : {result[10]}, E-Mail: {result[8]}')
            await asyncio.sleep(15)
async def main():
    first_iteration = True
    while True:
        try:
            if not first_iteration:
                print("Second iteration, sending messages")
                asyncio.create_task(get_events())
                asyncio.create_task(process_telegram_queue())
                asyncio.create_task(process_email_queue())
                await asyncio.sleep(10)
                await asyncio.gather(get_events(), process_telegram_queue(), process_email_queue())
            else:
                await asyncio.sleep(5)
                print("First iteration, skip sending messages")
                first_iteration = False
                await asyncio.sleep(5)
                pass
        except Exception as e:
            logger_system.error(f'Error : {e}')
            await asyncio.sleep(15)
if __name__ == "__main__":
    asyncio.run(main())