Fix exporter: correct --dir flag, handle full prompt sequence, find mail_* dir
- Replace --export-dir (doesn't exist) with --dir (correct v1.0.6 flag) - Add handlers for the operation prompt (send 'B') and 'Do you wish to proceed?' (send 'Yes') which were previously unhandled - After export, locate the actual mail_<timestamp> directory the CLI creates under <base>/<Email>/ and return it so callers get the right path - Update prompt pattern for 'Username:' (CLI uses 'Username' not 'Email') - _clean_export_dir now uses shutil.rmtree since the returned path is the mail_* subdir, not a flat file directory https://claude.ai/code/session_01KjaNo9RXevw6x1DjJD8mj6
This commit is contained in:
Claude
@@ -384,12 +384,13 @@ def _load_config() -> dict:
|
||||
|
||||
|
||||
def _clean_export_dir(export_dir: Path) -> None:
|
||||
"""Delete all files in the export directory after a sync cycle."""
|
||||
for f in export_dir.iterdir():
|
||||
try:
|
||||
f.unlink()
|
||||
except OSError as exc:
|
||||
log.warning("Could not delete export file %s: %s", f.name, exc)
|
||||
"""Remove the mail_* export directory tree after a sync cycle."""
|
||||
import shutil
|
||||
try:
|
||||
shutil.rmtree(export_dir)
|
||||
log.debug("Removed export directory: %s", export_dir.name)
|
||||
except OSError as exc:
|
||||
log.warning("Could not remove export directory %s: %s", export_dir.name, exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
+98
-42
@@ -1,19 +1,21 @@
|
||||
"""Automate proton-mail-export-cli via pexpect.
|
||||
|
||||
The Proton Export CLI interactive prompt sequence (observed order):
|
||||
1. Email address
|
||||
The Proton Export CLI interactive prompt sequence (v1.0.6):
|
||||
1. Username
|
||||
2. Password
|
||||
3. Two-factor authentication code (only if 2FA is enabled)
|
||||
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.
|
||||
Each message produces two files:
|
||||
{messageID}.eml
|
||||
{messageID}.metadata.json
|
||||
The CLI creates the following directory structure under the base export dir:
|
||||
<base>/<Email>/mail_<YYYYMMDD_HHMMSS>/
|
||||
<messageID>.eml
|
||||
<messageID>.metadata.json
|
||||
...
|
||||
|
||||
Because the exact prompt strings can vary between CLI versions and account
|
||||
configurations, each pattern is defined as a regex so minor wording differences
|
||||
are tolerated. If the CLI changes its prompt wording, update PROMPTS below.
|
||||
run_export() returns the innermost mail_* directory so callers can
|
||||
scan .eml files directly.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
@@ -37,12 +39,14 @@ EXPORT_DIR = Path(__file__).parent.parent / "data" / "exports"
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PROMPTS = {
|
||||
"email": r"[Ee]mail\s*(address)?[\s:>]+",
|
||||
"password": r"[Pp]assword[\s:>]+",
|
||||
"totp": r"[Tt]wo.factor|[Oo]ne.time|[Tt][Oo][Tt][Pp]|[Aa]uth.*code",
|
||||
"mailbox_password": r"[Mm]ailbox\s*[Pp]assword[\s:>]+",
|
||||
"done": r"[Ee]xport\s*(complete|finished|done)|[Ss]uccessfully\s*export",
|
||||
"error": r"[Ee]rror|[Ff]ailed|[Ii]nvalid",
|
||||
"username": r"[Uu]sername\s*:",
|
||||
"password": r"[Pp]assword\s*:",
|
||||
"totp": r"[Ee]nter the code from your authenticator",
|
||||
"mailbox_password": r"[Mm]ailbox [Pp]assword\s*:",
|
||||
"operation": r"[Oo]peration\s*\(\(B\)ackup",
|
||||
"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)
|
||||
@@ -66,12 +70,15 @@ def run_export(
|
||||
mailbox_password: str = "",
|
||||
export_dir: Optional[Path] = None,
|
||||
) -> 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.
|
||||
"""
|
||||
out_dir = export_dir or EXPORT_DIR
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
base_dir = export_dir or EXPORT_DIR
|
||||
base_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
log.info("Starting Proton Mail export...")
|
||||
|
||||
@@ -80,8 +87,8 @@ def run_export(
|
||||
except Exception as exc:
|
||||
raise ExportError(str(exc)) from exc
|
||||
|
||||
cmd = [str(binary), "--export-dir", str(out_dir)]
|
||||
log.info("Starting Proton export CLI: %s", " ".join(cmd))
|
||||
cmd = [str(binary), "--dir", str(base_dir)]
|
||||
log.debug("Running: %s", " ".join(cmd))
|
||||
|
||||
child = pexpect.spawn(
|
||||
cmd[0], cmd[1:], timeout=PROMPT_TIMEOUT, encoding="utf-8"
|
||||
@@ -100,13 +107,17 @@ def run_export(
|
||||
raise ExportError(
|
||||
f"CLI exited with code {child.exitstatus}. Output: {output.strip()}"
|
||||
) from exc
|
||||
# EOF after a successful export is normal
|
||||
log.info("CLI process finished (EOF).")
|
||||
log.debug("CLI process finished (EOF).")
|
||||
|
||||
log.info("Export complete. Files in: %s", out_dir)
|
||||
return out_dir
|
||||
mail_dir = _find_mail_dir(base_dir, email)
|
||||
log.info("Proton Mail export complete. Files in: %s", mail_dir)
|
||||
return mail_dir
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Interactive prompt driver
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _drive_cli(
|
||||
child: pexpect.spawn,
|
||||
email: str,
|
||||
@@ -116,14 +127,16 @@ def _drive_cli(
|
||||
) -> None:
|
||||
"""Respond to each interactive prompt in sequence."""
|
||||
patterns = [
|
||||
pexpect.TIMEOUT,
|
||||
pexpect.EOF,
|
||||
PROMPTS["email"],
|
||||
PROMPTS["password"],
|
||||
PROMPTS["totp"],
|
||||
PROMPTS["mailbox_password"],
|
||||
PROMPTS["done"],
|
||||
PROMPTS["error"],
|
||||
pexpect.TIMEOUT, # 0
|
||||
pexpect.EOF, # 1
|
||||
PROMPTS["username"], # 2
|
||||
PROMPTS["password"], # 3
|
||||
PROMPTS["totp"], # 4
|
||||
PROMPTS["mailbox_password"], # 5
|
||||
PROMPTS["operation"], # 6
|
||||
PROMPTS["proceed"], # 7
|
||||
PROMPTS["starting"], # 8
|
||||
PROMPTS["error"], # 9
|
||||
]
|
||||
|
||||
totp_sent = False
|
||||
@@ -138,13 +151,11 @@ def _drive_cli(
|
||||
if idx == 1: # EOF — process exited
|
||||
return
|
||||
|
||||
if idx == 2: # email prompt
|
||||
log.debug("CLI requested email.")
|
||||
if idx == 2: # Username
|
||||
log.debug("CLI requested username.")
|
||||
child.sendline(email)
|
||||
|
||||
elif idx == 3: # password prompt
|
||||
# The CLI may show a password prompt for both the account password
|
||||
# and the mailbox password. We track which we've already sent.
|
||||
elif idx == 3: # Password (account or mailbox)
|
||||
if not mailbox_sent and mailbox_password and totp_sent:
|
||||
log.debug("CLI requested mailbox password.")
|
||||
child.sendline(mailbox_password)
|
||||
@@ -153,21 +164,30 @@ def _drive_cli(
|
||||
log.debug("CLI requested account password.")
|
||||
child.sendline(password)
|
||||
|
||||
elif idx == 4: # TOTP prompt
|
||||
elif idx == 4: # TOTP
|
||||
log.debug("CLI requested TOTP code.")
|
||||
child.sendline(totp_code)
|
||||
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).")
|
||||
child.sendline(mailbox_password or "")
|
||||
mailbox_sent = True
|
||||
|
||||
elif idx == 6: # export done
|
||||
log.info("CLI reported export complete.")
|
||||
elif idx == 6: # Operation prompt
|
||||
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
|
||||
|
||||
elif idx == 7: # error line
|
||||
elif idx == 9: # Error line
|
||||
snippet = (child.before or "").strip().splitlines()[-1]
|
||||
raise ExportError(f"CLI reported an error: {snippet}")
|
||||
|
||||
@@ -176,6 +196,42 @@ def _drive_cli(
|
||||
# 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:
|
||||
"""Thin adapter so pexpect writes its read data to our logger at DEBUG."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user