Python Full Stack Web Developer Course

from collections import Counter

# Sample log data (mimics Apache access log format)
logs = [
    '192.168.1.10 - - [2023-05-15 10:30:45] "GET /index.html HTTP/1.1" 200 2048',
    '192.168.1.11 - - [2023-05-15 10:31:02] "GET /images/logo.png HTTP/1.1" 200 4096',
    '192.168.1.10 - - [2023-05-15 10:32:15] "GET /about.html HTTP/1.1" 200 1024',
    '192.168.1.12 - - [2023-05-15 10:35:22] "GET /products HTTP/1.1" 404 512',
    '192.168.1.10 - - [2023-05-15 10:36:48] "POST /contact.php HTTP/1.1" 500 256',
    '192.168.1.11 - - [2023-05-15 10:40:33] "GET /index.html HTTP/1.1" 304 0',
    '192.168.1.13 - - [2023-05-15 10:45:21] "PUT /api/resource HTTP/1.1" 201 128',
    '192.168.1.14 - - [2023-05-15 10:47:05] "DELETE /api/resource/1 HTTP/1.1" 204 0',
    '192.168.1.10 - - [2023-05-15 10:50:37] "GET /nonexistent.html HTTP/1.1" 404 768',
    '192.168.1.15 - - [2023-05-15 10:55:11] "GET /admin HTTP/1.1" 403 1024'
]

def analyze_logs(log_entries):
    """Analyze log entries and return statistics."""
    # Initialize counters
    ip_counter = Counter()
    method_counter = Counter()
    path_counter = Counter()
    status_counter = Counter()
    
    # Parse logs
    for log in log_entries:
        # Split log entry into parts
        parts = log.split()
        
        # Extract IP
        ip = parts[0]
        ip_counter[ip] += 1
        
        # Extract method and path
        request = parts[6]
        method = request.strip('"')
        method_counter[method] += 1
        
        # Extract path
        path = parts[7]
        path_counter[path] += 1
        
        # Extract status code
        status = parts[9]
        status_counter[status] += 1
    
    return {
        'ip_addresses': ip_counter,
        'methods': method_counter,
        'paths': path_counter,
        'status_codes': status_counter,
        'total_requests': len(log_entries)
    }

# Analyze the logs
log_stats = analyze_logs(logs)

# Print the results
print("Log Analysis Results:")
print(f"Total Requests: {log_stats['total_requests']}")

print("\nTop IP Addresses:")
for ip, count in log_stats['ip_addresses'].most_common(3):
    print(f"  {ip}: {count} requests")

print("\nHTTP Methods:")
for method, count in log_stats['methods'].most_common():
    print(f"  {method}: {count} requests")

print("\nMost Requested Paths:")
for path, count in log_stats['paths'].most_common(3):
    print(f"  {path}: {count} requests")

print("\nStatus Code Distribution:")
for status, count in log_stats['status_codes'].most_common():
    print(f"  {status}: {count} requests")

from collections import defaultdict, OrderedDict, namedtuple
import datetime

# Define data structures
Customer = namedtuple('Customer', [
    'id', 'name', 'email', 'phone', 'address', 'registered_on'
])

Product = namedtuple('Product', [
    'id', 'name', 'description', 'price', 'category', 'in_stock'
])

Order = namedtuple('Order', [
    'id', 'customer_id', 'order_date', 'status', 'items', 'total'
])

