Additional functions

.gitignore

@@ -0,0 +1 @@
+.env

README.md

@@ -72,3 +72,18 @@ openproject_importer.exe
 - If needed, adjust the `type` value in the script (default is `1` for `Task`).
 
 For enhancements (such as better error handling, additional field mappings, or logging), feel free to contact or extend the script!
+
+## Additional updates
+- Allow the presence of a ".env" file in the current directory for values matching those parameters
+- Also allow the script to accept command line parameters under the same names. (ie --api_token)
+- If both sources are missing, only then prompt the user to input the parameter values.
+- If no project ID is given, display a list of existing projects for user to choose.
+- Look for header row in CSV file and match each column with Work Package field (ie "subject")
+- Allow for csv file to have its own ID and Parent columns, and will create child tasks based off which task has parent id defined.  (ID in csv will be for reference only)
+- Will scan for all members listed under project
+- Admin assigned to the project will have all inserted tasks assigned to admin
+- If the "assignee" or "accountable" column exist in CSV, it'll try to match the name with what's found in the member list
+
+## Additional requirements
+- Python packages:
+  - `dotenv`

env.sample

@@ -0,0 +1,4 @@
+OPENPROJECT_URL=
+API_TOKEN=
+PROJECT_ID=
+CSV_FILE=

import requests_basicauth.py

@@ -3,73 +3,220 @@
 import sys
 import os
 import base64
+import argparse
+from dotenv import load_dotenv
+import csv
 
-# Query user for configuration
-openproject_url = input('Enter your OpenProject URL (e.g., https://openproject.local): ').strip().rstrip('/')
-api_token = input('Enter your OpenProject API token: ').strip()
-project_id = input('Enter your OpenProject numeric project ID (e.g., 5): ').strip()
-csv_file = input('Enter the full path to the CSV file to import (e.g., D:/path/to/file.csv): ').strip()
+# Load .env file if it exists
+load_dotenv()
 
-# Remove any surrounding quotes from the input path
+# Set up argument parser for command-line arguments
+parser = argparse.ArgumentParser(description='Import work packages to OpenProject from CSV')
+parser.add_argument('--openproject_url', help='OpenProject URL (e.g., https://openproject.local)')
+parser.add_argument('--api_token', help='OpenProject API token')
+parser.add_argument('--project_id', help='OpenProject numeric project ID (e.g., 5)')
+parser.add_argument('--csv_file', help='Full path to the CSV file to import (e.g., D:/path/to/file.csv)')
+args = parser.parse_args()
+
+# Function to get parameter value: .env first, then command-line, then prompt
+def get_param(env_key, arg_value, prompt_message):
+    value = os.getenv(env_key) or arg_value
+    if not value:
+        value = input(prompt_message).strip()
+    return value.rstrip('/') if env_key == 'OPENPROJECT_URL' else value
+
+# Retrieve parameters
+openproject_url = get_param('OPENPROJECT_URL', args.openproject_url, 'Enter your OpenProject URL (e.g., https://openproject.local): ')
+api_token = get_param('API_TOKEN', args.api_token, 'Enter your OpenProject API token: ')
+project_id = get_param('PROJECT_ID', args.project_id, 'Enter your OpenProject numeric project ID or leave blank to search: ')
+csv_file = get_param('CSV_FILE', args.csv_file, 'Enter the full path to the CSV file to import (e.g., D:/path/to/file.csv): ')
+
+# Remove any surrounding quotes from the csv_file path
 csv_file = csv_file.strip('"').strip("'")
 
 # Check if file exists
 if not os.path.isfile(csv_file):
-    print(f" The file '{csv_file}' does not exist. Please check the path and try again.")
+    print(f"The file '{csv_file}' does not exist. Please check the path and try again.")
     sys.exit(1)
 
-# Load CSV
-df = pd.read_csv(csv_file)
-
 # Set Basic Auth header using 'apikey' as username and API token as password
 auth_header = base64.b64encode(f'apikey:{api_token}'.encode('utf-8')).decode('utf-8')
 headers = {
     'Authorization': f'Basic {auth_header}',
     'Content-Type': 'application/json'
 }
 
