We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add jitter on scheduled jobs
hourly, daily long processes if all started at once can cause sudden increase in workload if you have many sites/benches. 


Adding simple jitter to scheduled time can lessen the impact of such issues. Jitter is common pattern used for solving problems with "frequency" becomes a problem. E.g. gunicorn adds jitter to avoid restarting all workers at same time, profilers add jitter to avoid amplifying some pattern of repeated work. retry/backoff implementations also use to avoid creating patterns.


Possible implementation: When importing scheduled job types add some random delays in cron. E.g. daily jobs will start in the range of  12:00-12:15 AM instead of all starting at 12:00 AM.


Cons: Some jobs are required to be executed at specific times e.g. birthday reminders. So adding negative offset can introduce bugs for them, positive offset however should be fine AFAIK. 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `frappe/core/doctype/scheduled_job_type/scheduled_job_type.py`
Content:
```
1 # Copyright (c) 2021, Frappe Technologies and contributors
2 # License: MIT. See LICENSE
3 
4 import json
5 from datetime import datetime
6 
7 import click
8 from croniter import CroniterBadCronError, croniter
9 
10 import frappe
11 from frappe import _
12 from frappe.model.document import Document
13 from frappe.utils import get_datetime, now_datetime
14 from frappe.utils.background_jobs import enqueue, is_job_enqueued
15 
16 
17 class ScheduledJobType(Document):
18 	def autoname(self):
19 		self.name = ".".join(self.method.split(".")[-2:])
20 
21 	def validate(self):
22 		if self.frequency != "All":
23 			# force logging for all events other than continuous ones (ALL)
24 			self.create_log = 1
25 
26 		if self.frequency == "Cron":
27 			if not self.cron_format:
28 				frappe.throw(_("Cron format is required for job types with Cron frequency."))
29 			try:
30 				croniter(self.cron_format)
31 			except CroniterBadCronError:
32 				frappe.throw(
33 					_("{0} is not a valid Cron expression.").format(f"<code>{self.cron_format}</code>"),
34 					title=_("Bad Cron Expression"),
35 				)
36 
37 	def enqueue(self, force=False) -> bool:
38 		# enqueue event if last execution is done
39 		if self.is_event_due() or force:
40 			if frappe.flags.enqueued_jobs:
41 				frappe.flags.enqueued_jobs.append(self.method)
42 
43 			if frappe.flags.execute_job:
44 				self.execute()
45 			else:
46 				if not self.is_job_in_queue():
47 					enqueue(
48 						"frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job",
49 						queue=self.get_queue_name(),
50 						job_type=self.method,
51 						job_id=self.rq_job_id,
52 					)
53 					return True
54 				else:
55 					frappe.logger("scheduler").error(
56 						f"Skipped queueing {self.method} because it was found in queue for {frappe.local.site}"
57 					)
58 		return False
59 
60 	def is_event_due(self, current_time=None):
61 		"""Return true if event is due based on time lapsed since last execution"""
62 		# if the next scheduled event is before NOW, then its due!
63 		return self.get_next_execution() <= (current_time or now_datetime())
64 
65 	def is_job_in_queue(self) -> bool:
66 		return is_job_enqueued(self.rq_job_id)
67 
68 	@property
69 	def rq_job_id(self):
70 		"""Unique ID created to deduplicate jobs with single RQ call."""
71 		return f"scheduled_job::{self.method}"
72 
73 	@property
74 	def next_execution(self):
75 		return self.get_next_execution()
76 
77 	def get_next_execution(self):
78 		CRON_MAP = {
79 			"Yearly": "0 0 1 1 *",
80 			"Annual": "0 0 1 1 *",
81 			"Monthly": "0 0 1 * *",
82 			"Monthly Long": "0 0 1 * *",
83 			"Weekly": "0 0 * * 0",
84 			"Weekly Long": "0 0 * * 0",
85 			"Daily": "0 0 * * *",
86 			"Daily Long": "0 0 * * *",
87 			"Hourly": "0 * * * *",
88 			"Hourly Long": "0 * * * *",
89 			"All": "0/" + str((frappe.get_conf().scheduler_interval or 240) // 60) + " * * * *",
90 		}
91 
92 		if not self.cron_format:
93 			self.cron_format = CRON_MAP[self.frequency]
94 
95 		# If this is a cold start then last_execution will not be set.
96 		# Creation is set as fallback because if very old fallback is set job might trigger
97 		# immediately, even when it's meant to be daily.
98 		# A dynamic fallback like current time might miss the scheduler interval and job will never start.
99 		last_execution = get_datetime(self.last_execution or self.creation)
100 		return croniter(self.cron_format, last_execution).get_next(datetime)
101 
102 	def execute(self):
103 		self.scheduler_log = None
104 		try:
105 			self.log_status("Start")
106 			if self.server_script:
107 				script_name = frappe.db.get_value("Server Script", self.server_script)
108 				if script_name:
109 					frappe.get_doc("Server Script", script_name).execute_scheduled_method()
110 			else:
111 				frappe.get_attr(self.method)()
112 			frappe.db.commit()
113 			self.log_status("Complete")
114 		except Exception:
115 			frappe.db.rollback()
116 			self.log_status("Failed")
117 
118 	def log_status(self, status):
119 		# log file
120 		frappe.logger("scheduler").info(f"Scheduled Job {status}: {self.method} for {frappe.local.site}")
121 		self.update_scheduler_log(status)
122 
123 	def update_scheduler_log(self, status):
124 		if not self.create_log:
125 			# self.get_next_execution will work properly iff self.last_execution is properly set
126 			if self.frequency == "All" and status == "Start":
127 				self.db_set("last_execution", now_datetime(), update_modified=False)
128 				frappe.db.commit()
129 			return
130 		if not self.scheduler_log:
131 			self.scheduler_log = frappe.get_doc(
132 				dict(doctype="Scheduled Job Log", scheduled_job_type=self.name)
133 			).insert(ignore_permissions=True)
134 		self.scheduler_log.db_set("status", status)
135 		if status == "Failed":
136 			self.scheduler_log.db_set("details", frappe.get_traceback())
137 		if status == "Start":
138 			self.db_set("last_execution", now_datetime(), update_modified=False)
139 		frappe.db.commit()
140 
141 	def get_queue_name(self):
142 		return "long" if ("Long" in self.frequency) else "default"
143 
144 	def on_trash(self):
145 		frappe.db.delete("Scheduled Job Log", {"scheduled_job_type": self.name})
146 
147 
148 @frappe.whitelist()
149 def execute_event(doc: str):
150 	frappe.only_for("System Manager")
151 	doc = json.loads(doc)
152 	frappe.get_doc("Scheduled Job Type", doc.get("name")).enqueue(force=True)
153 	return doc
154 
155 
156 def run_scheduled_job(job_type: str):
157 	"""This is a wrapper function that runs a hooks.scheduler_events method"""
158 	try:
159 		frappe.get_doc("Scheduled Job Type", dict(method=job_type)).execute()
160 	except Exception:
161 		print(frappe.get_traceback())
162 
163 
164 def sync_jobs(hooks: dict | None = None):
165 	frappe.reload_doc("core", "doctype", "scheduled_job_type")
166 	scheduler_events = hooks or frappe.get_hooks("scheduler_events")
167 	all_events = insert_events(scheduler_events)
168 	clear_events(all_events)
169 
170 
171 def insert_events(scheduler_events: dict) -> list:
172 	cron_jobs, event_jobs = [], []
173 	for event_type in scheduler_events:
174 		events = scheduler_events.get(event_type)
175 		if isinstance(events, dict):
176 			cron_jobs += insert_cron_jobs(events)
177 		else:
178 			# hourly, daily etc
179 			event_jobs += insert_event_jobs(events, event_type)
180 	return cron_jobs + event_jobs
181 
182 
183 def insert_cron_jobs(events: dict) -> list:
184 	cron_jobs = []
185 	for cron_format in events:
186 		for event in events.get(cron_format):
187 			cron_jobs.append(event)
188 			insert_single_event("Cron", event, cron_format)
189 	return cron_jobs
190 
191 
192 def insert_event_jobs(events: list, event_type: str) -> list:
193 	event_jobs = []
194 	for event in events:
195 		event_jobs.append(event)
196 		frequency = event_type.replace("_", " ").title()
197 		insert_single_event(frequency, event)
198 	return event_jobs
199 
200 
201 def insert_single_event(frequency: str, event: str, cron_format: str | None = None):
202 	cron_expr = {"cron_format": cron_format} if cron_format else {}
203 
204 	try:
205 		frappe.get_attr(event)
206 	except Exception as e:
207 		click.secho(f"{event} is not a valid method: {e}", fg="yellow")
208 
209 	doc = frappe.get_doc(
210 		{
211 			"doctype": "Scheduled Job Type",
212 			"method": event,
213 			"cron_format": cron_format,
214 			"frequency": frequency,
215 		}
216 	)
217 
218 	if not frappe.db.exists("Scheduled Job Type", {"method": event, "frequency": frequency, **cron_expr}):
219 		savepoint = "scheduled_job_type_creation"
220 		try:
221 			frappe.db.savepoint(savepoint)
222 			doc.insert()
223 		except frappe.DuplicateEntryError:
224 			frappe.db.rollback(save_point=savepoint)
225 			doc.delete()
226 			doc.insert()
227 
228 
229 def clear_events(all_events: list):
230 	for event in frappe.get_all("Scheduled Job Type", fields=["name", "method", "server_script"]):
231 		is_server_script = event.server_script
232 		is_defined_in_hooks = event.method in all_events
233 
234 		if not (is_defined_in_hooks or is_server_script):
235 			frappe.delete_doc("Scheduled Job Type", event.name)
236
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
--- a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
+++ b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
@@ -2,7 +2,8 @@
 # License: MIT. See LICENSE
 
 import json
-from datetime import datetime
+from datetime import datetime, timedelta
+from random import randint
 
 import click
 from croniter import CroniterBadCronError, croniter
@@ -97,7 +98,12 @@
 		# immediately, even when it's meant to be daily.
 		# A dynamic fallback like current time might miss the scheduler interval and job will never start.
 		last_execution = get_datetime(self.last_execution or self.creation)
-		return croniter(self.cron_format, last_execution).get_next(datetime)
+		next_execution = croniter(self.cron_format, last_execution).get_next(datetime)
+
+		jitter = 0
+		if self.frequency in ("Hourly Long", "Daily Long"):
+			jitter = randint(1, 600)
+		return next_execution + timedelta(seconds=jitter)
 
 	def execute(self):
 		self.scheduler_log = None

{"golden_diff": "diff --git a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py\n--- a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py\n+++ b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py\n@@ -2,7 +2,8 @@\n # License: MIT. See LICENSE\n \n import json\n-from datetime import datetime\n+from datetime import datetime, timedelta\n+from random import randint\n \n import click\n from croniter import CroniterBadCronError, croniter\n@@ -97,7 +98,12 @@\n \t\t# immediately, even when it's meant to be daily.\n \t\t# A dynamic fallback like current time might miss the scheduler interval and job will never start.\n \t\tlast_execution = get_datetime(self.last_execution or self.creation)\n-\t\treturn croniter(self.cron_format, last_execution).get_next(datetime)\n+\t\tnext_execution = croniter(self.cron_format, last_execution).get_next(datetime)\n+\n+\t\tjitter = 0\n+\t\tif self.frequency in (\"Hourly Long\", \"Daily Long\"):\n+\t\t\tjitter = randint(1, 600)\n+\t\treturn next_execution + timedelta(seconds=jitter)\n \n \tdef execute(self):\n \t\tself.scheduler_log = None\n", "issue": "Add jitter on scheduled jobs\nhourly, daily long processes if all started at once can cause sudden increase in workload if you have many sites/benches. \r\n\r\n\r\nAdding simple jitter to scheduled time can lessen the impact of such issues. Jitter is common pattern used for solving problems with \"frequency\" becomes a problem. E.g. gunicorn adds jitter to avoid restarting all workers at same time, profilers add jitter to avoid amplifying some pattern of repeated work. retry/backoff implementations also use to avoid creating patterns.\r\n\r\n\r\nPossible implementation: When importing scheduled job types add some random delays in cron. E.g. daily jobs will start in the range of  12:00-12:15 AM instead of all starting at 12:00 AM.\r\n\r\n\r\nCons: Some jobs are required to be executed at specific times e.g. birthday reminders. So adding negative offset can introduce bugs for them, positive offset however should be fine AFAIK. \n", "before_files": [{"content": "# Copyright (c) 2021, Frappe Technologies and contributors\n# License: MIT. See LICENSE\n\nimport json\nfrom datetime import datetime\n\nimport click\nfrom croniter import CroniterBadCronError, croniter\n\nimport frappe\nfrom frappe import _\nfrom frappe.model.document import Document\nfrom frappe.utils import get_datetime, now_datetime\nfrom frappe.utils.background_jobs import enqueue, is_job_enqueued\n\n\nclass ScheduledJobType(Document):\n\tdef autoname(self):\n\t\tself.name = \".\".join(self.method.split(\".\")[-2:])\n\n\tdef validate(self):\n\t\tif self.frequency != \"All\":\n\t\t\t# force logging for all events other than continuous ones (ALL)\n\t\t\tself.create_log = 1\n\n\t\tif self.frequency == \"Cron\":\n\t\t\tif not self.cron_format:\n\t\t\t\tfrappe.throw(_(\"Cron format is required for job types with Cron frequency.\"))\n\t\t\ttry:\n\t\t\t\tcroniter(self.cron_format)\n\t\t\texcept CroniterBadCronError:\n\t\t\t\tfrappe.throw(\n\t\t\t\t\t_(\"{0} is not a valid Cron expression.\").format(f\"<code>{self.cron_format}</code>\"),\n\t\t\t\t\ttitle=_(\"Bad Cron Expression\"),\n\t\t\t\t)\n\n\tdef enqueue(self, force=False) -> bool:\n\t\t# enqueue event if last execution is done\n\t\tif self.is_event_due() or force:\n\t\t\tif frappe.flags.enqueued_jobs:\n\t\t\t\tfrappe.flags.enqueued_jobs.append(self.method)\n\n\t\t\tif frappe.flags.execute_job:\n\t\t\t\tself.execute()\n\t\t\telse:\n\t\t\t\tif not self.is_job_in_queue():\n\t\t\t\t\tenqueue(\n\t\t\t\t\t\t\"frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job\",\n\t\t\t\t\t\tqueue=self.get_queue_name(),\n\t\t\t\t\t\tjob_type=self.method,\n\t\t\t\t\t\tjob_id=self.rq_job_id,\n\t\t\t\t\t)\n\t\t\t\t\treturn True\n\t\t\t\telse:\n\t\t\t\t\tfrappe.logger(\"scheduler\").error(\n\t\t\t\t\t\tf\"Skipped queueing {self.method} because it was found in queue for {frappe.local.site}\"\n\t\t\t\t\t)\n\t\treturn False\n\n\tdef is_event_due(self, current_time=None):\n\t\t\"\"\"Return true if event is due based on time lapsed since last execution\"\"\"\n\t\t# if the next scheduled event is before NOW, then its due!\n\t\treturn self.get_next_execution() <= (current_time or now_datetime())\n\n\tdef is_job_in_queue(self) -> bool:\n\t\treturn is_job_enqueued(self.rq_job_id)\n\n\t@property\n\tdef rq_job_id(self):\n\t\t\"\"\"Unique ID created to deduplicate jobs with single RQ call.\"\"\"\n\t\treturn f\"scheduled_job::{self.method}\"\n\n\t@property\n\tdef next_execution(self):\n\t\treturn self.get_next_execution()\n\n\tdef get_next_execution(self):\n\t\tCRON_MAP = {\n\t\t\t\"Yearly\": \"0 0 1 1 *\",\n\t\t\t\"Annual\": \"0 0 1 1 *\",\n\t\t\t\"Monthly\": \"0 0 1 * *\",\n\t\t\t\"Monthly Long\": \"0 0 1 * *\",\n\t\t\t\"Weekly\": \"0 0 * * 0\",\n\t\t\t\"Weekly Long\": \"0 0 * * 0\",\n\t\t\t\"Daily\": \"0 0 * * *\",\n\t\t\t\"Daily Long\": \"0 0 * * *\",\n\t\t\t\"Hourly\": \"0 * * * *\",\n\t\t\t\"Hourly Long\": \"0 * * * *\",\n\t\t\t\"All\": \"0/\" + str((frappe.get_conf().scheduler_interval or 240) // 60) + \" * * * *\",\n\t\t}\n\n\t\tif not self.cron_format:\n\t\t\tself.cron_format = CRON_MAP[self.frequency]\n\n\t\t# If this is a cold start then last_execution will not be set.\n\t\t# Creation is set as fallback because if very old fallback is set job might trigger\n\t\t# immediately, even when it's meant to be daily.\n\t\t# A dynamic fallback like current time might miss the scheduler interval and job will never start.\n\t\tlast_execution = get_datetime(self.last_execution or self.creation)\n\t\treturn croniter(self.cron_format, last_execution).get_next(datetime)\n\n\tdef execute(self):\n\t\tself.scheduler_log = None\n\t\ttry:\n\t\t\tself.log_status(\"Start\")\n\t\t\tif self.server_script:\n\t\t\t\tscript_name = frappe.db.get_value(\"Server Script\", self.server_script)\n\t\t\t\tif script_name:\n\t\t\t\t\tfrappe.get_doc(\"Server Script\", script_name).execute_scheduled_method()\n\t\t\telse:\n\t\t\t\tfrappe.get_attr(self.method)()\n\t\t\tfrappe.db.commit()\n\t\t\tself.log_status(\"Complete\")\n\t\texcept Exception:\n\t\t\tfrappe.db.rollback()\n\t\t\tself.log_status(\"Failed\")\n\n\tdef log_status(self, status):\n\t\t# log file\n\t\tfrappe.logger(\"scheduler\").info(f\"Scheduled Job {status}: {self.method} for {frappe.local.site}\")\n\t\tself.update_scheduler_log(status)\n\n\tdef update_scheduler_log(self, status):\n\t\tif not self.create_log:\n\t\t\t# self.get_next_execution will work properly iff self.last_execution is properly set\n\t\t\tif self.frequency == \"All\" and status == \"Start\":\n\t\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\t\t\tfrappe.db.commit()\n\t\t\treturn\n\t\tif not self.scheduler_log:\n\t\t\tself.scheduler_log = frappe.get_doc(\n\t\t\t\tdict(doctype=\"Scheduled Job Log\", scheduled_job_type=self.name)\n\t\t\t).insert(ignore_permissions=True)\n\t\tself.scheduler_log.db_set(\"status\", status)\n\t\tif status == \"Failed\":\n\t\t\tself.scheduler_log.db_set(\"details\", frappe.get_traceback())\n\t\tif status == \"Start\":\n\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\tfrappe.db.commit()\n\n\tdef get_queue_name(self):\n\t\treturn \"long\" if (\"Long\" in self.frequency) else \"default\"\n\n\tdef on_trash(self):\n\t\tfrappe.db.delete(\"Scheduled Job Log\", {\"scheduled_job_type\": self.name})\n\n\n@frappe.whitelist()\ndef execute_event(doc: str):\n\tfrappe.only_for(\"System Manager\")\n\tdoc = json.loads(doc)\n\tfrappe.get_doc(\"Scheduled Job Type\", doc.get(\"name\")).enqueue(force=True)\n\treturn doc\n\n\ndef run_scheduled_job(job_type: str):\n\t\"\"\"This is a wrapper function that runs a hooks.scheduler_events method\"\"\"\n\ttry:\n\t\tfrappe.get_doc(\"Scheduled Job Type\", dict(method=job_type)).execute()\n\texcept Exception:\n\t\tprint(frappe.get_traceback())\n\n\ndef sync_jobs(hooks: dict | None = None):\n\tfrappe.reload_doc(\"core\", \"doctype\", \"scheduled_job_type\")\n\tscheduler_events = hooks or frappe.get_hooks(\"scheduler_events\")\n\tall_events = insert_events(scheduler_events)\n\tclear_events(all_events)\n\n\ndef insert_events(scheduler_events: dict) -> list:\n\tcron_jobs, event_jobs = [], []\n\tfor event_type in scheduler_events:\n\t\tevents = scheduler_events.get(event_type)\n\t\tif isinstance(events, dict):\n\t\t\tcron_jobs += insert_cron_jobs(events)\n\t\telse:\n\t\t\t# hourly, daily etc\n\t\t\tevent_jobs += insert_event_jobs(events, event_type)\n\treturn cron_jobs + event_jobs\n\n\ndef insert_cron_jobs(events: dict) -> list:\n\tcron_jobs = []\n\tfor cron_format in events:\n\t\tfor event in events.get(cron_format):\n\t\t\tcron_jobs.append(event)\n\t\t\tinsert_single_event(\"Cron\", event, cron_format)\n\treturn cron_jobs\n\n\ndef insert_event_jobs(events: list, event_type: str) -> list:\n\tevent_jobs = []\n\tfor event in events:\n\t\tevent_jobs.append(event)\n\t\tfrequency = event_type.replace(\"_\", \" \").title()\n\t\tinsert_single_event(frequency, event)\n\treturn event_jobs\n\n\ndef insert_single_event(frequency: str, event: str, cron_format: str | None = None):\n\tcron_expr = {\"cron_format\": cron_format} if cron_format else {}\n\n\ttry:\n\t\tfrappe.get_attr(event)\n\texcept Exception as e:\n\t\tclick.secho(f\"{event} is not a valid method: {e}\", fg=\"yellow\")\n\n\tdoc = frappe.get_doc(\n\t\t{\n\t\t\t\"doctype\": \"Scheduled Job Type\",\n\t\t\t\"method\": event,\n\t\t\t\"cron_format\": cron_format,\n\t\t\t\"frequency\": frequency,\n\t\t}\n\t)\n\n\tif not frappe.db.exists(\"Scheduled Job Type\", {\"method\": event, \"frequency\": frequency, **cron_expr}):\n\t\tsavepoint = \"scheduled_job_type_creation\"\n\t\ttry:\n\t\t\tfrappe.db.savepoint(savepoint)\n\t\t\tdoc.insert()\n\t\texcept frappe.DuplicateEntryError:\n\t\t\tfrappe.db.rollback(save_point=savepoint)\n\t\t\tdoc.delete()\n\t\t\tdoc.insert()\n\n\ndef clear_events(all_events: list):\n\tfor event in frappe.get_all(\"Scheduled Job Type\", fields=[\"name\", \"method\", \"server_script\"]):\n\t\tis_server_script = event.server_script\n\t\tis_defined_in_hooks = event.method in all_events\n\n\t\tif not (is_defined_in_hooks or is_server_script):\n\t\t\tfrappe.delete_doc(\"Scheduled Job Type\", event.name)\n", "path": "frappe/core/doctype/scheduled_job_type/scheduled_job_type.py"}], "after_files": [{"content": "# Copyright (c) 2021, Frappe Technologies and contributors\n# License: MIT. See LICENSE\n\nimport json\nfrom datetime import datetime, timedelta\nfrom random import randint\n\nimport click\nfrom croniter import CroniterBadCronError, croniter\n\nimport frappe\nfrom frappe import _\nfrom frappe.model.document import Document\nfrom frappe.utils import get_datetime, now_datetime\nfrom frappe.utils.background_jobs import enqueue, is_job_enqueued\n\n\nclass ScheduledJobType(Document):\n\tdef autoname(self):\n\t\tself.name = \".\".join(self.method.split(\".\")[-2:])\n\n\tdef validate(self):\n\t\tif self.frequency != \"All\":\n\t\t\t# force logging for all events other than continuous ones (ALL)\n\t\t\tself.create_log = 1\n\n\t\tif self.frequency == \"Cron\":\n\t\t\tif not self.cron_format:\n\t\t\t\tfrappe.throw(_(\"Cron format is required for job types with Cron frequency.\"))\n\t\t\ttry:\n\t\t\t\tcroniter(self.cron_format)\n\t\t\texcept CroniterBadCronError:\n\t\t\t\tfrappe.throw(\n\t\t\t\t\t_(\"{0} is not a valid Cron expression.\").format(f\"<code>{self.cron_format}</code>\"),\n\t\t\t\t\ttitle=_(\"Bad Cron Expression\"),\n\t\t\t\t)\n\n\tdef enqueue(self, force=False) -> bool:\n\t\t# enqueue event if last execution is done\n\t\tif self.is_event_due() or force:\n\t\t\tif frappe.flags.enqueued_jobs:\n\t\t\t\tfrappe.flags.enqueued_jobs.append(self.method)\n\n\t\t\tif frappe.flags.execute_job:\n\t\t\t\tself.execute()\n\t\t\telse:\n\t\t\t\tif not self.is_job_in_queue():\n\t\t\t\t\tenqueue(\n\t\t\t\t\t\t\"frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job\",\n\t\t\t\t\t\tqueue=self.get_queue_name(),\n\t\t\t\t\t\tjob_type=self.method,\n\t\t\t\t\t\tjob_id=self.rq_job_id,\n\t\t\t\t\t)\n\t\t\t\t\treturn True\n\t\t\t\telse:\n\t\t\t\t\tfrappe.logger(\"scheduler\").error(\n\t\t\t\t\t\tf\"Skipped queueing {self.method} because it was found in queue for {frappe.local.site}\"\n\t\t\t\t\t)\n\t\treturn False\n\n\tdef is_event_due(self, current_time=None):\n\t\t\"\"\"Return true if event is due based on time lapsed since last execution\"\"\"\n\t\t# if the next scheduled event is before NOW, then its due!\n\t\treturn self.get_next_execution() <= (current_time or now_datetime())\n\n\tdef is_job_in_queue(self) -> bool:\n\t\treturn is_job_enqueued(self.rq_job_id)\n\n\t@property\n\tdef rq_job_id(self):\n\t\t\"\"\"Unique ID created to deduplicate jobs with single RQ call.\"\"\"\n\t\treturn f\"scheduled_job::{self.method}\"\n\n\t@property\n\tdef next_execution(self):\n\t\treturn self.get_next_execution()\n\n\tdef get_next_execution(self):\n\t\tCRON_MAP = {\n\t\t\t\"Yearly\": \"0 0 1 1 *\",\n\t\t\t\"Annual\": \"0 0 1 1 *\",\n\t\t\t\"Monthly\": \"0 0 1 * *\",\n\t\t\t\"Monthly Long\": \"0 0 1 * *\",\n\t\t\t\"Weekly\": \"0 0 * * 0\",\n\t\t\t\"Weekly Long\": \"0 0 * * 0\",\n\t\t\t\"Daily\": \"0 0 * * *\",\n\t\t\t\"Daily Long\": \"0 0 * * *\",\n\t\t\t\"Hourly\": \"0 * * * *\",\n\t\t\t\"Hourly Long\": \"0 * * * *\",\n\t\t\t\"All\": \"0/\" + str((frappe.get_conf().scheduler_interval or 240) // 60) + \" * * * *\",\n\t\t}\n\n\t\tif not self.cron_format:\n\t\t\tself.cron_format = CRON_MAP[self.frequency]\n\n\t\t# If this is a cold start then last_execution will not be set.\n\t\t# Creation is set as fallback because if very old fallback is set job might trigger\n\t\t# immediately, even when it's meant to be daily.\n\t\t# A dynamic fallback like current time might miss the scheduler interval and job will never start.\n\t\tlast_execution = get_datetime(self.last_execution or self.creation)\n\t\tnext_execution = croniter(self.cron_format, last_execution).get_next(datetime)\n\n\t\tjitter = 0\n\t\tif self.frequency in (\"Hourly Long\", \"Daily Long\"):\n\t\t\tjitter = randint(1, 600)\n\t\treturn next_execution + timedelta(seconds=jitter)\n\n\tdef execute(self):\n\t\tself.scheduler_log = None\n\t\ttry:\n\t\t\tself.log_status(\"Start\")\n\t\t\tif self.server_script:\n\t\t\t\tscript_name = frappe.db.get_value(\"Server Script\", self.server_script)\n\t\t\t\tif script_name:\n\t\t\t\t\tfrappe.get_doc(\"Server Script\", script_name).execute_scheduled_method()\n\t\t\telse:\n\t\t\t\tfrappe.get_attr(self.method)()\n\t\t\tfrappe.db.commit()\n\t\t\tself.log_status(\"Complete\")\n\t\texcept Exception:\n\t\t\tfrappe.db.rollback()\n\t\t\tself.log_status(\"Failed\")\n\n\tdef log_status(self, status):\n\t\t# log file\n\t\tfrappe.logger(\"scheduler\").info(f\"Scheduled Job {status}: {self.method} for {frappe.local.site}\")\n\t\tself.update_scheduler_log(status)\n\n\tdef update_scheduler_log(self, status):\n\t\tif not self.create_log:\n\t\t\t# self.get_next_execution will work properly iff self.last_execution is properly set\n\t\t\tif self.frequency == \"All\" and status == \"Start\":\n\t\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\t\t\tfrappe.db.commit()\n\t\t\treturn\n\t\tif not self.scheduler_log:\n\t\t\tself.scheduler_log = frappe.get_doc(\n\t\t\t\tdict(doctype=\"Scheduled Job Log\", scheduled_job_type=self.name)\n\t\t\t).insert(ignore_permissions=True)\n\t\tself.scheduler_log.db_set(\"status\", status)\n\t\tif status == \"Failed\":\n\t\t\tself.scheduler_log.db_set(\"details\", frappe.get_traceback())\n\t\tif status == \"Start\":\n\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\tfrappe.db.commit()\n\n\tdef get_queue_name(self):\n\t\treturn \"long\" if (\"Long\" in self.frequency) else \"default\"\n\n\tdef on_trash(self):\n\t\tfrappe.db.delete(\"Scheduled Job Log\", {\"scheduled_job_type\": self.name})\n\n\n@frappe.whitelist()\ndef execute_event(doc: str):\n\tfrappe.only_for(\"System Manager\")\n\tdoc = json.loads(doc)\n\tfrappe.get_doc(\"Scheduled Job Type\", doc.get(\"name\")).enqueue(force=True)\n\treturn doc\n\n\ndef run_scheduled_job(job_type: str):\n\t\"\"\"This is a wrapper function that runs a hooks.scheduler_events method\"\"\"\n\ttry:\n\t\tfrappe.get_doc(\"Scheduled Job Type\", dict(method=job_type)).execute()\n\texcept Exception:\n\t\tprint(frappe.get_traceback())\n\n\ndef sync_jobs(hooks: dict | None = None):\n\tfrappe.reload_doc(\"core\", \"doctype\", \"scheduled_job_type\")\n\tscheduler_events = hooks or frappe.get_hooks(\"scheduler_events\")\n\tall_events = insert_events(scheduler_events)\n\tclear_events(all_events)\n\n\ndef insert_events(scheduler_events: dict) -> list:\n\tcron_jobs, event_jobs = [], []\n\tfor event_type in scheduler_events:\n\t\tevents = scheduler_events.get(event_type)\n\t\tif isinstance(events, dict):\n\t\t\tcron_jobs += insert_cron_jobs(events)\n\t\telse:\n\t\t\t# hourly, daily etc\n\t\t\tevent_jobs += insert_event_jobs(events, event_type)\n\treturn cron_jobs + event_jobs\n\n\ndef insert_cron_jobs(events: dict) -> list:\n\tcron_jobs = []\n\tfor cron_format in events:\n\t\tfor event in events.get(cron_format):\n\t\t\tcron_jobs.append(event)\n\t\t\tinsert_single_event(\"Cron\", event, cron_format)\n\treturn cron_jobs\n\n\ndef insert_event_jobs(events: list, event_type: str) -> list:\n\tevent_jobs = []\n\tfor event in events:\n\t\tevent_jobs.append(event)\n\t\tfrequency = event_type.replace(\"_\", \" \").title()\n\t\tinsert_single_event(frequency, event)\n\treturn event_jobs\n\n\ndef insert_single_event(frequency: str, event: str, cron_format: str | None = None):\n\tcron_expr = {\"cron_format\": cron_format} if cron_format else {}\n\n\ttry:\n\t\tfrappe.get_attr(event)\n\texcept Exception as e:\n\t\tclick.secho(f\"{event} is not a valid method: {e}\", fg=\"yellow\")\n\n\tdoc = frappe.get_doc(\n\t\t{\n\t\t\t\"doctype\": \"Scheduled Job Type\",\n\t\t\t\"method\": event,\n\t\t\t\"cron_format\": cron_format,\n\t\t\t\"frequency\": frequency,\n\t\t}\n\t)\n\n\tif not frappe.db.exists(\"Scheduled Job Type\", {\"method\": event, \"frequency\": frequency, **cron_expr}):\n\t\tsavepoint = \"scheduled_job_type_creation\"\n\t\ttry:\n\t\t\tfrappe.db.savepoint(savepoint)\n\t\t\tdoc.insert()\n\t\texcept frappe.DuplicateEntryError:\n\t\t\tfrappe.db.rollback(save_point=savepoint)\n\t\t\tdoc.delete()\n\t\t\tdoc.insert()\n\n\ndef clear_events(all_events: list):\n\tfor event in frappe.get_all(\"Scheduled Job Type\", fields=[\"name\", \"method\", \"server_script\"]):\n\t\tis_server_script = event.server_script\n\t\tis_defined_in_hooks = event.method in all_events\n\n\t\tif not (is_defined_in_hooks or is_server_script):\n\t\t\tfrappe.delete_doc(\"Scheduled Job Type\", event.name)\n", "path": "frappe/core/doctype/scheduled_job_type/scheduled_job_type.py"}]}