# -*- coding: utf-8 -*-
"""SQLite store of the addon library (docs/plan/milionar-library/030).

The only module that runs SQL. One connection per call (plugin and service
processes both use it), WAL journal, every write one transaction.
"""
import json
import os
import sqlite3
import time

SCHEMA_VERSION = 4
DB_NAME = 'library.db'
_override_path = None


def set_path(path):
    """Tests point the store at a temporary file."""
    global _override_path
    _override_path = path


def db_path():
    if _override_path:
        return _override_path
    import xbmcvfs
    return os.path.join(xbmcvfs.translatePath('special://profile/addon_data/plugin.video.milionar/'), DB_NAME)


SCHEMA_V1 = '''
CREATE TABLE titles(
  media_type TEXT NOT NULL CHECK(media_type IN ('movie','tv')),
  tmdb_id INTEGER NOT NULL,
  in_list INTEGER NOT NULL DEFAULT 0,
  list_added_at INTEGER,
  list_pos INTEGER,
  legacy_json TEXT,
  hidden_from_continue_at INTEGER,
  last_activity_at INTEGER,
  PRIMARY KEY(media_type, tmdb_id));
CREATE TABLE movie_state(
  tmdb_id INTEGER PRIMARY KEY,
  playcount INTEGER NOT NULL DEFAULT 0, last_played_at INTEGER,
  resume_pos REAL NOT NULL DEFAULT 0, resume_total REAL NOT NULL DEFAULT 0,
  last_ident TEXT, last_provider TEXT, last_file_name TEXT, mirrored_at INTEGER);
CREATE TABLE episode_state(
  show_id INTEGER NOT NULL, season INTEGER NOT NULL, episode INTEGER NOT NULL,
  playcount INTEGER NOT NULL DEFAULT 0, last_played_at INTEGER,
  resume_pos REAL NOT NULL DEFAULT 0, resume_total REAL NOT NULL DEFAULT 0,
  last_ident TEXT, last_provider TEXT, last_file_name TEXT, mirrored_at INTEGER,
  PRIMARY KEY(show_id, season, episode));
CREATE TABLE play_log(
  id INTEGER PRIMARY KEY, media_type TEXT, tmdb_id INTEGER, season INTEGER, episode INTEGER,
  started_at INTEGER, ended_at INTEGER, position REAL, total REAL, outcome TEXT);
CREATE TABLE meta(key TEXT PRIMARY KEY, json TEXT NOT NULL, fetched_at INTEGER NOT NULL, ttl INTEGER NOT NULL);
CREATE TABLE flags(key TEXT PRIMARY KEY, value TEXT);
'''


SCHEMA_V2 = '''
ALTER TABLE movie_state ADD COLUMN mirrored_playcount INTEGER;
ALTER TABLE movie_state ADD COLUMN mirrored_resume_pos REAL;
ALTER TABLE episode_state ADD COLUMN mirrored_playcount INTEGER;
ALTER TABLE episode_state ADD COLUMN mirrored_resume_pos REAL;
'''

SCHEMA_V3 = '''
ALTER TABLE movie_state ADD COLUMN state_version INTEGER NOT NULL DEFAULT 0;
ALTER TABLE episode_state ADD COLUMN state_version INTEGER NOT NULL DEFAULT 0;
CREATE TABLE mirror_pending(
  media_type TEXT NOT NULL, tmdb_id INTEGER NOT NULL,
  season INTEGER NOT NULL, episode INTEGER NOT NULL,
  version INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
  next_attempt REAL NOT NULL DEFAULT 0,
  PRIMARY KEY(media_type,tmdb_id,season,episode));
CREATE TABLE pending_marks(
  id INTEGER PRIMARY KEY, media_type TEXT NOT NULL, tmdb_id INTEGER NOT NULL,
  kind TEXT NOT NULL, season INTEGER, episode INTEGER, watched INTEGER NOT NULL,
  attempts INTEGER NOT NULL DEFAULT 0, next_attempt REAL NOT NULL DEFAULT 0);
'''

SCHEMA_V4 = '''
CREATE TABLE feedback(
  media_type TEXT NOT NULL CHECK(media_type IN ('movie','tv')),
  tmdb_id INTEGER NOT NULL,
  value TEXT NOT NULL CHECK(value IN ('like','dislike','hide','pick')),
  at INTEGER NOT NULL,
  PRIMARY KEY(media_type, tmdb_id));
'''


