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 — returns Loads with their Pallets."""
loads = frappe.get_list("Load",
fields=["name", "load_number", "incoming_date", "customer", "customer_name",
"total_devices", "total_weight", "data_status", "red_r2"],
order_by="incoming_date desc",
limit_page_length=100,
)
for load in loads:
pallets = frappe.get_list("Pallet",
filters={"load": load.name},
fields=["name", "pallet_number", "received_date", "inbound_weight",
"total_items", "data_status", "red_r2", "description", "status"],
limit_page_length=50,
)
load["pallets"] = pallets
load["pallet_count"] = len(pallets)
return {"checkins": loads}
@frappe.whitelist()
def checkin_load(pickup_name, received_date, actual_pallets, total_weight, load_contents, data_status=None, red_r2=None):
"""Check in a load: create Load + Pallets, mark pickup Complete."""
# Get the pickup
pickup = frappe.get_doc("Scheduled Pickup", pickup_name)
if not actual_pallets or int(actual_pallets) < 1:
frappe.throw("Actual pallet count must be at least 1")
actual_pallets = int(actual_pallets)
# Resolve customer - customer_number on pickup is a Link to Customer
customer_id = pickup.customer_number
if not customer_id or not frappe.db.exists("Customer", customer_id):
frappe.throw("Customer {} not found. Please verify the customer on the pickup.".format(customer_id))
# Generate Load name: MMDDYYYY-CustomerNumber format
from datetime import datetime
try:
dt = datetime.strptime(received_date, "%Y-%m-%d")
date_part = dt.strftime("%m%d%Y")
except (ValueError, TypeError):
date_part = received_date.replace("-", "")
cust_num = customer_id
load_name = "{}-{}".format(date_part, cust_num)
# Make unique if name already exists
base_name = load_name
counter = 1
while frappe.db.exists("Load", load_name):
load_name = "{}-{}".format(base_name, counter)
counter += 1
# Create Load
load = frappe.get_doc({
"doctype": "Load",
"name": load_name,
"load_number": load_name,
"incoming_date": received_date,
"customer": customer_id,
"customer_name": pickup.company_name or "",
"customer_number": customer_id,
"data_status": data_status or pickup.data_status or "",
"red_r2": red_r2 or pickup.red_r2 or "",
"total_weight": float(total_weight) if total_weight else 0,
"total_devices": actual_pallets,
"material_items": [{
"material_type": "# Of Pallets",
"total_count": actual_pallets,
"weight": float(total_weight) if total_weight else 0,
"initial_data_status": data_status or pickup.data_status or "D0",
}],
})
load.insert()
frappe.db.commit()
# Create Pallets — autoname=pallet_number, so set name=pallet_number
for i in range(actual_pallets):
pallet_num = "{}-P{}".format(load_name, i + 1)
pallet = frappe.get_doc({
"doctype": "Pallet",
"name": pallet_num,
"pallet_number": pallet_num,
"received_date": received_date,
"load": load.name,
"company_name": pickup.company_name or "",
"inbound_weight": str(round(float(total_weight) / actual_pallets, 1)) if total_weight and actual_pallets else "",
"description": load_contents or "",
"data_status": data_status or pickup.data_status or "",
"red_r2": red_r2 or pickup.red_r2 or "",
"contact_name": pickup.contact_name or "",
"contact_number": pickup.contact_phone or "",
"contact_email": pickup.contact_email or "",
"address_line": (pickup.address_line or "") + ((", " + pickup.city) if pickup.city else "") + ((", " + pickup.state) if pickup.state else ""),
"needs_aor": pickup.needs_aor or 0,
"needs_cod": pickup.needs_cod or 0,
"notes": pickup.notes or "",
"pickup": pickup.pickup_type or "",
"status": "Received",
})
pallet.insert()
frappe.db.commit()
# Update pickup status
pickup.status = "Complete"
pickup.save()
frappe.db.commit()
return {
"success": True,
"load": load.name,
"pallets_created": actual_pallets,
}
@frappe.whitelist()
def get_pickup_details(pickup_name):
"""Get full details of a Scheduled Pickup for the check-in form."""
pickup = frappe.get_doc("Scheduled Pickup", pickup_name)
return {
"name": pickup.name,
"pickup_date": pickup.pickup_date,
"pickup_type": pickup.pickup_type,
"customer_number": pickup.customer_number,
"company_name": pickup.company_name,
"contact_name": pickup.contact_name,
"contact_phone": pickup.contact_phone,
"contact_email": pickup.contact_email,
"address_line": pickup.address_line,
"city": pickup.city,
"state": pickup.state,
"zip_code": pickup.zip_code,
"estimated_items": pickup.estimated_items,
"estimated_weight": pickup.estimated_weight,
"data_status": pickup.data_status,
"red_r2": pickup.red_r2,
"needs_aor": pickup.needs_aor,
"needs_cod": pickup.needs_cod,
"notes": pickup.notes,
}
@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 += "| Date | Customer | Items | Weight | "
html += "Contents | Data Status | RED/R2 | AoR | CoD |
"
for p in pickups:
html += "| " + str(p.get("pickup_date", "")) + " | "
html += "" + str(p.get("company_name", "")) + " | "
html += "" + str(p.get("estimated_items", "")) + " | "
html += "" + str(p.get("estimated_weight", "")) + " | "
html += "" + str(p.get("load_contents", "")) + " | "
html += "" + str(p.get("data_status", "")) + " | "
html += "" + str(p.get("red_r2", "")) + " | "
html += "" + ("✓" if p.get("needs_aor") else "") + " | "
html += "" + ("✓" if p.get("needs_cod") else "") + " |
"
html += "
"
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 += '"
html += "| # | Customer | Address | Contact | "
html += "Items | Weight | Data | RED/R2 | AoR | CoD | Notes |
"
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 += "| " + str(i) + " | " + str(s.get("company_name", "")) + " | "
html += "" + addr + " | " + str(s.get("contact_name", "")) + " " + str(s.get("contact_phone", "")) + " | "
html += "" + str(s.get("estimated_items", "")) + " | " + str(s.get("estimated_weight", "")) + " | "
html += "" + str(s.get("data_status", "")) + " | " + str(s.get("red_r2", "")) + " | "
html += "" + ("✓" if s.get("needs_aor") else "") + " | "
html += "" + ("✓" if s.get("needs_cod") else "") + " | "
html += "" + str(s.get("notes", "")) + " |
"
html += "
"
if unassigned:
html += "Unassigned
| Customer | Address | Notes |
"
for s in unassigned:
html += "| " + str(s.get("company_name", "")) + " | "
html += "" + str(s.get("address_line", "")) + " | "
html += "" + str(s.get("notes", "")) + " |
"
html += "
"
html += ""
frappe.local.response["type"] = "html"
frappe.local.response["page_content"] = html
@frappe.whitelist()
def print_green_sheet(date=None):
"""Generate Green Sheet printout for each pallet.
Shows customer info, service level banner, driver instructions, RED LINE instructions."""
if not date:
date = today()
# Get completed loads for this date
loads = frappe.get_list("Load",
filters={"incoming_date": date},
fields=["name", "load_number", "customer", "customer_name",
"incoming_date", "total_weight", "data_status", "red_r2"],
order_by="name asc",
limit_page_length=200,
)
html = "Green Sheets"
html += ""
for load in loads:
pallets = frappe.get_list("Pallet",
filters={"load": load.name},
fields=["name", "pallet_number", "inbound_weight", "total_items",
"data_status", "red_r2", "description", "needs_aor", "needs_cod", "notes", "status"],
limit_page_length=50,
)
for pallet in pallets:
html += ''
html += '
🟢 GREEN SHEET — Data-Bearing Equipment Tracking
'
html += '
Pallet # ' + str(pallet.get("pallet_number", "")) + ' | Load # ' + str(load.get("name", "")) + ' | ' + str(load.get("incoming_date", "")) + '
'
# Customer block
html += '
'
service_level = ""
if pallet.get("red_r2"):
service_level = pallet.get("red_r2", "")
html += '(' + str(load.get("customer", "")) + ') — ' + str(service_level) + ' — ' + str(load.get("customer_name", "")) + ''
html += '
'
# Service level banner (only for RED/NIST)
rr = pallet.get("red_r2", "")
if rr and rr != "Neither":
html += '
SERVICE LEVEL: ' + str(rr).upper() + '
'
# Driver instructions (notes)
if pallet.get("notes"):
html += '
Driver Instructions:
' + str(pallet.get("notes", "")) + '
'
# RED LINE instructions (for RED/NIST)
if rr and rr not in ("", "Neither", "R2"):
html += '
⚠ RED LINE INSTRUCTIONS
All data-bearing equipment must be tracked. Destruction method per customer specification.
'
# Pallet details table
html += '
| Pallet Designation | Data Status |
'
html += '| ' + str(pallet.get("status", "Received")) + ' | ' + str(pallet.get("data_status", "")) + ' |
'
html += '| Inbound Weight | Total Items |
'
html += '| ' + str(pallet.get("inbound_weight", "")) + ' | ' + str(pallet.get("total_items", "")) + ' |
'
html += '| AoR/CoR | Contents |
'
aor_cor = ""
if pallet.get("needs_aor"): aor_cor += "✓ AoR "
if pallet.get("needs_cod"): aor_cor += "✓ CoD"
html += '| ' + (aor_cor or "None") + ' | ' + str(pallet.get("description", "")) + ' |
'
html += '
'
# Material tracking (hand-write on paper)
html += '
| Material | % | Weight | Sign Off | Date |
'
for _ in range(4):
html += '| | | | | |
'
html += '
'
# R2 warning
html += '
⚠ R2 REQUIREMENT: This pallet contains data-bearing equipment. All devices must be tracked through erasure with 5% verification audit.
'
# Signatures
html += '
| Received By | Inspected By | Verified By |
'
html += '| | | |
'
# Footer
html += ''
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