6 Commits

Author SHA1 Message Date
vegardbm f4ae9f6c9e import os 2026-03-18 18:44:14 +01:00
vegardbm 9e0065d849 debug 2026-03-18 18:30:47 +01:00
vegardbm b996be7c7f submissive and breedable 2026-03-18 18:09:23 +01:00
vegardbm a0da34de7e test if password is read correctly 2026-03-18 17:43:56 +01:00
vegardbm c564ee0b99 strip from password file 2026-03-18 17:21:39 +01:00
vegardbm 9bdaaf6c51 allow using stable dep versions 2026-03-18 12:17:27 +01:00
18 changed files with 182 additions and 157 deletions
+1 -4
View File
@@ -3,7 +3,7 @@ debug = true
debug_sql = false
[database]
# One of (sqlite, postgresql)
# One of (sqlite, postgres)
type = 'sqlite'
[database.sqlite]
@@ -38,6 +38,3 @@ dryrun = false
warn_days_before_borrowing_deadline = [ 5, 1 ]
days_before_queue_position_expires = 14
warn_days_before_expiring_queue_position_deadline = [ 3, 1 ]
[general]
quit_allowed = true
Generated
+4 -4
View File
@@ -7,11 +7,11 @@
]
},
"locked": {
"lastModified": 1780178524,
"narHash": "sha256-2PcNyNqbGCWBpAMdCU1HxSQmhQiG6evdjxVnPA7w5bQ=",
"lastModified": 1769338528,
"narHash": "sha256-t18ZoSt9kaI1yde26ok5s7aFLkap1Q9+/2icVh2zuaE=",
"ref": "refs/heads/main",
"rev": "2406de41ce9d0a1404cbf4e55537e3f720f37f23",
"revCount": 15,
"rev": "7218348163fd8d84df4a6f682c634793e67a3fed",
"revCount": 13,
"type": "git",
"url": "https://git.pvv.ntnu.no/Projects/libdib.git"
},
@@ -1,22 +0,0 @@
# base fetcher.
from abc import ABC, abstractmethod
from .BookData import BookData
class BookDataFetcher(ABC):
"""
A base class for adapters that fetch book data from external sources.
"""
@classmethod
@abstractmethod
def fetcher_id(cls) -> str:
"""Returns a unique identifier for this specific fetcher, to identify where the data came from."""
pass
@classmethod
@abstractmethod
def try_fetch_data(cls, isbn: str) -> BookData | None:
"""Tries to fetch data for the given ISBN."""
pass
@@ -1,3 +0,0 @@
from .book_data_fetcher import fetch_book_data_from_multiple_sources
__all__ = ["fetch_book_data_from_multiple_sources"]
@@ -1,72 +0,0 @@
"""
this module contains the fetch_book_data_from_multiple_sources() function which combines all fetchers and returns ranked results (if any)
"""
from concurrent.futures import ThreadPoolExecutor
from worblehat.book_data_fetchers.BookData import BookData
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
from worblehat.book_data_fetchers.fetchers.GoogleBooksFetcher import GoogleBooksFetcher
from worblehat.book_data_fetchers.fetchers.OpenLibraryFetcher import OpenLibraryFetcher
from worblehat.book_data_fetchers.fetchers.OutlandScraperFetcher import (
OutlandScraperFetcher,
)
# The order of these fetchers determines the priority of the sources.
# The first fetcher in the list has the highest priority.
FETCHERS: list[BookDataFetcher] = [
OpenLibraryFetcher,
GoogleBooksFetcher,
OutlandScraperFetcher,
]
FETCHER_SOURCE_IDS: list[str] = [fetcher.fetcher_id() for fetcher in FETCHERS]
def sort_data_by_priority(data: list[BookData]) -> list[BookData]:
"""
Sorts the given data by the priority of the sources.
The order of the data is the same as the order of the sources in the FETCHERS list.
"""
# Note that this function is O(n^2) but the number of fetchers is small so it's fine.
return sorted(data, key=lambda m: FETCHER_SOURCE_IDS.index(m.source))
def fetch_book_data_from_multiple_sources(isbn: str, strict: bool=False) -> list[BookData]:
"""
Returns a list of data fetched from multiple fetchers.
Fetchers that are not able to retrieve any data for the given ISBN will be ignored.
There is no guarantee that there will be any book data.
The results are always ordered in the same way as the fetchers are listed in the FETCHERS list.
"""
isbn = isbn.replace("-", "").replace("_", "").strip().lower()
if len(isbn) != 10 and len(isbn) != 13 and not isbn.isnumeric():
raise ValueError("Invalid ISBN")
results: list[BookData] = []
with ThreadPoolExecutor() as executor:
futures = [executor.submit(fetcher.try_fetch_data, isbn) for fetcher in FETCHERS]
for future in futures:
result = future.result()
if result is not None:
results.append(result)
for result in results:
try:
result.validate()
except ValueError as e:
if strict:
raise e
print(f"Invalid data: {e}")
results.remove(result)
return sort_data_by_priority(results)
+19 -14
View File
@@ -16,7 +16,6 @@ from worblehat.models import *
from worblehat.services import (
create_bookcase_item_from_isbn,
is_valid_isbn,
Config,
)
from .subclis import (
@@ -48,13 +47,26 @@ class WorblehatCli(NumberedCmd):
self.sql_session_dirty = False
self.prompt_header = None
def run_with_safe_exit_wrapper(self) -> None:
@classmethod
def run_with_safe_exit_wrapper(cls, sql_session: Session) -> None:
tool = cls(sql_session)
while True:
try:
self.cmdloop()
except KeyboardInterrupt:
print("\n\n-----------------\n")
self.do_exit("Exit")
tool.cmdloop()
except KeyboardInterrupt:
if not tool.sql_session_dirty:
exit(0)
try:
print()
if prompt_yes_no(
"Are you sure you want to exit without saving?",
default=False,
):
raise KeyboardInterrupt
except KeyboardInterrupt:
if tool.sql_session is not None:
tool.sql_session.rollback()
exit(0)
def do_show_bookcase(self, arg: str) -> None:
bookcase_selector = InteractiveItemSelector(
@@ -63,8 +75,6 @@ class WorblehatCli(NumberedCmd):
)
bookcase_selector.cmdloop()
bookcase = bookcase_selector.result
if bookcase == None:
return
for shelf in bookcase.shelfs:
print(shelf.short_str())
@@ -128,8 +138,6 @@ class WorblehatCli(NumberedCmd):
)
bookcase_selector.cmdloop()
bookcase = bookcase_selector.result
if bookcase == None:
return
bookcase_item.shelf = select_bookcase_shelf(bookcase, self.sql_session)
@@ -144,8 +152,6 @@ class WorblehatCli(NumberedCmd):
media_type_selector.cmdloop()
bookcase_item.media_type = media_type_selector.result
if bookcase_item.media_type == None:
return
username = input("Who owns this book? [PVV]> ")
if username != "":
@@ -234,8 +240,7 @@ class WorblehatCli(NumberedCmd):
self.sql_session.commit()
else:
self.sql_session.rollback()
if Config["general.quit_allowed"]:
exit(0)
exit(0)
funcs = {
0: {
@@ -384,8 +384,6 @@ class EditBookcaseCli(NumberedCmd):
)
bookcase_selector.cmdloop()
bookcase = bookcase_selector.result
if bookcase == None:
return
assert isinstance(bookcase, Bookcase)
shelf = select_bookcase_shelf(bookcase, self.sql_session)
+1 -2
View File
@@ -65,8 +65,7 @@ def main() -> None:
if args.command == "cli":
sql_session = _connect_to_database(echo=Config["logging.debug_sql"])
worblehat = WorblehatCli(sql_session)
worblehat.run_with_safe_exit_wrapper()
WorblehatCli.run_with_safe_exit_wrapper(sql_session)
exit(0)
if args.command == "create-db":
+2 -3
View File
@@ -2,13 +2,12 @@ import isbnlib
from sqlalchemy import select
from sqlalchemy.orm import Session
from worblehat.book_data_fetchers import fetch_book_data_from_multiple_sources
from ..models import (
Author,
BookcaseItem,
Language,
)
from .metadata_fetchers import fetch_metadata_from_multiple_sources
def is_valid_pvv_isbn(isbn: str) -> bool:
@@ -42,7 +41,7 @@ def create_bookcase_item_from_isbn(
Please not that the returned BookcaseItem will likely not be fully populated with the required
data, such as the book's location in the library, and the owner of the book, etc.
"""
metadata = fetch_book_data_from_multiple_sources(isbn)
metadata = fetch_metadata_from_multiple_sources(isbn)
if len(metadata) == 0:
return None
+6 -2
View File
@@ -2,7 +2,7 @@ import tomllib
from pathlib import Path
from pprint import pformat
from typing import Any
import os
class Config:
"""
@@ -38,10 +38,14 @@ class Config:
@staticmethod
def read_password(password_field: str) -> str:
if Path(password_field).is_file():
file: Path = Path(password_field)
if file.is_file() and any([file.stat().st_mode & 0o400 and file.stat().st_uid == os.getuid(), file.stat().st_mode & 0o040 and file.stat().st_gid == os.getgid(), file.stat().st_mode & 0o004]):
with Path(password_field).open() as f:
return f.read().strip()
else:
raise RuntimeError(
f"Testing, should only use file. {password_field}",
)
return password_field
@classmethod
@@ -1,5 +1,4 @@
from dataclasses import dataclass
from typing import Any
# TODO: Add more languages
LANGUAGES: set[str] = {
@@ -20,14 +19,12 @@ LANGUAGES: set[str] = {
@dataclass
class BookData:
"""
A class representing metadata for a book that we might want to fetch from external sources
"""
class BookMetadata:
"""A class representing metadata for a book."""
isbn: str
title: str
# ID of the data fetcher used to fetch this instance
# The source of the metadata provider
source: str
authors: set[str]
language: str | None
@@ -35,11 +32,11 @@ class BookData:
num_pages: int | None
subjects: set[str]
def to_dict(self) -> dict[str, Any]:
def to_dict(self) -> dict[str, any]:
return {
"isbn": self.isbn,
"title": self.title,
"source": self.source,
"source": self.metadata_source_id(),
"authors": set() if self.authors is None else self.authors,
"language": self.language,
"publish_date": self.publish_date,
@@ -0,0 +1,22 @@
# base fetcher.
from abc import ABC, abstractmethod
from .BookMetadata import BookMetadata
class BookMetadataFetcher(ABC):
"""
A base class for metadata fetchers.
"""
@classmethod
@abstractmethod
def metadata_source_id(cls) -> str:
"""Returns a unique identifier for the metadata source, to identify where the metadata came from."""
pass
@classmethod
@abstractmethod
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
"""Tries to fetch metadata for the given ISBN."""
pass
@@ -4,17 +4,17 @@ A BookMetadataFetcher for the Google Books API.
import requests
from worblehat.book_data_fetchers.BookData import BookData
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
class GoogleBooksFetcher(BookDataFetcher):
class GoogleBooksFetcher(BookMetadataFetcher):
@classmethod
def fetcher_id(_cls) -> str:
def metadata_source_id(_cls) -> str:
return "google_books"
@classmethod
def try_fetch_data(cls, isbn: str) -> BookData | None:
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
try:
jsonInput = requests.get(
"https://www.googleapis.com/books/v1/volumes",
@@ -33,13 +33,19 @@ class GoogleBooksFetcher(BookDataFetcher):
except Exception:
return None
return BookData(
return BookMetadata(
isbn=isbn,
title=title,
source=cls.fetcher_id(),
source=cls.metadata_source_id(),
authors=authors,
language=languages,
publish_date=publishDate,
num_pages=numberOfPages,
subjects=subjects,
)
if __name__ == "__main__":
book_data = GoogleBooksFetcher.fetch_metadata("0132624788")
book_data.validate()
print(book_data)
@@ -4,21 +4,21 @@ A BookMetadataFetcher for the Open Library API.
import requests
from worblehat.book_data_fetchers.BookData import BookData
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
LANGUAGE_MAP = {
"Norwegian": "no",
}
class OpenLibraryFetcher(BookDataFetcher):
class OpenLibraryFetcher(BookMetadataFetcher):
@classmethod
def fetcher_id(_cls) -> str:
def metadata_source_id(_cls) -> str:
return "open_library"
@classmethod
def try_fetch_data(cls, isbn: str) -> BookData | None:
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
try:
jsonInput = requests.get(f"https://openlibrary.org/isbn/{isbn}.json").json()
@@ -48,13 +48,19 @@ class OpenLibraryFetcher(BookDataFetcher):
except Exception:
return None
return BookData(
return BookMetadata(
isbn=isbn,
title=title,
source=cls.fetcher_id(),
source=cls.metadata_source_id(),
authors=author_names,
language=language,
publish_date=publishDate,
num_pages=numberOfPages,
subjects=subjects,
)
if __name__ == "__main__":
book_data = OpenLibraryFetcher.fetch_metadata("9788205530751")
book_data.validate()
print(book_data)
@@ -5,8 +5,8 @@ A BookMetadataFetcher that webscrapes https://outland.no/
import requests
from bs4 import BeautifulSoup
from worblehat.book_data_fetchers.BookData import BookData
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
LANGUAGE_MAP = {
"Norsk": "no",
@@ -25,13 +25,13 @@ LANGUAGE_MAP = {
}
class OutlandScraperFetcher(BookDataFetcher):
class OutlandScraperFetcher(BookMetadataFetcher):
@classmethod
def fetcher_id(_cls) -> str:
def metadata_source_id(_cls) -> str:
return "outland_scraper"
@classmethod
def try_fetch_data(cls, isbn: str) -> BookData | None:
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
try:
# Find the link to the product page
response = requests.get(f"https://outland.no/{isbn}")
@@ -89,13 +89,19 @@ class OutlandScraperFetcher(BookDataFetcher):
except Exception:
return None
return BookData(
return BookMetadata(
isbn=isbn,
title=bookData.get("Title"),
source=cls.fetcher_id(),
source=cls.metadata_source_id(),
authors=bookData.get("Authors"),
language=bookData.get("Language"),
publish_date=bookData.get("PublishDate"),
num_pages=bookData.get("NumberOfPages"),
subjects=bookData.get("Subjects"),
)
if __name__ == "__main__":
book_data = OutlandScraperFetcher.fetch_metadata("9781947808225")
book_data.validate()
print(book_data)
@@ -0,0 +1,3 @@
from .book_metadata_fetcher import fetch_metadata_from_multiple_sources
__all__ = ["fetch_metadata_from_multiple_sources"]
@@ -0,0 +1,80 @@
"""
this module contains the fetch_book_metadata() function which fetches book metadata from multiple sources in threads and returns the higest ranked non-None result.
"""
from concurrent.futures import ThreadPoolExecutor
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
from worblehat.services.metadata_fetchers.GoogleBooksFetcher import GoogleBooksFetcher
from worblehat.services.metadata_fetchers.OpenLibraryFetcher import OpenLibraryFetcher
from worblehat.services.metadata_fetchers.OutlandScraperFetcher import (
OutlandScraperFetcher,
)
# The order of these fetchers determines the priority of the sources.
# The first fetcher in the list has the highest priority.
FETCHERS: list[BookMetadataFetcher] = [
OpenLibraryFetcher,
GoogleBooksFetcher,
OutlandScraperFetcher,
]
FETCHER_SOURCE_IDS: list[str] = [fetcher.metadata_source_id() for fetcher in FETCHERS]
def sort_metadata_by_priority(metadata: list[BookMetadata]) -> list[BookMetadata]:
"""
Sorts the given metadata by the priority of the sources.
The order of the metadata is the same as the order of the sources in the FETCHERS list.
"""
# Note that this function is O(n^2) but the number of fetchers is small so it's fine.
return sorted(metadata, key=lambda m: FETCHER_SOURCE_IDS.index(m.source))
def fetch_metadata_from_multiple_sources(isbn: str, strict: bool=False) -> list[BookMetadata]:
"""
Returns a list of metadata fetched from multiple sources.
Sources that does not have metadata for the given ISBN will be ignored.
There is no guarantee that there will be any metadata.
The results are always ordered in the same way as the fetchers are listed in the FETCHERS list.
"""
isbn = isbn.replace("-", "").replace("_", "").strip().lower()
if len(isbn) != 10 and len(isbn) != 13 and not isbn.isnumeric():
raise ValueError("Invalid ISBN")
results: list[BookMetadata] = []
with ThreadPoolExecutor() as executor:
futures = [executor.submit(fetcher.fetch_metadata, isbn) for fetcher in FETCHERS]
for future in futures:
result = future.result()
if result is not None:
results.append(result)
for result in results:
try:
result.validate()
except ValueError as e:
if strict:
raise e
print(f"Invalid metadata: {e}")
results.remove(result)
return sort_metadata_by_priority(results)
if __name__ == "__main__":
from pprint import pprint
isbn = "0132624788"
metadata = fetch_metadata_from_multiple_sources(isbn)
pprint(metadata)