class _Connection(sqlite3.Connection):
    pass


def connect(path=None):
    path = path or db_path()
    folder = os.path.dirname(path)
    if folder:
        os.makedirs(folder, exist_ok=True)
    conn = sqlite3.connect(path, timeout=5.0, isolation_level=None, factory=_Connection)
    conn.widget_rows = set()
    conn.row_factory = sqlite3.Row
    conn.execute('PRAGMA busy_timeout=5000')
    conn.execute('PRAGMA journal_mode=WAL')
    conn.execute('PRAGMA foreign_keys=ON')
    _migrate(conn)
    return conn


def _migrate(conn):
    version = conn.execute('PRAGMA user_version').fetchone()[0]
    if version >= SCHEMA_VERSION:
        return
    conn.execute('BEGIN IMMEDIATE')
    try:
        version = conn.execute('PRAGMA user_version').fetchone()[0]  # another process may have migrated
        if version < 1:
            for statement in SCHEMA_V1.strip().split(';'):
                if statement.strip():
                    conn.execute(statement)
        if version < 2:
            for statement in SCHEMA_V2.strip().split(';'):
                if statement.strip():
                    conn.execute(statement)
        if version < 3:
            for statement in SCHEMA_V3.strip().split(';'):
                if statement.strip():
                    conn.execute(statement)
        if version < 4:
            for statement in SCHEMA_V4.strip().split(';'):
                if statement.strip():
                    conn.execute(statement)
        conn.execute('PRAGMA user_version={}'.format(SCHEMA_VERSION))
        conn.execute('COMMIT')
    except Exception:
        conn.execute('ROLLBACK')
        raise


class _Tx:
    """with store.tx() as conn: one IMMEDIATE transaction."""

    def __init__(self, path=None):
        self.conn = connect(path)

    def __enter__(self):
        self.conn.execute('BEGIN IMMEDIATE')
        return self.conn

    def __exit__(self, exc_type, exc, tb):
        changed = set(self.conn.widget_rows) if not exc_type else set()
        try:
            if not exc_type:
                for row in changed:
                    key = 'widget.rev.' + row
                    self.conn.execute('INSERT OR IGNORE INTO flags(key,value) VALUES(?,0)', (key,))
                    self.conn.execute('UPDATE flags SET value=CAST(value AS INTEGER)+1 WHERE key=?', (key,))
            self.conn.execute('ROLLBACK' if exc_type else 'COMMIT')
        finally:
            self.conn.close()
        if changed:
            try:
                from wcs.skin.refresh import notify
                notify('skin_rows_changed', changed)
            except Exception:
                pass  # committed revisions are recovered by the service on startup / its next check
        return False


def tx(path=None):
    return _Tx(path)


def _now():
    return int(time.time())


def dirty(conn, rows):
    """Collect rows once per transaction; revisions commit with the corresponding data."""
    conn.widget_rows.update(rows)


def _media_dirty(conn, media_type, membership=False):
    from wcs.skin.refresh import media_rows
    dirty(conn, media_rows(media_type, membership))


def _queue_mirror(conn, media_type, tmdb_id, season=None, episode=None):
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    conn.execute('UPDATE {} SET state_version=state_version+1 WHERE {}'.format(table, where), args)
    version = conn.execute('SELECT state_version FROM {} WHERE {}'.format(table, where), args).fetchone()[0]
    conn.execute('INSERT OR REPLACE INTO mirror_pending(media_type,tmdb_id,season,episode,version,next_attempt) '
                 'VALUES(?,?,?,?,?,?)', (media_type, int(tmdb_id), season if season is not None else -1,
                                         episode if episode is not None else -1, version, time.time() + 0.5))


# --------------------------------------------------------------------------- lists (030 §4)

def _ensure_title(conn, media_type, tmdb_id):
    conn.execute('INSERT OR IGNORE INTO titles(media_type, tmdb_id) VALUES(?, ?)', (media_type, int(tmdb_id)))


def get_list(media_type):
    """Legacy dicts of My movies ('movie') / My series ('tv') in list order."""
    conn = connect()
    try:
        rows = conn.execute('SELECT legacy_json, tmdb_id FROM titles WHERE media_type=? AND in_list=1 '
                            'ORDER BY list_pos, list_added_at, tmdb_id', (media_type,)).fetchall()
    finally:
        conn.close()
    result = []
    for row in rows:
        try:
            entry = json.loads(row['legacy_json']) if row['legacy_json'] else {}
        except ValueError:
            entry = {}
        entry.setdefault('id', row['tmdb_id'])
        result.append(entry)
    return result


