MailSweep Python Domain API Reference

MailSweep provides side-effect free domain modules in Python without external package dependencies.

---

1. Quick Start: One-Shot Pipeline Execution

The MailboxPipeline.connect() context manager coordinates IMAP connection lifecycles, scanning, classification, safety auditing, backup creation, and non-destructive trashing:

from mailsweep import MailboxPipeline, load_env

env_vars = load_env()

with MailboxPipeline.connect(env_vars) as pipeline:
    # Run full sweep (Dry-Run by default)
    summary = pipeline.sweep(
        folders=["INBOX"],
        action="trash",
        backup=True,
        dry_run=True
    )
    print(summary.render_text())
    
    # Assert zero critical personal or financial messages were affected
    assert summary.is_clean, "Safety audit failed!"

---

2. Step-by-Step Domain Lifecycle

You can execute each phase of the domain lifecycle independently on the pipeline:

from mailsweep import (
    MailboxPipeline,
    UnsubscribeHub,
    SafetyAuditor,
    load_env,
)

env_vars = load_env()

with MailboxPipeline.connect(env_vars) as pipeline:
    # 1. Scan mailbox headers (in-memory, returns ScanResult)
    scan_res = pipeline.scan(folders=["INBOX"])

    # 2. Evaluate 13-language safety tiers and build a typed CleanupPlan
    plan = pipeline.plan(scan_res)

    # 3. Audit safety before execution (returns SafetyAuditReport)
    audit_report = pipeline.audit(plan)
    assert audit_report.is_clean, "Critical messages detected in cleanup targets!"

    # 4. Create an offline SHA-256 verified backup archive (returns BackupArchive)
    archive = pipeline.backup(
        plan=plan,
        output_dir="backups",
        user_identifier="myuser"
    )

    # 5. Non-destructively move targeted clutter to Trash (returns CleanupReport)
    cleanup_report = pipeline.clean(
        plan=plan,
        action="trash",
        dry_run=False
    )

    # 6. Restore from an offline backup archive with SHA-256 integrity verification
    restored_count = pipeline.restore(
        archive=archive,
        query="important"
    )

    # 7. Aggregate brand subscriptions and verify unsubscription URLs
    hub = UnsubscribeHub()
    brands = hub.aggregate([scan_res])
    verified_brands = hub.verify_links(brands)
    html_dashboard = hub.render_dashboard(verified_brands)

---

3. Core Domain Models & Modules

MailboxPipeline (mailsweep.pipeline)

The top-level coordinator managing end-to-end mailbox operations.

• MailboxPipeline.connect(config, transport=None, rules=None, user_domains=(), property_keywords=()) \rightarrow Context manager yielding connected MailboxPipeline.
• MailboxPipeline.offline(config=None, rules=None, user_domains=(), property_keywords=()) \rightarrow Offline MailboxPipeline instance.
• pipeline.sweep(folders, action, dest_folder, trash_folder, backup_dir, backup, create_zip, dry_run, save_snapshots, cache_dir, user_identifier, fresh, progress_callback) \rightarrow SweepSummaryReport
• pipeline.scan(folders, save_path, cache_dir, user_identifier, fresh, progress_callback) \rightarrow ScanResult
• pipeline.plan(scan_data, save_path) \rightarrow CleanupPlan
• pipeline.audit(target) \rightarrow SafetyAuditReport
• pipeline.backup(plan, output_dir, user_identifier, create_zip, fresh, progress_callback) \rightarrow BackupArchive
• pipeline.clean(plan, action, dest_folder, trash_folder, dry_run, progress_callback) \rightarrow CleanupReport
• pipeline.restore(archive, query, sender, subject, dry_run, progress_callback) \rightarrow int
• pipeline.verify_backup(archive, scan_result=None, folders=None, trash_folder=None) \rightarrow BackupVerificationReport
• pipeline.unsubscribe(sources=None, summary_path=..., backups_dir=..., output_path=..., skip_check=False, default_user_email="") \rightarrow tuple[BrandSubscription, ...]

EmailHeader (mailsweep.gateway)

Immutable domain model encapsulating parsed message headers, attachments, and Gmail smart labels.

• header.dedup_key \rightarrow str (Canonical deduplication key across virtual folders)
• header.normalized_labels \rightarrow tuple[str, ...] (Lowercase Gmail labels)
• header.has_label(label) \rightarrow bool
• header.is_promotional \rightarrow bool
• header.is_social \rightarrow bool
• header.with_merged_labels(new_labels) \rightarrow EmailHeader (Returns new instance with merged labels)

ScanResult (mailsweep.gateway)

Immutable domain model encapsulating mailbox header scan data and UIDVALIDITY snapshots.

• ScanResult.load(source) \rightarrow ScanResult
• scan_result.save(output_path) \rightarrow None
• scan_result.to_dict() \rightarrow dict

ClassificationEngine (mailsweep.classifier)

The heuristic evaluation engine applying 13-language safety tiers.

• ClassificationEngine(rules=None, user_domains=(), property_keywords=(), extra_org_keywords=())
• ClassificationEngine.from_env_and_rules(env_vars=None, rules_path="rules.json") \rightarrow ClassificationEngine
• engine.is_critical_subject(subject) \rightarrow bool
• engine.audit_item_safety(email_item) \rightarrow (finding_type: AuditFindingType, ClassificationResult)
• engine.audit_safety(items) \rightarrow list[tuple[AuditFindingType, ClassificationResult]]
• engine.classify(email_item) \rightarrow ClassificationResult
• engine.create_plan(scan_data) \rightarrow CleanupPlan

SafetyAuditor (mailsweep.audit)

The classification safety verification module.

• SafetyAuditor(rules=None)
• auditor.audit_scan(scan_res) \rightarrow SafetyAuditReport
• auditor.audit_plan(plan) \rightarrow SafetyAuditReport
• auditor.audit_backup(archive_path) \rightarrow SafetyAuditReport

BackupArchive (mailsweep.archive)

The offline archive creator and verifier.

• BackupArchive(path, is_zip=True, manifest=None)
• BackupArchive.open(path) \rightarrow BackupArchive
• BackupArchive.calculate_sha256(path) \rightarrow str
• archive.verify_integrity() \rightarrow (bool, str)
• archive.iter_messages(query, sender, subject) \rightarrow Generator of ArchiveMessage

UnsubscribeHub (mailsweep.unsubscribe)

The brand aggregator and unsubscription link health verifier.

• UnsubscribeHub(health_checker=None)
• hub.aggregate(scan_results) \rightarrow list[BrandSubscription]
• hub.verify_links(brands) \rightarrow list[BrandSubscription]
• hub.render_dashboard(brands) \rightarrow str (HTML)

RulesRegistry (mailsweep.rules)

The governance and synchronization module for categorization rules.

• RulesRegistry(transport=None)
• registry.load_ruleset(path) \rightarrow RuleSet
• registry.validate_rules(ruleset) \rightarrow RulesValidationReport
• registry.sync_upstream(source_url) \rightarrow RuleSet
• registry.sync_sources(manifest) \rightarrow RuleSet