-# Test API connectivity
+# If project_id is empty, fetch projects and prompt user to select one
+if not project_id:
+    try:
+        projects_response = requests.get(f"{openproject_url}/api/v3/projects", headers=headers)
+        if projects_response.status_code != 200:
+            print(f"Failed to fetch projects — Status {projects_response.status_code}: {projects_response.text}")
+            sys.exit(1)
+        projects_data = projects_response.json()
+        projects = projects_data['_embedded']['elements']
+        if not projects:
+            print("No projects found in OpenProject. Please create a project or specify a project ID.")
+            sys.exit(1)
+        print("\nAvailable Projects:")
+        for project in projects:
+            print(f"ID: {project['id']}, Name: {project['name']}")
+        project_id = input("\nEnter the ID of the project to import data to: ").strip()
+        if not project_id or not project_id.isdigit():
+            print("Invalid project ID. Please enter a numeric project ID.")
+            sys.exit(1)
+    except Exception as e:
+        print(f"Failed to fetch projects: {e}")
+        sys.exit(1)
+
+# Test API connectivity with the selected project_id
 try:
     test_response = requests.get(f"{openproject_url}/api/v3/projects/{project_id}", headers=headers)
     if test_response.status_code != 200:
-        print(f" API token test failed — Status {test_response.status_code}: {test_response.text}")
+        print(f"API token test failed — Status {test_response.status_code}: {test_response.text}")
         sys.exit(1)
     else:
-        print(f" API token verified. Proceeding with import.")
+        print(f"API token verified. Proceeding with import to project ID {project_id}.")
 except Exception as e:
-    print(f" Failed to connect to API: {e}")
+    print(f"Failed to connect to API: {e}")
     sys.exit(1)
 
-# Loop over each row and create work package
-for idx, row in df.iterrows():
+# Fetch available assignees
+try:
+    assignees_response = requests.get(f"{openproject_url}/api/v3/projects/{project_id}/available_assignees", headers=headers)
+    if assignees_response.status_code != 200:
+        print(f"Failed to fetch assignees — Status {assignees_response.status_code}: {assignees_response.text}")
+        sys.exit(1)
+    assignees_data = assignees_response.json()
+    assignees = []
+    for user in assignees_data['_embedded']['elements']:
+        # Fetch user details using full URL
+        user_href = user['_links']['self']['href']
+        if not user_href.startswith('http'):
+            user_href = f"{openproject_url}{user_href}"
+        user_response = requests.get(user_href, headers=headers)
+        if user_response.status_code == 200:
+            user_data = user_response.json()
+            roles = []
+            if user_data.get('admin') == True:
+                roles.append('admin') 
+            assignees.append({
+                'id': user_data['id'],
+                'firstName': user_data.get('firstName', ''),
+                'lastName': user_data.get('lastName', ''),
+                'fullName': f"{user_data.get('firstName', '')} {user_data.get('lastName', '')}".strip(),
+                'roles': roles
+            })
+    # Find the first admin (case-insensitive partial match for "admin" in role name)
+    default_assignee = next(
+        (user for user in assignees if any('admin' in role.lower() for role in user['roles'])),
+        assignees[0] if assignees else None
+    )
+except Exception as e:
+    print(f"Failed to fetch assignees: {e}")
+    sys.exit(1)
+
+# Load CSV with proper quote handling and normalize headers to lowercase
+df = pd.read_csv(csv_file, quoting=csv.QUOTE_ALL, quotechar='"', escapechar='\\')
+# Map parentID and childID to parent and child, then normalize to lowercase
+header_mapping = {col.lower(): 'parent' if col.lower() == 'parentid' else 'child' if col.lower() == 'childid' else col.lower() for col in df.columns}
+df.columns = [header_mapping[col.lower()] for col in df.columns]
+
+# Validate CSV headers
+required_headers = ['subject']  # Minimum required field
+if not all(header in df.columns for header in required_headers):
+    print(f"CSV file must contain the following headers: {required_headers}")
+    sys.exit(1)
+
+# Map CSV IDs to OpenProject IDs
+id_mapping = {}
+
+# Separate parent and child tasks
+parent_tasks = df[df['parent'].isna() | (df['parent'] == '')]
+child_tasks = df[~(df['parent'].isna() | (df['parent'] == ''))]
+
+# Function to create work package
+def create_work_package(row, parent_id=None):
     payload = {
         "_links": {
-            "project": {
-                "href": f"/api/v3/projects/{project_id}"
-            },
-            "type": {
-                "href": f"/api/v3/types/1"  # usually '1' is Task, adjust if needed
-            }
-        },
-        "subject": row['Subject'],
-        "description": {
-            "format": "markdown",
-            "raw": row['Description']
-        },
-        "status": {
-            "name": row['Status']
+            "project": {"href": f"/api/v3/projects/{project_id}"},
+            "type": {"href": f"/api/v3/types/1"}  # Default to Task, adjust if needed
         },
-        "priority": {
-            "name": row['Priority']
-        }
+        "subject": str(row['subject']).strip('"')  # Strip any residual quotes
     }
 