def list_times(media_type):
    """{tmdb_id: {'added', 'activity'}} (epoch seconds or None) of the titles in My movies / My series."""
    conn = connect()
    try:
        rows = conn.execute('SELECT tmdb_id, list_added_at, last_activity_at FROM titles WHERE media_type=? AND in_list=1',
                            (media_type,)).fetchall()
    finally:
        conn.close()
    return {r['tmdb_id']: {'added': r['list_added_at'], 'activity': r['last_activity_at']} for r in rows}


def replace_list(media_type, entries):
    """Make `entries` (legacy dicts with 'id') the whole list, keeping their order."""
    now = _now()
    with tx() as conn:
        keep = []
        changed = 0
        for pos, entry in enumerate(entries):
            tmdb_id = int(entry['id'])
            keep.append(tmdb_id)
            _ensure_title(conn, media_type, tmdb_id)
            encoded = json.dumps(entry, ensure_ascii=False)
            changed += conn.execute('UPDATE titles SET in_list=1, list_pos=?, legacy_json=?, '
                         'list_added_at=COALESCE(list_added_at, ?) WHERE media_type=? AND tmdb_id=? '
                         'AND (in_list!=1 OR list_pos IS NOT ? OR legacy_json IS NOT ?)',
                         (pos, encoded, now, media_type, tmdb_id, pos, encoded)).rowcount
        if keep:
            marks = ','.join('?' * len(keep))
            changed += conn.execute('UPDATE titles SET in_list=0, list_pos=NULL WHERE media_type=? AND in_list=1 '
                         'AND tmdb_id NOT IN ({})'.format(marks), [media_type] + keep).rowcount
        else:
            changed += conn.execute('UPDATE titles SET in_list=0, list_pos=NULL WHERE media_type=? AND in_list=1', (media_type,)).rowcount
        if changed:
            _media_dirty(conn, media_type, True)


def set_in_list(media_type, tmdb_id, in_list, legacy=None):
    """Add / remove one title. Added titles go to the end of the list (like the old append)."""
    now = _now()
    with tx() as conn:
        _ensure_title(conn, media_type, tmdb_id)
        if in_list:
            end = conn.execute('SELECT COALESCE(MAX(list_pos), -1) + 1 FROM titles WHERE media_type=? AND in_list=1',
                               (media_type,)).fetchone()[0]
            changed = conn.execute('UPDATE titles SET in_list=1, list_pos=?, list_added_at=?, legacy_json=COALESCE(?, legacy_json) '
                         'WHERE media_type=? AND tmdb_id=? AND in_list=0',
                         (end, now, json.dumps(legacy, ensure_ascii=False) if legacy else None, media_type, int(tmdb_id))).rowcount
        else:
            changed = conn.execute('UPDATE titles SET in_list=0, list_pos=NULL WHERE media_type=? AND tmdb_id=? AND in_list!=0',
                         (media_type, int(tmdb_id))).rowcount
        if changed:
            _media_dirty(conn, media_type, True)


def is_in_list(media_type, tmdb_id):
    conn = connect()
    try:
        row = conn.execute('SELECT in_list FROM titles WHERE media_type=? AND tmdb_id=?',
                           (media_type, int(tmdb_id))).fetchone()
    finally:
        conn.close()
    return bool(row and row['in_list'])


# --------------------------------------------------------------------------- watch state (030 §2 rules)

def _state_keys(media_type, tmdb_id, season=None, episode=None):
    if media_type == 'movie':
        return 'movie_state', 'tmdb_id=?', (int(tmdb_id),), ('tmdb_id',), (int(tmdb_id),)
    return ('episode_state', 'show_id=? AND season=? AND episode=?', (int(tmdb_id), int(season), int(episode)),
            ('show_id', 'season', 'episode'), (int(tmdb_id), int(season), int(episode)))


def _ensure_state(conn, media_type, tmdb_id, season, episode):
    table, _where, _args, cols, vals = _state_keys(media_type, tmdb_id, season, episode)
    conn.execute('INSERT OR IGNORE INTO {}({}) VALUES({})'.format(table, ','.join(cols), ','.join('?' * len(cols))), vals)


