import json import frappe from frappe.utils import today, getdate, add_days from datetime import timedelta @frappe.whitelist() def get_pickups(date=None): """Fetch scheduled pickups with optional date filter. Returns pickups, calendar (next 30 days), and weekly chart data.""" filters = [] if date: filters.append(["Scheduled Pickup", "pickup_date", "=", date]) fields = [ "name", "pickup_date", "pickup_type", "status", "truck", "stop_order", "customer_number", "company_name", "contact_name", "contact_phone", "contact_email", "address_line", "city", "state", "zip_code", "latitude", "longitude", "estimated_items", "estimated_weight", "load_contents", "num_labels", "data_status", "red_r2", "notes", "legacy_notes", "needs_aor", "needs_cod", ] pickups = frappe.get_list("Scheduled Pickup", fields=fields, filters=filters if filters else None, order_by="pickup_date asc, stop_order asc", limit_page_length=500, ) # Build calendar data (next 30 days) from_date = getdate(today()) to_date = add_days(from_date, 30) all_pickups = frappe.get_list("Scheduled Pickup", fields=["pickup_date"], filters=[["Scheduled Pickup", "pickup_date", ">=", str(from_date)], ["Scheduled Pickup", "pickup_date", "<=", str(to_date)]], limit_page_length=500, ) pickup_counts = {} for p in all_pickups: d = p.get("pickup_date", "") if d: pickup_counts[d] = pickup_counts.get(d, 0) + 1 calendar = [] for i in range(30): d = add_days(from_date, i) ds = str(d) calendar.append({"date": ds, "count": pickup_counts.get(ds, 0)}) # Build weekly chart data (last 12 weeks) weekly = [] for i in range(11, -1, -1): week_start = add_days(from_date, -(from_date.weekday() + 7 * i)) week_end = add_days(week_start, 6) count = 0 for d_str, c in pickup_counts.items(): try: d = getdate(d_str) if week_start <= d <= week_end: count += c except (ValueError, TypeError): pass weekly.append({"label": week_start.strftime("%m/%d"), "count": count}) return { "pickups": pickups, "calendar": calendar, "weekly": weekly, } @frappe.whitelist() def auto_route(date=None): """Auto-assign pickups to trucks based on capacity and proximity.""" if not date: date = today() pickups = frappe.get_list("Scheduled Pickup", filters={"pickup_date": date}, fields=["name", "company_name", "estimated_items", "estimated_weight", "latitude", "longitude", "pickup_type"], limit_page_length=200, ) if not pickups: return {"success": True, "assigned": 0} trucks = ["Truck 1", "Truck 2", "Truck 3"] sorted_p = sorted(pickups, key=lambda p: (float(p.get("latitude") or 0), float(p.get("longitude") or 0))) n = len(sorted_p) assigned = 0 for i, p in enumerate(sorted_p): if p.get("pickup_type") == "Drop-off": truck = "" else: truck = trucks[i % 3] if n <= 3 else trucks[min(i * 3 // n, 2)] doc = frappe.get_doc("Scheduled Pickup", p["name"]) doc.truck = truck doc.status = "Routed" if truck else "Scheduled" doc.stop_order = i + 1 doc.save() assigned += 1 frappe.db.commit() return {"success": True, "assigned": assigned} @frappe.whitelist() def get_checkins(): """Fetch completed check-ins.""" checkins = frappe.get_list("Scheduled Pickup", filters={"status": ["in", ["Complete", "In Progress"]]}, fields=["name", "pickup_date", "pickup_type", "status", "company_name", "customer_number", "estimated_items", "estimated_weight", "load_contents", "data_status", "red_r2", "notes"], order_by="pickup_date desc", limit_page_length=100, ) return {"checkins": checkins} @frappe.whitelist() def cor_report(): """Generate Certificate of Recycling report.""" pickups = frappe.get_list("Scheduled Pickup", filters={"status": "Complete"}, fields=["name", "pickup_date", "company_name", "customer_number", "estimated_items", "estimated_weight", "load_contents", "data_status", "red_r2", "needs_aor", "needs_cod"], order_by="pickup_date desc", limit_page_length=200, ) html = "CoR Report" html += "" html += "

Certificate of Recycling (CoR) Report

" html += "

Generated: " + frappe.utils.now() + "

" html += "

Total completed loads: " + str(len(pickups)) + "

" if pickups: html += "" html += "" for p in pickups: html += "" html += "" html += "" html += "" html += "" html += "" html += "" html += "" html += "" html += "
DateCustomerItemsWeightContentsData StatusRED/R2AoRCoD
" + str(p.get("pickup_date", "")) + "" + str(p.get("company_name", "")) + "" + str(p.get("estimated_items", "")) + "" + str(p.get("estimated_weight", "")) + "" + str(p.get("load_contents", "")) + "" + str(p.get("data_status", "")) + "" + str(p.get("red_r2", "")) + "" + ("✓" if p.get("needs_aor") else "") + "" + ("✓" if p.get("needs_cod") else "") + "
" else: html += "

