(
inbox_doc: InboxDoc,
inbox_emails: typing.Sequence[InboxEmail],
workdir_path: pathlib.Path,
)
| 102 | |
| 103 | |
| 104 | def compute_missing_emails( |
| 105 | inbox_doc: InboxDoc, |
| 106 | inbox_emails: typing.Sequence[InboxEmail], |
| 107 | workdir_path: pathlib.Path, |
| 108 | ) -> typing.Generator[tuple[InboxEmail, pathlib.Path], None, None]: |
| 109 | template_env = SandboxedEnvironment() |
| 110 | workdir = workdir_path.resolve().absolute() |
| 111 | for inbox_email in inbox_emails: |
| 112 | action = process_inbox_email( |
| 113 | template_env=template_env, |
| 114 | inbox_email=inbox_email, |
| 115 | inbox_configs=inbox_doc.inbox, |
| 116 | ) |
| 117 | if action is None: |
| 118 | continue |
| 119 | if isinstance(action, IgnoreInboxAction): |
| 120 | continue |
| 121 | elif isinstance(action, ArchiveInboxAction): |
| 122 | output_file = (workdir / action.output_file).resolve().absolute() |
| 123 | if not output_file.is_relative_to(workdir): |
| 124 | logger.error( |
| 125 | "The email archive output path [green]%s[/] for email [green]%s[/] is not a sub-path of workdir [green]%s[/]", |
| 126 | output_file, |
| 127 | inbox_email.id, |
| 128 | workdir, |
| 129 | extra={"markup": True, "highlighter": None}, |
| 130 | ) |
| 131 | sys.exit(-1) |
| 132 | rel_output_file = output_file.relative_to(workdir) |
| 133 | if output_file.exists(): |
| 134 | logger.info( |
| 135 | "The email [green]%s[/] archive output path [green]%s[/] already exists, skip", |
| 136 | inbox_email.id, |
| 137 | rel_output_file, |
| 138 | extra={"markup": True, "highlighter": None}, |
| 139 | ) |
| 140 | continue |
| 141 | yield inbox_email, rel_output_file |
| 142 | else: |
| 143 | raise ValueError(f"Unexpected action type {type(action)}") |
| 144 | |
| 145 | |
| 146 | def load_inbox_doc(config: str) -> InboxDoc: |
no outgoing calls