DEV Community

ByteLedger
ByteLedger

Posted on • Edited on

python mail format parser

#cli #python #tooling 

Get-ChildItem -Path $env:USERPROFILE -Recurse -Filter kubectl.exe -ErrorAction SilentlyContinue | Select-Object -First 10 FullName

& "C:\Users\Administrator\AppData\Roaming\Lens\binaries\kubectl\1.29.3\kubectl.exe" version --client

$kubectl = "C:\Users\Administrator\AppData\Roaming\Lens\binaries\kubectl\1.29.3\kubectl.exe"
$kubeconfig = Get-ChildItem "$env:APPDATA\Lens\kubeconfigs" -Recurse -Include *.config -ErrorAction SilentlyContinue | Select-Object -First 1 -ExpandProperty FullName
if ($kubeconfig) { $env:KUBECONFIG = $kubeconfig }
& $kubectl version --client
& $kubectl get ns
Enter fullscreen mode Exit fullscreen mode

"$env:APPDATA\Lens\kubeconfigs" -Recurse -Include *.config -ErrorAction SilentlyContinue | Select-Object -First 1 -ExpandProperty FullName; if ($lkc) { $env:KUBECONFIG = $lkc } }; if ($kubectl) { & $kubectl version --client; & $kubectl get ns } else { Write-Error "kubectl.exe not found (PATH or Lens cache)." }

$kubectl = (Get-Command kubectl -ErrorAction SilentlyContinue).Source; if (-not $kubectl) { $kubectl = Get-ChildItem "$env:APPDATA\Lens" -Recurse -Filter kubectl.exe -ErrorAction SilentlyContinue | Select-Object -First 1 -ExpandProperty FullName }; if (-not $env:KUBECONFIG) { $lkc = Get-ChildItem "$env:APPDATA\Lens\kubeconfigs" -Recurse -Include *.config -ErrorAction SilentlyContinue | Select-Object -First 1 -ExpandProperty FullName; if ($lkc) { $env:KUBECONFIG = $lkc } }; if ($kubectl) { & $kubectl version --client; & $kubectl get ns } else { Write-Error "kubectl.exe not found (PATH or Lens cache)." }


https://knowledge.broadcom.com/external/article/297346/how-to-limit-the-per-message-max-size-in.html?


emlgen/
  __init__.py
  cli.py              # argparse CLI
  io_utils.py         # read/write EML, CRLF-safe
  ids.py              # new GUID/Message-ID/boundary/timestamp
  transform.py        # token map, replacements, text injection

emlgen/io_utils.py
Enter fullscreen mode Exit fullscreen mode

from email import policy
from email.parser import BytesParser
from email.generator import BytesGenerator
from io import BytesIO
import os

def read_eml(path: str):
with open(path, "rb") as f:
return BytesParser(policy=policy.default).parse(f)

def write_eml(msg, path: str):
os.makedirs(os.path.dirname(path), exist_ok=True)
bio = BytesIO()
BytesGenerator(bio, policy=policy.SMTP).flatten(msg) # CRLF
with open(path, "wb") as f:
f.write(bio.getvalue())

emlgen/ids.py
Enter fullscreen mode Exit fullscreen mode

import uuid, datetime
from email.utils import make_msgid

def new_guid(): return str(uuid.uuid4())

def new_message_id(domain="example.test"):
return make_msgid(domain=domain) # includes <...>

def boundary(tag="MESSAGE_ID"): # e.g. MESSAGE_IDabc123
return f"{tag}{uuid.uuid4().hex[:10]}"

def ts_compact():
dt = datetime.datetime.utcnow()
return dt.strftime("%Y%m%d%H%M%S") + f"{int(dt.microsecond/1000):03d}"

emlgen/transform.py
Enter fullscreen mode Exit fullscreen mode

import re
from email import encoders

def get_boundary(msg):
b = msg.get_boundary()
if b: return b
m = re.search(r'boundary="?(.*?)"(;|$)', msg.get('Content-Type',''), re.I)
return m.group(1) if m else None

def set_boundary(mp, val):
if mp.is_multipart():
mp.set_boundary(val)

def walk_leaf_parts(msg):
for p in msg.walk():
if not p.is_multipart():
return_part = p
yield return_part

def collect_tokens(msg):
toks = {}
mid = (msg.get("Message-ID") or "").strip()
toks["Message-ID"] = mid
toks["Message-ID-unbr"] = mid[1:-1] if mid.startswith("<") and mid.endswith(">") else ""
toks["ConversationID"] = (msg.get("X-Header-ConversationID") or "").strip()
toks["Boundary"] = get_boundary(msg) or ""
toks["Filenames"] = [p.get_filename() for p in walk_leaf_parts(msg) if p.get_filename()]
return toks

def rename_attachment_headers(part, new_filename):
if part.get("Content-Disposition"):
part.set_param("filename", new_filename, header="Content-Disposition")
part.set_param("name", new_filename, header="Content-Type")

def replace_text_payload(part, replace_map, keep_b64=True):
orig_cte = (part.get("Content-Transfer-Encoding") or "").lower()
charset = part.get_content_charset() or "utf-8"
raw = part.get_payload(decode=True) or b""
try: text = raw.decode(charset, errors="replace")
except: text = raw.decode("utf-8", errors="replace")

for old, new in replace_map.items():
    if old: text = text.replace(old, new)

part.set_payload(text, charset=charset)
if part["Content-Transfer-Encoding"]:
    del part["Content-Transfer-Encoding"]
if keep_b64 and orig_cte == "base64":
    encoders.encode_base64(part)
Enter fullscreen mode Exit fullscreen mode

def inject_text(msg, text, where="html-end", once_marker="<!--emlgen-injected-->"):
"""
Adds text into text/html or text/plain part.
where: 'html-top' | 'html-end' | 'text-top' | 'text-end'
Adds a marker to avoid duplicate injections on re-runs.
"""
for p in walk_leaf_parts(msg):
ctype = (p.get_content_type() or "").lower()
if ctype == "text/html":
raw = p.get_payload(decode=True) or b""
charset = p.get_content_charset() or "utf-8"
try: html = raw.decode(charset, errors="replace")
except: html = raw.decode("utf-8", errors="replace")
if once_marker in html: continue
if where == "html-top":
html = once_marker + text + html
elif where == "html-end":
# insert before

Top comments (0)