class OrderManagementSystem:
    """Customer order management system."""
    
    def __init__(self):
        # Store customers and products
        self.customers = {}  # customer_id -> Customer
        self.products = {}   # product_id -> Product
        
        # Store orders
        self.orders = {}  # order_id -> Order
        
        # Track order history by customer
        self.customer_orders = defaultdict(list)  # customer_id -> list of order_ids
        
        # Track recent orders (ordered by time)
        self.recent_orders = OrderedDict()  # order_id -> timestamp
        
        # Track product sales
        self.product_sales = defaultdict(int)  # product_id -> quantity sold
        
        # Track current cart for each customer
        self.shopping_carts = defaultdict(lambda: defaultdict(int))  # customer_id -> {product_id -> quantity}
    
    def add_customer(self, id, name, email, phone=None, address=None):
        """Add a new customer."""
        customer = Customer(
            id=id,
            name=name,
            email=email,
            phone=phone,
            address=address,
            registered_on=datetime.datetime.now()
        )
        self.customers[id] = customer
        return customer
    
    def add_product(self, id, name, description, price, category, in_stock=True):
        """Add a new product."""
        product = Product(
            id=id,
            name=name,
            description=description,
            price=price,
            category=category,
            in_stock=in_stock
        )
        self.products[id] = product
        return product
    
    def add_to_cart(self, customer_id, product_id, quantity=1):
        """Add a product to a customer's cart."""
        if customer_id not in self.customers:
            raise ValueError(f"Customer ID {customer_id} not found")
        
        if product_id not in self.products:
            raise ValueError(f"Product ID {product_id} not found")
        
        product = self.products[product_id]
        if not product.in_stock:
            raise ValueError(f"Product {product.name} is out of stock")
        
        # Add to cart
        self.shopping_carts[customer_id][product_id] += quantity
    
    def remove_from_cart(self, customer_id, product_id, quantity=None):
        """Remove a product from a customer's cart."""
        if customer_id not in self.shopping_carts:
            return False
        
        if product_id not in self.shopping_carts[customer_id]:
            return False
        
        if quantity is None or quantity >= self.shopping_carts[customer_id][product_id]:
            # Remove completely
            del self.shopping_carts[customer_id][product_id]
        else:
            # Reduce quantity
            self.shopping_carts[customer_id][product_id] -= quantity
        
        return True
    
    def get_cart(self, customer_id):
        """Get the current cart for a customer."""
        cart = self.shopping_carts[customer_id]
        
        # Convert to list of (product, quantity) tuples
        cart_items = []
        cart_total = 0
        
        for product_id, quantity in cart.items():
            product = self.products[product_id]
            item_total = product.price * quantity
            cart_items.append((product, quantity, item_total))
            cart_total += item_total
        
        return {
            'items': cart_items,
            'total': cart_total
        }
    
    def checkout(self, customer_id, order_id=None):
        """Create an order from the customer's cart."""
        if customer_id not in self.customers:
            raise ValueError(f"Customer ID {customer_id} not found")
        
        if customer_id not in self.shopping_carts or not self.shopping_carts[customer_id]:
            raise ValueError("Shopping cart is empty")
        
        # Generate order ID if not provided
        if order_id is None:
            order_id = f"ORD-{len(self.orders) + 1:04d}"
        
        # Create order items
        items = {}
        total = 0
        
        for product_id, quantity in self.shopping_carts[customer_id].items():
            product = self.products[product_id]
            item_total = product.price * quantity
            items[product_id] = (quantity, item_total)
            total += item_total
            
            # Update product sales
            self.product_sales[product_id] += quantity
        
        # Create order
        order = Order(
            id=order_id,
            customer_id=customer_id,
            order_date=datetime.datetime.now(),
            status='PLACED',
            items=items,
            total=total
        )
        
        # Store order
        self.orders[order_id] = order
        self.customer_orders[customer_id].append(order_id)
        # Add to recent orders
                self.recent_orders[order_id] = order.order_date
                
                # Limit recent orders to 100
                if len(self.recent_orders) > 100:
                    self.recent_orders.popitem(last=False)  # Remove oldest
                
                # Clear the cart
                del self.shopping_carts[customer_id]
                
                return order
            
            def update_order_status(self, order_id, new_status):
                """Update the status of an order."""
                if order_id not in self.orders:
                    raise ValueError(f"Order ID {order_id} not found")
                
                old_order = self.orders[order_id]
                
                # Create updated order
                new_order = old_order._replace(status=new_status)
                self.orders[order_id] = new_order
                
                return new_order
            
            def get_customer_orders(self, customer_id):
                """Get all orders for a customer."""
                if customer_id not in self.customers:
                    raise ValueError(f"Customer ID {customer_id} not found")
                
                orders = []
                for order_id in self.customer_orders[customer_id]:
                    orders.append(self.orders[order_id])
                
                return orders
            
            def get_recent_orders(self, limit=10):
                """Get the most recent orders."""
                recent = list(self.recent_orders.keys())[-limit:]
                return [self.orders[order_id] for order_id in reversed(recent)]
            
            def get_popular_products(self, limit=5):
                """Get the most popular products by sales."""
                popular = sorted(self.product_sales.items(), key=lambda x: x[1], reverse=True)[:limit]
                return [(self.products[product_id], sales) for product_id, sales in popular]
            
            def get_sales_by_category(self):
                """Get total sales by product category."""
                category_sales = defaultdict(int)
                
                for product_id, quantity in self.product_sales.items():
                    product = self.products[product_id]
                    category_sales[product.category] += quantity
                
                return category_sales
        
        # Test the order management system
        def test_order_system():
            """Test the order management system with sample data."""
            system = OrderManagementSystem()
            
            print("Order Management System Test\n")
            
            # Add customers
            print("Adding customers...")
            system.add_customer(1, "John Smith", "john@example.com", "555-1234", "123 Main St")
            system.add_customer(2, "Jane Doe", "jane@example.com", "555-5678", "456 Oak Ave")
            system.add_customer(3, "Bob Johnson", "bob@example.com", "555-9012", "789 Pine Rd")
            
            # Add products
            print("Adding products...")
            system.add_product(101, "Laptop", "Powerful laptop for developers", 1299.99, "Electronics")
            system.add_product(102, "Smartphone", "Latest smartphone model", 799.99, "Electronics")
            system.add_product(103, "Headphones", "Noise-cancelling headphones", 249.99, "Electronics")
            system.add_product(104, "Coffee Maker", "Automatic coffee maker", 89.99, "Kitchen")
            system.add_product(105, "Desk Lamp", "LED desk lamp", 49.99, "Home Office")
            
            # Add items to carts
            print("\nAdding items to carts...")
            system.add_to_cart(1, 101, 1)  # John buys a laptop
            system.add_to_cart(1, 103, 1)  # And headphones
            
            system.add_to_cart(2, 102, 1)  # Jane buys a smartphone
            system.add_to_cart(2, 105, 2)  # And two desk lamps
            
            system.add_to_cart(3, 104, 1)  # Bob buys a coffee maker
            
            # Check carts
            for customer_id in [1, 2, 3]:
                cart = system.get_cart(customer_id)
                customer = system.customers[customer_id]
                print(f"\n{customer.name}'s Cart:")
                for product, quantity, total in cart['items']:
                    print(f"  {quantity}x {product.name} (${product.price:.2f}) = ${total:.2f}")
                print(f"  Total: ${cart['total']:.2f}")
            
            # Process checkouts
            print("\nProcessing checkouts...")
            order1 = system.checkout(1)
            order2 = system.checkout(2)
            order3 = system.checkout(3)
            
            print(f"Order {order1.id}: ${order1.total:.2f}")
            print(f"Order {order2.id}: ${order2.total:.2f}")
            print(f"Order {order3.id}: ${order3.total:.2f}")
            
            # Update order statuses
            system.update_order_status(order1.id, "SHIPPED")
            system.update_order_status(order2.id, "PROCESSING")
            
            # Get customer order history
            customer = system.customers[1]
            orders = system.get_customer_orders(1)
            print(f"\n{customer.name}'s Order History:")
            for order in orders:
                print(f"  Order {order.id} ({order.status}): ${order.total:.2f}")
                for product_id, (quantity, item_total) in order.items.items():
                    product = system.products[product_id]
                    print(f"    {quantity}x {product.name} = ${item_total:.2f}")
            
            # Get recent orders
            print("\nRecent Orders:")
            recent_orders = system.get_recent_orders()
            for order in recent_orders:
                customer = system.customers[order.customer_id]
                print(f"  {order.order_date.strftime('%H:%M:%S')}: {customer.name} - ${order.total:.2f} ({order.status})")
            
            # Get popular products
            print("\nPopular Products:")
            popular = system.get_popular_products()
            for product, sales in popular:
                print(f"  {product.name}: {sales} sold")
            
            # Get sales by category
            print("\nSales by Category:")
            category_sales = system.get_sales_by_category()
            for category, sales in category_sales.items():
                print(f"  {category}: {sales} items sold")
        
        # Run the test
        test_order_system()