def _touch(conn, media_type, tmdb_id, when):
    _ensure_title(conn, media_type, tmdb_id)
    conn.execute('UPDATE titles SET last_activity_at=?, hidden_from_continue_at=NULL WHERE media_type=? AND tmdb_id=?',
                 (when, media_type, int(tmdb_id)))
    _media_dirty(conn, media_type)


def get_state(media_type, tmdb_id, season=None, episode=None):
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    conn = connect()
    try:
        row = conn.execute('SELECT * FROM {} WHERE {}'.format(table, where), args).fetchone()
    finally:
        conn.close()
    return dict(row) if row else None


def get_movie_states():
    """{tmdb_id: movie_state row} of every movie with a state."""
    conn = connect()
    try:
        rows = conn.execute('SELECT * FROM movie_state').fetchall()
    finally:
        conn.close()
    return {r['tmdb_id']: dict(r) for r in rows}


def get_all_episode_states():
    """{show_id: {(season, episode): episode_state row}} for every show with states."""
    conn = connect()
    try:
        rows = conn.execute('SELECT * FROM episode_state').fetchall()
    finally:
        conn.close()
    result = {}
    for r in rows:
        result.setdefault(r['show_id'], {})[(r['season'], r['episode'])] = dict(r)
    return result


def get_episode_states(show_id):
    conn = connect()
    try:
        rows = conn.execute('SELECT * FROM episode_state WHERE show_id=? ORDER BY season, episode',
                            (int(show_id),)).fetchall()
    finally:
        conn.close()
    return [dict(r) for r in rows]


def save_resume(media_type, tmdb_id, position, total, season=None, episode=None, source=None, when=None):
    """In-progress position; also records the source file used (`source` = dict ident/provider/file_name)."""
    when = when or _now()
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    source = source or {}
    with tx() as conn:
        _ensure_state(conn, media_type, tmdb_id, season, episode)
        changed = conn.execute('UPDATE {} SET resume_pos=?, resume_total=?, last_played_at=?, last_ident=COALESCE(?, last_ident), '
                     'last_provider=COALESCE(?, last_provider), last_file_name=COALESCE(?, last_file_name) '
                     'WHERE {} AND (resume_pos!=? OR resume_total!=? OR COALESCE(?,last_ident) IS NOT last_ident '
                     'OR COALESCE(?,last_provider) IS NOT last_provider OR COALESCE(?,last_file_name) IS NOT last_file_name)'.format(table, where),
                     (float(position), float(total), when, source.get('ident'), source.get('provider'),
                      source.get('file_name')) + args + (float(position), float(total), source.get('ident'),
                                                        source.get('provider'), source.get('file_name'))).rowcount
        if changed:
            _touch(conn, media_type, tmdb_id, when)
            _queue_mirror(conn, media_type, tmdb_id, season, episode)


def mark_watched(media_type, tmdb_id, season=None, episode=None, when=None, source=None):
    """Watched: playcount + 1, resume cleared."""
    when = when or _now()
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    source = source or {}
    with tx() as conn:
        _ensure_state(conn, media_type, tmdb_id, season, episode)
        conn.execute('UPDATE {} SET playcount=playcount+1, resume_pos=0, resume_total=0, last_played_at=?, '
                     'last_ident=COALESCE(?, last_ident), last_provider=COALESCE(?, last_provider), '
                     'last_file_name=COALESCE(?, last_file_name) WHERE {}'.format(table, where),
                     (when, source.get('ident'), source.get('provider'), source.get('file_name')) + args)
        _touch(conn, media_type, tmdb_id, when)
        _queue_mirror(conn, media_type, tmdb_id, season, episode)


def mark_unwatched(media_type, tmdb_id, season=None, episode=None, when=None):
    """Unwatched: playcount 0 and resume cleared."""
    when = when or _now()
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    with tx() as conn:
        _ensure_state(conn, media_type, tmdb_id, season, episode)
        conn.execute('UPDATE {} SET playcount=0, resume_pos=0, resume_total=0 WHERE {}'.format(table, where), args)
        _touch(conn, media_type, tmdb_id, when)
        _queue_mirror(conn, media_type, tmdb_id, season, episode)


