Empowering ERPNext with Python: Advanced Data Manipulation

# Fetching specific fields and applying filters
tasks = frappe.get_list("Task",
                     filters={"status": "Open", "priority": "High"},
                     fields=["name", "subject", "owner", "due_date"],
                     order_by="due_date ASC",
                     limit=10)
 
for task in tasks:
    frappe.msgprint(f"Task: {task.subject} assigned to {task.owner} is due on {task.due_date}")
 
# Fetching a single value
company_name = frappe.db.get_value("Company", "MyERPNextCompany", "address")
frappe.log_error(f"Company address: {company_name}")

# Example: Find customers with outstanding invoices exceeding a certain amount AND not contacted in the last 30 days
from frappe.utils import date_diff
from datetime import datetime, timedelta
 
thirty_days_ago = datetime.now() - timedelta(days=30)
 
# Get customers with outstanding invoices over $1000
customers_with_large_invoices = set()
for invoice in frappe.get_list("Sales Invoice",
                               filters={"status": "Unpaid"},
                               fields=["name", "customer", "grand_total"]):
    if invoice.grand_total > 1000:
        customers_with_large_invoices.add(invoice.customer)
 
# Get customers contacted recently
recent_contacts = set()
for communication in frappe.get_list("Communication",
                                    filters={"reference_doctype": "Customer", "communication_type": "Email"},
                                    fields=["reference_name", "creation"]):
    if date_diff(datetime.now(), communication.creation) < 30:
        recent_contacts.add(communication.reference_name)
 
# Find customers who meet both criteria
customers_to_target = list(customers_with_large_invoices - recent_contacts)
 
if customers_to_target:
    frappe.log_info(f"Customers to target: {customers_to_target}")
else:
    frappe.log_info("No customers to target based on criteria.")

# hooks.py
 
def after_sales_order_submit(doc, *args, **kwargs):
    if doc.status != "Shipped":
        # Create a follow-up task
        frappe.get_doc({
            "doctype": "Task",
            "subject": f"Follow up on Sales Order {doc.name}",
            "assigned_to": doc.owner, # Or a specific user/team
            "status": "Open",
            "due_date": frappe.utils.add_days(frappe.utils.today(), 3),
            "reference_doctype": "Sales Order",
            "reference_name": doc.name
        }).insert(ignore_permissions=True)
 
frappe.msgprint("Follow-up task created.")
 
after_insert = [
    "your_app.your_module.your_script.after_sales_order_submit",
]

# your_app/your_module/batch_statements.py
 
def send_monthly_statements():
    customers = frappe.get_list("Customer",
                              filters={"credit_limit": (">", 0)},
                              fields=["name", "email_id"])
 
    for customer in customers:
        outstanding_amount = frappe.db.get_value("Sales Invoice",
                                              filters={"customer": customer.name, "status": "Unpaid"},
                                              fieldname="SUM(grand_total)")
 
        if outstanding_amount and outstanding_amount > 0:
            # Logic to generate statement PDF (complex, may involve reporting tools)
            # For simplicity, we'll just log an email trigger
            frappe.sendmail(
                recipients=[customer.email_id],
                subject=f"Your Monthly Statement - Outstanding Balance: {outstanding_amount:.2f}",
                message=f"Dear {customer.name},
 
Your current outstanding balance is {outstanding_amount:.2f}. Please find your statement attached.
 
Regards,
Your Company",
                # attachments=[statement_pdf_path]
            )
            frappe.log_info(f"Statement sent to {customer.name}")
 
# To schedule this job, you would add it to your app's scheduler.json or use frappe.schedule decorator
# Example (if using a dedicated scheduler file):
# {
#   "ref_doctype": "Your DocType",
#   "method": "your_app.your_module.batch_statements.send_monthly_statements",
#   "schedule_interval": "Daily", # Or Monthly, Weekly, etc.
#   "time_of_day": "02:00:00"
# }

# Example: Standardize phone numbers in Contacts DocType
def standardize_phone_numbers():
    contacts = frappe.get_list("Contact",
                              fields=["name", "phone", "phone_mobile"])
 
    for contact in contacts:
        if contact.phone:
            standardized_phone = re.sub(r'[^0-9]', '', contact.phone) # Remove non-digits
            if len(standardized_phone) == 10:
                standardized_phone = "+1" + standardized_phone # Add country code if needed
            frappe.db.set_value("Contact", contact.name, "phone", standardized_phone)
 
        if contact.phone_mobile:
            standardized_mobile = re.sub(r'[^0-9]', '', contact.phone_mobile)
            if len(standardized_mobile) == 10:
                standardized_mobile = "+1" + standardized_mobile
            frappe.db.set_value("Contact", contact.name, "phone_mobile", standardized_mobile)
 
    frappe.log_info("Phone numbers standardized.")
 
# This could be run as a server script or a scheduled job.

# Example: Geocode customer addresses (requires an external geocoding API)
# Assuming you have a function `geocode_address(address)` that returns lat/lon
def geocode_customer_addresses():
    customers = frappe.get_list("Customer",
                              fields=["name", "address_line1", "city", "state", "pincode"])
 
    for customer in customers:
        full_address = f"{customer.address_line1}, {customer.city}, {customer.state} {customer.pincode}"
        if customer.pincode and not customer.get('latitude'): # Avoid re-geocoding
            try:
                # Replace with your actual geocoding API call
                # For demonstration, assuming a mock function `get_lat_lon_from_address`
                lat_lon = get_lat_lon_from_address(full_address)
                if lat_lon:
                    frappe.db.set_value("Customer", customer.name, "latitude", lat_lon['lat'])
                    frappe.db.set_value("Customer", customer.name, "longitude", lat_lon['lon'])
                    frappe.log_info(f"Geocoded customer {customer.name}")
            except Exception as e:
                frappe.log_error(f"Could not geocode {customer.name}: {e}")
 
# Mock function for demonstration
def get_lat_lon_from_address(address):
    # In a real scenario, this would call a service like Google Maps Geocoding API, OpenStreetMap Nominatim, etc.
    # For testing, return dummy data if address seems valid
    if "New York" in address: return {'lat': 40.7128, 'lon': -74.0060}
    return None