feat(email_auditor): scan Spam folder and fix TG proxy connection error

This commit is contained in:
MatrixHasYou 2026-06-21 13:52:19 +04:00
parent cf8fb1b8ab
commit 4f355e55f4
2 changed files with 188 additions and 61 deletions

View file

@ -37,6 +37,9 @@ DOMAINS = [
"ulgss.ru", "ulgov.ru", "ulgkh.ru", "dgi.ru", "mos.ru", "pfrf.ru", "ulgss.ru", "ulgov.ru", "ulgkh.ru", "dgi.ru", "mos.ru", "pfrf.ru",
"fsin.gov.ru", "skrf.ru", "epp.genproc.gov.ru", "ugpr.ru", "fsin.gov.ru", "skrf.ru", "epp.genproc.gov.ru", "ugpr.ru",
"fas.gov.ru", "minstroyrf.gov.ru", "gkh.ru", "sudrf.ru", "fas.gov.ru", "minstroyrf.gov.ru", "gkh.ru", "sudrf.ru",
"msud.ru", "arbitr.ru", "fssp.gov.ru", "fssprus.ru", "mirsud",
"msud73.ru", "mirsud73.ru", "court.ru", "sud.ru", "vkks.ru",
"uloblsud.ru", "sledcom.ru"
] ]
# From this date # From this date
@ -331,60 +334,74 @@ def main():
# Connect # Connect
log.info("Connecting to Gmail IMAP...") log.info("Connecting to Gmail IMAP...")
try: try:
mailbox = MailBox(IMAP_SERVER).login(imap_user, imap_pass, initial_folder=GMAIL_FOLDER) mailbox = MailBox(IMAP_SERVER).login(imap_user, imap_pass)
log.info("Connected successfully!") log.info("Connected successfully!")
except Exception as e: except Exception as e:
log.error(f"Failed to connect: {e}") log.error(f"Failed to connect: {e}")
sys.exit(1) sys.exit(1)
# Phase 1: Fetch headers only to find matching emails # Auto-detect folder names
log.info("Phase 1: Scanning emails (headers only) to find matches...") all_mail_folder = "[Gmail]/Вся почта"
scan_start = time.time() spam_folder = "[Gmail]/Спам"
try:
for folder in mailbox.folder.list():
if r'\All' in folder.flags:
all_mail_folder = folder.name
elif r'\Junk' in folder.flags:
spam_folder = folder.name
except Exception as e:
log.warning(f"Failed to auto-detect folder names, using defaults: {e}")
folders_to_scan = [all_mail_folder, spam_folder]
matching_uids = [] matching_uids = []
total_scanned = 0 total_scanned = 0
# Phase 1: Fetch headers from all target folders
log.info("Phase 1: Scanning emails (headers only) to find matches...")
scan_start = time.time()
last_log_time = time.time() last_log_time = time.time()
# Fetch headers only first — much faster for folder_name in folders_to_scan:
for msg in mailbox.fetch( log.info(f"Scanning folder: {folder_name}")
AND(date_gte=SINCE_DATE), try:
bulk=True, mailbox.folder.set(folder_name)
headers_only=True, except Exception as e:
mark_seen=False, log.error(f"Failed to set folder {folder_name}: {e}")
): continue
total_scanned += 1
# Log every 10 seconds for msg in mailbox.fetch(
now = time.time() AND(date_gte=SINCE_DATE),
if now - last_log_time > 10: bulk=True,
elapsed = now - scan_start headers_only=True,
rate = total_scanned / elapsed if elapsed > 0 else 0 mark_seen=False,
log.info(f" Scanned: {total_scanned} ({rate:.0f}/sec), Matching so far: {len(matching_uids)}") ):
last_log_time = now total_scanned += 1
now = time.time()
if now - last_log_time > 10:
elapsed = now - scan_start
rate = total_scanned / elapsed if elapsed > 0 else 0
log.info(f" Scanned in {folder_name}: {total_scanned} ({rate:.0f}/sec), Matching so far: {len(matching_uids)}")
last_log_time = now
# Check if matches any domain # Check if matches any domain
from_match = matches_domain(msg.from_, DOMAINS) from_match = matches_domain(msg.from_, DOMAINS)
to_match = None to_match = None
if not from_match: if not from_match:
for to_addr in (msg.to or []): for to_addr in (msg.to or []):
to_match = matches_domain(to_addr, DOMAINS) to_match = matches_domain(to_addr, DOMAINS)
if to_match: if to_match:
break break
if from_match or to_match: if from_match or to_match:
# Also verify date since IMAP date filtering can be imprecise msg_date = msg.date
msg_date = msg.date if msg_date and hasattr(msg_date, 'date'):
if msg_date and hasattr(msg_date, 'date'): msg_date = msg_date.date()
msg_date = msg_date.date() if not msg_date or msg_date >= SINCE_DATE:
elif isinstance(msg_date, datetime): matching_uids.append((msg.uid, folder_name))
pass # ok
if not msg_date or msg_date >= SINCE_DATE:
matching_uids.append(msg.uid)
scan_elapsed = time.time() - scan_start scan_elapsed = time.time() - scan_start
log.info(f"Phase 1 complete: scanned {total_scanned} emails in {scan_elapsed/60:.1f}min") log.info(f"Phase 1 complete: scanned {total_scanned} emails in {scan_elapsed/60:.1f}min")
log.info(f"Found {len(matching_uids)} matching emails") log.info(f"Found {len(matching_uids)} matching emails across folders")
if not matching_uids: if not matching_uids:
log.info("No matching emails found. Exiting.") log.info("No matching emails found. Exiting.")
@ -400,13 +417,16 @@ def main():
total_processed = 0 total_processed = 0
total_errors = 0 total_errors = 0
def reconnect_mailbox(): current_folder = ""
"""Reconnect to IMAP with retry."""
def reconnect_mailbox(target_folder):
"""Reconnect to IMAP with retry and set folder."""
for attempt in range(5): for attempt in range(5):
try: try:
log.info(f"Reconnecting... attempt {attempt+1}") log.info(f"Reconnecting... attempt {attempt+1}")
mb = MailBox(IMAP_SERVER).login(imap_user, imap_pass, initial_folder=GMAIL_FOLDER) mb = MailBox(IMAP_SERVER).login(imap_user, imap_pass)
log.info("Reconnected!") mb.folder.set(target_folder)
log.info(f"Reconnected and folder set to {target_folder}!")
return mb return mb
except Exception as e: except Exception as e:
log.warning(f"Reconnect failed: {e}") log.warning(f"Reconnect failed: {e}")
@ -414,30 +434,35 @@ def main():
log.error("Failed to reconnect after 5 attempts") log.error("Failed to reconnect after 5 attempts")
sys.exit(1) sys.exit(1)
for idx, uid in enumerate(matching_uids, 1): for idx, (uid, f_name) in enumerate(matching_uids, 1):
if str(uid) in processed: if current_folder != f_name:
log.info(f"[{idx}/{len(matching_uids)}] SKIP (already processed) UID={uid}") try:
mailbox.folder.set(f_name)
current_folder = f_name
except Exception as e:
log.error(f"Failed to switch to folder {f_name}: {e}")
continue
progress_key = f"{f_name}:{uid}" if f_name == spam_folder else str(uid)
if progress_key in processed or str(uid) in processed:
log.info(f"[{idx}/{len(matching_uids)}] SKIP (already processed) UID={uid} in {f_name}")
continue continue
elapsed = time.time() - start_time elapsed = time.time() - start_time
if idx > 1: eta = (elapsed / (idx - 1)) * (len(matching_uids) - idx + 1) if idx > 1 else 0
eta = (elapsed / (idx - 1)) * (len(matching_uids) - idx + 1)
else:
eta = 0
log.info(f"[{idx}/{len(matching_uids)}] Downloading UID={uid} | ETA: {eta/60:.1f}m") log.info(f"[{idx}/{len(matching_uids)}] Downloading UID={uid} from {f_name} | ETA: {eta/60:.1f}m")
try: try:
# Fetch full message
msgs = list(mailbox.fetch( msgs = list(mailbox.fetch(
AND(uid=uid), AND(uid=uid),
headers_only=False, headers_only=False,
mark_seen=False, mark_seen=False,
)) ))
if not msgs: if not msgs:
log.warning(f" Could not fetch UID={uid}") log.warning(f" Could not fetch UID={uid} in {f_name}")
total_errors += 1 total_errors += 1
save_progress(uid, processed) save_progress(progress_key, processed)
continue continue
msg = msgs[0] msg = msgs[0]
@ -514,20 +539,20 @@ def main():
total_errors += 1 total_errors += 1
# Save progress # Save progress
save_progress(uid, processed) save_progress(progress_key, processed)
except Exception as e: except Exception as e:
# Check if it's a socket error — try to reconnect # Check if it's a socket error — try to reconnect
if 'socket error' in str(e) or 'EOF' in str(e) or 'abort' in str(e): if 'socket error' in str(e) or 'EOF' in str(e) or 'abort' in str(e):
log.warning(f"Connection lost, trying to reconnect...") log.warning(f"Connection lost, trying to reconnect...")
mailbox = reconnect_mailbox() mailbox = reconnect_mailbox(f_name)
# Don't mark as processed, retry next time current_folder = f_name
total_errors += 1 total_errors += 1
continue continue
else: else:
log.error(f" ERROR processing UID={uid}: {e}", exc_info=True) log.error(f" ERROR processing UID={uid} in {f_name}: {e}", exc_info=True)
total_errors += 1 total_errors += 1
save_progress(uid, processed) save_progress(progress_key, processed)
mailbox.logout() mailbox.logout()
@ -566,9 +591,8 @@ def main():
"text": "📬 <b>ВАЖНОЕ ОБНОВЛЕНИЕ ПОЧТЫ!</b>\n" + summary_text + "\n🚀 <i>Пора обновить источники в NotebookLM!</i>", "text": "📬 <b>ВАЖНОЕ ОБНОВЛЕНИЕ ПОЧТЫ!</b>\n" + summary_text + "\n🚀 <i>Пора обновить источники в NotebookLM!</i>",
"parse_mode": "HTML" "parse_mode": "HTML"
} }
# Use proxy if needed # Send directly (NanoPi routes to VPN transparently)
proxies = {"https": "socks5h://127.0.0.1:10808"} requests.post(url, json=params, timeout=20)
requests.post(url, json=params, proxies=proxies, timeout=20)
log.info("✅ Telegram notification sent.") log.info("✅ Telegram notification sent.")
except Exception as e: except Exception as e:
log.error(f"❌ Failed to send Telegram notification: {e}") log.error(f"❌ Failed to send Telegram notification: {e}")