def clear_resume(media_type, tmdb_id, season=None, episode=None):
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    with tx() as conn:
        changed = conn.execute('UPDATE {} SET resume_pos=0, resume_total=0 WHERE {} '
                               'AND (resume_pos!=0 OR resume_total!=0)'.format(table, where), args).rowcount
        if changed:
            _media_dirty(conn, media_type)
            _queue_mirror(conn, media_type, tmdb_id, season, episode)


def mark_many(media_type, tmdb_id, targets, watched=True, when=None, pending_id=None):
    """Manual mark: one transaction, no network/RPC, and no second playcount for a repeated mark."""
    when = when or _now()
    changed = 0
    with tx() as conn:
        if pending_id is not None and not conn.execute('SELECT 1 FROM pending_marks WHERE id=?', (pending_id,)).fetchone():
            return 0
        table = 'movie_state' if media_type == 'movie' else 'episode_state'
        existing = ({(None, None): dict(r) for r in conn.execute('SELECT * FROM movie_state WHERE tmdb_id=?', (int(tmdb_id),))}
                    if media_type == 'movie' else
                    {(r['season'], r['episode']): dict(r) for r in conn.execute(
                        'SELECT * FROM episode_state WHERE show_id=?', (int(tmdb_id),))})
        for season, episode in set(targets):
            state = existing.get((season, episode)) or {}
            played = int(state.get('playcount') or 0) > 0
            if (watched and played) or (not watched and not played and not float(state.get('resume_pos') or 0)):
                continue
            _ensure_state(conn, media_type, tmdb_id, season, episode)
            _table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
            if watched:
                conn.execute('UPDATE {} SET playcount=1,resume_pos=0,resume_total=0,last_played_at=? WHERE {}'.format(table, where), (when,) + args)
            else:
                conn.execute('UPDATE {} SET playcount=0,resume_pos=0,resume_total=0 WHERE {}'.format(table, where), args)
            _queue_mirror(conn, media_type, tmdb_id, season, episode)
            changed += 1
        if changed:
            _touch(conn, media_type, tmdb_id, when)
        if pending_id is not None:
            conn.execute('DELETE FROM pending_marks WHERE id=?', (pending_id,))
    return changed


def log_play(media_type, tmdb_id, season, episode, started_at, ended_at, position, total, outcome):
    with tx() as conn:
        conn.execute('INSERT INTO play_log(media_type,tmdb_id,season,episode,started_at,ended_at,position,total,outcome) '
                     'VALUES(?,?,?,?,?,?,?,?,?)',
                     (media_type, int(tmdb_id), season, episode, started_at, ended_at, position, total, outcome))
        dirty(conn, {'recent_played'})


def log_start(media_type, tmdb_id, season, episode, started_at):
    """Open a play_log row (ended_at NULL); returns its id. An open row after a crash is closed by close_open_logs()."""
    with tx() as conn:
        cur = conn.execute('INSERT INTO play_log(media_type,tmdb_id,season,episode,started_at) VALUES(?,?,?,?,?)',
                           (media_type, int(tmdb_id), season, episode, started_at))
        dirty(conn, {'recent_played'})
        return cur.lastrowid


def log_update(row_id, position, total):
    with tx() as conn:
        conn.execute('UPDATE play_log SET position=?, total=? WHERE id=?', (position, total, row_id))


def log_finish(row_id, ended_at, position, total, outcome):
    with tx() as conn:
        conn.execute('UPDATE play_log SET ended_at=?, position=?, total=?, outcome=? WHERE id=?',
                     (ended_at, position, total, outcome, row_id))


def close_open_logs(when=None):
    """Service start: rows left open by a crash / power loss get outcome 'crash' (their last position stays)."""
    with tx() as conn:
        cur = conn.execute("UPDATE play_log SET ended_at=?, outcome='crash' WHERE ended_at IS NULL",
                           (when or _now(),))
        return cur.rowcount


def recent_logs(limit=20):
    conn = connect()
    try:
        rows = conn.execute('SELECT * FROM play_log ORDER BY id DESC LIMIT ?', (limit,)).fetchall()
    finally:
        conn.close()
    return [dict(r) for r in rows]


