add links from file

kontor-scripts/add_links.py

@@ -0,0 +1,72 @@
+"""
+read file with links and store it in DB
+"""
+import logging.config
+from typing import List
+import yaml
+from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
+from pathlib import Path
+from platformdirs import PlatformDirs
+from pathlib import Path
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session
+from db.models.base import Base
+from db.models import registry
+from psycopg2.errors import NotNullViolation
+from config import get_logger
+import os
+import json
+
+from db.models.media import MediaFile
+
+parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
+parser.add_argument('--file', '-f', help='file with links', default='~/.sync/media/list.txt')
+parser.add_argument('--video', help='store Url as VideoFile', action="store_true")
+parser.add_argument('--config', '-c', default='kontor-docker')
+parser.add_argument('--verbose', '-v', action='count', default=0)
+args = parser.parse_args()
+
+DB_USER: str = os.getenv("DB_USER", "kontor")
+DB_PASSWORD: str = os.getenv("DB_PASSWORD", "kontor")
+DB_SERVER: str = os.getenv("DB_SERVER", "127.0.0.1")
+DB_PORT: int = int(os.getenv("DB_PORT", 5432))
+DB_DBNAME: str = os.getenv("DB_DBNAME", "kontor")
+DATABASE_URL: str = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_SERVER}:{DB_PORT}/{DB_DBNAME}"
+
+def load_data(filename: str, log) -> List[str]:
+    links: List[str] = []
+    log.debug("load_data")
+    import_file = Path(filename)
+    if not import_file.exists():
+        log.info(f"File {filename} does not exist. Do nothing.")
+        raise FileNotFoundError()
+    log.info("read txt file")
+    with open(filename, 'r') as txt_file:
+        while line := txt_file.readline():
+            # log.info(line.rstrip())
+            links.append(line.rstrip())
+    return links
+
+
+if __name__ == '__main__':
+    logger = get_logger(args.verbose, "kontor")
+    logger.info('kontor.add_links started')
+    engine = create_engine(DATABASE_URL)
+    Base.metadata.create_all(bind=engine, checkfirst=True)
+    SessionLocal = sessionmaker(bind=engine)
+    with SessionLocal() as db:
+        links = load_data(args.file, logger)
+        for link in links:
+            logger.info(f"process {link}")
+            media_files = db.query(MediaFile).filter(MediaFile.url == link).all()
+            if len(media_files) == 0:
+                logger.info("no entry is found")
+                media_file = MediaFile()
+                media_file.url = link
+                media_file.review = True
+                media_file.should_download = True
+                db.add(media_file)
+                db.commit()
+            # else:
+            #     logger.info("entry is found")
+    logger.info('kontor.add_link finished')