Update sprint_scrum.py

Changed cycles to modules, functionality is working, looking for feedback on how Nick would like this to run
This commit is contained in:
2025-11-17 20:49:40 +00:00
parent 3e26fe1c17
commit 60b0ec826a

View File

@@ -1,255 +1,194 @@
#----------------------------------Imports----------------------------------- #----------------------------------Imports-----------------------------------
import json import json
import requests import requests
from datetime import timedelta, datetime, timezone from datetime import timedelta, datetime, timezone
from dotenv import load_dotenv from dotenv import load_dotenv
import os import os
#---------------------------Set Up-------------------------------------------- #---------------------------Set Up--------------------------------------------
PLANE_URL = "https://project-management.cui-secure.us/api/v1/workspaces/midwatch" PLANE_URL = "https://project-management.cui-secure.us/api/v1"
load_dotenv() load_dotenv()
TOKEN = os.getenv("TOKEN") TOKEN = os.getenv("TOKEN")
PROJECT_ID = "2a3b613e-a182-4278-8547-9bd5250bf67f" PROJECT_ID = "2a3b613e-a182-4278-8547-9bd5250bf67f"
WORKSPACE_ID = "midwatch" WORKSPACE_ID = "midwatch"
TESTING_PROJECT = "a0c36f9c-65a5-4f50-beae-b9badc8cf11d" TESTING_PROJECT = "a0c36f9c-65a5-4f50-beae-b9badc8cf11d"
SPRINT_LENGTH = 7 SPRINT_LENGTH = 7
HEADER = { HEADER = {
"X-API-Key": f"{TOKEN}", "X-API-Key": f"{TOKEN}",
"Content-Type": "application/json" "Content-Type": "application/json"
} }
#---------------------------Testing Grounds------------------------------------ #---------------------------Testing Grounds------------------------------------
# test = requests.get(f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles", headers=HEADER) # test = requests.get(f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/modules", headers=HEADER)
# if test.status_code == 200: # if test.status_code == 200:
# print("Connected to Plane API successfully.") # print("Connected to Plane API successfully.")
# print(test.json()) # print(test.json())
# else: # else:
# print(f"API connection failed. Status code: {test.status_code}") # print(f"API connection failed. Status code: {test.status_code}")
# print("Response:", test.text) # print("Response:", test.text)
# exit() # exit()
#-----------------------Functionality----------------------------------------------- #-----------------------Functionality-----------------------------------------------
def get_last_sprint(): def get_last_sprint():
url = f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles" url = f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/modules"
response = requests.get(url, headers=HEADER) response = requests.get(url, headers=HEADER)
data = response.json() data = response.json()
if response.status_code != 200: if response.status_code != 200:
print(f"Failed to fetch sprints: {response.status_code}") print(f"Failed to fetch sprints: {response.status_code}")
print("Response:", data) print("Response:", data)
return None return None
sprints = data if isinstance(data, list) else data.get("results", []) sprints = data if isinstance(data, list) else data.get("results", [])
if not sprints: if not sprints:
print("No sprints found.") print("No sprints found.")
return None return None
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
completed = [] completed = []
active = [] upcoming = []
upcoming = []
for s in sprints:
for s in sprints: raw = s["target_date"]
start = datetime.fromisoformat(s["start_date"].replace("Z", "+00:00"))
end = datetime.fromisoformat(s["end_date"].replace("Z", "+00:00")) end = datetime.strptime(raw, "%Y-%m-%d").replace(tzinfo=timezone.utc)
if end < now:
completed.append(s) if end < now:
elif start <= now <= end: completed.append(s)
active.append(s) else:
else: upcoming.append(s)
upcoming.append(s)
completed.sort(key=lambda s: s["target_date"], reverse=True)
completed.sort(key=lambda s: s["end_date"], reverse=True)
if completed:
if completed: last = completed[0]
last = completed[0] print(f"Last completed sprint: {last['name']} (ended {last['target_date']})")
print(f"Last completed sprint: {last['name']} (ended {last['end_date']})") return last
return last else:
elif active: print("No active or completed sprints found.")
current = active[0] return None
print(f"Current sprint in progress: {current['name']} (ends {current['end_date']})")
return current
else: def get_all_issues():
print("No active or completed sprints found.") response = requests.get(
return None f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/issues?expand=state",
headers=HEADER
)
def get_all_issues(): response.raise_for_status()
response = requests.get( data = response.json()
f"{PLANE_URL}/projects/{TESTING_PROJECT}/issues?expand=state",
headers=HEADER all_issues = data.get('results', data)
) uncompleted = []
response.raise_for_status()
data = response.json() for issue in all_issues:
state = issue.get('state', {})
all_issues = data.get('results', data) group = state.get('group', 'unknown')
uncompleted = []
if group.lower() != 'completed':
for issue in all_issues: uncompleted.append(issue)
state = issue.get('state', {})
group = state.get('group', 'unknown') print(f"Retrieved {len(uncompleted)} uncompleted issues from project {TESTING_PROJECT}")
return uncompleted
if group.lower() != 'completed':
uncompleted.append(issue) def get_unfinished_issues(sprint, all_issues):
sprint_id = sprint['id']
print(f"Retrieved {len(uncompleted)} uncompleted issues from project {TESTING_PROJECT}")
return uncompleted response = requests.get(
f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/modules/{sprint_id}/module-issues/",
headers=HEADER
def get_unfinished_issues(sprint, all_issues): )
sprint_id = sprint['id'] response.raise_for_status()
response = requests.get( sprint_data = response.json()
f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/{sprint_id}/cycle-issues/", sprint_issues = sprint_data.get('results', sprint_data)
headers=HEADER
) sprint_issue_ids = {i.get('id') for i in sprint_issues if i.get('id')}
response.raise_for_status()
print(f"Found {len(sprint_issue_ids)} issues linked to sprint {sprint_id}")
sprint_data = response.json()
sprint_issues = sprint_data.get('results', sprint_data) unfinished_ids = [
issue.get('id')
sprint_issue_ids = {i.get('id') for i in sprint_issues if i.get('id')} for issue in all_issues
if issue.get('id') in sprint_issue_ids
print(f"Found {len(sprint_issue_ids)} issues linked to sprint {sprint_id}") ]
unfinished_ids = [ print(f"Found {len(unfinished_ids)} unfinished issues in sprint {sprint_id}")
issue.get('id')
for issue in all_issues archive_last_sprint = requests.patch(
if issue.get('id') in sprint_issue_ids f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/modules/{sprint_id}/",
] headers=HEADER,
json={"status": "completed"}
print(f"Found {len(unfinished_ids)} unfinished issues in sprint {sprint_id}") )
return unfinished_ids print(f"Patch Last Sprint: {archive_last_sprint.status_code}")
return unfinished_ids
def create_cycle_minimal(name="Auto-Sprint"):
url = f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/"
payload = {"name": name} def create_new_sprint(last_sprint, unresolved_issue_ids):
last_end = datetime.fromisoformat(last_sprint["target_date"])
print("Creating new cycle:", payload) new_start = last_end + timedelta(days=1)
res = requests.post(url, headers=HEADER, json=payload) new_end = new_start + timedelta(days=SPRINT_LENGTH)
print("Create response:", res.text)
res.raise_for_status() starting_str = new_start.strftime("%Y-%m-%d")
ending_str = new_end.strftime("%Y-%m-%d")
cycle = res.json()
return cycle
print("Creating cycle with name only...")
def update_cycle_details(cycle_id, start_date, end_date, description): create_payload = {
url = f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/{cycle_id}" "name": f"Auto-Sprint {starting_str}-{ending_str}",
payload = { "description": f"This is an Automated Sprint for the dates {starting_str} to {ending_str}",
"start_date": start_date, "start_date": starting_str,
"end_date": end_date, "target_date": ending_str,
"description": description, "status":"in-progress"
"name": f"Auto-Sprint {start_date}" }
}
create_res = requests.post(
print(f"Updating cycle {cycle_id}:", payload) f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/modules/",
res = requests.patch(url, headers=HEADER, json=payload) headers=HEADER,
print("Patch response:", res.text) json=create_payload
res.raise_for_status() )
return res.json()
print("Create response:", create_res.text)
create_res.raise_for_status()
def attach_issues_to_cycle(cycle_id, issue_ids):
url = f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/{cycle_id}/cycle-issues/" cycle = create_res.json()
payload = {"issues": issue_ids} cycle_id = cycle["id"]
print("Cycle created:", cycle_id)
print(f"Attaching issues to cycle {cycle_id}:", payload)
res = requests.post(url, headers=HEADER, json=payload)
print("Attach response:", res.text) if unresolved_issue_ids:
res.raise_for_status() attach_payload = {"issues": unresolved_issue_ids}
return res.json() print("Attaching issues:", attach_payload)
attach_res = requests.post(
f"{PLANE_URL}/workspaces/midwatch/projects/{TESTING_PROJECT}/modules/{cycle_id}/module-issues/",
def create_new_sprint(last_sprint, unresolved_issue_ids): headers=HEADER,
last_end = datetime.fromisoformat(last_sprint["end_date"].replace("Z", "+00:00")) json=attach_payload
new_start = last_end + timedelta(days=1) )
new_end = new_start + timedelta(days=SPRINT_LENGTH) print("Attach response:", attach_res.text)
attach_res.raise_for_status()
start_str = new_start.strftime("%Y-%m-%dT00:00:00Z") else:
end_str = new_end.strftime("%Y-%m-%dT23:59:59Z") print("No unresolved issues to attach.")
return {
print("Creating cycle with name only...") "cycle_id": cycle_id,
"attached_issues": unresolved_issue_ids
create_payload = { }
"name": f"Auto-Sprint {new_start}"
}
create_res = requests.post(
f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/", all_uncompleted = get_all_issues()
headers=HEADER,
json=create_payload last_sprint = get_last_sprint()
)
get_unfinished_issues(last_sprint, all_uncompleted)
print("Create response:", create_res.text)
create_res.raise_for_status() print(create_new_sprint(last_sprint, get_unfinished_issues(last_sprint, all_uncompleted)))
cycle = create_res.json()
cycle_id = cycle["id"]
print("Cycle created:", cycle_id)
patch_payload = {
"start_date": "2025-11-15",
"end_date": "2025-11-16",
"description": f"Automated sprint starting {start_str}"
}
print("➡ Patching cycle with:", patch_payload)
patch_res = requests.patch(
f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/{cycle_id}",
headers=HEADER,
json=patch_payload
)
print("Patch response:", patch_res.text)
patch_res.raise_for_status()
patched_cycle = patch_res.json()
if unresolved_issue_ids:
attach_payload = {"issues": unresolved_issue_ids}
print("Attaching issues:", attach_payload)
attach_res = requests.post(
f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/{cycle_id}/cycle-issues/",
headers=HEADER,
json=attach_payload
)
print("Attach response:", attach_res.text)
attach_res.raise_for_status()
else:
print("No unresolved issues to attach.")
return {
"cycle_id": cycle_id,
"updated": patched_cycle,
"attached_issues": unresolved_issue_ids
}
all_uncompleted = get_all_issues()
last_sprint = get_last_sprint()
get_unfinished_issues(last_sprint, all_uncompleted)
print(create_new_sprint(last_sprint, get_unfinished_issues(last_sprint, all_uncompleted)))
# patch_try = requests.patch(f"{PLANE_URL}/projects/{TESTING_PROJECT}/cycles/52261070-73ae-4ac6-aeed-805dfa5fb3d0",
# headers=HEADER,
# json= {
# "name": "Auto-Sprint 2025-11-15",
# "start_date": "2015-11-17",
# "end_date": "2025-11-19"
# })
# print(f"Status: {patch_try.status_code}")
# print(f"Response: {patch_try.text}")