def set_mirrored(media_type, tmdb_id, playcount, resume_pos, season=None, episode=None, when=None, version=None):
    """Remember what was written into Kodi (so a later difference means the user changed it in Kodi)."""
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    with tx() as conn:
        changed = conn.execute('UPDATE {} SET mirrored_at=?, mirrored_playcount=?, mirrored_resume_pos=? WHERE {}{}'.format(
                     table, where, ' AND state_version=?' if version is not None else ''),
                     (when or _now(), int(playcount), float(resume_pos)) + args + ((version,) if version is not None else ())).rowcount
        if changed and version is not None:
            conn.execute('DELETE FROM mirror_pending WHERE media_type=? AND tmdb_id=? AND season=? AND episode=? AND version=?',
                         (media_type, int(tmdb_id), season if season is not None else -1,
                          episode if episode is not None else -1, version))
        return bool(changed)


def pending_mirrors(limit=5, now=None):
    conn = connect()
    try:
        return [dict(r) for r in conn.execute('SELECT * FROM mirror_pending WHERE next_attempt<=? '
                                              'ORDER BY next_attempt LIMIT ?', (time.time() if now is None else now, limit))]
    finally:
        conn.close()


def mirror_pending(media_type, tmdb_id, season=None, episode=None):
    conn = connect()
    try:
        return conn.execute('SELECT 1 FROM mirror_pending WHERE media_type=? AND tmdb_id=? AND season=? AND episode=?',
                            (media_type, int(tmdb_id), season if season is not None else -1,
                             episode if episode is not None else -1)).fetchone() is not None
    finally:
        conn.close()


def retry_mirror(pending):
    delay = min(600, 2 ** min(pending['attempts'] + 1, 10))
    with tx() as conn:
        conn.execute('UPDATE mirror_pending SET attempts=attempts+1,next_attempt=? '
                     'WHERE media_type=? AND tmdb_id=? AND season=? AND episode=? AND version=?',
                     (time.time() + delay, pending['media_type'], pending['tmdb_id'], pending['season'],
                      pending['episode'], pending['version']))


def adopt_from_kodi(media_type, tmdb_id, playcount, resume_pos, resume_total, season=None, episode=None, when=None):
    """Apply a change the user made through Kodi (mark watched / unwatched / remove resume point)."""
    when = when or _now()
    table, where, args, _cols, _vals = _state_keys(media_type, tmdb_id, season, episode)
    with tx() as conn:
        _ensure_state(conn, media_type, tmdb_id, season, episode)
        conn.execute('UPDATE {} SET playcount=?, resume_pos=?, resume_total=? WHERE {}'.format(table, where),
                     (int(playcount), float(resume_pos), float(resume_total)) + args)
        _touch(conn, media_type, tmdb_id, when)
        _queue_mirror(conn, media_type, tmdb_id, season, episode)


def set_hidden_from_continue(media_type, tmdb_id, hidden, when=None):
    with tx() as conn:
        _ensure_title(conn, media_type, tmdb_id)
        changed = conn.execute('UPDATE titles SET hidden_from_continue_at=? WHERE media_type=? AND tmdb_id=? '
                               'AND hidden_from_continue_at IS {}NULL'.format('' if hidden else 'NOT '),
                     ((when or _now()) if hidden else None, media_type, int(tmdb_id))).rowcount
        if changed:
            _media_dirty(conn, media_type)


def all_state_keys():
    """[('movie', id, None, None), ('tv', show, season, episode)] for every stored watch state."""
    conn = connect()
    try:
        keys = [('movie', r[0], None, None) for r in conn.execute('SELECT tmdb_id FROM movie_state')]
        keys += [('tv', r[0], r[1], r[2]) for r in conn.execute('SELECT show_id, season, episode FROM episode_state')]
    finally:
        conn.close()
    return keys


def reset_watch_state():
    """Delete every watched / resume state and the play log; the lists stay. Returns the removed row counts."""
    with tx() as conn:
        counts = {'movies': conn.execute('SELECT COUNT(*) FROM movie_state').fetchone()[0],
                  'episodes': conn.execute('SELECT COUNT(*) FROM episode_state').fetchone()[0],
                  'plays': conn.execute('SELECT COUNT(*) FROM play_log').fetchone()[0]}
        for table in ('movie_state', 'episode_state', 'play_log', 'mirror_pending', 'pending_marks'):
            conn.execute('DELETE FROM {}'.format(table))
        conn.execute('UPDATE titles SET last_activity_at=NULL, hidden_from_continue_at=NULL')
        _media_dirty(conn, 'movie')
        _media_dirty(conn, 'tv')
    return counts


