| 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() |
|
|
|
|