#!/usr/bin/env python3 import imaplib import email from email.header import decode_header import re import json import time import logging import os from datetime import datetime, timedelta from flask import Flask, request, jsonify from threading import Thread # ===== НАСТРОЙКИ ===== IMAP_SERVER = "mail.gtn.ru" IMAP_PORT = 993 EMAIL_USER = "M.Fedorov@gtn.ru" EMAIL_PASSWORD = "cloutfan8" FOLDER_NAME = "Новые договоры" DATA_DIR = "/opt/contract-parser/data" EMAILS_FILE = f"{DATA_DIR}/emails.json" SHOWN_FILE = f"{DATA_DIR}/shown_banners.json" SCRIPT_FILE = f"{DATA_DIR}/script.user.js" LOG_DIR = "/opt/contract-parser/logs" LOG_FILE = f"{LOG_DIR}/parser.log" # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler() ] ) app = Flask(__name__) # ===== ФУНКЦИИ РАБОТЫ С ПОЧТОЙ ===== def decode_header_value(value): """Декодирует заголовки письма""" if not value: return "" decoded_parts = decode_header(value) result = "" for part, encoding in decoded_parts: if isinstance(part, bytes): if encoding: result += part.decode(encoding, errors='ignore') else: result += part.decode('utf-8', errors='ignore') else: result += str(part) return result def parse_email_body(body_text): """Извлекает имя оператора и номер договора из текста письма""" pattern = r'Оператор\s+([А-Яа-я\s]+?)\s+создал\s+новый\s+договор\s+(\d+)' match = re.search(pattern, body_text, re.IGNORECASE) if match: operator_name = match.group(1).strip() contract_number = match.group(2).strip() operator_name = re.sub(r'\s+', ' ', operator_name) return operator_name, contract_number return None, None def fetch_new_emails(): """Проверяет папку 'Новые договоры' и возвращает новые письма""" try: mail = imaplib.IMAP4_SSL(IMAP_SERVER, IMAP_PORT) mail.login(EMAIL_USER, EMAIL_PASSWORD) folder_found = False for folder in [FOLDER_NAME, f'"{FOLDER_NAME}"', 'Новые договоры']: try: mail.select(folder, readonly=False) folder_found = True logging.info(f"Открыта папка: {folder}") break except imaplib.IMAP4.error: continue if not folder_found: status, folders = mail.list() logging.error("Не найдена папка 'Новые договоры'. Доступные папки:") for f in folders: logging.error(f" {f.decode('utf-8')}") mail.logout() return [] status, messages = mail.search(None, 'UNSEEN') if status != 'OK' or not messages[0]: mail.logout() return [] email_ids = messages[0].split() logging.info(f"Найдено непрочитанных писем: {len(email_ids)}") new_emails = [] for e_id in email_ids: status, msg_data = mail.fetch(e_id, '(RFC822)') if status != 'OK': continue for response_part in msg_data: if isinstance(response_part, tuple): msg = email.message_from_bytes(response_part[1]) # Получаем тело письма (только text/plain) body = "" if msg.is_multipart(): for part in msg.walk(): if part.get_content_type() == "text/plain": payload = part.get_payload(decode=True) if payload: body += payload.decode('utf-8', errors='ignore') break else: payload = msg.get_payload(decode=True) if payload: body = payload.decode('utf-8', errors='ignore') operator_name, contract_number = parse_email_body(body) if operator_name and contract_number: new_emails.append({ "contract_number": contract_number, "operator_name": operator_name, "received_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) logging.info(f"✓ Найдено: договор {contract_number}, оператор {operator_name}") else: logging.warning(f"Не удалось распарсить письмо") mail.store(e_id, '+FLAGS', '\\Seen') mail.close() mail.logout() return new_emails except Exception as e: logging.error(f"Ошибка при подключении к почте: {e}") return [] # ===== ФУНКЦИИ РАБОТЫ С JSON ===== def load_emails(): if not os.path.exists(EMAILS_FILE): return {"last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "emails": []} with open(EMAILS_FILE, 'r', encoding='utf-8') as f: return json.load(f) def save_emails(data): data["last_updated"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(EMAILS_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def load_shown(): if not os.path.exists(SHOWN_FILE): return {"last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "shown": []} with open(SHOWN_FILE, 'r', encoding='utf-8') as f: return json.load(f) def save_shown(data): data["last_updated"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(SHOWN_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def cleanup_old_emails(): """Удаляет записи старше 4 месяцев""" four_months_ago = datetime.now() - timedelta(days=120) emails_data = load_emails() original_count = len(emails_data["emails"]) emails_data["emails"] = [ e for e in emails_data["emails"] if datetime.strptime(e["received_date"], "%Y-%m-%d %H:%M:%S") > four_months_ago ] if original_count != len(emails_data["emails"]): logging.info(f"Очистка emails: удалено {original_count - len(emails_data['emails'])} записей") save_emails(emails_data) shown_data = load_shown() original_shown = len(shown_data["shown"]) shown_data["shown"] = [ s for s in shown_data["shown"] if datetime.strptime(s["closed_at"], "%Y-%m-%d %H:%M:%S") > four_months_ago ] if original_shown != len(shown_data["shown"]): logging.info(f"Очистка shown: удалено {original_shown - len(shown_data['shown'])} записей") save_shown(shown_data) def add_new_emails(new_emails): """Добавляет новые письма в очередь (без дубликатов)""" if not new_emails: return emails_data = load_emails() shown_data = load_shown() shown_contracts = [s["contract_number"] for s in shown_data["shown"]] existing_contracts = [e["contract_number"] for e in emails_data["emails"]] added_count = 0 for email in new_emails: contract_number = email["contract_number"] if contract_number in shown_contracts: logging.info(f"Пропуск договора {contract_number} — уже был показан") continue if contract_number in existing_contracts: logging.info(f"Пропуск договора {contract_number} — уже в очереди") continue emails_data["emails"].append(email) added_count += 1 logging.info(f"Добавлен договор {contract_number} (оператор: {email['operator_name']})") if added_count > 0: save_emails(emails_data) # ===== FLASK ЭНДПОИНТЫ ===== @app.route('/get-banner', methods=['GET']) def get_banner(): contract_number = request.args.get('contract', '') if not contract_number: return jsonify({"error": "Не указан номер договора"}), 400 logging.info(f"Запрос плашки для договора {contract_number}") emails_data = load_emails() shown_data = load_shown() shown_contracts = [s["contract_number"] for s in shown_data["shown"]] if contract_number in shown_contracts: return jsonify({"show": False, "reason": "already_shown"}) for email in emails_data["emails"]: if email["contract_number"] == contract_number: return jsonify({ "show": True, "operator_name": email["operator_name"], "contract_number": email["contract_number"], "received_date": email["received_date"] }) return jsonify({"show": False, "reason": "not_found"}) @app.route('/close-banner', methods=['POST']) def close_banner(): data = request.get_json() contract_number = data.get('contract_number', '') if not contract_number: return jsonify({"error": "Не указан номер договора"}), 400 logging.info(f"Закрытие плашки для договора {contract_number}") emails_data = load_emails() shown_data = load_shown() # Находим оператора перед удалением operator_name = "" for email in emails_data["emails"]: if email["contract_number"] == contract_number: operator_name = email["operator_name"] break # Удаляем из очереди emails_data["emails"] = [ e for e in emails_data["emails"] if e["contract_number"] != contract_number ] save_emails(emails_data) # Добавляем в показанные shown_contracts = [s["contract_number"] for s in shown_data["shown"]] if contract_number not in shown_contracts: shown_data["shown"].append({ "contract_number": contract_number, "operator_name": operator_name, "closed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }) save_shown(shown_data) cleanup_old_emails() return jsonify({"success": True}) @app.route('/script.user.js', methods=['GET']) def get_script(): script_path = "/opt/contract-parser/data/script.user.js" if os.path.exists(script_path): with open(script_path, 'r', encoding='utf-8') as f: script_content = f.read() return script_content, 200, {'Content-Type': 'application/javascript'} else: return jsonify({"error": "Скрипт не найден"}), 404 @app.route('/health', methods=['GET']) def health(): return jsonify({"status": "ok", "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}) # ===== ФОНОВЫЙ ПОТОК ===== def background_check(): logging.info("Фоновый поток запущен. Проверка почты каждую минуту.") while True: try: new_emails = fetch_new_emails() if new_emails: add_new_emails(new_emails) time.sleep(60) except Exception as e: logging.error(f"Ошибка в фоновом потоке: {e}") time.sleep(60) # ===== ЗАПУСК ===== def main(): logging.info("=" * 50) logging.info("Запуск сервера парсинга договоров") logging.info(f"IMAP сервер: {IMAP_SERVER}") logging.info(f"Папка: {FOLDER_NAME}") logging.info(f"Порт API: 8766") logging.info("=" * 50) os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(LOG_DIR, exist_ok=True) if not os.path.exists(EMAILS_FILE): save_emails({"last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "emails": []}) if not os.path.exists(SHOWN_FILE): save_shown({"last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "shown": []}) logging.info("Первая проверка почты...") try: new_emails = fetch_new_emails() if new_emails: add_new_emails(new_emails) except Exception as e: logging.error(f"Ошибка при первой проверке: {e}") thread = Thread(target=background_check, daemon=True) thread.start() app.run(host='0.0.0.0', port=8766, debug=False, use_reloader=False) if __name__ == '__main__': main()