import frappe import json import copy from frappe.model.document import Document from frappe.utils import getdate, nowdate, flt import re import importlib @frappe.whitelist() def get_doctype_fields(doctype): """Получение числовых полей DocType для создания формул""" meta = frappe.get_meta(doctype) fields = [] # Получаем все числовые поля for field in meta.fields: if field.fieldtype in ["Int", "Float", "Currency"]: fields.append({ "fieldname": field.fieldname, "label": field.label or field.fieldname, # ✅ Используем fieldname если нет label "fieldtype": field.fieldtype }) # Проверяем таблицы внутри DocType for child in meta.fields: if child.fieldtype == "Table": child_meta = frappe.get_meta(child.options) for sub_field in child_meta.fields: if sub_field.fieldtype in ["Int", "Float", "Currency"]: # ✅ Используем fieldname если нет label для обоих полей child_label = child.label or child.fieldname sub_field_label = sub_field.label or sub_field.fieldname fields.append({ "fieldname": f"{child.fieldname}_{sub_field.fieldname}", "label": f"{child_label} → {sub_field_label}", "fieldtype": sub_field.fieldtype }) return fields @frappe.whitelist() def get_all_doctype_fields(doctype): """Получение отфильтрованных полей DocType включая расширенные типы для фильтров""" meta = frappe.get_meta(doctype) fields = [] # Типы полей, которые нужно исключить excluded_types = [ "Tab Break", "Column Break", "Button", "Section Break", "Column Break", "HTML", "Table", "Attach", "Attach Image", "Code", "Text Editor", "Image", "Heading", "Fold", "Table MultiSelect" ] # Получаем все поля, которые не входят в исключенные типы for field in meta.fields: if field.fieldname and field.fieldtype not in excluded_types: field_data = { "fieldname": field.fieldname, "label": field.label or field.fieldname, # ✅ Используем fieldname если нет label "fieldtype": field.fieldtype } # Добавляем опции для полей типа Select и Link if field.fieldtype in ["Select", "Link"] and field.options: field_data["options"] = field.options fields.append(field_data) return fields def get_report_module(report_name): """Универсальное получение модуля отчета""" try: # Получаем документ отчета report_doc = frappe.get_doc("Report", report_name) if report_doc.report_type != "Script Report": return None # Пробуем найти модуль отчета через различные пути module_name = report_name.replace(" ", "_").replace("-", "_").lower() # Возможные пути к модулям отчетов possible_paths = [ # ERPNext модули f"erpnext.accounts.report.{module_name}.{module_name}", f"erpnext.selling.report.{module_name}.{module_name}", f"erpnext.buying.report.{module_name}.{module_name}", f"erpnext.stock.report.{module_name}.{module_name}", f"erpnext.manufacturing.report.{module_name}.{module_name}", f"erpnext.projects.report.{module_name}.{module_name}", f"erpnext.hr.report.{module_name}.{module_name}", f"erpnext.crm.report.{module_name}.{module_name}", f"erpnext.assets.report.{module_name}.{module_name}", f"erpnext.setup.report.{module_name}.{module_name}", f"erpnext.utilities.report.{module_name}.{module_name}", # Frappe модули f"frappe.desk.query_report.{module_name}", f"frappe.core.report.{module_name}.{module_name}", # Пользовательские приложения - получаем список всех установленных приложений ] # Добавляем пути для всех установленных приложений apps = frappe.get_installed_apps() for app in apps: if app not in ['frappe', 'erpnext']: # уже добавлены выше possible_paths.extend([ f"{app}.{app}.report.{module_name}.{module_name}", f"{app}.report.{module_name}.{module_name}", f"{app}.reports.{module_name}.{module_name}", ]) # Пробуем импортировать модуль for path in possible_paths: try: module = importlib.import_module(path) return module except ImportError: continue return None except Exception as e: frappe.log_error(f"Error getting report module for {report_name}: {str(e)}") return None @frappe.whitelist() def get_report_filters_dynamic(report_name): """Универсальное получение фильтров отчета""" if not report_name: return [] try: # Способ 1: Пробуем получить через модуль отчета module = get_report_module(report_name) if module and hasattr(module, 'get_filters'): try: filters_meta = module.get_filters() or [] if filters_meta: return format_filters(filters_meta) except Exception as e: frappe.log_error(f"Error calling get_filters for {report_name}: {str(e)}") # Способ 2: Получаем из метаданных отчета report_doc = frappe.get_doc("Report", report_name) if report_doc.report_type != "Script Report": frappe.throw("Выбранный отчет должен быть типа 'Script Report'") filters_meta = report_doc.get('filters') or [] if filters_meta: return format_filters(filters_meta) # Способ 3: Выполняем отчет с пустыми фильтрами и анализируем ошибки # (этот способ сложен в реализации, поэтому возвращаем базовые фильтры) return get_default_report_filters() except Exception as e: frappe.log_error(f"Error in get_report_filters_dynamic: {str(e)}") return get_default_report_filters() @frappe.whitelist() def get_report_fields(report_name, filters=None): """Универсальное получение полей отчета""" if not report_name: frappe.throw("Не указано имя отчета") try: # Обрабатываем фильтры if filters and isinstance(filters, str): try: filters = json.loads(filters) except json.JSONDecodeError: frappe.throw(f"Ошибка при разборе JSON фильтров: {filters}") if not filters: filters = get_basic_filters() # Способ 1: Пробуем получить через модуль отчета module = get_report_module(report_name) if module and hasattr(module, 'get_columns'): try: columns = module.get_columns(filters) or [] if columns: return format_columns(columns) except Exception as e: frappe.log_error(f"Error calling get_columns for {report_name}: {str(e)}") # Способ 2: Выполняем отчет для получения колонок from frappe.desk.query_report import run # Добавляем базовые фильтры если их нет if not filters.get('company'): filters['company'] = frappe.defaults.get_user_default("Company") or frappe.db.get_single_value("Global Defaults", "default_company") try: result = run(report_name, filters=filters) except Exception as e: # Если отчет не выполняется с фильтрами, пробуем с базовыми try: basic_filters = get_basic_filters() result = run(report_name, filters=basic_filters) except Exception as e2: frappe.throw(f"Не удалось выполнить отчет: {str(e2)}") if not result: frappe.throw("Отчет не вернул результатов") columns = result.get('columns', []) if not columns: frappe.throw("В результатах отчета отсутствуют колонки") return format_columns(columns) except Exception as e: frappe.throw(f"Ошибка при получении полей отчета: {str(e)}") def format_filters(filters_meta): """Форматирует фильтры в унифицированный формат""" formatted_filters = [] for filter_meta in filters_meta: # Обработка фильтров в формате списка [fieldname, label, fieldtype, options, default, reqd] if isinstance(filter_meta, list) and len(filter_meta) > 0: filter_data = { 'fieldname': filter_meta[0], 'label': filter_meta[1] if len(filter_meta) > 1 else filter_meta[0], 'fieldtype': filter_meta[2] if len(filter_meta) > 2 else 'Data', 'options': filter_meta[3] if len(filter_meta) > 3 else None, 'default': filter_meta[4] if len(filter_meta) > 4 else None, 'reqd': filter_meta[5] if len(filter_meta) > 5 else 0 } formatted_filters.append(filter_data) # Обработка фильтров в формате словаря elif isinstance(filter_meta, dict) and filter_meta.get('fieldname'): filter_data = { 'fieldname': filter_meta.get('fieldname'), 'label': filter_meta.get('label') or filter_meta.get('fieldname'), 'fieldtype': filter_meta.get('fieldtype') or 'Data', 'options': filter_meta.get('options'), 'default': filter_meta.get('default'), 'reqd': filter_meta.get('reqd') or 0, 'description': filter_meta.get('description') } formatted_filters.append(filter_data) return formatted_filters def format_columns(columns): """Форматирует колонки в унифицированный формат""" fields = [] for i, col in enumerate(columns): field_data = {} if isinstance(col, dict): field_data = { "fieldname": col.get('fieldname') or f"column_{i}", "label": col.get('label') or f"Column {i+1}", "fieldtype": col.get('fieldtype') or "Data" } elif isinstance(col, str): field_data = { "fieldname": f"column_{i}", "label": col, "fieldtype": "Data" } else: field_data = { "fieldname": f"column_{i}", "label": f"Column {i+1}", "fieldtype": "Data" } fields.append(field_data) return fields def get_basic_filters(): """Возвращает базовые фильтры для выполнения отчетов""" return { 'company': frappe.defaults.get_user_default("Company") or frappe.db.get_single_value("Global Defaults", "default_company"), 'from_date': frappe.utils.get_first_day(frappe.utils.nowdate()), 'to_date': frappe.utils.nowdate() } def get_default_report_filters(): """Возвращает базовые фильтры по умолчанию для интерфейса""" return [ { 'fieldname': 'company', 'label': 'Company', 'fieldtype': 'Link', 'options': 'Company', 'default': frappe.defaults.get_user_default("Company"), 'reqd': 1 }, { 'fieldname': 'from_date', 'label': 'From Date', 'fieldtype': 'Date', 'default': frappe.utils.get_first_day(frappe.utils.nowdate()), 'reqd': 1 }, { 'fieldname': 'to_date', 'label': 'To Date', 'fieldtype': 'Date', 'default': frappe.utils.nowdate(), 'reqd': 1 } ] @frappe.whitelist() def run_report_for_formula(report_name, filters, field_name=None): """Универсальное выполнение отчета для формулы""" try: # Парсим фильтры, если они переданы как строка if isinstance(filters, str): filters = json.loads(filters) # Выполняем отчет from frappe.desk.query_report import run report_result = run(report_name, filters=filters) # Проверяем результат на ошибки if isinstance(report_result, dict) and report_result.get('message'): return {"error": report_result.get('message')} columns = report_result.get('columns', []) data = report_result.get('result', []) # Если указано конкретное поле, извлекаем его значение if field_name: result = extract_field_from_report(data, columns, field_name) return result return { "columns": columns, "data": data } except Exception as e: frappe.log_error(f"Error in run_report_for_formula: {str(e)}") return {"error": str(e)} def extract_field_from_report(data, columns, field_name): """Универсальное извлечение значения поля из результатов отчета""" if not data: return 0 # Определяем индекс колонки по имени поля column_idx = None for i, col in enumerate(columns): if isinstance(col, dict): if col.get('fieldname') == field_name or col.get('label') == field_name: column_idx = i break elif isinstance(col, str) and (col == field_name or field_name.lower() in col.lower()): column_idx = i break # Если колонка не найдена, ищем по частичному совпадению if column_idx is None: for i, col in enumerate(columns): col_name = "" if isinstance(col, dict): col_name = col.get('fieldname', '') or col.get('label', '') elif isinstance(col, str): col_name = col if field_name.lower() in col_name.lower(): column_idx = i break if column_idx is not None: # Ищем итоговые строки сначала total_markers = ['total', 'итого', 'grand total', 'общий итог', 'итого по', 'total row'] for row in data: is_total_row = False if isinstance(row, dict): # Проверяем специальные атрибуты итоговых строк if (row.get('is_total_row') or row.get('bold') or row.get('total_row')): is_total_row = True # Проверяем ключи и значения на наличие маркеров итога if not is_total_row: for key, value in row.items(): if isinstance(value, str) and any(marker in value.lower() for marker in total_markers): is_total_row = True break elif isinstance(row, list) and len(row) > 0: # Для списков проверяем первый элемент first_element = str(row[0]) if row[0] is not None else "" if any(marker in first_element.lower() for marker in total_markers): is_total_row = True # Если нашли итоговую строку, извлекаем значение if is_total_row: if isinstance(row, dict): if field_name in row: value = row[field_name] else: # Поиск с игнорированием регистра value = None for key in row: if key.lower() == field_name.lower(): value = row[key] break elif isinstance(row, list) and column_idx < len(row): value = row[column_idx] else: value = None # Преобразуем в число и округляем до 2 знаков try: if value is not None: num_value = float(value) return round(num_value, 2) except (ValueError, TypeError): pass # Если итоговую строку не нашли, суммируем все значения total = 0 count = 0 for row in data: value = None if isinstance(row, dict) and field_name in row: value = row[field_name] elif isinstance(row, dict): # Поиск поля с игнорированием регистра for key in row: if key.lower() == field_name.lower(): value = row[key] break elif isinstance(row, list) and column_idx < len(row): value = row[column_idx] # Преобразуем в число try: if value is not None: num_value = float(value) total += num_value count += 1 except (ValueError, TypeError): pass # Округляем итоговую сумму до 2 знаков return round(total, 2) return 0 @frappe.whitelist() def on_trash(doc, method=None): """Удаляет клиентские скрипты, созданные для DocType при удалении Formula Editor""" if not doc.target_doctype: return # Ищем клиентские скрипты, созданные для целевого DocType client_scripts = frappe.get_all( "Client Script", filters={ "dt": doc.target_doctype, "name": ["like", "Formula_Editor_Script_for_%"] }, fields=["name"] ) # Удаляем найденные скрипты for script in client_scripts: try: frappe.delete_doc("Client Script", script.name, force=True) frappe.db.commit() except Exception as e: pass # --- Client Script generation ------------------------------------------------- # # The generated script is what actually runs on a declaration form. It used to be # built in the browser from the editor's rendered DOM, which meant a half-drawn # form could serialise empty formulas over good ones. It is built here from the # stored JSON instead, so saving a Formula Editor record always regenerates the # script from the data of record. # # The emitted script carries only a CONFIG object and a call into taxes_az.formula. # It contains no reference to this app, so formula_editor can be uninstalled from # production machines and the declarations keep calculating. CLIENT_SCRIPT_PREFIX = "Formula_Editor_Script_for_" # Client Scripts ship to production through taxes_az's fixtures, which select on # this module. CLIENT_SCRIPT_MODULE = "Taxes Az" SCRIPT_TEMPLATE = """// Generated by Formula Editor // Hesablamanı taxes_az.formula aparır (bax: taxes_az/public/js/formula_runtime.js). const CONFIG = {config}; frappe.ui.form.on({doctype}, taxes_az.formula.handlers(CONFIG)); """ def build_config(doc): """Project a Formula Editor record onto the config its runtime consumes. This is a copy of the stored JSON, not a new format — the storage fields are the contract and are passed through untouched. """ return { "doctype": doc.target_doctype, "formulas": json.loads(doc.formula_storage or "{}"), "table_row_formulas": json.loads(doc.table_row_formulas_storage or "{}"), "report_fields": normalize_report_filters(json.loads(doc.report_fields_storage or "{}")), "universal_fields": json.loads(doc.universal_fields_storage or "{}"), } def normalize_report_filters(report_fields): """Force report filters onto the placeholders the runtime knows how to resolve. `company` becomes $DefaultCompany and anything date-shaped becomes $AyIlSpecial, exactly as the previous generator did while baking filters into the script. This is not cosmetic: a different filter value means the report returns different figures, and those figures go on a tax return. Operates on the parsed copy; the stored `filters_json` is never rewritten. """ for field in report_fields.values(): if not field: continue try: filters = json.loads(field.get("filters_json") or "{}") except ValueError: field["filters_json"] = "{}" continue company = filters.get("company") if isinstance(company, str) and not company.startswith("$"): filters["company"] = "$DefaultCompany" for key in list(filters): lowered = key.lower() if not ("date" in lowered or "from" in lowered or "to" in lowered): continue value = filters[key] if not isinstance(value, str): continue if value in ("$AyIlSpecial", "$RubIlSpecial") or value.startswith("doc."): continue filters[key] = "$AyIlSpecial" field["filters_json"] = json.dumps(filters) return report_fields def regenerate_client_script(doc, method=None): """Rebuild the Client Script for a Formula Editor record. Bound to on_update.""" if not doc.target_doctype: return script = SCRIPT_TEMPLATE.format( config=json.dumps(build_config(doc), ensure_ascii=False, indent=1), doctype=json.dumps(doc.target_doctype, ensure_ascii=False), ) name = CLIENT_SCRIPT_PREFIX + doc.target_doctype if frappe.db.exists("Client Script", name): client_script = frappe.get_doc("Client Script", name) client_script.dt = doc.target_doctype client_script.view = "Form" client_script.enabled = 1 client_script.script = script client_script.save(ignore_permissions=True) return frappe.get_doc( { "doctype": "Client Script", "name": name, "dt": doc.target_doctype, "view": "Form", "script_type": "Client", "module": CLIENT_SCRIPT_MODULE, "enabled": 1, "script": script, } ).insert(ignore_permissions=True) @frappe.whitelist() def regenerate_all(): """Regenerate every declaration's Client Script from its Formula Editor record.""" regenerated = [] for name in frappe.get_all("Formula Editor", pluck="name"): doc = frappe.get_doc("Formula Editor", name) regenerate_client_script(doc) regenerated.append(doc.target_doctype) frappe.db.commit() return regenerated