# -*- coding: utf-8 -*- """Build the Pro tebe snapshot (docs/plan/pro-tebe/010 §1). signals -> seed details -> taste profile -> candidates from TMDb -> light score -> details of the best -> full score, match %, reason -> rows per chip mode -> snapshot in `meta` (key `foryou:snapshot`). Runs in the service (P3) or the harness; widgets only read the snapshot. """ import time from datetime import date, datetime, timedelta from wcs.foryou import api, compose, features as F, score as S, signals from wcs.skin import userfirst SNAPSHOT_KEY = 'foryou:snapshot' SNAPSHOT_TTL = 30 * 86400 VERSION = 1 SEEDS_POS = 14 SEEDS_NEG = 8 DETAILED = 200 # candidates that get a detail call (cached 14 days) TOP_DETAILED = 30 KEYWORDS = 3 # strongest taste keywords used for discover MAKERS = 4 SAGA_RANK = 3 # the next part of a watched saga counts like the 4th TMDb recommendation PER_KIND = 40 # details reserved per special row kind and media SANE_LANGS = S.HOME_LANGS | {'fr', 'de', 'es', 'it', 'da', 'sv', 'no', 'pl', 'nl', 'is', 'fi'} # movie genre id -> tv genre id for discover/tv (films-only genres have no series twin) GEM_VOTES = {'movie': 6000, 'tv': 1500} # a hidden gem has fewer votes than this TV_DISCOVER = {28: 10759, 12: 10759, 878: 10765, 14: 10765, 10752: 10768, 35: 35, 18: 18, 80: 80, 9648: 9648, 99: 99, 16: 16, 37: 37, 10751: 10751} def _entry(pool, media, data): if not data.get('id') or media not in ('movie', 'tv'): return None key = F.key(media, data['id']) entry = pool.get(key) if entry is None: entry = pool[key] = {'card': F.card(data, media), 'features': F.features(data, media), 'sources': {}, 'kinds': set(), 'names': {}} return entry def _media_of(item, default): return item.get('media_type') or default def _add_list(pool, items, media, kind, source_key=None): for rank, item in enumerate(items): m = _media_of(item, media) entry = _entry(pool, m, item) if entry is None: continue entry['kinds'].add(kind) if source_key: src = (kind, source_key) entry['sources'][src] = min(rank, entry['sources'].get(src, rank)) def _taste_calls(pos): """Discover by the profile's strongest keywords ('sitcom', 'artificial intelligence') - TMDb recommendations miss many obvious neighbours that the keywords find.""" keywords = [n[2:] for n, _v in sorted(((n, v) for n, v in pos.items() if n.startswith('k:')), key=lambda kv: -kv[1])[:KEYWORDS]] calls = [] for kw in keywords: calls.append(('movie', 'keyword', 'discover/movie', {'with_keywords': kw, 'sort_by': 'popularity.desc', 'vote_count.gte': 300})) calls.append(('tv', 'keyword', 'discover/tv', {'with_keywords': kw, 'sort_by': 'popularity.desc', 'vote_count.gte': 150})) return calls def _maker_titles(pos, seed_features): """Other works of the household's series creators (Chuck Lorre -> Malý Sheldon, Graham Linehan -> Otec Ted); a film director only when two watched titles share him (an auteur, not "the director of HP 1").""" count, tv_makers = {}, set() for seed, feats in seed_features: if seed['weight'] <= 0: continue for name in feats: if name.startswith('d:'): count[name] = count.get(name, 0) + 1 if seed['media'] == 'tv': tv_makers.add(name) eligible = {n: v for n, v in pos.items() if n.startswith('d:') and (n in tv_makers or count.get(n, 0) >= 2)} makers = [n[2:] for n, _v in sorted(eligible.items(), key=lambda kv: -kv[1])[:MAKERS]] answers = api.parallel([lambda m=m: api.get('person/{}/combined_credits'.format(m), None, api.TTL_DETAIL) for m in makers]) out = [] for maker, data in zip(makers, answers): works = [w for w in (data or {}).get('crew') or [] if w.get('job') in ('Director', 'Creator') and w.get('media_type') in ('movie', 'tv') and int(w.get('vote_count') or 0) >= 50] works.sort(key=lambda w: -float(w.get('popularity') or 0)) out.append((maker, works[:15])) return out def _discover_calls(top_genres, today): """(media, kind, path, params) of the discover queries that fill the pool around the taste.""" calls = [] for gid in top_genres[:3]: calls.append(('movie', 'genre', 'discover/movie', {'with_genres': gid, 'sort_by': 'popularity.desc', 'vote_count.gte': 400, 'vote_average.gte': 6.6})) if gid in TV_DISCOVER: calls.append(('tv', 'genre', 'discover/tv', {'with_genres': TV_DISCOVER[gid], 'sort_by': 'popularity.desc', 'vote_count.gte': 150, 'vote_average.gte': 6.8})) any_genre = '|'.join(str(g) for g in top_genres[:4]) tv_genres = '|'.join(sorted({str(TV_DISCOVER[g]) for g in top_genres[:4] if g in TV_DISCOVER})) if not top_genres: # cold start: the same rows over all genres calls = [(m, 'cold', p, q) for m, p, q in COLD_CALLS] niche = ','.join(str(g) for g in sorted(F.NICHE_GENRES)) # gems are for grown-ups, in the household's genres for page in (1, 2, 3, 4, 5): # many: a lesser known title often has no Czech version on Webshare calls.append(('movie', 'gem', 'discover/movie', {'with_genres': any_genre, 'without_genres': niche, 'sort_by': 'vote_average.desc', 'vote_average.gte': 7.0, 'vote_count.gte': 300, 'vote_count.lte': GEM_VOTES['movie'], 'primary_release_date.gte': '1975-01-01', 'page': page})) for page in (1, 2, 3): calls.append(('tv', 'gem', 'discover/tv', {'with_genres': tv_genres, 'without_genres': niche, 'sort_by': 'vote_average.desc', 'vote_average.gte': 7.4, 'vote_count.gte': 100, 'vote_count.lte': GEM_VOTES['tv'], 'page': page})) since = (today - timedelta(days=300)).isoformat() until = (today - timedelta(days=20)).isoformat() for page in (1, 2): calls.append(('movie', 'new', 'discover/movie', {'primary_release_date.gte': since, 'primary_release_date.lte': until, 'sort_by': 'popularity.desc', 'vote_count.gte': 60, 'page': page})) for page in (1, 2): calls.append(('tv', 'new', 'discover/tv', {'first_air_date.gte': (today - timedelta(days=300)).isoformat(), 'first_air_date.lte': until, 'sort_by': 'popularity.desc', 'vote_count.gte': 20, 'page': page})) for _media, _kind, _path, params in calls: if params.get('with_genres') == '': del params['with_genres'] return calls TOP_CALLS = (('movie', 'trending/movie/week', {}), ('tv', 'trending/tv/week', {}), ('movie', 'discover/movie', {'watch_region': 'CZ', 'with_watch_monetization_types': 'flatrate', 'sort_by': 'popularity.desc'}), ('tv', 'trending/tv/day', {})) COLD_CALLS = (('movie', 'movie/top_rated', {}), ('tv', 'tv/top_rated', {})) def _top_order(lists): """Top 10 v ČR: blend of TMDb trending (global) and popularity on CZ streaming, by reciprocal rank.""" points = {} for media, items in lists: for rank, item in enumerate(items): key = F.key(_media_of(item, media), item['id']) points[key] = points.get(key, 0.0) + 1.0 / (rank + 4) return [k for k, _p in sorted(points.items(), key=lambda kv: -kv[1])] def _sane(card, pos): """Top 10 is not personal, but it must look like a Czech household's chart (no foreign anime / talk shows).""" taste_langs = {n[5:] for n in pos if n.startswith('lang:')} if set(card['genres']) & F.BLOCKED_TV_GENRES or card['adult']: return False if card['language'] and card['language'] not in SANE_LANGS and card['language'] not in taste_langs: return False if card['votes'] >= 30 and card['rating'] < 6.0: return False return card['votes'] >= 20 and card['year'] >= date.today().year - 3 # a chart of now, not 1999 def _gem_ok(card, pos): """A gem must be watchable for this household: its languages and no niche genre outside the taste.""" taste = S.taste_of(pos) if card['language'] not in S.HOME_LANGS and card['language'] not in taste['langs']: return False return not any(taste['share'].get(g, 0.0) < 0.35 for g in set(card['genres']) & F.NICHE_GENRES) def _released(card, today): return bool(card['date']) and card['date'] <= today.isoformat() MIN_FILM_MINUTES = 60 def _collapse(pool, seed_features, exclude, today, seed_collections=None): """One title per collection: its first part not seen yet (Harry Potter 2, not all seven, even when part 2 did not make it into the pool); a collection whose earlier part waits unwatched in the list is left out. Films under an hour (shorts, TV specials) go too.""" waiting = {name for seed, feats in seed_features if seed['kind'] == 'list' for name in feats if name.startswith('col:')} groups = {} out = {} for key, entry in pool.items(): card = entry['card'] if card['media'] == 'movie' and card['runtime'] and card['runtime'] < MIN_FILM_MINUTES: continue if not card['collection']: out[key] = entry elif 'col:{}'.format(card['collection']) not in waiting: groups.setdefault(card['collection'], []).append(key) for collection in seed_collections or {}: if 'col:{}'.format(collection) not in waiting: groups.setdefault(collection, []) parts = api.parallel([lambda c=c: api.get('collection/{}'.format(c), None, api.TTL_DETAIL) for c in groups]) for (collection, members), data in zip(groups.items(), parts): order = sorted((p for p in (data or {}).get('parts') or [] if p.get('id')), key=lambda p: p.get('release_date') or '9999') released = [p for p in order if p.get('release_date') and p['release_date'] <= today.isoformat()] seed = (seed_collections or {}).get(collection) watched_at = [i for i, p in enumerate(released) if seed and F.key('movie', p['id']) == seed] start = watched_at[-1] + 1 if watched_at else 0 # after Terminátor 2 comes 3, not the 1984 original nxt = next((F.key('movie', p['id']) for p in released[start:] + released[:start] if ('movie', int(p['id'])) not in exclude), None) if nxt is None: if not members: continue nxt = min(members, key=lambda k: pool[k]['card']['date']) if nxt not in pool: detail = api.detail('movie', nxt.split(':')[1]) if not detail or not detail.get('poster_path') or not detail.get('backdrop_path'): continue pool[nxt] = {'card': F.card(detail, 'movie'), 'features': F.features(detail, 'movie'), 'sources': {}, 'kinds': set(), 'names': F.names(detail, 'movie')} entry = pool[nxt] for key in members: # the next part inherits why its siblings were in the pool (not that for src, rank in pool[key]['sources'].items(): # they are new or trending: those are their own) entry['sources'][src] = min(rank, entry['sources'].get(src, rank)) entry['kinds'] |= pool[key]['kinds'] - {'new', 'top', 'trend', 'gem'} if seed: entry['sources'][('rec', seed)] = min(SAGA_RANK, entry['sources'].get(('rec', seed), SAGA_RANK)) entry['kinds'].add('saga') out[nxt] = entry return out WEAK_REASONS = ('genre', 'quality', 'trending') WEAK_MATCH_CAP = 85 LANG_FACTOR = {'cz_dub': 1.0, 'cz_sub': 0.92, 'sk_dub': 0.85} # a Czech household: CZ dubbing first MIN_VERIFIED = 40 # fewer playable titles than this: the check failed somewhere, keep the old snapshot class NotEnough(Exception): """The Webshare check left too few titles; the caller keeps the previous snapshot.""" def _check_order(pool, top_order): """Titles in the order their availability matters: the chart, then by score, with each special kind's best ones early enough to fill their rows.""" ranked = sorted(pool, key=lambda k: -pool[k]['score']) first = [k for k in top_order if k in pool][:TOP_DETAILED] for kind in ('new', 'gem'): first += [k for k in ranked if kind in pool[k]['kinds']][:40] return list(dict.fromkeys(first + ranked)) LOOSE_LINK = 0.25 # a keyword / "similar" link to a seed needs at least this much in common with it def _prune_loose_links(pool, seed_features): """TMDb "similar" and the seed-keyword queries share a single tag at times (Čarodějky for Dva a půl chlapa); such a link to the seed is dropped unless the title has enough in common with the seed (its features, with the full detail now). TMDb recommendations are behaviour-based and stay.""" vectors = {F.key(s['media'], s['id']): feats for s, feats in seed_features if s['weight'] > 0} for entry in pool.values(): for source in [src for src in entry['sources'] if src[0] in ('sim', 'kw')]: seed = vectors.get(source[1]) if seed is None or S.similarity(seed, entry['features']) < LOOSE_LINK: del entry['sources'][source] def build(now=None, hour=None, today=None, log=None, verify=True): """Snapshot dict; does not store it (see `save`). `verify`: only titles playable from Webshare (P2).""" started = time.time() now = now or time.time() today = today or date.fromtimestamp(now) hour = datetime.fromtimestamp(now).hour if hour is None else hour log = log or (lambda msg: None) seed_list, exclude = signals.load(now) positive = [s for s in seed_list if s['weight'] > 0][:SEEDS_POS] negative = [s for s in seed_list if s['weight'] < 0][:SEEDS_NEG] used_seeds = positive + negative details = api.parallel([lambda s=s: api.detail(s['media'], s['id']) for s in used_seeds]) seed_features, seed_titles, seed_weight, seed_kind = [], {}, {}, {} seed_collections = {} # collection id -> seed key: the saga continues with its next unseen part seed_genre_ids = {} seed_audience = {} pool = {} userfirst.breathe() # background rebuild: the user first (wcs/skin/userfirst.py) for seed, data in zip(used_seeds, details): if not data: continue key = F.key(seed['media'], seed['id']) seed_features.append((seed, F.features(data, seed['media']))) seed_titles[key] = data.get('title') or data.get('name') or '' seed_weight[key] = seed['weight'] seed_kind[key] = seed['kind'] seed_genre_ids[key] = F.genre_ids(data) seed_audience[key] = F.audience(data, seed['media']) if seed['weight'] > 0: if (data.get('belongs_to_collection') or {}).get('id') and seed['kind'] in ('watched', 'like', 'pick'): seed_collections.setdefault(int(data['belongs_to_collection']['id']), key) _add_list(pool, (data.get('recommendations') or {}).get('results') or [], seed['media'], 'rec', key) primary = set(F.folded_genres(F.genre_ids(data))[:1]) similar = [i for i in (data.get('similar') or {}).get('results') or [] if primary & set(F.folded_genres(F.genre_ids(i)))] # TMDb 'similar' = shared keywords only _add_list(pool, similar, seed['media'], 'sim', key) pos, neg = S.profile(seed_features) top_genres = [int(n[2:]) for n in S.taste_summary(pos, 4)] log('seeds {} (+{} / -{}), taste genres {}'.format(len(seed_features), len(positive), len(negative), top_genres)) calls = [(m, 'trend', p, q) for m, p, q in TOP_CALLS] calls += _discover_calls(top_genres, today) + _taste_calls(pos) # more like each strong seed: its own keywords (Matrix -> dystopia, man vs machine -> Blade Runner, Počátek) for seed, data in zip(used_seeds, details): if data and seed['weight'] > 0 and seed in positive[:6]: kws = F.seed_keywords(data) if kws: calls.append((seed['media'], 'kwseed:' + F.key(seed['media'], seed['id']), 'discover/{}'.format(seed['media']), {'with_keywords': '|'.join(str(k) for k in kws), 'sort_by': 'vote_count.desc', 'vote_average.gte': 6.8, 'vote_count.gte': 300 if seed['media'] == 'movie' else 100})) # the second page of TMDb recommendations of the strongest seeds for seed in positive[:5]: calls.append((seed['media'], 'rec2:' + F.key(seed['media'], seed['id']), '{}/{}/recommendations'.format(seed['media'], seed['id']), {'page': 2})) answers = api.parallel([lambda c=c: api.results(c[2], c[3]) for c in calls]) top_lists = [] userfirst.breathe() # background rebuild: the user first (wcs/skin/userfirst.py) for (media, kind, _path, _params), items in zip(calls, answers): items = items or [] if kind.startswith('kwseed:'): seed_key = kind[7:] primary = F.folded_genres(seed_genre_ids.get(seed_key, []))[:1] for rank, item in enumerate(items): if not set(primary) & set(F.folded_genres(F.genre_ids(item))): continue # a shared keyword ('dream') does not make Počátek a film like V hlavě entry = _entry(pool, media, item) if entry is not None: entry['kinds'].add('kw') entry['sources'].setdefault(('kw', seed_key), 4 + rank) continue if kind.startswith('rec2:'): seed_key = kind[5:] for rank, item in enumerate(items): entry = _entry(pool, media, item) if entry is not None: entry['kinds'].add('rec') entry['sources'].setdefault(('rec', seed_key), 20 + rank) continue _add_list(pool, items, media, kind) if kind == 'trend': top_lists.append((media, items)) for _maker, works in _maker_titles(pos, seed_features): for work in works: entry = _entry(pool, work['media_type'], work) if entry is not None: entry['kinds'].add('maker') top_order = _top_order(top_lists) userfirst.breathe() # background rebuild: the user first (wcs/skin/userfirst.py) for key in top_order: if key in pool: pool[key]['kinds'].add('top') for key in list(pool): media, tmdb_id = key.split(':') card = pool[key]['card'] if (media, int(tmdb_id)) in exclude or card['adult'] or not card['poster'] or not card['backdrop'] \ or not _released(card, today): del pool[key] log('pool {} candidates from {} lists'.format(len(pool), len(calls))) # light score -> details of the best + the Top 10 candidates -> full score S.score_pool(pool, pos, neg, seed_weight, today.isoformat()) ranked = sorted(pool, key=lambda k: -pool[k]['score']) best = ranked[:DETAILED] tops = [k for k in top_order if k in pool][:TOP_DETAILED] # every special row gets its own share of details, per media, so no chip mode runs dry shares = [k for kind in ('new', 'gem', 'keyword', 'maker') for media in ('movie', 'tv') for k in [k for k in ranked if kind in pool[k]['kinds'] and pool[k]['card']['media'] == media][:PER_KIND]] wanted = list(dict.fromkeys(best + tops + shares)) full = api.parallel([lambda k=k: api.detail(*k.split(':')) for k in wanted]) detailed = {} userfirst.breathe() # background rebuild: the user first (wcs/skin/userfirst.py) for key, data in zip(wanted, full): if not data: continue media = key.split(':')[0] entry = pool[key] entry['card'] = F.card(data, media) entry['features'] = F.features(data, media) entry['names'] = F.names(data, media) detailed[key] = entry pool = _collapse(detailed, seed_features, exclude, today, seed_collections) _prune_loose_links(pool, seed_features) S.score_pool(pool, pos, neg, seed_weight, today.isoformat()) best_score = max([e['score'] for e in pool.values()] + [0]) feature_names = {} userfirst.breathe() # background rebuild: the user first (wcs/skin/userfirst.py) for entry in pool.values(): feature_names.update(entry['names']) personal = any(s['weight'] > 0 for s in used_seeds) seed_vectors = {F.key(s['media'], s['id']): feats for s, feats in seed_features if s['weight'] > 0} for key, entry in pool.items(): entry['match'] = S.match_percent(entry['score'], best_score) if personal else 0 # no taste, no "% shoda" entry['reason'] = S.reason(entry, seed_weight, seed_kind, seed_titles, pos, feature_names, seed_vectors) if entry['reason']['kind'] in WEAK_REASONS: # a genre or a rating alone is no 95 % match entry['match'] = min(entry['match'], WEAK_MATCH_CAP) entry['sane'] = _sane(entry['card'], pos) entry['gem_ok'] = _gem_ok(entry['card'], pos) slot = compose.slot_of(hour) share = S.taste_of(pos)['share'] avail, verified = {}, False if verify: from wcs.foryou import availability try: order = _check_order(pool, top_order) avail = availability.check_many([dict(pool[k]['card']) for k in order], today=today, log=log) verified = True except availability.Unavailable as exc: # no account / no login: show everything, unverified log('webshare check off: {}'.format(exc)) if verified: playable = {k for k, r in avail.items() if r.get('status') == 'ok' and k in pool} if len(playable) < MIN_VERIFIED: raise NotEnough('only {} playable titles'.format(len(playable))) def available(key): return key in playable for key in playable: # same taste, the Czech version first pool[key]['score'] *= LANG_FACTOR.get(avail[key].get('lang'), 1.0) def dubbed(key): return avail.get(key, {}).get('lang') == 'cz_dub' else: available = dubbed = None main_genres = {key: F.folded_genres(ids) for key, ids in seed_genre_ids.items()} modes = {mode: compose.rows(pool, used_seeds, top_order, slot, available, mode=mode, taste_share=share, dubbed=dubbed, seed_genres=main_genres, seed_audience=seed_audience) for mode in compose.MODES} shown = {k for rows in modes.values() for row in rows for k in row['items']} ordered = sorted(shown) logos = dict(zip(ordered, api.parallel([lambda k=k: api.logo(*k.split(':')) for k in ordered]))) items = {} userfirst.breathe() # background rebuild: the user first (wcs/skin/userfirst.py) for key in shown: entry = pool[key] item = dict(entry['card']) item.update({'match': entry['match'], 'reason': entry['reason'], 'kinds': sorted(entry['kinds']), 'score': round(entry['score'], 4), 'parts': entry['parts'], 'lang': avail.get(key, {}).get('lang', ''), 'quality': avail.get(key, {}).get('quality', ''), 'logo': logos.get(key) or ''}) items[key] = item snapshot = {'version': VERSION, 'built_at': int(now), 'personal': personal, 'verified': verified, 'day': today.isoformat(), 'slot': slot, 'seeds': [{'key': F.key(s['media'], s['id']), 'title': seed_titles.get(F.key(s['media'], s['id']), ''), 'weight': round(s['weight'], 3), 'kind': s['kind']} for s in used_seeds], 'seed_titles': seed_titles, 'genres': api.genre_names(), 'taste': top_genres, 'modes': modes, 'items': items, 'stats': {'pool': len(pool), 'seconds': round(time.time() - started, 1), 'tmdb': dict(api.STATS)}} return snapshot def save(snapshot): from wcs.library2 import meta meta.put(SNAPSHOT_KEY, snapshot, SNAPSHOT_TTL) def load(): from wcs.library2 import meta data, _stale = meta.get(SNAPSHOT_KEY) return data if isinstance(data, dict) and data.get('version') == VERSION else None