st01765
374 lines
12 KiB
Python
374 lines
12 KiB
Python
from django.forms import model_to_dict
|
|
|
|
from ..models import *
|
|
|
|
|
|
def _to_date(date):
|
|
if date is None:
|
|
return None
|
|
return date.strftime("%Y-%m-%d")
|
|
|
|
def serialize_model(instance, ignore_fields=None, extra_mappings=None):
|
|
|
|
if ignore_fields is None:
|
|
ignore_fields = []
|
|
if extra_mappings is None:
|
|
extra_mappings = {}
|
|
|
|
data = model_to_dict(instance)
|
|
|
|
for field in ignore_fields:
|
|
data.pop(field, None)
|
|
|
|
for field, value in extra_mappings.items():
|
|
data[field] = value
|
|
|
|
return data
|
|
|
|
def serialize_suburb(suburb):
|
|
return serialize_model(
|
|
suburb
|
|
)
|
|
|
|
def serialize_school(school):
|
|
return serialize_model(
|
|
school,
|
|
extra_mappings={
|
|
"suburb": school.suburb.name if school.suburb else None,
|
|
}
|
|
)
|
|
|
|
def serialize_bus(bus):
|
|
bus_dict = serialize_model(
|
|
bus,
|
|
ignore_fields=["company"],
|
|
)
|
|
|
|
bus_dict["drivers"] = [
|
|
serialize_model(driver, ignore_fields=["bus"])
|
|
for driver in Driver.objects.filter(bus=bus)
|
|
]
|
|
|
|
bus_dict["shuttles"] = [
|
|
serialize_model(shuttle, ignore_fields=["bus"])
|
|
for shuttle in Shuttle.objects.filter(bus=bus)
|
|
]
|
|
|
|
bus_dict["bus_stops"] = [
|
|
serialize_model(
|
|
bus_stop,
|
|
ignore_fields=["bus"],
|
|
extra_mappings={
|
|
"am_time": bus_stop.am_time.strftime("%H:%M:%S") if bus_stop.am_time else None,
|
|
"pm_time": bus_stop.pm_time.strftime("%H:%M:%S") if bus_stop.pm_time else None,
|
|
},
|
|
)
|
|
for bus_stop in BusStop.objects.filter(bus=bus)
|
|
]
|
|
|
|
return bus_dict
|
|
|
|
|
|
def serialize_company(company):
|
|
company_dict = serialize_model(
|
|
company,
|
|
extra_mappings={
|
|
"suburb": company.suburb.name if company.suburb else None,
|
|
},
|
|
)
|
|
company_dict["buses"] = [serialize_bus(bus) for bus in Bus.objects.filter(company=company)]
|
|
return company_dict
|
|
|
|
|
|
def serialize_traveller(traveller):
|
|
traveller_dict = serialize_model(
|
|
traveller,
|
|
ignore_fields=['travel_start_date', 'travel_end_date', 'is_active', 'is_archived', 'bus_stops', 'fee_per_term', 'address'],
|
|
extra_mappings = {
|
|
"school": traveller.school.name if traveller.school else None,
|
|
"dob": _to_date(traveller.dob),
|
|
"assessment_date": _to_date(traveller.assessment_date),
|
|
"created_on": traveller.created_on.strftime("%Y-%m-%d %H:%M:%S %Z"),
|
|
"last_edit": traveller.last_edit.strftime("%Y-%m-%d %H:%M:%S %Z"),
|
|
}
|
|
)
|
|
|
|
traveller_dict["families"] = [
|
|
serialize_model(
|
|
family,
|
|
ignore_fields=["traveller"],
|
|
extra_mappings={
|
|
"residential_suburb": family.residential_suburb.name if family.residential_suburb else None,
|
|
"postal_suburb": family.postal_suburb.name if family.postal_suburb else None,
|
|
},
|
|
)
|
|
for family in Family.objects.filter(traveller=traveller)
|
|
]
|
|
|
|
traveller_dict["bus_stops"] = [
|
|
serialize_model(
|
|
route,
|
|
ignore_fields=["traveller"],
|
|
extra_mappings={
|
|
"travel_start_date": _to_date(traveller.travel_start_date),
|
|
}
|
|
)
|
|
for route in TravellerRoute.objects.filter(traveller=traveller)
|
|
]
|
|
|
|
return traveller_dict
|
|
|
|
|
|
def serialize_all():
|
|
return {
|
|
"suburbs": [serialize_suburb(suburb) for suburb in Suburb.objects.all()],
|
|
"schools": [serialize_school(school) for school in School.objects.all()],
|
|
"companies": [serialize_company(company) for company in Company.objects.all()],
|
|
"travellers": [serialize_traveller(traveller) for traveller in Traveller.objects.all()],
|
|
}
|
|
|
|
|
|
def compare_backup(backup_content):
|
|
import json
|
|
|
|
# Parse uploaded file
|
|
backup_data = json.loads(backup_content)
|
|
|
|
# Current DB state
|
|
current_data = serialize_all()
|
|
|
|
def compare_dicts_shallow(current, backup, ignore_keys=None):
|
|
"""Compare two dicts at a single level."""
|
|
ignore_keys = ignore_keys or []
|
|
modifications = []
|
|
|
|
keys = set(current.keys()) | set(backup.keys())
|
|
|
|
for key in keys:
|
|
if key in ignore_keys:
|
|
continue
|
|
cur_val = current.get(key)
|
|
bak_val = backup.get(key)
|
|
|
|
if cur_val != bak_val:
|
|
if cur_val is None:
|
|
mod_type = "New"
|
|
elif bak_val is None:
|
|
mod_type = "Removed"
|
|
else:
|
|
mod_type = "Updated"
|
|
|
|
modifications.append({
|
|
"type": mod_type,
|
|
"key": key,
|
|
"old_value": cur_val,
|
|
"new_value": bak_val
|
|
})
|
|
return modifications
|
|
|
|
def compare_backup_dicts(current, backup, key, nested_keys=None):
|
|
current_list = current.get(key, [])
|
|
backup_list = backup.get(key, [])
|
|
|
|
if not (current_list and backup_list):
|
|
return []
|
|
|
|
nested_keys = nested_keys or []
|
|
|
|
current_map = {item[key]: item for item in current_list}
|
|
backup_map = {item[key]: item for item in backup_list}
|
|
|
|
results = []
|
|
|
|
# Check for added items
|
|
for name in backup_map:
|
|
if name not in current_map:
|
|
results.append({
|
|
"name": name,
|
|
"type": "New",
|
|
"modifications": [],
|
|
"inline_modifications": []
|
|
})
|
|
|
|
# Check for removed items
|
|
for name in current_map:
|
|
if name not in backup_map:
|
|
results.append({
|
|
"name": name,
|
|
"type": "Removed",
|
|
"modifications": [],
|
|
"inline_modifications": []
|
|
})
|
|
|
|
# Check for updated items
|
|
for name in current_map:
|
|
if name in backup_map:
|
|
cur_item = current_map[name]
|
|
bak_item = backup_map[name]
|
|
|
|
modifications = compare_dicts_shallow(cur_item, bak_item, ignore_keys=nested_keys)
|
|
inline_modifications = []
|
|
|
|
for nested_key in nested_keys:
|
|
nested_diff = compare_backup_dicts(current_list, current_list, nested_key, nested_keys)
|
|
if nested_diff:
|
|
inline_modifications.append({
|
|
"name": nested_key.capitalize(),
|
|
"modifications": nested_diff
|
|
})
|
|
|
|
if modifications or inline_modifications:
|
|
results.append({
|
|
"name": name,
|
|
"type": "Updated",
|
|
"modifications": modifications,
|
|
"inline_modifications": inline_modifications
|
|
})
|
|
|
|
return results
|
|
|
|
diff_suburbs = compare_backup_dicts(
|
|
current_data,
|
|
backup_data,
|
|
"suburbs"
|
|
)
|
|
diff_schools = compare_backup_dicts(
|
|
current_data,
|
|
backup_data,
|
|
"schools"
|
|
)
|
|
diff_companies = compare_backup_dicts(
|
|
current_data,
|
|
backup_data,
|
|
"companies",
|
|
["buses", "drivers", "shuttles", "bus_stops"]
|
|
)
|
|
diff_travellers = compare_backup_dicts(
|
|
current_data,
|
|
backup_data,
|
|
"travellers",
|
|
["families", "bus_stops"]
|
|
)
|
|
|
|
# Helper function to compare lists by key
|
|
def compare_lists(current, backup, key="name"):
|
|
current_map = {item[key]: item for item in current}
|
|
backup_map = {item[key]: item for item in backup}
|
|
|
|
added = [backup_map[k] for k in backup_map if k not in current_map]
|
|
removed = [current_map[k] for k in current_map if k not in backup_map]
|
|
modified = [
|
|
(k, {"db": current_map[k], "backup": backup_map[k]})
|
|
for k in current_map
|
|
if k in backup_map and current_map[k] != backup_map[k]
|
|
]
|
|
|
|
return {
|
|
"add": len(added),
|
|
"remove": len(removed),
|
|
"modify": len(modified),
|
|
"details": {
|
|
"added": added,
|
|
"removed": removed,
|
|
"modified": modified,
|
|
},
|
|
}
|
|
|
|
results = {
|
|
"suburbs": compare_lists(current_data["suburbs"], backup_data["suburbs"], key="name"),
|
|
"schools": compare_lists(current_data["schools"], backup_data["schools"], key="name"),
|
|
"companies": compare_lists(current_data["companies"], backup_data["companies"], key="name"),
|
|
"travellers": compare_lists(current_data["travellers"], backup_data["travellers"], key="name"),
|
|
}
|
|
|
|
# Summary counts only
|
|
summary = {k: {m: v[m] for m in ["add", "remove", "modify"]} for k, v in results.items()}
|
|
|
|
return summary, results
|
|
|
|
|
|
def get_export_dict():
|
|
suburbs = []
|
|
schools = []
|
|
companies = []
|
|
travellers = []
|
|
|
|
for suburb in Suburb.objects.all():
|
|
suburbs.append(model_to_dict(suburb))
|
|
|
|
for school in School.objects.all():
|
|
school_dict = model_to_dict(school)
|
|
school_dict['suburb'] = school.suburb.name
|
|
schools.append(school_dict)
|
|
|
|
for company in Company.objects.all():
|
|
company_dict = model_to_dict(company)
|
|
company_dict['suburb'] = company.suburb.name
|
|
buses = []
|
|
for bus in Bus.objects.filter(company=company):
|
|
bus_dict = model_to_dict(bus)
|
|
bus_dict.pop('company')
|
|
drivers = []
|
|
shuttles = []
|
|
bus_stops = []
|
|
|
|
for driver in Driver.objects.filter(bus=bus):
|
|
driver_dict = model_to_dict(driver)
|
|
driver_dict.pop('bus')
|
|
drivers.append(driver_dict)
|
|
|
|
for shuttle in Shuttle.objects.filter(bus=bus):
|
|
shuttle_dict = model_to_dict(shuttle)
|
|
shuttle_dict.pop('bus')
|
|
shuttles.append(shuttle_dict)
|
|
|
|
for busStop in BusStop.objects.filter(bus=bus):
|
|
bus_stop_dict = model_to_dict(busStop)
|
|
bus_stop_dict.pop('bus')
|
|
bus_stop_dict['am_time'] = busStop.am_time.strftime("%H:%M:%S")
|
|
bus_stop_dict['pm_time'] = busStop.pm_time.strftime("%H:%M:%S")
|
|
bus_stops.append(bus_stop_dict)
|
|
|
|
bus_dict['drivers'] = drivers
|
|
bus_dict['shuttles'] = shuttles
|
|
bus_dict['bus_stops'] = bus_stops
|
|
buses.append(bus_dict)
|
|
company_dict['buses'] = buses
|
|
companies.append(company_dict)
|
|
|
|
ignore_fields = ['travel_start_date', 'travel_end_date', 'is_active', 'is_archived', 'bus_stops', 'fee_per_term', 'address']
|
|
|
|
for traveller in Traveller.objects.all():
|
|
traveller_dict = model_to_dict(traveller)
|
|
traveller_dict['school'] = traveller.school.name
|
|
traveller_dict['dob'] = _to_date(traveller.dob)
|
|
traveller_dict['assessment_date'] = _to_date(traveller.assessment_date)
|
|
traveller_dict['created_on'] = traveller.created_on.strftime("%Y-%m-%d %H:%M:%S %Z")
|
|
traveller_dict['last_edit'] = traveller.last_edit.strftime("%Y-%m-%d %H:%M:%S %Z")
|
|
for field in ignore_fields:
|
|
traveller_dict.pop(field)
|
|
|
|
families = []
|
|
for family in Family.objects.filter(traveller=traveller):
|
|
family_dict = model_to_dict(family)
|
|
family_dict.pop('traveller')
|
|
if family.residential_suburb is not None:
|
|
family_dict['residential_suburb'] = family.residential_suburb.name
|
|
if family.postal_suburb is not None:
|
|
family_dict['postal_suburb'] = family.postal_suburb.name
|
|
families.append(family_dict)
|
|
traveller_dict['families'] = families
|
|
|
|
routes = []
|
|
for route in TravellerRoute.objects.filter(traveller=traveller):
|
|
route_dict = model_to_dict(route)
|
|
route_dict.pop('traveller')
|
|
route_dict['name'] = route.busStop.__str__()
|
|
route_dict['travel_start_date'] = _to_date(route.travel_start_date)
|
|
route_dict['travel_end_date'] = _to_date(route.travel_end_date)
|
|
routes.append(route_dict)
|
|
traveller_dict['bus_stops'] = routes
|
|
travellers.append(traveller_dict)
|
|
|
|
return {'suburbs': suburbs, 'schools': schools, 'companies': companies, 'travellers': travellers}
|