<commit_before>from PyPDF2 import PdfFileReader
from datetime import datetime
from datetime import timedelta
from email import encoders
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
from email.utils import formatdate
from pdfminer.converter import TextConverter
from pdfminer.layout import LAParams
from pdfminer.pdfinterp import PDFResourceManager, process_pdf
from reportlab.pdfgen import canvas
from subprocess import Popen, PIPE

import io
import logging
import inspect
import os
import requests
import shutil
import sys
import tempfile
import unittest


class Email2PDFTestCase(unittest.TestCase):
    isOnline = None
    examineDir = None

    NONEXIST_IMG = 'http://www.andrewferrier.com/nonexist.jpg'
    EXIST_IMG = 'https://raw.githubusercontent.com/andrewferrier/email2pdf/master/tests/basi2c16.png'

    def setUp(self):
        self.workingDir = tempfile.mkdtemp(dir='/tmp')
        self.command = os.path.normpath(os.path.join(os.getcwd(), 'email2pdf'))
        self.checkOnline()
        self.checkExamineDir()

    @classmethod
    def checkExamineDir(cls):
        if Email2PDFTestCase.examineDir is None:
            Email2PDFTestCase.examineDir = '/tmp'
            Email2PDFTestCase.examineDir = tempfile.mkdtemp(dir=Email2PDFTestCase.examineDir)
            print("Output examination directory: " + Email2PDFTestCase.examineDir)

    @classmethod
    def checkOnline(cls):
        if Email2PDFTestCase.isOnline is None:
            print("Checking if online... ", end="")
            sys.stdout.flush()
            ONLINE_URL = "https://raw.githubusercontent.com/andrewferrier/email2pdf/master"
            try:
                requests.get(ONLINE_URL, timeout=1)
                Email2PDFTestCase.isOnline = True
                print("Yes.")
            except requests.exceptions.RequestException:
                Email2PDFTestCase.isOnline = False
                print("No.")

        return Email2PDFTestCase.isOnline

    def getOriginalScriptPath(self):
        module_path = inspect.getfile(inspect.currentframe())
        module_path = os.path.join(os.path.dirname(os.path.dirname(module_path)), 'email2pdf')

        return module_path

    def getTimeStamp(self, myTime):
        return myTime.strftime("%Y-%m-%dT%H-%M-%S")

    def existsByTime(self, path=None):
        if self.getTimedFilename(path):
            return True
        else:
            return False

    def getTimedFilename(self, path=None):
        if path is None:
            path = self.workingDir

        for single_time in self.timerange(self.timeInvoked, self.timeCompleted):
            filename = os.path.join(path, self.getTimeStamp(single_time) + ".pdf")
            if os.path.exists(filename):
                return filename

        return None

    def addHeaders(self, frm="<EMAIL>", to="<EMAIL>", subject="Subject of the email"):
        if(subject):
            self.msg['Subject'] = subject

        if(frm):
            self.msg['From'] = frm

        if(to):
            self.msg['To'] = to

        self.msg['Date'] = formatdate()

    def invokeAsSubprocess(self, inputFile=False, outputDirectory=None, outputFile=None, extraParams=[]):
        bytesMessage = bytes(self.msg.as_string(), 'UTF-8')

        options = [self.command]

        if inputFile:
            inputFile_handle = tempfile.NamedTemporaryFile()
            options.extend(['-i', inputFile_handle.name])
            inputFile_handle.write(bytesMessage)
            inputFile_handle.flush()
            myStdin = None
            myInput = None
        else:
            myStdin = PIPE
            myInput = bytesMessage

        if outputDirectory:
            options.extend(['-d', outputDirectory])

        if outputFile:
            options.extend(['-o', outputFile])

        options.extend(extraParams)

        self.timeInvoked = datetime.now()
        if outputDirectory is None:
            myCwd = self.workingDir
        else:
            myCwd = None

        p = Popen(options, stdin=myStdin, stdout=PIPE, stderr=PIPE, cwd=myCwd)

        output, error = p.communicate(myInput)
        p.wait()
        self.timeCompleted = datetime.now()

        output = str(output, "utf-8")
        error = str(error, "utf-8")

        self.assertEqual("", output)

        if inputFile:
            inputFile_handle.close()

        return (p.returncode, output, error)

    def invokeDirectly(self, outputDirectory=None, outputFile=None, extraParams=[]):
        import importlib.machinery
        module_path = self.getOriginalScriptPath()
        loader = importlib.machinery.SourceFileLoader('email2pdf', module_path)
        email2pdf = loader.load_module()

        with tempfile.NamedTemporaryFile() as inputFile_handle:
            options = [module_path, '-i', inputFile_handle.name]
            inputFile_handle.write(bytes(self.msg.as_string(), 'UTF-8'))
            inputFile_handle.flush()

            if outputDirectory:
                options.extend(['-d', outputDirectory])
            else:
                options.extend(['-d', self.workingDir])

            if outputFile:
                options.extend(['-o', outputFile])

            options.extend(extraParams)
