feat(JW Import): match invoice references to ERPNext docs

This commit is contained in:
Glen Whitney 2020-11-19 23:55:10 +00:00
parent a3dfdc31ff
commit b9106e89d4
3 changed files with 201 additions and 1 deletions

View File

@ -0,0 +1,55 @@
[
{
"allow_in_quick_entry": 0,
"allow_on_submit": 0,
"bold": 0,
"collapsible": 0,
"collapsible_depends_on": null,
"columns": 0,
"default": null,
"depends_on": null,
"description": "Used for matching Justworks invoice details; if set, should be identical to the \"Reference\" field of a Justworks record for this Employee.",
"docstatus": 0,
"doctype": "Custom Field",
"dt": "Employee",
"fetch_from": null,
"fetch_if_empty": 0,
"fieldname": "justworks_name",
"fieldtype": "Data",
"hidden": 0,
"hide_border": 0,
"hide_days": 0,
"hide_seconds": 0,
"ignore_user_permissions": 0,
"ignore_xss_filter": 0,
"in_global_search": 0,
"in_list_view": 0,
"in_preview": 0,
"in_standard_filter": 0,
"insert_after": "employee_number",
"label": "Justworks Name",
"length": 0,
"mandatory_depends_on": null,
"modified": "2020-11-19 17:06:54.757674",
"name": "Employee-justworks_name",
"no_copy": 0,
"non_negative": 0,
"options": null,
"parent": null,
"parentfield": null,
"parenttype": null,
"permlevel": 0,
"precision": "",
"print_hide": 0,
"print_hide_if_no_value": 0,
"print_width": null,
"read_only": 0,
"read_only_depends_on": null,
"report_hide": 0,
"reqd": 0,
"search_index": 0,
"translatable": 0,
"unique": 0,
"width": null
}
]

View File

@ -12,7 +12,7 @@ app_email = "code@studioinfinity.org"
app_license = "GNU General Public License (v3)"
fixtures = ['Custom Field']
accounting_dimension_doctypes = ['JW Employee Allocation']
accounting_dimension_doctypes = ['JW Employee Allocation', 'JW Import']
after_install = "justworks_erpnext.install.after_install"

View File

