administrator/mail_exp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve export diagnostics and splash

Browse Source
This commit is contained in:
allebonvi
2026-04-21 19:17:04 +02:00
parent 908f679317
commit 8009b9d16f
3 changed files with 381 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

387
mail_exporter.py
View File

@@ -18,6 +18,7 @@ from tkinter import ttk
try:
import pythoncom
import win32timezone # required by pywin32 when Outlook returns timezone-aware COM dates
import win32com.client
except ImportError:
pythoncom = None
@@ -30,6 +31,7 @@ MAIL_CLASS = 43 # OlObjectClass.olMail
MAX_LOG_LINES = 3000
PREFS_APP_DIR = "OutlookExporter"
PREFS_FILE_NAME = "prefs.json"
LOG_FILE_NAME = "mail_exporter.log"
OUTLOOK_NOT_RUNNING_MESSAGE = (
"Outlook non risulta aperto.\n\n"
"Apri Microsoft Outlook con il profilo email da esportare, attendi che abbia completato il caricamento, "
@@ -61,9 +63,21 @@ ALLOWED_ATTACHMENT_EXTENSIONS = {
".potx",
".potm",
".rtf",
".fodp",
".fods",
".fodt",
".odb",
".odf",
".odg",
".odt",
".ods",
".odp",
".otg",
".oth",
".otp",
".ots",
".ott",
".sxc",
".jpg",
".jpeg",
".png",
@@ -73,6 +87,24 @@ ALLOWED_ATTACHMENT_EXTENSIONS = {
".tiff",
".webp",
}
IMAGE_ATTACHMENT_EXTENSIONS = {
".jpg",
".jpeg",
".png",
".gif",
".bmp",
".tif",
".tiff",
".webp",
}
MAPI_ATTACHMENT_HIDDEN = "http://schemas.microsoft.com/mapi/proptag/0x7FFE000B"
MAPI_ATTACH_CONTENT_ID = "http://schemas.microsoft.com/mapi/proptag/0x3712001F"
MAPI_ATTACH_CONTENT_LOCATION = "http://schemas.microsoft.com/mapi/proptag/0x3713001F"
MAPI_ATTACH_CONTENT_DISPOSITION = "http://schemas.microsoft.com/mapi/proptag/0x3716001F"
INLINE_ATTACHMENT_NAME_RE = re.compile(
r"^(image\d{3,}|att\d{5}|oledata|logo|spacer|facebook|linkedin|twitter|instagram|youtube)\b",
re.IGNORECASE,
)
@dataclass
@@ -111,6 +143,10 @@ class PreferenceStore:
base_dir = Path(appdata) if appdata else Path.home() / ".config"
return base_dir / PREFS_APP_DIR / PREFS_FILE_NAME
@classmethod
def app_data_dir(cls) -> Path:
return cls._prefs_path().parent
def _load_or_create(self) -> dict:
if self.path.exists():
try:
@@ -279,9 +315,10 @@ class NameCodec:
return f"mail_{stamp}_{seq:05d}"
@classmethod
def mail_folder_name(cls, base_mail_name: str, subject: str) -> str:
def mail_folder_name(cls, base_mail_name: str, subject: str, attachment_count: int = 0) -> str:
safe_subject = cls.sanitize(subject or "senza oggetto", max_len=50)
return f"{base_mail_name}__{safe_subject}"
attachment_marker = f"CON_ALLEGATI_{attachment_count:02d}" if attachment_count else "SOLO_MAIL"
return f"{base_mail_name}__{attachment_marker}__{safe_subject}"
@classmethod
def attachment_name(cls, base_mail_name: str, original_filename: str, att_index: int) -> str:
@@ -318,6 +355,8 @@ class Exporter:
self.global_seq = 0
self.exported_mail_count = 0
self.exported_attachment_count = 0
self.datetime_diagnostic_count = 0
self.datetime_diagnostic_limit = 20
def cancel(self) -> None:
self.cancel_requested = True
@@ -406,12 +445,16 @@ class Exporter:
return local_mail_count, local_attachment_count
def _export_mail(self, item: object, folder_out_path: Path) -> Tuple[int, int]:
received_dt = self._convert_received_time(self._safe_get(item, "ReceivedTime"))
received_dt = self._get_best_mail_datetime(item)
subject = self._safe_get(item, "Subject") or ""
body = self._safe_get(item, "Body") or ""
html_body = self._safe_get(item, "HTMLBody") or ""
exportable_attachments = self._collect_exportable_attachments(item, html_body)
base_name = NameCodec.email_base_name(received_dt, self.global_seq)
mail_dir = self._unique_path(folder_out_path / NameCodec.mail_folder_name(base_name, subject))
mail_dir = self._unique_path(
folder_out_path / NameCodec.mail_folder_name(base_name, subject, len(exportable_attachments))
)
mail_dir.mkdir(parents=True, exist_ok=True)
txt_path = mail_dir / f"{base_name}.txt"
@@ -426,24 +469,11 @@ class Exporter:
self.app.log(f"Mail salvata: {txt_path.name}")
attachment_count = 0
attachments = self._safe_get(item, "Attachments")
if attachments is not None:
try:
attachment_total = attachments.Count
except Exception as exc:
self.app.log(f" Errore leggendo il numero di allegati: {exc}")
attachment_total = 0
if exportable_attachments:
attachments_dir = mail_dir / "allegati"
for att_index in range(1, attachment_total + 1):
for att_index, attachment, original_filename in exportable_attachments:
self.check_cancel()
try:
attachment = attachments.Item(att_index)
original_filename = self._safe_get(attachment, "FileName") or f"attachment_{att_index}"
if not NameCodec.is_allowed_attachment(original_filename):
self.app.log(f" Allegato ignorato: {original_filename}")
continue
export_name = NameCodec.attachment_name(base_name, original_filename, att_index)
attachments_dir.mkdir(parents=True, exist_ok=True)
export_path = self._unique_path(attachments_dir / export_name)
@@ -455,6 +485,173 @@ class Exporter:
return 1, attachment_count
def _collect_exportable_attachments(self, item: object, html_body: str) -> List[Tuple[int, object, str]]:
exportable = []
attachments = self._safe_get(item, "Attachments")
if attachments is None:
return exportable
try:
attachment_total = attachments.Count
except Exception as exc:
self.app.log(f" Errore leggendo il numero di allegati: {exc}")
return exportable
for att_index in range(1, attachment_total + 1):
self.check_cancel()
try:
attachment = attachments.Item(att_index)
original_filename = self._safe_get(attachment, "FileName") or f"attachment_{att_index}"
inline_score, inline_reasons = self._inline_signature_score(
attachment, original_filename, html_body
)
if inline_score >= 3:
self.app.log(
f" Allegato inline/firma ignorato: {original_filename} | "
f"score={inline_score} | motivi={', '.join(inline_reasons)}"
)
elif NameCodec.is_allowed_attachment(original_filename):
exportable.append((att_index, attachment, original_filename))
else:
self.app.log(f" Allegato ignorato: {original_filename}")
except Exception as exc:
self.app.log(f" Errore leggendo allegato {att_index}: {exc}")
return exportable
def _inline_signature_score(self, attachment: object, filename: str, html_body: str) -> Tuple[int, List[str]]:
ext = NameCodec.sanitize_extension(os.path.splitext(filename or "")[1])
if ext not in IMAGE_ATTACHMENT_EXTENSIONS:
return 0, []
score = 0
reasons = []
html_lower = (html_body or "").lower()
filename_lower = os.path.basename(filename or "").lower()
hidden = self._get_attachment_mapi_property(attachment, MAPI_ATTACHMENT_HIDDEN)
if hidden is True:
score += 4
reasons.append("hidden")
content_id = self._get_attachment_mapi_property(attachment, MAPI_ATTACH_CONTENT_ID)
if self._has_value(content_id):
cid = str(content_id).strip()
if self._html_references_cid(html_lower, cid):
score += 3
reasons.append("cid nel corpo HTML")
else:
score += 1
reasons.append("cid presente")
content_location = self._get_attachment_mapi_property(attachment, MAPI_ATTACH_CONTENT_LOCATION)
if self._has_value(content_location):
location = str(content_location).strip()
if location.lower() in html_lower:
score += 2
reasons.append("content-location nel corpo HTML")
else:
score += 1
reasons.append("content-location presente")
disposition = self._get_attachment_mapi_property(attachment, MAPI_ATTACH_CONTENT_DISPOSITION)
if self._has_value(disposition) and "inline" in str(disposition).lower():
score += 3
reasons.append("disposition inline")
if filename_lower and filename_lower in html_lower:
score += 2
reasons.append("nome file nel corpo HTML")
stem = os.path.splitext(filename_lower)[0]
if INLINE_ATTACHMENT_NAME_RE.match(stem or "") is not None:
score += 2
reasons.append("nome tipico inline/firma")
return score, reasons
@staticmethod
def _html_references_cid(html_lower: str, content_id: str) -> bool:
if not content_id:
return False
cid = content_id.strip().strip("<>").lower()
if not cid:
return False
return f"cid:{cid}" in html_lower or cid in html_lower
def _get_attachment_mapi_property(self, attachment: object, schema: str):
try:
accessor = attachment.PropertyAccessor
return accessor.GetProperty(schema)
except Exception:
return None
@staticmethod
def _has_value(value: object) -> bool:
if value is None:
return False
try:
return bool(str(value).strip())
except Exception:
return True
def _get_best_mail_datetime(self, item: object) -> Optional[datetime]:
for attr in ("ReceivedTime", "SentOn", "CreationTime", "LastModificationTime"):
value = self._safe_get(item, attr)
converted = self._convert_outlook_datetime(value)
if converted is not None:
if attr != "ReceivedTime":
self.app.log(f" Data ReceivedTime non disponibile, uso {attr}.")
return converted
self._log_datetime_diagnostics(item)
return None
def _log_datetime_diagnostics(self, item: object) -> None:
if self.datetime_diagnostic_count >= self.datetime_diagnostic_limit:
if self.datetime_diagnostic_count == self.datetime_diagnostic_limit:
self.app.log(
" Diagnostica date: limite raggiunto, ulteriori mail senza data non verranno dettagliate."
)
self.datetime_diagnostic_count += 1
return
self.datetime_diagnostic_count += 1
subject = self._safe_get(item, "Subject") or "(senza oggetto)"
self.app.log(f" Diagnostica date mail #{self.global_seq}: {subject}")
for label, value in self._mail_identity_details(item):
self.app.log(f" {label}: {value}")
for attr in ("ReceivedTime", "SentOn", "CreationTime", "LastModificationTime"):
ok, value_or_error = self._safe_get_with_error(item, attr)
if not ok:
self.app.log(f" {attr}: ERRORE accesso: {value_or_error}")
continue
value = value_or_error
self.app.log(
f" {attr}: tipo={type(value).__name__}; repr={repr(value)}; str={self._safe_str(value)}"
)
def _mail_identity_details(self, item: object) -> List[Tuple[str, str]]:
parent = self._safe_get(item, "Parent")
folder_name = self._safe_get(parent, "Name") if parent is not None else None
store_id = self._safe_get(parent, "StoreID") if parent is not None else None
details = [
("EntryID", self._safe_str(self._safe_get(item, "EntryID"))),
("ConversationID", self._safe_str(self._safe_get(item, "ConversationID"))),
("InternetMessageID", self._safe_str(self._get_mail_mapi_property(item, "http://schemas.microsoft.com/mapi/proptag/0x1035001F"))),
("Cartella Outlook", self._safe_str(folder_name)),
("StoreID", self._safe_str(store_id)),
]
return details
def _get_mail_mapi_property(self, item: object, schema: str):
try:
accessor = item.PropertyAccessor
return accessor.GetProperty(schema)
except Exception:
return None
@staticmethod
def _unique_path(path: Path) -> Path:
if not path.exists():
@@ -478,7 +675,24 @@ class Exporter:
return None
@staticmethod
def _convert_received_time(value) -> Optional[datetime]:
def _safe_get_with_error(obj: object, attr: str) -> Tuple[bool, object]:
try:
return True, getattr(obj, attr)
except Exception as exc:
return False, exc
@staticmethod
def _safe_str(value: object, max_len: int = 300) -> str:
try:
text = str(value)
except Exception as exc:
text = f"<errore str: {exc}>"
if len(text) > max_len:
return text[:max_len] + "..."
return text
@staticmethod
def _convert_outlook_datetime(value) -> Optional[datetime]:
if value is None:
return None
if isinstance(value, datetime):
@@ -492,6 +706,20 @@ class Exporter:
value.minute,
value.second,
)
except Exception:
pass
for converter in (getattr(value, "Format", None), getattr(value, "strftime", None)):
if converter is None:
continue
try:
text = str(converter("%Y-%m-%d %H:%M:%S"))
return datetime.strptime(text, "%Y-%m-%d %H:%M:%S")
except Exception:
continue
try:
return datetime.fromisoformat(str(value))
except Exception:
return None
@@ -517,9 +745,12 @@ class OutlookExporterApp(ctk.CTk):
self.connected = False
self.tree_folder_map = {}
self.tree_path_map = {}
self.empty_folder_alert_shown = set()
self.selected_folder_obj = None
default_output_dir = Path.home() / "Desktop" / "outlook_export"
self.prefs = PreferenceStore(default_output_dir)
self.log_file_path = PreferenceStore.app_data_dir() / LOG_FILE_NAME
self._init_file_log()
self.selected_folder_display = ctk.StringVar(value="Nessuna cartella selezionata")
self.output_dir_var = ctk.StringVar(value=self.prefs.get_output_dir())
self.status_var = ctk.StringVar(value="Pronto")
@@ -529,6 +760,7 @@ class OutlookExporterApp(ctk.CTk):
self._build_ui()
self.hide_startup_splash()
self.deiconify()
self.close_pyinstaller_splash()
self.after(100, self.process_ui_queue)
self.protocol("WM_DELETE_WINDOW", self.on_close)
@@ -574,6 +806,15 @@ class OutlookExporterApp(ctk.CTk):
pass
self.splash = None
@staticmethod
def close_pyinstaller_splash() -> None:
try:
import pyi_splash
pyi_splash.close()
except Exception:
pass
def _build_ui(self) -> None:
self.grid_columnconfigure(0, weight=1)
self.grid_rowconfigure(1, weight=1)
@@ -741,6 +982,7 @@ class OutlookExporterApp(ctk.CTk):
self.tree.delete(*self.tree.get_children())
self.tree_folder_map.clear()
self.tree_path_map.clear()
self.empty_folder_alert_shown.clear()
self.selected_folder_obj = None
self.selected_folder_display.set("Nessuna cartella selezionata")
self.btn_export.configure(state="disabled")
@@ -796,6 +1038,89 @@ class OutlookExporterApp(ctk.CTk):
self.selected_folder_display.set(f"Selezionata: {folder_path}")
self.btn_export.configure(state="normal")
self.maybe_show_empty_folder_server_alert(node_id, folder_obj)
def maybe_show_empty_folder_server_alert(self, node_id: str, folder_obj: object) -> None:
if node_id in self.empty_folder_alert_shown:
return
mail_count = OutlookService.count_mail_items(folder_obj)
if mail_count != 0:
return
self.empty_folder_alert_shown.add(node_id)
self.show_empty_folder_server_alert()
def show_empty_folder_server_alert(self) -> None:
dialog = ctk.CTkToplevel(self)
dialog.title("Cartella Outlook vuota")
dialog.geometry("720x430")
dialog.minsize(640, 380)
dialog.transient(self)
dialog.grab_set()
dialog.grid_columnconfigure(0, weight=0)
dialog.grid_columnconfigure(1, weight=1)
dialog.grid_rowconfigure(1, weight=1)
alert = ctk.CTkLabel(
dialog,
text="!",
width=72,
height=72,
corner_radius=36,
fg_color="#b45309",
text_color="white",
font=ctk.CTkFont(size=42, weight="bold"),
)
alert.grid(row=0, column=0, rowspan=2, sticky="n", padx=(24, 16), pady=(24, 12))
ctk.CTkLabel(
dialog,
text="La cartella selezionata risulta vuota",
font=ctk.CTkFont(size=22, weight="bold"),
anchor="w",
).grid(row=0, column=1, sticky="ew", padx=(0, 24), pady=(26, 8))
message = (
"La cartella selezionata risulta vuota, ma potrebbero esserci email sul server di Outlook "
"che non hai sincronizzato.\n\n"
"Per sincronizzare vai su:\n\n"
"File -> Impostazioni account -> Impostazioni di sincronizzazione e Account -> Tutto\n\n"
"Poi attendi la sincronizzazione completa e riprova l'esportazione.\n\n"
"Nota: se Outlook mostra un link tipo 'ci sono altri elementi sul server', il programma puo "
"vedere solo le email gia sincronizzate localmente."
)
ctk.CTkLabel(dialog, text=message, justify="left", anchor="nw", wraplength=560).grid(
row=1, column=1, sticky="nsew", padx=(0, 24), pady=(0, 16)
)
buttons = ctk.CTkFrame(dialog, fg_color="transparent")
buttons.grid(row=2, column=0, columnspan=2, sticky="ew", padx=24, pady=(0, 24))
buttons.grid_columnconfigure(0, weight=1)
buttons.grid_columnconfigure(1, weight=0)
buttons.grid_columnconfigure(2, weight=0)
def open_outlook() -> None:
try:
os.startfile("outlook")
except Exception as exc:
messagebox.showerror(APP_TITLE, f"Impossibile aprire Outlook.\n\n{exc}")
ctk.CTkButton(buttons, text="Apri Outlook", width=140, command=open_outlook).grid(
row=0, column=1, sticky="e", padx=(0, 12)
)
ctk.CTkButton(
buttons,
text="Chiudi",
width=120,
command=dialog.destroy,
fg_color="#5f6368",
hover_color="#4b4f52",
).grid(row=0, column=2, sticky="e")
dialog.protocol("WM_DELETE_WINDOW", dialog.destroy)
dialog.after(100, dialog.focus_force)
def expand_all(self) -> None:
def _expand(node_id: str) -> None:
@@ -1085,9 +1410,23 @@ class OutlookExporterApp(ctk.CTk):
self._write_log(message)
def _init_file_log(self) -> None:
try:
self.log_file_path.parent.mkdir(parents=True, exist_ok=True)
header = (
f"MailExporter log avviato: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
f"File log: {self.log_file_path}\n"
"=" * 80
+ "\n"
)
self.log_file_path.write_text(header, encoding="utf-8", errors="replace")
except Exception:
pass
def _write_log(self, message: str) -> None:
timestamp = datetime.now().strftime("%H:%M:%S")
line = f"[{timestamp}] {message}\n"
self._append_file_log(line)
self.log_box.configure(state="normal")
self.log_box.insert("end", line)
current_text = self.log_box.get("1.0", "end")
@@ -1100,6 +1439,14 @@ class OutlookExporterApp(ctk.CTk):
self.log_box.configure(state="disabled")
self.pump_ui()
def _append_file_log(self, line: str) -> None:
try:
self.log_file_path.parent.mkdir(parents=True, exist_ok=True)
with self.log_file_path.open("a", encoding="utf-8", errors="replace") as fh:
fh.write(line)
except Exception:
pass
def set_status(self, message: str) -> None:
if threading.get_ident() != self.main_thread_id:
self.ui_queue.put(("status", message))