103
master_ai_sync.py Executable file
View file

@ -0,0 +1,103 @@
#!/usr/bin/env python3
import subprocess
import os
import sys
import logging
import requests
from pathlib import Path
from datetime import datetime
# === CONFIG ===
BASE_DIR = Path("/home/matrixhasyou/mutt")
PYTHON_BIN = "/usr/bin/python3"
GOOGLE_PYTHON_BIN = "/home/matrixhasyou/swarm-services/domovoy-bot/venv-google/bin/python"
EXPORT_SCRIPT = BASE_DIR / "export_emails.py"
CONSOLIDATE_SCRIPT = BASE_DIR / "consolidate_to_mega_pdf.py"
DRIVE_SYNC_SCRIPT = Path("/home/matrixhasyou/domovoy_drive_sync.py")
MASTER_PDF = BASE_DIR / "ALL_EMAILS_CONSOLIDATED.pdf"
BOT_TOKEN = "8725618164:AAH1tGalq-pw1l0t4P5c0sdLkCVJdh7IE7M"
CHAT_ID = "197957361"
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("MasterSync")
def send_tg(text):
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
try:
requests.post(url, json={"chat_id": CHAT_ID, "text": text, "parse_mode": "HTML"},
timeout=20)
except Exception as e:
log.error(f"TG notification failed: {e}")
def run_step(name, cmd):
log.info(f"--- Step: {name} ---")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
log.error(f"Step {name} FAILED!")
log.error(result.stderr)
return False, result.stderr
log.info(f"Step {name} completed.")
return True, result.stdout
def main():
start_time = datetime.now()
log.info(f"Starting Master Sync at {start_time}")
# 1. Export Emails
ok, out_export = run_step("Export Emails", [PYTHON_BIN, str(EXPORT_SCRIPT)])
if not ok:
send_tg(f"❌ <b>Mutt Master Sync: FAILED</b>\nStep: Export Emails\nError: <code>{out_export[:200]}</code>")
return
# 2. Consolidate PDF
ok, out_consolidate = run_step("Consolidate PDF", [PYTHON_BIN, str(CONSOLIDATE_SCRIPT)])
if not ok:
send_tg(f"❌ <b>Mutt Master Sync: FAILED</b>\nStep: Consolidate PDF\nError: <code>{out_consolidate[:200]}</code>")
return
# 3. Sync to G-Drive
ok, out_drive = run_step("G-Drive Sync", [GOOGLE_PYTHON_BIN, str(DRIVE_SYNC_SCRIPT), str(MASTER_PDF)])
if not ok:
send_tg(f"❌ <b>Mutt Master Sync: FAILED</b>\nStep: G-Drive Sync\nError: <code>{out_drive[:200]}</code>")
return
end_time = datetime.now()
# Parsing Stats
new_emails = 0
total_emails = 0
for line in out_export.split('\n'):
if "Processed this run:" in line:
try: new_emails = int(line.split(':')[-1].strip())
except: pass
for line in out_consolidate.split('\n'):
if "Found" in line and "PDFs" in line:
try: total_emails = int(line.split('Found')[-1].split('PDFs')[0].strip())
except: pass
pdf_size_mb = os.path.getsize(MASTER_PDF) / (1024 * 1024)
summary = f"""
🚀 <b>AI ПОЧТОВЫЙ ТУРБО-СИНХРОН: OK</b> 🚀
📥 <b>Новых писем:</b> <code>{new_emails}</code>
📚 <b>Всего в архиве:</b> <code>{total_emails}</code>
🐘 <b>Размер PDF:</b> <code>{pdf_size_mb:.2f} МБ</code>
<b>G-Drive:</b> Обновлено
📅 {end_time.strftime("%Y-%m-%d %H:%M")}
<i>Робот закончил работу. NotebookLM готов к анализу.</i>
"""
send_tg(summary)
log.info("Master Sync completed successfully.")
# Save last sync time for web UI
sync_time_file = Path("/home/matrixhasyou/mutt/last_ai_sync.txt")
with open(sync_time_file, "w") as f:
f.write(end_time.strftime("%Y-%m-%d %H:%M:%S"))
if __name__ == "__main__":
main()