@ -7,15 +7,22 @@ import frappe
from frappe.model.document import Document
from frappe.utils.csvutils import read_csv_content
from frappe.utils.dateutils import parse_date
from fnmatch import fnmatch as wildcard_match
from operator import itemgetter
from os.path import commonprefix
import re
class JWImport(Document):
account_mapping_fields = 'charge_type charge_subtype map_to_account'.split()
def process_invoice_file(self):
fcontent = frappe.get_doc('File',
dict(file_url=self.load_from_file)).get_content()
rows = read_csv_content(fcontent)
header = [frappe.scrub(h) for h in rows.pop(0)]
self.set('lines', [])
# Load the base data from the file
for row in rows:
rec = self.append('lines', {})
for fieldname, val in zip(header, row):
@ -25,3 +32,141 @@ class JWImport(Document):
elif fieldname == 'amount':
val = re.sub('^[^.,\d]*', '', val)
rec.set(fieldname, val)
# Now determine all of the derived data
self.map_accounts(self.lines)
self.determine_references(self.lines)
#self.check_data(self.lines)
#self.determine_disposition(self.lines)
def map_accounts(self, recs):
amf = self.account_mapping_fields
acc_maps = frappe.get_all('JW Account Mapping', fields=amf)
for rec in recs:
for amap in acc_maps:
if all(wildcard_match((rec.get(fld,'') or ''),
amap.get(fld))
for fld in amf[:-1]):
rec.set('account', amap.get(amf[-1]))
break
def determine_references(self, recs):
expense_acct = frappe.get_value('Company',
self.justworks_company,
'default_expense_claim_payable_account')
# First determine the reference type for each record
for rec in recs:
acct = rec.get('account', '')
if not acct: continue
if acct == expense_acct:
rec.set('reference_doctype', 'Expense Claim')
continue
if frappe.get_value('Account', acct, 'account_type') == 'Payable':
rec.set('reference_doctype', 'Purchase Invoice')
continue
if frappe.get_value('Account', acct, 'root_type') == 'Expense':
rec.set('reference_doctype', 'Employee')
# Now try to determine the matching doc for each record
for rec in recs:
doctype = rec.get('reference_doctype', '')
if not doctype or not rec.get('reference', ''):
continue
if doctype == 'Purchase Invoice':
self.search_invoice(rec)
continue
emp = self.search_employee(rec)
if not emp: continue
if doctype == 'Employee':
rec.set('reference_doc', emp)
continue
# Ok, this is an expense claim:
self.search_expense_claim(rec, emp)
# Look for and return an employee matching the reference of the given
# record, if any
def search_employee(self, rec):
fullref = rec.reference
(last, comma, restref) = fullref.partition(',')
emps = (frappe.get_all('Employee',
filters={'justworks_name': fullref})
or frappe.get_all('Employee',
filters={'last_name': last},
fields=['name', 'first_name']))
if not emps: return ''
if len(emps) == 1: return emps[0].name
# We have multiple employees with same last name, take
# best match for first name
first = restref.partition(' ')[0]
closest = max(emps, key=lambda e: len(commonprefix(e, first)))
return closest.name
# Look for a purchase invoice with the same supplier and amount
# as rec and fill in the reference_doc if found
def search_invoice(self, rec):
# Reference should be a supplier
suppliers = (frappe.get_all('Supplier',
{'supplier_name': rec.reference}) or
frappe.get_all('Supplier', {
'supplier_name':
['like', f"%{rec.reference}%"]})
)
if not suppliers or len(suppliers) != 1: return
suplr = suppliers[0].name
invoices = frappe.get_all('Purchase Invoice',
filters={'supplier': suplr,
'company': self.justworks_company},
fields=['name', 'posting_date as date', 'grand_total',
'rounded_total', 'outstanding_amount'])
if not invoices: return
if len(invoices) == 1:
rec.set('reference_doc', invoices[0].name)
return
match_inv = ([inv for inv in invoices
if inv.outstanding_amount == rec.amount]
or [inv for inv in invoices
if inv.outstanding_amount > rec.amount])
if len(match_inv) > 0:
# Best match is earliest unpaid invoice
earliest = min(match_inv, key=itemgetter('date'))
rec.set('reference_doc', earliest.name)
return
match_inv = [inv for inv in invoices
if inv.grand_total == rec.amount or
inv.rounded_total == rec.amount]
if len(match_inv) > 0:
latest = max(match_inv, key=itemgetter('date'))
rec.set('reference_doc', latest.name)
# Hmmm, supplier has invoices but none with
# enough left to be paid and none with matching
# amount, so punt.
# Look for an expense claim with given emp and the same amount as
# the given rec, and if one is found, fill in the reference_doc
def search_expense_claim(self, rec, emp):
claims = frappe.get_all('Expense Claim',
filters={'employee':emp,
'company': self.justworks_company},
fields=['name', 'total_amount_reimbursed as total',
'is_paid', 'posting_date as date'])
if not claims: return
if len(claims) == 1:
rec.set('reference_doc', claims[0].name)
return
match_claims = {c for c in claims if c.total == rec.amount}
unpaid_claims = {c for c in claims
if c.total >= rec.amount and not c.is_paid}
mu_claims = match_claims & unpaid_claims
if len(mu_claims) > 0:
earliest = min(mu_claims, key=itemgetter('date'))
rec.set('reference_doc', earliest.name)
return
if len(match_claims) > 0:
latest = max(match_claims, key=itemgetter('date'))
rec.set('reference_doc', latest.name)
return
if len(unpaid_claims) > 0:
earliest = min(unpaid_claims, key=itemgetter('date'))
rec.set('reference_doc', earliest.name)
return
# Hmmm, employee has claims but no large enough unpaid ones
# and none with matching amount, so punt.