150 lines
5.8 KiB
Python
Executable File
150 lines
5.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Скрипт для приведения текста постов к "сырому" виду.
|
||
Удаляет форматирование, добавленное функцией get_text_message(), оставляя только исходный текст.
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import html
|
||
import os
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
project_root = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
import aiosqlite
|
||
|
||
from logs.custom_logger import logger
|
||
|
||
DEFAULT_DB_PATH = "database/tg-bot-database.db"
|
||
|
||
# Паттерны для определения форматированного текста
|
||
PREFIX = "Пост из ТГ:\n"
|
||
ANONYMOUS_SUFFIX = "\n\nПост опубликован анонимно"
|
||
AUTHOR_SUFFIX_PATTERN = re.compile(r"\n\nАвтор поста: .+$")
|
||
|
||
|
||
def extract_raw_text(formatted_text: str) -> str:
|
||
"""
|
||
Извлекает сырой текст из форматированного текста поста.
|
||
|
||
Args:
|
||
formatted_text: Форматированный текст поста
|
||
|
||
Returns:
|
||
str: Сырой текст или исходный текст, если форматирование не обнаружено
|
||
"""
|
||
if not formatted_text:
|
||
return ""
|
||
|
||
# Проверяем, начинается ли текст с префикса
|
||
if not formatted_text.startswith(PREFIX):
|
||
# Текст уже в сыром виде или имеет другой формат
|
||
return formatted_text
|
||
|
||
# Извлекаем текст после префикса
|
||
text_after_prefix = formatted_text[len(PREFIX) :]
|
||
|
||
# Проверяем, заканчивается ли текст на "Пост опубликован анонимно"
|
||
if text_after_prefix.endswith(ANONYMOUS_SUFFIX):
|
||
raw_text = text_after_prefix[: -len(ANONYMOUS_SUFFIX)]
|
||
# Проверяем, заканчивается ли текст на "Автор поста: ..."
|
||
elif AUTHOR_SUFFIX_PATTERN.search(text_after_prefix):
|
||
raw_text = AUTHOR_SUFFIX_PATTERN.sub("", text_after_prefix)
|
||
else:
|
||
# Не удалось определить формат, возвращаем текст без префикса
|
||
raw_text = text_after_prefix
|
||
|
||
# Декодируем HTML-экранирование
|
||
raw_text = html.unescape(raw_text)
|
||
|
||
return raw_text
|
||
|
||
|
||
async def main(db_path: str, dry_run: bool = False) -> None:
|
||
db_path = os.path.abspath(db_path)
|
||
if not os.path.exists(db_path):
|
||
logger.error("База данных не найдена: %s", db_path)
|
||
print(f"Ошибка: база данных не найдена: {db_path}")
|
||
return
|
||
|
||
async with aiosqlite.connect(db_path) as conn:
|
||
await conn.execute("PRAGMA foreign_keys = ON")
|
||
|
||
# Получаем все записи с текстом
|
||
cursor = await conn.execute(
|
||
"SELECT message_id, text FROM post_from_telegram_suggest WHERE text IS NOT NULL AND text != ''"
|
||
)
|
||
posts = await cursor.fetchall()
|
||
await cursor.close()
|
||
|
||
updated_count = 0
|
||
skipped_count = 0
|
||
error_count = 0
|
||
|
||
print(f"Найдено записей для обработки: {len(posts)}")
|
||
if dry_run:
|
||
print("РЕЖИМ ПРОВЕРКИ (dry-run): изменения не будут сохранены")
|
||
|
||
# Обрабатываем каждую запись
|
||
for message_id, formatted_text in posts:
|
||
try:
|
||
# Извлекаем сырой текст
|
||
raw_text = extract_raw_text(formatted_text)
|
||
|
||
# Проверяем, изменился ли текст
|
||
if raw_text == formatted_text:
|
||
skipped_count += 1
|
||
continue
|
||
|
||
if dry_run:
|
||
print(f"\n[DRY-RUN] message_id={message_id}:")
|
||
print(f" Было: {formatted_text[:100]}...")
|
||
print(f" Станет: {raw_text[:100]}...")
|
||
else:
|
||
# Обновляем запись
|
||
await conn.execute(
|
||
"UPDATE post_from_telegram_suggest SET text = ? WHERE message_id = ?",
|
||
(raw_text, message_id),
|
||
)
|
||
updated_count += 1
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при обработке поста message_id={message_id}: {e}")
|
||
error_count += 1
|
||
|
||
if not dry_run:
|
||
await conn.commit()
|
||
|
||
total_processed = updated_count + skipped_count + error_count
|
||
logger.info(
|
||
f"Обработка завершена. Всего записей: {total_processed}, "
|
||
f"обновлено: {updated_count}, пропущено: {skipped_count}, ошибок: {error_count}"
|
||
)
|
||
print(f"\nОбработка завершена:")
|
||
print(f" - Всего записей: {total_processed}")
|
||
print(f" - Обновлено: {updated_count}")
|
||
print(f" - Пропущено (уже в сыром виде): {skipped_count}")
|
||
print(f" - Ошибок: {error_count}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(
|
||
description="Приведение текста постов к 'сырому' виду"
|
||
)
|
||
parser.add_argument(
|
||
"--db",
|
||
default=os.environ.get("DB_PATH", DEFAULT_DB_PATH),
|
||
help="Путь к БД (или DB_PATH)",
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="Режим проверки без сохранения изменений",
|
||
)
|
||
args = parser.parse_args()
|
||
asyncio.run(main(args.db, args.dry_run))
|