parser_version6

1
import asyncio
2
from aiogram import Bot, Dispatcher, types
3
from aiogram.types import Message
4
from aiogram.fsm.storage.memory import MemoryStorage
5
from aiogram.fsm.context import FSMContext
6
from aiogram.filters import Command
7
from aiogram.fsm.state import StatesGroup, State
8
from datetime import date, timedelta
9
from test2 import parse_and_save
10
import mysql.connector
11
import hashlib
12

13
# Замените на ваш токен и ID группы
14
TOKEN = "7032031280:AAEjXQh-WNlqfY2el6a7cj0PtUvJNh-vLM0"
15
GROUP_ID = "--1002448868558"  
16

17
# Инициализация бота и диспетчера
18
bot = Bot(token=TOKEN, timeout=60)
19
storage = MemoryStorage()
20
dp = Dispatcher(storage=storage)
21

22
# Конфигурация базы данных
23
config = {
24
    'user': 'root',
25
    'password': 'Uz94deco!',
26
    'host': '127.0.0.1',
27
    'database': 'parser'
28
}
29

30
# Класс для состояний FSM
31
class DaysForm(StatesGroup):
32
    waiting_for_days = State()
33

34
# Функция для нормализации текста сообщения
35
def normalize_text(text):
36
    return ' '.join(text.strip().split()).lower()  # Убираем лишние пробелы и приводим текст к нижнему регистру
37

38
# Функция для вычисления хеша текста сообщения
39
def get_text_hash(text):
40
    normalized_text = normalize_text(text)
41
    return hashlib.sha256(normalized_text.encode('utf-8')).hexdigest()
42

43
# Функция для получения истории сообщений группы (до 1000 сообщений)
44
async def get_group_messages(limit=1000):
45
    all_messages = []
46
    all_hashes = []
47
    try:
48
        offset_message_id = None  # ID последнего полученного сообщения для смещения
49

50
        while len(all_messages) < limit:
51
            # Получаем 100 сообщений с возможным смещением
52
            messages = await bot.get_chat_history(chat_id=GROUP_ID, limit=100, offset_message_id=offset_message_id)
53
            
54
            if not messages:
55
                break  # Если сообщений больше нет, выходим из цикла
56
            
57
            for message in messages:
58
                if message.text:
59
                    normalized_message = normalize_text(message.text)
60
                    all_messages.append(normalized_message)
61
                    all_hashes.append(get_text_hash(normalized_message))
62
            
63
            # Устанавливаем ID последнего сообщения для следующего смещения
64
            offset_message_id = messages[-1].message_id
65
        
66
        # Ограничиваем общее количество сообщений до запрашиваемого лимита
67
        return all_messages[:limit], all_hashes[:limit]
68
    
69
    except Exception as e:
70
        print(f"Ошибка при получении истории сообщений группы: {e}")
71
        return [], []
72

73
# Проверка наличия дубликатов по хешам сообщений в группе или базе данных
74
async def is_duplicate(post_text, group_messages_hashes, conn):
75
    post_text_hash = get_text_hash(post_text)
76

77
    # Проверка на дубликат в истории сообщений группы
78
    if post_text_hash in group_messages_hashes:
79
        return True
80

81
    # Проверка на дубликат в базе данных
82
    cursor = conn.cursor()
83
    cursor.execute("SELECT COUNT(*) FROM sent_posts WHERE post_hash = %s", (post_text_hash,))
84
    count = cursor.fetchone()[0]
85
    cursor.close()
86

87
    return count > 0
88

89
# Ограничиваем скорость запросов, чтобы избежать лимитов
90
async def send_message_with_delay(text, conn):
91
    try:
92
        await bot.send_message(GROUP_ID, text)
93

94
        # Сохраняем хеш отправленного сообщения в базе данных
95
        post_text_hash = get_text_hash(text)
96
        cursor = conn.cursor()
97
        cursor.execute("INSERT INTO sent_posts (post_hash) VALUES (%s)", (post_text_hash,))
98
        conn.commit()
99
        cursor.close()
100

101
        await asyncio.sleep(5)  # Задержка между запросами, чтобы избежать лимита
102
    except Exception as e:
103
        print(f"Ошибка при отправке сообщения: {e}")
104

105
async def send_unique_posts():
106
    try:
107
        conn = mysql.connector.connect(**config)
108
        cursor = conn.cursor(dictionary=True)
109
        cursor.execute("SELECT post_text, time, channel_name FROM posts")  # Добавили post_date и channel_name
110
        posts = cursor.fetchall()
111

112
        # Получаем до 1000 сообщений из группы и их хеши
113
        group_messages, group_messages_hashes = await get_group_messages(1000)
114

115
        for post in posts:
