diff --git a/README.md b/README.md index f620e0ce..c298a644 100644 --- a/README.md +++ b/README.md @@ -84,6 +84,7 @@ This is a repository of all the tutorials of [The Python Code](https://www.thepy - [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) @@ -146,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)) @@ -190,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)) @@ -262,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/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]")