parent
1806785700
commit
220256dd91
133
update.py
133
update.py
|
@ -11,7 +11,6 @@ from time import time
|
||||||
import deepdiff
|
import deepdiff
|
||||||
import flask
|
import flask
|
||||||
import requests
|
import requests
|
||||||
import yaml
|
|
||||||
|
|
||||||
import agenda.bristol_waste
|
import agenda.bristol_waste
|
||||||
import agenda.fx
|
import agenda.fx
|
||||||
|
@ -109,7 +108,7 @@ Agenda: https://edwardbetts.com/agenda/
|
||||||
agenda.mail.send_mail(config, subject, body)
|
agenda.mail.send_mail(config, subject, body)
|
||||||
|
|
||||||
|
|
||||||
def format_launch_changes(differences: dict) -> str:
|
def format_launch_changes(differences: StrDict) -> str:
|
||||||
"""Convert deepdiff output to human-readable format."""
|
"""Convert deepdiff output to human-readable format."""
|
||||||
changes = []
|
changes = []
|
||||||
# Handle value changes
|
# Handle value changes
|
||||||
|
@ -121,9 +120,14 @@ def format_launch_changes(differences: dict) -> str:
|
||||||
# Format specific fields nicely
|
# Format specific fields nicely
|
||||||
if field == "net":
|
if field == "net":
|
||||||
try:
|
try:
|
||||||
old_dt = datetime.fromisoformat(old_val.replace('Z', '+00:00'))
|
old_dt = datetime.fromisoformat(old_val.replace("Z", "+00:00"))
|
||||||
new_dt = datetime.fromisoformat(new_val.replace('Z', '+00:00'))
|
new_dt = datetime.fromisoformat(new_val.replace("Z", "+00:00"))
|
||||||
changes.append(f"Launch time changed from {old_dt.strftime('%d %b %Y at %H:%M UTC')} to {new_dt.strftime('%d %b %Y at %H:%M UTC')}")
|
changes.append(
|
||||||
|
"Launch time changed from "
|
||||||
|
+ old_dt.strftime("%d %b %Y at %H:%M UTC")
|
||||||
|
+ " to "
|
||||||
|
+ new_dt.strftime("%d %b %Y at %H:%M UTC")
|
||||||
|
)
|
||||||
except:
|
except:
|
||||||
changes.append(f"Launch time changed from {old_val} to {new_val}")
|
changes.append(f"Launch time changed from {old_val} to {new_val}")
|
||||||
elif field == "name":
|
elif field == "name":
|
||||||
|
@ -134,11 +138,16 @@ def format_launch_changes(differences: dict) -> str:
|
||||||
elif new_val is None:
|
elif new_val is None:
|
||||||
changes.append("Launch probability removed")
|
changes.append("Launch probability removed")
|
||||||
else:
|
else:
|
||||||
changes.append(f"Launch probability changed from {old_val}% to {new_val}%")
|
changes.append(
|
||||||
|
f"Launch probability changed from {old_val}% to {new_val}%"
|
||||||
|
)
|
||||||
elif "status" in field:
|
elif "status" in field:
|
||||||
changes.append(f"Status changed from '{old_val}' to '{new_val}'")
|
changes.append(f"Status changed from '{old_val}' to '{new_val}'")
|
||||||
else:
|
else:
|
||||||
changes.append(f"{field.replace('_', ' ').title()} changed from '{old_val}' to '{new_val}'")
|
changes.append(
|
||||||
|
f"{field.replace('_', ' ').title()} changed "
|
||||||
|
+ f"from '{old_val}' to '{new_val}'"
|
||||||
|
)
|
||||||
# Handle additions
|
# Handle additions
|
||||||
if "dictionary_item_added" in differences:
|
if "dictionary_item_added" in differences:
|
||||||
for path in differences["dictionary_item_added"]:
|
for path in differences["dictionary_item_added"]:
|
||||||
|
@ -153,26 +162,118 @@ def format_launch_changes(differences: dict) -> str:
|
||||||
if "type_changes" in differences:
|
if "type_changes" in differences:
|
||||||
for path, change in differences["type_changes"].items():
|
for path, change in differences["type_changes"].items():
|
||||||
field = path.replace("root['", "").replace("']", "").replace("root.", "")
|
field = path.replace("root['", "").replace("']", "").replace("root.", "")
|
||||||
changes.append(f"{field.replace('_', ' ').title()} type changed from {change['old_type'].__name__} to {change['new_type'].__name__}")
|
changes.append(
|
||||||
return "\n".join(f"• {change}" for change in changes) if changes else "No specific changes detected"
|
f"{field.replace('_', ' ').title()} type changed "
|
||||||
|
+ f"from {change['old_type'].__name__} to {change['new_type'].__name__}"
|
||||||
|
)
|
||||||
|
return (
|
||||||
|
"\n".join(f"• {change}" for change in changes)
|
||||||
|
if changes
|
||||||
|
else "No specific changes detected"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def report_space_launch_change(
|
def report_space_launch_change(
|
||||||
config: flask.config.Config, prev_launch: StrDict | None, cur_launch: StrDict | None
|
config: flask.config.Config, prev_launch: StrDict | None, cur_launch: StrDict | None
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Send mail to announce change to space launch data."""
|
"""Send mail to announce change to space launch data."""
|
||||||
|
|
||||||
|
# Handle case where launch disappeared from upcoming list
|
||||||
|
if prev_launch and not cur_launch:
|
||||||
|
# Launch is no longer in upcoming list - could be completed, cancelled, or failed
|
||||||
|
# Check if we can determine status from previous data
|
||||||
|
prev_status_id = (
|
||||||
|
prev_launch.get("status", {}).get("id", 0)
|
||||||
|
if isinstance(prev_launch.get("status"), dict)
|
||||||
|
else 0
|
||||||
|
)
|
||||||
|
name = prev_launch["name"]
|
||||||
|
launch_date = prev_launch.get("net", "Unknown")
|
||||||
|
location = (
|
||||||
|
prev_launch.get("pad", {}).get("location", {}).get("name", "Unknown")
|
||||||
|
if isinstance(prev_launch.get("pad"), dict)
|
||||||
|
else "Unknown"
|
||||||
|
)
|
||||||
|
|
||||||
|
# Since launch is no longer in upcoming list, it likely completed
|
||||||
|
# We can't know the exact outcome, so provide helpful message
|
||||||
|
subject = f"🚀 Space Launch Completed: {name}"
|
||||||
|
|
||||||
|
# Format launch date nicely
|
||||||
|
formatted_date = "Unknown"
|
||||||
|
if launch_date and launch_date != "Unknown":
|
||||||
|
try:
|
||||||
|
dt = datetime.fromisoformat(launch_date.replace("Z", "+00:00"))
|
||||||
|
formatted_date = dt.strftime("%d %b %Y at %H:%M UTC")
|
||||||
|
except:
|
||||||
|
formatted_date = launch_date
|
||||||
|
|
||||||
|
body = f"""🚀 Space Launch Completed
|
||||||
|
|
||||||
|
Mission: {name}
|
||||||
|
Launch Date: {formatted_date}
|
||||||
|
Location: {location}
|
||||||
|
|
||||||
|
This launch is no longer appearing in the upcoming launches list, which typically means it has taken place.
|
||||||
|
|
||||||
|
To check if the launch was successful or failed, visit:
|
||||||
|
https://edwardbetts.com/agenda/launches
|
||||||
|
|
||||||
|
View all launches: https://edwardbetts.com/agenda/launches
|
||||||
|
"""
|
||||||
|
agenda.mail.send_mail(config, subject, body)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Handle regular status updates
|
||||||
if cur_launch:
|
if cur_launch:
|
||||||
name = cur_launch["name"]
|
name = cur_launch["name"]
|
||||||
status = cur_launch.get("status", {}).get("name", "Unknown") if isinstance(cur_launch.get("status"), dict) else "Unknown"
|
status = (
|
||||||
|
cur_launch.get("status", {}).get("name", "Unknown")
|
||||||
|
if isinstance(cur_launch.get("status"), dict)
|
||||||
|
else "Unknown"
|
||||||
|
)
|
||||||
|
status_id = (
|
||||||
|
cur_launch.get("status", {}).get("id", 0)
|
||||||
|
if isinstance(cur_launch.get("status"), dict)
|
||||||
|
else 0
|
||||||
|
)
|
||||||
launch_date = cur_launch.get("net", "Unknown")
|
launch_date = cur_launch.get("net", "Unknown")
|
||||||
location = cur_launch.get("pad", {}).get("location", {}).get("name", "Unknown") if isinstance(cur_launch.get("pad"), dict) else "Unknown"
|
location = (
|
||||||
|
cur_launch.get("pad", {}).get("location", {}).get("name", "Unknown")
|
||||||
|
if isinstance(cur_launch.get("pad"), dict)
|
||||||
|
else "Unknown"
|
||||||
|
)
|
||||||
|
|
||||||
|
# Check for specific status changes that deserve special attention
|
||||||
|
prev_status_id = 0
|
||||||
|
if prev_launch and isinstance(prev_launch.get("status"), dict):
|
||||||
|
prev_status_id = prev_launch.get("status", {}).get("id", 0)
|
||||||
|
|
||||||
|
# Customize subject based on status changes
|
||||||
|
if status_id == 3: # Launch Successful
|
||||||
|
subject = f"🎉 Launch Successful: {name}"
|
||||||
|
elif status_id == 4: # Launch Failure
|
||||||
|
subject = f"💥 Launch Failed: {name}"
|
||||||
|
elif status_id == 7: # Partial Failure
|
||||||
|
subject = f"⚠️ Launch Partial Failure: {name}"
|
||||||
|
elif status_id == 6: # In Flight
|
||||||
|
subject = f"🚀 Launch In Flight: {name}"
|
||||||
|
elif status_id == 5: # On Hold
|
||||||
|
subject = f"⏸️ Launch On Hold: {name}"
|
||||||
|
else:
|
||||||
|
subject = f"Space Launch Update: {name}"
|
||||||
else:
|
else:
|
||||||
|
# This shouldn't happen with the new logic above, but keep as fallback
|
||||||
assert prev_launch
|
assert prev_launch
|
||||||
name = prev_launch["name"]
|
name = prev_launch["name"]
|
||||||
status = "Cancelled/Removed"
|
status = "Unknown"
|
||||||
launch_date = prev_launch.get("net", "Unknown")
|
launch_date = prev_launch.get("net", "Unknown")
|
||||||
location = prev_launch.get("pad", {}).get("location", {}).get("name", "Unknown") if isinstance(prev_launch.get("pad"), dict) else "Unknown"
|
location = (
|
||||||
subject = f"Space Launch Update: {name}"
|
prev_launch.get("pad", {}).get("location", {}).get("name", "Unknown")
|
||||||
|
if isinstance(prev_launch.get("pad"), dict)
|
||||||
|
else "Unknown"
|
||||||
|
)
|
||||||
|
subject = f"Space Launch Update: {name}"
|
||||||
|
|
||||||
differences = deepdiff.DeepDiff(prev_launch, cur_launch)
|
differences = deepdiff.DeepDiff(prev_launch, cur_launch)
|
||||||
changes_text = format_launch_changes(differences)
|
changes_text = format_launch_changes(differences)
|
||||||
|
@ -180,8 +281,8 @@ def report_space_launch_change(
|
||||||
formatted_date = "Unknown"
|
formatted_date = "Unknown"
|
||||||
if launch_date and launch_date != "Unknown":
|
if launch_date and launch_date != "Unknown":
|
||||||
try:
|
try:
|
||||||
dt = datetime.fromisoformat(launch_date.replace('Z', '+00:00'))
|
dt = datetime.fromisoformat(launch_date.replace("Z", "+00:00"))
|
||||||
formatted_date = dt.strftime('%d %b %Y at %H:%M UTC')
|
formatted_date = dt.strftime("%d %b %Y at %H:%M UTC")
|
||||||
except:
|
except:
|
||||||
formatted_date = launch_date
|
formatted_date = launch_date
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue