preparemoodle uses moodle utilities

preparemoodle.py

 #!/usr/bin/env python
 
-import csv
 import os
 import time
 import shutil  # copyfile, make_archive
-import argparse
+import argparse  # argument parsing
 import sys
+
+import utils.moodle as moodle
 import utils.matnum as matnum_utils
 
 
-def find_unmatched_pdfs(infolder, matnums, nowarn):
-    """Finds matnumbers not present in CSV but in PDF folder
+def sanity_check(matnums_csv, matnums_folder):
+    """Check two cases for sanity:
+    - Are there PDF files with no corresponding CSV entries?
+    - Are there CSV entries with no provided PDF file?
 
     Args:
-        infolder (str): path to input folder
-        matnums (list): list of matriculation numbers
-        nowarn (int): flag
+        matnums_csv (list): Matnums of all CSV entries
+        matnums_folder (list): Matnums of all provided PDF files
     """
 
-    print("\nSearching for matnumbers not present in CSV but in PDF folder:")
-
-    # Loop over all PDFs:
-    notfoundmatnums = []
-    for root, dirs, files in os.walk(infolder):
-        for pdffile in files:
-            if pdffile.endswith(".pdf"):
-                # Get matriculation number from file
-                matnum = matnum_utils.get_matnum(pdffile)
+    # PDF files with no entry in CSV:
+    notfoundcsv = list(set(matnums_folder).difference(matnums_csv))
 
-                # Search matriculation number in CSV
-                if matnum not in matnums:
-                    notfoundmatnums.append(matnum)
-                    if not nowarn:
-                        print("Warning: {} not in CSV".format(matnum))
+    # Entries in CSV without PDF file
+    notfoundpdf = list(set(matnums_csv).difference(matnums_folder))
 
     # Report back
-    if len(notfoundmatnums) > 0:
-        print('''Could not find following {} matnumbers in CSV:
-            {}'''.format(len(notfoundmatnums), ", ".join(notfoundmatnums)))
+    if len(notfoundcsv) > 0:
+        print('''Warning: Following {} matnums have PDFs but no entry in CSV:
+            {}'''.format(len(notfoundcsv), ", ".join(notfoundcsv)))
+
+    if len(notfoundpdf) > 0:
+        print('''Warning: Following {} matnums have CSV entries but no PDF:
+            {}'''.format(len(notfoundpdf), ", ".join(notfoundpdf)))
 
     print("Done.\n")
 
+    return notfoundcsv, notfoundpdf
+
 
 def main(args):
     """Main routine
