Github issues / MIlestones creation

github_helper.py

import pandas as pd
import requests
import json
import pdb
import time
from datetime import datetime


def fetch_issues(owner, repo, token, per_page=100):
    url = f"https://api.github.com/repos/{owner}/{repo}/issues"
    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {token}",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    issues = {}
    page = 1

    while True:
        params = {"page": page, "per_page": per_page}
        response = requests.get(url, headers=headers, params=params)
        if response.status_code != 200:
            raise Exception(f"Failed to fetch issues: {response.status_code} {response.text}")
        if not response.json():
            break
        
        for res in response.json():
            if res["state"] =="open":
              issues[res["title"]] = res["node_id"]
        page += 1
    
    return issues


def create_github_issue(owner, repo, token, title, body=None, assignees=None, milestone=None, labels=None):
    """
    Creates a GitHub issue using the GitHub API.

    Args:
        owner (str): The owner of the repository (e.g., "octocat").
        repo (str): The name of the repository (e.g., "Spoon-Knife").
        token (str): Your GitHub personal access token.
        title (str): The title of the issue.
        body (str): The body/description of the issue.
        assignees (list, optional): A list of GitHub usernames to assign to the issue. Defaults to None.
        milestone (int, optional): The milestone number to associate with the issue. Defaults to None.
        labels (list, optional): A list of labels to add to the issue. Defaults to None.

    Returns:
        dict: The JSON response from the GitHub API or None if the request failed.
    """

    url = f"https://api.github.com/repos/{owner}/{repo}/issues"

    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {token}",
        "X-GitHub-Api-Version": "2022-11-28",
        "Content-Type": "application/json" # Important to specify JSON content
    }

    data = {
        "title": title,
        "body": body,
    }

    if milestone:
      data["milestone"] = milestone
    if labels:
      data["labels"] = labels
    
    try:
      response = requests.post(url, headers=headers, data=json.dumps(data))
      response.raise_for_status()  # Raise an exception for bad status codes (4xx or 5xx)
      return response.json()
    except requests.exceptions.RequestException as e:
      print(f"Error creating issue: {e}")
      if response.status_code :
        print(f"Status Code : {response.status_code}")
        try:
          error_body= response.json()
          print (f"Github Error : {error_body}")
        except:
          print(f"Response : {response.text}")

      return None

def fetch_milestones(owner, repo, token, per_page=100):
    url = f"https://api.github.com/repos/{owner}/{repo}/milestones"
    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {token}",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    milestones = {}
    page = 1

    while True:
        params = {"page": page, "per_page": per_page}
        response = requests.get(url, headers=headers, params=params)
        if response.status_code != 200:
            raise Exception(f"Failed to fetch milestones: {response.status_code} {response.text}")
        if not response.json():
            break
        
        for res in response.json():
            milestones[res["title"]] = res["number"]
        page += 1

    return milestones


def create_milestone_with_curl(owner, repo, github_token, title, state, due_on, description =None):
    """
    Creates a GitHub milestone using the equivalent of the provided curl command.

    Args:
        github_token (str): Your GitHub Personal Access Token.
        owner (str): The repository owner's username.
        repo (str): The name of the repository.
        title (str): The title of the milestone.
        state (str): The state of the milestone ("open" or "closed").
        description (str): The description of the milestone.
        due_on (str): The due date of the milestone in ISO 8601 format (e.g., "2012-10-09T23:39:01Z").

    Returns:
        dict: The JSON response from the GitHub API, or None if an error occurs.
    """

    url = f"https://api.github.com/repos/{owner}/{repo}/milestones"
    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {github_token}",
        "X-GitHub-Api-Version": "2022-11-28",
        "Content-Type": "application/json"
    }
    data = {
        "title": title,
        "state": state,
        "due_on": due_on
    }

    try:
        response = requests.post(url, headers=headers, data=json.dumps(data))
        response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
        return response.json()
    except requests.exceptions.HTTPError as e:
        print(f"Error creating milestone: {e}")
        print(f"Response content: {response.content}")
        return None
    except Exception as e:
         print(f"An unexpected error occurred: {e}")
         return None
     


"""
# ------------------- CSV file -------------------
2024 Timeline,Milestones,Start Date,End Date,Team
Conference Venue Contract Signed,Venue,12/1/2024,2025/04/26,Board
Conference Catering Contract Signed,Catering,12/1/2024,2025/06/21,Board
"""


