Created
March 11, 2026 12:42
-
-
Save szepeviktor/0698f561c92601b4487c44b545beadce to your computer and use it in GitHub Desktop.
Attach Billingo.hu PDF invoices, to be used in an email pipe
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| import sys | |
| import re | |
| import copy | |
| import syslog | |
| import gzip | |
| import zlib | |
| import subprocess | |
| import urllib.request | |
| from email import message_from_bytes | |
| from email.header import decode_header, make_header | |
| from email.message import Message | |
| from email.mime.multipart import MIMEMultipart | |
| from email.mime.application import MIMEApplication | |
| from email.utils import parseaddr | |
| TAG = "billingo-attacher" | |
| UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0" | |
| def log(msg): | |
| syslog.openlog(TAG) | |
| syslog.syslog(msg) | |
| def die(msg, code=1): | |
| log("ERROR: " + msg) | |
| sys.exit(code) | |
| def dec_hdr(v): | |
| if not v: | |
| return "" | |
| try: | |
| return str(make_header(decode_header(v))) | |
| except Exception: | |
| return v | |
| def first_text_plain(msg): | |
| for p in msg.walk(): | |
| if p.is_multipart(): | |
| continue | |
| if p.get_content_type() != "text/plain": | |
| continue | |
| raw = p.get_payload(decode=True) | |
| if raw is None: | |
| raw = p.get_payload() | |
| if isinstance(raw, bytes): | |
| return raw.decode("utf-8", "replace") | |
| return raw or "" | |
| cs = p.get_content_charset() or "utf-8" | |
| try: | |
| return raw.decode(cs, "replace") | |
| except Exception: | |
| return raw.decode("utf-8", "replace") | |
| return "" | |
| def as_body_part(msg): | |
| part = Message() | |
| for k, v in msg.items(): | |
| lk = k.lower() | |
| if lk.startswith("content-") or lk == "mime-version": | |
| part[k] = v | |
| part.set_payload(msg.get_payload()) | |
| return part | |
| def fetch(url): | |
| req = urllib.request.Request(url) | |
| req.add_header("User-Agent", UA) | |
| req.add_header("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8") | |
| req.add_header("Accept-Language", "en-US,en;q=0.5") | |
| req.add_header("Accept-Encoding", "gzip, deflate") | |
| req.add_header("DNT", "1") | |
| req.add_header("Connection", "keep-alive") | |
| r = urllib.request.urlopen(req, timeout=30) | |
| data = r.read() | |
| enc = (r.info().get("Content-Encoding") or "").lower() | |
| if enc == "gzip": | |
| data = gzip.decompress(data) | |
| elif enc == "deflate": | |
| try: | |
| data = zlib.decompress(data) | |
| except Exception: | |
| data = zlib.decompress(data, -zlib.MAX_WBITS) | |
| return data | |
| if len(sys.argv) < 2: | |
| die("usage: billingo-attacher.py recipient@example.com", 64) | |
| rcpt = sys.argv[1] | |
| raw = sys.stdin.buffer.read() | |
| if not raw: | |
| die("empty stdin") | |
| msg = message_from_bytes(raw) | |
| text = first_text_plain(msg) | |
| if not text: | |
| die("no text/plain part") | |
| m = re.search(r'https?://\S+', text) | |
| if not m: | |
| die("no url found") | |
| url = m.group(0).rstrip(").,;]>\"'") + "/download" | |
| try: | |
| pdf = fetch(url) | |
| except Exception as e: | |
| die("download failed: " + str(e)) | |
| if not pdf.startswith(b"%PDF"): | |
| die("downloaded file is not a pdf") | |
| inv = re.search(r"Számla sorszáma:\s*([A-Z0-9-]+)", text) | |
| fn = (inv.group(1) if inv else "szamla") + ".pdf" | |
| out = MIMEMultipart("mixed") | |
| out["From"] = parseaddr(dec_hdr(msg.get("To")))[1] or "root@localhost" | |
| out["To"] = rcpt | |
| out["Subject"] = dec_hdr(msg.get("Subject")) or "PDF" | |
| if msg.get("Reply-To"): | |
| out["Reply-To"] = dec_hdr(msg.get("Reply-To")) | |
| if msg.is_multipart(): | |
| for p in msg.get_payload(): | |
| out.attach(copy.deepcopy(p)) | |
| else: | |
| out.attach(as_body_part(msg)) | |
| pdf_part = MIMEApplication(pdf, _subtype="pdf") | |
| pdf_part.add_header("Content-Disposition", "attachment", filename=fn) | |
| out.attach(pdf_part) | |
| p = subprocess.Popen( | |
| ["/usr/sbin/sendmail", "-t", "-i"], | |
| stdin=subprocess.PIPE, | |
| stdout=subprocess.DEVNULL, | |
| stderr=subprocess.PIPE | |
| ) | |
| stderr = p.communicate(out.as_bytes())[1] | |
| if p.returncode != 0: | |
| die("sendmail failed: " + stderr.decode("utf-8", "ignore").strip()) | |
| log("OK: sent {0} to {1} from url={2}".format(fn, rcpt, url)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment