wallabag-tools/wallabag_mass_import.py

import asyncio
import configparser
import logging
import json
import os
from typing import Any, Dict, List

import aiohttp
from wallabag_api.wallabag import Wallabag

logging.basicConfig(level=logging.DEBUG)

ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(ROOT_DIR, "config.ini")
ARTICLES_PATH = os.path.join(ROOT_DIR, "articles.json")


def load_configuration(path):
    config = configparser.ConfigParser()
    config.read(path)
    return config["Wallabag Configuration"]


async def connect_to_wallabag(
    configuration: Dict[str, str], session: aiohttp.ClientSession
) -> Wallabag:
    configuration["access_token"] = await Wallabag.get_token(**configuration)

    return Wallabag(
        host=configuration["host"],
        token=configuration["access_token"],
        client_id=configuration["client_id"],
        client_secret=configuration["client_secret"],
        aio_sess=session,
    )


async def post_entry(w_api: Wallabag, item: Dict[str, str]):
    entry = await w_api.post_entries(
        url=item["url"],
        archive=item["is_archived"],
        starred=item["is_starred"],
        original_url=item["origin_url"],
        tags=",".join(item["tags"]),
    )
    logging.info(f"Entry url {item['url']} posted to wallabag with id {entry['id']}")
    return entry["id"]


async def delete_entry(w_api: Wallabag, item_id: int):
    entry = await w_api.delete_entries(item_id)
    logging.info(f"Deleted wallabag entry with id {entry['id']}")


async def import_articles(w_api: Wallabag, path: str, limit=0):
    entries_id = []
    tasks = []
    with open(path) as f:
        for index, item in enumerate(json.load(f)):
            if limit > 0 and index == limit:
                break
            task = asyncio.ensure_future(post_entry(w_api, item))
            tasks.append(task)

    entries_id = await asyncio.gather(*tasks)
    return entries_id


async def delete_all_entries(ids: List[int], w_api: Wallabag):
    tasks = []
    for id in ids:
        tasks.append(delete_entry(w_api, id))
    await asyncio.gather(*tasks)


async def main():
    async with aiohttp.ClientSession(loop=asyncio.get_event_loop()) as session:
        configuration = dict(load_configuration(CONFIG_PATH))
        w_api = await connect_to_wallabag(configuration, session)

        entries_id = await import_articles(w_api, ARTICLES_PATH, limit=4)
        await delete_all_entries(entries_id, w_api)


if __name__ == "__main__":
    asyncio.run(main())