diff --git a/actions/rag-basic.yaml b/actions/rag-basic.yaml new file mode 100644 index 00000000..b634671e --- /dev/null +++ b/actions/rag-basic.yaml @@ -0,0 +1,15 @@ +name: local/basic-rag +description: > + Build a basic RAG (Retrieval Augmented Generation) application in Python. + Use OpenAI embeddings for vectorizing documents and FAISS for the vector store. + Include a document loader that reads .txt files from a directory, chunks them, + embeds them, stores in FAISS, and answers questions by retrieving relevant + chunks and passing them to OpenAI GPT for answer generation. Include a CLI + interface that accepts a question as input. Include requirements.txt. +strategy_actor: anthropic/claude-sonnet-4-20250514 +execution_actor: anthropic/claude-sonnet-4-20250514 +definition_of_done: > + A rag.py implements document loading, chunking, embedding with OpenAI, + FAISS vector storage, retrieval, and answer generation. + CLI accepts a question via argparse and a --docs-dir flag for the documents directory. + A requirements.txt lists all dependencies (openai, faiss-cpu, tiktoken, etc). diff --git a/run_all_sims.py b/run_all_sims.py new file mode 100644 index 00000000..09204fac --- /dev/null +++ b/run_all_sims.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Run all 7 simulations through the V3 lifecycle, collecting files after each.""" + +import os +import shutil +import subprocess +import sys +import time +from pathlib import Path + +CLI = [sys.executable, "-m", "cleveragents.cli.main"] +PROJECT_ROOT = Path.cwd() + +SIMS = [ + ("sim1", "local/todo-cli"), + ("sim2", "local/bookstore-api"), + ("sim3", "local/websocket-chat"), + ("sim4", "local/hn-scraper"), + ("sim5", "local/flask-auth"), + ("sim6", "local/weather-cli"), + ("sim7", "local/url-shortener"), +] + +# Track files that existed before we started +IGNORE_PATTERNS = { + ".venv", ".git", "src", "docs", "actions", "simulations", ".cleveragents", + ".claude", ".env", ".gitignore", "run_all_sims.py", "features", "tests", + "benchmarks", "pyproject.toml", "poetry.lock", "alembic.ini", "Makefile", + "Dockerfile", "docker-compose.yml", "README.md", "CLAUDE.md", "LICENSE", + "cleveragents.db", "__pycache__", +} + + +def run(args, timeout=300): + r = subprocess.run(CLI + args, capture_output=True, text=True, timeout=timeout) + return r.stdout + r.stderr + + +def extract_plan_id(output): + for line in output.splitlines(): + if "ID:" in line: + parts = line.split("ID:") + if len(parts) > 1: + return parts[1].strip().strip("│").strip() + return None + + +def extract_file_count(output): + for line in output.splitlines(): + if "file_count=" in line: + for part in line.split(): + if part.startswith("file_count="): + return int(part.split("=")[1]) + return 0 + + +def snapshot_cwd(): + """Return set of all files in CWD (relative paths), excluding known dirs.""" + files = set() + for item in PROJECT_ROOT.iterdir(): + if item.name in IGNORE_PATTERNS: + continue + if item.is_file(): + files.add(item.name) + elif item.is_dir(): + for f in item.rglob("*"): + if f.is_file(): + files.add(str(f.relative_to(PROJECT_ROOT))) + return files + + +def collect_new_files(before_snapshot, sim_dir): + """Move newly created files to sim_dir.""" + after = snapshot_cwd() + new_files = after - before_snapshot + sim_dir.mkdir(parents=True, exist_ok=True) + + for rel_path in sorted(new_files): + src = PROJECT_ROOT / rel_path + dst = sim_dir / rel_path + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(src), str(dst)) + + # Clean up any empty dirs left behind + for rel_path in sorted(new_files, reverse=True): + src_parent = (PROJECT_ROOT / rel_path).parent + if src_parent != PROJECT_ROOT and src_parent.exists(): + try: + src_parent.rmdir() # only removes if empty + except OSError: + pass + + return sorted(new_files) + + +results = {} + +for sim_name, action_name in SIMS: + print(f"\n{'='*60}") + print(f" {sim_name.upper()}: {action_name}") + print(f"{'='*60}") + + sim_dir = PROJECT_ROOT / "simulations" / sim_name + if sim_dir.exists(): + shutil.rmtree(sim_dir) + + t0 = time.time() + + # Plan use + out = run(["plan", "use", action_name]) + plan_id = extract_plan_id(out) + if not plan_id: + print(f" FAILED: Could not extract plan ID") + results[sim_name] = {"status": "FAILED", "error": "No plan ID"} + continue + print(f" Plan ID: {plan_id}") + + # Execute + print(f" Executing...", end="", flush=True) + before = snapshot_cwd() + out = run(["plan", "execute", plan_id], timeout=300) + elapsed = time.time() - t0 + file_count = extract_file_count(out) + + if "Execute completed" not in out and "Plan Executed" not in out: + print(f" FAILED ({elapsed:.0f}s)") + results[sim_name] = {"status": "FAILED", "error": out[-300:]} + continue + print(f" done ({elapsed:.0f}s, {file_count} files in sandbox)") + + # Apply + out = run(["plan", "lifecycle-apply", plan_id]) + if "file(s) written" not in out: + print(f" Apply FAILED") + results[sim_name] = {"status": "FAILED", "error": "Apply failed"} + continue + + # Collect files + new_files = collect_new_files(before, sim_dir) + print(f" Applied {len(new_files)} files:") + for f in new_files: + print(f" {f}") + + results[sim_name] = { + "status": "PASS", + "plan_id": plan_id, + "sandbox_count": file_count, + "applied_files": new_files, + "elapsed": f"{elapsed:.0f}s", + } + +print(f"\n{'='*60}") +print(" SUMMARY") +print(f"{'='*60}") +for sim, info in results.items(): + status = info["status"] + files = info.get("applied_files", []) + elapsed = info.get("elapsed", "?") + print(f" {sim}: {status} — {len(files)} files ({elapsed})") + for f in files: + print(f" {f}") diff --git a/simulations/sim1/storage.py b/simulations/sim1/storage.py deleted file mode 100644 index 567aaace..00000000 --- a/simulations/sim1/storage.py +++ /dev/null @@ -1,79 +0,0 @@ -import json -import os -from typing import List, Dict, Any, Optional -from task import Task - - -class TaskStorage: - """Handles persistence of tasks to JSON file.""" - - def __init__(self, filename: str = "tasks.json"): - self.filename = filename - self._ensure_file_exists() - - def _ensure_file_exists(self) -> None: - """Create the JSON file if it doesn't exist.""" - if not os.path.exists(self.filename): - with open(self.filename, 'w') as f: - json.dump([], f) - - def load_tasks(self) -> List[Task]: - """Load all tasks from the JSON file.""" - try: - with open(self.filename, 'r') as f: - data = json.load(f) - return [Task.from_dict(task_data) for task_data in data] - except (json.JSONDecodeError, FileNotFoundError) as e: - print(f"Error loading tasks: {e}") - return [] - except Exception as e: - print(f"Unexpected error loading tasks: {e}") - return [] - - def save_tasks(self, tasks: List[Task]) -> bool: - """Save all tasks to the JSON file.""" - try: - task_dicts = [task.to_dict() for task in tasks] - with open(self.filename, 'w') as f: - json.dump(task_dicts, f, indent=2) - return True - except Exception as e: - print(f"Error saving tasks: {e}") - return False - - def add_task(self, task: Task) -> bool: - """Add a single task to storage.""" - tasks = self.load_tasks() - tasks.append(task) - return self.save_tasks(tasks) - - def get_task_by_id(self, task_id: str) -> Optional[Task]: - """Retrieve a task by its ID.""" - tasks = self.load_tasks() - for task in tasks: - if task.id == task_id or task.id.startswith(task_id): - return task - return None - - def update_task(self, updated_task: Task) -> bool: - """Update an existing task in storage.""" - tasks = self.load_tasks() - for i, task in enumerate(tasks): - if task.id == updated_task.id: - tasks[i] = updated_task - return self.save_tasks(tasks) - return False - - def delete_task(self, task_id: str) -> bool: - """Delete a task by its ID.""" - tasks = self.load_tasks() - original_count = len(tasks) - tasks = [task for task in tasks if not (task.id == task_id or task.id.startswith(task_id))] - - if len(tasks) < original_count: - return self.save_tasks(tasks) - return False - - def get_all_tasks(self) -> List[Task]: - """Get all tasks from storage.""" - return self.load_tasks() \ No newline at end of file diff --git a/simulations/sim1/task.py b/simulations/sim1/task.py deleted file mode 100644 index 45cd8fe3..00000000 --- a/simulations/sim1/task.py +++ /dev/null @@ -1,49 +0,0 @@ -from dataclasses import dataclass, asdict -from datetime import datetime -from typing import Dict, Any -import uuid - - -@dataclass -class Task: - id: str - title: str - completed: bool - created_at: str - - @classmethod - def create(cls, title: str) -> 'Task': - """Create a new task with generated ID and current timestamp.""" - return cls( - id=str(uuid.uuid4()), - title=title, - completed=False, - created_at=datetime.now().isoformat() - ) - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> 'Task': - """Create a Task instance from dictionary data.""" - return cls(**data) - - def to_dict(self) -> Dict[str, Any]: - """Convert task to dictionary for JSON serialization.""" - return asdict(self) - - def toggle_completion(self) -> None: - """Toggle the completion status of the task.""" - self.completed = not self.completed - - def mark_complete(self) -> None: - """Mark the task as completed.""" - self.completed = True - - def mark_incomplete(self) -> None: - """Mark the task as incomplete.""" - self.completed = False - - def __str__(self) -> str: - """String representation of the task.""" - status = "✓" if self.completed else "○" - created = datetime.fromisoformat(self.created_at).strftime("%Y-%m-%d %H:%M") - return f"[{status}] {self.title} (ID: {self.id[:8]}..., Created: {created})" \ No newline at end of file diff --git a/simulations/sim1/todo.py b/simulations/sim1/todo.py index 3f60daed..079e5bf5 100644 --- a/simulations/sim1/todo.py +++ b/simulations/sim1/todo.py @@ -1,130 +1,253 @@ +#!/usr/bin/env python3 + import argparse -import sys -from typing import Optional -from task import Task -from storage import TaskStorage +import json +import uuid +from dataclasses import dataclass, asdict +from datetime import datetime +from pathlib import Path +from typing import List, Optional -class TodoCLI: - """Command-line interface for the todo application.""" - - def __init__(self): - self.storage = TaskStorage() +@dataclass +class Task: + id: str + title: str + completed: bool + created_at: str - def add_task(self, title: str) -> None: - """Add a new task.""" - if not title.strip(): - print("Error: Task title cannot be empty.") - return - - task = Task.create(title.strip()) - if self.storage.add_task(task): - print(f"Task added: {task.title}") - print(f"ID: {task.id}") - else: - print("Error: Failed to add task.") - - def list_tasks(self) -> None: - """List all tasks.""" - tasks = self.storage.get_all_tasks() - - if not tasks: - print("No tasks found.") - return - - print(f"\nFound {len(tasks)} task(s):") - print("-" * 60) - - # Sort tasks: incomplete first, then by creation date - sorted_tasks = sorted(tasks, key=lambda t: (t.completed, t.created_at)) - - for task in sorted_tasks: - print(task) - - def complete_task(self, task_id: str) -> None: - """Mark a task as completed.""" - task = self.storage.get_task_by_id(task_id) - - if not task: - print(f"Error: Task with ID '{task_id}' not found.") - return - - if task.completed: - print(f"Task '{task.title}' is already completed.") - else: - task.mark_complete() - if self.storage.update_task(task): - print(f"Task completed: {task.title}") - else: - print("Error: Failed to update task.") - - def delete_task(self, task_id: str) -> None: - """Delete a task.""" - task = self.storage.get_task_by_id(task_id) - - if not task: - print(f"Error: Task with ID '{task_id}' not found.") - return - - if self.storage.delete_task(task_id): - print(f"Task deleted: {task.title}") - else: - print("Error: Failed to delete task.") - - def run(self) -> None: - """Main entry point for the CLI application.""" - parser = argparse.ArgumentParser( - description="A simple command-line todo application", - prog="todo" + @classmethod + def create(cls, title: str) -> 'Task': + """Create a new task with auto-generated ID and timestamp.""" + return cls( + id=str(uuid.uuid4()), + title=title, + completed=False, + created_at=datetime.now().isoformat() ) - subparsers = parser.add_subparsers(dest="command", help="Available commands") - # Add command - add_parser = subparsers.add_parser("add", help="Add a new task") - add_parser.add_argument("title", nargs="+", help="Task title") + @classmethod + def from_dict(cls, data: dict) -> 'Task': + """Create Task instance from dictionary.""" + return cls(**data) - # List command - subparsers.add_parser("list", help="List all tasks") - # Complete command - complete_parser = subparsers.add_parser("complete", help="Mark a task as completed") - complete_parser.add_argument("id", help="Task ID (full ID or partial ID)") +class TaskManager: + def __init__(self, storage_file: str = "tasks.json"): + self.storage_file = Path(storage_file) + self.tasks: List[Task] = [] + self.load_tasks() - # Delete command - delete_parser = subparsers.add_parser("delete", help="Delete a task") - delete_parser.add_argument("id", help="Task ID (full ID or partial ID)") - - # Parse arguments - if len(sys.argv) == 1: - parser.print_help() - return - - args = parser.parse_args() - - # Route to appropriate command handler + def load_tasks(self) -> None: + """Load tasks from JSON file.""" try: - if args.command == "add": - title = " ".join(args.title) - self.add_task(title) - elif args.command == "list": - self.list_tasks() - elif args.command == "complete": - self.complete_task(args.id) - elif args.command == "delete": - self.delete_task(args.id) + if self.storage_file.exists(): + with open(self.storage_file, 'r', encoding='utf-8') as f: + data = json.load(f) + self.tasks = [Task.from_dict(task_dict) for task_dict in data] else: - parser.print_help() - except KeyboardInterrupt: - print("\nOperation cancelled.") - except Exception as e: - print(f"An error occurred: {e}") + self.tasks = [] + except (json.JSONDecodeError, KeyError) as e: + print(f"Error loading tasks: {e}") + print("Starting with empty task list.") + self.tasks = [] + + def save_tasks(self) -> None: + """Save tasks to JSON file.""" + try: + with open(self.storage_file, 'w', encoding='utf-8') as f: + json.dump([asdict(task) for task in self.tasks], f, indent=2, ensure_ascii=False) + except IOError as e: + print(f"Error saving tasks: {e}") + + def add_task(self, title: str) -> Task: + """Add a new task.""" + task = Task.create(title) + self.tasks.append(task) + self.save_tasks() + return task + + def list_tasks(self) -> List[Task]: + """Get all tasks.""" + return self.tasks + + def find_task(self, task_id: str) -> Optional[Task]: + """Find task by ID.""" + for task in self.tasks: + if task.id == task_id: + return task + return None + + def complete_task(self, task_id: str) -> bool: + """Mark task as completed.""" + task = self.find_task(task_id) + if task: + task.completed = True + self.save_tasks() + return True + return False + + def delete_task(self, task_id: str) -> bool: + """Delete task by ID.""" + task = self.find_task(task_id) + if task: + self.tasks.remove(task) + self.save_tasks() + return True + return False + + +def format_task(task: Task) -> str: + """Format task for display.""" + status = "✓" if task.completed else "○" + created_date = datetime.fromisoformat(task.created_at).strftime("%Y-%m-%d %H:%M") + return f"{status} [{task.id[:8]}] {task.title} (created: {created_date})" + + +def cmd_add(args, task_manager: TaskManager) -> None: + """Handle add command.""" + if not args.title: + print("Error: Task title cannot be empty.") + return + + task = task_manager.add_task(args.title) + print(f"Added task: {task.title}") + print(f"Task ID: {task.id}") + + +def cmd_list(args, task_manager: TaskManager) -> None: + """Handle list command.""" + tasks = task_manager.list_tasks() + + if not tasks: + print("No tasks found.") + return + + print(f"Total tasks: {len(tasks)}") + print() + + for task in tasks: + print(format_task(task)) + + +def cmd_complete(args, task_manager: TaskManager) -> None: + """Handle complete command.""" + if task_manager.complete_task(args.id): + task = task_manager.find_task(args.id) + print(f"Completed task: {task.title}") + else: + print(f"Error: Task with ID '{args.id}' not found.") + + +def cmd_delete(args, task_manager: TaskManager) -> None: + """Handle delete command.""" + task = task_manager.find_task(args.id) + if task and task_manager.delete_task(args.id): + print(f"Deleted task: {task.title}") + else: + print(f"Error: Task with ID '{args.id}' not found.") + + +def create_parser() -> argparse.ArgumentParser: + """Create and configure argument parser.""" + parser = argparse.ArgumentParser( + description="Simple command-line TODO application", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + %(prog)s add "Buy groceries" + %(prog)s list + %(prog)s complete abc12345 + %(prog)s delete abc12345 + """ + ) + + subparsers = parser.add_subparsers(dest='command', help='Available commands') + + # Add command + add_parser = subparsers.add_parser('add', help='Add a new task') + add_parser.add_argument('title', help='Task title') + + # List command + subparsers.add_parser('list', help='List all tasks') + + # Complete command + complete_parser = subparsers.add_parser('complete', help='Mark task as completed') + complete_parser.add_argument('id', help='Task ID (or first 8 characters)') + + # Delete command + delete_parser = subparsers.add_parser('delete', help='Delete a task') + delete_parser.add_argument('id', help='Task ID (or first 8 characters)') + + return parser + + +def find_task_by_partial_id(task_manager: TaskManager, partial_id: str) -> Optional[Task]: + """Find task by partial ID (supports short IDs).""" + # First try exact match + task = task_manager.find_task(partial_id) + if task: + return task + + # Try partial match + matching_tasks = [task for task in task_manager.tasks if task.id.startswith(partial_id)] + + if len(matching_tasks) == 1: + return matching_tasks[0] + elif len(matching_tasks) > 1: + print(f"Error: Multiple tasks match ID '{partial_id}':") + for task in matching_tasks: + print(f" {task.id[:8]} - {task.title}") + return None + + return None def main(): - """Entry point for the application.""" - cli = TodoCLI() - cli.run() + """Main application entry point.""" + parser = create_parser() + args = parser.parse_args() + + if not args.command: + parser.print_help() + return + + task_manager = TaskManager() + + try: + if args.command == 'add': + cmd_add(args, task_manager) + elif args.command == 'list': + cmd_list(args, task_manager) + elif args.command == 'complete': + # Handle partial ID matching + task = find_task_by_partial_id(task_manager, args.id) + if task: + if task_manager.complete_task(task.id): + print(f"Completed task: {task.title}") + else: + print(f"Error: Could not complete task '{task.id}'.") + else: + print(f"Error: Task with ID '{args.id}' not found.") + elif args.command == 'delete': + # Handle partial ID matching + task = find_task_by_partial_id(task_manager, args.id) + if task: + task_title = task.title + if task_manager.delete_task(task.id): + print(f"Deleted task: {task_title}") + else: + print(f"Error: Could not delete task '{task.id}'.") + else: + print(f"Error: Task with ID '{args.id}' not found.") + else: + parser.print_help() + + except KeyboardInterrupt: + print("\nOperation cancelled.") + except Exception as e: + print(f"An unexpected error occurred: {e}") -if __name__ == "__main__": +if __name__ == '__main__': main() \ No newline at end of file diff --git a/simulations/sim2/README.md b/simulations/sim2/README.md deleted file mode 100644 index 0f265870..00000000 --- a/simulations/sim2/README.md +++ /dev/null @@ -1,15 +0,0 @@ -# Book Management API - -A RESTful API for managing books built with FastAPI. - -## Features - -- Full CRUD operations for books -- Input validation with Pydantic -- Thread-safe in-memory storage -- Comprehensive error handling -- Interactive API documentation - -## Installation - -1. Install dependencies: \ No newline at end of file diff --git a/simulations/sim2/main.py b/simulations/sim2/main.py index c159344b..9ed8a821 100644 --- a/simulations/sim2/main.py +++ b/simulations/sim2/main.py @@ -1,175 +1,120 @@ -from fastapi import FastAPI, HTTPException, status -from fastapi.responses import JSONResponse -from typing import List, Dict -import threading -from models import Book, BookCreate, BookUpdate +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel, Field +from typing import Dict, List, Optional +import uvicorn -# Thread-safe in-memory storage -class BookStorage: - def __init__(self): - self._books: Dict[int, Book] = {} - self._next_id = 1 - self._lock = threading.RLock() - - def create_book(self, book_data: BookCreate) -> Book: - with self._lock: - # Check for duplicate ISBN - for existing_book in self._books.values(): - if existing_book.isbn == book_data.isbn: - raise ValueError(f"Book with ISBN {book_data.isbn} already exists") - - book = Book(id=self._next_id, **book_data.dict()) - self._books[self._next_id] = book - self._next_id += 1 - return book - - def get_book(self, book_id: int) -> Book: - with self._lock: - if book_id not in self._books: - raise KeyError(f"Book with id {book_id} not found") - return self._books[book_id] - - def get_all_books(self) -> List[Book]: - with self._lock: - return list(self._books.values()) - - def update_book(self, book_id: int, book_data: BookUpdate) -> Book: - with self._lock: - if book_id not in self._books: - raise KeyError(f"Book with id {book_id} not found") - - # Check for duplicate ISBN if updating ISBN - if book_data.isbn: - for existing_id, existing_book in self._books.items(): - if existing_id != book_id and existing_book.isbn == book_data.isbn: - raise ValueError(f"Book with ISBN {book_data.isbn} already exists") - - current_book = self._books[book_id] - update_data = book_data.dict(exclude_unset=True) - updated_book = current_book.copy(update=update_data) - self._books[book_id] = updated_book - return updated_book - - def delete_book(self, book_id: int) -> bool: - with self._lock: - if book_id not in self._books: - raise KeyError(f"Book with id {book_id} not found") - del self._books[book_id] - return True - -# Initialize FastAPI app and storage app = FastAPI( - title="Book Management API", - description="A RESTful API for managing books with CRUD operations", + title="Bookstore API", + description="A simple REST API for managing books in a bookstore", version="1.0.0" ) -book_storage = BookStorage() +# Pydantic models +class Book(BaseModel): + id: int = Field(..., description="Unique identifier for the book") + title: str = Field(..., min_length=1, description="Title of the book") + author: str = Field(..., min_length=1, description="Author of the book") + price: float = Field(..., gt=0, description="Price of the book in dollars") + isbn: str = Field(..., min_length=10, max_length=17, description="ISBN of the book") -# Exception handlers -@app.exception_handler(ValueError) -async def value_error_handler(request, exc): - return JSONResponse( - status_code=status.HTTP_400_BAD_REQUEST, - content={"detail": str(exc)} - ) +class BookCreate(BaseModel): + title: str = Field(..., min_length=1, description="Title of the book") + author: str = Field(..., min_length=1, description="Author of the book") + price: float = Field(..., gt=0, description="Price of the book in dollars") + isbn: str = Field(..., min_length=10, max_length=17, description="ISBN of the book") -@app.exception_handler(KeyError) -async def key_error_handler(request, exc): - return JSONResponse( - status_code=status.HTTP_404_NOT_FOUND, - content={"detail": str(exc)} - ) +class BookUpdate(BaseModel): + title: Optional[str] = Field(None, min_length=1, description="Title of the book") + author: Optional[str] = Field(None, min_length=1, description="Author of the book") + price: Optional[float] = Field(None, gt=0, description="Price of the book in dollars") + isbn: Optional[str] = Field(None, min_length=10, max_length=17, description="ISBN of the book") -# API Endpoints -@app.get("/books", response_model=List[Book], tags=["Books"]) -async def get_all_books(): - """Retrieve all books""" - return book_storage.get_all_books() +# In-memory storage +books_db: Dict[int, Book] = {} +next_book_id = 1 -@app.get("/books/{book_id}", response_model=Book, tags=["Books"]) -async def get_book(book_id: int): - """Retrieve a specific book by ID""" - if book_id <= 0: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Book ID must be a positive integer" - ) - - try: - return book_storage.get_book(book_id) - except KeyError as e: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(e) - ) +# Sample data +sample_books = [ + {"title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "price": 12.99, "isbn": "978-0-7432-7356-5"}, + {"title": "To Kill a Mockingbird", "author": "Harper Lee", "price": 14.99, "isbn": "978-0-06-112008-4"}, + {"title": "1984", "author": "George Orwell", "price": 13.99, "isbn": "978-0-452-28423-4"} +] -@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED, tags=["Books"]) -async def create_book(book: BookCreate): - """Create a new book""" - try: - return book_storage.create_book(book) - except ValueError as e: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=str(e) - ) +# Initialize sample data +for book_data in sample_books: + book = Book(id=next_book_id, **book_data) + books_db[next_book_id] = book + next_book_id += 1 -@app.put("/books/{book_id}", response_model=Book, tags=["Books"]) -async def update_book(book_id: int, book: BookUpdate): - """Update an existing book""" - if book_id <= 0: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Book ID must be a positive integer" - ) - - # Check if at least one field is provided for update - if not any(book.dict(exclude_unset=True).values()): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="At least one field must be provided for update" - ) - - try: - return book_storage.update_book(book_id, book) - except KeyError as e: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(e) - ) - except ValueError as e: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=str(e) - ) - -@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["Books"]) -async def delete_book(book_id: int): - """Delete a book by ID""" - if book_id <= 0: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Book ID must be a positive integer" - ) - - try: - book_storage.delete_book(book_id) - except KeyError as e: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(e) - ) - -@app.get("/", tags=["Root"]) +@app.get("/") async def root(): - """API root endpoint""" - return { - "message": "Book Management API", - "version": "1.0.0", - "docs": "/docs" - } + return {"message": "Welcome to the Bookstore API", "docs": "/docs"} + +@app.get("/books", response_model=List[Book]) +async def get_all_books(): + """Retrieve all books from the bookstore.""" + return list(books_db.values()) + +@app.get("/books/{book_id}", response_model=Book) +async def get_book(book_id: int): + """Retrieve a specific book by its ID.""" + if book_id not in books_db: + raise HTTPException(status_code=404, detail="Book not found") + return books_db[book_id] + +@app.post("/books", response_model=Book, status_code=201) +async def create_book(book: BookCreate): + """Create a new book in the bookstore.""" + global next_book_id + + # Check if ISBN already exists + for existing_book in books_db.values(): + if existing_book.isbn == book.isbn: + raise HTTPException(status_code=400, detail="Book with this ISBN already exists") + + new_book = Book(id=next_book_id, **book.dict()) + books_db[next_book_id] = new_book + next_book_id += 1 + + return new_book + +@app.put("/books/{book_id}", response_model=Book) +async def update_book(book_id: int, book_update: BookUpdate): + """Update an existing book by its ID.""" + if book_id not in books_db: + raise HTTPException(status_code=404, detail="Book not found") + + existing_book = books_db[book_id] + + # Check if ISBN already exists for another book + if book_update.isbn: + for bid, existing in books_db.items(): + if bid != book_id and existing.isbn == book_update.isbn: + raise HTTPException(status_code=400, detail="Book with this ISBN already exists") + + # Update only provided fields + update_data = book_update.dict(exclude_unset=True) + updated_book_data = existing_book.dict() + updated_book_data.update(update_data) + + updated_book = Book(**updated_book_data) + books_db[book_id] = updated_book + + return updated_book + +@app.delete("/books/{book_id}", status_code=204) +async def delete_book(book_id: int): + """Delete a book from the bookstore by its ID.""" + if book_id not in books_db: + raise HTTPException(status_code=404, detail="Book not found") + + del books_db[book_id] + return None + +# Health check endpoint +@app.get("/health") +async def health_check(): + return {"status": "healthy", "total_books": len(books_db)} if __name__ == "__main__": - import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) \ No newline at end of file diff --git a/simulations/sim2/models.py b/simulations/sim2/models.py deleted file mode 100644 index 63eb83c3..00000000 --- a/simulations/sim2/models.py +++ /dev/null @@ -1,67 +0,0 @@ -from pydantic import BaseModel, Field, validator -from typing import Optional -import re - -class BookBase(BaseModel): - title: str = Field(..., min_length=1, max_length=200, description="Book title") - author: str = Field(..., min_length=1, max_length=100, description="Book author") - price: float = Field(..., gt=0, description="Book price (must be positive)") - isbn: str = Field(..., min_length=10, max_length=17, description="ISBN-10 or ISBN-13") - - @validator('isbn') - def validate_isbn(cls, v): - # Remove hyphens and spaces for validation - isbn_clean = re.sub(r'[-\s]', '', v) - - # Check if it's numeric and correct length - if not isbn_clean.isdigit(): - raise ValueError('ISBN must contain only digits, hyphens, and spaces') - - if len(isbn_clean) not in [10, 13]: - raise ValueError('ISBN must be 10 or 13 digits long') - - return v - - @validator('price') - def validate_price(cls, v): - # Round to 2 decimal places for currency - return round(v, 2) - -class BookCreate(BookBase): - pass - -class BookUpdate(BaseModel): - title: Optional[str] = Field(None, min_length=1, max_length=200) - author: Optional[str] = Field(None, min_length=1, max_length=100) - price: Optional[float] = Field(None, gt=0) - isbn: Optional[str] = Field(None, min_length=10, max_length=17) - - @validator('isbn') - def validate_isbn(cls, v): - if v is not None: - isbn_clean = re.sub(r'[-\s]', '', v) - if not isbn_clean.isdigit(): - raise ValueError('ISBN must contain only digits, hyphens, and spaces') - if len(isbn_clean) not in [10, 13]: - raise ValueError('ISBN must be 10 or 13 digits long') - return v - - @validator('price') - def validate_price(cls, v): - if v is not None: - return round(v, 2) - return v - -class Book(BookBase): - id: int = Field(..., description="Unique book identifier") - - class Config: - schema_extra = { - "example": { - "id": 1, - "title": "The Python Guide", - "author": "John Doe", - "price": 29.99, - "isbn": "978-0123456789" - } - } \ No newline at end of file diff --git a/simulations/sim2/requirements.txt b/simulations/sim2/requirements.txt index c06c1b64..85205d6e 100644 --- a/simulations/sim2/requirements.txt +++ b/simulations/sim2/requirements.txt @@ -1,3 +1,3 @@ fastapi>=0.104.0 -uvicorn[standard]>=0.24.0 +uvicorn>=0.24.0 pydantic>=2.0.0 \ No newline at end of file diff --git a/simulations/sim3/index.html b/simulations/sim3/index.html new file mode 100644 index 00000000..88d5ab99 --- /dev/null +++ b/simulations/sim3/index.html @@ -0,0 +1,423 @@ + + + + + + WebSocket Chat + + + +
+
+

