import sqlite3 from profile import TinderProfile # See 'row_factory' in https://docs.python.org/3/library/sqlite3.html#connection-objects def dict_factory(cursor, row): d = {} for idx, col in enumerate(cursor.description): d[col[0]] = row[idx] return d class HumanKind(object): def __init__(self): """Create a database for storing Tinder profiles.""" self._db = sqlite3.connect('profiles.db', detect_types=sqlite3.PARSE_DECLTYPES) sqlite3.register_adapter(bool, int) sqlite3.register_converter("BOOLEAN", lambda v: bool(int(v))) self._db.row_factory = dict_factory sql = ''' CREATE TABLE IF NOT EXISTS People( uid TEXT PRIMARY KEY, name TEXT, gender INTEGER, birth_date TEXT, ping_time TEXT, school TEXT, job_title TEXT, job_company TEXT, distance_mi INTEGER, bio TEXT, liked BOOLEAN, superliked BOOLEAN, passed BOOLEAN, extracted BOOLEAN ) ''' cursor = self._db.cursor() cursor.execute(sql) self._db.commit() sql = ''' CREATE TABLE IF NOT EXISTS Photos( uid TEXT, url TEXT, main BOOLEAN ) ''' cursor = self._db.cursor() cursor.execute(sql) self._db.commit() def has_profile_id(self, profile): """Tell is a given Tinder profile is known in the database.""" values = (profile._id, ) cursor = self._db.cursor() cursor.execute('SELECT uid FROM People WHERE uid = ?', values) found = cursor.fetchone() return not(found is None) def store_profile(self, profile): """Store a Tinder profile into the database.""" exist = self.has_profile_id(profile) if exist: fields = 'name = ?, gender = ?, birth_date = ?, ping_time = ?, school = ?,' \ ' job_title = ?, job_company = ?, distance_mi = ?, bio = ?' values = (profile._name, profile._gender, profile._birth_date, profile._ping_time, profile._school, profile._job_title, profile._job_company, profile._distance_mi, profile._bio, profile._id) cursor = self._db.cursor() cursor.execute('UPDATE People SET ' + fields + ' WHERE uid = ?', values) else: values = (profile._id, profile._name, profile._gender, profile._birth_date, profile._ping_time, profile._school, profile._job_title, profile._job_company, profile._distance_mi, profile._bio, False, False, False, False) cursor = self._db.cursor() cursor.execute('INSERT INTO People VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', values) for p in profile._photos: values = (profile._id, p['url']) cursor = self._db.cursor() cursor.execute('SELECT uid FROM Photos WHERE uid = ? AND url = ?', values) found = cursor.fetchone() if found is None: values = (profile._id, p['url'], p['main']) cursor = self._db.cursor() cursor.execute('INSERT INTO Photos VALUES (?, ?, ?)', values) self._db.commit() def load_profile(self, uid): """Load a Tinder profile from the database.""" values = (uid, ) cursor = self._db.cursor() cursor.execute('SELECT * FROM People WHERE uid = ?', values) found = cursor.fetchone() if not(found): profile = None else: data = {} data['jobs'] = {} data['schools'] = {} data['photos'] = [] data['ping_time'] = found['ping_time'] if found['job_title']: data['jobs']['title'] = found['job_title'] if found['job_company']: data['jobs']['company'] = found['job_company'] if found['school']: data['schools']['name'] = found['school'] data['gender'] = found['gender'] data['distance_mi'] = found['distance_mi'] data['_id'] = found['uid'] if found['bio'] is None: data['bio'] = '' else: data['bio'] = found['bio'] data['name'] = found['name'] data['birth_date'] = found['birth_date'] values = (uid, ) cursor = self._db.cursor() cursor.execute('SELECT url, main FROM Photos WHERE uid = ?', values) for found in cursor.fetchall(): photo = { 'url': found['url'], 'width': 1, 'height': 1 } rec = {} rec['processedFiles'] = [ photo ] if not(found['main']): rec['main'] = False data['photos'].append(rec) profile = TinderProfile(data) return profile def load_all_profiles(self): """Load all stored Tinder profiles.""" new = [] cursor = self._db.cursor() cursor.execute('SELECT uid FROM People') for found in cursor.fetchall(): profile = self.load_profile(found['uid']) new.append(profile) return new def load_unknown_profiles(self): """Load all new Tinder profiles.""" new = [] values = (False, False, False) cursor = self._db.cursor() cursor.execute('SELECT uid FROM People WHERE liked = ? AND passed = ? AND superliked = ?', values) for found in cursor.fetchall(): profile = self.load_profile(found['uid']) new.append(profile) return new def load_new_profiles(self): """Load all new Tinder profiles.""" new = [] values = (False, ) cursor = self._db.cursor() cursor.execute('SELECT uid FROM People WHERE extracted = ?', values) for found in cursor.fetchall(): profile = self.load_profile(found['uid']) new.append(profile) return new def mark_profile_as_liked(self, profile): """Update information about a Tinder profile in the database.""" values = (True, profile._id) cursor = self._db.cursor() cursor.execute('UPDATE People SET liked = ? WHERE uid = ?', values) self._db.commit() def mark_profile_as_superliked(self, profile): """Update information about a Tinder profile in the database.""" values = (True, profile._id) cursor = self._db.cursor() cursor.execute('UPDATE People SET superliked = ? WHERE uid = ?', values) self._db.commit() def mark_profile_as_passed(self, profile): """Update information about a Tinder profile in the database.""" values = (True, profile._id) cursor = self._db.cursor() cursor.execute('UPDATE People SET passed = ? WHERE uid = ?', values) self._db.commit() def mark_profile_as_extracted(self, profile): """Update information about a Tinder profile in the database.""" values = (True, profile._id) cursor = self._db.cursor() cursor.execute('UPDATE People SET extracted = ? WHERE uid = ?', values) self._db.commit()