send_feedback.py

#!/usr/bin/env python3
# This script sends feedback to students' forks by opening issues with the Gitlab API

# The target forks and associated metadata are read from stdin. The script expects JSON
# data as outputted by collect_forks.py

# The issues are based on a template, with the content read from a separate CSV table
# with columns "id", "grade", "comments"

import sys
import json
from string import Formatter
from typing import Dict, Optional
import requests
from pathlib import Path
import click
import pandas as pd
from teach_auto.common_requests import parse_repo_url
from teach_auto.utils import ingest_token


FEEDBACK_TEMPLATE = """
# {project}, group {group}

## Members:
{members}

Please find below your grade and specific comments for this project:

## Grade:
{grade}

## Comments:
{comments}
"""


def fill_template(content: Dict[str, str], template: str) -> str:
    """Replace fields surrounded by {} in template by the value
    of corresponding keys in content.

    Examples
    --------
    >>> content = {'name': 'Bob'}
    >>> fill_template(content, 'I am {name}')
    'I am Bob'
    >>> fill_template(content | {'age': 13}, 'I am {age}')
    'I am 13'
    >>> fill_template(content, 'I am {age}') # doctest: +IGNORE_EXCEPTION_DETAIL
    Traceback (most recent call last):
    ...
    ValueError: Fields missing from content: ['age']
    """
    try:
        # Replace content in issue template
        return template.format(**content)
    except KeyError:
        templ_flds = [fld for _, fld, _, _ in Formatter().parse(template) if fld]
        missing_flds = [fld for fld in templ_flds if fld not in content.keys()]
        raise ValueError(f"Fields missing from content: {missing_flds}") from None


def gather_fill_data(fork: Dict, feedback_csv: Optional[Path] = None) -> Dict:
    """Gather metadata and feedback for a single fork into a dictionary
    to use for filling the template. The resulting dictionary contains
    fields: project, group, members, url, commit and visibility from the fork
    metadata, as well as all columns present in the feedback csv file.
    """
    # Get basic fields in fork metadata
    _, group, repo = parse_repo_url(fork["url"])
    members = "\n".join([f"* {member['name']}" for member in fork["members"]])
    fill_data = {
        "project": repo.removesuffix(".git"),
        "group": group,
        "members": members,
    }
    # add commit, url and visibility fields from fork data
    fill_data |= {key: fork.get(key) for key in ["url", "commit", "visibility"]}
    # If a CSV file with feedback is provided, read the fields
    # of the current fork (row) to the fill data.
    if feedback_csv:
        fill_data |= (
            pd.read_csv(feedback_csv)
            .query(f'id == {fork["id"]}')
            .iloc[0]
            .drop("id")
            .to_dict()
        )
    return fill_data


@click.command()
@click.argument("forks", type=click.File(mode="r"), default=sys.stdin)
@click.option(
    "--token-path",
    type=click.Path(exists=True),
    help="Path to a file containing the Gitlab API token. If not provided, you will be prompted for the token.",
)
@click.option(
    "--template",
    type=click.Path(exists=True),
    help="Markdown template for the issue. Fields surrounded by {} in the template will be replaced "
    "by content from corresponding columns in feedback.csv. By default, a template with fields {project}, "
    "{group}, {members}, {grade} and {comments} is used.",
    default=None,
)
@click.option(
    "--feedback-csv",
    type=click.Path(exists=True),
    help="CSV file with column 'id', and additional columns matching fields in the template"
    "The id column should match a gitlab project ID from the JSON input.",
    default=None,
)
@click.option(
    "--dry-run", is_flag=True, help="Print feedback to stdout instead of opening issues"
)
def send_feedback(forks, feedback_csv, token_path, template, dry_run):
    """
    Send feedback to repositories in input JSON by opening issues.
    """

    # Build header with authentication token
    token = ingest_token(token_path)
    header = {"PRIVATE-TOKEN": token}

    forks = json.load(forks)

    # Load issue template if provided
    if template:
        feedback_template = open(template).read()
    else:
        feedback_template = FEEDBACK_TEMPLATE

    for fork in forks:
        base = parse_repo_url(fork["url"])[0]
        fill_data = gather_fill_data(fork, feedback_csv)
        # Replace fields in the template by their value for the current fork
        description = fill_template(fill_data, feedback_template)
        # Create issue, or print to stdout if dry run
        if dry_run:
            print(fork)
            print(description)
        else:
            requests.post(
                f"{base}/api/v4/projects/{fork['id']}/issues?confidential=true&title=Feedback",
                headers=header,
                data={"description": description},
            )


if __name__ == "__main__":
    send_feedback()