WebSocket Chat Room

+
+ +
+ Disconnected +
+ +
+
+
Welcome to the chat room! Enter your username and click Connect to start chatting.
+
+
+ +
+
+ + + +
+
+ + + +
+
+
+ + + + \ No newline at end of file diff --git a/simulations/sim3/requirements.txt b/simulations/sim3/requirements.txt new file mode 100644 index 00000000..f4fc2af4 --- /dev/null +++ b/simulations/sim3/requirements.txt @@ -0,0 +1 @@ +websockets>=11.0.0 \ No newline at end of file diff --git a/simulations/sim3/server.py b/simulations/sim3/server.py new file mode 100644 index 00000000..1c140458 --- /dev/null +++ b/simulations/sim3/server.py @@ -0,0 +1,139 @@ +import asyncio +import websockets +import json +import logging +from typing import Set + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Global set to store connected clients +connected_clients: Set[websockets.WebSocketServerProtocol] = set() + +async def register_client(websocket: websockets.WebSocketServerProtocol) -> None: + """Register a new client connection.""" + connected_clients.add(websocket) + logger.info(f"Client {websocket.remote_address} connected. Total clients: {len(connected_clients)}") + + # Notify all clients about new connection + if len(connected_clients) > 1: + notification = { + "type": "system", + "message": f"User from {websocket.remote_address[0]} joined the chat", + "timestamp": asyncio.get_event_loop().time() + } + await broadcast_message(json.dumps(notification), exclude=websocket) + +async def unregister_client(websocket: websockets.WebSocketServerProtocol) -> None: + """Unregister a client connection.""" + if websocket in connected_clients: + connected_clients.remove(websocket) + logger.info(f"Client {websocket.remote_address} disconnected. Total clients: {len(connected_clients)}") + + # Notify remaining clients about disconnection + if connected_clients: + notification = { + "type": "system", + "message": f"User from {websocket.remote_address[0]} left the chat", + "timestamp": asyncio.get_event_loop().time() + } + await broadcast_message(json.dumps(notification)) + +async def broadcast_message(message: str, exclude: websockets.WebSocketServerProtocol = None) -> None: + """Broadcast a message to all connected clients except the excluded one.""" + if not connected_clients: + return + + # Create a copy of the set to avoid modification during iteration + clients_to_notify = connected_clients.copy() + if exclude: + clients_to_notify.discard(exclude) + + if clients_to_notify: + # Send message to all clients concurrently + await asyncio.gather( + *[send_safe(client, message) for client in clients_to_notify], + return_exceptions=True + ) + +async def send_safe(websocket: websockets.WebSocketServerProtocol, message: str) -> None: + """Safely send a message to a client with error handling.""" + try: + await websocket.send(message) + except websockets.exceptions.ConnectionClosed: + logger.warning(f"Failed to send message to {websocket.remote_address}: connection closed") + await unregister_client(websocket) + except Exception as e: + logger.error(f"Error sending message to {websocket.remote_address}: {e}") + await unregister_client(websocket) + +async def handle_client(websocket: websockets.WebSocketServerProtocol, path: str) -> None: + """Handle a client connection.""" + await register_client(websocket) + + try: + async for message in websocket: + try: + # Parse the incoming message + data = json.loads(message) + + # Create a broadcast message with metadata + broadcast_data = { + "type": "chat", + "message": data.get("message", ""), + "sender": data.get("sender", "Anonymous"), + "timestamp": asyncio.get_event_loop().time() + } + + logger.info(f"Broadcasting message from {broadcast_data['sender']}: {broadcast_data['message']}") + + # Broadcast to all connected clients + await broadcast_message(json.dumps(broadcast_data)) + + except json.JSONDecodeError: + logger.warning(f"Invalid JSON received from {websocket.remote_address}") + error_response = { + "type": "error", + "message": "Invalid message format", + "timestamp": asyncio.get_event_loop().time() + } + await send_safe(websocket, json.dumps(error_response)) + + except Exception as e: + logger.error(f"Error processing message from {websocket.remote_address}: {e}") + + except websockets.exceptions.ConnectionClosed: + logger.info(f"Connection closed by {websocket.remote_address}") + except Exception as e: + logger.error(f"Unexpected error with {websocket.remote_address}: {e}") + finally: + await unregister_client(websocket) + +async def main(): + """Start the WebSocket server.""" + host = "localhost" + port = 8765 + + logger.info(f"Starting WebSocket server on {host}:{port}") + + try: + async with websockets.serve(handle_client, host, port): + logger.info(f"WebSocket server started successfully on ws://{host}:{port}") + logger.info("Press Ctrl+C to stop the server") + + # Keep the server running + await asyncio.Future() # Run forever + + except OSError as e: + logger.error(f"Failed to start server: {e}") + except KeyboardInterrupt: + logger.info("Server shutdown requested") + except Exception as e: + logger.error(f"Unexpected server error: {e}") + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + logger.info("Server stopped") \ No newline at end of file diff --git a/simulations/sim4/requirements.txt b/simulations/sim4/requirements.txt index bddfc493..cf671b42 100644 --- a/simulations/sim4/requirements.txt +++ b/simulations/sim4/requirements.txt @@ -1,3 +1,3 @@ -requests>=2.31.0 -beautifulsoup4>=4.12.0 +requests>=2.28.0 +beautifulsoup4>=4.11.0 lxml>=4.9.0 \ No newline at end of file diff --git a/simulations/sim4/scraper.py b/simulations/sim4/scraper.py index 5cd7f3ad..1093b327 100644 --- a/simulations/sim4/scraper.py +++ b/simulations/sim4/scraper.py @@ -1,17 +1,18 @@ #!/usr/bin/env python3 """ -Hacker News Web Scraper +Hacker News Scraper -Scrapes the Hacker News front page for top stories and exports them as JSON. -Includes proper error handling, rate limiting, and data validation. +A web scraper that fetches stories from the Hacker News front page +and outputs them in JSON format. """ import json import logging -import time +import re +import sys from dataclasses import dataclass, asdict from typing import List, Optional -from urllib.parse import urljoin +from urllib.parse import urljoin, urlparse import requests from bs4 import BeautifulSoup @@ -23,16 +24,19 @@ logging.basicConfig( ) logger = logging.getLogger(__name__) +# Constants +HACKER_NEWS_URL = "https://news.ycombinator.com/" +REQUEST_TIMEOUT = 10 +USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" + @dataclass class Story: - """Represents a Hacker News story with its metadata.""" + """Represents a Hacker News story with metadata.""" title: str url: str points: int comments: int - story_id: str - rank: int def to_dict(self) -> dict: """Convert Story to dictionary for JSON serialization.""" @@ -40,120 +44,143 @@ class Story: class HackerNewsScraper: - """ - Scraper for Hacker News front page stories. + """Scraper for Hacker News front page stories.""" - Implements respectful scraping with error handling and retry logic. - """ - - BASE_URL = "https://news.ycombinator.com" - FRONT_PAGE_URL = "https://news.ycombinator.com/" - - def __init__(self, delay: float = 1.0, max_retries: int = 3): - """ - Initialize the scraper. - - Args: - delay: Delay between requests in seconds - max_retries: Maximum number of retry attempts - """ - self.delay = delay - self.max_retries = max_retries + def __init__(self): self.session = requests.Session() - self.session.headers.update({ - 'User-Agent': 'HackerNews-Scraper/1.0 (Educational Purpose)', - 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', - 'Accept-Language': 'en-US,en;q=0.5', - 'Accept-Encoding': 'gzip, deflate', - 'Connection': 'keep-alive', - }) - - def _make_request(self, url: str) -> Optional[requests.Response]: + self.session.headers.update({'User-Agent': USER_AGENT}) + + def fetch_page(self, url: str) -> Optional[str]: """ - Make HTTP request with retry logic and error handling. + Fetch HTML content from the given URL. Args: - url: URL to fetch + url: The URL to fetch Returns: - Response object or None if all attempts failed + HTML content as string, or None if request failed """ - for attempt in range(self.max_retries): - try: - logger.info(f"Fetching {url} (attempt {attempt + 1}/{self.max_retries})") - response = self.session.get(url, timeout=10) - response.raise_for_status() - - # Respectful delay - time.sleep(self.delay) - return response - - except requests.exceptions.RequestException as e: - logger.warning(f"Request failed (attempt {attempt + 1}): {e}") - if attempt < self.max_retries - 1: - time.sleep(2 ** attempt) # Exponential backoff - else: - logger.error(f"All {self.max_retries} attempts failed for {url}") - - return None - - def _extract_story_data(self, story_row, subtext_row, rank: int) -> Optional[Story]: + try: + logger.info(f"Fetching page: {url}") + response = self.session.get(url, timeout=REQUEST_TIMEOUT) + response.raise_for_status() + return response.text + except requests.exceptions.RequestException as e: + logger.error(f"Failed to fetch {url}: {e}") + return None + + def extract_number(self, text: str) -> int: """ - Extract story data from HTML elements. + Extract numeric value from text string. Args: - story_row: BeautifulSoup element containing story title and URL + text: Text containing a number + + Returns: + Extracted number, or 0 if no number found + """ + if not text: + return 0 + + match = re.search(r'(\d+)', text.strip()) + return int(match.group(1)) if match else 0 + + def parse_story_row(self, title_row, subtext_row) -> Optional[Story]: + """ + Parse a story from title row and subtext row elements. + + Args: + title_row: BeautifulSoup element containing title and URL subtext_row: BeautifulSoup element containing points and comments - rank: Story rank on the front page Returns: - Story object or None if extraction failed + Story object or None if parsing failed """ try: # Extract title and URL - title_link = story_row.find('span', class_='titleline').find('a') - title = title_link.get_text().strip() - url = title_link.get('href', '') + title_link = title_row.find('span', class_='titleline') + if not title_link: + return None + + link_elem = title_link.find('a') + if not link_elem: + return None + + title = link_elem.get_text(strip=True) + url = link_elem.get('href', '') # Handle relative URLs if url.startswith('item?'): - url = urljoin(self.BASE_URL, url) + url = urljoin(HACKER_NEWS_URL, url) + elif not urlparse(url).netloc: + url = urljoin(HACKER_NEWS_URL, url) - # Extract story ID from the row - story_id = story_row.get('id', '') - - # Extract points (handle cases where points might not exist for new stories) + # Extract points points = 0 - points_span = subtext_row.find('span', class_='score') - if points_span: - points_text = points_span.get_text() - points = int(''.join(filter(str.isdigit, points_text))) + score_elem = subtext_row.find('span', class_='score') + if score_elem: + points = self.extract_number(score_elem.get_text()) - # Extract comment count + # Extract comments count comments = 0 comment_links = subtext_row.find_all('a') for link in comment_links: - link_text = link.get_text() - if 'comment' in link_text: - if link_text == 'discuss': - comments = 0 - else: - comments = int(''.join(filter(str.isdigit, link_text))) + link_text = link.get_text(strip=True) + if 'comment' in link_text.lower(): + comments = self.extract_number(link_text) break return Story( title=title, url=url, points=points, - comments=comments, - story_id=story_id, - rank=rank + comments=comments ) - - except (AttributeError, ValueError, TypeError) as e: - logger.warning(f"Failed to extract story data: {e}") + + except Exception as e: + logger.warning(f"Failed to parse story row: {e}") return None - + + def parse_stories(self, html: str) -> List[Story]: + """ + Parse stories from Hacker News HTML content. + + Args: + html: HTML content string + + Returns: + List of Story objects + """ + try: + soup = BeautifulSoup(html, 'lxml') + stories = [] + + # Find the main table containing stories + main_table = soup.find('table', id='hnmain') + if not main_table: + logger.error("Could not find main stories table") + return stories + + # Find all story rows (they have class 'athing') + story_rows = main_table.find_all('tr', class_='athing') + + for story_row in story_rows: + # The subtext row immediately follows the story row + subtext_row = story_row.find_next_sibling('tr') + + if subtext_row and subtext_row.find('td', class_='subtext'): + story = self.parse_story_row(story_row, subtext_row) + if story: + stories.append(story) + logger.debug(f"Parsed story: {story.title}") + + logger.info(f"Successfully parsed {len(stories)} stories") + return stories + + except Exception as e: + logger.error(f"Failed to parse HTML: {e}") + return [] + def scrape_front_page(self) -> List[Story]: """ Scrape stories from Hacker News front page. @@ -161,106 +188,64 @@ class HackerNewsScraper: Returns: List of Story objects """ - response = self._make_request(self.FRONT_PAGE_URL) - if not response: - logger.error("Failed to fetch front page") + html = self.fetch_page(HACKER_NEWS_URL) + if not html: + logger.error("Failed to fetch Hacker News front page") return [] + + return self.parse_stories(html) + + def close(self): + """Close the requests session.""" + self.session.close() - soup = BeautifulSoup(response.content, 'lxml') - stories = [] - - # Find all story rows (they have class 'athing' and numeric id) - story_rows = soup.find_all('tr', class_='athing') - - logger.info(f"Found {len(story_rows)} story rows") - - for i, story_row in enumerate(story_rows, 1): - try: - # Find the corresponding subtext row (contains points, comments) - subtext_row = story_row.find_next_sibling('tr') - if not subtext_row or not subtext_row.find('td', class_='subtext'): - logger.warning(f"No subtext found for story {i}") - continue - - story = self._extract_story_data(story_row, subtext_row, i) - if story: - stories.append(story) - logger.debug(f"Extracted story {i}: {story.title}") - - except Exception as e: - logger.error(f"Error processing story {i}: {e}") - continue - logger.info(f"Successfully extracted {len(stories)} stories") - return stories - - def save_to_json(self, stories: List[Story], filename: str = 'stories.json') -> bool: - """ - Save stories to JSON file. +def stories_to_json(stories: List[Story]) -> str: + """ + Convert list of stories to JSON string. + + Args: + stories: List of Story objects - Args: - stories: List of Story objects - filename: Output filename - - Returns: - True if successful, False otherwise - """ - try: - data = { - 'scraped_at': time.strftime('%Y-%m-%d %H:%M:%S UTC'), - 'total_stories': len(stories), - 'stories': [story.to_dict() for story in stories] - } - - with open(filename, 'w', encoding='utf-8') as f: - json.dump(data, f, indent=2, ensure_ascii=False) - - logger.info(f"Saved {len(stories)} stories to {filename}") - return True - - except (IOError, TypeError) as e: - logger.error(f"Failed to save stories to {filename}: {e}") - return False + Returns: + JSON string representation + """ + try: + stories_dict = [story.to_dict() for story in stories] + return json.dumps(stories_dict, indent=2, ensure_ascii=False) + except Exception as e: + logger.error(f"Failed to serialize stories to JSON: {e}") + return "[]" def main(): """Main function to run the scraper.""" - logger.info("Starting Hacker News scraper...") - - # Initialize scraper with respectful settings - scraper = HackerNewsScraper(delay=1.0, max_retries=3) + scraper = HackerNewsScraper() try: # Scrape stories stories = scraper.scrape_front_page() if not stories: - logger.error("No stories were scraped") - return 1 + logger.warning("No stories were scraped") + print("[]") + sys.exit(1) + + # Output as JSON + json_output = stories_to_json(stories) + print(json_output) + + logger.info(f"Successfully scraped and output {len(stories)} stories") - # Save to JSON - if scraper.save_to_json(stories): - logger.info("Scraping completed successfully!") - - # Print summary - print(f"\nScraping Summary:") - print(f"Total stories: {len(stories)}") - print(f"Output file: stories.json") - print(f"\nTop 5 stories:") - for story in stories[:5]: - print(f"{story.rank}. {story.title} ({story.points} points, {story.comments} comments)") - - return 0 - else: - return 1 - except KeyboardInterrupt: logger.info("Scraping interrupted by user") - return 1 + sys.exit(1) except Exception as e: logger.error(f"Unexpected error: {e}") - return 1 + sys.exit(1) + finally: + scraper.close() -if __name__ == '__main__': - exit(main()) \ No newline at end of file +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/simulations/sim5/app.py b/simulations/sim5/app.py new file mode 100644 index 00000000..3688ecd8 --- /dev/null +++ b/simulations/sim5/app.py @@ -0,0 +1,361 @@ +from flask import Flask, request, jsonify +from config import Config +from database import init_database +from models import db, User +from auth import ( + generate_jwt_token, + token_required, + validate_email, + validate_username, + validate_password_strength +) +import traceback + +app = Flask(__name__) +app.config.from_object(Config) + +# Initialize database +init_database(app) + +@app.errorhandler(404) +def not_found(error): + return jsonify({ + 'error': 'Not found', + 'message': 'The requested resource was not found' + }), 404 + +@app.errorhandler(500) +def internal_error(error): + db.session.rollback() + return jsonify({ + 'error': 'Internal server error', + 'message': 'An unexpected error occurred' + }), 500 + +@app.route('/health', methods=['GET']) +def health_check(): + """Health check endpoint.""" + return jsonify({ + 'status': 'healthy', + 'message': 'Application is running' + }), 200 + +@app.route('/api/register', methods=['POST']) +def register(): + """User registration endpoint.""" + try: + data = request.get_json() + + if not data: + return jsonify({ + 'error': 'Invalid input', + 'message': 'JSON data is required' + }), 400 + + username = data.get('username', '').strip() + email = data.get('email', '').strip().lower() + password = data.get('password', '') + + # Validate required fields + if not username or not email or not password: + return jsonify({ + 'error': 'Missing required fields', + 'message': 'Username, email, and password are required' + }), 400 + + # Validate username format + if not validate_username(username): + return jsonify({ + 'error': 'Invalid username', + 'message': 'Username must be 3-80 characters and contain only letters, numbers, and underscores' + }), 400 + + # Validate email format + if not validate_email(email): + return jsonify({ + 'error': 'Invalid email', + 'message': 'Please provide a valid email address' + }), 400 + + # Validate password strength + is_valid, password_message = validate_password_strength(password) + if not is_valid: + return jsonify({ + 'error': 'Weak password', + 'message': password_message + }), 400 + + # Check if user already exists + existing_user = User.query.filter( + (User.username == username) | (User.email == email) + ).first() + + if existing_user: + if existing_user.username == username: + return jsonify({ + 'error': 'Username taken', + 'message': 'Username already exists' + }), 409 + else: + return jsonify({ + 'error': 'Email taken', + 'message': 'Email already registered' + }), 409 + + # Create new user + new_user = User( + username=username, + email=email, + password=password + ) + + db.session.add(new_user) + db.session.commit() + + # Generate JWT token + token = generate_jwt_token(new_user.id) + + return jsonify({ + 'message': 'User registered successfully', + 'user': new_user.to_dict(), + 'token': token + }), 201 + + except Exception as e: + db.session.rollback() + app.logger.error(f"Registration error: {str(e)}") + app.logger.error(traceback.format_exc()) + return jsonify({ + 'error': 'Registration failed', + 'message': 'An error occurred during registration' + }), 500 + +@app.route('/api/login', methods=['POST']) +def login(): + """User login endpoint.""" + try: + data = request.get_json() + + if not data: + return jsonify({ + 'error': 'Invalid input', + 'message': 'JSON data is required' + }), 400 + + login_identifier = data.get('username') or data.get('email', '') + password = data.get('password', '') + + if not login_identifier or not password: + return jsonify({ + 'error': 'Missing credentials', + 'message': 'Username/email and password are required' + }), 400 + + login_identifier = login_identifier.strip().lower() + + # Find user by username or email + user = User.query.filter( + (User.username == login_identifier) | (User.email == login_identifier) + ).first() + + if not user or not user.check_password(password): + return jsonify({ + 'error': 'Invalid credentials', + 'message': 'Username/email or password is incorrect' + }), 401 + + if not user.is_active: + return jsonify({ + 'error': 'Account disabled', + 'message': 'Your account has been disabled' + }), 401 + + # Generate JWT token + token = generate_jwt_token(user.id) + + return jsonify({ + 'message': 'Login successful', + 'user': user.to_dict(), + 'token': token + }), 200 + + except Exception as e: + app.logger.error(f"Login error: {str(e)}") + app.logger.error(traceback.format_exc()) + return jsonify({ + 'error': 'Login failed', + 'message': 'An error occurred during login' + }), 500 + +@app.route('/api/profile', methods=['GET']) +@token_required +def get_profile(current_user): + """Get current user profile (protected endpoint).""" + return jsonify({ + 'message': 'Profile retrieved successfully', + 'user': current_user.to_dict() + }), 200 + +@app.route('/api/profile', methods=['PUT']) +@token_required +def update_profile(current_user): + """Update current user profile (protected endpoint).""" + try: + data = request.get_json() + + if not data: + return jsonify({ + 'error': 'Invalid input', + 'message': 'JSON data is required' + }), 400 + + username = data.get('username') + email = data.get('email') + + # Update username if provided + if username is not None: + username = username.strip() + if not validate_username(username): + return jsonify({ + 'error': 'Invalid username', + 'message': 'Username must be 3-80 characters and contain only letters, numbers, and underscores' + }), 400 + + # Check if username is already taken by another user + existing_user = User.query.filter( + User.username == username, + User.id != current_user.id + ).first() + + if existing_user: + return jsonify({ + 'error': 'Username taken', + 'message': 'Username already exists' + }), 409 + + current_user.username = username + + # Update email if provided + if email is not None: + email = email.strip().lower() + if not validate_email(email): + return jsonify({ + 'error': 'Invalid email', + 'message': 'Please provide a valid email address' + }), 400 + + # Check if email is already taken by another user + existing_user = User.query.filter( + User.email == email, + User.id != current_user.id + ).first() + + if existing_user: + return jsonify({ + 'error': 'Email taken', + 'message': 'Email already registered' + }), 409 + + current_user.email = email + + db.session.commit() + + return jsonify({ + 'message': 'Profile updated successfully', + 'user': current_user.to_dict() + }), 200 + + except Exception as e: + db.session.rollback() + app.logger.error(f"Profile update error: {str(e)}") + app.logger.error(traceback.format_exc()) + return jsonify({ + 'error': 'Update failed', + 'message': 'An error occurred while updating profile' + }), 500 + +@app.route('/api/change-password', methods=['POST']) +@token_required +def change_password(current_user): + """Change user password (protected endpoint).""" + try: + data = request.get_json() + + if not data: + return jsonify({ + 'error': 'Invalid input', + 'message': 'JSON data is required' + }), 400 + + current_password = data.get('current_password', '') + new_password = data.get('new_password', '') + + if not current_password or not new_password: + return jsonify({ + 'error': 'Missing password fields', + 'message': 'Current password and new password are required' + }), 400 + + # Verify current password + if not current_user.check_password(current_password): + return jsonify({ + 'error': 'Invalid current password', + 'message': 'Current password is incorrect' + }), 401 + + # Validate new password strength + is_valid, password_message = validate_password_strength(new_password) + if not is_valid: + return jsonify({ + 'error': 'Weak password', + 'message': password_message + }), 400 + + # Check if new password is different from current + if current_user.check_password(new_password): + return jsonify({ + 'error': 'Same password', + 'message': 'New password must be different from current password' + }), 400 + + # Update password + current_user.set_password(new_password) + db.session.commit() + + return jsonify({ + 'message': 'Password changed successfully' + }), 200 + + except Exception as e: + db.session.rollback() + app.logger.error(f"Password change error: {str(e)}") + app.logger.error(traceback.format_exc()) + return jsonify({ + 'error': 'Password change failed', + 'message': 'An error occurred while changing password' + }), 500 + +@app.route('/api/users', methods=['GET']) +@token_required +def get_users(current_user): + """Get all users (protected endpoint).""" + try: + users = User.query.filter_by(is_active=True).all() + users_data = [user.to_dict() for user in users] + + return jsonify({ + 'message': 'Users retrieved successfully', + 'users': users_data, + 'count': len(users_data) + }), 200 + + except Exception as e: + app.logger.error(f"Get users error: {str(e)}") + app.logger.error(traceback.format_exc()) + return jsonify({ + 'error': 'Failed to retrieve users', + 'message': 'An error occurred while fetching users' + }), 500 + +if __name__ == '__main__': + app.run(debug=True, host='0.0.0.0', port=5000) \ No newline at end of file diff --git a/simulations/sim5/auth.py b/simulations/sim5/auth.py new file mode 100644 index 00000000..abf064d3 --- /dev/null +++ b/simulations/sim5/auth.py @@ -0,0 +1,119 @@ +import jwt +from datetime import datetime, timedelta +from functools import wraps +from flask import request, jsonify, current_app +from models import User +import re + +def generate_jwt_token(user_id): + """Generate JWT token for user.""" + payload = { + 'user_id': user_id, + 'exp': datetime.utcnow() + timedelta(hours=current_app.config['JWT_EXPIRATION_HOURS']), + 'iat': datetime.utcnow() + } + + token = jwt.encode( + payload, + current_app.config['JWT_SECRET_KEY'], + algorithm='HS256' + ) + + return token + +def verify_jwt_token(token): + """Verify JWT token and return user_id if valid.""" + try: + payload = jwt.decode( + token, + current_app.config['JWT_SECRET_KEY'], + algorithms=['HS256'] + ) + return payload['user_id'] + except jwt.ExpiredSignatureError: + return None + except jwt.InvalidTokenError: + return None + +def token_required(f): + """Decorator to require valid JWT token for protected routes.""" + @wraps(f) + def decorated(*args, **kwargs): + token = None + auth_header = request.headers.get('Authorization') + + if auth_header and auth_header.startswith('Bearer '): + token = auth_header.split(' ')[1] + + if not token: + return jsonify({ + 'error': 'Token is missing', + 'message': 'Authorization token is required' + }), 401 + + user_id = verify_jwt_token(token) + if user_id is None: + return jsonify({ + 'error': 'Token is invalid or expired', + 'message': 'Please log in again' + }), 401 + + current_user = User.query.get(user_id) + if not current_user or not current_user.is_active: + return jsonify({ + 'error': 'User not found or inactive', + 'message': 'Invalid user credentials' + }), 401 + + return f(current_user, *args, **kwargs) + + return decorated + +def validate_email(email): + """Validate email format.""" + pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' + return re.match(pattern, email) is not None + +def validate_username(username): + """Validate username format.""" + if not username or len(username) < 3 or len(username) > 80: + return False + + pattern = r'^[a-zA-Z0-9_]+$' + return re.match(pattern, username) is not None + +def validate_password_strength(password): + """Validate password strength.""" + min_length = current_app.config.get('PASSWORD_MIN_LENGTH', 8) + + if not password or len(password) < min_length: + return False, f'Password must be at least {min_length} characters long' + + if not re.search(r'[A-Z]', password): + return False, 'Password must contain at least one uppercase letter' + + if not re.search(r'[a-z]', password): + return False, 'Password must contain at least one lowercase letter' + + if not re.search(r'\d', password): + return False, 'Password must contain at least one digit' + + if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password): + return False, 'Password must contain at least one special character' + + return True, 'Password is valid' + +def get_current_user_from_token(): + """Extract current user from request token.""" + auth_header = request.headers.get('Authorization') + + if not auth_header or not auth_header.startswith('Bearer '): + return None + + token = auth_header.split(' ')[1] + user_id = verify_jwt_token(token) + + if user_id: + return User.query.get(user_id) + + return None \ No newline at end of file diff --git a/simulations/sim5/config.py b/simulations/sim5/config.py new file mode 100644 index 00000000..18503af9 --- /dev/null +++ b/simulations/sim5/config.py @@ -0,0 +1,12 @@ +import os +from dotenv import load_dotenv + +load_dotenv() + +class Config: + SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production' + SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db' + SQLALCHEMY_TRACK_MODIFICATIONS = False + JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret-key-change-in-production' + JWT_EXPIRATION_HOURS = int(os.environ.get('JWT_EXPIRATION_HOURS', 24)) + PASSWORD_MIN_LENGTH = int(os.environ.get('PASSWORD_MIN_LENGTH', 8)) \ No newline at end of file diff --git a/simulations/sim5/database.py b/simulations/sim5/database.py new file mode 100644 index 00000000..b080063c --- /dev/null +++ b/simulations/sim5/database.py @@ -0,0 +1,36 @@ +from flask import Flask +from models import db, User +import os + +def create_app_context(app): + """Create application context and initialize database.""" + with app.app_context(): + db.create_all() + return app + +def init_database(app): + """Initialize database with Flask app.""" + db.init_app(app) + + # Create tables if they don't exist + with app.app_context(): + db.create_all() + + # Create default admin user if it doesn't exist + admin_user = User.query.filter_by(username='admin').first() + if not admin_user and os.environ.get('CREATE_ADMIN_USER', 'false').lower() == 'true': + admin_email = os.environ.get('ADMIN_EMAIL', 'admin@example.com') + admin_password = os.environ.get('ADMIN_PASSWORD', 'admin123') + + admin = User( + username='admin', + email=admin_email, + password=admin_password + ) + db.session.add(admin) + db.session.commit() + print("Admin user created successfully") + +def get_db(): + """Get database instance.""" + return db \ No newline at end of file diff --git a/simulations/sim5/models.py b/simulations/sim5/models.py new file mode 100644 index 00000000..1277210f --- /dev/null +++ b/simulations/sim5/models.py @@ -0,0 +1,45 @@ +from flask_sqlalchemy import SQLAlchemy +from datetime import datetime +import bcrypt + +db = SQLAlchemy() + +class User(db.Model): + __tablename__ = 'users' + + id = db.Column(db.Integer, primary_key=True) + username = db.Column(db.String(80), unique=True, nullable=False, index=True) + email = db.Column(db.String(120), unique=True, nullable=False, index=True) + password_hash = db.Column(db.String(128), nullable=False) + created_at = db.Column(db.DateTime, default=datetime.utcnow) + is_active = db.Column(db.Boolean, default=True) + + def __init__(self, username, email, password): + self.username = username + self.email = email + self.set_password(password) + + def set_password(self, password): + """Hash and set the user's password.""" + password_bytes = password.encode('utf-8') + salt = bcrypt.gensalt() + self.password_hash = bcrypt.hashpw(password_bytes, salt).decode('utf-8') + + def check_password(self, password): + """Check if the provided password matches the stored hash.""" + password_bytes = password.encode('utf-8') + hash_bytes = self.password_hash.encode('utf-8') + return bcrypt.checkpw(password_bytes, hash_bytes) + + def to_dict(self): + """Convert user object to dictionary (excluding sensitive data).""" + return { + 'id': self.id, + 'username': self.username, + 'email': self.email, + 'created_at': self.created_at.isoformat(), + 'is_active': self.is_active + } + + def __repr__(self): + return f'' \ No newline at end of file diff --git a/simulations/sim5/requirements.txt b/simulations/sim5/requirements.txt new file mode 100644 index 00000000..043d8f61 --- /dev/null +++ b/simulations/sim5/requirements.txt @@ -0,0 +1,5 @@ +flask>=2.3.0 +flask-sqlalchemy>=3.0.0 +bcrypt>=4.0.0 +pyjwt>=2.8.0 +python-dotenv>=1.0.0 \ No newline at end of file diff --git a/simulations/sim6/requirements.txt b/simulations/sim6/requirements.txt index 0c7f68f1..39aae79e 100644 --- a/simulations/sim6/requirements.txt +++ b/simulations/sim6/requirements.txt @@ -1,2 +1 @@ -requests>=2.31.0 -urllib3>=2.0.0 \ No newline at end of file +requests>=2.28.0 \ No newline at end of file diff --git a/simulations/sim6/weather.py b/simulations/sim6/weather.py index cd949bfb..82974776 100644 --- a/simulations/sim6/weather.py +++ b/simulations/sim6/weather.py @@ -1,105 +1,45 @@ #!/usr/bin/env python3 """ -Weather CLI Application - -A command-line tool to fetch and display current weather conditions -using the OpenWeatherMap API. - -Usage: - python weather.py - python weather.py "New York" - -Environment Variables: - OPENWEATHER_API_KEY: Required API key from OpenWeatherMap +Weather CLI application using OpenWeatherMap API """ import argparse +import json import os import sys from dataclasses import dataclass from typing import Optional + import requests -from requests.adapters import HTTPAdapter -from urllib3.util.retry import Retry @dataclass class WeatherData: - """Data class to represent weather information.""" + """Data class to hold weather information""" temperature: float humidity: int wind_speed: float description: str - - def __str__(self) -> str: - """Format weather data for terminal display.""" - return f""" -╔══════════════════════════════════╗ -║ Weather Report ║ -╠══════════════════════════════════╣ -║ Temperature: {self.temperature:6.1f}°C ║ -║ Humidity: {self.humidity:6d}% ║ -║ Wind Speed: {self.wind_speed:6.1f} m/s ║ -║ Conditions: {self.description:<15} ║ -╚══════════════════════════════════╝ -""" + city: str + country: str -class WeatherAPIError(Exception): - """Custom exception for weather API related errors.""" - pass - - -class WeatherClient: - """Client for interacting with OpenWeatherMap API.""" +class WeatherAPIClient: + """Client for OpenWeatherMap API""" BASE_URL = "https://api.openweathermap.org/data/2.5/weather" - TIMEOUT = 10 # seconds + TIMEOUT = 10 def __init__(self, api_key: str): - """ - Initialize the weather client. - - Args: - api_key: OpenWeatherMap API key - """ self.api_key = api_key - self.session = self._create_session() - - def _create_session(self) -> requests.Session: - """Create a requests session with retry strategy.""" - session = requests.Session() - - # Retry strategy for handling temporary failures - retry_strategy = Retry( - total=3, - backoff_factor=1, - status_forcelist=[429, 500, 502, 503, 504], - ) - - adapter = HTTPAdapter(max_retries=retry_strategy) - session.mount("http://", adapter) - session.mount("https://", adapter) - - return session + self.session = requests.Session() def get_weather(self, city: str) -> WeatherData: - """ - Fetch weather data for the specified city. - - Args: - city: Name of the city to get weather for - - Returns: - WeatherData object containing weather information - - Raises: - WeatherAPIError: If API request fails or returns invalid data - """ + """Fetch weather data for a given city""" params = { - 'q': city.strip(), + 'q': city, 'appid': self.api_key, - 'units': 'metric' # Use Celsius + 'units': 'metric' } try: @@ -108,93 +48,98 @@ class WeatherClient: params=params, timeout=self.TIMEOUT ) + response.raise_for_status() - # Handle different HTTP status codes + except requests.exceptions.Timeout: + raise WeatherAPIError("Request timed out. Please try again.") + except requests.exceptions.ConnectionError: + raise WeatherAPIError("Unable to connect to weather service. Check your internet connection.") + except requests.exceptions.HTTPError as e: if response.status_code == 401: - raise WeatherAPIError("Invalid API key. Please check your OPENWEATHER_API_KEY.") + raise WeatherAPIError("Invalid API key. Please check your OpenWeatherMap API key.") elif response.status_code == 404: raise WeatherAPIError(f"City '{city}' not found. Please check the spelling.") elif response.status_code == 429: raise WeatherAPIError("API rate limit exceeded. Please try again later.") - elif not response.ok: + else: raise WeatherAPIError(f"API request failed with status {response.status_code}") - + except requests.exceptions.RequestException: + raise WeatherAPIError("An unexpected error occurred while fetching weather data.") + + try: data = response.json() return self._parse_weather_data(data) - - except requests.exceptions.Timeout: - raise WeatherAPIError("Request timed out. Please check your internet connection.") - except requests.exceptions.ConnectionError: - raise WeatherAPIError("Connection error. Please check your internet connection.") - except requests.exceptions.RequestException as e: - raise WeatherAPIError(f"Network error: {str(e)}") - except ValueError as e: - raise WeatherAPIError(f"Invalid response format: {str(e)}") + except json.JSONDecodeError: + raise WeatherAPIError("Invalid response format from weather service.") + except KeyError as e: + raise WeatherAPIError(f"Unexpected response format: missing {e}") def _parse_weather_data(self, data: dict) -> WeatherData: - """ - Parse API response data into WeatherData object. - - Args: - data: JSON response from OpenWeatherMap API - - Returns: - WeatherData object - - Raises: - WeatherAPIError: If response data is malformed - """ + """Parse API response into WeatherData object""" try: return WeatherData( - temperature=float(data['main']['temp']), - humidity=int(data['main']['humidity']), - wind_speed=float(data.get('wind', {}).get('speed', 0.0)), - description=data['weather'][0]['description'].title() + temperature=data['main']['temp'], + humidity=data['main']['humidity'], + wind_speed=data.get('wind', {}).get('speed', 0.0), + description=data['weather'][0]['description'].title(), + city=data['name'], + country=data['sys']['country'] ) - except (KeyError, IndexError, ValueError) as e: - raise WeatherAPIError(f"Malformed API response: {str(e)}") + except (KeyError, IndexError, TypeError) as e: + raise WeatherAPIError(f"Failed to parse weather data: {e}") -def get_api_key() -> str: - """ - Get API key from environment variables. +class WeatherAPIError(Exception): + """Custom exception for weather API errors""" + pass + + +class WeatherFormatter: + """Format weather data for terminal output""" - Returns: - API key string - - Raises: - SystemExit: If API key is not found - """ - api_key = os.getenv('OPENWEATHER_API_KEY') - if not api_key: - print("❌ Error: OPENWEATHER_API_KEY environment variable not set.", file=sys.stderr) - print("\n📝 To fix this:") - print("1. Get an API key from: https://openweathermap.org/api") - print("2. Set the environment variable:") - print(" export OPENWEATHER_API_KEY='your_api_key_here'") - sys.exit(1) - return api_key + @staticmethod + def format_weather(weather: WeatherData) -> str: + """Format weather data for display""" + return f""" +Weather for {weather.city}, {weather.country} +{'=' * (len(weather.city) + len(weather.country) + 12)} + +Temperature: {weather.temperature:.1f}°C +Humidity: {weather.humidity}% +Wind Speed: {weather.wind_speed:.1f} m/s +Conditions: {weather.description} +""" + + +def get_api_key() -> Optional[str]: + """Get API key from environment variable""" + return os.environ.get('OPENWEATHERMAP_API_KEY') def create_parser() -> argparse.ArgumentParser: - """Create and configure argument parser.""" + """Create and configure argument parser""" parser = argparse.ArgumentParser( - description="Get current weather conditions for a city", + description='Get current weather information for any city', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: - python weather.py London - python weather.py "New York" - python weather.py "São Paulo" - + %(prog)s "New York" + %(prog)s London + %(prog)s "San Francisco" --api-key YOUR_API_KEY + Environment Variables: - OPENWEATHER_API_KEY Required API key from OpenWeatherMap + OPENWEATHERMAP_API_KEY Your OpenWeatherMap API key """ ) parser.add_argument( 'city', - help='Name of the city to get weather for (use quotes for cities with spaces)' + help='City name to get weather for (use quotes for multi-word cities)' + ) + + parser.add_argument( + '--api-key', + help='OpenWeatherMap API key (overrides environment variable)' ) parser.add_argument( @@ -206,36 +151,43 @@ Environment Variables: return parser -def main(): - """Main application entry point.""" +def main() -> int: + """Main application entry point""" parser = create_parser() args = parser.parse_args() - # Validate city name - if not args.city.strip(): - print("❌ Error: City name cannot be empty.", file=sys.stderr) - sys.exit(1) + # Get API key from command line or environment + api_key = args.api_key or get_api_key() + + if not api_key: + print("Error: OpenWeatherMap API key required.", file=sys.stderr) + print("Either set OPENWEATHERMAP_API_KEY environment variable", file=sys.stderr) + print("or provide --api-key argument.", file=sys.stderr) + print("\nGet your free API key at: https://openweathermap.org/api", file=sys.stderr) + return 1 try: - # Get API key and create client - api_key = get_api_key() - client = WeatherClient(api_key) + # Initialize API client and fetch weather + client = WeatherAPIClient(api_key) + weather_data = client.get_weather(args.city) - # Fetch and display weather - print(f"🌤️ Fetching weather for {args.city}...") - weather = client.get_weather(args.city) - print(weather) + # Format and display results + formatter = WeatherFormatter() + output = formatter.format_weather(weather_data) + print(output) + + return 0 except WeatherAPIError as e: - print(f"❌ Weather Error: {e}", file=sys.stderr) - sys.exit(1) + print(f"Error: {e}", file=sys.stderr) + return 1 except KeyboardInterrupt: - print("\n⚠️ Operation cancelled by user.", file=sys.stderr) - sys.exit(1) + print("\nOperation cancelled by user.", file=sys.stderr) + return 1 except Exception as e: - print(f"❌ Unexpected error: {e}", file=sys.stderr) - sys.exit(1) + print(f"Unexpected error: {e}", file=sys.stderr) + return 1 -if __name__ == "__main__": - main() \ No newline at end of file +if __name__ == '__main__': + sys.exit(main()) \ No newline at end of file diff --git a/simulations/sim7/app.py b/simulations/sim7/app.py index 69bf047d..48e45804 100644 --- a/simulations/sim7/app.py +++ b/simulations/sim7/app.py @@ -1,155 +1,121 @@ -from flask import Flask, request, redirect, render_template, jsonify, url_for -from urllib.parse import urlparse -import re -import threading +from flask import Flask, request, jsonify, render_template, redirect +import validators +import string app = Flask(__name__) # In-memory storage -url_storage = {} -counter = 0 -counter_lock = threading.Lock() +url_mapping = {} # counter -> original_url +counter = 1 -# Base62 alphabet -BASE62_ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" +# Base62 characters for encoding +BASE62_CHARS = string.ascii_lowercase + string.ascii_uppercase + string.digits -def base62_encode(number): - """Convert a number to base62 string.""" - if number == 0: - return BASE62_ALPHABET[0] +def encode_base62(num): + """Convert integer to base62 string""" + if num == 0: + return BASE62_CHARS[0] - result = "" - while number > 0: - result = BASE62_ALPHABET[number % 62] + result - number //= 62 + result = [] + while num > 0: + result.append(BASE62_CHARS[num % 62]) + num //= 62 + + return ''.join(reversed(result)) + +def decode_base62(encoded): + """Convert base62 string back to integer""" + result = 0 + for char in encoded: + if char in BASE62_CHARS: + result = result * 62 + BASE62_CHARS.index(char) + else: + return None return result -def base62_decode(string): - """Convert a base62 string to number.""" - number = 0 - for char in string: - number = number * 62 + BASE62_ALPHABET.index(char) - return number - -def is_valid_url(url): - """Validate URL format and structure.""" - try: - result = urlparse(url) - # Check if URL has scheme and netloc - if not all([result.scheme, result.netloc]): - return False - - # Check for valid scheme - if result.scheme not in ['http', 'https']: - return False - - # Basic domain validation - domain_pattern = re.compile( - r'^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$' - ) - - return bool(domain_pattern.match(result.netloc.split(':')[0])) - - except Exception: - return False - -def generate_short_code(): - """Generate next short code using thread-safe counter.""" - global counter - with counter_lock: - counter += 1 - return base62_encode(counter) - @app.route('/') def index(): - """Serve the main landing page.""" + """Serve the landing page""" return render_template('index.html') @app.route('/shorten', methods=['POST']) def shorten_url(): - """ - Shorten a URL endpoint. - Accepts JSON or form data with 'url' field. - """ - # Handle both JSON and form data + """Shorten a URL and return the short code""" + global counter + + # Get URL from JSON request or form data if request.is_json: data = request.get_json() - url = data.get('url', '').strip() - else: - url = request.form.get('url', '').strip() - - # Validate URL presence - if not url: - if request.is_json: + if not data or 'url' not in data: + return jsonify({'error': 'URL is required'}), 400 + url = data['url'] + else: + url = request.form.get('url') + if not url: return jsonify({'error': 'URL is required'}), 400 - return render_template('index.html', error='URL is required'), 400 - # Validate URL format - if not is_valid_url(url): - error_msg = 'Invalid URL format. Please enter a valid HTTP or HTTPS URL.' - if request.is_json: - return jsonify({'error': error_msg}), 400 - return render_template('index.html', error=error_msg, url=url), 400 + # Validate URL + if not validators.url(url): + return jsonify({'error': 'Invalid URL format'}), 400 - # Generate short code and store - short_code = generate_short_code() - url_storage[short_code] = url + # Check if URL already exists + for existing_counter, existing_url in url_mapping.items(): + if existing_url == url: + short_code = encode_base62(existing_counter) + if request.is_json: + return jsonify({ + 'short_code': short_code, + 'short_url': request.host_url + short_code, + 'original_url': url + }) + else: + return render_template('index.html', + short_code=short_code, + short_url=request.host_url + short_code, + original_url=url) - # Build shortened URL - shortened_url = url_for('redirect_url', code=short_code, _external=True) + # Store new URL + url_mapping[counter] = url + short_code = encode_base62(counter) + counter += 1 if request.is_json: return jsonify({ - 'shortened_url': shortened_url, 'short_code': short_code, + 'short_url': request.host_url + short_code, 'original_url': url - }), 200 - - return render_template('index.html', - shortened_url=shortened_url, - original_url=url, - success=True) + }) + else: + return render_template('index.html', + short_code=short_code, + short_url=request.host_url + short_code, + original_url=url) -@app.route('/') -def redirect_url(code): - """ - Redirect to original URL using short code. - Returns 404 if code not found. - """ - # Validate short code format (base62) - if not code or not all(c in BASE62_ALPHABET for c in code): - return render_template('error.html', - message='Invalid short code format'), 404 +@app.route('/') +def redirect_url(short_code): + """Redirect to original URL using short code""" + # Decode the short code + counter_id = decode_base62(short_code) - # Look up original URL - original_url = url_storage.get(code) + if counter_id is None: + return jsonify({'error': 'Invalid short code format'}), 404 - if not original_url: - return render_template('error.html', - message='Short code not found'), 404 + # Look up the original URL + if counter_id not in url_mapping: + return jsonify({'error': 'Short code not found'}), 404 + original_url = url_mapping[counter_id] return redirect(original_url, code=302) @app.errorhandler(404) def not_found(error): - """Handle 404 errors.""" - return render_template('error.html', - message='Page not found'), 404 + """Handle 404 errors""" + return jsonify({'error': 'Endpoint not found'}), 404 @app.errorhandler(500) def internal_error(error): - """Handle 500 errors.""" - return render_template('error.html', - message='Internal server error'), 500 - -# Health check endpoint -@app.route('/health') -def health_check(): - """Simple health check endpoint.""" - return jsonify({ - 'status': 'healthy', - 'urls_stored': len(url_storage) - }), 200 + """Handle 500 errors""" + return jsonify({'error': 'Internal server error'}), 500 if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000) \ No newline at end of file diff --git a/simulations/sim7/requirements.txt b/simulations/sim7/requirements.txt index 2f841bdc..0754555a 100644 --- a/simulations/sim7/requirements.txt +++ b/simulations/sim7/requirements.txt @@ -1 +1,2 @@ -Flask>=2.0.0 \ No newline at end of file +flask>=2.0.0 +validators>=0.20.0 \ No newline at end of file diff --git a/simulations/sim7/templates/error.html b/simulations/sim7/templates/error.html deleted file mode 100644 index bc99aafa..00000000 --- a/simulations/sim7/templates/error.html +++ /dev/null @@ -1,58 +0,0 @@ - - - - - - Error - URL Shortener - - - -
-

❌

-

{{ message }}

- ← Go Back Home -
- - \ No newline at end of file diff --git a/simulations/sim7/templates/index.html b/simulations/sim7/templates/index.html index 5f1a23e5..8a48eff3 100644 --- a/simulations/sim7/templates/index.html +++ b/simulations/sim7/templates/index.html @@ -6,120 +6,89 @@ URL Shortener