| # ERPNext - web based ERP (http://erpnext.com) |
| # Copyright (C) 2012 Web Notes Technologies Pvt Ltd |
| # |
| # This program is free software: you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation, either version 3 of the License, or |
| # (at your option) any later version. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License |
| # along with this program. If not, see <http://www.gnu.org/licenses/>. |
| |
| from __future__ import unicode_literals |
| import webnotes |
| from webnotes import msgprint, _ |
| from webnotes.utils import load_json, cstr, flt, now_datetime |
| from webnotes.model.doc import addchild |
| |
| from controllers.status_updater import StatusUpdater |
| |
| class TransactionBase(StatusUpdater): |
| def get_default_address_and_contact(self, party_type): |
| """get a dict of default field values of address and contact for a given party type |
| party_type can be one of: customer, supplier""" |
| ret = {} |
| |
| # {customer: self.doc.fields.get("customer")} |
| args = {party_type: self.doc.fields.get(party_type)} |
| |
| address_text, address_name = self.get_address_text(**args) |
| ret.update({ |
| # customer_address |
| (party_type + "_address"): address_name, |
| "address_display": address_text |
| }) |
| ret.update(self.get_contact_text(**args)) |
| return ret |
| |
| # Get Customer Default Primary Address - first load |
| def get_default_customer_address(self, args=''): |
| address_text, address_name = self.get_address_text(customer=self.doc.customer) |
| self.doc.customer_address = address_name or '' |
| self.doc.address_display = address_text or '' |
| self.doc.fields.update(self.get_contact_text(customer=self.doc.customer)) |
| |
| if args != 'onload': |
| self.get_customer_details(self.doc.customer) |
| self.get_sales_person(self.doc.customer) |
| |
| # Get Customer Default Shipping Address - first load |
| # ----------------------- |
| def get_default_customer_shipping_address(self, args=''): |
| address_text, address_name = self.get_address_text(customer=self.doc.customer,is_shipping_address=1) |
| self.doc.customer_address = address_name or '' |
| self.doc.address_display = address_text or '' |
| self.doc.fields.update(self.get_contact_text(customer=self.doc.customer)) |
| |
| if self.doc.doctype != 'Quotation' and args != 'onload': |
| self.get_customer_details(self.doc.customer) |
| self.get_sales_person(self.doc.customer) |
| |
| # Get Customer Address |
| # ----------------------- |
| def get_customer_address(self, args): |
| args = load_json(args) |
| address_text, address_name = self.get_address_text(address_name=args['address']) |
| ret = { |
| 'customer_address' : address_name, |
| 'address_display' : address_text, |
| } |
| |
| ret.update(self.get_contact_text(contact_name=args['contact'])) |
| |
| return ret |
| |
| # Get Address Text |
| # ----------------------- |
| def get_address_text(self, customer=None, address_name=None, supplier=None, is_shipping_address=None): |
| if customer: |
| cond = customer and 'customer="%s"' % customer or 'name="%s"' % address_name |
| elif supplier: |
| cond = supplier and 'supplier="%s"' % supplier or 'name="%s"' % address_name |
| else: |
| cond = 'name="%s"' % address_name |
| |
| if is_shipping_address: |
| details = webnotes.conn.sql("select name, address_line1, address_line2, city, country, pincode, state, phone, fax from `tabAddress` where %s and docstatus != 2 order by is_shipping_address desc, is_primary_address desc limit 1" % cond, as_dict = 1) |
| else: |
| details = webnotes.conn.sql("select name, address_line1, address_line2, city, country, pincode, state, phone, fax from `tabAddress` where %s and docstatus != 2 order by is_primary_address desc limit 1" % cond, as_dict = 1) |
| |
| extract = lambda x: details and details[0] and details[0].get(x,'') or '' |
| address_fields = [('','address_line1'),('\n','address_line2'),('\n','city'),('\n','state'),(' ','pincode'),('\n','country'),('\nPhone: ','phone'),('\nFax: ', 'fax')] |
| address_display = ''.join([a[0]+extract(a[1]) for a in address_fields if extract(a[1])]) |
| if address_display.startswith('\n'): address_display = address_display[1:] |
| |
| address_name = details and details[0]['name'] or '' |
| return address_display, address_name |
| |
| # Get Contact Text |
| # ----------------------- |
| def get_contact_text(self, customer=None, contact_name=None, supplier=None): |
| if customer: |
| cond = customer and 'customer="%s"' % customer or 'name="%s"' % contact_name |
| elif supplier: |
| cond = supplier and 'supplier="%s"' % supplier or 'name="%s"' % contact_name |
| else: |
| cond = 'name="%s"' % contact_name |
| |
| details = webnotes.conn.sql("select name, first_name, last_name, email_id, phone, mobile_no, department, designation from `tabContact` where %s and docstatus != 2 order by is_primary_contact desc limit 1" % cond, as_dict = 1) |
| |
| extract = lambda x: details and details[0] and details[0].get(x,'') or '' |
| contact_fields = [('','first_name'),(' ','last_name')] |
| contact_display = ''.join([a[0]+cstr(extract(a[1])) for a in contact_fields if extract(a[1])]) |
| if contact_display.startswith('\n'): contact_display = contact_display[1:] |
| |
| return { |
| "contact_display": contact_display, |
| "contact_person": details and details[0]["name"] or "", |
| "contact_email": details and details[0]["email_id"] or "", |
| "contact_mobile": details and details[0]["mobile_no"] or "", |
| "contact_designation": details and details[0]["designation"] or "", |
| "contact_department": details and details[0]["department"] or "", |
| } |
| |
| def get_customer_details(self, name): |
| """ |
| Get customer details like name, group, territory |
| and other such defaults |
| """ |
| customer_details = webnotes.conn.sql("""\ |
| select |
| customer_name, customer_group, territory, |
| default_sales_partner, default_commission_rate, default_currency, |
| default_price_list |
| from `tabCustomer` |
| where name = %s and docstatus < 2""", name, as_dict=1) |
| if customer_details: |
| for f in ['customer_name', 'customer_group', 'territory']: |
| self.doc.fields[f] = customer_details[0][f] or self.doc.fields.get(f) |
| |
| # fields prepended with default in Customer doctype |
| for f in ['sales_partner', 'commission_rate', 'currency']: |
| self.doc.fields[f] = customer_details[0]["default_%s" % f] or self.doc.fields.get(f) |
| |
| # optionally fetch default price list from Customer Group |
| self.doc.price_list_name = (customer_details[0]['default_price_list'] |
| or webnotes.conn.get_value('Customer Group', self.doc.customer_group, |
| 'default_price_list') |
| or self.doc.fields.get('price_list_name')) |
| |
| # Get Customer Shipping Address |
| # ----------------------- |
| def get_shipping_address(self, name): |
| details = webnotes.conn.sql("select name, address_line1, address_line2, city, country, pincode, state, phone from `tabAddress` where customer = '%s' and docstatus != 2 order by is_shipping_address desc, is_primary_address desc limit 1" %(name), as_dict = 1) |
| |
| extract = lambda x: details and details[0] and details[0].get(x,'') or '' |
| address_fields = [('','address_line1'),('\n','address_line2'),('\n','city'),(' ','pincode'),('\n','state'),('\n','country'),('\nPhone: ','phone')] |
| address_display = ''.join([a[0]+extract(a[1]) for a in address_fields if extract(a[1])]) |
| if address_display.startswith('\n'): address_display = address_display[1:] |
| |
| ret = { |
| 'shipping_address_name' : details and details[0]['name'] or '', |
| 'shipping_address' : address_display |
| } |
| return ret |
| |
| # Get Lead Details |
| # ----------------------- |
| def get_lead_details(self, name): |
| details = webnotes.conn.sql("select name, lead_name, address_line1, address_line2, city, country, state, pincode, territory, phone, mobile_no, email_id, company_name from `tabLead` where name = '%s'" %(name), as_dict = 1) |
| |
| extract = lambda x: details and details[0] and details[0].get(x,'') or '' |
| address_fields = [('','address_line1'),('\n','address_line2'),('\n','city'),(' ','pincode'),('\n','state'),('\n','country'),('\nPhone: ','contact_no')] |
| address_display = ''.join([a[0]+extract(a[1]) for a in address_fields if extract(a[1])]) |
| if address_display.startswith('\n'): address_display = address_display[1:] |
| |
| ret = { |
| 'contact_display' : extract('lead_name'), |
| 'address_display' : address_display, |
| 'territory' : extract('territory'), |
| 'contact_mobile' : extract('mobile_no'), |
| 'contact_email' : extract('email_id'), |
| 'customer_name' : extract('company_name') or extract('lead_name') |
| } |
| return ret |
| |
| |
| # Get Supplier Default Primary Address - first load |
| # ----------------------- |
| def get_default_supplier_address(self, args): |
| if isinstance(args, basestring): |
| args = load_json(args) |
| address_text, address_name = self.get_address_text(supplier=args['supplier']) |
| ret = { |
| 'supplier_address' : address_name, |
| 'address_display' : address_text, |
| } |
| ret.update(self.get_contact_text(supplier=args['supplier'])) |
| ret.update(self.get_supplier_details(args['supplier'])) |
| return ret |
| |
| # Get Supplier Address |
| # ----------------------- |
| def get_supplier_address(self, args): |
| args = load_json(args) |
| address_text, address_name = self.get_address_text(address_name=args['address']) |
| ret = { |
| 'supplier_address' : address_name, |
| 'address_display' : address_text, |
| } |
| ret.update(self.get_contact_text(contact_name=args['contact'])) |
| return ret |
| |
| # Get Supplier Details |
| # ----------------------- |
| def get_supplier_details(self, name): |
| supplier_details = webnotes.conn.sql("""\ |
| select supplier_name, default_currency |
| from `tabSupplier` |
| where name = %s and docstatus < 2""", name, as_dict=1) |
| if supplier_details: |
| return { |
| 'supplier_name': (supplier_details[0]['supplier_name'] |
| or self.doc.fields.get('supplier_name')), |
| 'currency': (supplier_details[0]['default_currency'] |
| or self.doc.fields.get('currency')), |
| } |
| else: |
| return {} |
| |
| # Get Sales Person Details of Customer |
| # ------------------------------------ |
| def get_sales_person(self, name): |
| self.doclist = self.doc.clear_table(self.doclist,'sales_team') |
| idx = 0 |
| for d in webnotes.conn.sql("select sales_person, allocated_percentage, allocated_amount, incentives from `tabSales Team` where parent = '%s'" % name): |
| ch = addchild(self.doc, 'sales_team', 'Sales Team', self.doclist) |
| ch.sales_person = d and cstr(d[0]) or '' |
| ch.allocated_percentage = d and flt(d[1]) or 0 |
| ch.allocated_amount = d and flt(d[2]) or 0 |
| ch.incentives = d and flt(d[3]) or 0 |
| ch.idx = idx |
| idx += 1 |
| |
| def load_notification_message(self): |
| dt = self.doc.doctype.lower().replace(" ", "_") |
| if int(webnotes.conn.get_value("Notification Control", None, dt) or 0): |
| self.doc.fields["__notification_message"] = \ |
| webnotes.conn.get_value("Notification Control", None, dt + "_message") |
| |
| def add_communication_list(self): |
| # remove communications if present |
| self.doclist = webnotes.doclist(self.doclist).get({ |
| "doctype": ["!=", "Communcation"]}) |
| |
| comm_list = webnotes.conn.sql("""select * from tabCommunication |
| where %s=%s order by modified desc limit 20""" \ |
| % (self.doc.doctype.replace(" ", "_").lower(), "%s"), |
| self.doc.name, as_dict=1, update={"doctype":"Communication"}) |
| |
| self.doclist.extend(webnotes.doclist([webnotes.doc(fielddata=d) \ |
| for d in comm_list])) |
| |
| def validate_posting_time(self): |
| if not self.doc.posting_time: |
| self.doc.posting_time = now_datetime().strftime('%H:%M:%S') |
| |
| def add_calendar_event(self, opts, force=False): |
| if self.doc.contact_by != cstr(self._prev.contact_by) or \ |
| self.doc.contact_date != cstr(self._prev.contact_date) or force: |
| |
| self.delete_events() |
| self._add_calendar_event(opts) |
| |
| def delete_events(self): |
| webnotes.delete_doc("Event", webnotes.conn.sql_list("""select name from `tabEvent` |
| where ref_type=%s and ref_name=%s""", (self.doc.doctype, self.doc.name))) |
| |
| def _add_calendar_event(self, opts): |
| opts = webnotes._dict(opts) |
| |
| if self.doc.contact_date: |
| event_doclist = [{ |
| "doctype": "Event", |
| "owner": opts.owner or self.doc.owner, |
| "subject": opts.subject, |
| "description": opts.description, |
| "starts_on": self.doc.contact_date + " 10:00:00", |
| "event_type": "Private", |
| "ref_type": self.doc.doctype, |
| "ref_name": self.doc.name |
| }] |
| |
| if webnotes.conn.exists("Profile", self.doc.contact_by): |
| event_doclist.append({ |
| "doctype": "Event User", |
| "parentfield": "event_individuals", |
| "person": self.doc.contact_by |
| }) |
| |
| webnotes.bean(event_doclist).insert() |
| |
| def validate_conversion_rate(currency, conversion_rate, conversion_rate_label, company): |
| """common validation for currency and price list currency""" |
| if conversion_rate == 0: |
| msgprint(conversion_rate_label + _(' cannot be 0'), raise_exception=True) |
| |
| company_currency = webnotes.conn.get_value("Company", company, "default_currency") |
| |
| # parenthesis for 'OR' are necessary as we want it to evaluate as |
| # mandatory valid condition and (1st optional valid condition |
| # or 2nd optional valid condition) |
| valid_conversion_rate = (conversion_rate and |
| ((currency == company_currency and conversion_rate == 1.00) |
| or (currency != company_currency and conversion_rate != 1.00))) |
| |
| if not valid_conversion_rate: |
| msgprint(_('Please enter valid ') + conversion_rate_label + (': ') |
| + ("1 %s = [?] %s" % (currency, company_currency)), |
| raise_exception=True) |
| |
| def validate_item_fetch(args, item): |
| from stock.utils import validate_end_of_life |
| validate_end_of_life(item.name, item.end_of_life) |
| |
| # validate company |
| if not args.company: |
| msgprint(_("Please specify Company"), raise_exception=True) |
| |
| def validate_currency(args, item, meta=None): |
| from webnotes.model.meta import get_field_precision |
| if not meta: |
| meta = webnotes.get_doctype(args.doctype) |
| |
| # validate conversion rate |
| if meta.get_field("currency"): |
| validate_conversion_rate(args.currency, args.conversion_rate, |
| meta.get_label("conversion_rate"), args.company) |
| |
| # round it |
| args.conversion_rate = flt(args.conversion_rate, |
| get_field_precision(meta.get_field("conversion_rate"), |
| webnotes._dict({"fields": args}))) |
| |
| # validate price list conversion rate |
| if meta.get_field("price_list_currency") and args.price_list_name and \ |
| args.price_list_currency: |
| validate_conversion_rate(args.price_list_currency, args.plc_conversion_rate, |
| meta.get_label("plc_conversion_rate"), args.company) |
| |
| # round it |
| args.plc_conversion_rate = flt(args.plc_conversion_rate, |
| get_field_precision(meta.get_field("plc_conversion_rate"), |
| webnotes._dict({"fields": args}))) |
| |
| def delete_events(ref_type, ref_name): |
| webnotes.delete_doc("Event", webnotes.conn.sql_list("""select name from `tabEvent` |
| where ref_type=%s and ref_name=%s""", (ref_type, ref_name)), for_reload=True) |