270 lines
11 KiB
Python
270 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Детальный анализ экспортированных писем — полная таблица для Telegram.
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
import fitz
|
||
|
||
BASE_OUTPUT = Path("/home/matrixhasyou/mutt/exported_emails")
|
||
|
||
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 (первые 5 страниц)."""
|
||
try:
|
||
doc = fitz.open(filepath)
|
||
text = ""
|
||
for page_num in range(min(5, len(doc))):
|
||
text += doc[page_num].get_text()
|
||
doc.close()
|
||
return text
|
||
except Exception as e:
|
||
return f"ERROR: {e}"
|
||
|
||
def extract_incoming_number(text):
|
||
"""Извлечь номер входящего/исходящего."""
|
||
patterns = [
|
||
r'[№N]\s*[\.:]?\s*([A-Za-zА-Яа-я0-9\-\/\.]+)',
|
||
r'исх\.?\s*[\.:№\s]*\s*([A-Za-zА-Яа-я0-9\-\/\.]+)',
|
||
r'вх\.?\s*[\.:№\s]*\s*([A-Za-zА-Яа-я0-9\-\/\.]+)',
|
||
r'регистрации\s*[№:]\s*([A-Za-zА-Яа-я0-9\-\/\.]+)',
|
||
r'([0-9]{2,4}[-\/][A-Za-zА-я0-9]+[-\/][0-9]+)',
|
||
]
|
||
|
||
for pattern in patterns:
|
||
matches = re.finditer(pattern, text[:3000], re.IGNORECASE)
|
||
for match in matches:
|
||
val = match.group(1).strip()
|
||
# Фильтруем слишком короткие или мусорные значения
|
||
if len(val) > 3 and val.lower() not in ['reply', 'pdf', 'notification', 'fo']:
|
||
return val
|
||
return None
|
||
|
||
def extract_response_date(text):
|
||
"""Извлечь дату ответа."""
|
||
date_patterns = [
|
||
r'от\s+(\d{2}[\.\/]\d{2}[\.\/]\d{4})',
|
||
r'(\d{2}[\.\/]\d{2}[\.\/]\d{4})',
|
||
r'«(\d+)»\s+(\w+)\s+(\d{4})',
|
||
]
|
||
for pattern in date_patterns:
|
||
match = re.search(pattern, text[:2000])
|
||
if match:
|
||
return match.group(0)[:15]
|
||
return None
|
||
|
||
def classify_reply(text):
|
||
"""Определить тип ответа."""
|
||
text_lower = text.lower()
|
||
|
||
formal_markers = [
|
||
'ваше обращение рассмотрено', 'сообщаем вам', 'уведомляем',
|
||
'в соответствии с федеральным законом', 'в пределах компетенции',
|
||
'направляем в ваш адрес', 'в рамках действующего законодательства',
|
||
'уважаемый', 'благодарим за обращение', 'ваше предложение',
|
||
'разъясняем', 'доводим до вашего сведения',
|
||
'не усматривается оснований', 'не представляет возможным',
|
||
'в соответствии с положением', 'регулирующим',
|
||
'ваше предложение не подлежит', 'отказано',
|
||
]
|
||
|
||
substantive_markers = [
|
||
'удовлетворено', 'приняты меры', 'выявлены нарушения',
|
||
'обязать устранить', 'проведена проверка', 'выдано предписание',
|
||
'направлено предписание', 'привлечен к ответственности',
|
||
'возбуждено дело', 'штраф наложен', 'материалы переданы',
|
||
'установлено нарушение', 'признано незаконным',
|
||
'подлежит устранению', 'в срок до', 'контроль исполнения',
|
||
'исполнено в полном объеме', 'восстановлено',
|
||
'перерасчет', 'возмещено', 'компенсация',
|
||
]
|
||
|
||
formal_count = sum(1 for kw in formal_markers if kw in text_lower)
|
||
substantive_count = sum(1 for kw in substantive_markers if kw in text_lower)
|
||
|
||
if formal_count == 0 and substantive_count == 0:
|
||
# Проверяем есть ли вообще ответ
|
||
has_response_indicators = any(kw in text_lower for kw in [
|
||
'ответ', 'рассмотрение', 'сообщаем', 'уведомляем',
|
||
'уважаемый', 'обращение', 'письмо ваше'
|
||
])
|
||
if has_response_indicators:
|
||
return "Формальная ⚠️", formal_count, substantive_count
|
||
return None, 0, 0
|
||
|
||
if substantive_count > formal_count:
|
||
return "По существу ✅", formal_count, substantive_count
|
||
return "Формальная ⚠️", formal_count, substantive_count
|
||
|
||
def analyze_email(filepath):
|
||
"""Полный анализ одного письма."""
|
||
filename = filepath.name
|
||
|
||
# Парсим имя файла
|
||
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
|
||
|
||
# Анализируем текст
|
||
text = parse_pdf_metadata(filepath)
|
||
|
||
incoming_num = extract_incoming_number(text)
|
||
response_date = extract_response_date(text)
|
||
reply_type, formal_score, substantive_score = classify_reply(text)
|
||
|
||
is_sent = True # Это исходящее письмо
|
||
has_reply = reply_type is not None
|
||
|
||
return {
|
||
'date': date_str,
|
||
'sender': sender,
|
||
'subject': subject[:55],
|
||
'incoming_number': incoming_num,
|
||
'response_date': response_date,
|
||
'reply_type': reply_type,
|
||
'has_reply': has_reply,
|
||
'is_sent': is_sent,
|
||
'formal_score': formal_score,
|
||
'substantive_score': substantive_score,
|
||
}
|
||
|
||
def build_telegram_report():
|
||
"""Построить полный отчёт для Telegram."""
|
||
all_emails = []
|
||
|
||
# Сканируем все папки
|
||
for folder_name, folder_emoji 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:
|
||
info = analyze_email(pdf_file)
|
||
info['folder_name'] = folder_name
|
||
info['folder_display'] = folder_emoji
|
||
all_emails.append(info)
|
||
|
||
# Сортируем по дате
|
||
all_emails.sort(key=lambda x: x['date'])
|
||
|
||
# Считаем статистику
|
||
total = len(all_emails)
|
||
with_replies = sum(1 for e in all_emails if e['has_reply'])
|
||
without_replies = total - with_replies
|
||
formal = sum(1 for e in all_emails if e['reply_type'] and 'Формальная' in str(e['reply_type']))
|
||
substantive = sum(1 for e in all_emails if e['reply_type'] and 'существу' in str(e['reply_type']))
|
||
|
||
# Строим отчёт
|
||
report = []
|
||
|
||
report.append("📊 **ПОЛНЫЙ ОТЧЁТ ПО ПЕРЕПИСКЕ С ОРГАНАМИ ВЛАСТИ**")
|
||
report.append(f"📅 Дата формирования: {datetime.now().strftime('%d.%m.%Y %H:%M')}")
|
||
report.append("━" * 48)
|
||
report.append("")
|
||
|
||
# Группируем по органам
|
||
by_folder = {}
|
||
for email in all_emails:
|
||
folder = email['folder_display']
|
||
if folder not in by_folder:
|
||
by_folder[folder] = []
|
||
by_folder[folder].append(email)
|
||
|
||
# Для каждого органа
|
||
for folder_display, emails in by_folder.items():
|
||
report.append(f"{'─' * 48}")
|
||
report.append(f"**{folder_display}**")
|
||
report.append(f"{'─' * 48}")
|
||
report.append("")
|
||
|
||
replies_count = sum(1 for e in emails if e['has_reply'])
|
||
|
||
# Таблица писем
|
||
report.append(f"{'№':<3} │ {'Дата':^10} │ {'Статус':^13} │ {'Вх. №':^18} │ Тема")
|
||
report.append(f"{'─' * 3}─┼─{'─' * 10}─┼─{'─' * 13}─┼─{'─' * 18}─┼─{'─' * 30}")
|
||
|
||
for idx, email in enumerate(emails, 1):
|
||
status = ""
|
||
if email['has_reply']:
|
||
status = email['reply_type'][:13]
|
||
else:
|
||
status = "⏳ Нет ответа"
|
||
|
||
incoming = email['incoming_number'] or "—"
|
||
if len(incoming) > 18:
|
||
incoming = incoming[:15] + "..."
|
||
|
||
subject = email['subject']
|
||
if len(subject) > 35:
|
||
subject = subject[:32] + "..."
|
||
|
||
report.append(f"{idx:<3} │ {email['date']:^10} │ {status:^13} │ {incoming:^18} │ {subject}")
|
||
|
||
report.append("")
|
||
report.append(f"📊 Итого по органу: {len(emails)} писем | Ответов: {replies_count} ({replies_count/len(emails)*100:.0f}%)")
|
||
report.append("")
|
||
|
||
# ИТОГО
|
||
report.append("=" * 48)
|
||
report.append("📈 **СВОДНАЯ СТАТИСТИКА**")
|
||
report.append("=" * 48)
|
||
report.append("")
|
||
report.append(f"✉️ **Всего отправлено писем:** {total}")
|
||
report.append(f"✅ **Получено ответов:** {with_replies} ({with_replies/total*100:.0f}%)")
|
||
report.append(f"⏳ **Без ответа:** {without_replies} ({without_replies/total*100:.0f}%)")
|
||
report.append("")
|
||
|
||
if with_replies > 0:
|
||
report.append("📊 **КАЧЕСТВО ОТВЕТОВ:**")
|
||
report.append(f" ⚠️ Формальные отписки: {formal} ({formal/with_replies*100:.0f}%)")
|
||
report.append(f" ✅ Ответы по существу: {substantive} ({substantive/with_replies*100:.0f}%)")
|
||
report.append("")
|
||
|
||
report.append("🎯 **ВЫВОД:**")
|
||
if substantive / with_replies > 0.5:
|
||
report.append(" Большинство ответов — по существу. Органы работают! 💪")
|
||
elif formal / with_replies > 0.7:
|
||
report.append(" ⚠️ Преобладают формальные отписки. Требуется:")
|
||
report.append(" • Повторные обращения с требованием ответа по существу")
|
||
report.append(" • Жалобы в вышестоящие органы")
|
||
report.append(" • Обращение в прокуратуру на бездействие")
|
||
else:
|
||
report.append(" Смешанная картина. Требуется работа с каждым ответом.")
|
||
|
||
report.append("")
|
||
report.append("=" * 48)
|
||
report.append(f"🕐 Отчёт сформирован: {datetime.now().strftime('%H:%M %d.%m.%Y')}")
|
||
report.append("=" * 48)
|
||
|
||
return "\n".join(report)
|
||
|
||
if __name__ == '__main__':
|
||
print("Формирую детальный отчёт...")
|
||
report = build_telegram_report()
|
||
print(report)
|
||
|
||
# Сохраняем
|
||
output_file = Path(__file__).parent / "report_detailed.txt"
|
||
output_file.write_text(report, encoding='utf-8')
|
||
print(f"\n\n📄 Отчёт сохранён: {output_file}")
|