+    # Map CSV headers to API fields
+    for column in df.columns:
+        if column == 'id' or column == 'child':
+            continue  # Skip ID and child fields
+        elif column == 'description':
+            payload['description'] = {"format": "markdown", "raw": str(row[column]).strip('"')}
+        elif column == 'status':
+            payload['status'] = {"name": str(row[column]).strip('"')}
+        elif column == 'priority':
+            payload['priority'] = {"name": str(row[column]).strip('"')}
+        elif column == 'assignee' and pd.notna(row[column]):
+            # Match assignee by first name or full name
+            assignee_name = str(row[column]).strip('"').strip()
+            matched_assignee = next(
+                (user for user in assignees if assignee_name.lower() in (user['firstName'].lower(), user['fullName'].lower())),
+                default_assignee
+            )
+            if matched_assignee:
+                payload['_links']['assignee'] = {"href": f"/api/v3/users/{matched_assignee['id']}"}
+        elif column == 'responsible' and pd.notna(row[column]):
+            # Match responsible by first name or full name
+            responsible_name = str(row[column]).strip('"').strip()
+            matched_responsible = next(
+                (user for user in assignees if responsible_name.lower() in (user['firstName'].lower(), user['fullName'].lower())),
+                None
+            )
+            if matched_responsible:
+                payload['_links']['responsible'] = {"href": f"/api/v3/users/{matched_responsible['id']}"}
+        elif column == 'parent' and parent_id:
+            payload['_links']['parent'] = {"href": f"/api/v3/work_packages/{parent_id}"}
+        else:
+            # Include other fields as custom fields or direct attributes if applicable
+            value = str(row[column]).strip('"') if pd.notna(row[column]) else None
+            if value:
+                payload[column] = value
+
+    # Set default assignee if not specified
+    if 'assignee' not in df.columns and default_assignee:
+        payload['_links']['assignee'] = {"href": f"/api/v3/users/{default_assignee['id']}"}
+
     response = requests.post(f"{openproject_url}/api/v3/work_packages", headers=headers, json=payload)
+    return response
+
+# Insert parent tasks first
+for idx, row in parent_tasks.iterrows():
+    response = create_work_package(row)
+    if response.status_code == 201:
+        openproject_id = response.json()['id']
+        # Only use 'id' column if it exists
+        csv_id = row.get('id') if 'id' in df.columns and pd.notna(row.get('id')) else None
+        if csv_id:
+            id_mapping[csv_id] = openproject_id
+        print(f"Created parent work package: {row['subject']} (OpenProject ID: {openproject_id})")
+    else:
+        print(f"Failed to create {row['subject']} — Status {response.status_code}: {response.text}")
 
+# Insert child tasks
+for idx, row in child_tasks.iterrows():
+    csv_parent_id = row['parent'] if 'parent' in df.columns and pd.notna(row['parent']) else None
+    parent_id = id_mapping.get(csv_parent_id) if csv_parent_id else None
+    if csv_parent_id and not parent_id:
+        print(f"Warning: Parent ID {csv_parent_id} for {row['subject']} not found in id_mapping. Skipping parent assignment.")
+    response = create_work_package(row, parent_id)
     if response.status_code == 201:
-        print(f" Created work package: {row['Subject']}")
+        openproject_id = response.json()['id']
+        # Only use 'id' column if it exists
+        csv_id = row.get('id') if 'id' in df.columns and pd.notna(row.get('id')) else None
+        if csv_id:
+            id_mapping[csv_id] = openproject_id
+        print(f"Created child work package: {row['subject']} (OpenProject ID: {openproject_id})")
     else:
-        print(f" Failed to create {row['Subject']} — Status {response.status_code}: {response.text}")
+        print(f"Failed to create {row['subject']} — Status {response.status_code}: {response.text}")
 
 # Built by @SOCKS for OpenProject Ingestion 
 # https://github.com/773-process-312/OpenProject_CSV_Import.git