"""Position knowledge and portable study folders layered on the PGN library."""
import json
import os
import threading
from .chess import Chess
from . import pgnutil
from .store import slugify
SCHEMA = '''
CREATE TABLE IF NOT EXISTS positions (
hash TEXT NOT NULL, game_id INTEGER REFERENCES games(id) ON DELETE CASCADE,
ply INTEGER NOT NULL, next_san TEXT, PRIMARY KEY(hash, game_id, ply));
CREATE INDEX IF NOT EXISTS positions_game ON positions(game_id);
-- Covers the position lookup, which reads the next move as well as the game: without
-- it every game reaching a position is a separate fetch from the positions table.
CREATE INDEX IF NOT EXISTS positions_hash_next ON positions(hash, game_id, next_san);
CREATE TABLE IF NOT EXISTS pins (
id INTEGER PRIMARY KEY, hash TEXT NOT NULL, fen TEXT NOT NULL,
title TEXT NOT NULL, url TEXT NOT NULL DEFAULT '', note TEXT NOT NULL DEFAULT '');
CREATE INDEX IF NOT EXISTS pins_hash ON pins(hash);
CREATE TABLE IF NOT EXISTS folders (
id INTEGER PRIMARY KEY, name TEXT NOT NULL, path TEXT NOT NULL UNIQUE,
parent_id INTEGER REFERENCES folders(id), created_at TEXT DEFAULT CURRENT_TIMESTAMP);
CREATE TABLE IF NOT EXISTS collection_folders (
collection_id INTEGER PRIMARY KEY REFERENCES collections(id) ON DELETE CASCADE,
folder_id INTEGER REFERENCES folders(id));
CREATE TABLE IF NOT EXISTS game_tags (
game_id INTEGER REFERENCES games(id) ON DELETE CASCADE, tag TEXT NOT NULL,
PRIMARY KEY(game_id, tag));
'''
[docs]
class Study:
def __init__(self, library):
self.library = library
library.connect().executescript(SCHEMA)
if 'category' not in {r[1] for r in library.connect().execute('PRAGMA table_info(folders)')}:
library.connect().execute("ALTER TABLE folders ADD COLUMN category TEXT NOT NULL DEFAULT 'Games to study'")
library.connect().commit()
self.lock = threading.Lock()
self.state = {'running': False, 'done': 0, 'errors': 0, 'total': 0, 'collection': None}
[docs]
def index(self, collection):
info = self.library.collection(collection)
if not info:
raise ValueError('Choose an existing collection')
with self.lock:
if self.state['running']:
raise ValueError('Position indexing is already running')
self.state = dict(running=True, done=0, errors=0, total=0, collection=info['name'])
threading.Thread(target=self._index, args=(info['id'],), daemon=True).start()
return self.state.copy()
def _index(self, collection_id):
db = self.library.connect()
pending = []
def flush():
with self.library._write_lock, db:
for game_id, signature, rows in pending:
current = db.execute('SELECT signature FROM games WHERE id=?', (game_id,)).fetchone()
if not current or current[0] != signature:
continue # An edit or deletion during replay must not get a stale index.
db.execute('DELETE FROM positions WHERE game_id=?', (game_id,))
db.executemany('INSERT INTO positions VALUES (?,?,?,?)', rows)
pending.clear()
try:
ids = [r[0] for r in db.execute('''SELECT g.id FROM games g
WHERE (g.collection_id=? OR g.id IN
(SELECT game_id FROM game_collections WHERE collection_id=?))
AND NOT EXISTS (SELECT 1 FROM positions p WHERE p.game_id=g.id)''',
(collection_id, collection_id))]
self.state['total'] = len(ids)
for game_id in ids:
try:
signature = db.execute('SELECT signature FROM games WHERE id=?', (game_id,)).fetchone()[0]
pgn = self.library.game_pgn(game_id)
game = Chess(pgnutil.headers(pgn).get('FEN') or Chess().fen())
moves = pgnutil.moves(pgn)
rows = []
for ply in range(len(moves) + 1):
rows.append((game.key(), game_id, ply, moves[ply] if ply < len(moves) else None))
if ply < len(moves):
game.move(moves[ply])
pending.append((game_id, signature, rows))
if len(pending) >= 25:
flush()
except (ValueError, TypeError):
self.state['errors'] += 1
self.state['done'] += 1
flush()
except Exception as err:
self.state['error'] = str(err)
finally:
self.state['running'] = False
self.library.close()
[docs]
def position(self, fen):
key = Chess(fen).key()
db = self.library.connect()
# One read of the games reaching this position, summarised here, rather than
# three joins that each fetched the same rows: the starting position is reached
# by every indexed game, and this runs on every move made on the board.
rows = db.execute('''SELECT p.game_id, p.next_san, g.white, g.black, g.date, g.event,
g.result, g.eco, g.opening FROM positions p JOIN games g ON g.id = p.game_id
WHERE p.hash = ?''', (key,)).fetchall()
seen, unique, continuations = set(), [], set()
for r in rows:
if r['next_san'] is not None:
continuations.add((r['game_id'], r['next_san'], r['result']))
if r['game_id'] not in seen:
seen.add(r['game_id'])
unique.append(r)
undated = lambda d: not d or d.startswith('0000')
ordered = sorted(unique, key=lambda r: (undated(r['date']), r['date'] or '', r['game_id']))
games = [dict(id=r['game_id'], white=r['white'], black=r['black'], date=r['date'],
event=r['event'], result=r['result'], eco=r['eco'], opening=r['opening'])
for r in ordered[:100]]
tally = {}
for _, san, result in continuations:
m = tally.setdefault(san, dict(san=san, games=0, white=0, draws=0, black=0))
m['games'] += 1
m['white'] += result == '1-0'
m['draws'] += result == '1/2-1/2'
m['black'] += result == '0-1'
moves = sorted(tally.values(), key=lambda m: -m['games'])
by_decade = {}
for r in unique:
year = (r['date'] or '')[:4]
if year > '0000' and year[:1].isdigit():
decade = int(''.join(c for c in year if c.isdigit()) or 0) // 10 * 10
by_decade[decade] = by_decade.get(decade, 0) + 1
decades = [dict(decade=d, games=n) for d, n in sorted(by_decade.items())]
return dict(hash=key, games=games, moves=moves, decades=decades,
pins=[dict(r) for r in db.execute('SELECT * FROM pins WHERE hash=?', (key,))],
indexed_games=db.execute('SELECT COUNT(DISTINCT game_id) FROM positions').fetchone()[0])
[docs]
def pin(self, body):
fen = body['fen']
title = str(body.get('title', '')).strip()
url = str(body.get('url', '')).strip()
if not title or (url and not url.startswith(('https://', 'http://'))):
raise ValueError('Give a title and an HTTP(S) link, or leave the link empty')
with self.library._write_lock, self.library.connect() as db:
cur = db.execute('INSERT INTO pins(hash,fen,title,url,note) VALUES(?,?,?,?,?)',
(Chess(fen).key(), fen, title, url, str(body.get('note', ''))))
return {'id': cur.lastrowid}
[docs]
def folders(self):
db = self.library.connect()
return dict(folders=[dict(r) for r in db.execute('SELECT * FROM folders ORDER BY path')],
assignments=[dict(r) for r in db.execute('SELECT * FROM collection_folders')],
root=os.path.join(self.library.dir, 'studies'))
[docs]
def create_folder(self, body):
title = str(body.get('name', '')).strip()
if not title:
raise ValueError('Give the study folder a name')
parent_id = body.get('parent_id') or None
db = self.library.connect()
parent = db.execute('SELECT path FROM folders WHERE id=?', (parent_id,)).fetchone() if parent_id else None
if parent_id and not parent:
raise ValueError('Parent folder does not exist')
relative = '/'.join(filter(None, [parent['path'] if parent else '', slugify(title)]))
base = os.path.realpath(os.path.join(self.library.dir, 'studies'))
target = os.path.realpath(os.path.join(base, relative))
if os.path.commonpath([base, target]) != base:
raise ValueError('Invalid folder path')
with self.library._write_lock, db:
if db.execute('SELECT id FROM folders WHERE path=?', (relative,)).fetchone():
raise ValueError('A folder with that name already exists here')
os.makedirs(target, exist_ok=True)
cur = db.execute('INSERT INTO folders(name,path,parent_id) VALUES(?,?,?)', (title, relative, parent_id))
db.execute('UPDATE folders SET category=? WHERE id=?',(str(body.get('category') or 'Games to study').strip()[:100],cur.lastrowid))
self.write_manifest(cur.lastrowid)
return {'id': cur.lastrowid, 'path': target}
[docs]
def delete_folder(self, folder_id):
"""Remove a folder and everything nested inside it.
Collections are only unassigned: their PGN files live in the library's
collections directory, not in the study folder. On disk we take back the
manifest we wrote and the directories we created, but a directory the user
has put their own files in is left alone and reported back.
"""
db = self.library.connect()
folder = db.execute('SELECT * FROM folders WHERE id=?', (int(folder_id),)).fetchone()
if not folder:
raise ValueError('No such folder')
prefix = folder['path'] + '/'
doomed = [dict(r) for r in db.execute('SELECT id, name, path FROM folders')
if r['path'] == folder['path'] or r['path'].startswith(prefix)]
# Children first: parent_id is a real foreign key.
doomed.sort(key=lambda f: f['path'].count('/'), reverse=True)
with self.library._write_lock, db:
for row in doomed:
db.execute('DELETE FROM collection_folders WHERE folder_id=?', (row['id'],))
for row in doomed:
db.execute('DELETE FROM folders WHERE id=?', (row['id'],))
base = os.path.realpath(os.path.join(self.library.dir, 'studies'))
kept = []
for row in doomed:
if not row['path']:
continue
target = os.path.realpath(os.path.join(base, row['path']))
if os.path.commonpath([base, target]) != base or not os.path.isdir(target):
continue
manifest = os.path.join(target, 'study.json')
if os.path.isfile(manifest):
os.remove(manifest)
try:
os.rmdir(target)
except OSError: # the folder still holds the user's own files
kept.append(target)
return {'deleted': len(doomed), 'kept': kept}
[docs]
def shelve(self, collection_id, folder_name, category='Reference'):
"""File a collection under a study folder, making the folder if it is missing.
A base that only exists in a drop-down is a base you forget you have. Study
folders are where this library keeps the things it is working from, so an
attached base is filed there on arrival like anything else.
"""
db = self.library.connect()
row = db.execute('SELECT id FROM folders WHERE lower(name)=lower(?) AND parent_id IS NULL',
(folder_name,)).fetchone()
folder_id = row['id'] if row else self.create_folder({'name': folder_name})['id']
if not row:
try:
self.categorize_folder(folder_id, category)
except (ValueError, KeyError):
pass # an unknown category is not worth failing over
self.assign({'folder_id': folder_id, 'collection_id': collection_id})
return folder_id
[docs]
def assign(self, body):
db = self.library.connect()
folder_id, collection_id = int(body['folder_id']), int(body['collection_id'])
if not db.execute('SELECT 1 FROM folders WHERE id=?', (folder_id,)).fetchone():
raise ValueError('No such folder')
if not self.library.collection(collection_id):
raise ValueError('No such collection')
old = db.execute('SELECT folder_id FROM collection_folders WHERE collection_id=?', (collection_id,)).fetchone()
with self.library._write_lock, db:
db.execute('INSERT OR REPLACE INTO collection_folders VALUES(?,?)', (collection_id, folder_id))
self.write_manifest(folder_id)
if old and old[0] != folder_id:
self.write_manifest(old[0])
return {'saved': True}
[docs]
def write_manifest(self, folder_id):
db = self.library.connect()
folder = db.execute('SELECT * FROM folders WHERE id=?', (folder_id,)).fetchone()
if not folder:
return
target = os.path.join(self.library.dir, 'studies', folder['path'])
entries = []
for row in db.execute('''SELECT c.name FROM collection_folders f JOIN collections c
ON c.id=f.collection_id WHERE f.folder_id=?''', (folder_id,)):
entries.append(dict(name=row['name'], pgn=os.path.relpath(self.library._pgn_path(row['name']), target)))
with open(os.path.join(target, 'study.json'), 'w', encoding='utf-8') as f:
json.dump(dict(name=folder['name'], category=folder['category'], collections=entries), f, indent=2)
[docs]
def categorize_folder(self, ident, category):
category=str(category).strip()[:100]
if not category:
raise ValueError('Choose a category')
with self.library._write_lock, self.library.connect() as db:
if not db.execute('UPDATE folders SET category=? WHERE id=?',(category,int(ident))).rowcount:
raise ValueError('No such folder')
self.write_manifest(int(ident))
return {'saved':True}