Add utm tags to announcement links

src/octoprint/plugins/announcements/__init__.py

@@ -22,7 +22,9 @@ from collections import OrderedDict
 
 from octoprint.server import admin_permission
 from octoprint.server.util.flask import restricted_access, with_revalidation_checking, check_etag
+from octoprint.util import utmify
 from flask.ext.babel import gettext
+from octoprint import __version__ as OCTOPRINT_VERSION
 
 class AnnouncementPlugin(octoprint.plugin.AssetPlugin,
                          octoprint.plugin.SettingsPlugin,
@@ -173,6 +175,7 @@ class AnnouncementPlugin(octoprint.plugin.AssetPlugin,
 			hash = hashlib.sha1()
 			hash.update(repr(sorted(enabled)))
 			hash.update(repr(sorted(forced)))
+			hash.update(OCTOPRINT_VERSION)
 
 			for channel in sorted(channel_configs.keys()):
 				hash.update(repr(channel_configs[channel]))
@@ -386,7 +389,7 @@ class AnnouncementPlugin(octoprint.plugin.AssetPlugin,
 		            summary=_lazy_images(entry["summary"]),
 		            summary_without_images=_strip_images(entry["summary"]),
 		            published=published,
-		            link=entry["link"],
+		            link=utmify(entry["link"], source="octoprint", medium="announcements", content=OCTOPRINT_VERSION),
 		            read=read)
 
 	def _get_channel_cache_path(self, key):

src/octoprint/util/__init__.py

@@ -875,14 +875,14 @@ except ImportError:
 	# no glob.escape - we need to implement our own
 	_glob_escape_check = re.compile("([*?[])")
 	_glob_escape_check_bytes = re.compile(b"([*?[])")
-	
+
 	def glob_escape(pathname):
 		"""
 		Ported from Python 3.4
-		
+
 		See https://github.com/python/cpython/commit/fd32fffa5ada8b8be8a65bd51b001d989f99a3d3
 		"""
-		
+
 		drive, pathname = os.path.splitdrive(pathname)
 		if isinstance(pathname, bytes):
 			pathname = _glob_escape_check_bytes.sub(br"[\1]", pathname)
@@ -900,6 +900,40 @@ except RuntimeError:
 	monotonic_time = time.time
 
 
+def utmify(link, source=None, medium=None, name=None, term=None, content=None):
+	if source is None:
+		return link
+
+	from collections import OrderedDict
+	try:
+		import urlparse
+		from urllib import urlencode
+	except ImportError:
+		# python 3
+		import urllib.parse as urlparse
+		from urllib.parse import urlencode
+
+	# inspired by https://stackoverflow.com/a/2506477
+	parts = list(urlparse.urlparse(link))
+
+	# parts[4] is the url query
+	query = OrderedDict(urlparse.parse_qs(parts[4]))
+
+	query["utm_source"] = source
+	if medium is not None:
+		query["utm_medium"] = medium
+	if name is not None:
+		query["utm_name"] = name
+	if term is not None:
+		query["utm_term"] = term
+	if content is not None:
+		query["utm_content"] = content
+
+	parts[4] = urlencode(query, doseq=True)
+
+	return urlparse.urlunparse(parts)
+
+
 class RepeatedTimer(threading.Thread):
 	"""
 	This class represents an action that should be run repeatedly in an interval. It is similar to python's

tests/util/test_misc.py

@@ -6,9 +6,11 @@ __copyright__ = "Copyright (C) 2017 The OctoPrint Project - Released under terms
 
 
 import unittest
+import ddt
 
 import octoprint.util
 
+@ddt.ddt
 class MiscTestCase(unittest.TestCase):
 
 	def test_get_class(self):
@@ -29,3 +31,17 @@ class MiscTestCase(unittest.TestCase):
 		except ImportError:
 			# success
 			pass
+
+	@ddt.data(
+		("http://example.com", dict(source="source"), "http://example.com?utm_source=source"),
+		("http://example.com?q=1", dict(source="source"), "http://example.com?q=1&utm_source=source"),
+		("http://example.com", dict(source="source", medium="medium"), "http://example.com?utm_source=source&utm_medium=medium"),
+		("http://example.com", dict(source="source", medium="medium", content="content with spaces"), "http://example.com?utm_source=source&utm_medium=medium&utm_content=content+with+spaces"),
+
+		# no handling
+		("http://example.com", dict(), "http://example.com"),
+	)
+	@ddt.unpack
+	def test_utmify(self, link, kwargs, expected):
+		actual = octoprint.util.utmify(link, **kwargs)
+		self.assertEqual(actual, expected)