Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f4ae9f6c9e | |||
| 9e0065d849 | |||
| b996be7c7f | |||
| a0da34de7e | |||
| c564ee0b99 | |||
| 9bdaaf6c51 |
@@ -3,7 +3,7 @@ debug = true
|
|||||||
debug_sql = false
|
debug_sql = false
|
||||||
|
|
||||||
[database]
|
[database]
|
||||||
# One of (sqlite, postgresql)
|
# One of (sqlite, postgres)
|
||||||
type = 'sqlite'
|
type = 'sqlite'
|
||||||
|
|
||||||
[database.sqlite]
|
[database.sqlite]
|
||||||
@@ -38,6 +38,3 @@ dryrun = false
|
|||||||
warn_days_before_borrowing_deadline = [ 5, 1 ]
|
warn_days_before_borrowing_deadline = [ 5, 1 ]
|
||||||
days_before_queue_position_expires = 14
|
days_before_queue_position_expires = 14
|
||||||
warn_days_before_expiring_queue_position_deadline = [ 3, 1 ]
|
warn_days_before_expiring_queue_position_deadline = [ 3, 1 ]
|
||||||
|
|
||||||
[general]
|
|
||||||
quit_allowed = true
|
|
||||||
|
|||||||
Generated
+4
-4
@@ -7,11 +7,11 @@
|
|||||||
]
|
]
|
||||||
},
|
},
|
||||||
"locked": {
|
"locked": {
|
||||||
"lastModified": 1780178524,
|
"lastModified": 1769338528,
|
||||||
"narHash": "sha256-2PcNyNqbGCWBpAMdCU1HxSQmhQiG6evdjxVnPA7w5bQ=",
|
"narHash": "sha256-t18ZoSt9kaI1yde26ok5s7aFLkap1Q9+/2icVh2zuaE=",
|
||||||
"ref": "refs/heads/main",
|
"ref": "refs/heads/main",
|
||||||
"rev": "2406de41ce9d0a1404cbf4e55537e3f720f37f23",
|
"rev": "7218348163fd8d84df4a6f682c634793e67a3fed",
|
||||||
"revCount": 15,
|
"revCount": 13,
|
||||||
"type": "git",
|
"type": "git",
|
||||||
"url": "https://git.pvv.ntnu.no/Projects/libdib.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)
|
|
||||||
+17
-12
@@ -16,7 +16,6 @@ from worblehat.models import *
|
|||||||
from worblehat.services import (
|
from worblehat.services import (
|
||||||
create_bookcase_item_from_isbn,
|
create_bookcase_item_from_isbn,
|
||||||
is_valid_isbn,
|
is_valid_isbn,
|
||||||
Config,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
from .subclis import (
|
from .subclis import (
|
||||||
@@ -48,13 +47,26 @@ class WorblehatCli(NumberedCmd):
|
|||||||
self.sql_session_dirty = False
|
self.sql_session_dirty = False
|
||||||
self.prompt_header = None
|
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:
|
while True:
|
||||||
try:
|
try:
|
||||||
self.cmdloop()
|
tool.cmdloop()
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
print("\n\n-----------------\n")
|
if not tool.sql_session_dirty:
|
||||||
self.do_exit("Exit")
|
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:
|
def do_show_bookcase(self, arg: str) -> None:
|
||||||
bookcase_selector = InteractiveItemSelector(
|
bookcase_selector = InteractiveItemSelector(
|
||||||
@@ -63,8 +75,6 @@ class WorblehatCli(NumberedCmd):
|
|||||||
)
|
)
|
||||||
bookcase_selector.cmdloop()
|
bookcase_selector.cmdloop()
|
||||||
bookcase = bookcase_selector.result
|
bookcase = bookcase_selector.result
|
||||||
if bookcase == None:
|
|
||||||
return
|
|
||||||
|
|
||||||
for shelf in bookcase.shelfs:
|
for shelf in bookcase.shelfs:
|
||||||
print(shelf.short_str())
|
print(shelf.short_str())
|
||||||
@@ -128,8 +138,6 @@ class WorblehatCli(NumberedCmd):
|
|||||||
)
|
)
|
||||||
bookcase_selector.cmdloop()
|
bookcase_selector.cmdloop()
|
||||||
bookcase = bookcase_selector.result
|
bookcase = bookcase_selector.result
|
||||||
if bookcase == None:
|
|
||||||
return
|
|
||||||
|
|
||||||
bookcase_item.shelf = select_bookcase_shelf(bookcase, self.sql_session)
|
bookcase_item.shelf = select_bookcase_shelf(bookcase, self.sql_session)
|
||||||
|
|
||||||
@@ -144,8 +152,6 @@ class WorblehatCli(NumberedCmd):
|
|||||||
|
|
||||||
media_type_selector.cmdloop()
|
media_type_selector.cmdloop()
|
||||||
bookcase_item.media_type = media_type_selector.result
|
bookcase_item.media_type = media_type_selector.result
|
||||||
if bookcase_item.media_type == None:
|
|
||||||
return
|
|
||||||
|
|
||||||
username = input("Who owns this book? [PVV]> ")
|
username = input("Who owns this book? [PVV]> ")
|
||||||
if username != "":
|
if username != "":
|
||||||
@@ -234,7 +240,6 @@ class WorblehatCli(NumberedCmd):
|
|||||||
self.sql_session.commit()
|
self.sql_session.commit()
|
||||||
else:
|
else:
|
||||||
self.sql_session.rollback()
|
self.sql_session.rollback()
|
||||||
if Config["general.quit_allowed"]:
|
|
||||||
exit(0)
|
exit(0)
|
||||||
|
|
||||||
funcs = {
|
funcs = {
|
||||||
|
|||||||
@@ -384,8 +384,6 @@ class EditBookcaseCli(NumberedCmd):
|
|||||||
)
|
)
|
||||||
bookcase_selector.cmdloop()
|
bookcase_selector.cmdloop()
|
||||||
bookcase = bookcase_selector.result
|
bookcase = bookcase_selector.result
|
||||||
if bookcase == None:
|
|
||||||
return
|
|
||||||
assert isinstance(bookcase, Bookcase)
|
assert isinstance(bookcase, Bookcase)
|
||||||
|
|
||||||
shelf = select_bookcase_shelf(bookcase, self.sql_session)
|
shelf = select_bookcase_shelf(bookcase, self.sql_session)
|
||||||
|
|||||||
@@ -65,8 +65,7 @@ def main() -> None:
|
|||||||
|
|
||||||
if args.command == "cli":
|
if args.command == "cli":
|
||||||
sql_session = _connect_to_database(echo=Config["logging.debug_sql"])
|
sql_session = _connect_to_database(echo=Config["logging.debug_sql"])
|
||||||
worblehat = WorblehatCli(sql_session)
|
WorblehatCli.run_with_safe_exit_wrapper(sql_session)
|
||||||
worblehat.run_with_safe_exit_wrapper()
|
|
||||||
exit(0)
|
exit(0)
|
||||||
|
|
||||||
if args.command == "create-db":
|
if args.command == "create-db":
|
||||||
|
|||||||
@@ -2,13 +2,12 @@ import isbnlib
|
|||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
from worblehat.book_data_fetchers import fetch_book_data_from_multiple_sources
|
|
||||||
|
|
||||||
from ..models import (
|
from ..models import (
|
||||||
Author,
|
Author,
|
||||||
BookcaseItem,
|
BookcaseItem,
|
||||||
Language,
|
Language,
|
||||||
)
|
)
|
||||||
|
from .metadata_fetchers import fetch_metadata_from_multiple_sources
|
||||||
|
|
||||||
|
|
||||||
def is_valid_pvv_isbn(isbn: str) -> bool:
|
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
|
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.
|
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:
|
if len(metadata) == 0:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import tomllib
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from pprint import pformat
|
from pprint import pformat
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
import os
|
||||||
|
|
||||||
class Config:
|
class Config:
|
||||||
"""
|
"""
|
||||||
@@ -38,10 +38,14 @@ class Config:
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def read_password(password_field: str) -> str:
|
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:
|
with Path(password_field).open() as f:
|
||||||
return f.read().strip()
|
return f.read().strip()
|
||||||
else:
|
else:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Testing, should only use file. {password_field}",
|
||||||
|
)
|
||||||
return password_field
|
return password_field
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
+5
-8
@@ -1,5 +1,4 @@
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
# TODO: Add more languages
|
# TODO: Add more languages
|
||||||
LANGUAGES: set[str] = {
|
LANGUAGES: set[str] = {
|
||||||
@@ -20,14 +19,12 @@ LANGUAGES: set[str] = {
|
|||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class BookData:
|
class BookMetadata:
|
||||||
"""
|
"""A class representing metadata for a book."""
|
||||||
A class representing metadata for a book that we might want to fetch from external sources
|
|
||||||
"""
|
|
||||||
|
|
||||||
isbn: str
|
isbn: str
|
||||||
title: str
|
title: str
|
||||||
# ID of the data fetcher used to fetch this instance
|
# The source of the metadata provider
|
||||||
source: str
|
source: str
|
||||||
authors: set[str]
|
authors: set[str]
|
||||||
language: str | None
|
language: str | None
|
||||||
@@ -35,11 +32,11 @@ class BookData:
|
|||||||
num_pages: int | None
|
num_pages: int | None
|
||||||
subjects: set[str]
|
subjects: set[str]
|
||||||
|
|
||||||
def to_dict(self) -> dict[str, Any]:
|
def to_dict(self) -> dict[str, any]:
|
||||||
return {
|
return {
|
||||||
"isbn": self.isbn,
|
"isbn": self.isbn,
|
||||||
"title": self.title,
|
"title": self.title,
|
||||||
"source": self.source,
|
"source": self.metadata_source_id(),
|
||||||
"authors": set() if self.authors is None else self.authors,
|
"authors": set() if self.authors is None else self.authors,
|
||||||
"language": self.language,
|
"language": self.language,
|
||||||
"publish_date": self.publish_date,
|
"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
|
||||||
+13
-7
@@ -4,17 +4,17 @@ A BookMetadataFetcher for the Google Books API.
|
|||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
from worblehat.book_data_fetchers.BookData import BookData
|
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
|
||||||
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
|
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
|
||||||
|
|
||||||
|
|
||||||
class GoogleBooksFetcher(BookDataFetcher):
|
class GoogleBooksFetcher(BookMetadataFetcher):
|
||||||
@classmethod
|
@classmethod
|
||||||
def fetcher_id(_cls) -> str:
|
def metadata_source_id(_cls) -> str:
|
||||||
return "google_books"
|
return "google_books"
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def try_fetch_data(cls, isbn: str) -> BookData | None:
|
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
|
||||||
try:
|
try:
|
||||||
jsonInput = requests.get(
|
jsonInput = requests.get(
|
||||||
"https://www.googleapis.com/books/v1/volumes",
|
"https://www.googleapis.com/books/v1/volumes",
|
||||||
@@ -33,13 +33,19 @@ class GoogleBooksFetcher(BookDataFetcher):
|
|||||||
except Exception:
|
except Exception:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return BookData(
|
return BookMetadata(
|
||||||
isbn=isbn,
|
isbn=isbn,
|
||||||
title=title,
|
title=title,
|
||||||
source=cls.fetcher_id(),
|
source=cls.metadata_source_id(),
|
||||||
authors=authors,
|
authors=authors,
|
||||||
language=languages,
|
language=languages,
|
||||||
publish_date=publishDate,
|
publish_date=publishDate,
|
||||||
num_pages=numberOfPages,
|
num_pages=numberOfPages,
|
||||||
subjects=subjects,
|
subjects=subjects,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
book_data = GoogleBooksFetcher.fetch_metadata("0132624788")
|
||||||
|
book_data.validate()
|
||||||
|
print(book_data)
|
||||||
+13
-7
@@ -4,21 +4,21 @@ A BookMetadataFetcher for the Open Library API.
|
|||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
from worblehat.book_data_fetchers.BookData import BookData
|
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
|
||||||
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
|
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
|
||||||
|
|
||||||
LANGUAGE_MAP = {
|
LANGUAGE_MAP = {
|
||||||
"Norwegian": "no",
|
"Norwegian": "no",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
class OpenLibraryFetcher(BookDataFetcher):
|
class OpenLibraryFetcher(BookMetadataFetcher):
|
||||||
@classmethod
|
@classmethod
|
||||||
def fetcher_id(_cls) -> str:
|
def metadata_source_id(_cls) -> str:
|
||||||
return "open_library"
|
return "open_library"
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def try_fetch_data(cls, isbn: str) -> BookData | None:
|
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
|
||||||
try:
|
try:
|
||||||
jsonInput = requests.get(f"https://openlibrary.org/isbn/{isbn}.json").json()
|
jsonInput = requests.get(f"https://openlibrary.org/isbn/{isbn}.json").json()
|
||||||
|
|
||||||
@@ -48,13 +48,19 @@ class OpenLibraryFetcher(BookDataFetcher):
|
|||||||
except Exception:
|
except Exception:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return BookData(
|
return BookMetadata(
|
||||||
isbn=isbn,
|
isbn=isbn,
|
||||||
title=title,
|
title=title,
|
||||||
source=cls.fetcher_id(),
|
source=cls.metadata_source_id(),
|
||||||
authors=author_names,
|
authors=author_names,
|
||||||
language=language,
|
language=language,
|
||||||
publish_date=publishDate,
|
publish_date=publishDate,
|
||||||
num_pages=numberOfPages,
|
num_pages=numberOfPages,
|
||||||
subjects=subjects,
|
subjects=subjects,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
book_data = OpenLibraryFetcher.fetch_metadata("9788205530751")
|
||||||
|
book_data.validate()
|
||||||
|
print(book_data)
|
||||||
+13
-7
@@ -5,8 +5,8 @@ A BookMetadataFetcher that webscrapes https://outland.no/
|
|||||||
import requests
|
import requests
|
||||||
from bs4 import BeautifulSoup
|
from bs4 import BeautifulSoup
|
||||||
|
|
||||||
from worblehat.book_data_fetchers.BookData import BookData
|
from worblehat.services.metadata_fetchers.BookMetadata import BookMetadata
|
||||||
from worblehat.book_data_fetchers.BookDataFetcher import BookDataFetcher
|
from worblehat.services.metadata_fetchers.BookMetadataFetcher import BookMetadataFetcher
|
||||||
|
|
||||||
LANGUAGE_MAP = {
|
LANGUAGE_MAP = {
|
||||||
"Norsk": "no",
|
"Norsk": "no",
|
||||||
@@ -25,13 +25,13 @@ LANGUAGE_MAP = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
class OutlandScraperFetcher(BookDataFetcher):
|
class OutlandScraperFetcher(BookMetadataFetcher):
|
||||||
@classmethod
|
@classmethod
|
||||||
def fetcher_id(_cls) -> str:
|
def metadata_source_id(_cls) -> str:
|
||||||
return "outland_scraper"
|
return "outland_scraper"
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def try_fetch_data(cls, isbn: str) -> BookData | None:
|
def fetch_metadata(cls, isbn: str) -> BookMetadata | None:
|
||||||
try:
|
try:
|
||||||
# Find the link to the product page
|
# Find the link to the product page
|
||||||
response = requests.get(f"https://outland.no/{isbn}")
|
response = requests.get(f"https://outland.no/{isbn}")
|
||||||
@@ -89,13 +89,19 @@ class OutlandScraperFetcher(BookDataFetcher):
|
|||||||
except Exception:
|
except Exception:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return BookData(
|
return BookMetadata(
|
||||||
isbn=isbn,
|
isbn=isbn,
|
||||||
title=bookData.get("Title"),
|
title=bookData.get("Title"),
|
||||||
source=cls.fetcher_id(),
|
source=cls.metadata_source_id(),
|
||||||
authors=bookData.get("Authors"),
|
authors=bookData.get("Authors"),
|
||||||
language=bookData.get("Language"),
|
language=bookData.get("Language"),
|
||||||
publish_date=bookData.get("PublishDate"),
|
publish_date=bookData.get("PublishDate"),
|
||||||
num_pages=bookData.get("NumberOfPages"),
|
num_pages=bookData.get("NumberOfPages"),
|
||||||
subjects=bookData.get("Subjects"),
|
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)
|
||||||
Reference in New Issue
Block a user