roadmapMerge / main.py
Rakshitjan's picture
Update main.py
e9b26f6 verified
# # Additional imports
# from typing import List, Dict
# from fastapi import FastAPI, HTTPException
# from pydantic import BaseModel
# import json
# from datetime import datetime
# # Initialize FastAPI
# app = FastAPI()
# # Merge roadmaps
# def merge_roadmaps(regular_roadmap: Dict, test_roadmap: Dict) -> Dict:
# """Merge two roadmaps into one, combining overlapping and non-overlapping dates."""
# def schedule_to_dict(schedule: List[Dict]) -> Dict:
# """Convert a roadmap's schedule into a dictionary indexed by dates."""
# return {day["date"]: day for day in schedule}
# # Convert schedules to dictionaries for quick lookup
# regular_schedule_dict = schedule_to_dict(regular_roadmap["schedule"])
# test_schedule_dict = schedule_to_dict(test_roadmap["schedule"])
# # Parse dates from schedules
# regular_dates = [datetime.strptime(day["date"], "%Y-%m-%d") for day in regular_roadmap["schedule"]]
# test_dates = [datetime.strptime(day["date"], "%Y-%m-%d") for day in test_roadmap["schedule"]]
# # Identify start and end dates for both roadmaps
# regular_start, regular_end = min(regular_dates), max(regular_dates)
# test_start, test_end = min(test_dates), max(test_dates)
# # Merge overlapping dates
# for date, test_day in test_schedule_dict.items():
# if date in regular_schedule_dict:
# # Append test roadmap subjects to matching days in regular roadmap
# regular_schedule_dict[date]["subjects"].extend(test_day["subjects"])
# else:
# # Add non-overlapping test roadmap dates
# regular_schedule_dict[date] = test_day
# # Convert the merged schedule dictionary back to a sorted list
# merged_schedule = sorted(regular_schedule_dict.values(), key=lambda x: x["date"])
# return {"schedule": merged_schedule}
# # New endpoint for merging roadmaps
# @app.post("/merge-roadmaps/")
# def merge_roadmaps_endpoint(roadmap1: Dict, roadmap2: Dict):
# """
# Merge two roadmaps into one, combining overlapping and non-overlapping dates.
# """
# try:
# merged_roadmap = merge_roadmaps(roadmap1, roadmap2)
# return merged_roadmap
# except Exception as e:
# raise HTTPException(status_code=500, detail=f"Error merging roadmaps: {e}")
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import json
app = FastAPI()
class MergeRoadmapRequest(BaseModel):
regular_roadmap: str
test_roadmaps: List[str]
class MergeRoadmapResponse(BaseModel):
merged_roadmap: str
def merge_roadmaps(regular_roadmap_str, test_roadmaps_str_list):
# Convert input strings to JSON
regular_roadmap = json.loads(regular_roadmap_str)
test_roadmaps = [json.loads(roadmap) for roadmap in test_roadmaps_str_list]
# Helper to convert a roadmap's schedule into a dictionary indexed by dates
def schedule_to_dict(schedule):
return {day["date"]: day for day in schedule}
# Convert the regular roadmap's schedule to a dictionary
regular_schedule_dict = schedule_to_dict(regular_roadmap["schedule"])
# Process each test roadmap
for test_roadmap in test_roadmaps:
# Convert the test roadmap's schedule to a dictionary
test_schedule_dict = schedule_to_dict(test_roadmap["schedule"])
# Merge overlapping dates
for date, test_day in test_schedule_dict.items():
if date in regular_schedule_dict:
# Append test roadmap subjects to matching day in regular roadmap
if "test_study" not in regular_schedule_dict[date]:
regular_schedule_dict[date]["test_study"] = []
regular_schedule_dict[date]["test_study"].extend(test_day["subjects"])
else:
# Add test roadmap dates missing in the regular schedule with 'test_study' field
test_day["test_study"] = test_day.get("test_study", []) + test_day["subjects"]
del test_day["subjects"] # Remove 'subjects' as they're moved to 'test_study'
regular_schedule_dict[date] = test_day
# Extend the regular roadmap to include missing days from the test roadmap
# for date, test_day in test_schedule_dict.items():
# if date not in regular_schedule_dict:
# regular_schedule_dict[date] = test_day
# Convert the dictionary back to a sorted list
merged_schedule = sorted(
(day for day in regular_schedule_dict.values() if day["date"] != 'YYYY-MM-DD'),
key=lambda x: x["date"]
)
for i, day in enumerate(merged_schedule, start=1):
day["dayNumber"] = i
return {"schedule": merged_schedule}
@app.post("/merge-roadmaps", response_model=MergeRoadmapResponse)
def merge_roadmaps_endpoint(request: MergeRoadmapRequest):
try:
# Merge the roadmaps
merged_roadmap = merge_roadmaps(request.regular_roadmap, request.test_roadmaps)
# Return the merged roadmap as a JSON string
return MergeRoadmapResponse(merged_roadmap=json.dumps(merged_roadmap))
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error merging roadmaps: {str(e)}")