from collections import deque, namedtuple, Counter
        import re
        import time
        import random
        
        # Define document structure
        Document = namedtuple('Document', [
            'id', 'title', 'content', 'author', 'date', 'metadata'
        ])
        
        # Define processing result
        ProcessingResult = namedtuple('ProcessingResult', [
            'document_id', 'stage', 'status', 'result', 'timestamp'
        ])
        
        class DocumentProcessor:
            """A pipeline for processing text documents."""
            
            def __init__(self, max_queue_size=100):
                # Processing queues
                self.input_queue = deque(maxlen=max_queue_size)
                self.processing_queue = deque()
                self.completed_queue = deque(maxlen=max_queue_size)
                
                # Processing stages
                self.stages = [
                    self.clean_text,
                    self.count_words,
                    self.extract_entities,
                    self.sentiment_analysis,
                    self.topic_modeling
                ]
                
                # Results by document
                self.results = {}
            
            def add_document(self, document):
                """Add a document to the input queue."""
                self.input_queue.append(document)
                # Initialize results entry
                self.results[document.id] = {
                    'document': document,
                    'stages': {},
                    'completed': False
                }
            
            def process_next_batch(self, batch_size=5):
                """Process the next batch of documents."""
                # Move documents from input to processing queue
                processed_count = 0
                
                # Process up to batch_size documents
                for _ in range(min(batch_size, len(self.input_queue))):
                    document = self.input_queue.popleft()
                    self.processing_queue.append((document, 0))  # (document, stage_index)
                    processed_count += 1
                
                # Process documents in the processing queue
                documents_to_remove = []
                
                for item in self.processing_queue:
                    document, stage_index = item
                    
                    if stage_index >= len(self.stages):
                        # Document has completed all stages
                        documents_to_remove.append(item)
                        self.completed_queue.append(document)
                        self.results[document.id]['completed'] = True
                        continue
                    
                    # Process the current stage
                    stage_func = self.stages[stage_index]
                    stage_name = stage_func.__name__
                    
                    try:
                        # Process the document
                        result = stage_func(document)
                        status = 'completed'
                    except Exception as e:
                        # Handle processing error
                        result = {'error': str(e)}
                        status = 'failed'
                    
                    # Store the result
                    processing_result = ProcessingResult(
                        document_id=document.id,
                        stage=stage_name,
                        status=status,
                        result=result,
                        timestamp=time.time()
                    )
                    
                    self.results[document.id]['stages'][stage_name] = processing_result
                    
                    # Move to the next stage if successful
                    if status == 'completed':
                        item_index = self.processing_queue.index(item)
                        self.processing_queue[item_index] = (document, stage_index + 1)
                
                # Remove completed documents from processing queue
                for item in documents_to_remove:
                    self.processing_queue.remove(item)
                
                return processed_count, len(documents_to_remove)
            
            def get_results(self, document_id):
                """Get processing results for a document."""
                return self.results.get(document_id)
            
            def get_status(self):
                """Get the current status of the processor."""
                return {
                    'input_queue': len(self.input_queue),
                    'processing_queue': len(self.processing_queue),
                    'completed_queue': len(self.completed_queue),
                    'total_documents': len(self.results),
                    'completed_documents': sum(1 for r in self.results.values() if r['completed'])
                }
            
            # Processing stages
            def clean_text(self, document):
                """Clean the document text."""
                # Simulate processing time
                time.sleep(0.01)
                
                content = document.content
                
                # Convert to lowercase
                content = content.lower()
                
                # Remove special characters
                content = re.sub(r'[^a-zA-Z0-9\s]', '', content)
                
                # Remove extra whitespace
                content = re.sub(r'\s+', ' ', content).strip()
                
                return {
                    'cleaned_text': content,
                    'original_length': len(document.content),
                    'cleaned_length': len(content)
                }
            
            def count_words(self, document):
                """Count words in the document."""
                # Simulate processing time
                time.sleep(0.01)
                
                # Get cleaned text from previous stage
                cleaned_text = self.results[document.id]['stages']['clean_text'].result['cleaned_text']
                
                # Split into words
                words = cleaned_text.split()
                
                # Count word frequencies
                word_counter = Counter(words)
                
                return {
                    'word_count': len(words),
                    'unique_words': len(word_counter),
                    'word_frequencies': word_counter,
                    'most_common': word_counter.most_common(10)
                }
            
            def extract_entities(self, document):
                """Extract entities from the document."""
                # Simulate processing time
                time.sleep(0.02)
                
                # In a real system, this would use NLP to extract entities
                # Here we'll just simulate it with random extraction
                
                # Get cleaned text from previous stage
                cleaned_text = self.results[document.id]['stages']['clean_text'].result['cleaned_text']
                words = cleaned_text.split()
                
                # Randomly select some "entities"
                entity_count = min(5, len(words))
                entities = random.sample(words, entity_count)
                
                # Categorize entities (simulated)
                entity_types = ['PERSON', 'ORGANIZATION', 'LOCATION', 'DATE', 'MISC']
                categorized = {}
                
                for entity in entities:
                    category = random.choice(entity_types)
                    if category not in categorized:
                        categorized[category] = []
                    categorized[category].append(entity)
                
                return {
                    'entities': entities,
                    'categorized': categorized,
                    'entity_count': entity_count
                }
            
            def sentiment_analysis(self, document):
                """Analyze sentiment of the document."""
                # Simulate processing time
                time.sleep(0.03)
                
                # In a real system, this would use an NLP model
                # Here we'll just simulate sentiment analysis
                
                # Randomly assign sentiment (simulated)
                sentiment_score = random.uniform(-1.0, 1.0)
                
                # Map score to sentiment category
                if sentiment_score < -0.3:
                    sentiment = 'negative'
                elif sentiment_score > 0.3:
                    sentiment = 'positive'
                else:
                    sentiment = 'neutral'
                
                return {
                    'sentiment_score': sentiment_score,
                    'sentiment': sentiment,
                    'confidence': random.uniform(0.7, 0.95)
                }
            
            def topic_modeling(self, document):
                """Identify topics in the document."""
                # Simulate processing time
                time.sleep(0.02)
                
                # In a real system, this would use techniques like LDA
                # Here we'll just simulate topic modeling
                
                # Possible topics
                all_topics = [
                    'Technology', 'Science', 'Politics', 'Business', 
                    'Entertainment', 'Sports', 'Health', 'Education'
                ]
                
                # Randomly assign topics and scores
                topic_count = random.randint(1, 3)
                topics = random.sample(all_topics, topic_count)
                
                topic_scores = {}
                total = 0
                
                for topic in topics:
                    score = random.uniform(0.5, 1.0)
                    topic_scores[topic] = score
                    total += score
                
                # Normalize scores
                normalized_scores = {t: s/total for t, s in topic_scores.items()}
                
                return {
                    'topics': topics,
                    'topic_scores': normalized_scores,
                    'primary_topic': max(normalized_scores.items(), key=lambda x: x[1])[0]
                }
        
        # Test the document processor
        def test_document_processor():
            """Test the document processor with sample documents."""
            processor = DocumentProcessor()
            
            print("Document Processor Test\n")
            
            # Create sample documents
            documents = [
                Document(
                    id="doc-001",
                    title="The Impact of Artificial Intelligence",
                    content="Artificial intelligence (AI) is transforming industries around the world. From healthcare to finance, AI technologies are improving efficiency and enabling new capabilities. However, the rapid adoption of AI also raises important ethical considerations.",
                    author="John Smith",
                    date="2023-01-15",
                    metadata={"category": "Technology", "tags": ["AI", "machine learning", "ethics"]}
                ),
                Document(
                    id="doc-002",
                    title="Sustainable Energy Solutions",
                    content="Renewable energy sources like solar and wind power are essential for combating climate change. These technologies have seen significant cost reductions in recent years, making them competitive with fossil fuels in many markets.",
                    author="Jane Doe",
                    date="2023-02-20",
                    metadata={"category": "Environment", "tags": ["renewable energy", "climate change"]}
                ),
                Document(
                    id="doc-003",
                    title="Modern Web Development Practices",
                    content="Modern web development encompasses a wide range of practices and technologies. Front-end frameworks like React and Vue.js have revolutionized user interface development, while containerization tools like Docker have simplified deployment processes.",
                    author="Bob Johnson",
                    date="2023-03-10",
                    metadata={"category": "Technology", "tags": ["web development", "frameworks", "containers"]}
                )
            ]
            
            # Add documents to the processor
            print("Adding documents to the processor...")
            for doc in documents:
                processor.add_document(doc)
                print(f"  Added: {doc.title}")
            
            # Process documents
            print("\nProcessing documents...")
            total_processed = 0
            total_completed = 0
            
            while processor.input_queue or processor.processing_queue:
                processed, completed = processor.process_next_batch()
                total_processed += processed
                total_completed += completed
                
                status = processor.get_status()
                print(f"  Batch processed. Queue status: input={status['input_queue']}, processing={status['processing_queue']}, completed={status['completed_queue']}")
            
            print(f"\nProcessing complete. {total_completed} documents fully processed.")
            
            # Print results for each document
            for doc in documents:
                results = processor.get_results(doc.id)
                
                print(f"\nResults for '{doc.title}':")
                if not results['completed']:
                    print("  Processing incomplete")
                    continue
                
                # Word count results
                word_count = results['stages']['count_words'].result
                print(f"  Word count: {word_count['word_count']} words, {word_count['unique_words']} unique")
                print(f"  Most common words: {word_count['most_common'][:3]}")
                
                # Entity extraction results
                entities = results['stages']['extract_entities'].result
                print(f"  Extracted entities: {entities['entity_count']}")
                for entity_type, entity_list in entities['categorized'].items():
                    print(f"    {entity_type}: {', '.join(entity_list)}")
                
                # Sentiment analysis results
                sentiment = results['stages']['sentiment_analysis'].result
                print(f"  Sentiment: {sentiment['sentiment']} (score: {sentiment['sentiment_score']:.2f}, confidence: {sentiment['confidence']:.2f})")
                
                # Topic modeling results
                topics = results['stages']['topic_modeling'].result
                print(f"  Primary topic: {topics['primary_topic']}")
                print(f"  All topics: {', '.join(topics['topics'])}")
        
        # Run the test
        test_document_processor()