|
|
import json |
|
|
import re |
|
|
from datetime import date, datetime |
|
|
from enum import IntEnum |
|
|
from pathlib import Path |
|
|
from typing import Any, List, Optional, Callable |
|
|
from uuid import UUID |
|
|
|
|
|
import httpx |
|
|
|
|
|
FORUM_BASE_URL = "https://brestok-lab4.hf.space" |
|
|
LIBRARY_BASE_URL = "https://brestok-vika-server.hf.space" |
|
|
DOWNLOADS_DIR = Path(__file__).parent / "downloads" |
|
|
|
|
|
SEPARATOR = "=" * 70 |
|
|
SEPARATOR_60 = "=" * 60 |
|
|
SEPARATOR_90 = "-" * 90 |
|
|
DASH_LINE_60 = "-" * 60 |
|
|
PROMPT = "Select option: " |
|
|
|
|
|
TABLE_PARTICIPANTS = "participants" |
|
|
TABLE_TOPICS = "topics" |
|
|
|
|
|
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}$") |
|
|
|
|
|
|
|
|
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} is required.") |
|
|
if len(value) < 2: |
|
|
raise ValidationError(f"{field_name} must be at least 2 characters.") |
|
|
if len(value) > 50: |
|
|
raise ValidationError(f"{field_name} must be at most 50 characters.") |
|
|
if not NAME_PATTERN.match(value): |
|
|
raise ValidationError(f"{field_name} must contain only letters, spaces, hyphens, and apostrophes.") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_title(value: str, field_name: str = "Title") -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
raise ValidationError(f"{field_name} is required.") |
|
|
if len(value) < 2: |
|
|
raise ValidationError(f"{field_name} must be at least 2 characters.") |
|
|
if len(value) > 100: |
|
|
raise ValidationError(f"{field_name} must be at most 100 characters.") |
|
|
if not TITLE_PATTERN.match(value): |
|
|
raise ValidationError(f"{field_name} contains invalid characters.") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_author(value: str) -> str: |
|
|
value = value.strip() |
|
|
if not value: |
|
|
raise ValidationError("Author is required.") |
|
|
if len(value) < 2: |
|
|
raise ValidationError("Author must be at least 2 characters.") |
|
|
if len(value) > 100: |
|
|
raise ValidationError("Author must be at most 100 characters.") |
|
|
if not NAME_PATTERN.match(value): |
|
|
raise ValidationError("Author must contain only letters, spaces, hyphens, and apostrophes.") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_pages(value: int) -> int: |
|
|
if value < 1: |
|
|
raise ValidationError("Pages must be at least 1.") |
|
|
if value > 10000: |
|
|
raise ValidationError("Pages must be at most 10000.") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_year(value: int) -> int: |
|
|
current_year = date.today().year |
|
|
if value < 1000: |
|
|
raise ValidationError("Year must be at least 1000.") |
|
|
if value > current_year: |
|
|
raise ValidationError(f"Year cannot be greater than {current_year}.") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_genre(value: str) -> str: |
|
|
if value not in GENRES: |
|
|
raise ValidationError(f"Genre must be one of: {', '.join(GENRES)}") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_status(value: str) -> str: |
|
|
if value not in BOOK_STATUSES: |
|
|
raise ValidationError(f"Status must be one of: {', '.join(BOOK_STATUSES)}") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_experience(value: int) -> int: |
|
|
if value < 0: |
|
|
raise ValidationError("Experience cannot be negative.") |
|
|
if value > 60: |
|
|
raise ValidationError("Experience must be at most 60 years.") |
|
|
return value |
|
|
|
|
|
|
|
|
def validate_work_days(days: List[str]) -> List[str]: |
|
|
if not days: |
|
|
raise ValidationError("At least one work day is required.") |
|
|
invalid = [d for d in days if d not in WORK_DAYS] |
|
|
if invalid: |
|
|
raise ValidationError(f"Invalid work days: {', '.join(invalid)}. Valid options: {', '.join(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} is required.") |
|
|
if not UUID_PATTERN.match(value): |
|
|
raise ValidationError(f"{field_name} must be a valid UUID format.") |
|
|
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("Date must be in YYYY-MM-DD format.") |
|
|
|
|
|
|
|
|
def validate_ids_list(ids: List[str]) -> List[str]: |
|
|
if not ids: |
|
|
raise ValidationError("At least one item is required.") |
|
|
validated = [] |
|
|
for i, id_val in enumerate(ids, 1): |
|
|
id_val = id_val.strip() |
|
|
if not id_val: |
|
|
raise ValidationError(f"Item #{i} is empty.") |
|
|
if not UUID_PATTERN.match(id_val): |
|
|
raise ValidationError(f"Item #{i} ({id_val}) is not a valid UUID.") |
|
|
validated.append(id_val) |
|
|
return validated |
|
|
|
|
|
|
|
|
def validate_activity_rating(value: float) -> float: |
|
|
if value < 0.1 or value > 5.0: |
|
|
raise ValidationError("Activity rating must be between 0.1 and 5.0.") |
|
|
return value |
|
|
|
|
|
|
|
|
class ForumClient: |
|
|
def __init__(self, base_url: str = FORUM_BASE_URL): |
|
|
self.base_url = base_url |
|
|
self.client = httpx.Client(timeout=30.0) |
|
|
|
|
|
@staticmethod |
|
|
def _link(table: str, value: str) -> dict: |
|
|
return {"table": table, "value": value} |
|
|
|
|
|
def get_participants(self) -> list: |
|
|
response = self.client.get(f"{self.base_url}/participants/") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_participant(self, participant_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/participants/{participant_id}") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def create_participant(self, first_name: str, last_name: str, nickname: str, activity_rating: float) -> dict: |
|
|
payload = { |
|
|
"first_name": first_name, |
|
|
"last_name": last_name, |
|
|
"nickname": nickname, |
|
|
"activity_rating": activity_rating, |
|
|
} |
|
|
response = self.client.post(f"{self.base_url}/participants/", json=payload) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_topics(self) -> list: |
|
|
response = self.client.get(f"{self.base_url}/topics/") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_topic(self, topic_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/topics/{topic_id}") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def get_messages(self) -> list: |
|
|
response = self.client.get(f"{self.base_url}/messages/") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def create_topic(self, title: str, description: str, participants: list = None) -> dict: |
|
|
payload = { |
|
|
"title": title, |
|
|
"description": description, |
|
|
"participants": [self._link(TABLE_PARTICIPANTS, p) for p in (participants or [])], |
|
|
"messages": [], |
|
|
} |
|
|
response = self.client.post(f"{self.base_url}/topics/", json=payload) |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def publish_message(self, topic_id: str, participant_id: str, content: str) -> dict | str: |
|
|
payload = { |
|
|
"participant_id": self._link(TABLE_PARTICIPANTS, participant_id), |
|
|
"content": content, |
|
|
} |
|
|
response = self.client.post(f"{self.base_url}/topics/{topic_id}/messages", json=payload) |
|
|
if response.status_code == 400: |
|
|
error_detail = response.json().get("detail", "Unknown error") |
|
|
return f"ERROR: {error_detail}" |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def download_participant(self, participant_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/participants/{participant_id}/download") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def download_topic(self, topic_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/topics/{topic_id}/download") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def download_message(self, message_id: str) -> dict: |
|
|
response = self.client.get(f"{self.base_url}/messages/{message_id}/download") |
|
|
response.raise_for_status() |
|
|
return response.json() |
|
|
|
|
|
def close(self): |
|
|
self.client.close() |
|
|
|
|
|
|
|
|
class LibraryClient: |
|
|
def __init__(self, base_url: str = LIBRARY_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 ForumMenuChoice(IntEnum): |
|
|
LIST_PARTICIPANTS = 1 |
|
|
GET_PARTICIPANT = 2 |
|
|
CREATE_PARTICIPANT = 3 |
|
|
DOWNLOAD_PARTICIPANT = 4 |
|
|
LIST_TOPICS = 5 |
|
|
GET_TOPIC = 6 |
|
|
CREATE_TOPIC = 7 |
|
|
DOWNLOAD_TOPIC = 8 |
|
|
PUBLISH_MESSAGE = 9 |
|
|
LIST_MESSAGES = 10 |
|
|
DOWNLOAD_MESSAGE = 11 |
|
|
BACK = 0 |
|
|
|
|
|
|
|
|
FORUM_MENU_TEXT = { |
|
|
ForumMenuChoice.LIST_PARTICIPANTS: "List all participants", |
|
|
ForumMenuChoice.GET_PARTICIPANT: "Get participant by ID", |
|
|
ForumMenuChoice.CREATE_PARTICIPANT: "Create new participant", |
|
|
ForumMenuChoice.DOWNLOAD_PARTICIPANT: "Download participant as file", |
|
|
ForumMenuChoice.LIST_TOPICS: "List all topics", |
|
|
ForumMenuChoice.GET_TOPIC: "Get topic by ID", |
|
|
ForumMenuChoice.CREATE_TOPIC: "Create new topic", |
|
|
ForumMenuChoice.DOWNLOAD_TOPIC: "Download topic as file", |
|
|
ForumMenuChoice.PUBLISH_MESSAGE: "Publish message to topic", |
|
|
ForumMenuChoice.LIST_MESSAGES: "List all messages", |
|
|
ForumMenuChoice.DOWNLOAD_MESSAGE: "Download message as file", |
|
|
ForumMenuChoice.BACK: "Back to server selection", |
|
|
} |
|
|
|
|
|
|
|
|
class LibraryMenuChoice(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 |
|
|
BACK = 0 |
|
|
|
|
|
|
|
|
LIBRARY_MENU_TEXT = { |
|
|
LibraryMenuChoice.LIST_BOOKS: "List books", |
|
|
LibraryMenuChoice.VIEW_BOOK: "Get book by ID", |
|
|
LibraryMenuChoice.CREATE_BOOK: "Create book", |
|
|
LibraryMenuChoice.UPDATE_BOOK: "Update book", |
|
|
LibraryMenuChoice.DELETE_BOOK: "Delete book", |
|
|
LibraryMenuChoice.BORROW_BOOKS: "Borrow books", |
|
|
LibraryMenuChoice.RETURN_BOOKS: "Return books", |
|
|
LibraryMenuChoice.DOWNLOAD_BOOK: "Download book as file", |
|
|
LibraryMenuChoice.LIST_VISITORS: "List visitors", |
|
|
LibraryMenuChoice.VIEW_VISITOR: "Get visitor by ID", |
|
|
LibraryMenuChoice.CREATE_VISITOR: "Create visitor", |
|
|
LibraryMenuChoice.UPDATE_VISITOR: "Update visitor", |
|
|
LibraryMenuChoice.DELETE_VISITOR: "Delete visitor", |
|
|
LibraryMenuChoice.DOWNLOAD_VISITOR: "Download visitor as file", |
|
|
LibraryMenuChoice.LIST_WORKERS: "List workers", |
|
|
LibraryMenuChoice.VIEW_WORKER: "Get worker by ID", |
|
|
LibraryMenuChoice.CREATE_WORKER: "Create worker", |
|
|
LibraryMenuChoice.UPDATE_WORKER: "Update worker", |
|
|
LibraryMenuChoice.DELETE_WORKER: "Delete worker", |
|
|
LibraryMenuChoice.WORKERS_BY_DAYS: "Find workers by work days", |
|
|
LibraryMenuChoice.DOWNLOAD_WORKER: "Download worker as file", |
|
|
LibraryMenuChoice.BACK: "Back to server selection", |
|
|
} |
|
|
|
|
|
|
|
|
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(" Error: Enter a valid integer.") |
|
|
except ValidationError as e: |
|
|
print(f" Error: {e}") |
|
|
|
|
|
|
|
|
def prompt_float_with_validation(label: str, validator: Callable[[float], float]) -> float: |
|
|
while True: |
|
|
raw = input(label).strip() |
|
|
try: |
|
|
value = float(raw) |
|
|
return validator(value) |
|
|
except ValueError: |
|
|
print(" Error: Enter a valid number.") |
|
|
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(" Error: Enter a 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("Pick number: ").strip() |
|
|
try: |
|
|
selected = int(raw) |
|
|
if 1 <= selected <= len(options): |
|
|
return options[selected - 1] |
|
|
except ValueError: |
|
|
pass |
|
|
print(" 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("Pick number (0 to skip): ").strip() |
|
|
try: |
|
|
selected = int(raw) |
|
|
if selected == 0: |
|
|
return None |
|
|
if 1 <= selected <= len(options): |
|
|
return options[selected - 1] |
|
|
except ValueError: |
|
|
pass |
|
|
print(" Error: Invalid choice.") |
|
|
|
|
|
|
|
|
def prompt_multi_choice(label: str, options: List[str]) -> List[str]: |
|
|
print(label) |
|
|
print("Use comma to separate values or type * for all.") |
|
|
for idx, option in enumerate(options, 1): |
|
|
print(f" {idx}. {option}") |
|
|
while True: |
|
|
raw = input("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(f" Error: Invalid number {p}.") |
|
|
valid = False |
|
|
break |
|
|
else: |
|
|
print(f" Error: Invalid value '{p}'.") |
|
|
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(" Error: At least one value is required.") |
|
|
|
|
|
|
|
|
def prompt_optional_multi_choice(label: str, options: List[str]) -> Optional[List[str]]: |
|
|
print(label) |
|
|
print("Use comma to separate values, * for all, or leave empty to skip.") |
|
|
for idx, option in enumerate(options, 1): |
|
|
print(f" {idx}. {option}") |
|
|
while True: |
|
|
raw = input("Enter numbers or names (empty to 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(f" Error: Invalid number {p}.") |
|
|
valid = False |
|
|
break |
|
|
else: |
|
|
print(f" Error: Invalid value '{p}'.") |
|
|
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(f" Error: Book with ID {book_id} 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(f" Error: Book with ID {validated} not found.") |
|
|
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(f" Error: Visitor with ID {validated} 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(f" Error: Worker with ID {validated} 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("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("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("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 print_participants(participants: list): |
|
|
if not participants: |
|
|
print("No participants found.") |
|
|
return |
|
|
header = f"{'ID':<40} {'Name':<25} {'Nickname':<15} {'Rating':<10}" |
|
|
print(header) |
|
|
print(SEPARATOR_90) |
|
|
for p in participants: |
|
|
full_name = f"{p['first_name']} {p['last_name']}" |
|
|
print(f"{p['id']:<40} {full_name:<25} {p['nickname']:<15} {p['activity_rating']:<10}") |
|
|
|
|
|
|
|
|
def print_topics(topics: list): |
|
|
if not topics: |
|
|
print("No topics found.") |
|
|
return |
|
|
for topic in topics: |
|
|
print(f"\nTopic: {topic['title']}") |
|
|
print(f" ID: {topic['id']}") |
|
|
print(f" Description: {topic['description']}") |
|
|
print(f" Created: {topic['created_at']}") |
|
|
print(f" Messages ({len(topic['messages'])}):") |
|
|
for msg in topic["messages"]: |
|
|
print(f" {msg.get('order_in_topic', 0)}. [{msg['participant_name']}]: {msg['content']}") |
|
|
|
|
|
|
|
|
def print_topic_detail(topic: dict): |
|
|
print(f"\nTopic: {topic['title']}") |
|
|
print(f" ID: {topic['id']}") |
|
|
print(f" Description: {topic['description']}") |
|
|
print(f" Created: {topic['created_at']}") |
|
|
print(f" Participants: {len(topic['participants'])}") |
|
|
print(f"\n Messages ({len(topic['messages'])}):") |
|
|
if not topic["messages"]: |
|
|
print(" No messages yet.") |
|
|
for msg in topic["messages"]: |
|
|
print(f" {msg.get('order_in_topic', 0)}. [{msg['participant_name']}]: {msg['content']}") |
|
|
|
|
|
|
|
|
def print_messages(messages: list): |
|
|
if not messages: |
|
|
print(" No messages yet.") |
|
|
return |
|
|
for msg in messages: |
|
|
print(f"{msg['topic_title']} | #{msg['order_in_topic']} [{msg['participant_name']}]: {msg['content']}") |
|
|
|
|
|
|
|
|
def server_selection_menu() -> Optional[int]: |
|
|
print("\n" + SEPARATOR) |
|
|
print(" UNIFIED CLIENT - SERVER SELECTION") |
|
|
print(SEPARATOR) |
|
|
print("1. Forum Server (Topics, Participants, Messages)") |
|
|
print("2. Library Server (Books, Visitors, Workers)") |
|
|
print("0. Exit") |
|
|
print(DASH_LINE_60) |
|
|
raw = input(PROMPT).strip() |
|
|
try: |
|
|
choice = int(raw) |
|
|
if choice in (0, 1, 2): |
|
|
return choice |
|
|
except ValueError: |
|
|
pass |
|
|
return None |
|
|
|
|
|
|
|
|
def forum_menu() -> Optional[ForumMenuChoice]: |
|
|
print("\n" + SEPARATOR_60) |
|
|
print(" FORUM CLIENT - MAIN MENU") |
|
|
print(SEPARATOR_60) |
|
|
for choice in ForumMenuChoice: |
|
|
print(f"{choice.value}. {FORUM_MENU_TEXT[choice]}") |
|
|
print(DASH_LINE_60) |
|
|
raw = input(PROMPT).strip() |
|
|
try: |
|
|
return ForumMenuChoice(int(raw)) |
|
|
except Exception: |
|
|
return None |
|
|
|
|
|
|
|
|
def library_menu() -> Optional[LibraryMenuChoice]: |
|
|
print("\n" + SEPARATOR) |
|
|
for choice in LibraryMenuChoice: |
|
|
print(f"{choice.value}. {LIBRARY_MENU_TEXT[choice]}") |
|
|
raw = input(PROMPT).strip() |
|
|
try: |
|
|
return LibraryMenuChoice(int(raw)) |
|
|
except Exception: |
|
|
return None |
|
|
|
|
|
|
|
|
def run_forum_client(): |
|
|
client = ForumClient() |
|
|
try: |
|
|
while True: |
|
|
choice = forum_menu() |
|
|
if choice is None: |
|
|
print("Invalid option. Please try again.") |
|
|
continue |
|
|
if choice == ForumMenuChoice.BACK: |
|
|
break |
|
|
|
|
|
try: |
|
|
if choice == ForumMenuChoice.LIST_PARTICIPANTS: |
|
|
print(SEPARATOR_60) |
|
|
print("ALL PARTICIPANTS:") |
|
|
print_participants(client.get_participants()) |
|
|
|
|
|
elif choice == ForumMenuChoice.GET_PARTICIPANT: |
|
|
print(SEPARATOR_60) |
|
|
participant_id = input("Enter participant ID: ").strip() |
|
|
try: |
|
|
UUID(participant_id) |
|
|
participant = client.get_participant(participant_id) |
|
|
print(f"\nParticipant: {participant['first_name']} {participant['last_name']}") |
|
|
print(f" ID: {participant['id']}") |
|
|
print(f" Nickname: {participant['nickname']}") |
|
|
print(f" Rating: {participant['activity_rating']}") |
|
|
print(f" Registered: {participant['registered_at']}") |
|
|
except ValueError: |
|
|
print("Invalid UUID format.") |
|
|
|
|
|
elif choice == ForumMenuChoice.CREATE_PARTICIPANT: |
|
|
print(SEPARATOR_60) |
|
|
print("CREATE NEW PARTICIPANT:") |
|
|
first_name = prompt_with_validation(" First name: ", |
|
|
lambda v: v if re.fullmatch(r"[A-Za-zА-Яа-яЁё-]+", v) else (_ for _ in ()).throw( |
|
|
ValidationError("First name must contain only letters and '-'."))) |
|
|
last_name = prompt_with_validation(" Last name: ", |
|
|
lambda v: v if re.fullmatch(r"[A-Za-zА-Яа-яЁё-]+", v) else (_ for _ in ()).throw( |
|
|
ValidationError("Last name must contain only letters and '-'."))) |
|
|
nickname = prompt_with_validation(" Nickname: ", |
|
|
lambda v: v if re.fullmatch(r"[A-Za-zА-Яа-яЁё-]+", v) else (_ for _ in ()).throw( |
|
|
ValidationError("Nickname must contain only letters and '-'."))) |
|
|
activity_rating = prompt_float_with_validation(" Activity rating: ", validate_activity_rating) |
|
|
participant = client.create_participant(first_name, last_name, nickname, activity_rating) |
|
|
print("\nParticipant created successfully!") |
|
|
print(f" ID: {participant['id']}") |
|
|
|
|
|
elif choice == ForumMenuChoice.LIST_TOPICS: |
|
|
print(SEPARATOR_60) |
|
|
print("ALL TOPICS:") |
|
|
print_topics(client.get_topics()) |
|
|
|
|
|
elif choice == ForumMenuChoice.GET_TOPIC: |
|
|
print(SEPARATOR_60) |
|
|
topic_id = input("Enter topic ID: ").strip() |
|
|
try: |
|
|
UUID(topic_id) |
|
|
topic = client.get_topic(topic_id) |
|
|
print_topic_detail(topic) |
|
|
except ValueError: |
|
|
print("Invalid UUID format.") |
|
|
|
|
|
elif choice == ForumMenuChoice.CREATE_TOPIC: |
|
|
print(SEPARATOR_60) |
|
|
print("CREATE NEW TOPIC:") |
|
|
title = input(" Title: ").strip() |
|
|
description = input(" Description: ").strip() |
|
|
topic = client.create_topic(title, description) |
|
|
print("\nTopic created successfully!") |
|
|
print(f" ID: {topic['id']}") |
|
|
|
|
|
elif choice == ForumMenuChoice.PUBLISH_MESSAGE: |
|
|
print(SEPARATOR_60) |
|
|
print("PUBLISH MESSAGE TO TOPIC:") |
|
|
while True: |
|
|
topic_id = input(" Topic ID: ").strip() |
|
|
if not topic_id: |
|
|
print("Topic ID is required.") |
|
|
continue |
|
|
try: |
|
|
client.get_topic(topic_id) |
|
|
break |
|
|
except httpx.HTTPStatusError as e: |
|
|
if e.response.status_code in (404, 422): |
|
|
print("Topic not found.") |
|
|
else: |
|
|
print(f"Error: {e.response.status_code} - {e.response.text}") |
|
|
|
|
|
while True: |
|
|
participant_id = input(" Participant ID: ").strip() |
|
|
if not participant_id: |
|
|
print("Participant ID is required.") |
|
|
continue |
|
|
try: |
|
|
client.get_participant(participant_id) |
|
|
break |
|
|
except httpx.HTTPStatusError as e: |
|
|
if e.response.status_code in (404, 422): |
|
|
print("Participant not found.") |
|
|
else: |
|
|
print(f"Error: {e.response.status_code} - {e.response.text}") |
|
|
|
|
|
content = input(" Message content: ").strip() |
|
|
result = client.publish_message(topic_id, participant_id, content) |
|
|
if isinstance(result, str): |
|
|
print(f"\n{result}") |
|
|
else: |
|
|
print("\nMessage published successfully!") |
|
|
print(f"Topic now has {len(result['messages'])} message(s).") |
|
|
|
|
|
elif choice == ForumMenuChoice.LIST_MESSAGES: |
|
|
print(SEPARATOR_60) |
|
|
print("ALL MESSAGES:") |
|
|
print_messages(client.get_messages()) |
|
|
|
|
|
elif choice == ForumMenuChoice.DOWNLOAD_PARTICIPANT: |
|
|
print(SEPARATOR_60) |
|
|
participant_id = input("Enter participant ID: ").strip() |
|
|
try: |
|
|
UUID(participant_id) |
|
|
content = client.download_participant(participant_id) |
|
|
filepath = save_to_file("participants", participant_id, content) |
|
|
print(f"File saved to: {filepath}") |
|
|
except ValueError: |
|
|
print("Invalid UUID format.") |
|
|
|
|
|
elif choice == ForumMenuChoice.DOWNLOAD_TOPIC: |
|
|
print(SEPARATOR_60) |
|
|
topic_id = input("Enter topic ID: ").strip() |
|
|
try: |
|
|
UUID(topic_id) |
|
|
content = client.download_topic(topic_id) |
|
|
filepath = save_to_file("topics", topic_id, content) |
|
|
print(f"File saved to: {filepath}") |
|
|
except ValueError: |
|
|
print("Invalid UUID format.") |
|
|
|
|
|
elif choice == ForumMenuChoice.DOWNLOAD_MESSAGE: |
|
|
print(SEPARATOR_60) |
|
|
message_id = input("Enter message ID: ").strip() |
|
|
try: |
|
|
UUID(message_id) |
|
|
content = client.download_message(message_id) |
|
|
filepath = save_to_file("messages", message_id, content) |
|
|
print(f"File saved to: {filepath}") |
|
|
except ValueError: |
|
|
print("Invalid UUID format.") |
|
|
|
|
|
except httpx.HTTPStatusError as e: |
|
|
print(f"Error: {e.response.status_code} - {e.response.text}") |
|
|
finally: |
|
|
client.close() |
|
|
|
|
|
|
|
|
def run_library_client(): |
|
|
client = LibraryClient() |
|
|
try: |
|
|
while True: |
|
|
choice = library_menu() |
|
|
if choice is None: |
|
|
print("Invalid option.") |
|
|
continue |
|
|
if choice == LibraryMenuChoice.BACK: |
|
|
break |
|
|
|
|
|
try: |
|
|
if choice == LibraryMenuChoice.LIST_BOOKS: |
|
|
print_books(client.list_books()) |
|
|
|
|
|
elif choice == LibraryMenuChoice.VIEW_BOOK: |
|
|
book_id = prompt_book_id("Book ID: ", client) |
|
|
print_books([client.get_book(book_id)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.CREATE_BOOK: |
|
|
title = prompt_with_validation("Title: ", validate_title) |
|
|
author = prompt_with_validation("Author: ", validate_author) |
|
|
pages = prompt_int_with_validation("Pages: ", validate_pages) |
|
|
year = prompt_int_with_validation("Year: ", validate_year) |
|
|
genre = prompt_choice("Genre:", GENRES) |
|
|
status = prompt_choice("Status:", BOOK_STATUSES) |
|
|
print_books([client.create_book(title, author, pages, year, genre, status)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.UPDATE_BOOK: |
|
|
book_id = prompt_book_id("Book ID: ", client) |
|
|
print("Leave fields empty to skip them.") |
|
|
title = prompt_optional_with_validation("Title (empty to skip): ", validate_title) |
|
|
author = prompt_optional_with_validation("Author (empty to skip): ", validate_author) |
|
|
pages = prompt_optional_int_with_validation("Pages (empty to skip): ", validate_pages) |
|
|
year = prompt_optional_int_with_validation("Year (empty to skip): ", validate_year) |
|
|
genre = prompt_optional_choice("Genre:", GENRES) |
|
|
status = prompt_optional_choice("Status:", BOOK_STATUSES) |
|
|
if not any([title, author, pages, year, genre, status]): |
|
|
print("No fields to update.") |
|
|
continue |
|
|
print_books([client.update_book(book_id, title, author, pages, year, genre, status)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.DELETE_BOOK: |
|
|
book_id = prompt_book_id("Book ID: ", client) |
|
|
result = client.delete_book(book_id) |
|
|
print(result.get("message") if isinstance(result, dict) else "Deleted.") |
|
|
|
|
|
elif choice == LibraryMenuChoice.BORROW_BOOKS: |
|
|
book_ids = prompt_book_ids("Book IDs (comma separated): ", client) |
|
|
visitor_id = prompt_visitor_id("Visitor ID: ", client) |
|
|
worker_id = prompt_worker_id("Worker ID: ", client) |
|
|
borrow_date = prompt_date("Borrow date (YYYY-MM-DD)") |
|
|
result = client.borrow_books(book_ids, visitor_id, worker_id, borrow_date) |
|
|
print(result.get("message") if isinstance(result, dict) else result) |
|
|
|
|
|
elif choice == LibraryMenuChoice.RETURN_BOOKS: |
|
|
book_ids = prompt_book_ids("Book IDs (comma separated): ", client) |
|
|
visitor_id = prompt_visitor_id("Visitor ID: ", client) |
|
|
worker_id = prompt_worker_id("Worker ID: ", client) |
|
|
return_date = prompt_date("Return date (YYYY-MM-DD)") |
|
|
result = client.return_books(book_ids, visitor_id, worker_id, return_date) |
|
|
print(result.get("message") if isinstance(result, dict) else result) |
|
|
|
|
|
elif choice == LibraryMenuChoice.DOWNLOAD_BOOK: |
|
|
book_id = prompt_book_id("Book ID: ", client) |
|
|
content = client.download_book(book_id) |
|
|
filepath = save_to_file("books", book_id, content) |
|
|
print(f"File saved to: {filepath}") |
|
|
|
|
|
elif choice == LibraryMenuChoice.LIST_VISITORS: |
|
|
print_visitors(client.list_visitors()) |
|
|
|
|
|
elif choice == LibraryMenuChoice.VIEW_VISITOR: |
|
|
visitor_id = prompt_visitor_id("Visitor ID: ", client) |
|
|
visitor = client.get_visitor(visitor_id) |
|
|
print_visitors([visitor]) |
|
|
current = visitor.get("currentBooks") or [] |
|
|
history = visitor.get("history") or [] |
|
|
if current: |
|
|
print("\nCurrent books:") |
|
|
print_books(current) |
|
|
if history: |
|
|
print("\nHistory:") |
|
|
print_books(history) |
|
|
|
|
|
elif choice == LibraryMenuChoice.CREATE_VISITOR: |
|
|
name = prompt_with_validation("Name: ", lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_with_validation("Surname: ", lambda v: validate_name(v, "Surname")) |
|
|
print_visitors([client.create_visitor(name, surname)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.UPDATE_VISITOR: |
|
|
visitor_id = prompt_visitor_id("Visitor ID: ", client) |
|
|
print("Leave fields empty to skip them.") |
|
|
name = prompt_optional_with_validation("Name (empty to skip): ", lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_optional_with_validation("Surname (empty to skip): ", lambda v: validate_name(v, "Surname")) |
|
|
if not name and not surname: |
|
|
print("No fields to update.") |
|
|
continue |
|
|
print_visitors([client.update_visitor(visitor_id, name, surname)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.DELETE_VISITOR: |
|
|
visitor_id = prompt_visitor_id("Visitor ID: ", client) |
|
|
client.delete_visitor(visitor_id) |
|
|
print("Visitor deleted.") |
|
|
|
|
|
elif choice == LibraryMenuChoice.DOWNLOAD_VISITOR: |
|
|
visitor_id = prompt_visitor_id("Visitor ID: ", client) |
|
|
content = client.download_visitor(visitor_id) |
|
|
filepath = save_to_file("visitors", visitor_id, content) |
|
|
print(f"File saved to: {filepath}") |
|
|
|
|
|
elif choice == LibraryMenuChoice.LIST_WORKERS: |
|
|
print_workers(client.list_workers()) |
|
|
|
|
|
elif choice == LibraryMenuChoice.VIEW_WORKER: |
|
|
worker_id = prompt_worker_id("Worker ID: ", 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 == LibraryMenuChoice.CREATE_WORKER: |
|
|
name = prompt_with_validation("Name: ", lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_with_validation("Surname: ", lambda v: validate_name(v, "Surname")) |
|
|
experience = prompt_int_with_validation("Experience (years): ", validate_experience) |
|
|
work_days = prompt_multi_choice("Work days:", WORK_DAYS) |
|
|
print_workers([client.create_worker(name, surname, experience, work_days)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.UPDATE_WORKER: |
|
|
worker_id = prompt_worker_id("Worker ID: ", client) |
|
|
print("Leave fields empty to skip them.") |
|
|
name = prompt_optional_with_validation("Name (empty to skip): ", lambda v: validate_name(v, "Name")) |
|
|
surname = prompt_optional_with_validation("Surname (empty to skip): ", lambda v: validate_name(v, "Surname")) |
|
|
experience = prompt_optional_int_with_validation("Experience (empty to skip): ", validate_experience) |
|
|
work_days = prompt_optional_multi_choice("Work days:", WORK_DAYS) |
|
|
if not any([name, surname, experience is not None, work_days]): |
|
|
print("No fields to update.") |
|
|
continue |
|
|
print_workers([client.update_worker(worker_id, name, surname, experience, work_days)]) |
|
|
|
|
|
elif choice == LibraryMenuChoice.DELETE_WORKER: |
|
|
worker_id = prompt_worker_id("Worker ID: ", client) |
|
|
client.delete_worker(worker_id) |
|
|
print("Worker deleted.") |
|
|
|
|
|
elif choice == LibraryMenuChoice.WORKERS_BY_DAYS: |
|
|
work_days = prompt_multi_choice("Select work days:", WORK_DAYS) |
|
|
print_workers(client.workers_by_days(work_days)) |
|
|
|
|
|
elif choice == LibraryMenuChoice.DOWNLOAD_WORKER: |
|
|
worker_id = prompt_worker_id("Worker ID: ", client) |
|
|
content = client.download_worker(worker_id) |
|
|
filepath = save_to_file("workers", worker_id, content) |
|
|
print(f"File saved to: {filepath}") |
|
|
|
|
|
except (RuntimeError, httpx.HTTPError, ValueError) as error: |
|
|
print(f"Error: {error}") |
|
|
finally: |
|
|
client.close() |
|
|
|
|
|
|
|
|
def main(): |
|
|
try: |
|
|
while True: |
|
|
server_choice = server_selection_menu() |
|
|
if server_choice is None: |
|
|
print("Invalid option. Please try again.") |
|
|
continue |
|
|
if server_choice == 0: |
|
|
print("Goodbye!") |
|
|
break |
|
|
elif server_choice == 1: |
|
|
run_forum_client() |
|
|
elif server_choice == 2: |
|
|
run_library_client() |
|
|
except KeyboardInterrupt: |
|
|
print("\nInterrupted. Goodbye!") |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|
|
|
|
