227 lines
9.4 KiB
Python
227 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Анализ экспортированных писем.
|
||
Извлекает метаданные из PDF файлов и создаёт таблицу переписки.
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import json
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
import fitz # PyMuPDF
|
||
|
||
BASE_OUTPUT = Path("/home/matrixhasyou/mutt/exported_emails")
|
||
ALL_FOLDER = BASE_OUTPUT / "all"
|
||
|
||
# Папки с ответами (органы которым писали)
|
||
REPLY_FOLDERS = {
|
||
"gosuslugi.ru": "Госуслуги",
|
||
"gov.ru": "Правительство РФ",
|
||
"mailop.ru": "Почта России (mailop)",
|
||
"mvd.ru": "МВД России",
|
||
"rospotrebnadzor.ru": "Роспотребнадзор",
|
||
"russianpost.ru": "Почта России",
|
||
"ulgkh.ru": "УЛГКХ (Ульяновск)",
|
||
"ulgss.ru": "УЛГСС (Ульяновск)",
|
||
}
|
||
|
||
def parse_pdf_metadata(filepath):
|
||
"""Извлечь текст и метаданные из PDF."""
|
||
try:
|
||
doc = fitz.open(filepath)
|
||
text = ""
|
||
for page_num in range(min(3, len(doc))): # Первые 3 страницы
|
||
text += doc[page_num].get_text()
|
||
doc.close()
|
||
return text
|
||
except Exception as e:
|
||
return f"ERROR: {e}"
|
||
|
||
def extract_email_info(filename, text):
|
||
"""Извлечь информацию о письме из имени файла и текста."""
|
||
# Парсим имя файла: YYYY-MM-DD_sender_subject.pdf
|
||
parts = Path(filename).stem.split('_', 2)
|
||
if len(parts) >= 3:
|
||
date_str = parts[0]
|
||
sender = parts[1]
|
||
subject = parts[2]
|
||
else:
|
||
date_str = "unknown"
|
||
sender = "unknown"
|
||
subject = Path(filename).stem
|
||
|
||
# Анализируем текст на наличие признаков ответа
|
||
is_reply = False
|
||
reply_type = None # "formal" или "substantive"
|
||
|
||
text_lower = text.lower()
|
||
|
||
# Ключевые слова для определения типа ответа
|
||
formal_keywords = [
|
||
'ваше обращение', 'рассмотрено', 'сообщаем', 'уведомляем',
|
||
'в соответствии с', 'на основании', 'руководствуясь',
|
||
'отказано', 'не усматривается', 'не имеется оснований',
|
||
'в пределах компетенции', 'направляем в ваш адрес',
|
||
'информация предоставляется', 'согласно', 'в рамках',
|
||
'уважаемый', 'благодарим за обращение',
|
||
]
|
||
|
||
substantive_keywords = [
|
||
'удовлетворено', 'приняты меры', 'выявлены нарушения',
|
||
'обязать', 'устранить', 'проведена проверка',
|
||
'направлено предписание', 'штраф', 'предписани',
|
||
'срок устранения', 'контроль', 'исполнено',
|
||
'возбуждено дело', 'материалы переданы',
|
||
'установлено', 'подтверждено', 'выдано',
|
||
]
|
||
|
||
formal_count = sum(1 for kw in formal_keywords if kw in text_lower)
|
||
substantive_count = sum(1 for kw in substantive_keywords if kw in text_lower)
|
||
|
||
# Определяем тип
|
||
if formal_count > 0 or substantive_count > 0:
|
||
is_reply = True
|
||
if substantive_count > formal_count:
|
||
reply_type = "По существу ✓"
|
||
else:
|
||
reply_type = "Формальная отписка ⚠"
|
||
|
||
# Извлекаем номер входящего (если есть)
|
||
incoming_number = None
|
||
number_patterns = [
|
||
r'[№N]\s*[\.:]?\s*([A-Za-zА-Яа-я0-9\-\/]+)',
|
||
r'исх\.\s*[\.:]?\s*([A-Za-zА-Яа-я0-9\-\/]+)',
|
||
r'вх\.\s*[\.:]?\s*([A-Za-zА-Яа-я0-9\-\/]+)',
|
||
r'регистрационн\w*\s*[\.:]?\s*([A-Za-zА-Яа-я0-9\-\/]+)',
|
||
]
|
||
for pattern in number_patterns:
|
||
match = re.search(pattern, text[:2000], re.IGNORECASE)
|
||
if match:
|
||
incoming_number = match.group(1)
|
||
break
|
||
|
||
return {
|
||
'date': date_str,
|
||
'sender': sender,
|
||
'subject': subject[:60],
|
||
'is_reply': is_reply,
|
||
'reply_type': reply_type,
|
||
'incoming_number': incoming_number,
|
||
'formal_score': formal_count,
|
||
'substantive_score': substantive_count,
|
||
}
|
||
|
||
def analyze_all_emails():
|
||
"""Анализировать все письма и вернуть статистику."""
|
||
results = []
|
||
|
||
# Сканируем папки с ответами
|
||
for folder_name, folder_desc in REPLY_FOLDERS.items():
|
||
folder_path = BASE_OUTPUT / folder_name
|
||
if not folder_path.exists():
|
||
continue
|
||
|
||
pdf_files = sorted(folder_path.glob("*.pdf"))
|
||
|
||
for pdf_file in pdf_files:
|
||
text = parse_pdf_metadata(pdf_file)
|
||
info = extract_email_info(pdf_file.name, text)
|
||
info['folder'] = folder_desc
|
||
info['filename'] = pdf_file.name
|
||
results.append(info)
|
||
|
||
return results
|
||
|
||
def format_table(results):
|
||
"""Форматировать результаты в красивую таблицу для Telegram."""
|
||
# Сортируем по дате
|
||
results.sort(key=lambda x: x['date'])
|
||
|
||
# Заголовок
|
||
table = "📊 **ОТЧЁТ ПО ПЕРЕПИСКЕ С ОРГАНАМИ**\n"
|
||
table += f"📅 Дата формирования: {datetime.now().strftime('%d.%m.%Y')}\n"
|
||
table += "━" * 50 + "\n\n"
|
||
|
||
# Группируем по папкам
|
||
by_folder = {}
|
||
for r in results:
|
||
folder = r['folder']
|
||
if folder not in by_folder:
|
||
by_folder[folder] = []
|
||
by_folder[folder].append(r)
|
||
|
||
total_sent = len(results)
|
||
total_replies = sum(1 for r in results if r['is_reply'])
|
||
total_formal = sum(1 for r in results if r['reply_type'] and 'Формальная' in r['reply_type'])
|
||
total_substantive = sum(1 for r in results if r['reply_type'] and 'существу' in r['reply_type'])
|
||
|
||
# Таблица по органам
|
||
table += "📮 **ПО ОРГАНАМ:**\n\n"
|
||
|
||
for folder, items in by_folder.items():
|
||
replies = sum(1 for i in items if i['is_reply'])
|
||
reply_pct = (replies / len(items) * 100) if items else 0
|
||
|
||
table += f"**{folder}**\n"
|
||
table += f"├─ Писем: {len(items)}\n"
|
||
table += f"├─ Ответов: {replies} ({reply_pct:.0f}%)\n"
|
||
|
||
# Последние 3 письма с деталями
|
||
recent = items[-3:] if len(items) > 3 else items
|
||
for idx, item in enumerate(recent):
|
||
num = len(items) - len(recent) + idx + 1
|
||
status = "✅ Ответ" if item['is_reply'] else "⏳ Нет ответа"
|
||
reply_detail = f" → {item['reply_type']}" if item['reply_type'] else ""
|
||
incoming = f" [Вх.№ {item['incoming_number']}]" if item['incoming_number'] else ""
|
||
|
||
table += f"├─ #{num} | {item['date']} | {status}{reply_detail}{incoming}\n"
|
||
table += f"│ 📝 {item['subject'][:45]}...\n"
|
||
|
||
if len(items) > 3:
|
||
table += f"└─ ... и ещё {len(items) - 3} писем\n"
|
||
|
||
table += "\n"
|
||
|
||
# ИТОГО
|
||
table += "━" * 50 + "\n"
|
||
table += "📈 **ИТОГО:**\n\n"
|
||
|
||
reply_pct_total = (total_replies / total_sent * 100) if total_sent else 0
|
||
formal_pct = (total_formal / total_replies * 100) if total_replies else 0
|
||
substantive_pct = (total_substantive / total_replies * 100) if total_replies else 0
|
||
|
||
table += f"✉️ Всего отправлено: **{total_sent}**\n"
|
||
table += f"✅ Получено ответов: **{total_replies}** ({reply_pct_total:.0f}%)\n"
|
||
table += f"⏳ Без ответа: **{total_sent - total_replies}**\n\n"
|
||
|
||
if total_replies > 0:
|
||
table += "📊 **КАЧЕСТВО ОТВЕТОВ:**\n\n"
|
||
table += f"⚠️ Формальные отписки: **{total_formal}** ({formal_pct:.0f}%)\n"
|
||
table += f"✅ Ответы по существу: **{total_substantive}** ({substantive_pct:.0f}%)\n\n"
|
||
|
||
table += "🎯 **ВЫВОД:**\n\n"
|
||
if substantive_pct > 50:
|
||
table += "Большинство ответов — **по существу**. Органы работают! 💪\n"
|
||
elif formal_pct > 70:
|
||
table += "Преобладают **формальные отписки**. Требуется escalation. ⚠️\n"
|
||
else:
|
||
table += "Смешанная картина. Нужно работать с каждым ответом отдельно.\n"
|
||
|
||
table += "\n" + "━" * 50 + "\n"
|
||
table += f"🕐 Сформировано: {datetime.now().strftime('%H:%M %d.%m.%Y')}\n"
|
||
|
||
return table
|
||
|
||
if __name__ == '__main__':
|
||
print("Анализирую письма...")
|
||
results = analyze_all_emails()
|
||
print(f"Проанализировано {len(results)} писем\n")
|
||
|
||
table = format_table(results)
|
||
print(table)
|
||
|
||
# Сохраняем в файл
|
||
output_file = Path(__file__).parent / "report_table.txt"
|
||
output_file.write_text(table, encoding='utf-8')
|
||
print(f"\nОтчёт сохранён в {output_file}")
|