TySS-Dev/MailRelay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #5 from TySP-Dev/claude/create-data-directories-1ktvx

Browse Source 
Fix Proton export CLI integration for v1.0.6
This commit is contained in:
Tyler
2026-03-24 18:08:44 -04:00
committed by GitHub
parent a703af9743 83eb9752a5
commit 7ca1ebd466
2 changed files with 105 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+5 -4
View File
@@ -384,12 +384,13 @@ def _load_config() -> dict:
def _clean_export_dir(export_dir: Path) -> None: def _clean_export_dir(export_dir: Path) -> None:
"""Delete all files in the export directory after a sync cycle.""" """Remove the mail_* export directory tree after a sync cycle."""
for f in export_dir.iterdir(): import shutil
try: try:
f.unlink() shutil.rmtree(export_dir)
log.debug("Removed export directory: %s", export_dir.name)
except OSError as exc: except OSError as exc:
log.warning("Could not delete export file %s: %s", f.name, exc) log.warning("Could not remove export directory %s: %s", export_dir.name, exc)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
modules/exporter.py
+98 -42
View File
@@ -1,19 +1,21 @@
"""Automate proton-mail-export-cli via pexpect. """Automate proton-mail-export-cli via pexpect.
The Proton Export CLI interactive prompt sequence (observed order): The Proton Export CLI interactive prompt sequence (v1.0.6):
1. Email address 1. Username
2. Password 2. Password
3. Two-factor authentication code (only if 2FA is enabled) 3. Two-factor authentication code (only if 2FA is enabled)
4. Mailbox password (only if separate mailbox password is set) 4. Mailbox password (only if separate mailbox password is set)
5. Operation (B)ackup/(R)estore — we always send 'B'
6. "Do you wish to proceed?" — we always send 'Yes'
After authentication the CLI exports all mail to the export directory. The CLI creates the following directory structure under the base export dir:
Each message produces two files: <base>/<Email>/mail_<YYYYMMDD_HHMMSS>/
{messageID}.eml <messageID>.eml
{messageID}.metadata.json <messageID>.metadata.json
...
Because the exact prompt strings can vary between CLI versions and account run_export() returns the innermost mail_* directory so callers can
configurations, each pattern is defined as a regex so minor wording differences scan .eml files directly.
are tolerated. If the CLI changes its prompt wording, update PROMPTS below.
""" """
from pathlib import Path from pathlib import Path
@@ -37,12 +39,14 @@ EXPORT_DIR = Path(__file__).parent.parent / "data" / "exports"
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
PROMPTS = { PROMPTS = {
"email": r"[Ee]mail\s*(address)?[\s:>]+", "username": r"[Uu]sername\s*:",
"password": r"[Pp]assword[\s:>]+", "password": r"[Pp]assword\s*:",
"totp": r"[Tt]wo.factor|[Oo]ne.time|[Tt][Oo][Tt][Pp]|[Aa]uth.*code", "totp": r"[Ee]nter the code from your authenticator",
"mailbox_password": r"[Mm]ailbox\s*[Pp]assword[\s:>]+", "mailbox_password": r"[Mm]ailbox [Pp]assword\s*:",
"done": r"[Ee]xport\s*(complete|finished|done)|[Ss]uccessfully\s*export", "operation": r"[Oo]peration\s*\(\(B\)ackup",
"error": r"[Ee]rror|[Ff]ailed|[Ii]nvalid", "proceed": r"[Dd]o you wish to proceed\?",
"starting": r"Starting Export",
"error": r"[Ee]rror|[Ff]ailed|[Ii]nvalid|[Uu]nexpected",
} }
# Maximum time to wait for each prompt (seconds) # Maximum time to wait for each prompt (seconds)
@@ -66,12 +70,15 @@ def run_export(
mailbox_password: str = "", mailbox_password: str = "",
export_dir: Optional[Path] = None, export_dir: Optional[Path] = None,
) -> Path: ) -> Path:
"""Drive proton-mail-export-cli and return the export directory path. """Drive proton-mail-export-cli and return the mail_* export directory.
The CLI creates <base>/<Email>/mail_<timestamp>/ — this function finds
and returns that innermost directory.
Raises ExportError on authentication failure or unexpected CLI output. Raises ExportError on authentication failure or unexpected CLI output.
""" """
out_dir = export_dir or EXPORT_DIR base_dir = export_dir or EXPORT_DIR
out_dir.mkdir(parents=True, exist_ok=True) base_dir.mkdir(parents=True, exist_ok=True)
log.info("Starting Proton Mail export...") log.info("Starting Proton Mail export...")
@@ -80,8 +87,8 @@ def run_export(
except Exception as exc: except Exception as exc:
raise ExportError(str(exc)) from exc raise ExportError(str(exc)) from exc
cmd = [str(binary), "--export-dir", str(out_dir)] cmd = [str(binary), "--dir", str(base_dir)]
log.info("Starting Proton export CLI: %s", " ".join(cmd)) log.debug("Running: %s", " ".join(cmd))
child = pexpect.spawn( child = pexpect.spawn(
cmd[0], cmd[1:], timeout=PROMPT_TIMEOUT, encoding="utf-8" cmd[0], cmd[1:], timeout=PROMPT_TIMEOUT, encoding="utf-8"
@@ -100,13 +107,17 @@ def run_export(
raise ExportError( raise ExportError(
f"CLI exited with code {child.exitstatus}. Output: {output.strip()}" f"CLI exited with code {child.exitstatus}. Output: {output.strip()}"
) from exc ) from exc
# EOF after a successful export is normal log.debug("CLI process finished (EOF).")
log.info("CLI process finished (EOF).")
log.info("Export complete. Files in: %s", out_dir) mail_dir = _find_mail_dir(base_dir, email)
return out_dir log.info("Proton Mail export complete. Files in: %s", mail_dir)
return mail_dir
# ---------------------------------------------------------------------------
# Interactive prompt driver
# ---------------------------------------------------------------------------
def _drive_cli( def _drive_cli(
child: pexpect.spawn, child: pexpect.spawn,
email: str, email: str,
@@ -116,14 +127,16 @@ def _drive_cli(
) -> None: ) -> None:
"""Respond to each interactive prompt in sequence.""" """Respond to each interactive prompt in sequence."""
patterns = [ patterns = [
pexpect.TIMEOUT, pexpect.TIMEOUT, # 0
pexpect.EOF, pexpect.EOF, # 1
PROMPTS["email"], PROMPTS["username"], # 2
PROMPTS["password"], PROMPTS["password"], # 3
PROMPTS["totp"], PROMPTS["totp"], # 4
PROMPTS["mailbox_password"], PROMPTS["mailbox_password"], # 5
PROMPTS["done"], PROMPTS["operation"], # 6
PROMPTS["error"], PROMPTS["proceed"], # 7
PROMPTS["starting"], # 8
PROMPTS["error"], # 9
] ]
totp_sent = False totp_sent = False
@@ -138,13 +151,11 @@ def _drive_cli(
if idx == 1: # EOF — process exited if idx == 1: # EOF — process exited
return return
if idx == 2: # email prompt if idx == 2: # Username
log.debug("CLI requested email.") log.debug("CLI requested username.")
child.sendline(email) child.sendline(email)
elif idx == 3: # password prompt elif idx == 3: # Password (account or mailbox)
# The CLI may show a password prompt for both the account password
# and the mailbox password. We track which we've already sent.
if not mailbox_sent and mailbox_password and totp_sent: if not mailbox_sent and mailbox_password and totp_sent:
log.debug("CLI requested mailbox password.") log.debug("CLI requested mailbox password.")
child.sendline(mailbox_password) child.sendline(mailbox_password)
@@ -153,21 +164,30 @@ def _drive_cli(
log.debug("CLI requested account password.") log.debug("CLI requested account password.")
child.sendline(password) child.sendline(password)
elif idx == 4: # TOTP prompt elif idx == 4: # TOTP
log.debug("CLI requested TOTP code.") log.debug("CLI requested TOTP code.")
child.sendline(totp_code) child.sendline(totp_code)
totp_sent = True totp_sent = True
elif idx == 5: # explicit mailbox password prompt elif idx == 5: # Explicit mailbox password prompt
log.debug("CLI requested mailbox password (explicit prompt).") log.debug("CLI requested mailbox password (explicit prompt).")
child.sendline(mailbox_password or "") child.sendline(mailbox_password or "")
mailbox_sent = True mailbox_sent = True
elif idx == 6: # export done elif idx == 6: # Operation prompt
log.info("CLI reported export complete.") log.debug("CLI requested operation — sending 'B' for backup.")
child.sendline("B")
elif idx == 7: # "Do you wish to proceed?"
log.debug("CLI asked to confirm export path — sending 'Yes'.")
child.sendline("Yes")
elif idx == 8: # "Starting Export" — export is underway, wait for EOF
log.debug("CLI reported export started — waiting for completion.")
child.expect(pexpect.EOF, timeout=EXPORT_TIMEOUT)
return return
elif idx == 7: # error line elif idx == 9: # Error line
snippet = (child.before or "").strip().splitlines()[-1] snippet = (child.before or "").strip().splitlines()[-1]
raise ExportError(f"CLI reported an error: {snippet}") raise ExportError(f"CLI reported an error: {snippet}")
@@ -176,6 +196,42 @@ def _drive_cli(
# Helpers # Helpers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _find_mail_dir(base_dir: Path, email: str) -> Path:
"""Locate the mail_* directory the CLI created under base_dir.
The CLI creates: base_dir/<Email>/mail_<YYYYMMDD_HHMMSS>/
The email folder name may differ in capitalisation from the supplied
address, so we match case-insensitively.
"""
# Find the email-named subdirectory
email_dirs = [
d for d in base_dir.iterdir()
if d.is_dir() and d.name.lower() == email.lower()
]
if not email_dirs:
# Fallback: take any subdirectory (there should be exactly one)
email_dirs = [d for d in base_dir.iterdir() if d.is_dir()]
if not email_dirs:
raise ExportError(
f"Export finished but no output directory found under {base_dir}."
)
email_dir = email_dirs[0]
mail_dirs = sorted(
[d for d in email_dir.iterdir() if d.is_dir() and d.name.startswith("mail_")],
key=lambda d: d.stat().st_mtime,
reverse=True,
)
if not mail_dirs:
raise ExportError(
f"Export finished but no mail_* directory found under {email_dir}."
)
return mail_dirs[0]
class _PexpectLogger: class _PexpectLogger:
"""Thin adapter so pexpect writes its read data to our logger at DEBUG.""" """Thin adapter so pexpect writes its read data to our logger at DEBUG."""