Over the last few years of using RCS Zetta , I’ve written a few random scripts that solve little problems here and there. I thought I’d share a few of them with you today.
For many of these scripts, you will need Python 2.7, PyODBC , and a configured ODBC source pointing to your primary Zetta SQL Database.
These are unofficial scripts, and not supported in any way by myself or RCS.
Log Import Email Alerts
Zetta has a good activity logger, but doesn’t provide much in the way of realtime alerting. This script sends a summary email whenever a log is imported to any station in your Zetta system. This can be particularly useful if you want to find out when a traffic-reload happens.
"""
Zetta Log Import Emailer
This script will email you whenever a new log is imported into Zetta
Written by Anthony Eden (https://mediarealm.com.au/)
"""
# You need to download this module: https://pypi.python.org/pypi/pypyodbc
# Email config options
EMAIL_TO = ""
EMAIL_FROM = ""
EMAIL_SERVER = ""
# Zetta database ODBC connection name:
SQLConnectionString = 'DSN=ZettaDb;'
import time
import datetime
import pypyodbc
# Email libraries
import smtplib
from email.mime.text import MIMEText
# Setup Zetta SQL DB Connection:
dbConn = pypyodbc.connect(SQLConnectionString)
cur = dbConn.cursor()
def listImportTimes():
eventsQuery = '''
SELECT
DISTINCT InfoCreatedTime,
Station.StationID,
UserName,
CpuDescription,
xLogger_ActivityLog.TrackerAccountID,
Station.Name AS StationName
FROM [ZettaDB].[dbo].[xLogger_ActivityLog],
[ZettaDB].[dbo].[xLogger_TrackerAccount],
[ZettaDB].[dbo].[Station]
WHERE ActivityLocationID = 103
AND xLogger_ActivityLog.TrackerAccountID = xLogger_TrackerAccount.TrackerAccountId
AND xLogger_ActivityLog.StationID = Station.StationID
AND InfoCreatedTime >= DATEADD(MINUTE, -10, GETUTCDATE())
ORDER BY InfoCreatedTime DESC
'''
cur.execute(eventsQuery)
events = []
for d in cur:
if d[2] == "":
UserName = "SYSTEM USER"
else:
UserName = d[2]
events.append({
"InfoCreatedTime": d[0],
"StationID": d[1],
"StationName": d[5],
"UserName": UserName,
"CpuDescription": d[3],
"TrackerAccountID": d[4]
})
return events
def listImportDate(eventTime, accountId):
assetTypesQuery = '''
SELECT TOP 1 Changes
FROM [ZettaDB].[dbo].[xLogger_ActivityLog]
WHERE ActivityLocationID = 103
AND InfoCreatedTime = ?
AND TrackerAccountID = ?
'''
cur.execute(assetTypesQuery, [eventTime, int(accountId)])
for d in cur:
dateString = d[0].split(" Hour ")
return dateString[0].split("log ")[1]
def listImportAssetTypes(eventTime, accountId):
assetTypesQuery = '''
SELECT Changes
FROM [ZettaDB].[dbo].[xLogger_ActivityLog]
WHERE ActivityLocationID = 103
AND InfoCreatedTime = ?
AND TrackerAccountID = ?
'''
cur.execute(assetTypesQuery, [eventTime, int(accountId)])
# Store a count of every asset type found in the new log
assetTypes = {
"Other/Unknown": 0
}
assetTypesMatch = {
"Song": "Play Song: ",
"Link": "Play Link: ",
"Spot": "Play Spot: ",
"Exact Time Marker": "Exact Time Marker: ",
"Unresolved Asset": "Unresolved Asset: ",
"Top of Hour": "Top of Hour: ",
"Spot Block": "Spot Block: ",
"Macro": "Macro: ",
"Empty Voice Track Slot": "Empty Voice Track Slot: ",
"Comment": "Comment: ",
}
for assetType in assetTypesMatch:
assetTypes[assetType] = 0
for d in cur:
foundMatch = False
for assetType in assetTypesMatch:
if assetTypesMatch[assetType] in d[0]:
assetTypes[assetType] += 1
foundMatch = True
break
if foundMatch is False:
print d[0]
assetTypes['Other/Unknown'] += 1
return assetTypes
def sendEmail(to, subject, body):
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = EMAIL_FROM
msg['To'] = to
s = smtplib.SMTP(EMAIL_SERVER)
s.sendmail(EMAIL_FROM, [to], msg.as_string())
s.quit()
if __name__ == "__main__":
notifiedImportTimes = []
while True:
try:
for event in listImportTimes():
if event['InfoCreatedTime'] not in notifiedImportTimes:
notifiedImportTimes.append(event['InfoCreatedTime'])
spotsOnly = True
emailSubject = "Log Imported: " + event['StationName']
emailBody = "A log was imported by " + event['UserName']
emailBody += " to station " + event['StationName']
emailBody += " on computer " + event['CpuDescription'] + "\r\n\r\n"
emailBody += "Log Date: " + listImportDate(event['InfoCreatedTime'], event['TrackerAccountID']) + "\r\n\r\n"
emailBody += "Here's a breakdown of what was imported:\r\n"
assetBreakdown = listImportAssetTypes(event['InfoCreatedTime'], event['TrackerAccountID'])
for assetType in assetBreakdown:
if assetType != "Spot" and assetBreakdown[assetType] > 0:
spotsOnly = False
emailBody += assetType + ": " + str(assetBreakdown[assetType]) + "\r\n"
emailBody += "\r\nImport Time: " + str(event['InfoCreatedTime']) + "\r\n"
if spotsOnly is True:
emailSubject += " (Spots Only)"
print emailSubject
print emailBody
sendEmail(EMAIL_TO, emailSubject, emailBody)
#print event
print
time.sleep(20)
except Exception, e:
print "Exception!!!", e
time.sleep(120)
Add Timestamps to Background Recorder Asset Titles
Zetta allows you to record audio files in the background and override assets – useful for capturing satellite and codec feeds. However, there’s no quick way for on-air talent to see when an asset was last updated. This script checks all background-recorded files in the system, and appends a timestamp to the filename.
"""
Zetta Background Recorder Timestamper
This script appends the last-modified timestamp string to assets that are updated by the background recorder.
Written by Anthony Eden (https://mediarealm.com.au)
"""
import pypyodbc
import time
import datetime
# Setup Zetta SQL DB Connection:
SQLConnectionString = 'DSN=ZettaDB;'
dbConn = pypyodbc.connect(SQLConnectionString)
cur = dbConn.cursor()
def utc2local(utc):
# From https://stackoverflow.com/a/19238551
epoch = time.mktime(utc.timetuple())
offset = datetime.datetime.fromtimestamp(epoch) - datetime.datetime.utcfromtimestamp(epoch)
return utc + offset
def listBackgroundRecorderAssets():
assetsQuery = '''
SELECT
RecordEvent.AssetID,
Asset.infoModifiedDate,
Asset.Title,
[Resource].infoModifiedDate
FROM
[RecordEvent],
[Asset],
[AssetToResource],
[Resource]
WHERE RefId IS NULL
AND RecordEvent.AssetID IS NOT NULL
AND RecordEvent.AssetID = Asset.AssetID
AND AssetToResource.AssetID = Asset.AssetID
AND AssetToResource.IsPrimaryResource = 1
AND AssetToResource.ResourceID = [Resource].ResourceID
'''
cur.execute(assetsQuery)
assets = []
rows = cur.fetchall()
for d in rows:
localTime = utc2local(d[3])
assets.append({
"AssetID": d[0],
"Title": d[2],
"LastModified": d[3],
"LastModifiedLocal": localTime,
"AppendDateString": "[" + localTime.strftime("%d/%m/%Y %I:%M:%S %p") + "]"
})
return assets
def updateAssetTitle(assetId, title):
# Sets a new title for an asset
assetQuery = ''' UPDATE TOP(1) Asset SET Title = ? WHERE AssetID = ? '''
cur.execute(assetQuery, (title, assetId))
dbConn.commit()
if __name__ == "__main__":
while True:
try:
# Loop over all background recorder assets
for x in listBackgroundRecorderAssets():
# Does this asset already have square braces in the title?
if "[" in x['Title'] and "]" in x['Title'][x['Title'].index("["):]:
# Replace existing square braces with datetime stamp
newTitle = x['Title'][:x['Title'].index("[")] + x['AppendDateString'] + x['Title'][x['Title'].index("]", x['Title'].index("[")) + 1:]
if newTitle != x['Title']:
print "Update title", newTitle, "for Asset ID", x['AssetID']
updateAssetTitle(x['AssetID'], newTitle)
else:
# Append the datetime stamp to the existing title
newTitle = x['Title'] + " " + x['AppendDateString']
print "New title", newTitle, "for Asset ID", x['AssetID']
updateAssetTitle(x['AssetID'], newTitle)
time.sleep(20)
except Exception, e:
print "Exception!!!", e
time.sleep(120)
Batch Export Assets Based On External ID
Let’s say you need to export a lot of assets from Zetta, and you have a list of their External IDs (perhaps you got this list from an external music scheduling program, or a reconciliation file). How can you export all these files to a folder without locating the assets manually in the Library?
Simply paste your list of External IDs into the CONFIG_AssetList variable in this script, specify the source and destination directories, and run it.
"""
Zetta Batch Audio Exporter
Given a list of External IDs, find each asset's file in the Zetta Content Store, and copy it to a separate folder.
Written by Anthony Eden (https://mediarealm.com.au)
"""
from shutil import copyfile
import os
import pypyodbc
import re
# The Zetta Content Store location:
CONFIG_StorageLocation = "\\\\zettaaudio1\\ZAUDIO\\"
# The output location:
CONFIG_OutputStorage = "C:\\SCRIPTS\\ZettaBatchExport\\"
# Paste a list of External IDs here (one per line)
CONFIG_AssetList = """
12345
67890
"""
# Setup Zetta SQL DB Connection:
SQLConnectionString = 'DSN=ZettaDB;'
dbConn = pypyodbc.connect(SQLConnectionString)
cur = dbConn.cursor()
def findAssetByExternalID(ExternalID):
assetQuery = '''
SELECT
Asset.ThirdPartyID,
Asset.Title,
Resource.StorageFile,
AutoGen_AssetReadonlyReferences.ArtistID_1,
Resource.ResourceID
FROM
Asset,
Resource,
AssetToResource,
AutoGen_AssetReadonlyReferences
WHERE
Asset.AssetID = AssetToResource.AssetID
AND AssetToResource.isPrimaryResource = 1
AND AssetToResource.ResourceID = Resource.ResourceID
AND Asset.AssetID = AutoGen_AssetReadonlyReferences.AssetID
AND ThirdPartyID = ?
'''
cur.execute(assetQuery, [str(ExternalID)])
assetList = []
for d in cur:
assetList.append({
"ExternalID": str(d[0]),
"Filename": d[2],
"Title": str(d[1]),
"Artist": getArtistName(d[3]),
"FileFolder": getStorageFolderFromFilename(d[4]),
"FileExtension": getExtensionFromFilename(d[2]),
"ResourceID": str(d[4])
})
return assetList
def getArtistName(artistId):
# Take the Artist ID and return the Artist's name
try:
artistId = int(artistId)
except:
return None
artistNameQuery = '''
SELECT Artist.Name
FROM Artist
WHERE ArtistID = ?'''
cur.execute(artistNameQuery, [artistId])
for d in cur:
return d[0]
return ""
def getStorageFolderFromFilename(filename):
# Zetta stores 1000 items in each folder (e.g. from 1000 to 1999)
# This is based on the ResourceID
folder = str(filename).split("~")[0][:-3]
if folder == "":
folder = 0
return int(folder)
def getExtensionFromFilename(fullname):
# Returns the file's extension name
filename, file_extension = os.path.splitext(fullname)
return file_extension
if __name__ == "__main__":
IDs = CONFIG_AssetList.split()
for thisExternalID in IDs:
# Loop over every specified External ID
asset = findAssetByExternalID(thisExternalID)[0]
assetTitle = re.sub(r'\W+', ' ', asset['Title']).replace(" ", " ").strip()
assetArtist = re.sub(r'\W+', ' ', asset['Artist']).replace(" ", " ").strip()
srcfile = CONFIG_StorageLocation + str(asset['FileFolder']) + "\\" + asset['Filename']
dstfile = CONFIG_OutputStorage + assetArtist + " - " + assetTitle + asset['FileExtension']
if os.path.isfile(dstfile):
print "DST FILE ALREADY EXISTS: ", dstfile
elif os.path.isfile(srcfile):
# Do the copy
print asset['ExternalID'], "||", dstfile
copyfile(srcfile, dstfile)
else:
print "SRC FILE DOESN'T EXIST: ", srcfile
ZettaView: HTML Status Feed Viewer
Update August 2018 : ZettaView is now available for download. Check it out on GitHub .
ZettaView is a simple Zetta Status Feed client , which allows you to view the current and upcoming assets for a specific station in a web browser. This can be useful for outside broadcasts, TV displays around the building, or simple remote troubleshooting.
ZettaView is a Python script, with an in-built Flask web server. For convenience, I’ve bundled this script as an EXE.
Simply download the EXE , rename ‘config-sample.json’ to ‘config.json’, put you Zetta Status Feed Server’s address into the config file, run the EXE, and then browse to http://127.0.0.1:4444.
Please note – this is a simple script, which shouldn’t be exposed to the public internet. Please do not port forward to this script.
