Python2017/homeworks/main.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import argparse
import logging
import git
import os
import importlib
import collections
import csv

logger = logging.getLogger('Homeworks')
logger.setLevel(logging.DEBUG)
fm = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch = logging.StreamHandler()
ch.setFormatter(fm)
logger.addHandler(ch)


def arg_parse():
    parser = argparse.ArgumentParser(description='Framework for checking homeworks.')
    parser.add_argument('-i', '--ids', type=str,
                        help='path to a file with student ids.')
    return parser.parse_args()

def get_ids(path):
    ids = []
    with open(path) as _file:
        for line in _file:
            ids.append(line.strip().split(' ')[::2])
    return ids


def clone_repo(username, repo_name):
    repo = "https://git.wmi.amu.edu.pl/{}/{}.git".format(username, repo_name)
    dest = "./Python2017-{}".format(username)
    if os.path.exists(dest):
        logger.debug("{}: repo exests.".format(username))
    else:
        try:
            git.Repo.clone_from(repo, dest)
        except git.GitCommandError:
            logger.info("Repository '{}' does not exist".format(repo))
            return False
    return True

def check_tasks(username, repo_name, tasks):
    logger.debug("Marking user: {}".format(username))
    scores = collections.defaultdict(dict)
    for lab in tasks:
        path = os.path.realpath("./Python2017-{}/{}".format(username, lab))
        for task, fun in tasks[lab]:
            task_file = path + "/{}". format(task)
            logger.debug(task_file)
            package = 'Python2017-{}.{}.{}'.format(username, lab, task)
            lib = importlib.import_module(package)
            tested_fun = getattr(lib, fun)
            tests = getattr(lib, 'tests')
            try:
                test_result = tests(tested_fun)
            except Exception:
                test_result = "FAILED"

            if test_result == 'TESTS PASSED':
                scores[lab][task] = 1
            else:
                scores[lab][task] = 0
    return scores

def get_fieldnames(tasks):
    fieldnames = ['ID']
    for lab in tasks:
        for task, _ in tasks[lab]:
            fieldnames.append('{}:{}'.format(lab, task))
    return fieldnames

def parse_scores(fieldnames, uname, scores):
    out = {}
    for name in fieldnames:
        if name == 'ID':
            out[name] = uname
            continue
        lab, task = name.split(':')
        out[name] = scores[lab][task]
    return out


def write_scores(scores, filename, tasks):
    fieldnames = get_fieldnames(tasks)

    with open(filename, 'w') as results:
        writer = csv.DictWriter(results, fieldnames, restval=0)
        writer.writeheader()
        for student in scores:
            writer.writerow(parse_scores(fieldnames, student, scores[student]))


def main():
    """ main """
    options = arg_parse()
    ids = get_ids(options.ids)
    logger.info("Liczba studentów: {}".format(len(ids)))
    scores = {}

    tasks = {
        'labs02' : [
            ('task07', 'char_sum'),
            ('task08', 'sum_div35'),
            ('task09', 'leet_speak'),
            ('task10', 'pokemon_speak'),
            ('task11', 'common_chars')
        ]
    }

    for repo_id in ids:
        st_uname = repo_id[0]
        repo_name = repo_id[-1]
        if clone_repo(st_uname, repo_name):
            scores[st_uname] = check_tasks(st_uname, repo_name, tasks)
    write_scores(scores, 'results.csv', tasks)


if __name__ == "__main__":
    main()