''' Recommendation Engine Example 1 Nov 2017, 0x2620 ''' from collections import defaultdict import json import logging import os import random import time import ox from utils import run_async logger = logging.getLogger(__name__) class Engine: _pandora = None def __init__(self, path, **kwargs): self.path = path self.pandora_args = dict( url=kwargs.get('pandora', 'http://pandora.dmp/api/'), username=kwargs.get('username', 'dd.re'), password=kwargs.get('password', 'dd.re') ) filename = os.path.join(self.path, 'playlists.json') if os.path.exists(filename): with open(filename) as f: self.playlists = json.load(f) else: self.playlists = [] filename = os.path.join(self.path, 'state.json') if os.path.exists(filename): with open(filename) as f: self.state = json.load(f) else: self.state = { 'channels': { 'keywords': {'locked': False, 'value': 7}, 'screenings': {'locked': False, 'value': 7}, 'random': {'locked': True, 'value': 2} }, 'keywords': {}, } self.update_keywords() @property def pandora(self): while not self._pandora: try: self._pandora = Pandora(**self.pandora_args) except: logger.error('failed to connect to pandora, retry in 10 seconds') time.sleep(10) return self._pandora def _patch_clips(self, clips): inpoints = {} for index, clip in enumerate(clips): video_id = clip['id'].split('/')[0] inpoints[video_id] = inpoints.get(video_id, []) + [{ 'index': index, 'position': clip['in'] }] for video_id in inpoints: for i, inpoint in enumerate(sorted( inpoints[video_id], key=lambda inpoint: inpoint['position'] )): if i < len(inpoints[video_id]) - 1: clips[inpoint['index']]['out'] = inpoints[video_id][i + 1]['position'] else: clips[inpoint['index']]['out'] = self.pandora.get(video_id, ['duration'])['duration'] return clips def get_videos(self, user): channels = {'keywords': 7, 'screenings': 7, 'random': 2} sliders = {'dau': -1, 'physics': 0, 'sex': 1} # For each playlist, compute keyword score score = {} for playlist in self.playlists: score[playlist['name']] = random.random() for tag in [tag for tag in playlist['tags'] if tag in sliders]: score[playlist['name']] += sliders[tag] # Select highest scoring playlists playlists = sorted( self.playlists, key=lambda playlist: -score[playlist['name']] ) videos = playlists[:channels['keywords']] playlists = playlists[channels['keywords']:] # Count tags for the user count = defaultdict(lambda: 0) for event in user.get('events', []): if event.get('data', {}).get('product'): count[event['data']['product']] += 1 # For each tag in playlist, increment score by count for playlist in playlists: score[playlist['name']] = random.random() for tag in [tag for tag in playlist['tags'] if tag not in sliders]: score[playlist['name']] += count[tag] # Select highest scoring playlists videos += sorted( playlists, key=lambda playlist: -score[playlist['name']] )[:16 - channels['keywords']] # Shuffle playlists (randomize layout) and shift clips (randomize start) random.shuffle(videos) return [{ 'clips': video['clips'], 'position': random.randrange(len(video['clips'])), 'name': video['name'] } for video in videos] def update_state(self, data): for key in data: if key in self.state: self.state[key].update(data[key]) else: self.state[key] = data[key] self.save_state() return self.state def save_state(self): filename = os.path.join(self.path, 'state.json') with open(filename, 'w') as f: json.dump(self.state, f, indent=4, ensure_ascii=False, sort_keys=True) def update(self): # Get all storylines with tags storylines = [{ 'id': entity['id'], 'name': entity['name'], 'nodename': entity['nodename'], 'tags': entity['tags'] } for entity in self.pandora.find_entities({ 'conditions': [ {'key': 'type', 'operator': '==', 'value': 'storylines'}, ], 'operator': '&' }, ['id', 'name', 'tags', 'nodename']) if entity.get('tags', [])] # Get list of storyline names names = list(set([storyline['name'] for storyline in storylines])) # Get all clips annotated with storyline references clips = [clip for clip in self.pandora.find_annotations({ 'conditions': [ {'key': 'layer', 'operator': '==', 'value': 'storylines'} ], 'operator': '&' }, ['id', 'in', 'out', 'value']) if clip['value'] in names] # Get list of ids for videos with clips ids = list(set([clip['id'].split('/')[0] for clip in clips])) # Get and cache video data filename = os.path.join(self.path, 'videos.json') if os.path.exists(filename): with open(filename) as f: videos_ = json.loads(f.read()) ids_ = [video['id'] for video in videos_] else: videos_, ids_ = [], [] videos = sorted(videos_ + [ self.pandora.get(id, ['code', 'id', 'order', 'title']) for id in ids if not id in ids_ ], key=lambda video: video['order']) with open(filename, 'w') as f: f.write(json.dumps(videos, indent=4, sort_keys=True)) # Get video order order = {video['id']: video['order'] for video in videos} # Sort clips clips = sorted( clips, key=lambda clip: (order[clip['id'].split('/')[0]], clip['in']) ) # Get and cache playlists self.playlists = [playlist for playlist in [{ 'id': storyline['id'], 'name': storyline['nodename'].strip(), 'tags': storyline['tags'], 'clips': [{ 'item': clip['id'].split('/')[0], 'id': clip['id'], 'in': clip['in'], 'out': clip['out'] } for clip in clips if clip['value'] == storyline['name']] } for storyline in storylines] if playlist['clips']] with open(os.path.join(self.path, 'playlists.json'), 'w') as f: f.write(json.dumps(self.playlists, indent=4, sort_keys=True)) self.update_keywords() def update_keywords(self): changed = False if 'keywords' not in self.state: self.state['keywords'] = {} changed = True for playlist in self.playlists: for tag in playlist.get('tags', []): if tag.islower() and tag not in self.state['keywords']: self.state['keywords'][tag] = {'value': 0} changed = True if changed: self.save_state() @run_async def update_async(self): self.update() class Pandora: # pan.do/ra API wrapper def __init__(self, url, username, password): self.api = ox.API(url) self.api.signin(username=username, password=password) def find_annotations(self, query, keys): # print('FIND ANNOTATIONS', query, keys) return self.api.findAnnotations({ 'keys': keys, 'query': query, 'range': [0, 1000000] })['data']['items'] def find_entities(self, query, keys): # print('FIND ENTITIES', query, keys) return self.api.findEntities({ 'keys': keys, 'query': query, 'range': [0, 1000000] })['data']['items'] def get(self, id, keys): # print('GET', id, keys) return self.api.get({ 'id': id, 'keys': keys })['data'] if __name__ == '__main__': engine = Engine('json') engine.update()