86 lines
2.3 KiB
Python
86 lines
2.3 KiB
Python
|
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,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def read_urls_file(f):
|
||
|
|
while True:
|
||
|
|
line = f.readline()
|
||
|
|
if line:
|
||
|
|
yield line
|
||
|
|
else:
|
||
|
|
break
|
||
|
|
|
||
|
|
|
||
|
|
async def import_articles(w_api: Wallabag, path: str, limit=0):
|
||
|
|
entries_id = []
|
||
|
|
with open(path) as f:
|
||
|
|
for index, item in enumerate(json.load(f)):
|
||
|
|
if limit > 0 and index == limit:
|
||
|
|
break
|
||
|
|
|
||
|
|
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']}"
|
||
|
|
)
|
||
|
|
entries_id.append(entry["id"])
|
||
|
|
|
||
|
|
return entries_id
|
||
|
|
|
||
|
|
|
||
|
|
async def delete_all_entries(ids: List[int], w_api: Wallabag):
|
||
|
|
for id in ids:
|
||
|
|
entry = await w_api.delete_entries(id)
|
||
|
|
logging.info(f"Deleted wallabag entry with id {entry['id']}")
|
||
|
|
|
||
|
|
|
||
|
|
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.json", limit=1)
|
||
|
|
await delete_all_entries(entries_id, w_api)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|