def get_title(media_type, tmdb_id):
    conn = connect()
    try:
        row = conn.execute('SELECT * FROM titles WHERE media_type=? AND tmdb_id=?',
                           (media_type, int(tmdb_id))).fetchone()
    finally:
        conn.close()
    return dict(row) if row else None


# --------------------------------------------------------------------------- taste feedback (docs/plan/pro-tebe)

FEEDBACK_VALUES = ('like', 'dislike', 'hide', 'pick')


def set_feedback(media_type, tmdb_id, value, when=None):
    """like / dislike / hide (not interested) / pick (first-run favourite); None clears the title's feedback."""
    if value is not None and value not in FEEDBACK_VALUES:
        raise ValueError('Unknown feedback {!r}'.format(value))
    with tx() as conn:
        if value is None:
            conn.execute('DELETE FROM feedback WHERE media_type=? AND tmdb_id=?', (media_type, int(tmdb_id)))
        else:
            conn.execute('INSERT OR REPLACE INTO feedback(media_type, tmdb_id, value, at) VALUES(?,?,?,?)',
                         (media_type, int(tmdb_id), value, int(when if when is not None else _now())))


def reset_taste():
    """Delete the likes and the first-run picks, so Pro tebe offers the pick again; dislikes and hidden stay."""
    with tx() as conn:
        return conn.execute("DELETE FROM feedback WHERE value IN ('like', 'pick')").rowcount


def get_feedback():
    """{(media_type, tmdb_id): (value, at)} of every title with feedback."""
    conn = connect()
    try:
        rows = conn.execute('SELECT media_type, tmdb_id, value, at FROM feedback').fetchall()
    finally:
        conn.close()
    return {(r['media_type'], r['tmdb_id']): (r['value'], r['at']) for r in rows}


# --------------------------------------------------------------------------- flags + legacy import

def get_flag(key):
    conn = connect()
    try:
        row = conn.execute('SELECT value FROM flags WHERE key=?', (key,)).fetchone()
    finally:
        conn.close()
    return row['value'] if row else None


def set_flag(key, value):
    with tx() as conn:
        conn.execute('INSERT OR REPLACE INTO flags(key, value) VALUES(?, ?)', (key, str(value)))


LEGACY_FILES = (('movie', 'watchlist_movies.json'), ('tv', 'watchlist_series.json'))


def import_watchlists(profile_dir, snapshot=None):
    """Explicit, atomic append/merge. Originals and immutable backup copies stay.

    Return valid unique source counts ({} for an already committed fingerprint).
    A failed read/backup or a changed reviewed snapshot raises without importing.
    Existing membership, snapshots, order and watch state take precedence.
    """
    current = inspect_watchlist_import(profile_dir)
    if snapshot is not None and snapshot['fingerprint'] != current['fingerprint']:
        raise ValueError('Legacy lists changed after confirmation')
    snapshot = current
    if not snapshot['files']:
        return {}
    if get_flag('watchlists.import.v2') == snapshot['fingerprint']:
        return {}
    for source in snapshot['files']:
        _import_backup(source)
    with tx() as conn:
        flag = conn.execute('SELECT value FROM flags WHERE key=?', ('watchlists.import.v2',)).fetchone()
        if flag and flag['value'] == snapshot['fingerprint']:
            return {}
        for media_type, _name in LEGACY_FILES:
            end = conn.execute('SELECT COALESCE(MAX(list_pos), -1) + 1 FROM titles '
                               'WHERE media_type=? AND in_list=1', (media_type,)).fetchone()[0]
            changed = False
            for entry in snapshot['entries'][media_type]:
                tmdb_id = entry['id']
                _ensure_title(conn, media_type, tmdb_id)
                updated = conn.execute('UPDATE titles SET in_list=1, list_pos=?, '
                                       'list_added_at=COALESCE(list_added_at, ?), '
                                       'legacy_json=COALESCE(legacy_json, ?) '
                                       'WHERE media_type=? AND tmdb_id=? AND in_list=0',
                                       (end, _now(), json.dumps(entry, ensure_ascii=False), media_type, tmdb_id)).rowcount
                if updated:
                    changed = True
                    end += 1
            if changed:
                _media_dirty(conn, media_type, True)
        conn.execute('INSERT OR REPLACE INTO flags(key,value) VALUES(?,?)',
                     ('watchlists.import.v2', snapshot['fingerprint']))
        conn.execute('INSERT OR REPLACE INTO flags(key,value) VALUES(?,?)',
                     ('watchlists_imported', str(_now())))
    return {kind: len(snapshot['entries'][kind]) for kind, _name in LEGACY_FILES}


