|
|
import json |
|
|
import re |
|
|
from datetime import date, datetime |
|
|
from enum import IntEnum |
|
|
from pathlib import Path |
|
|
from typing import Any, List, Optional, Callable |
|
|
|
|
|
import httpx |
|
|
|
|
|
BASE_URL = "https://brestok-vika-server.hf.space" |
|
|
DOWNLOADS_DIR = Path(__file__).parent / "downloads" |
|
|
SEPARATOR = "=" * 70 |
|
|
PROMPT = "Select option: " |
|
|
|
|
|
GENRES = [ |
|
|
"Fiction", |
|
|
"Non-Fiction", |
|
|
"Mystery", |
|
|
"Sci-Fi", |
|
|
"Fantasy", |
|
|
"Biography", |
|
|
"History", |
|
|
"Romance", |
|
|
"Thriller", |
|
|
"Horror", |
|
|
"Poetry", |
|
|
"Drama", |
|
|
"Comics", |
|
|
"Other", |
|
|
] |
|
|
|
|
|
BOOK_STATUSES = ["Available", "Borrowed"] |
|
|
|
|
|
WORK_DAYS = [ |
|
|
"Monday", |
|
|
"Tuesday", |
|
|
"Wednesday", |
|
|
"Thursday", |
|
|
"Friday", |
|
|
"Saturday", |
|
|
"Sunday", |
|
|
] |
|
|
|
|
|
NAME_PATTERN = re.compile(r"^[A-Za-zА-Яа-яЁёІіЇїЄєҐґ\s\-']+$") |
|
|
TITLE_PATTERN = re.compile(r"^[A-Za-zА-Яа-яЁёІіЇїЄєҐґ0-9\s\-'.,!?:;\"()]+$") |
|
|
UUID_PATTERN = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") |
|
|
|
|
|
|
|
|
MSG_IS_REQUIRED = "is required." |
|
|
MSG_AT_LEAST_2_CHARS = "must be at least 2 characters." |
|
|
MSG_AT_MOST_50_CHARS = "must be at most 50 characters." |
|
|
MSG_ONLY_LETTERS_SPACES_HYPHENS_APOSTROPHES = "must contain only letters, spaces, hyphens, and apostrophes." |
|
|
MSG_AT_MOST_100_CHARS = "must be at most 100 characters." |
|
|
MSG_CONTAINS_INVALID_CHARS = "contains invalid characters." |
|
|
MSG_AUTHOR_REQUIRED = "Author is required." |
|
|
MSG_AUTHOR_AT_LEAST_2_CHARS = "Author must be at least 2 characters." |
|
|
MSG_AUTHOR_AT_MOST_100_CHARS = "Author must be at most 100 characters." |
|
|
MSG_AUTHOR_ONLY_LETTERS = "Author must contain only letters, spaces, hyphens, and apostrophes." |
|
|
MSG_PAGES_AT_LEAST_1 = "Pages must be at least 1." |
|
|
MSG_PAGES_AT_MOST_10000 = "Pages must be at most 10000." |
|
|
MSG_YEAR_AT_LEAST_1000 = "Year must be at least 1000." |
|
|
MSG_YEAR_CANNOT_BE_GREATER = "Year cannot be greater than {current_year}." |
|
|
MSG_GENRE_MUST_BE_ONE_OF = "Genre must be one of: {', '.join(GENRES)}" |
|
|
MSG_STATUS_MUST_BE_ONE_OF = "Status must be one of: {', '.join(BOOK_STATUSES)}" |
|
|
MSG_EXPERIENCE_NEGATIVE = "Experience cannot be negative." |
|
|
MSG_EXPERIENCE_AT_MOST_60 = "Experience must be at most 60 years." |
|
|
MSG_AT_LEAST_ONE_WORK_DAY = "At least one work day is required." |
|
|
MSG_INVALID_WORK_DAYS = "Invalid work days: {', '.join(invalid)}. Valid options: {', '.join(WORK_DAYS)}" |
|
|
MSG_MUST_BE_VALID_UUID = "must be a valid UUID format." |
|
|
MSG_DATE_YYYY_MM_DD = "Date must be in YYYY-MM-DD format." |
|
|
MSG_AT_LEAST_ONE_ITEM_REQUIRED = "At least one item is required." |
|
|
MSG_ITEM_EMPTY = "Item #{i} is empty." |
|
|
MSG_ITEM_INVALID_UUID = "Item #{i} ({id_val}) is not a valid UUID." |
|
|
|
|
|
|
|
|
MSG_ERROR_ENTER_VALID_INTEGER = " Error: Enter a valid integer." |
|
|
MSG_ERROR_INVALID_CHOICE = " Error: Invalid choice." |
|
|
MSG_ERROR_INVALID_NUMBER = " Error: Invalid number {p}." |
|
|
MSG_ERROR_INVALID_VALUE = " Error: Invalid value '{p}'." |
|
|
MSG_ERROR_AT_LEAST_ONE_VALUE = " Error: At least one value is required." |
|
|
MSG_ERROR_BOOK_NOT_FOUND = " Error: Book with ID {book_id} not found." |
|
|
MSG_ERROR_BOOK_NOT_FOUND_VALIDATED = " Error: Book with ID {validated} not found." |
|
|
MSG_ERROR_VISITOR_NOT_FOUND = " Error: Visitor with ID {validated} not found." |
|
|
MSG_ERROR_WORKER_NOT_FOUND = " Error: Worker with ID {validated} not found." |
|
|
MSG_NO_BOOKS_FOUND = "No books found." |
|
|
MSG_NO_VISITORS_FOUND = "No visitors found." |
|
|
MSG_NO_WORKERS_FOUND = "No workers found." |
|
|
MSG_ERROR_PREFIX = "Error: " |
|
|
MSG_INVALID_OPTION = "Invalid option." |
|
|
MSG_GOODBYE = "Goodbye." |
|
|
MSG_NO_FIELDS_TO_UPDATE = "No fields to update." |
|
|
MSG_DELETED = "Deleted." |
|
|
MSG_FILE_SAVED_TO = "File saved to: " |
|
|
MSG_CURRENT_BOOKS = "\nCurrent books:" |
|
|
MSG_HISTORY = "\nHistory:" |
|
|
MSG_VISITOR_DELETED = "Visitor deleted." |
|
|
MSG_WORKER_DELETED = "Worker deleted." |
|
|
MSG_INTERRUPTED_GOODBYE = "\nInterrupted. Goodbye." |
|
|
|
|
|
|
|
|
MSG_USE_COMMA_OR_STAR = "Use comma to separate values or type * for all." |
|
|
MSG_ENTER_NUMBERS_OR_NAMES = "Enter numbers or names: " |
|
|
MSG_USE_COMMA_STAR_OR_EMPTY = "Use comma to separate values, * for all, or leave empty to skip." |
|
|
MSG_ENTER_NUMBERS_OR_NAMES_EMPTY_SKIP = "Enter numbers or names (empty to skip): " |
|
|
MSG_PICK_NUMBER = "Pick number: " |
|
|
MSG_PICK_NUMBER_SKIP = "Pick number (0 to skip): " |
|
|
MSG_BOOK_ID_PROMPT = "Book ID: " |
|
|
MSG_BOOK_IDS_COMMA_SEPARATED = "Book IDs (comma separated): " |
|
|
MSG_VISITOR_ID_PROMPT = "Visitor ID: " |
|
|
MSG_WORKER_ID_PROMPT = "Worker ID: " |
|
|
MSG_TITLE_PROMPT = "Title: " |
|
|
MSG_AUTHOR_PROMPT = "Author: " |
|
|
MSG_PAGES_PROMPT = "Pages: " |
|
|
MSG_YEAR_PROMPT = "Year: " |
|
|
MSG_GENRE_PROMPT = "Genre:" |
|
|
MSG_STATUS_PROMPT = "Status:" |
|
|
MSG_NAME_PROMPT = "Name: " |
|
|
MSG_SURNAME_PROMPT = "Surname: " |
|
|
MSG_EXPERIENCE_PROMPT = "Experience (years): " |
|
|
MSG_WORK_DAYS_PROMPT = "Work days:" |
|
|
MSG_SELECT_WORK_DAYS = "Select work days:" |
|
|
MSG_TITLE_SKIP_PROMPT = "Title (empty to skip): " |
|
|
MSG_AUTHOR_SKIP_PROMPT = "Author (empty to skip): " |
|
|
MSG_PAGES_SKIP_PROMPT = "Pages (empty to skip): " |
|
|
MSG_YEAR_SKIP_PROMPT = "Year (empty to skip): " |
|
|
MSG_NAME_SKIP_PROMPT = "Name (empty to skip): " |
|
|
MSG_SURNAME_SKIP_PROMPT = "Surname (empty to skip): " |
|
|
MSG_EXPERIENCE_SKIP_PROMPT = "Experience (empty to skip): " |
|
|
MSG_LEAVE_EMPTY_TO_SKIP = "Leave fields empty to skip them." |
|
|
MSG_BORROW_DATE_PROMPT = "Borrow date (YYYY-MM-DD)" |
|
|
MSG_RETURN_DATE_PROMPT = "Return date (YYYY-MM-DD)" |
|
|
|
|
|
|
|
|
class ValidationError(Exception): |
|
|
pass |
|
|
|
|
|
|
|
|
def ensure_downloads_dir(): |
|
|
DOWNLOADS_DIR.mkdir(parents=True, exist_ok=True) |
|
|
|
|
|
|
|
|
def save_to_file(table: str, entity_id: str, content: str) -> str: |
|
|
ensure_downloads_dir() |
|
|
safe_id = entity_id.replace("/", "_").replace("\\", "_") |
|
|
filename = f"{table}-{safe_id}.json" |
|
|
filepath = DOWNLOADS_DIR / filename |
|
|
if isinstance(content, (dict, list)): |
|
|
text = json.dumps(content, ensure_ascii=False, indent=2) |
|
|
else: |
|
|
try: |
|
|
parsed = json.loads(content) |
|
|
text = json.dumps(parsed, ensure_ascii=False, indent=2) |
|
|
except Exception: |
|
|
text = content |
|
|
filepath.write_text(text, encoding="utf-8") |
|
|
return str(filepath) |
|
|
|
|
|
|
|
|
def validate_name(value: str, field_name: str = "Name") -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
raise ValidationError(f"{field_name} {MSG_IS_REQUIRED}") |
|
|
if len(value) < 2: |
|
|
raise ValidationError(f"{field_name} {MSG_AT_LEAST_2_CHARS}") |
|
|
if len(value) > 50: |
|
|
raise ValidationError(f"{field_name} {MSG_AT_MOST_50_CHARS}") |
|
|
if not NAME_PATTERN.match(value): |
|
|
raise ValidationError(f"{field_name} {MSG_ONLY_LETTERS_SPACES_HYPHENS_APOSTROPHES}") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_title(value: str, field_name: str = "Title") -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
raise ValidationError(f"{field_name} {MSG_IS_REQUIRED}") |
|
|
if len(value) < 2: |
|
|
raise ValidationError(f"{field_name} {MSG_AT_LEAST_2_CHARS}") |
|
|
if len(value) > 100: |
|
|
raise ValidationError(f"{field_name} {MSG_AT_MOST_100_CHARS}") |
|
|
if not TITLE_PATTERN.match(value): |
|
|
raise ValidationError(f"{field_name} {MSG_CONTAINS_INVALID_CHARS}") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_author(value: str) -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
raise ValidationError(MSG_AUTHOR_REQUIRED) |
|
|
if len(value) < 2: |
|
|
raise ValidationError(MSG_AUTHOR_AT_LEAST_2_CHARS) |
|
|
if len(value) > 100: |
|
|
raise ValidationError(MSG_AUTHOR_AT_MOST_100_CHARS) |
|
|
if not NAME_PATTERN.match(value): |
|
|
raise ValidationError(MSG_AUTHOR_ONLY_LETTERS) |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_pages(value: int) -> int: |
|
|
if value < 1: |
|
|
raise ValidationError(MSG_PAGES_AT_LEAST_1) |
|
|
if value > 10000: |
|
|
raise ValidationError(MSG_PAGES_AT_MOST_10000) |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_year(value: int) -> int: |
|
|
current_year = date.today().year |
|
|
if value < 1000: |
|
|
raise ValidationError(MSG_YEAR_AT_LEAST_1000) |
|
|
if value > current_year: |
|
|
raise ValidationError(MSG_YEAR_CANNOT_BE_GREATER) |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_genre(value: str) -> str: |
|
|
if value not in GENRES: |
|
|
raise ValidationError(MSG_GENRE_MUST_BE_ONE_OF) |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_status(value: str) -> str: |
|
|
if value not in BOOK_STATUSES: |
|
|
raise ValidationError(MSG_STATUS_MUST_BE_ONE_OF) |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_experience(value: int) -> int: |
|
|
if value < 0: |
|
|
raise ValidationError(MSG_EXPERIENCE_NEGATIVE) |
|
|
if value > 60: |
|
|
raise ValidationError(MSG_EXPERIENCE_AT_MOST_60) |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_work_days(days: List[str]) -> List[str]: |
|
|
if not days: |
|
|
raise ValidationError(MSG_AT_LEAST_ONE_WORK_DAY) |
|
|
invalid = [d for d in days if d not in WORK_DAYS] |
|
|
if invalid: |
|
|
raise ValidationError(MSG_INVALID_WORK_DAYS) |
|
|
return days |
|
|
|
|
|
|
|
|
def validate_uuid(value: str, field_name: str = "ID") -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
raise ValidationError(f"{field_name} {MSG_IS_REQUIRED}") |
|
|
if not UUID_PATTERN.match(value): |
|
|
raise ValidationError(f"{field_name} {MSG_MUST_BE_VALID_UUID}") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_date(value: str) -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
return date.today().isoformat() |
|
|
try: |
|
|
datetime.strptime(value, "%Y-%m-%d") |
|
|
return value |
|
|
except ValueError: |
|
|
raise ValidationError(MSG_DATE_YYYY_MM_DD) |
|
|
|
|
|
|
|
|
def validate_ids_list(ids: List[str]) -> List[str]: |
|
|
if not ids: |
|
|
raise ValidationError(MSG_AT_LEAST_ONE_ITEM_REQUIRED) |
|
|
validated = [] |
|
|
for i, id_val in enumerate(ids, 1): |
|
|
id_val = id_val.strip() |
|
|
if not id_val: |
|
|
raise ValidationError(MSG_ITEM_EMPTY) |
|
|
if not UUID_PATTERN.match(id_val): |
|
|
raise ValidationError(MSG_ITEM_INVALID_UUID) |
|
|
validated.append(id_val) |
|
|
return validated |
|
|
|
|
|
|
|
|
class LibraryClient: |
|
|
def __init__(self, base_url: str = BASE_URL): |
|
|
self.client = httpx.Client(base_url=base_url, timeout=30.0) |
|
|
|
|
|
def close(self): |
|
|
self.client.close() |
|
|
|
|
|
def _handle_response(self, response: httpx.Response): |
|
|
content_type = response.headers.get("content-type", "").lower() |
|
|
if response.status_code >= 400: |
|
|
message = response.text |
|
|
try: |
|
|
payload = response.json() |
|
|
if isinstance(payload, dict): |
|
|
message = ( |
|
|
payload.get("error", {}).get("message") |
|
|
or payload.get("message") |
|
|
or str(payload) |
|
|
) |
|
|
except Exception: |
|
|
message = response.text |
|
|
raise RuntimeError(f"{response.status_code}: {message}") |
|
|
if "application/json" in content_type: |
|
|
payload = response.json() |
|
|
if isinstance(payload, dict) and "successful" in payload: |
|
|
if payload.get("successful"): |
|
|
return payload.get("data") |
|
|
error = payload.get("error") or {} |
|
|
raise RuntimeError(error.get("message") or str(payload)) |
|
|
return payload |
|
|
return response |
|
|
|
|
|
def get(self, path: str, **kwargs): |
|
|
response = self.client.get(path, **kwargs) |
|
|
return self._handle_response(response) |
|
|
|
|
|
def post(self, path: str, **kwargs): |
|
|
response = self.client.post(path, **kwargs) |
|
|
return self._handle_response(response) |
|
|
|
|
|
def patch(self, path: str, **kwargs): |
|
|
response = self.client.patch(path, **kwargs) |
|
|
return self._handle_response(response) |
|
|
|
|
|
def delete(self, path: str, **kwargs): |
|
|
response = self.client.delete(path, **kwargs) |
|
|
return self._handle_response(response) |
|
|
|
|
|
def list_books(self): |
|
|
return self.get("/books/all") |
|
|
|
|
|
def get_book(self, book_id: str): |
|
|
return self.get(f"/books/{book_id}") |
|
|
|
|
|
def book_exists(self, book_id: str) -> bool: |
|
|
try: |
|
|
self.get_book(book_id) |
|
|
return True |
|
|
except RuntimeError: |
|
|
return False |
|
|
|
|
|
def create_book( |
|
|
self, |
|
|
title: str, |
|
|
author: str, |
|
|
pages: int, |
|
|
year: int, |
|
|
genre: str, |
|
|
status: str = "Available", |
|
|
): |
|
|
payload = { |
|
|
"title": title, |
|
|
"author": author, |
|
|
"pages": pages, |
|
|
"year": year, |
|
|
"genre": genre, |
|
|
"status": status, |
|
|
} |
|
|
return self.post("/books/create", json=payload) |
|
|
|
|
|
def update_book( |
|
|
self, |
|
|
book_id: str, |
|
|
title: Optional[str] = None, |
|
|
author: Optional[str] = None, |
|
|
pages: Optional[int] = None, |
|
|
year: Optional[int] = None, |
|
|
genre: Optional[str] = None, |
|
|
status: Optional[str] = None, |
|
|
): |
|
|
payload = { |
|
|
k: v |
|
|
for k, v in { |
|
|
"title": title, |
|
|
"author": author, |
|
|
"pages": pages, |
|
|
"year": year, |
|
|
"genre": genre, |
|
|
"status": status, |
|
|
}.items() |
|
|
if v not in (None, "") |
|
|
} |
|
|
return self.patch(f"/books/{book_id}", json=payload) |
|
|
|
|
|
def delete_book(self, book_id: str): |
|
|
return self.delete(f"/books/{book_id}") |
|
|
|
|
|
def borrow_books( |
|
|
self, book_ids: List[str], visitor_id: str, worker_id: str, borrow_date: str |
|
|
): |
|
|
payload = { |
|
|
"bookIds": book_ids, |
|
|
"visitorId": visitor_id, |
|
|
"workerId": worker_id, |
|
|
"borrowDate": borrow_date, |
|
|
} |
|
|
return self.post("/books/borrow", json=payload) |
|
|
|
|
|
def return_books( |
|
|
self, book_ids: List[str], visitor_id: str, worker_id: str, return_date: str |
|
|
): |
|
|
payload = { |
|
|
"bookIds": book_ids, |
|
|
"visitorId": visitor_id, |
|
|
"workerId": worker_id, |
|
|
"returnDate": return_date, |
|
|
} |
|
|
return self.post("/books/return", json=payload) |
|
|
|
|
|
def download_book(self, book_id: str) -> str: |
|
|
response = self.get(f"/books/{book_id}/download") |
|
|
return response.text if isinstance(response, httpx.Response) else str(response) |
|
|
|
|
|
def list_visitors(self): |
|
|
return self.get("/visitors/all") |
|
|
|
|
|
def get_visitor(self, visitor_id: str): |
|
|
return self.get(f"/visitors/{visitor_id}") |
|
|
|
|
|
def visitor_exists(self, visitor_id: str) -> bool: |
|
|
try: |
|
|
self.get_visitor(visitor_id) |
|
|
return True |
|
|
except RuntimeError: |
|
|
return False |
|
|
|
|
|
def create_visitor(self, name: str, surname: str): |
|
|
payload = {"name": name, "surname": surname} |
|
|
return self.post("/visitors/create", json=payload) |
|
|
|
|
|
def update_visitor( |
|
|
self, |
|
|
visitor_id: str, |
|
|
name: Optional[str] = None, |
|
|
surname: Optional[str] = None, |
|
|
): |
|
|
payload = {k: v for k, v in {"name": name, "surname": surname}.items() if v} |
|
|
return self.patch(f"/visitors/{visitor_id}", json=payload) |
|
|
|
|
|
def delete_visitor(self, visitor_id: str): |
|
|
return self.delete(f"/visitors/delete/{visitor_id}") |
|
|
|
|
|
def download_visitor(self, visitor_id: str) -> str: |
|
|
response = self.get(f"/visitors/{visitor_id}/download") |
|
|
return response.text if isinstance(response, httpx.Response) else str(response) |
|
|
|
|
|
def list_workers(self): |
|
|
return self.get("/workers/all") |
|
|
|
|
|
def get_worker(self, worker_id: str): |
|
|
return self.get(f"/workers/{worker_id}") |
|
|
|
|
|
def worker_exists(self, worker_id: str) -> bool: |
|
|
try: |
|
|
self.get_worker(worker_id) |
|
|
return True |
|
|
except RuntimeError: |
|
|
return False |
|
|
|
|
|
def create_worker( |
|
|
self, name: str, surname: str, experience: int, work_days: List[str] |
|
|
): |
|
|
payload = { |
|
|
"name": name, |
|
|
"surname": surname, |
|
|
"experience": experience, |
|
|
"workDays": work_days, |
|
|
} |
|
|
return self.post("/workers/create", json=payload) |
|
|
|
|
|
def update_worker( |
|
|
self, |
|
|
worker_id: str, |
|
|
name: Optional[str] = None, |
|
|
surname: Optional[str] = None, |
|
|
experience: Optional[int] = None, |
|
|
work_days: Optional[List[str]] = None, |
|
|
): |
|
|
payload = { |
|
|
k: v |
|
|
for k, v in { |
|
|
"name": name, |
|
|
"surname": surname, |
|
|
"experience": experience, |
|
|
"workDays": work_days, |
|
|
}.items() |
|
|
if v not in (None, "") |
|
|
} |
|
|
return self.patch(f"/workers/{worker_id}", json=payload) |
|
|
|
|
|
def delete_worker(self, worker_id: str): |
|
|
return self.delete(f"/workers/{worker_id}") |
|
|
|
|
|
def workers_by_days(self, work_days: List[str]): |
|
|
params = [("workDays", day) for day in work_days] |
|
|
return self.get("/workers/by-work-days", params=params) |
|
|
|
|
|
def download_worker(self, worker_id: str) -> str: |
|
|
response = self.get(f"/workers/{worker_id}/download") |
|
|
return response.text if isinstance(response, httpx.Response) else str(response) |
|
|
|
|
|
|
|
|
class MenuChoice(IntEnum): |
|
|
LIST_BOOKS = 1 |
|
|
VIEW_BOOK = 2 |
|
|
CREATE_BOOK = 3 |
|
|
UPDATE_BOOK = 4 |
|
|
DELETE_BOOK = 5 |
|
|
BORROW_BOOKS = 6 |
|
|
RETURN_BOOKS = 7 |
|
|
DOWNLOAD_BOOK = 8 |
|
|
LIST_VISITORS = 9 |
|
|
VIEW_VISITOR = 10 |
|
|
CREATE_VISITOR = 11 |
|
|
UPDATE_VISITOR = 12 |
|
|
DELETE_VISITOR = 13 |
|
|
DOWNLOAD_VISITOR = 14 |
|
|
LIST_WORKERS = 15 |
|
|
VIEW_WORKER = 16 |
|
|
CREATE_WORKER = 17 |
|
|
UPDATE_WORKER = 18 |
|
|
DELETE_WORKER = 19 |
|
|
WORKERS_BY_DAYS = 20 |
|
|
DOWNLOAD_WORKER = 21 |
|
|
EXIT = 0 |
|
|
|
|
|
|
|
|
MENU_TEXT = { |
|
|
MenuChoice.LIST_BOOKS: "List books", |
|
|
MenuChoice.VIEW_BOOK: "Get book by ID", |
|
|
MenuChoice.CREATE_BOOK: "Create book", |
|
|
MenuChoice.UPDATE_BOOK: "Update book", |
|
|
MenuChoice.DELETE_BOOK: "Delete book", |
|
|
MenuChoice.BORROW_BOOKS: "Borrow books", |
|
|
MenuChoice.RETURN_BOOKS: "Return books", |
|
|
MenuChoice.DOWNLOAD_BOOK: "Download book as file", |
|
|
MenuChoice.LIST_VISITORS: "List visitors", |
|
|
MenuChoice.VIEW_VISITOR: "Get visitor by ID", |
|
|
MenuChoice.CREATE_VISITOR: "Create visitor", |
|
|
MenuChoice.UPDATE_VISITOR: "Update visitor", |
|
|
MenuChoice.DELETE_VISITOR: "Delete visitor", |
|
|
MenuChoice.DOWNLOAD_VISITOR: "Download visitor as file", |
|
|
MenuChoice.LIST_WORKERS: "List workers", |
|
|
MenuChoice.VIEW_WORKER: "Get worker by ID", |
|
|
MenuChoice.CREATE_WORKER: "Create worker", |
|
|
MenuChoice.UPDATE_WORKER: "Update worker", |
|
|
MenuChoice.DELETE_WORKER: "Delete worker", |
|
|
MenuChoice.WORKERS_BY_DAYS: "Find workers by work days", |
|
|
MenuChoice.DOWNLOAD_WORKER: "Download worker as file", |
|
|
MenuChoice.EXIT: "Exit", |
|
|
} |
|
|
|
|
|
|
|
|
def prompt_with_validation(label: str, validator: Callable[[str], str]) -> str: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
try: |
|
|
return validator(raw) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_int_with_validation(label: str, validator: Callable[[int], int]) -> int: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
try: |
|
|
value = int(raw) |
|
|
return validator(value) |
|
|
except ValueError: |
|
|
print(MSG_ERROR_ENTER_VALID_INTEGER) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_optional_with_validation(label: str, validator: Callable[[str], str]) -> Optional[str]: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
if not raw: |
|
|
return None |
|
|
try: |
|
|
return validator(raw) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_optional_int_with_validation(label: str, validator: Callable[[int], int]) -> Optional[int]: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
if not raw: |
|
|
return None |
|
|
try: |
|
|
value = int(raw) |
|
|
return validator(value) |
|
|
except ValueError: |
|
|
print(MSG_ERROR_ENTER_VALID_INTEGER) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_choice(label: str, options: List[str]) -> str: |
|
|
while True: |
|
|
print(label) |
|
|
for idx, option in enumerate(options, 1): |
|
|
print(f" {idx}. {option}") |
|
|
raw = input(MSG_PICK_NUMBER).strip() |
|
|
try: |
|
|
selected = int(raw) |
|
|
if 1 <= selected <= len(options): |
|
|
return options[selected - 1] |
|
|
except ValueError: |
|
|
pass |
|
|
print(MSG_ERROR_INVALID_CHOICE) |
|
|
|
|
|
|
|
|
def prompt_optional_choice(label: str, options: List[str]) -> Optional[str]: |
|
|
while True: |
|
|
print(label) |
|
|
print(" 0. Skip") |
|
|
for idx, option in enumerate(options, 1): |
|
|
print(f" {idx}. {option}") |
|
|
raw = input(MSG_PICK_NUMBER_SKIP).strip() |
|
|
try: |
|
|
selected = int(raw) |
|
|
if selected == 0: |
|
|
return None |
|
|
if 1 <= selected <= len(options): |
|
|
return options[selected - 1] |
|
|
except ValueError: |
|
|
pass |
|
|
print(MSG_ERROR_INVALID_CHOICE) |
|
|
|
|
|
|
|
|
def prompt_multi_choice(label: str, options: List[str]) -> List[str]: |
|
|
print(label) |
|
|
print(MSG_USE_COMMA_OR_STAR) |
|
|
for idx, option in enumerate(options, 1): |
|
|
print(f" {idx}. {option}") |
|
|
while True: |
|
|
raw = input(MSG_ENTER_NUMBERS_OR_NAMES).strip() |
|
|
if raw == "*": |
|
|
return options[:] |
|
|
parts = [p.strip() for p in raw.split(",") if p.strip()] |
|
|
result = [] |
|
|
valid = True |
|
|
for p in parts: |
|
|
if p in options: |
|
|
result.append(p) |
|
|
elif p.isdigit(): |
|
|
idx = int(p) |
|
|
if 1 <= idx <= len(options): |
|
|
result.append(options[idx - 1]) |
|
|
else: |
|
|
print(MSG_ERROR_INVALID_NUMBER) |
|
|
valid = False |
|
|
break |
|
|
else: |
|
|
print(MSG_ERROR_INVALID_VALUE) |
|
|
valid = False |
|
|
break |
|
|
if valid and result: |
|
|
try: |
|
|
return validate_work_days(result) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
elif valid and not result: |
|
|
print(MSG_ERROR_AT_LEAST_ONE_VALUE) |
|
|
|
|
|
|
|
|
def prompt_optional_multi_choice(label: str, options: List[str]) -> Optional[List[str]]: |
|
|
print(label) |
|
|
print(MSG_USE_COMMA_STAR_OR_EMPTY) |
|
|
for idx, option in enumerate(options, 1): |
|
|
print(f" {idx}. {option}") |
|
|
while True: |
|
|
raw = input(MSG_ENTER_NUMBERS_OR_NAMES_EMPTY_SKIP).strip() |
|
|
if not raw: |
|
|
return None |
|
|
if raw == "*": |
|
|
return options[:] |
|
|
parts = [p.strip() for p in raw.split(",") if p.strip()] |
|
|
result = [] |
|
|
valid = True |
|
|
for p in parts: |
|
|
if p in options: |
|
|
result.append(p) |
|
|
elif p.isdigit(): |
|
|
idx = int(p) |
|
|
if 1 <= idx <= len(options): |
|
|
result.append(options[idx - 1]) |
|
|
else: |
|
|
print(MSG_ERROR_INVALID_NUMBER) |
|
|
valid = False |
|
|
break |
|
|
else: |
|
|
print(MSG_ERROR_INVALID_VALUE) |
|
|
valid = False |
|
|
break |
|
|
if valid and result: |
|
|
try: |
|
|
return validate_work_days(result) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
elif valid and not result: |
|
|
return None |
|
|
|
|
|
|
|
|
def prompt_book_ids(label: str, client: LibraryClient) -> List[str]: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
parts = [p.strip() for p in raw.split(",") if p.strip()] |
|
|
try: |
|
|
validated = validate_ids_list(parts) |
|
|
for book_id in validated: |
|
|
if not client.book_exists(book_id): |
|
|
print(MSG_ERROR_BOOK_NOT_FOUND) |
|
|
raise ValidationError("Book not found") |
|
|
return validated |
|
|
except ValidationError as e: |
|
|
if "not found" not in str(e): |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_book_id(label: str, client: LibraryClient, check_exists: bool = True) -> str: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
try: |
|
|
validated = validate_uuid(raw, "Book ID") |
|
|
if check_exists and not client.book_exists(validated): |
|
|
print(MSG_ERROR_BOOK_NOT_FOUND_VALIDATED) |
|
|
continue |
|
|
return validated |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_visitor_id(label: str, client: LibraryClient, check_exists: bool = True) -> str: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
try: |
|
|
validated = validate_uuid(raw, "Visitor ID") |
|
|
if check_exists and not client.visitor_exists(validated): |
|
|
print(MSG_ERROR_VISITOR_NOT_FOUND) |
|
|
continue |
|
|
return validated |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_worker_id(label: str, client: LibraryClient, check_exists: bool = True) -> str: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
try: |
|
|
validated = validate_uuid(raw, "Worker ID") |
|
|
if check_exists and not client.worker_exists(validated): |
|
|
print(MSG_ERROR_WORKER_NOT_FOUND) |
|
|
continue |
|
|
return validated |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_date(label: str) -> str: |
|
|
today = date.today().isoformat() |
|
|
while True: |
|
|
raw = input(f"{label} [{today}]: ").strip() |
|
|
try: |
|
|
return validate_date(raw) |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def print_books(books: Any): |
|
|
if not books: |
|
|
print(MSG_NO_BOOKS_FOUND) |
|
|
return |
|
|
for book in books: |
|
|
print(SEPARATOR) |
|
|
print(f"ID: {book.get('id')}") |
|
|
print(f"Title: {book.get('title')}") |
|
|
print(f"Author: {book.get('author')}") |
|
|
print(f"Pages: {book.get('pages')}") |
|
|
print(f"Year: {book.get('year')}") |
|
|
print(f"Genre: {book.get('genre')}") |
|
|
print(f"Status: {book.get('status')}") |
|
|
|
|
|
|
|
|
def print_visitors(visitors: Any): |
|
|
if not visitors: |
|
|
print(MSG_NO_VISITORS_FOUND) |
|
|
return |
|
|
for visitor in visitors: |
|
|
print(SEPARATOR) |
|
|
print(f"ID: {visitor.get('id')}") |
|
|
print(f"Name: {visitor.get('name')} {visitor.get('surname')}") |
|
|
print(f"Registered: {visitor.get('registrationDate')}") |
|
|
current = visitor.get("currentBooks") or [] |
|
|
history = visitor.get("history") or [] |
|
|
print(f"Current books: {len(current)}") |
|
|
print(f"History: {len(history)}") |
|
|
|
|
|
|
|
|
def print_workers(workers: Any): |
|
|
if not workers: |
|
|
print(MSG_NO_WORKERS_FOUND) |
|
|
return |
|
|
for worker in workers: |
|
|
print(SEPARATOR) |
|
|
print(f"ID: {worker.get('id')}") |
|
|
print(f"Name: {worker.get('name')} {worker.get('surname')}") |
|
|
print(f"Experience: {worker.get('experience')} years") |
|
|
print(f"Work days: {', '.join(worker.get('workDays', []))}") |
|
|
issued = worker.get("issuedBooks") or [] |
|
|
print(f"Issued books: {len(issued)}") |
|
|
|
|
|
|
|
|
def main_menu() -> Optional[MenuChoice]: |
|
|
print("\n" + SEPARATOR) |
|
|
for choice in MenuChoice: |
|
|
print(f"{choice.value}. {MENU_TEXT[choice]}") |
|
|
raw = input(PROMPT).strip() |
|
|
try: |
|
|
return MenuChoice(int(raw)) |
|
|
except Exception: |
|
|
return None |
|
|
|
|
|
|
|
|
def handle_error(error: Exception): |
|
|
print(f"{MSG_ERROR_PREFIX}{error}") |
|
|
|
|
|
|
|
|
def main(): |
|
|
client = LibraryClient() |
|
|
try: |
|
|
while True: |
|
|
choice = main_menu() |
|
|
if choice is None: |
|
|
print(MSG_INVALID_OPTION) |
|
|
continue |
|
|
if choice == MenuChoice.EXIT: |
|
|
print(MSG_GOODBYE) |
|
|
break |
|
|
try: |
|
|
if choice == MenuChoice.LIST_BOOKS: |
|
|
print_books(client.list_books()) |
|
|
|
|
|
elif choice == MenuChoice.VIEW_BOOK: |
|
|
book_id = prompt_book_id(MSG_BOOK_ID_PROMPT, client) |
|
|
print_books([client.get_book(book_id)]) |
|
|
|
|
|
elif choice == MenuChoice.CREATE_BOOK: |
|
|
title = prompt_with_validation(MSG_TITLE_PROMPT, validate_title) |
|
|
author = prompt_with_validation(MSG_AUTHOR_PROMPT, validate_author) |
|
|
pages = prompt_int_with_validation(MSG_PAGES_PROMPT, validate_pages) |
|
|
year = prompt_int_with_validation(MSG_YEAR_PROMPT, validate_year) |
|
|
genre = prompt_choice(MSG_GENRE_PROMPT, GENRES) |
|
|
status = prompt_choice(MSG_STATUS_PROMPT, BOOK_STATUSES) |
|
|
print_books([client.create_book(title, author, pages, year, genre, status)]) |
|
|
|
|
|
elif choice == MenuChoice.UPDATE_BOOK: |
|
|
book_id = prompt_book_id(MSG_BOOK_ID_PROMPT, client) |
|
|
print(MSG_LEAVE_EMPTY_TO_SKIP) |
|
|
title = prompt_optional_with_validation(MSG_TITLE_SKIP_PROMPT, validate_title) |
|
|
author = prompt_optional_with_validation(MSG_AUTHOR_SKIP_PROMPT, validate_author) |
|
|
pages = prompt_optional_int_with_validation(MSG_PAGES_SKIP_PROMPT, validate_pages) |
|
|
year = prompt_optional_int_with_validation(MSG_YEAR_SKIP_PROMPT, validate_year) |
|
|
genre = prompt_optional_choice(MSG_GENRE_PROMPT, GENRES) |
|
|
status = prompt_optional_choice(MSG_STATUS_PROMPT, BOOK_STATUSES) |
|
|
if not any([title, author, pages, year, genre, status]): |
|
|
print(MSG_NO_FIELDS_TO_UPDATE) |
|
|
continue |
|
|
updated = client.update_book(book_id, title, author, pages, year, genre, status) |
|
|
print_books([updated]) |
|
|
|
|
|
elif choice == MenuChoice.DELETE_BOOK: |
|
|
book_id = prompt_book_id(MSG_BOOK_ID_PROMPT, client) |
|
|
result = client.delete_book(book_id) |
|
|
print(result.get("message") if isinstance(result, dict) else MSG_DELETED) |
|
|
|
|
|
elif choice == MenuChoice.BORROW_BOOKS: |
|
|
book_ids = prompt_book_ids(MSG_BOOK_IDS_COMMA_SEPARATED, client) |
|
|
visitor_id = prompt_visitor_id(MSG_VISITOR_ID_PROMPT, client) |
|
|
worker_id = prompt_worker_id(MSG_WORKER_ID_PROMPT, client) |
|
|
borrow_date = prompt_date(MSG_BORROW_DATE_PROMPT) |
|
|
result = client.borrow_books(book_ids, visitor_id, worker_id, borrow_date) |
|
|
print(result.get("message") if isinstance(result, dict) else result) |
|
|
|
|
|
elif choice == MenuChoice.RETURN_BOOKS: |
|
|
book_ids = prompt_book_ids(MSG_BOOK_IDS_COMMA_SEPARATED, client) |
|
|
visitor_id = prompt_visitor_id(MSG_VISITOR_ID_PROMPT, client) |
|
|
worker_id = prompt_worker_id(MSG_WORKER_ID_PROMPT, client) |
|
|
return_date = prompt_date(MSG_RETURN_DATE_PROMPT) |
|
|
result = client.return_books(book_ids, visitor_id, worker_id, return_date) |
|
|
print(result.get("message") if isinstance(result, dict) else result) |
|
|
|
|
|
elif choice == MenuChoice.DOWNLOAD_BOOK: |
|
|
book_id = prompt_book_id(MSG_BOOK_ID_PROMPT, client) |
|
|
content = client.download_book(book_id) |
|
|
filepath = save_to_file("books", book_id, content) |
|
|
print(f"{MSG_FILE_SAVED_TO}{filepath}") |
|
|
|
|
|
elif choice == MenuChoice.LIST_VISITORS: |
|
|
print_visitors(client.list_visitors()) |
|
|
|
|
|
elif choice == MenuChoice.VIEW_VISITOR: |
|
|
visitor_id = prompt_visitor_id(MSG_VISITOR_ID_PROMPT, client) |
|
|
visitor = client.get_visitor(visitor_id) |
|
|
print_visitors([visitor]) |
|
|
current = visitor.get("currentBooks") or [] |
|
|
history = visitor.get("history") or [] |
|
|
if current: |
|
|
print(MSG_CURRENT_BOOKS) |
|
|
print_books(current) |
|
|
if history: |
|
|
print(MSG_HISTORY) |
|
|
print_books(history) |
|
|
|
|
|
elif choice == MenuChoice.CREATE_VISITOR: |
|
|
name = prompt_with_validation(MSG_NAME_PROMPT, lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_with_validation(MSG_SURNAME_PROMPT, lambda v: validate_name(v, "Surname")) |
|
|
print_visitors([client.create_visitor(name, surname)]) |
|
|
|
|
|
elif choice == MenuChoice.UPDATE_VISITOR: |
|
|
visitor_id = prompt_visitor_id(MSG_VISITOR_ID_PROMPT, client) |
|
|
print(MSG_LEAVE_EMPTY_TO_SKIP) |
|
|
name = prompt_optional_with_validation(MSG_NAME_PROMPT + " (empty to skip): ", |
|
|
lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_optional_with_validation(MSG_SURNAME_PROMPT + " (empty to skip): ", |
|
|
lambda v: validate_name(v, "Surname")) |
|
|
if not name and not surname: |
|
|
print(MSG_NO_FIELDS_TO_UPDATE) |
|
|
continue |
|
|
visitor = client.update_visitor(visitor_id, name, surname) |
|
|
print_visitors([visitor]) |
|
|
|
|
|
elif choice == MenuChoice.DELETE_VISITOR: |
|
|
visitor_id = prompt_visitor_id(MSG_VISITOR_ID_PROMPT, client) |
|
|
client.delete_visitor(visitor_id) |
|
|
print(MSG_VISITOR_DELETED) |
|
|
|
|
|
elif choice == MenuChoice.DOWNLOAD_VISITOR: |
|
|
visitor_id = prompt_visitor_id(MSG_VISITOR_ID_PROMPT, client) |
|
|
content = client.download_visitor(visitor_id) |
|
|
filepath = save_to_file("visitors", visitor_id, content) |
|
|
print(f"{MSG_FILE_SAVED_TO}{filepath}") |
|
|
|
|
|
elif choice == MenuChoice.LIST_WORKERS: |
|
|
print_workers(client.list_workers()) |
|
|
|
|
|
elif choice == MenuChoice.VIEW_WORKER: |
|
|
worker_id = prompt_worker_id(MSG_WORKER_ID_PROMPT, client) |
|
|
worker = client.get_worker(worker_id) |
|
|
print_workers([worker]) |
|
|
issued = worker.get("issuedBooks") or [] |
|
|
if issued: |
|
|
print("\nIssued books:") |
|
|
print_books(issued) |
|
|
|
|
|
elif choice == MenuChoice.CREATE_WORKER: |
|
|
name = prompt_with_validation(MSG_NAME_PROMPT, lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_with_validation(MSG_SURNAME_PROMPT, lambda v: validate_name(v, "Surname")) |
|
|
experience = prompt_int_with_validation(MSG_EXPERIENCE_PROMPT, validate_experience) |
|
|
work_days = prompt_multi_choice(MSG_WORK_DAYS_PROMPT, WORK_DAYS) |
|
|
print_workers([client.create_worker(name, surname, experience, work_days)]) |
|
|
|
|
|
elif choice == MenuChoice.UPDATE_WORKER: |
|
|
worker_id = prompt_worker_id(MSG_WORKER_ID_PROMPT, client) |
|
|
print(MSG_LEAVE_EMPTY_TO_SKIP) |
|
|
name = prompt_optional_with_validation(MSG_NAME_PROMPT + " (empty to skip): ", |
|
|
lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_optional_with_validation(MSG_SURNAME_PROMPT + " (empty to skip): ", |
|
|
lambda v: validate_name(v, "Surname")) |
|
|
experience = prompt_optional_int_with_validation(MSG_EXPERIENCE_PROMPT + " (empty to skip): ", |
|
|
validate_experience) |
|
|
work_days = prompt_optional_multi_choice(MSG_WORK_DAYS_PROMPT, WORK_DAYS) |
|
|
if not any([name, surname, experience is not None, work_days]): |
|
|
print(MSG_NO_FIELDS_TO_UPDATE) |
|
|
continue |
|
|
worker = client.update_worker(worker_id, name, surname, experience, work_days) |
|
|
print_workers([worker]) |
|
|
|
|
|
elif choice == MenuChoice.DELETE_WORKER: |
|
|
worker_id = prompt_worker_id(MSG_WORKER_ID_PROMPT, client) |
|
|
client.delete_worker(worker_id) |
|
|
print(MSG_WORKER_DELETED) |
|
|
|
|
|
elif choice == MenuChoice.WORKERS_BY_DAYS: |
|
|
work_days = prompt_multi_choice(MSG_SELECT_WORK_DAYS, WORK_DAYS) |
|
|
print_workers(client.workers_by_days(work_days)) |
|
|
|
|
|
elif choice == MenuChoice.DOWNLOAD_WORKER: |
|
|
worker_id = prompt_worker_id(MSG_WORKER_ID_PROMPT, client) |
|
|
content = client.download_worker(worker_id) |
|
|
filepath = save_to_file("workers", worker_id, content) |
|
|
print(f"{MSG_FILE_SAVED_TO}{filepath}") |
|
|
|
|
|
else: |
|
|
print(MSG_INVALID_OPTION) |
|
|
|
|
|
except (RuntimeError, httpx.HTTPError, ValueError) as error: |
|
|
handle_error(error) |
|
|
except KeyboardInterrupt: |
|
|
print(MSG_INTERRUPTED_GOODBYE) |
|
|
finally: |
|
|
client.close() |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|