116
            post_text = post['post_text']
117
            time = post['time'].strftime('%d-%m-%Y')  # Форматируем дату
118
            channel_name = post['channel_name']  # Название канала
119

120
            # Формируем текст для отправки
121
            formatted_message = f"Канал: {channel_name}\nДата: {time}\n\n{post_text}"
122
            
123
            is_dup = await is_duplicate(formatted_message, group_messages_hashes, conn)
124

125
            if not is_dup:
126
                await send_message_with_delay(formatted_message, conn)
127
                print(f"Отправлен уникальный пост: {formatted_message}")
128
            else:
129
                print(f"Пост пропущен (дубликат): {formatted_message}")
130

131
    except Exception as e:
132
        print(f"Ошибка при отправке уникальных постов: {e}")
133
    finally:
134
        cursor.close()
135
        conn.close()
136

137
# Загрузка ссылок и ключевых слов из файлов
138
def load_links_and_keywords():
139
    try:
140
        with open('Links.txt', 'r', encoding='utf-8') as f:
141
            links = [line.strip() for line in f.readlines()]
142
        with open('keywords.txt', 'r', encoding='utf-8') as f:
143
            keywords = [line.strip() for line in f.readlines()]
144
        return links, keywords
145
    except Exception as e:
146
        print(f"Ошибка при загрузке ссылок или ключевых слов: {e}")
147
        return [], []
148

149
# Загрузка количества дней из файла
150
def load_days():
151
    try:
152
        with open('days.txt', 'r') as f:
153
            return int(f.read().strip())
154
    except Exception as e:
155
        print(f"Ошибка при загрузке дней: {e}")
156
        return 7  # Значение по умолчанию
157

158
# Сохранение количества дней в файл
159
def save_days(days):
160
    try:
161
        with open('days.txt', 'w') as f:
162
            f.write(str(days))
163
    except Exception as e:
164
        print(f"Ошибка при сохранении дней: {e}")
165

166
# Фоновая задача для парсинга каждые 15 минут
167
async def scheduled_parsing():
168
    while True:
169
        try:
170
            days = load_days()  # Загружаем количество дней из файла
171
            links, keywords = load_links_and_keywords()
172
            nazvanie = ["Бойлерная", "Кремлёвский безБашенник", "Силовики", "НЕЗЫГАРЬ", "ОПЕР Слил", "SHOT", "Mash", "Baza", "ВЧК-ОГПУ", "КОНТЕКСТ"]  # Названия каналов
173

174
            start_date = date.today() - timedelta(days=days)
175
            
176
            # Асинхронный запуск парсинга
177
            await asyncio.to_thread(parse_and_save, links, nazvanie, keywords, start_date)
178

179
            # Отправка уникальных постов
180
            await send_unique_posts()
181

182
        except Exception as e:
183
            print(f"Ошибка в фоновом парсинге: {e}")
184
        
185
        await asyncio.sleep(900)  # Повтор каждые 15 минут
186

187
# Команда /start
188
@dp.message(Command("start"))
189
async def start_command(message: Message, state: FSMContext):
190
    await message.answer("""Здравствуйте, вас привествует бот по Парсеру данных из разных каналов.
191
    Основные команды в боте:
192
    
193
    /start - запуск бота.
194
    /change_days - изменить время для парсера данных.""")
195
    await message.answer("Сколько дней вы хотите парсить данные?")
196
    await state.set_state(DaysForm.waiting_for_days)
197

198
# Обработка ответа на /start
199
@dp.message(DaysForm.waiting_for_days)
200
async def process_days(message: Message, state: FSMContext):
201
    if message.text.isdigit():
202
        days = int(message.text)
203
        save_days(days)
204
        await message.answer(f"Дни для парсинга установлены: {days}")
205
        await state.clear()  # Очищаем состояние
206
    else:
207
        await message.answer("Пожалуйста, введите корректное число.")
208

209
# Команда /change_days для изменения количества дней
210
@dp.message(Command("change_days"))
211
async def change_days_command(message: Message, state: FSMContext):
212
    await message.answer("Сколько дней вы хотите установить для парсинга?")
213
    await state.set_state(DaysForm.waiting_for_days)
214

215
# Закрытие сессии при завершении работы
216
async def on_shutdown():
217
    await bot.session.close()  # Закрываем сессию бота
218

219
# Запуск бота и фоновой задачи
220
async def main():
221
    asyncio.create_task(scheduled_parsing())  # Запуск фоновой задачи для парсинга
222
    await bot.delete_webhook(drop_pending_updates=True)
223
    try:
224
        await dp.start_polling(bot)
225
    finally:
226
        await on_shutdown()  # Корректное завершение работы
227

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