# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and contributors # License: MIT. See LICENSE from contextlib import suppress import redis import frappe from frappe.utils.data import cstr def publish_progress(percent, title=None, doctype=None, docname=None, description=None): publish_realtime( "progress", {"percent": percent, "title": title, "description": description}, user=frappe.session.user, doctype=doctype, docname=docname, ) def publish_realtime( event: str = None, message: dict = None, room: str = None, user: str = None, doctype: str = None, docname: str = None, task_id: str = None, after_commit: bool = False, ): """Publish real-time updates :param event: Event name, like `task_progress` etc. that will be handled by the client (default is `task_progress` if within task or `global`) :param message: JSON message object. For async must contain `task_id` :param room: Room in which to publish update (default entire site) :param user: Transmit to user :param doctype: Transmit to doctype, docname :param docname: Transmit to doctype, docname :param after_commit: (default False) will emit after current transaction is committed""" if message is None: message = {} if event is None: event = "task_progress" if frappe.local.task_id else "global" elif event == "msgprint" and not user: user = frappe.session.user elif event == "list_update": doctype = doctype or message.get("doctype") room = get_doctype_room(doctype) elif event == "docinfo_update": room = get_doc_room(doctype, docname) if not task_id and hasattr(frappe.local, "task_id"): task_id = frappe.local.task_id if not room: if task_id: after_commit = False if "task_id" not in message: message["task_id"] = task_id room = get_task_progress_room(task_id) elif user: # transmit to specific user: System, Website or Guest room = get_user_room(user) elif doctype and docname: room = get_doc_room(doctype, docname) else: # This will be broadcasted to all Desk users room = get_site_room() if after_commit: if not hasattr(frappe.local, "_realtime_log"): frappe.local._realtime_log = [] frappe.db.after_commit.add(flush_realtime_log) frappe.db.after_rollback.add(clear_realtime_log) params = [event, message, room] if params not in frappe.local._realtime_log: frappe.local._realtime_log.append(params) else: emit_via_redis(event, message, room) def flush_realtime_log(): for args in frappe.local._realtime_log: frappe.realtime.emit_via_redis(*args) clear_realtime_log() def clear_realtime_log(): if hasattr(frappe.local, "_realtime_log"): del frappe.local._realtime_log def emit_via_redis(event, message, room): """Publish real-time updates via redis :param event: Event name, like `task_progress` etc. :param message: JSON message object. For async must contain `task_id` :param room: name of the room""" from frappe.utils.background_jobs import get_redis_conn with suppress(redis.exceptions.ConnectionError): r = get_redis_conn() r.publish("events", frappe.as_json({"event": event, "message": message, "room": room})) @frappe.whitelist(allow_guest=True) def can_subscribe_doc(doctype: str, docname: str) -> bool: from frappe.exceptions import PermissionError if not frappe.has_permission(doctype=doctype, doc=docname, ptype="read"): raise PermissionError() return True @frappe.whitelist(allow_guest=True) def can_subscribe_doctype(doctype: str) -> bool: from frappe.exceptions import PermissionError if not frappe.has_permission(doctype=doctype, ptype="read"): raise PermissionError() return True @frappe.whitelist(allow_guest=True) def get_user_info(): return { "user": frappe.session.user, "user_type": frappe.session.data.user_type, } def get_doctype_room(doctype): return f"{frappe.local.site}:doctype:{doctype}" def get_doc_room(doctype, docname): return f"{frappe.local.site}:doc:{doctype}/{cstr(docname)}" def get_user_room(user): return f"{frappe.local.site}:user:{user}" def get_site_room(): return f"{frappe.local.site}:all" def get_task_progress_room(task_id): return f"{frappe.local.site}:task_progress:{task_id}" def get_website_room(): return f"{frappe.local.site}:website"