Source code for frostmark.importer
'''
Import bookmarks from various bookmarks database files into internal database.
'''
from ensure import ensure_annotations
from sqlalchemy import (
create_engine, Column, Integer, String
)
from sqlalchemy.orm.session import sessionmaker, Session
from sqlalchemy.ext.declarative.api import DeclarativeMeta as MetaBase
from frostmark.db import get_session
from frostmark.common import traverse
from frostmark.models import Folder, Bookmark
from frostmark.importer.firefox import FirefoxImporter
from frostmark.importer.opera import OperaImporter
from frostmark.importer.chrome import ChromeImporter
[docs]class Importer:
'''
Interface for retrieving and importing bookmarks into local storage
for multiple backends.
'''
# pylint: disable=too-few-public-methods
@ensure_annotations
def __init__(self, backend: str):
self.backend = None
if backend == 'firefox':
self.backend = FirefoxImporter()
elif backend == 'opera':
self.backend = OperaImporter()
elif backend == 'chrome':
self.backend = ChromeImporter()
@staticmethod
@ensure_annotations
def _path_session(path: str, base: MetaBase) -> Session:
'''
Create a SQLAlchemy session over a SQLite database with custom
MetaBase that connects the Python models to the engine and session.
'''
engine = create_engine(f'sqlite:///{path}')
base.prepare(engine)
maker = sessionmaker(bind=engine)
session = maker()
return session
@ensure_annotations
def import_from(self, path: str):
'''
Import bookmarks from particular path into internal storage.
'''
# pylint: disable=too-many-locals
backend = self.backend
# get the bookmarks tree from file
if isinstance(backend, FirefoxImporter):
source = self._path_session(path=path, base=backend.BASE)
elif isinstance(backend, OperaImporter):
source = path
elif isinstance(backend, ChromeImporter):
source = path
tree = backend.assemble_import_tree(source)
# open internal DB
nodes = traverse(tree)
folders = {
node.id: node
for node in nodes
if node.node_type == Folder
}
bookmarks = {
node.id: node
for node in nodes
if node.node_type == Bookmark
}
# sqla objects
frost = get_session()
sqla_folders = {}
# add folder structure to the database
sorted_folders = sorted(
folders.values(),
# parent_folder_id is None for root folder
key=lambda item: item.parent_folder_id or 0
)
# first sort the tree by parent IDs so that there is each parent
# available, however in case there is a kind-of circular relationship
# between the folders e.g. the ID of a child is smaller than ID of
# a parent which might be caused by browser importing old bookmarks
# from database directly or from a different browser while incorrectly
# setting IDs (or better said re-using already existing IDs when
# possible), therefore sorting by parent ID would work, but when trying
# to access the parent a KeyError would be raised because of parent
# not being available yet due to higher ID than the child has
#
# for that reason try to sort with parent ID first and postpone
# the relationship evaluation by using second/third/etc/... item
# in the sorted list until there is parent ID available (or throw
# IndexError in the end which would pretty much mean that the browser
# DB is just broken due to missing parent / dangling children)
idx = 0
while sorted_folders:
folder = sorted_folders[idx]
kwargs = {
'folder_name': folder.folder_name
}
if folder.parent_folder_id:
# in case there is a parent, get the Folder object
# and pull its ID after flush() (otherwise it's None)
try:
real_id = sqla_folders[folder.parent_folder_id].id
except KeyError:
idx += 1
continue
kwargs['parent_folder_id'] = real_id
new_folder = Folder(**kwargs)
frost.add(new_folder)
# flush to obtain folder ID,
# especially necessary for nested folders
frost.flush()
# preserve the original ID and point to a SQLA object
sqla_folders[folder.id] = new_folder
# remove current folder
idx = 0
sorted_folders.remove(folder)
# add bookmarks
for key in sorted(bookmarks.keys()):
book = bookmarks[key]
kwargs = {
'title': book.title,
'url': book.url,
'icon': b'',
'folder_id': sqla_folders[book.folder_id].id
}
# no need to flush, nothing required a bookmark
# ID to be present before final commit()
new_book = Bookmark(**kwargs)
frost.add(new_book)
# write data into internal DB
frost.commit()