forked from cleveragents/cleveragents-core
test: re-run all 7 simulations with improved pipeline
All sims re-run after pipeline fixes (better prompts, real syntax validation, fenced-block extraction fix). Results: 7/7 PASS, 20 files total, all Python files pass compile() syntax check. Includes run_all_sims.py runner script and rag-basic action config.
This commit is contained in:
@@ -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).
|
||||
+161
@@ -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}")
|
||||
@@ -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()
|
||||
@@ -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})"
|
||||
+236
-113
@@ -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()
|
||||
@@ -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:
|
||||
+105
-160
@@ -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)
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,3 @@
|
||||
fastapi>=0.104.0
|
||||
uvicorn[standard]>=0.24.0
|
||||
uvicorn>=0.24.0
|
||||
pydantic>=2.0.0
|
||||
@@ -0,0 +1,423 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>WebSocket Chat</title>
|
||||
<style>
|
||||
body {
|
||||
font-family: Arial, sans-serif;
|
||||
margin: 0;
|
||||
padding: 20px;
|
||||
background-color: #f5f5f5;
|
||||
}
|
||||
|
||||
.chat-container {
|
||||
max-width: 800px;
|
||||
margin: 0 auto;
|
||||
background: white;
|
||||
border-radius: 10px;
|
||||
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
|
||||
overflow: hidden;
|
||||
}
|
||||
|
||||
.chat-header {
|
||||
background: #4CAF50;
|
||||
color: white;
|
||||
padding: 20px;
|
||||
text-align: center;
|
||||
}
|
||||
|
||||
.connection-status {
|
||||
padding: 10px;
|
||||
text-align: center;
|
||||
font-weight: bold;
|
||||
}
|
||||
|
||||
.connected {
|
||||
background-color: #d4edda;
|
||||
color: #155724;
|
||||
}
|
||||
|
||||
.disconnected {
|
||||
background-color: #f8d7da;
|
||||
color: #721c24;
|
||||
}
|
||||
|
||||
.connecting {
|
||||
background-color: #fff3cd;
|
||||
color: #856404;
|
||||
}
|
||||
|
||||
.messages {
|
||||
height: 400px;
|
||||
overflow-y: auto;
|
||||
padding: 20px;
|
||||
border-bottom: 1px solid #eee;
|
||||
}
|
||||
|
||||
.message {
|
||||
margin-bottom: 15px;
|
||||
padding: 10px;
|
||||
border-radius: 5px;
|
||||
word-wrap: break-word;
|
||||
}
|
||||
|
||||
.message.chat {
|
||||
background-color: #e3f2fd;
|
||||
border-left: 4px solid #2196F3;
|
||||
}
|
||||
|
||||
.message.system {
|
||||
background-color: #f3e5f5;
|
||||
border-left: 4px solid #9c27b0;
|
||||
font-style: italic;
|
||||
}
|
||||
|
||||
.message.error {
|
||||
background-color: #ffebee;
|
||||
border-left: 4px solid #f44336;
|
||||
}
|
||||
|
||||
.message-sender {
|
||||
font-weight: bold;
|
||||
color: #333;
|
||||
margin-bottom: 5px;
|
||||
}
|
||||
|
||||
.message-content {
|
||||
color: #666;
|
||||
}
|
||||
|
||||
.message-timestamp {
|
||||
font-size: 0.8em;
|
||||
color: #999;
|
||||
margin-top: 5px;
|
||||
}
|
||||
|
||||
.input-area {
|
||||
padding: 20px;
|
||||
background-color: #fafafa;
|
||||
}
|
||||
|
||||
.input-row {
|
||||
display: flex;
|
||||
gap: 10px;
|
||||
margin-bottom: 10px;
|
||||
}
|
||||
|
||||
#usernameInput {
|
||||
flex: 1;
|
||||
padding: 10px;
|
||||
border: 1px solid #ddd;
|
||||
border-radius: 5px;
|
||||
font-size: 14px;
|
||||
}
|
||||
|
||||
#messageInput {
|
||||
flex: 3;
|
||||
padding: 10px;
|
||||
border: 1px solid #ddd;
|
||||
border-radius: 5px;
|
||||
font-size: 14px;
|
||||
}
|
||||
|
||||
#sendButton {
|
||||
padding: 10px 20px;
|
||||
background: #4CAF50;
|
||||
color: white;
|
||||
border: none;
|
||||
border-radius: 5px;
|
||||
cursor: pointer;
|
||||
font-size: 14px;
|
||||
}
|
||||
|
||||
#sendButton:hover:not(:disabled) {
|
||||
background: #45a049;
|
||||
}
|
||||
|
||||
#sendButton:disabled {
|
||||
background: #cccccc;
|
||||
cursor: not-allowed;
|
||||
}
|
||||
|
||||
.controls {
|
||||
display: flex;
|
||||
gap: 10px;
|
||||
justify-content: center;
|
||||
}
|
||||
|
||||
.btn {
|
||||
padding: 8px 16px;
|
||||
border: none;
|
||||
border-radius: 5px;
|
||||
cursor: pointer;
|
||||
font-size: 12px;
|
||||
}
|
||||
|
||||
.btn-primary {
|
||||
background: #2196F3;
|
||||
color: white;
|
||||
}
|
||||
|
||||
.btn-primary:hover {
|
||||
background: #1976D2;
|
||||
}
|
||||
|
||||
.btn-danger {
|
||||
background: #f44336;
|
||||
color: white;
|
||||
}
|
||||
|
||||
.btn-danger:hover {
|
||||
background: #d32f2f;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="chat-container">
|
||||
<div class="chat-header">
|
||||
<h1>WebSocket Chat Room</h1>
|
||||
</div>
|
||||
|
||||
<div id="connectionStatus" class="connection-status disconnected">
|
||||
Disconnected
|
||||
</div>
|
||||
|
||||
<div id="messages" class="messages">
|
||||
<div class="message system">
|
||||
<div class="message-content">Welcome to the chat room! Enter your username and click Connect to start chatting.</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="input-area">
|
||||
<div class="input-row">
|
||||
<input type="text" id="usernameInput" placeholder="Enter your username..." value="Anonymous">
|
||||
<input type="text" id="messageInput" placeholder="Type your message..." disabled>
|
||||
<button id="sendButton" disabled>Send</button>
|
||||
</div>
|
||||
<div class="controls">
|
||||
<button id="connectButton" class="btn btn-primary">Connect</button>
|
||||
<button id="disconnectButton" class="btn btn-danger" disabled>Disconnect</button>
|
||||
<button id="clearButton" class="btn">Clear Messages</button>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script>
|
||||
class ChatClient {
|
||||
constructor() {
|
||||
this.websocket = null;
|
||||
this.isConnected = false;
|
||||
this.username = 'Anonymous';
|
||||
this.serverUrl = 'ws://localhost:8765';
|
||||
|
||||
this.initializeElements();
|
||||
this.attachEventListeners();
|
||||
}
|
||||
|
||||
initializeElements() {
|
||||
this.elements = {
|
||||
connectionStatus: document.getElementById('connectionStatus'),
|
||||
messages: document.getElementById('messages'),
|
||||
usernameInput: document.getElementById('usernameInput'),
|
||||
messageInput: document.getElementById('messageInput'),
|
||||
sendButton: document.getElementById('sendButton'),
|
||||
connectButton: document.getElementById('connectButton'),
|
||||
disconnectButton: document.getElementById('disconnectButton'),
|
||||
clearButton: document.getElementById('clearButton')
|
||||
};
|
||||
}
|
||||
|
||||
attachEventListeners() {
|
||||
this.elements.connectButton.addEventListener('click', () => this.connect());
|
||||
this.elements.disconnectButton.addEventListener('click', () => this.disconnect());
|
||||
this.elements.sendButton.addEventListener('click', () => this.sendMessage());
|
||||
this.elements.clearButton.addEventListener('click', () => this.clearMessages());
|
||||
|
||||
this.elements.messageInput.addEventListener('keypress', (e) => {
|
||||
if (e.key === 'Enter' && !e.shiftKey) {
|
||||
e.preventDefault();
|
||||
this.sendMessage();
|
||||
}
|
||||
});
|
||||
|
||||
this.elements.usernameInput.addEventListener('keypress', (e) => {
|
||||
if (e.key === 'Enter') {
|
||||
this.connect();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async connect() {
|
||||
if (this.isConnected) return;
|
||||
|
||||
this.username = this.elements.usernameInput.value.trim() || 'Anonymous';
|
||||
this.updateConnectionStatus('connecting', 'Connecting...');
|
||||
|
||||
try {
|
||||
this.websocket = new WebSocket(this.serverUrl);
|
||||
|
||||
this.websocket.onopen = () => {
|
||||
this.isConnected = true;
|
||||
this.updateConnectionStatus('connected', 'Connected');
|
||||
this.updateButtonStates();
|
||||
this.addSystemMessage('Connected to chat server');
|
||||
};
|
||||
|
||||
this.websocket.onmessage = (event) => {
|
||||
this.handleMessage(event.data);
|
||||
};
|
||||
|
||||
this.websocket.onclose = (event) => {
|
||||
this.isConnected = false;
|
||||
this.updateConnectionStatus('disconnected', 'Disconnected');
|
||||
this.updateButtonStates();
|
||||
|
||||
if (event.wasClean) {
|
||||
this.addSystemMessage('Disconnected from chat server');
|
||||
} else {
|
||||
this.addSystemMessage('Connection lost unexpectedly');
|
||||
}
|
||||
|
||||
this.websocket = null;
|
||||
};
|
||||
|
||||
this.websocket.onerror = (error) => {
|
||||
console.error('WebSocket error:', error);
|
||||
this.addErrorMessage('Connection error occurred');
|
||||
};
|
||||
|
||||
} catch (error) {
|
||||
console.error('Failed to connect:', error);
|
||||
this.updateConnectionStatus('disconnected', 'Connection failed');
|
||||
this.addErrorMessage('Failed to connect to server');
|
||||
}
|
||||
}
|
||||
|
||||
disconnect() {
|
||||
if (this.websocket && this.isConnected) {
|
||||
this.websocket.close(1000, 'User disconnected');
|
||||
}
|
||||
}
|
||||
|
||||
sendMessage() {
|
||||
const message = this.elements.messageInput.value.trim();
|
||||
if (!message || !this.isConnected) return;
|
||||
|
||||
try {
|
||||
const messageData = {
|
||||
message: message,
|
||||
sender: this.username
|
||||
};
|
||||
|
||||
this.websocket.send(JSON.stringify(messageData));
|
||||
this.elements.messageInput.value = '';
|
||||
|
||||
} catch (error) {
|
||||
console.error('Failed to send message:', error);
|
||||
this.addErrorMessage('Failed to send message');
|
||||
}
|
||||
}
|
||||
|
||||
handleMessage(data) {
|
||||
try {
|
||||
const message = JSON.parse(data);
|
||||
|
||||
switch (message.type) {
|
||||
case 'chat':
|
||||
this.addChatMessage(message);
|
||||
break;
|
||||
case 'system':
|
||||
this.addSystemMessage(message.message);
|
||||
break;
|
||||
case 'error':
|
||||
this.addErrorMessage(message.message);
|
||||
break;
|
||||
default:
|
||||
console.warn('Unknown message type:', message.type);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Failed to parse message:', error);
|
||||
this.addErrorMessage('Received invalid message format');
|
||||
}
|
||||
}
|
||||
|
||||
addChatMessage(message) {
|
||||
const messageEl = this.createMessageElement('chat');
|
||||
messageEl.innerHTML = `
|
||||
<div class="message-sender">${this.escapeHtml(message.sender)}</div>
|
||||
<div class="message-content">${this.escapeHtml(message.message)}</div>
|
||||
<div class="message-timestamp">${this.formatTimestamp(message.timestamp)}</div>
|
||||
`;
|
||||
this.elements.messages.appendChild(messageEl);
|
||||
this.scrollToBottom();
|
||||
}
|
||||
|
||||
addSystemMessage(text) {
|
||||
const messageEl = this.createMessageElement('system');
|
||||
messageEl.innerHTML = `
|
||||
<div class="message-content">${this.escapeHtml(text)}</div>
|
||||
<div class="message-timestamp">${this.formatTimestamp(Date.now() / 1000)}</div>
|
||||
`;
|
||||
this.elements.messages.appendChild(messageEl);
|
||||
this.scrollToBottom();
|
||||
}
|
||||
|
||||
addErrorMessage(text) {
|
||||
const messageEl = this.createMessageElement('error');
|
||||
messageEl.innerHTML = `
|
||||
<div class="message-content">Error: ${this.escapeHtml(text)}</div>
|
||||
<div class="message-timestamp">${this.formatTimestamp(Date.now() / 1000)}</div>
|
||||
`;
|
||||
this.elements.messages.appendChild(messageEl);
|
||||
this.scrollToBottom();
|
||||
}
|
||||
|
||||
createMessageElement(type) {
|
||||
const messageEl = document.createElement('div');
|
||||
messageEl.className = `message ${type}`;
|
||||
return messageEl;
|
||||
}
|
||||
|
||||
updateConnectionStatus(status, text) {
|
||||
this.elements.connectionStatus.className = `connection-status ${status}`;
|
||||
this.elements.connectionStatus.textContent = text;
|
||||
}
|
||||
|
||||
updateButtonStates() {
|
||||
this.elements.connectButton.disabled = this.isConnected;
|
||||
this.elements.disconnectButton.disabled = !this.isConnected;
|
||||
this.elements.sendButton.disabled = !this.isConnected;
|
||||
this.elements.messageInput.disabled = !this.isConnected;
|
||||
this.elements.usernameInput.disabled = this.isConnected;
|
||||
}
|
||||
|
||||
clearMessages() {
|
||||
this.elements.messages.innerHTML = '';
|
||||
this.addSystemMessage('Messages cleared');
|
||||
}
|
||||
|
||||
scrollToBottom() {
|
||||
this.elements.messages.scrollTop = this.elements.messages.scrollHeight;
|
||||
}
|
||||
|
||||
escapeHtml(text) {
|
||||
const div = document.createElement('div');
|
||||
div.textContent = text;
|
||||
return div.innerHTML;
|
||||
}
|
||||
|
||||
formatTimestamp(timestamp) {
|
||||
const date = new Date(timestamp * 1000);
|
||||
return date.toLocaleTimeString();
|
||||
}
|
||||
}
|
||||
|
||||
// Initialize the chat client when the page loads
|
||||
document.addEventListener('DOMContentLoaded', () => {
|
||||
window.chatClient = new ChatClient();
|
||||
});
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1 @@
|
||||
websockets>=11.0.0
|
||||
@@ -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")
|
||||
@@ -1,3 +1,3 @@
|
||||
requests>=2.31.0
|
||||
beautifulsoup4>=4.12.0
|
||||
requests>=2.28.0
|
||||
beautifulsoup4>=4.11.0
|
||||
lxml>=4.9.0
|
||||
+155
-170
@@ -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())
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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))
|
||||
@@ -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
|
||||
@@ -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'<User {self.username}>'
|
||||
@@ -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
|
||||
@@ -1,2 +1 @@
|
||||
requests>=2.31.0
|
||||
urllib3>=2.0.0
|
||||
requests>=2.28.0
|
||||
+103
-151
@@ -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 <city_name>
|
||||
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()
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
+81
-115
@@ -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('/<code>')
|
||||
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('/<short_code>')
|
||||
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)
|
||||
@@ -1 +1,2 @@
|
||||
Flask>=2.0.0
|
||||
flask>=2.0.0
|
||||
validators>=0.20.0
|
||||
@@ -1,58 +0,0 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<title>Error - URL Shortener</title>
|
||||
<style>
|
||||
body {
|
||||
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', system-ui, sans-serif;
|
||||
max-width: 500px;
|
||||
margin: 100px auto;
|
||||
padding: 20px;
|
||||
text-align: center;
|
||||
color: #333;
|
||||
}
|
||||
|
||||
.error-container {
|
||||
background: #f8f9fa;
|
||||
padding: 40px;
|
||||
border-radius: 8px;
|
||||
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
|
||||
}
|
||||
|
||||
h1 {
|
||||
color: #dc3545;
|
||||
font-size: 48px;
|
||||
margin-bottom: 20px;
|
||||
}
|
||||
|
||||
p {
|
||||
font-size: 18px;
|
||||
margin-bottom: 30px;
|
||||
color: #666;
|
||||
}
|
||||
|
||||
.home-link {
|
||||
display: inline-block;
|
||||
background: #007bff;
|
||||
color: white;
|
||||
padding: 12px 24px;
|
||||
text-decoration: none;
|
||||
border-radius: 4px;
|
||||
font-weight: 500;
|
||||
}
|
||||
|
||||
.home-link:hover {
|
||||
background: #0056b3;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="error-container">
|
||||
<h1>❌</h1>
|
||||
<p>{{ message }}</p>
|
||||
<a href="/" class="home-link">← Go Back Home</a>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
@@ -6,120 +6,89 @@
|
||||
<title>URL Shortener</title>
|
||||
<style>
|
||||
body {
|
||||
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', system-ui, sans-serif;
|
||||
max-width: 600px;
|
||||
margin: 50px auto;
|
||||
font-family: Arial, sans-serif;
|
||||
max-width: 800px;
|
||||
margin: 0 auto;
|
||||
padding: 20px;
|
||||
line-height: 1.6;
|
||||
color: #333;
|
||||
background-color: #f5f5f5;
|
||||
}
|
||||
|
||||
.container {
|
||||
background: #f8f9fa;
|
||||
background-color: white;
|
||||
padding: 30px;
|
||||
border-radius: 8px;
|
||||
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
|
||||
border-radius: 10px;
|
||||
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
|
||||
}
|
||||
|
||||
h1 {
|
||||
color: #333;
|
||||
text-align: center;
|
||||
color: #2c3e50;
|
||||
margin-bottom: 30px;
|
||||
}
|
||||
|
||||
.form-group {
|
||||
margin-bottom: 20px;
|
||||
}
|
||||
|
||||
label {
|
||||
display: block;
|
||||
margin-bottom: 5px;
|
||||
font-weight: 500;
|
||||
font-weight: bold;
|
||||
color: #555;
|
||||
}
|
||||
|
||||
input[type="url"] {
|
||||
width: 100%;
|
||||
padding: 12px;
|
||||
border: 2px solid #ddd;
|
||||
border-radius: 4px;
|
||||
border-radius: 5px;
|
||||
font-size: 16px;
|
||||
box-sizing: border-box;
|
||||
}
|
||||
|
||||
input[type="url"]:focus {
|
||||
outline: none;
|
||||
border-color: #007bff;
|
||||
outline: none;
|
||||
}
|
||||
|
||||
button {
|
||||
background: #007bff;
|
||||
background-color: #007bff;
|
||||
color: white;
|
||||
padding: 12px 30px;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
cursor: pointer;
|
||||
border-radius: 5px;
|
||||
font-size: 16px;
|
||||
cursor: pointer;
|
||||
width: 100%;
|
||||
}
|
||||
|
||||
button:hover {
|
||||
background: #0056b3;
|
||||
background-color: #0056b3;
|
||||
}
|
||||
|
||||
.alert {
|
||||
padding: 15px;
|
||||
border-radius: 4px;
|
||||
margin: 20px 0;
|
||||
}
|
||||
|
||||
.alert-success {
|
||||
background: #d4edda;
|
||||
.result {
|
||||
margin-top: 30px;
|
||||
padding: 20px;
|
||||
background-color: #d4edda;
|
||||
border: 1px solid #c3e6cb;
|
||||
border-radius: 5px;
|
||||
}
|
||||
.result h2 {
|
||||
color: #155724;
|
||||
margin-top: 0;
|
||||
}
|
||||
.result p {
|
||||
margin: 10px 0;
|
||||
color: #155724;
|
||||
}
|
||||
|
||||
.alert-error {
|
||||
background: #f8d7da;
|
||||
border: 1px solid #f5c6cb;
|
||||
color: #721c24;
|
||||
}
|
||||
|
||||
.result-url {
|
||||
background: #fff;
|
||||
padding: 15px;
|
||||
border-radius: 4px;
|
||||
border: 2px solid #007bff;
|
||||
margin: 10px 0;
|
||||
.short-url {
|
||||
font-weight: bold;
|
||||
word-break: break-all;
|
||||
}
|
||||
|
||||
.result-url a {
|
||||
color: #007bff;
|
||||
text-decoration: none;
|
||||
font-weight: 500;
|
||||
}
|
||||
|
||||
.result-url a:hover {
|
||||
text-decoration: underline;
|
||||
}
|
||||
|
||||
.copy-button {
|
||||
background: #28a745;
|
||||
padding: 8px 16px;
|
||||
.copy-btn {
|
||||
background-color: #28a745;
|
||||
padding: 8px 15px;
|
||||
font-size: 14px;
|
||||
margin-top: 10px;
|
||||
width: auto;
|
||||
}
|
||||
|
||||
.copy-button:hover {
|
||||
background: #218838;
|
||||
.copy-btn:hover {
|
||||
background-color: #218838;
|
||||
}
|
||||
|
||||
.footer {
|
||||
text-align: center;
|
||||
margin-top: 30px;
|
||||
padding-top: 20px;
|
||||
border-top: 1px solid #ddd;
|
||||
color: #666;
|
||||
font-size: 14px;
|
||||
}
|
||||
@@ -129,82 +98,62 @@
|
||||
<div class="container">
|
||||
<h1>🔗 URL Shortener</h1>
|
||||
|
||||
{% if error %}
|
||||
<div class="alert alert-error">
|
||||
<strong>Error:</strong> {{ error }}
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{% if success and shortened_url %}
|
||||
<div class="alert alert-success">
|
||||
<strong>Success!</strong> Your URL has been shortened.
|
||||
|
||||
<div class="result-url">
|
||||
<strong>Shortened URL:</strong><br>
|
||||
<a href="{{ shortened_url }}" target="_blank" id="shortenedUrl">{{ shortened_url }}</a>
|
||||
<br>
|
||||
<button type="button" class="copy-button" onclick="copyToClipboard()">
|
||||
📋 Copy URL
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<div style="margin-top: 10px; font-size: 14px;">
|
||||
<strong>Original URL:</strong>
|
||||
<a href="{{ original_url }}" target="_blank">{{ original_url }}</a>
|
||||
</div>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<form method="POST" action="/shorten">
|
||||
<div class="form-group">
|
||||
<label for="url">Enter URL to shorten:</label>
|
||||
<input type="url"
|
||||
id="url"
|
||||
name="url"
|
||||
placeholder="https://example.com/your-long-url"
|
||||
value="{{ url if url else '' }}"
|
||||
required>
|
||||
<input type="url" id="url" name="url" placeholder="https://example.com" required>
|
||||
</div>
|
||||
|
||||
<button type="submit">Shorten URL</button>
|
||||
</form>
|
||||
|
||||
{% if short_code %}
|
||||
<div class="result">
|
||||
<h2>✅ Success!</h2>
|
||||
<p><strong>Original URL:</strong> {{ original_url }}</p>
|
||||
<p><strong>Short URL:</strong> <span class="short-url" id="shortUrl">{{ short_url }}</span></p>
|
||||
<p><strong>Short Code:</strong> {{ short_code }}</p>
|
||||
<button class="copy-btn" onclick="copyToClipboard()">📋 Copy Short URL</button>
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
<div class="footer">
|
||||
<p>Enter any HTTP or HTTPS URL to create a shortened version.</p>
|
||||
<p>Shortened URLs will redirect to your original link.</p>
|
||||
<p>Enter any valid URL above to generate a shortened version.</p>
|
||||
<p>The shortened URL will redirect to your original link.</p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script>
|
||||
function copyToClipboard() {
|
||||
const urlElement = document.getElementById('shortenedUrl');
|
||||
const url = urlElement.textContent;
|
||||
|
||||
navigator.clipboard.writeText(url).then(function() {
|
||||
const button = document.querySelector('.copy-button');
|
||||
const originalText = button.textContent;
|
||||
button.textContent = '✅ Copied!';
|
||||
button.style.background = '#28a745';
|
||||
|
||||
const shortUrl = document.getElementById('shortUrl').textContent;
|
||||
navigator.clipboard.writeText(shortUrl).then(function() {
|
||||
const btn = document.querySelector('.copy-btn');
|
||||
const originalText = btn.textContent;
|
||||
btn.textContent = '✅ Copied!';
|
||||
btn.style.backgroundColor = '#28a745';
|
||||
setTimeout(function() {
|
||||
button.textContent = originalText;
|
||||
button.style.background = '#28a745';
|
||||
btn.textContent = originalText;
|
||||
btn.style.backgroundColor = '#28a745';
|
||||
}, 2000);
|
||||
}).catch(function(err) {
|
||||
console.error('Failed to copy: ', err);
|
||||
// Fallback for older browsers
|
||||
const textArea = document.createElement('textarea');
|
||||
textArea.value = url;
|
||||
textArea.value = shortUrl;
|
||||
document.body.appendChild(textArea);
|
||||
textArea.focus();
|
||||
textArea.select();
|
||||
document.execCommand('copy');
|
||||
try {
|
||||
document.execCommand('copy');
|
||||
const btn = document.querySelector('.copy-btn');
|
||||
const originalText = btn.textContent;
|
||||
btn.textContent = '✅ Copied!';
|
||||
setTimeout(function() {
|
||||
btn.textContent = originalText;
|
||||
}, 2000);
|
||||
} catch (err) {
|
||||
console.error('Fallback copy failed: ', err);
|
||||
}
|
||||
document.body.removeChild(textArea);
|
||||
|
||||
const button = document.querySelector('.copy-button');
|
||||
button.textContent = '✅ Copied!';
|
||||
setTimeout(function() {
|
||||
button.textContent = '📋 Copy URL';
|
||||
}, 2000);
|
||||
});
|
||||
}
|
||||
</script>
|
||||
|
||||
Reference in New Issue
Block a user