def _import_id(value):
    """Only positive integer TMDb IDs; bool/fractional/overflow IDs are invalid."""
    if isinstance(value, bool) or not isinstance(value, (int, str)):
        return None
    try:
        number = int(value)
        return number if 0 < number <= 9223372036854775807 else None
    except ValueError:
        return None


def inspect_watchlist_import(profile_dir):
    """Read-only reviewed snapshot; unreadable/malformed whole files raise.

    Fall back to old *.migrated files only when the original is absent. Invalid
    individual entries are counted and retained in the original/backup bytes.
    """
    import hashlib
    digest = hashlib.sha256(b'watchlists-import-v2\0')
    result = {'entries': {}, 'counts': {}, 'files': [], 'duplicates': 0, 'invalid': 0, 'existing': 0}
    for media_type, name in LEGACY_FILES:
        path = os.path.join(profile_dir, name)
        if not os.path.exists(path):
            path += '.migrated'
        digest.update(name.encode('utf-8') + b'\0')
        entries, seen = [], set()
        count = 0
        try:
            with open(path, 'rb') as handle:
                raw = handle.read()
        except FileNotFoundError:
            raw = None
        if raw is not None:
            # Missing != valid empty list; include exact bytes in the consent identity.
            digest.update(b'present\0' + hashlib.sha256(raw).digest())
            data = json.loads(raw.decode('utf-8-sig'), parse_constant=_import_bad_constant)
            if not isinstance(data, list):
                raise ValueError('Legacy watchlist must be a JSON list: ' + name)
            count = len(data)
            result['files'].append({'path': path, 'raw': raw})
            for entry in data:
                tmdb_id = _import_id(entry.get('id')) if isinstance(entry, dict) else None
                if tmdb_id is None:
                    result['invalid'] += 1
                elif tmdb_id in seen:
                    result['duplicates'] += 1
                else:
                    seen.add(tmdb_id)
                    entries.append(dict(entry, id=tmdb_id))
        else:
            digest.update(b'absent\0')
        result['entries'][media_type] = entries
        result['counts'][media_type] = count
    conn = connect()
    try:
        for media_type, entries in result['entries'].items():
            existing = {r[0] for r in conn.execute('SELECT tmdb_id FROM titles WHERE media_type=? AND in_list=1',
                                                 (media_type,))}
            result['existing'] += sum(e['id'] in existing for e in entries)
    finally:
        conn.close()
    result['fingerprint'] = digest.hexdigest()
    return result


def _import_bad_constant(value):
    raise ValueError('Invalid JSON number: ' + value)


def watchlist_import_empty():
    """Automatic migration is offered only before either new list is populated."""
    conn = connect()
    try:
        return conn.execute('SELECT 1 FROM titles WHERE in_list>0 LIMIT 1').fetchone() is None
    finally:
        conn.close()


def _import_backup(source):
    """Never overwrite an older backup; fsync before the database commit.

    A truncated copy left by a crash is kept; a new copy is retried. Exclusive
    creation works on Android/external storage without hard-link support.
    """
    import hashlib
    path, raw = source['path'], source['raw']
    if path.endswith('.migrated'):
        return path
    target = path + '.migrated'
    fallback = path + '.migration-' + hashlib.sha256(raw).hexdigest() + '.bak'
    attempt = 0
    while True:
        try:
            with open(target, 'xb') as handle:
                handle.write(raw)
                handle.flush()
                os.fsync(handle.fileno())
            break
        except FileExistsError:
            with open(target, 'rb') as handle:
                if handle.read() == raw:
                    # A previous attempt may have failed during fsync.
                    os.fsync(handle.fileno())
                    return target
            target = fallback if attempt == 0 else fallback + '.' + str(attempt)
            attempt += 1
    # Some external filesystems cannot fsync directories. The original remains
    # untouched, and the complete backup itself has already been fsynced.
    try:
        folder = os.open(os.path.dirname(path) or '.', os.O_RDONLY)
        try:
            os.fsync(folder)
        finally:
            os.close(folder)
    except OSError:
        pass
    return target