import frappe from frappe import _ # All DocTypes managed in sidrah_business that support soft delete & multi tenancy SOFT_DELETE_DOCTYPES = [ "SB Incident", "SB Idea", "SB Order", "SB Process", "SB Organization", "SB Cycle", "SB KPI Input", "Managements", "Department", "Branch", "Incident Types", "Idea Types", "Order Types", "Process Types", "Incident Class", "Incident Field", "Incident Statuses", "Idea Statuses", "Order Statuses", "Process Statuses", "SB Delegation", "SB Delegation Log", "SB Group Company" ] WORKFLOW_STATUS_MAP = { "SB Incident": { "New": "New", "Initial Evaluation": "Initial Evaluation", "Accept & Under Progress": "Accept & Under Progress", "working on resolution": "Accept & Under Progress", "Executed & Closed": "Executed & Closed", "Rejected": "Rejected", }, "SB Idea": { "New": "New", "Initial Evaluation": "Initial Evaluation", "Accept & Under Progress": "Accept & Under Progress", "working on proposal": "Accept & Under Progress", "Executed & Closed": "Executed & Closed", "Rejected": "Rejected", "Moved to Study": "Moved to Study", }, "SB Order": { "Draft": "Bidding", "Pending Actor": "Execution", "In Study": "Execution", "Review": "Validation", "Closed": "Closed", }, "SB Process": { "Draft": "Active", "Pending Actor": "Active", "Pending Comment": "Active", "Review": "Active", "In Study": "Active", "Closed": "Closed", } } def is_ceo_user(user=None): """Checks if user has CEO role and is enabled (full access across the system).""" if not user: user = frappe.session.user if user in ("Administrator", "Guest"): return False enabled = frappe.db.get_value("User", user, "enabled") if not enabled: return False roles = frappe.get_roles(user) return any(r in roles for r in ("CEO", "Ceo", "SB CEO")) def is_platform_admin(user=None): """Platform-level admin: Administrator, System Manager, or enabled CEO.""" if not user: user = frappe.session.user if user == "Administrator": return True roles = frappe.get_roles(user) if any(r in roles for r in ("System Manager", "Administrator")): return True return is_ceo_user(user) def is_admin_user(user=None): if not user: user = frappe.session.user if is_platform_admin(user): return True roles = frappe.get_roles(user) return any(r in roles for r in ("System Manager", "Administrator", "SB Admin")) def is_business_owner_user(user=None): """Checks if user has Business Owner role or custom_category.""" if not user: user = frappe.session.user if user in ("Administrator", "Guest"): return False roles = frappe.get_roles(user) if any(r in roles for r in ("Business Owner", "SB Business Owner", "SB Owner", "Owner")): return True cat = frappe.db.get_value("User", user, "custom_category") return cat == "Business Owner" def is_group_company_admin(user=None, company=None): """ Checks if `user` is Admin of the Group Company. A user is Group Company Admin if: - is_platform_admin(user) is True, OR - User is listed as owner_user in SB Group Company, OR - User has role 'SB Admin' or 'Business Owner' and belongs to that group company. """ if not user: user = frappe.session.user if is_platform_admin(user): return True if is_business_owner_user(user): return True if not company: company = get_user_group_company(user) if not company: owned = frappe.db.get_value("SB Group Company", {"owner_user": user, "custom_is_deleted": 0}, "name") return bool(owned) owner = frappe.db.get_value("SB Group Company", company, "owner_user") if owner and owner == user: return True roles = frappe.get_roles(user) return "SB Admin" in roles def is_manager_user(user=None): if not user: user = frappe.session.user roles = frappe.get_roles(user) if "SB Manager" in roles: return True cat = frappe.db.get_value("User", user, "custom_category") return cat == "Manager" def is_auditor_user(user=None): if not user: user = frappe.session.user roles = frappe.get_roles(user) if any(r in roles for r in ("SB Auditor", "SB X-Auditor")): return True cat = frappe.db.get_value("User", user, "custom_category") return cat in ("Auditor", "X-Auditor") def get_subordinates_recursive(user): """ Recursively fetches all users reporting to `user` directly or indirectly following the hierarchy: Admin -> Manager -> Auditor -> Employee. - For Manager: returns all Auditors under him + all Employees under those Auditors. - For Auditor: returns all Employees under him. - Also includes auditors/delegates linked via accepted SB Delegation. """ if not user: return [] subordinates = set() queue = [user] visited = set([user]) while queue: curr = queue.pop(0) # Direct reports where custom_manager is curr direct = frappe.get_all( "User", filters={"custom_manager": curr, "enabled": 1}, pluck="name" ) # Delegated auditors where manager is curr delegated = frappe.get_all( "SB Delegation", filters={"manager": curr, "status": "Accepted"}, pluck="auditor" ) for sub in direct + delegated: if sub not in visited: visited.add(sub) subordinates.add(sub) queue.append(sub) return list(subordinates) def get_auditors_under_manager(manager_user): """Legacy helper returning all subordinates under manager_user.""" return get_subordinates_recursive(manager_user) def get_user_group_company(user=None): if not user: user = frappe.session.user if user == "Administrator" or is_ceo_user(user): return None co = frappe.db.get_value("User", user, "custom_group_company") if not co: co = frappe.db.get_value("SB Group Company", {"owner_user": user, "custom_is_deleted": 0}, "name") return co def prevent_delete(doc, method=None): """ Frappe before_delete hook. Blocks permanent deletion for non-admins or unflagged deletes. Allows permanent deletion if user is Admin AND frappe.flags.allow_permanent_delete is True, or during system operations (install, patch, migrate). """ if frappe.flags.in_install or frappe.flags.in_patch or frappe.flags.in_uninstall or frappe.flags.in_migrate: return if is_admin_user(frappe.session.user) and getattr(frappe.flags, "allow_permanent_delete", False): return frappe.throw( _("Permanent deletion is restricted. Please use Soft Delete instead."), frappe.ValidationError ) def auto_sync_status_from_workflow(doc, method=None): """ Frappe before_insert / before_save hook. Automatically fetches and fills group_company from logged-in user by default, and updates doc.status according to doc.workflow_state. """ # 1. Auto-fetch group_company from login user and fill by default if hasattr(doc, "group_company"): if not getattr(doc, "group_company", None): user_co = get_user_group_company(frappe.session.user) if user_co: doc.group_company = user_co else: org = getattr(doc, "organization", None) or getattr(doc, "custom_organization", None) if org and frappe.db.exists("SB Organization", org): org_co = frappe.db.get_value("SB Organization", org, "group_company") if org_co: doc.group_company = org_co if doc.doctype not in WORKFLOW_STATUS_MAP: return wf_state = getattr(doc, "workflow_state", None) if not wf_state or not hasattr(doc, "status"): return if doc.is_new() or doc.has_value_changed("workflow_state") or not doc.status: mapping = WORKFLOW_STATUS_MAP.get(doc.doctype, {}) target_status = mapping.get(wf_state, wf_state) doc.status = target_status @frappe.whitelist() def soft_delete(doctype, name, reason=None): """ Whitelisted API method to soft-delete a document. """ if not frappe.has_permission(doctype, "write", doc=name) and not frappe.has_permission(doctype, "delete", doc=name): if not is_admin_user(): frappe.throw(_("Permission denied to delete {0} {1}").format(doctype, name)) doc = frappe.get_doc(doctype, name) if hasattr(doc, "custom_is_deleted"): doc.db_set("custom_is_deleted", 1) doc.db_set("custom_deleted_by", frappe.session.user) doc.db_set("custom_deleted_on", frappe.utils.now()) if reason: doc.db_set("custom_deleted_reason", reason) else: frappe.db.set_value(doctype, name, { "custom_is_deleted": 1, "custom_deleted_by": frappe.session.user, "custom_deleted_on": frappe.utils.now(), "custom_deleted_reason": reason or "" }) frappe.db.commit() return { "success": True, "message": _("Record moved to Deleted.") } @frappe.whitelist() def restore_doc(doctype, name): """ Whitelisted API method to restore a soft-deleted document. Restricted to Administrator / System Manager. """ if not is_admin_user(): frappe.throw(_("Only Administrators can restore deleted records."), frappe.PermissionError) doc = frappe.get_doc(doctype, name) if hasattr(doc, "custom_is_deleted"): doc.db_set("custom_is_deleted", 0) doc.db_set("custom_deleted_by", None) doc.db_set("custom_deleted_on", None) doc.db_set("custom_deleted_reason", None) else: frappe.db.set_value(doctype, name, { "custom_is_deleted": 0, "custom_deleted_by": None, "custom_deleted_on": None, "custom_deleted_reason": None }) frappe.db.commit() return { "success": True, "message": _("Record restored successfully.") } def get_user_organization(user=None): if not user: user = frappe.session.user if user == "Administrator": return None return frappe.db.get_value("User", user, "custom_organization") or None def get_soft_delete_condition(user, doctype=None): if not user: user = frappe.session.user # Root Administrator has unrestricted access across the system if user == "Administrator": return "" user_org = get_user_organization(user) user_co = get_user_group_company(user) # Enabled CEO without specific org/company has full platform visibility if is_ceo_user(user) and not user_co and not user_org: return "" conditions = [] # 1. Multi-tenant Group Company isolation if user_co: safe_co = frappe.db.escape(user_co) if doctype: conditions.append(f"`tab{doctype}`.`group_company` = {safe_co}") else: conditions.append(f"`group_company` = {safe_co}") elif not user_org: # Non-admin user without a group company cannot view tenant data return "1=0" # 2. Organization scoping (applies strictly to ALL users including Admins if set!) if user_org: safe_org = frappe.db.escape(user_org) if doctype == "SB Organization": conditions.append(f"`tab{doctype}`.`name` = {safe_org}") elif doctype == "Department": conditions.append(f"`tab{doctype}`.`custom_organization` = {safe_org}") elif doctype in ("Managements", "Branch", "SB Cycle", "SB Incident", "SB Idea", "SB Order", "SB Process", "SB KPI Input"): conditions.append(f"`tab{doctype}`.`organization` = {safe_org}") elif not doctype: conditions.append(f"`organization` = {safe_org}") # 3. Soft delete filtering for non-admins if not is_group_company_admin(user, user_co) and not is_platform_admin(user): if doctype: conditions.append(f"`tab{doctype}`.`custom_is_deleted` = 0") else: conditions.append("`custom_is_deleted` = 0") # Master doctypes are visible to users within their group company and scoped org if doctype in ("SB Organization", "SB Cycle", "Managements", "Department", "Branch"): return " AND ".join(conditions) # 4. Group Company Admin / Platform Admin: sees ALL data in their company and scoped org if is_group_company_admin(user, user_co) or is_platform_admin(user): return " AND ".join(conditions) safe_user = frappe.db.escape(user) # 5. Manager: sees own data + data of all auditors & employees present under him (recursively) # Manager 1 sees only data present under him, NOT Manager 2's data! if is_manager_user(user): subs = get_subordinates_recursive(user) user_list = [safe_user] + [frappe.db.escape(s) for s in subs] users_in_clause = ", ".join(user_list) clauses = [f"`tab{doctype}`.`owner` IN ({users_in_clause})"] if doctype else [f"`owner` IN ({users_in_clause})"] if doctype in ("SB Incident", "SB Idea", "SB Order"): clauses.append(f"`tab{doctype}`.`assigned_to` IN ({users_in_clause})") if doctype in ("SB Order", "SB Process"): clauses.append(f"`tab{doctype}`.`actor` IN ({users_in_clause})") conditions.append(f"({' OR '.join(clauses)})") return " AND ".join(conditions) # 6. Auditor: sees own data + all data of employees present under him! # Auditor 1 sees E1, E2, E3, E4 under him. Auditor 1 CANNOT see Auditor 2's data or Auditor 2's employees! if is_auditor_user(user): subs = get_subordinates_recursive(user) user_list = [safe_user] + [frappe.db.escape(s) for s in subs] users_in_clause = ", ".join(user_list) clauses = [f"`tab{doctype}`.`owner` IN ({users_in_clause})"] if doctype else [f"`owner` IN ({users_in_clause})"] if doctype in ("SB Incident", "SB Idea", "SB Order"): clauses.append(f"`tab{doctype}`.`assigned_to` IN ({users_in_clause})") if doctype in ("SB Order", "SB Process"): clauses.append(f"`tab{doctype}`.`actor` IN ({users_in_clause})") conditions.append(f"({' OR '.join(clauses)})") return " AND ".join(conditions) # 7. Employee: employee1 sees ONLY his own data (not to see data created by employee2)! clauses = [f"`tab{doctype}`.`owner` = {safe_user}"] if doctype else [f"`owner` = {safe_user}"] if doctype in ("SB Incident", "SB Idea", "SB Order"): clauses.append(f"`tab{doctype}`.`assigned_to` = {safe_user}") if doctype in ("SB Order", "SB Process"): clauses.append(f"`tab{doctype}`.`actor` = {safe_user}") conditions.append(f"({' OR '.join(clauses)})") return " AND ".join(conditions) def sb_incident_condition(user): return get_soft_delete_condition(user, "SB Incident") def sb_idea_condition(user): return get_soft_delete_condition(user, "SB Idea") def sb_order_condition(user): return get_soft_delete_condition(user, "SB Order") def sb_process_condition(user): return get_soft_delete_condition(user, "SB Process") def sb_organization_condition(user): return get_soft_delete_condition(user, "SB Organization") def sb_cycle_condition(user): return get_soft_delete_condition(user, "SB Cycle") def sb_kpi_input_condition(user): return get_soft_delete_condition(user, "SB KPI Input") def managements_condition(user): return get_soft_delete_condition(user, "Managements") def department_condition(user): return get_soft_delete_condition(user, "Department") def branch_condition(user): return get_soft_delete_condition(user, "Branch") def sb_delegation_condition(user): return get_soft_delete_condition(user, "SB Delegation") def sb_group_company_condition(user): return get_soft_delete_condition(user, "SB Group Company") def setup_soft_delete_custom_fields(): """ Ensures custom soft-delete fields exist on all target DocTypes, and seeds missing BRD status records (e.g. Moved to Study for Idea Statuses). """ if frappe.db.exists("DocType", "Idea Statuses"): if not frappe.db.exists("Idea Statuses", "Moved to Study"): doc = frappe.new_doc("Idea Statuses") doc.name1 = "Moved to Study" doc.insert(ignore_permissions=True) frappe.db.commit() from frappe.custom.doctype.custom_field.custom_field import create_custom_fields custom_fields = {} fields_def = [ { "fieldname": "custom_is_deleted", "label": "Is Deleted", "fieldtype": "Check", "default": "0", "read_only": 1, "no_copy": 1, "hidden": 1, }, { "fieldname": "custom_deleted_by", "label": "Deleted By", "fieldtype": "Link", "options": "User", "read_only": 1, "no_copy": 1, "hidden": 1, }, { "fieldname": "custom_deleted_on", "label": "Deleted On", "fieldtype": "Datetime", "read_only": 1, "no_copy": 1, "hidden": 1, }, { "fieldname": "custom_deleted_reason", "label": "Deleted Reason", "fieldtype": "Small Text", "read_only": 1, "no_copy": 1, "hidden": 1, }, ] for dt in SOFT_DELETE_DOCTYPES: if frappe.db.exists("DocType", dt): custom_fields[dt] = fields_def create_custom_fields(custom_fields, ignore_validate=True)