<commit_msg>Simplify getting emails as bytes.<commit_after>from PyPDF2 import PdfFileReader
from datetime import datetime
from datetime import timedelta
from email import encoders
from email.mime.base import MIMEBase
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
from email.utils import formatdate
from pdfminer.converter import TextConverter
from pdfminer.layout import LAParams
from pdfminer.pdfinterp import PDFResourceManager, process_pdf
from reportlab.pdfgen import canvas
from subprocess import Popen, PIPE

import io
import logging
import inspect
import os
import requests
import shutil
import sys
import tempfile
import unittest


class Email2PDFTestCase(unittest.TestCase):
    isOnline = None
    examineDir = None

    NONEXIST_IMG = 'http://www.andrewferrier.com/nonexist.jpg'
    EXIST_IMG = 'https://raw.githubusercontent.com/andrewferrier/email2pdf/master/tests/basi2c16.png'

    def setUp(self):
        self.workingDir = tempfile.mkdtemp(dir='/tmp')
        self.command = os.path.normpath(os.path.join(os.getcwd(), 'email2pdf'))
        self.checkOnline()
        self.checkExamineDir()

    @classmethod
    def checkExamineDir(cls):
        if Email2PDFTestCase.examineDir is None:
            Email2PDFTestCase.examineDir = '/tmp'
            Email2PDFTestCase.examineDir = tempfile.mkdtemp(dir=Email2PDFTestCase.examineDir)
            print("Output examination directory: " + Email2PDFTestCase.examineDir)

    @classmethod
    def checkOnline(cls):
        if Email2PDFTestCase.isOnline is None:
            print("Checking if online... ", end="")
            sys.stdout.flush()
            ONLINE_URL = "https://raw.githubusercontent.com/andrewferrier/email2pdf/master"
            try:
                requests.get(ONLINE_URL, timeout=1)
                Email2PDFTestCase.isOnline = True
                print("Yes.")
            except requests.exceptions.RequestException:
                Email2PDFTestCase.isOnline = False
                print("No.")

        return Email2PDFTestCase.isOnline

    def getOriginalScriptPath(self):
        module_path = inspect.getfile(inspect.currentframe())
        module_path = os.path.join(os.path.dirname(os.path.dirname(module_path)), 'email2pdf')

        return module_path

    def getTimeStamp(self, myTime):
        return myTime.strftime("%Y-%m-%dT%H-%M-%S")

    def existsByTime(self, path=None):
        if self.getTimedFilename(path):
            return True
        else:
            return False

    def getTimedFilename(self, path=None):
        if path is None:
            path = self.workingDir

        for single_time in self.timerange(self.timeInvoked, self.timeCompleted):
            filename = os.path.join(path, self.getTimeStamp(single_time) + ".pdf")
            if os.path.exists(filename):
                return filename

        return None

    def addHeaders(self, frm="<EMAIL>", to="<EMAIL>", subject="Subject of the email"):
        if(subject):
            self.msg['Subject'] = subject

        if(frm):
            self.msg['From'] = frm

        if(to):
            self.msg['To'] = to

        self.msg['Date'] = formatdate()

    def invokeAsSubprocess(self, inputFile=False, outputDirectory=None, outputFile=None, extraParams=[]):
        bytesMessage = self.msg.as_bytes()

        options = [self.command]

        if inputFile:
            inputFile_handle = tempfile.NamedTemporaryFile()
            options.extend(['-i', inputFile_handle.name])
            inputFile_handle.write(bytesMessage)
            inputFile_handle.flush()
            myStdin = None
            myInput = None
        else:
            myStdin = PIPE
            myInput = bytesMessage

        if outputDirectory:
            options.extend(['-d', outputDirectory])

        if outputFile:
            options.extend(['-o', outputFile])

        options.extend(extraParams)

        self.timeInvoked = datetime.now()
        if outputDirectory is None:
            myCwd = self.workingDir
        else:
            myCwd = None

        p = Popen(options, stdin=myStdin, stdout=PIPE, stderr=PIPE, cwd=myCwd)

        output, error = p.communicate(myInput)
        p.wait()
        self.timeCompleted = datetime.now()

        output = str(output, "utf-8")
        error = str(error, "utf-8")

        self.assertEqual("", output)

        if inputFile:
            inputFile_handle.close()

        return (p.returncode, output, error)

    def invokeDirectly(self, outputDirectory=None, outputFile=None, extraParams=[]):
        import importlib.machinery
        module_path = self.getOriginalScriptPath()
        loader = importlib.machinery.SourceFileLoader('email2pdf', module_path)
        email2pdf = loader.load_module()

        bytesMessage = self.msg.as_bytes()

        with tempfile.NamedTemporaryFile() as inputFile_handle:
            options = [module_path, '-i', inputFile_handle.name]
            inputFile_handle.write(bytesMessage)
            inputFile_handle.flush()

            if outputDirectory:
                options.extend(['-d', outputDirectory])
            else:
                options.extend(['-d', self.workingDir])

            if outputFile:
                options.extend(['-o', outputFile])

            options.extend(extraParams)

{'fraction_non_alphanumeric': 0.10031288518062008, 'fraction_numerical': 0.004551057172655732, 'mean_word_length': 2.440313111545988, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2610575', 'n_tokens_mistral': 3070, 'n_tokens_neox': 2954, 'n_words': 628}