No completed loads found.

" html += "" frappe.local.response["type"] = "html" frappe.local.response["page_content"] = html @frappe.whitelist() def print_route_sheet(date=None): """Generate printable route sheet.""" if not date: date = today() filters = {"pickup_date": date} if date else {} pickups = frappe.get_list("Scheduled Pickup", filters=filters, fields=["name", "pickup_date", "pickup_type", "status", "truck", "stop_order", "company_name", "contact_name", "contact_phone", "contact_email", "address_line", "city", "state", "zip_code", "estimated_items", "estimated_weight", "load_contents", "data_status", "red_r2", "needs_aor", "needs_cod", "notes"], order_by="truck asc, stop_order asc", limit_page_length=200, ) trucks = {} unassigned = [] for p in pickups: t = p.get("truck", "") if t and t != "Unassigned": trucks.setdefault(t, []).append(p) else: unassigned.append(p) html = "Route Sheet" html += "" html += "

Route Sheet — " + str(date or "Today") + "

" for truck_name, stops in sorted(trucks.items()): html += '
🚛 ' + truck_name + " — " + str(len(stops)) + " stops
" html += "" html += "" for i, s in enumerate(stops, 1): addr = str(s.get("address_line", "")) + ", " + str(s.get("city", "")) + ", " + str(s.get("state", "")) + " " + str(s.get("zip_code", "")) html += "" html += "" html += "" html += "" html += "" html += "" html += "" html += "
#CustomerAddressContactItemsWeightDataRED/R2AoRCoDNotes
" + str(i) + "" + str(s.get("company_name", "")) + "" + addr + "" + str(s.get("contact_name", "")) + "
" + str(s.get("contact_phone", "")) + "
" + str(s.get("estimated_items", "")) + "" + str(s.get("estimated_weight", "")) + "" + str(s.get("data_status", "")) + "" + str(s.get("red_r2", "")) + "" + ("✓" if s.get("needs_aor") else "") + "" + ("✓" if s.get("needs_cod") else "") + "" + str(s.get("notes", "")) + "
" if unassigned: html += "

Unassigned

" for s in unassigned: html += "" html += "" html += "" html += "
CustomerAddressNotes
" + str(s.get("company_name", "")) + "" + str(s.get("address_line", "")) + "" + str(s.get("notes", "")) + "
" html += "" frappe.local.response["type"] = "html" frappe.local.response["page_content"] = html @frappe.whitelist() def print_green_sheet(date=None): """Generate Green Sheet printout. ⚠️ TODO: Green Sheet template needs to be filled in with the actual form layout.""" if not date: date = today() filters = {"pickup_date": date} if date else {} pickups = frappe.get_list("Scheduled Pickup", filters=filters, fields=["name", "pickup_date", "pickup_type", "company_name", "contact_name", "contact_phone", "address_line", "city", "state", "zip_code", "estimated_items", "estimated_weight", "load_contents", "data_status", "red_r2", "needs_aor", "needs_cod", "notes"], order_by="truck asc, stop_order asc", limit_page_length=200, ) html = "Green Sheet" html += "" html += "

🟢 Green Sheet — " + str(date or "Today") + "

" html += '

⚠️ This is a PLACEHOLDER template. Replace with actual Green Sheet layout once the form spec is provided.

' for p in pickups: html += '
' fields = [ ("Customer", p.get("company_name", "")), ("Contact", p.get("contact_name", "")), ("Phone", p.get("contact_phone", "")), ("Address", str(p.get("address_line", "")) + ", " + str(p.get("city", "")) + ", " + str(p.get("state", "")) + " " + str(p.get("zip_code", ""))), ("Est. Items", p.get("estimated_items", "")), ("Est. Weight", p.get("estimated_weight", "")), ("Load Contents", p.get("load_contents", "")), ("Data Status", p.get("data_status", "")), ("RED/R2", p.get("red_r2", "")), ("Needs AoR", "✓" if p.get("needs_aor") else ""), ("Needs CoD", "✓" if p.get("needs_cod") else ""), ("Notes", p.get("notes", "")), ] for label, val in fields: html += '
' + label + ':
' + str(val) + '
' html += '
' html += "" frappe.local.response["type"] = "html" frappe.local.response["page_content"] = html @frappe.whitelist() def print_labels(date=None): """Generate printable labels.""" if not date: date = today() filters = {"pickup_date": date} if date else {} pickups = frappe.get_list("Scheduled Pickup", filters=filters, fields=["name", "company_name", "pickup_date", "num_labels", "data_status", "red_r2"], limit_page_length=200, ) html = "Labels" html += "" for p in pickups: n = p.get("num_labels") or 1 for _ in range(n): html += '
' html += '
' + str(p.get("company_name", "")) + '
' html += '
' + str(p.get("pickup_date", "")) + '
' html += '
' + str(p.get("data_status", "")) + " | " + str(p.get("red_r2", "")) + '
' html += '
' html += "" frappe.local.response["type"] = "html" frappe.local.response["page_content"] = html