@@ -49,18 +47,30 @@ def main(args):
     # Parse input arguments
     parser = argparse.ArgumentParser(description='''
     prepares batch upload to Moodle via assignment module.
-    PDFs in folder 'in' are moved to folder 'tmp' with a certain folder structure and finally zipped to 'out'.
+    PDFs in folder 'in' are moved to folder 'tmp' with a certain folder
+    structure and finally zipped to 'out'.
     Attention: zip-archive 'out' will be overwritten in the following!
 
     ''')
     parser.add_argument("-i", "--infolder", default="./pdfs_encrypted",
-                        help="Input folder with PDFs. Default: ./pdfs_encrypted")
+                        help="Input folder with PDFs." +
+                             "Default: ./pdfs_encrypted")
     parser.add_argument("-c", "--csv", default="./Bewertungen.csv",
-                        help="Moodle grading CSV file, needed to construct the folder names. Default: ./Bewertungen.csv")
+                        help="Moodle grading sheet, needed to construct " +
+                             "submission folder names. " +
+                             "Default: ./Bewertungen.csv")
+    parser.add_argument("--csvdelim", default=",",
+                        help="CSV delimiter. Default: ','")
+    parser.add_argument("--csvquote", default='"', help="Quote character." +
+                        """Default: '"'""")
+    parser.add_argument("--csvencoding", default="utf-8",
+                        help="CSV encoding scheme. Typical encodings:" +
+                             "'utf-8', 'utf-8-sig', or 'cp1252' (Windows). " +
+                             "Default: 'utf-8'")
     parser.add_argument("-o", "--outzip", default="./moodle_feedbacks.zip",
-                        help="Output zip archive. Default: ./moodle_feedbacks.zip")
+                        help="Zip archive. Default: ./moodle_feedbacks.zip")
     parser.add_argument("-d", "--dry", action='store_true',
-                        help="Flag for dry run, displays only the folder structure inside the archive moodle_feedbacks.zip")
+                        help="Flag for dry run, displays folder structure")
     parser.add_argument("-t", "--tmp", default="./tmp",
                         help="tmp folder. Default: ./tmp")
     parser.add_argument("--nowarn", action='store_true',
@@ -68,23 +78,25 @@ def main(args):
 
     args = parser.parse_args(args)
     infolder = args.infolder
-    csvfilename = args.csv
+    sheet_csv = args.csv
     outzip = args.outzip
     tmpfolder = os.path.join(args.tmp, "to_be_zipped_for_moodle")
     dry = args.dry
     nowarn = args.nowarn
+    csvdelim = args.csvdelim
+    csvquote = args.csvquote
+    csvenc = args.csvencoding
 
+    # Print number of lines
     starttime = time.time()
-
-    # Print status with total number of lines
-    numlines = 0
-    with open(csvfilename, newline='') as csvfile:
-        numlines = sum(1 for line in csvfile)
+    numstudents = moodle.get_student_number(sheet_csv=sheet_csv,
+                                            csv_enc=csvenc)
 
     print('''Preparing for moodle upload
 Processing {} lines
-  '''.format(numlines))
+  '''.format(numstudents))
 
+    # Clean up and create temporary folder
     dryout = ""
     if dry:
         print("Dry run\n")
@@ -97,93 +109,97 @@ Processing {} lines
         if not os.path.isdir(tmpfolder):
             os.mkdir(tmpfolder)
 
-    # Open CSV file
-    with open(csvfilename, newline='') as csvfile:
-
-        numfoundpdfs = 0
-        matnums = []
-        line_cnt = 0
-        print("Start iterating...", sep='', end='', flush=True)
-
-        # Loop over all lines in CSV file
-        reader = csv.reader(csvfile, delimiter=',', quotechar='"')
-        next(reader)  # skip header CSV line
-        for row in reader:
-            # Parse required fields from CSV line
-            # Moodle has its own internal ID per participant alongside
-            # matriculation number
-            moodleid = row[0]
-            moodleid = moodleid.replace("Teilnehmer/in", "")  # German
-            moodleid = moodleid.replace("Participant ", "")  # English
-            name = row[1]  # Lastname, Firstname
-            matnum = row[2]  # matriculation number (6-digit)
-            matnums.append(matnum)  # save matriculation number for later
-
-            # Copy PDF files
-            # Find all PDFs starting with matriculation number, e.g.
-            # '123456_Lastname_sheet.pdf' and '123456_Lastname_exam.pdf'
-            # If pdf files for current student exists, create a directory and
-            # copy the pdf files to it. The resulting directories can be
-            # uploaded to Moodle
-            longpdffiles = matnum_utils.find_file(matnum + "*.pdf", infolder)
-            if len(longpdffiles) > 0:  # Found some file(s)
-                numfoundpdfs += 1
-
-                # Prepare folder
-                # For upload, Moodle accepts submission files per participant
-                folder = "{}_{}_assignsubmission_file_".format(name, moodleid)
-                longfolder = os.path.join(tmpfolder, folder)
-
-                # Create folder
+    # Parse input folder
+    # Only PDF files are considered with first digits
+    # containing matriculation number
+    matnums_folder = []
+    allfiles = os.listdir(infolder)
+    allfiles.sort()
+    allpdfs = []
+    for f in allfiles:
+        if f.lower().endswith('.pdf') and matnum_utils.starts_with_matnum(f):
+            allpdfs.append(f)
+            matnums_folder.append(matnum_utils.get_matnum(f))
+
+    # Parse grading infos from CSV file
+    infos = moodle.extract_info(sheet_csv=sheet_csv, csv_delim=csvdelim,
+                                csv_quote=csvquote, csv_enc=csvenc)
+
+    # Loop over grading infos
+    numfoundpdfs = 0
+    matnums_csv = []
+    moodleids = []
+    for cnt, info in enumerate(infos):
+        # Copy PDF files
+        # Find all PDFs starting with matriculation number, e.g.
+        # '123456_Lastname_sheet.pdf' and '123456_Lastname_exam.pdf'
+        # If pdf files for current student exists, create a directory and
+        # copy the pdf files to it. The resulting directories can be
+        # uploaded to Moodle
+        matnum = info['matnum']
+        matnums_csv.append(matnum)
+        moodleid = info['moodleid']
+        moodleids.append(moodleid)
+
+        pdfs_student = [_ for _ in allpdfs
+                        if matnum == matnum_utils.get_matnum(_)]
+        if len(pdfs_student) > 0:  # Found at least one pdf
+            numfoundpdfs += len(pdfs_student)
+
+            # Prepare submission folder
+            folder = moodle.submission_folder_name(info)
+            longfolder = os.path.join(tmpfolder, folder)
+
+            # Create folder
+            if not dry:
+                os.mkdir(longfolder)
+
+            # Copy all files to folder
+            for pdffile in pdfs_student:
+                longpdffile = os.path.join(infolder, pdffile)
+                longpdffiledest = os.path.join(longfolder, pdffile)
                 if not dry:
-                    os.mkdir(longfolder)
-
-                # Copy all files to folder
-                for longpdffile in longpdffiles:
-                    pdffile = os.path.basename(longpdffile)
-
-                    if not dry:
-                        shutil.copyfile(longpdffile,
-                                        os.path.join(longfolder, pdffile))
-                    else:
-                        dryout += "\n{}".format(os.path.join(folder, pdffile))
-            else:
-                if not nowarn:
-                    print("Warning: PDF corresponding to matnumber {} (moodleid={}, name={}) not available.".format(
-                        matnum, moodleid, name
-                    ))
-
-            # Print progress
-            if not (line_cnt % max(1, round(numlines/10))):
-                print(".", sep=' ', end='', flush=True)
-            line_cnt += 1
+                    shutil.copyfile(longpdffile, longpdffiledest)
+                else:
+                    dryout += "\n{}".format(os.path.join(folder, pdffile))
+
+        elif not nowarn:  # No PDF found
+            print("Warning: PDF for {matnum} (id={id}, name={name}) not found."
+                  .format(matnum=matnum, id=moodleid, name=info['fullname']))
+
+        # Print for-loop progress
+        if not (cnt % max(1, round(numstudents/10))):
+            print(".", sep=' ', end='', flush=True)
 
     # Print results
-    print("Found {} PDFs (CSV had {} entries)".format(numfoundpdfs, numlines))
+    print("Found {numpdf} PDFs (CSV had {numcsv} entries)"
+          .format(numpdf=numfoundpdfs, numcsv=numstudents))
     print("done.")
 
     # Sanity check:
     # Check for PDFs not reflected in CSV (student not registered in Moodle)
-    find_unmatched_pdfs(infolder, matnums, nowarn)
+    sanity_check(matnums_csv, matnums_folder)
 
-    # Zipping
+    # Zip
     if not dry:
         # Zip
         print("Zipping")
         shutil.make_archive(os.path.splitext(outzip)[0], 'zip', tmpfolder)
-        print('The Zip archive is available at: '+outzip)
+        print('Zip archive is stored at {}'.format(outzip))
 
         # Delete temporary folder
         shutil.rmtree(tmpfolder)
 
+    # Print dry run results
     else:
         print("\nDry run results:\n{}".format(dryout))
 
+    # Print status
     endtime = time.time()
-
     print("""Done.
 Time taken: {:.2f}""".format(endtime-starttime))
 
 
+# Main routine
 if __name__ == '__main__':
     main(sys.argv[1:])

tests/test_moodle.py

@@ -15,14 +15,14 @@ class MainTest(unittest.TestCase):
         from utils import moodle as moodle_utils
 
         csv_file = "./Bewertungen.csv"
-        gis = moodle_utils.extract_grading_info(sheet_csv=csv_file)
+        gis = moodle_utils.extract_info(sheet_csv=csv_file)
         self.assertEqual(gis[2]['lastname'], "Three")
 
-        num_students = moodle_utils.get_number_of_students(sheet_csv=csv_file)
+        num_students = moodle_utils.get_student_number(sheet_csv=csv_file)
         self.assertEqual(num_students, 3)
 
         csv_file = "./Grades.csv"
-        gis = moodle_utils.extract_grading_info(sheet_csv=csv_file)
+        gis = moodle_utils.extract_info(sheet_csv=csv_file)
 
         # Check that "d'Lastname" gets whitened to "dLastname"
         self.assertEqual(gis[3]['lastname'], "dLastname")

tests/test_prepare_moodle.py

+import unittest
+import time
+import os
+import tempfile
+import shutil
+
+
+class MainTest(unittest.TestCase):
+    def setUp(self):
+        self.tic = time.time()  # todo this is sooo ugly
+        self.test_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        self.toc = time.time()
+        t = self.toc - self.tic
+        print('Time: %.3f' % (t))
+
+        # Clean up
+        shutil.rmtree(self.test_dir)
+
+    def test_prepare_moodle(self):
+        import preparemoodle
+
+        expected_feedback_folder = 'Four, Student_107_assignsubmission_file_'
+        expected_feedback_file = '123456_Nachname.pdf'
+
+        # Prepare parameter
+        in_dir = './pdfs'
+        sheet_csv = "./Bewertungen.csv"
+        feedback_zip = 'feedbacks.zip'
+
+        tmp_dir = os.path.join(self.test_dir, 'tmp')
+        os.mkdir(tmp_dir)
+
+        out_zip = os.path.join(self.test_dir, feedback_zip)
+
+        # Call function
+        preparemoodle.main(["-i", in_dir, "--csv", sheet_csv,
+                            "-t", tmp_dir, "-o", out_zip])
+
+        # Unpack feedbacks
+        shutil.unpack_archive(out_zip, tmp_dir)
+        feedback_folders = os.listdir(tmp_dir)
+        self.assertTrue(feedback_folders[0], expected_feedback_folder)
+
+        feedbacks = os.listdir(os.path.join(tmp_dir, feedback_folders[0]))
+        self.assertTrue(feedbacks[0], expected_feedback_file)

utils/moodle.py

 import csv  # handles csv
 
 
-def get_number_of_students(sheet_csv, csv_encoding='utf-8'):
+def get_student_number(sheet_csv, csv_enc='utf-8'):
     """Count number of student entries in grading sheet
 
     Args:
         sheet_csv (str): filename of grading sheet CSV
+        csv_enc (str): CSV encoding
 
     Returns:
         int: number of entries
@@ -13,7 +14,7 @@ def get_number_of_students(sheet_csv, csv_encoding='utf-8'):
 
     # Open CSV file and count lines
     numlines = 0
-    with open(sheet_csv, newline='', encoding=csv_encoding) as csvfile:
+    with open(sheet_csv, newline='', encoding=csv_enc) as csvfile:
         numlines = sum(1 for _ in csvfile)
 
     numlines -= 1  # do not count header line
@@ -37,15 +38,16 @@ def submission_folder_name(grading_info):
     return foldername
 
 
-def extract_grading_info(sheet_csv,
-                         csv_delim=',', csv_quote='"', csv_encoding='utf-8'):
+def extract_info(sheet_csv, csv_delim=',', csv_quote='"', csv_enc='utf-8'):
     """Extract grading information from grading sheet
 
     Args:
         sheet_csv (str): filename of grading sheet CSV
         csv_delim (str, optional): CSV delimiter. Defaults to ','.
         csv_quote (str, optional): CSV quote char. Defaults to '"'.
-        csv_encoding (str, optional): CSV encoding. Defaults to 'utf-8'.
+        csv_enc (str, optional): CSV encoding. Defaults to 'utf-8'.
+                                 Typical values: "'utf-8', 'utf-8-sig',
+                                 or 'cp1252' (Windows). "
 
     Returns:
         list of dicts: grading information with following info per student:
@@ -57,7 +59,7 @@ def extract_grading_info(sheet_csv,
     fieldnames = ['longid', 'fullname', 'matnum', 'status', 'grade']
 
     # Check delimiter
-    with open(sheet_csv, newline='', encoding=csv_encoding) as csvfile:
+    with open(sheet_csv, newline='', encoding=csv_enc) as csvfile:
         try:
             csv.Sniffer().sniff(csvfile.read(1024),
                                 delimiters=csv_delim)
@@ -67,7 +69,7 @@ def extract_grading_info(sheet_csv,
             raise ValueError(error)
 
     # Open CSV file
-    with open(sheet_csv, newline='', encoding=csv_encoding) as csvfile:
+    with open(sheet_csv, newline='', encoding=csv_enc) as csvfile:
         # Convert CSV to list of dicts
         reader = csv.DictReader(csvfile, fieldnames=fieldnames,
                                 delimiter=csv_delim, quotechar=csv_quote)