242 lines
8.7 KiB
Python
242 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Fetch inbox emails filtered by authorized senders (conf.json tickets.authorized_emails).
|
|
Does not use UNSEEN; does not mark as read. Writes each matching message to projects/<id>/data/issues/
|
|
as JSON (<date>_<from_sanitized>_<uid>.pending). One file per message.
|
|
Usage: run from project root with GITEA_ISSUES_DIR and PROJECT_ROOT set (e.g. via tickets-fetch-inbox.sh).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import email
|
|
import imaplib
|
|
import json
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from email.header import decode_header
|
|
from email.utils import parsedate_to_datetime
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from mail_common import load_imap_config, imap_ssl_context
|
|
from project_config import authorized_emails, data_issues_dir, load_project_config
|
|
|
|
|
|
def decode_header_value(header: str | None) -> str:
|
|
if not header:
|
|
return ""
|
|
parts = decode_header(header)
|
|
result = []
|
|
for part, charset in parts:
|
|
if isinstance(part, bytes):
|
|
result.append(part.decode(charset or "utf-8", errors="replace"))
|
|
else:
|
|
result.append(part)
|
|
return "".join(result)
|
|
|
|
|
|
def parse_from_address(from_header: str) -> str:
|
|
"""Extract email address from From header (e.g. 'Name <user@host>' -> user@host)."""
|
|
if not from_header:
|
|
return ""
|
|
match = re.search(r"<([^>]+)>", from_header)
|
|
if match:
|
|
return match.group(1).strip().lower()
|
|
return from_header.strip().lower()
|
|
|
|
|
|
def get_text_body(msg: email.message.Message) -> str:
|
|
if msg.is_multipart():
|
|
for part in msg.walk():
|
|
if part.get_content_type() == "text/plain":
|
|
payload = part.get_payload(decode=True)
|
|
if payload:
|
|
return payload.decode(
|
|
part.get_content_charset() or "utf-8", errors="replace"
|
|
)
|
|
return ""
|
|
payload = msg.get_payload(decode=True)
|
|
if not payload:
|
|
return ""
|
|
return payload.decode(
|
|
msg.get_content_charset() or "utf-8", errors="replace"
|
|
)
|
|
|
|
|
|
def is_sent_to_alias(msg: email.message.Message, filter_to: str) -> bool:
|
|
if not filter_to:
|
|
return True
|
|
headers_to_check = ("To", "Delivered-To", "X-Original-To", "Cc", "Envelope-To")
|
|
for name in headers_to_check:
|
|
value = msg.get(name)
|
|
if value:
|
|
decoded = decode_header_value(value).lower()
|
|
if filter_to in decoded:
|
|
return True
|
|
return False
|
|
|
|
|
|
def sanitize_from_for_filename(email_addr: str) -> str:
|
|
"""Filesystem-safe string from email (e.g. user@example.com -> user_example.com)."""
|
|
return re.sub(r"[^a-zA-Z0-9._-]", "_", email_addr.replace("@", "_"))
|
|
|
|
|
|
def sanitize_attachment_filename(name: str) -> str:
|
|
"""Safe filename for attachment (no path, no dangerous chars)."""
|
|
if not name or not name.strip():
|
|
return "attachment"
|
|
base = Path(name).name
|
|
return re.sub(r"[^a-zA-Z0-9._-]", "_", base)[:200] or "attachment"
|
|
|
|
|
|
def get_attachments(msg: email.message.Message) -> list[tuple[str, bytes, str]]:
|
|
"""Return list of (filename, payload_bytes, content_type) for each attachment."""
|
|
result: list[tuple[str, bytes, str]] = []
|
|
for part in msg.walk():
|
|
content_type = (part.get_content_type() or "").lower()
|
|
if content_type.startswith("multipart/"):
|
|
continue
|
|
filename = part.get_filename()
|
|
if not filename:
|
|
# Optional: treat inline images etc. with Content-Disposition attachment
|
|
disp = part.get("Content-Disposition") or ""
|
|
if "attachment" in disp.lower():
|
|
ext = ""
|
|
if "image" in content_type:
|
|
ext = ".bin" if "octet-stream" in content_type else ".img"
|
|
filename = f"attachment{ext}"
|
|
else:
|
|
continue
|
|
filename = decode_header_value(filename).strip()
|
|
if not filename:
|
|
continue
|
|
payload = part.get_payload(decode=True)
|
|
if payload is None:
|
|
continue
|
|
result.append((filename, payload, content_type))
|
|
return result
|
|
|
|
|
|
def parse_references(refs: str | None) -> list[str]:
|
|
if not refs:
|
|
return []
|
|
return [x.strip() for x in re.split(r"\s+", refs) if x.strip()]
|
|
|
|
|
|
def main() -> int:
|
|
conf = load_project_config()
|
|
if not conf:
|
|
print("[tickets-fetch-inbox] No project config (projects/<id>/conf.json).", file=sys.stderr)
|
|
return 1
|
|
auth = authorized_emails()
|
|
filter_to = (auth.get("to") or "").strip().lower()
|
|
from_list = auth.get("from")
|
|
if isinstance(from_list, list):
|
|
allowed_from = {a.strip().lower() for a in from_list if a}
|
|
else:
|
|
allowed_from = set()
|
|
if not filter_to or not allowed_from:
|
|
print(
|
|
"[tickets-fetch-inbox] tickets.authorized_emails.to and .from required in conf.json.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
cfg = load_imap_config()
|
|
if not cfg["user"] or not cfg["password"]:
|
|
print("[tickets-fetch-inbox] IMAP_USER and IMAP_PASSWORD required.", file=sys.stderr)
|
|
return 1
|
|
|
|
spool = data_issues_dir()
|
|
spool.mkdir(parents=True, exist_ok=True)
|
|
|
|
mail = imaplib.IMAP4(cfg["host"], int(cfg["port"]))
|
|
if cfg["use_starttls"]:
|
|
mail.starttls(imap_ssl_context(cfg.get("ssl_verify", True)))
|
|
mail.login(cfg["user"], cfg["password"])
|
|
mail.select("INBOX")
|
|
# Do not use UNSEEN; fetch all (or SINCE to limit). Filter by authorized senders only.
|
|
_, nums = mail.search(None, "ALL")
|
|
ids = nums[0].split()
|
|
written = 0
|
|
for uid in ids:
|
|
uid_s = uid.decode("ascii")
|
|
_, data = mail.fetch(uid, "(RFC822)")
|
|
if not data or not data[0]:
|
|
continue
|
|
msg = email.message_from_bytes(data[0][1])
|
|
if not is_sent_to_alias(msg, filter_to):
|
|
continue
|
|
from_raw = decode_header_value(msg.get("From"))
|
|
from_addr = parse_from_address(from_raw)
|
|
if from_addr not in allowed_from:
|
|
continue
|
|
mid = (msg.get("Message-ID") or "").strip()
|
|
to_raw = decode_header_value(msg.get("To"))
|
|
to_addrs = [a.strip() for a in re.split(r"[,;]", to_raw) if a.strip()]
|
|
subj = decode_header_value(msg.get("Subject"))
|
|
date_h = decode_header_value(msg.get("Date"))
|
|
refs = parse_references(msg.get("References"))
|
|
in_reply_to = (msg.get("In-Reply-To") or "").strip() or None
|
|
body = get_text_body(msg)
|
|
try:
|
|
if date_h:
|
|
dt = parsedate_to_datetime(date_h)
|
|
date_str = dt.strftime("%Y-%m-%dT%H%M%S")
|
|
else:
|
|
date_str = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%S")
|
|
except Exception:
|
|
date_str = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%S")
|
|
from_safe = sanitize_from_for_filename(from_addr)
|
|
base = f"{date_str}_{from_safe}_{uid_s}"
|
|
path = spool / f"{base}.pending"
|
|
if path.exists():
|
|
continue
|
|
created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
attachments_meta: list[dict[str, str | int]] = []
|
|
attachment_parts = get_attachments(msg)
|
|
if attachment_parts:
|
|
att_dir = spool / f"{base}.d"
|
|
att_dir.mkdir(parents=True, exist_ok=True)
|
|
for idx, (orig_name, payload_bytes, content_type) in enumerate(attachment_parts):
|
|
safe_name = sanitize_attachment_filename(orig_name)
|
|
stored_name = f"{idx}_{safe_name}"
|
|
stored_path = att_dir / stored_name
|
|
stored_path.write_bytes(payload_bytes)
|
|
rel_path = f"{base}.d/{stored_name}"
|
|
attachments_meta.append({
|
|
"filename": orig_name,
|
|
"path": rel_path,
|
|
"content_type": content_type,
|
|
"size": len(payload_bytes),
|
|
})
|
|
payload = {
|
|
"version": 1,
|
|
"type": "incoming",
|
|
"message_id": mid or "",
|
|
"from": from_addr,
|
|
"to": to_addrs,
|
|
"subject": subj,
|
|
"date": date_h or "",
|
|
"body": body or "",
|
|
"references": refs,
|
|
"in_reply_to": in_reply_to,
|
|
"uid": uid_s,
|
|
"created_at": created_at,
|
|
"issue_number": None,
|
|
"status": "pending",
|
|
"attachments": attachments_meta,
|
|
}
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
written += 1
|
|
print(f"[tickets-fetch-inbox] Wrote {path.name}")
|
|
|
|
mail.logout()
|
|
print(f"[tickets-fetch-inbox] Done. Wrote {written} new message(s) to {spool}.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|