class TimelinePlanner:
    def __init__(self):
        self.data_path = "./data.csv"
        self.owner = "artisai"
        self.repo = "issueRepo"
        self.token = "ghp_token"
        self.HEADERS = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }

        self.GRAPHQL_URL = "https://api.github.com/graphql"
        self.df = pd.read_csv(self.data_path)
        self.milestones = self.create_milestones()
        self.issues = self.create_issues()

    def create_milestones(self):
        milestone_state = "open"
        existing_milestones = fetch_milestones(self.owner, self.repo, self.token, per_page=100)
        result_df = self.df.groupby("Milestones").agg({"End Date": "max"}).reset_index()
        result_df.columns = ["Milestones", "Max End Date"]
        for index, row in result_df.iterrows():
            if row["Milestones"] not in existing_milestones:      
                deadline = datetime.strptime(row["Max End Date"], "%Y/%m/%d").strftime("%Y-%m-%dT%H:%M:%SZ")
                milestone_data = create_milestone_with_curl(self.owner, self.repo, self.token, row["Milestones"],milestone_state, due_on=deadline )
                existing_milestones[milestone_data["title"]] = milestone_data["number"]
        print(f"Total milestones: {len(existing_milestones)}")
        return existing_milestones
        
    def create_issues(self):
        existing_issues = fetch_issues(self.owner, self.repo, self.token, per_page=100)
        for index, row in self.df.iterrows():
            if row["2024 Timeline"] not in existing_issues:
                body =  f"Details about the feature.\n\n**Start Date**: {row['Start Date']}\n**End Date**: {row['End Date']}"
                issue_data = create_github_issue(owner= self.owner, repo = self.repo, token=self.token, title=row["2024 Timeline"],body=body, milestone=self.milestones[row["Milestones"]], labels= [row["Team"]])
                time.sleep(3) # hit secondary rate limit
                existing_issues[issue_data["title"]] = issue_data["node_id"]
                # import pdb; pdb.set_trace()
        
        print(f"Total issues: {len(existing_issues)}")
        return existing_issues
    
    def get_project_node_id(self, project_name):
        query = """
        query($organization: String!) {
            organization(login: $organization) {
                projectsV2(first: 100) {
                    nodes {
                        id
                        title
                        fields(first: 100) {
                            nodes {
                                ... on ProjectV2Field {
                                    id
                                    name
                                    dataType
                                }
                            }
                        }
                        items(first: 100) {
                            nodes {
                                ... on ProjectV2Item {
                                    id
                                }
                            }
                        }
                    }
                }
            }
        }
        """
        variables = {
            "organization": self.owner
        }
        response = requests.post(self.GRAPHQL_URL, json={"query": query, "variables": variables}, headers=self.HEADERS)
        response.raise_for_status()
        
        # might be organization in the place of user
        response = response.json()
        # print(response)
        projects = response["data"]["organization"]["projectsV2"]["nodes"]
        field_map  = {}
        item_id = ""
        # import pdb 
        # pdb.set_trace()
        for project in projects:
            if project["title"] == project_name:
                fields = project["fields"]["nodes"]
                item_id = project["items"]["nodes"][0]["id"]
                for field in fields:
                    if field:
                        field_map[field["name"]] = field["id"]
                return project["id"],item_id,  field_map
        raise ValueError("Project not found")

    def add_issue_to_project(self, project_id, issue_node_id):
        mutation = """
        mutation($projectId: ID!, $contentId: ID!) {
            addProjectV2ItemById(input: {projectId: $projectId, contentId: $contentId}) {
                item {
                    id
                }
            }
        }
        """
        variables = {
            "projectId": project_id,
            "contentId": issue_node_id
        }
        response = requests.post(self.GRAPHQL_URL, json={"query": mutation, "variables": variables}, headers=self.HEADERS)
        response.raise_for_status()
        return response.json()["data"]["addProjectV2ItemById"]["item"]["id"]

    def set_field_value(self, project_id, item_id, field_id, value):
        # Step 2: Set the start_date or end_date field for the item in the project
        set_field_mutation = """
        mutation($projectId: ID!, $itemId: ID!, $fieldId: ID!, $value: ProjectV2FieldValue!) {
            updateProjectV2ItemFieldValue(input: {projectId: $projectId, itemId: $itemId, fieldId: $fieldId, value: $value}) {
                projectV2Item {
                    id
                }
            }
        }
        """
        variables = {
            "projectId": project_id,
            "itemId": item_id,
            "fieldId": field_id,
            "value": {"date": value}
        }
        response = requests.post(self.GRAPHQL_URL, json={"query": set_field_mutation, "variables": variables}, headers=self.HEADERS)
        response.raise_for_status()
        return response.json()

    def attach_issue_project(self, project_board_name):
        project_node_id, project_item_id, project_field_map = self.get_project_node_id(project_board_name)

        for issue, issue_node_id in self.issues.items():
            self.add_issue_to_project(project_node_id, issue_node_id)
            
            # Set the dates
            issue_row  = self.df[self.df['2024 Timeline'] == issue]
            
            try:
                start_date = datetime.strptime(issue_row["Start Date"].values[0], "%m/%d/%Y").strftime("%Y-%m-%d")
                end_date = datetime.strptime(issue_row["End Date"].values[0], "%Y/%m/%d").strftime("%Y-%m-%d")
                
                start = self.set_field_value(project_node_id, item_id=project_item_id,field_id=project_field_map["Start date"], value=start_date)
                end = self.set_field_value(project_node_id, project_item_id,field_id=project_field_map["End date"], value=end_date)
                print(f"---------------------\n{start}{end}")
            except Exception as e:
                print(f"error : {e} {issue_row}")
    
    def process(self, project_board_name):
        self.attach_issue_project(project_board_name)
        
if __name__ == "__main__":
    tp = TimelinePlanner()
    tp.process("testProject") # testProject is the board name

Note: issues are milestones are being created.
They are also linked to the project board.
Milestone enddate is being set based on the csv file.
:TODO: set Start / End date of the issue in the Project.