analitics

Pages

Showing posts with label threading. Show all posts
Showing posts with label threading. Show all posts

Tuesday, October 6, 2026

thinter : manage python scripts !!!

Today, this simple manager for python scripts with autodetect python modules and add to description for each script with Claude AI.
See the source code:
# -*- coding: utf-8 -*-
import os
import re
import ast
import sys
import json
import hashlib
import subprocess
import tkinter as tk
from tkinter import ttk, messagebox
import tkinter.font as tkfont
DEFAULT_ROOT = "/storage/emulated/0/Download"
CATALOG_NAME = ".scripts_catalog.json"
FONT_SIZE = 10
try:
    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
except NameError:
    BASE_DIR = os.getcwd()
STD = getattr(sys, "stdlib_module_names", None)
if STD is None:
    STD = set((
        "__future__ abc argparse array ast asyncio base64 bisect bz2 calendar cmd code codecs collections "
        "configparser contextlib copy csv ctypes curses dataclasses datetime decimal difflib email enum "
        "fnmatch fractions ftplib functools gc getopt getpass gettext glob gzip hashlib heapq hmac html http "
        "imaplib importlib inspect io itertools json locale logging lzma math mimetypes msvcrt multiprocessing "
        "operator optparse os pathlib pdb pickle platform pprint profile queue random re readline sched secrets "
        "select shelve shutil signal smtplib socket sqlite3 ssl stat statistics string struct subprocess sys "
        "tarfile telnetlib tempfile textwrap threading time timeit tkinter traceback turtle types typing "
        "unittest urllib uuid warnings weakref webbrowser winreg xml zipfile zlib"
    ).split())
ALIASES = {
    "PIL": "Pillow", "cv2": "opencv-python", "yt_dlp": "yt-dlp", "bs4": "beautifulsoup4",
    "sklearn": "scikit-learn", "yaml": "PyYAML", "dotenv": "python-dotenv", "serial": "pyserial",
    "win32api": "pywin32", "win32com": "pywin32", "Crypto": "pycryptodome", "skimage": "scikit-image",
    "dateutil": "python-dateutil", "OpenGL": "PyOpenGL", "bpy": "bpy (Blender)",
}
def safe(text):
    return str(text).encode("utf-8", "replace").decode("utf-8")
def read_json(path):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None
def is_local(mod, full, root_folder):
    root_n = os.path.normpath(root_folder)
    cur = os.path.normpath(os.path.dirname(full))
    while True:
        is_root = (cur == root_n)
        if os.path.isfile(os.path.join(cur, mod + ".py")):
            return True
        if not is_root and os.path.isdir(os.path.join(cur, mod)):
            return True
        if is_root or len(cur) <= len(root_n):
            return False
        cur = os.path.dirname(cur)
def analyze(full, root_folder):
    try:
        with open(full, "r", encoding="utf-8", errors="replace") as f:
            text = f.read(300000)
    except OSError:
        return "Libs: nu pot citi fișierul"
    mods = []
    try:
        for node in ast.walk(ast.parse(text)):
            if isinstance(node, ast.Import):
                for a in node.names:
                    mods.append(a.name.split(".")[0])
            elif isinstance(node, ast.ImportFrom):
                if node.module and node.level == 0:
                    mods.append(node.module.split(".")[0])
    except (SyntaxError, ValueError):
        pattern = r"^\s*(?:from\s+([A-Za-z_]\w*)[\w.]*\s+import\s|import\s+([A-Za-z_]\w*))"
        for m in re.finditer(pattern, text, re.M):
            mods.append(m.group(1) or m.group(2))
    libs = []
    for m in mods:
        if m in STD or is_local(m, full, root_folder):
            continue
        name = ALIASES.get(m, m)
        if name not in libs:
            libs.append(name)
    libs.sort(key=str.lower)
    return "Libs: " + (", ".join(libs) if libs else "doar biblioteca standard")
class App:
    def __init__(self, root):
        self.root = root
        root.title("Python Scripts Manager")
        root.geometry("900x600")
        self.scripts = []
        self.files = {}
        self.root_folder = ""
        self.font = tkfont.Font(family="Arial", size=FONT_SIZE)
        style = ttk.Style()
        style.configure("Treeview", font=self.font, rowheight=self.font.metrics("linespace") + 14)
        style.configure("Treeview.Heading", font=(self.font.actual("family"), FONT_SIZE, "bold"))
        top = tk.Frame(root)
        top.pack(fill="x", padx=6, pady=4)
        tk.Label(top, text="Folder:").pack(side="left")
        self.path_var = tk.StringVar(value=DEFAULT_ROOT)
        tk.Entry(top, textvariable=self.path_var, font=self.font).pack(side="left", fill="x", expand=True, padx=4)
        tk.Button(top, text="Scan", command=self.load_scripts).pack(side="left")
        search = tk.Frame(root)
        search.pack(fill="x", padx=6)
        tk.Label(search, text="Caută:").pack(side="left")
        self.search_var = tk.StringVar()
        self.search_var.trace_add("write", lambda *a: self.refresh_table())
        tk.Entry(search, textvariable=self.search_var, font=self.font).pack(side="left", fill="x", expand=True, padx=4)
        frame = tk.Frame(root)
        frame.pack(fill="both", expand=True, padx=6, pady=4)
        self.tree = ttk.Treeview(frame, columns=("name", "rel", "desc"), show="headings", selectmode="browse")
        self.tree.heading("name", text="Script")
        self.tree.heading("rel", text="Path relativ")
        self.tree.heading("desc", text="Descriere")
        self.tree.column("name", width=180)
        self.tree.column("rel", width=300)
        self.tree.column("desc", width=300)
        self.tree.tag_configure("auto", foreground="gray")
        vsb = ttk.Scrollbar(frame, orient="vertical", command=self.tree.yview)
        hsb = ttk.Scrollbar(frame, orient="horizontal", command=self.tree.xview)
        self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
        self.tree.grid(row=0, column=0, sticky="nsew")
        vsb.grid(row=0, column=1, sticky="ns")
        hsb.grid(row=1, column=0, sticky="ew")
        frame.rowconfigure(0, weight=1)
        frame.columnconfigure(0, weight=1)
        self.tree.bind("", lambda e: self.open_detail())
        btns = tk.Frame(root)
        btns.pack(fill="x", padx=6, pady=4)
        tk.Button(btns, text="Run", command=self.run_script, bg="#2196F3", fg="white", width=8).pack(side="left", padx=2)
        tk.Button(btns, text="Copy path", command=self.copy_path, width=8).pack(side="left", padx=2)
        tk.Button(btns, text="Detalii", command=self.open_detail, width=8).pack(side="left", padx=2)
        tk.Button(btns, text="Auto toate", command=self.auto_all, width=8).pack(side="left", padx=2)
        self.status = tk.Label(root, text="Ready", anchor="w")
        self.status.pack(fill="x", padx=6, pady=2)
        self.load_scripts()
    def primary_file(self):
        return os.path.join(self.root_folder, CATALOG_NAME)
    def fallback_file(self):
        h = hashlib.md5(self.root_folder.encode("utf-8", "replace")).hexdigest()[:8]
        return os.path.join(BASE_DIR, "catalog_" + h + ".json")
    def save_catalog(self):
        payload = {"version": 1, "root": self.root_folder, "files": self.files}
        for path in (self.primary_file(), self.fallback_file()):
            try:
                with open(path, "w", encoding="utf-8") as f:
                    json.dump(payload, f, indent=2, ensure_ascii=False)
                return path
            except OSError:
                continue
        messagebox.showerror("Eroare", "Nu pot salva catalogul.")
        return None
    def load_scripts(self):
        folder = self.path_var.get().strip()
        if not os.path.isdir(folder):
            messagebox.showerror("Eroare", "Folder inexistent:\n" + safe(folder))
            return
        self.root_folder = folder
        data = read_json(self.primary_file()) or read_json(self.fallback_file()) or {}
        old = data.get("files", {})
        found = []
        for dirpath, dirs, files in os.walk(folder):
            dirs[:] = [d for d in dirs if d != "__pycache__"]
            for file in files:
                if file.endswith(".py"):
                    full = os.path.join(dirpath, file)
                    rel = os.path.relpath(full, folder).replace("\\", "/")
                    found.append((file, rel, full))
        found.sort(key=lambda s: s[1].lower())
        found_rels = set(s[1] for s in found)
        orphans = {r: e for r, e in old.items() if r not in found_rels and e.get("desc")}
        new_files = {}
        for name, rel, full in found:
            try:
                st = os.stat(full)
                size, mtime = st.st_size, int(st.st_mtime)
            except OSError:
                size, mtime = 0, 0
            entry = old.get(rel)
            if entry is None:
                entry = {"desc": ""}
                for orel, oe in list(orphans.items()):
                    if os.path.basename(orel) == name and oe.get("size") == size:
                        entry = {"desc": oe.get("desc", ""), "auto": oe.get("auto", "")}
                        del orphans[orel]
                        break
            entry = dict(entry)
            entry["size"] = size
            entry["mtime"] = mtime
            new_files[rel] = entry
        new_files.update(orphans)
        self.files = new_files
        self.scripts = found
        saved = self.save_catalog()
        self.refresh_table()
        if saved:
            self.status.config(text=str(len(found)) + " scripturi | catalog: " + safe(saved))
    def refresh_table(self):
        q = self.search_var.get().strip().lower()
        self.tree.delete(*self.tree.get_children())
        shown = 0
        for i, (name, rel, full) in enumerate(self.scripts):
            entry = self.files.get(rel, {})
            desc = entry.get("desc", "")
            auto = entry.get("auto", "")
            text = desc if desc else ("~ " + auto if auto else "")
            if q and q not in name.lower() and q not in rel.lower() and q not in text.lower():
                continue
            tags = ("auto",) if (not desc and auto) else ()
            self.tree.insert("", "end", iid=str(i), values=(safe(name), safe(rel), safe(text)), tags=tags)
            shown += 1
        if q:
            self.status.config(text=str(shown) + " din " + str(len(self.scripts)) + " scripturi")
    def selected(self):
        sel = self.tree.selection()
        if not sel:
            messagebox.showinfo("Info", "Selectează un script din listă.")
            return None
        return int(sel[0])
    def open_detail(self):
        idx = self.selected()
        if idx is None:
            return
        name, rel, full = self.scripts[idx]
        entry = self.files.setdefault(rel, {})
        win = tk.Toplevel(self.root)
        win.title("Detalii")
        win.transient(self.root)
        w = max(self.root.winfo_width(), 320)
        win.geometry(str(w) + "x" + str(max(self.root.winfo_height() // 2, 280)))
        tk.Label(win, text=safe(rel), wraplength=w - 20, justify="left", anchor="w").pack(fill="x", padx=8, pady=4)
        bar = tk.Frame(win)
        bar.pack(fill="x", padx=8)
        box = tk.Text(win, wrap="word", height=8, font=self.font)
        box.pack(fill="both", expand=True, padx=8, pady=6)
        box.insert("1.0", entry.get("desc") or entry.get("auto", ""))
        def do_auto():
            if box.get("1.0", "end").strip():
                if not messagebox.askyesno("Auto detect", "Înlocuiesc textul din câmp?", parent=win):
                    return
            info = analyze(full, self.root_folder)
            entry["auto"] = info
            box.delete("1.0", "end")
            box.insert("1.0", info)
        def do_update():
            entry["desc"] = box.get("1.0", "end").strip()
            self.save_catalog()
            self.refresh_table()
            win.destroy()
        tk.Button(bar, text="Auto detect", command=do_auto, width=10).pack(side="left", padx=2)
        tk.Button(bar, text="Update", command=do_update, bg="#4CAF50", fg="white", width=8).pack(side="left", padx=2)
        tk.Button(bar, text="Închide", command=win.destroy, width=8).pack(side="left", padx=2)
        win.grab_set()
    def auto_all(self):
        total = len(self.scripts)
        if not total:
            return
        if not messagebox.askyesno("Auto toate", "Regenerez Libs pentru " + str(total) + " scripturi?\nDescrierile tale nu se modifică."):
            return
        for n, (name, rel, full) in enumerate(self.scripts, 1):
            self.files.setdefault(rel, {})["auto"] = analyze(full, self.root_folder)
            if n % 5 == 0:
                self.status.config(text="Analiză " + str(n) + "/" + str(total))
                self.root.update_idletasks()
        self.save_catalog()
        self.refresh_table()
        self.status.config(text="Libs generat pentru " + str(total) + " scripturi")
    def run_script(self):
        idx = self.selected()
        if idx is None:
            return
        full = self.scripts[idx][2]
        try:
            subprocess.Popen([sys.executable, full], cwd=os.path.dirname(full))
            self.status.config(text="Rulat: " + safe(os.path.basename(full)))
        except Exception as e:
            messagebox.showerror("Eroare", "Eroare la rulare:\n" + safe(e))
    def copy_path(self):
        idx = self.selected()
        if idx is None:
            return
        full = self.scripts[idx][2]
        self.root.clipboard_clear()
        self.root.clipboard_append(full)
        self.status.config(text="Copiat: " + safe(full))
if __name__ == "__main__":
    root = tk.Tk()
    App(root)
    root.mainloop()

Sunday, October 4, 2026

tkinter : PyTorch and Keras on pydroid3.

Today, simple python script with pytorch and keras on android pydroid application.
This script is a Tkinter graphical application designed to test and verify PyTorch and Keras packages in Pydroid 3. Right at launch, it tracks execution time down to milliseconds and displays the installed package versions in a status panel. All diagnostics run on a background thread to keep the user interface responsive. For PyTorch, the script tests matrix inversion, automatic gradient calculations using Autograd, data processing through a neural network linear layer, and eigenvalue decomposition. For Keras, the script verifies building a Sequential model with dense layers, compiling and training it on synthetic data for 3 epochs, and running a final prediction to confirm end-to-end execution.
This is the result of the running python script:
Let's see the source code:
import os
import time
import threading
import tkinter as tk
from tkinter import ttk, scrolledtext

# Set Keras backend to torch
os.environ["KERAS_BACKEND"] = "torch"

class AppCheck(tk.Tk):
    def __init__(self):
        self.start_time = time.time()
        
        super().__init__()
        self.title("PyTorch & Keras Inspector")
        self.geometry("480x600")
        
        self.style = ttk.Style(self)
        self.style.theme_use('clam')
        
        self.create_widgets()
        
        # Log UI startup time
        ui_ready = self.get_elapsed_str()
        self.log(f"{ui_ready} Tkinter UI initialized")
        
        # Initial package check
        self.check_initial_imports()

    def get_elapsed_str(self):
        """Returns elapsed time in MM:SS.mmm format without brackets."""
        elapsed = time.time() - self.start_time
        minutes = int(elapsed // 60)
        seconds = int(elapsed % 60)
        millis = int((elapsed - int(elapsed)) * 1000)
        return f"+{minutes:02d}:{seconds:02d}.{millis:03d}"

    def create_widgets(self):
        # Status Frame
        self.info_frame = ttk.LabelFrame(self, text=" Package Status ")
        self.info_frame.pack(fill="x", padx=8, pady=4)

        self.lbl_torch = ttk.Label(self.info_frame, text="PyTorch: Checking...", font=("Helvetica", 9))
        self.lbl_torch.pack(anchor="w", padx=8, pady=2)

        self.lbl_keras = ttk.Label(self.info_frame, text="Keras: Checking...", font=("Helvetica", 9))
        self.lbl_keras.pack(anchor="w", padx=8, pady=2)

        # Run Tests Button
        self.btn_run = ttk.Button(self, text="Run Full Test Suite", command=self.start_tests_thread)
        self.btn_run.pack(pady=6)

        # Console Log Header
        ttk.Label(self, text="Execution Log:", font=("Helvetica", 9, "bold")).pack(anchor="w", padx=8)
        
        # Font size reduced to 7 to fit ~10% more content line-by-line
        self.txt_log = scrolledtext.ScrolledText(self, height=25, font=("Courier", 7))
        self.txt_log.pack(fill="both", expand=True, padx=8, pady=4)

    def log(self, message):
        """Helper to append text to the log window."""
        self.txt_log.insert(tk.END, message + "\n")
        self.txt_log.see(tk.END)

    def check_initial_imports(self):
        t_str = self.get_elapsed_str()
        self.log(f"{t_str} Checking imports...")
        
        # Check PyTorch
        try:
            import torch
            self.lbl_torch.config(text=f"PyTorch: v{torch.__version__}", foreground="green")
            self.log(f"{self.get_elapsed_str()} PyTorch v{torch.__version__} loaded successfully")
        except ImportError as e:
            self.lbl_torch.config(text="PyTorch: Not installed", foreground="red")
            self.log(f"{self.get_elapsed_str()} PyTorch import error: {e}")

        # Check Keras
        try:
            import keras
            self.lbl_keras.config(text=f"Keras: v{keras.__version__}", foreground="green")
            self.log(f"{self.get_elapsed_str()} Keras v{keras.__version__} loaded successfully")
        except ImportError as e:
            self.lbl_keras.config(text="Keras: Not installed", foreground="red")
            self.log(f"{self.get_elapsed_str()} Keras import error: {e}")

    def start_tests_thread(self):
        self.btn_run.config(state="disabled")
        threading.Thread(target=self.run_tests, daemon=True).start()

    def run_tests(self):
        self.log(f"\n{self.get_elapsed_str()} STARTING FULL TEST SUITE")

        # PYTORCH TESTS
        try:
            import torch
            
            # Test 1: Matrix Inverse
            self.log(f"{self.get_elapsed_str()} PyTorch Test 1/4 Matrix Operations")
            A = torch.tensor([[2.0, 1.0], [1.0, 4.0]])
            A_inv = torch.inverse(A)
            self.log(f"   Inverted matrix output:")
            for row in A_inv.numpy():
                self.log(f"   {row}")

            # Test 2: Autograd
            self.log(f"{self.get_elapsed_str()} PyTorch Test 2/4 Autograd Gradient Calculation")
            x = torch.tensor([3.0], requires_grad=True)
            y = x**3 + 2*x
            y.backward()
            self.log(f"   dy/dx at x=3 (expected 29.0): {x.grad.item()}")

            # Test 3: Linear Layer
            self.log(f"{self.get_elapsed_str()} PyTorch Test 3/4 Neural Network Linear Layer")
            lin = torch.nn.Linear(3, 1)
            inp = torch.randn(2, 3)
            out = lin(inp)
            self.log(f"   Output tensor shape: {out.shape}")

            # Test 4: Eigenvalues
            self.log(f"{self.get_elapsed_str()} PyTorch Test 4/4 Eigenvalue Decomposition")
            e_vals = torch.linalg.eigvals(A)
            self.log(f"   Eigenvalues: {e_vals.numpy()}")

        except Exception as e:
            self.log(f"{self.get_elapsed_str()} PyTorch Test Failed: {str(e)}")
            
        # KERAS TESTS
        try:
            import keras
            import numpy as np
            
            # Test 1: Sequential Model Creation
            self.log(f"{self.get_elapsed_str()} Keras Test 1/3 Building Sequential Model")
            model = keras.Sequential([
                keras.layers.Input(shape=(4,)),
                keras.layers.Dense(8, activation="relu"),
                keras.layers.Dense(1, activation="sigmoid")
            ])
            self.log(f"   Model created successfully with {model.count_params()} parameters")

            # Test 2: Model Compilation & Training
            self.log(f"{self.get_elapsed_str()} Keras Test 2/3 Model Compilation and Training")
            model.compile(optimizer="adam", loss="binary_crossentropy")
            
            X_train = np.random.random((32, 4))
            y_train = np.random.randint(2, size=(32, 1))
            
            history = model.fit(X_train, y_train, epochs=3, batch_size=16, verbose=0)
            final_loss = history.history['loss'][-1]
            self.log(f"   3 epochs training completed. Final loss: {final_loss:.4f}")

            # Test 3: Inference / Prediction
            self.log(f"{self.get_elapsed_str()} Keras Test 3/3 Model Prediction")
            X_test = np.random.random((2, 4))
            preds = model.predict(X_test, verbose=0)
            self.log(f"   Prediction results shape: {preds.shape}")
            for pred in preds:
                self.log(f"   Output probabilities: {pred}")

        except Exception as e:
            self.log(f"{self.get_elapsed_str()} Keras Test Failed: {str(e)}")

        self.log(f"\n{self.get_elapsed_str()} TEST SUITE COMPLETED SUCCESSFULLY")
        self.btn_run.config(state="normal")

if __name__ == "__main__":
    app = AppCheck()
    app.mainloop()

Saturday, September 26, 2026

tkinter : basic portfolio for market with simulations ...

Today, one simple script tool for market. this allow to set one portofolio and use with simulations.
Let'see some screenshots:
The basic intro source code ...
import datetime
import json
import threading
import urllib.request
import tkinter as tk
from tkinter import ttk, messagebox

import matplotlib
matplotlib.use("TkAgg")
from matplotlib.figure import Figure
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg


def fetch_yahoo_history(symbol, period_years=1):
    """Descarcă istoricul zilnic de pe Yahoo Finance."""
    end_date = int(datetime.datetime.now().timestamp())
    start_date = int((datetime.datetime.now() - datetime.timedelta(days=365 * period_years)).timestamp())
    
    url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?period1={start_date}&period2={end_date}&interval=1d"
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
    
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read().decode("utf-8"))
            result = data["chart"]["result"][0]
            timestamps = result["timestamp"]
            closes = result["indicators"]["quote"][0]["close"]
            
            history = []
            for ts, cl in zip(timestamps, closes):
                if cl is not None:
                    dt = datetime.datetime.fromtimestamp(ts)
                    history.append((dt, float(cl)))
            return history
    except Exception as e:
        print(f"Eroare la descărcare {symbol}: {e}")
        return []
        ...

Saturday, September 12, 2026

News : Real-Time Stock & ETF Analyzer in using tvscreener and Tkinter and pydroid3.

The tvscreener is a lightweight Python library designed to extract and analyze real-time financial data directly from TradingView's API without requiring paid API keys or complex authentication. It provides quick access to technical indicators, fundamental metrics, historical performance, and volume data for international stocks and ETFs. Explore the source code and documentation on the official tvscreener GitHub repository. The script provides a desktop interface (Tkinter GUI) powered by tvscreener logic to query market data on demand. This allow to choose a ticker symbol from the dropdown list (such as GOOGL, NVDA, AAPL, or ETFs like VUAA). Clicking Check triggers a background HTTP request without freezing the user interface, while displaying real-time progress via the progress bar. The script formats the API response into four core metric categories: Performance, Technical Indicators, Fundamentals, and Volume and Liquidity. Clicking Copy transfers the complete output directly to your system clipboard for quick sharing or note-taking.

Tuesday, June 16, 2026

Python Qt : Network intrusion detection tool with alerts and real‑time HTTP attack detection.

This Python script acts as a local security monitor that watches incoming HTTP requests and alerts the user when suspicious activity is detected. It runs a lightweight HTTP server and inspects every GET and POST request for signs of intrusion, such as dangerous protocols (gopher://), command‑execution keywords (bash, powershell, cmd.exe), encoded payloads, or other attack patterns. Before scanning the request, the script verifies that the incoming connection comes from a specific trusted IP and MAC address. This prevents unauthorized devices from interacting with the server. If the IP or MAC does not match the configured values, the request is immediately rejected.
When a malicious pattern is detected, the script logs the event and displays a popup alert window on the user’s desktop. This popup is implemented using PyQt6 and appears as a small, always‑on‑top notification in the corner of the screen. It fades out automatically after a few seconds, ensuring the user is visually warned without relying on Windows system notifications.
The script also includes a tray icon with a menu that allows the user to update the trusted IP and MAC address. All events are written to a log file for later review. Overall, it functions as a simple intrusion‑detection and alerting tool for local network traffic.
This is a groper test on my browser and result was:
Let's see the source code:
import re
import threading
import queue
import subprocess
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
from datetime import datetime
import sys
from urllib.parse import unquote

# FIX pentru Windows 10/11 – AppID
import ctypes
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("Python.SecurityMonitor")

from PyQt6.QtWidgets import (
    QApplication, QSystemTrayIcon, QMenu, QDialog,
    QVBoxLayout, QLabel, QLineEdit, QPushButton, QWidget
)
from PyQt6.QtGui import QIcon
from PyQt6.QtCore import QTimer, Qt, QPropertyAnimation

# ---------------- CONFIG ---------------- #

ALERT_QUEUE = queue.Queue()
LOG_FILE = "security_log.txt"

MONITOR_IP = "192.168.0.136"
MONITOR_MAC = "60:45:cb:c3:4d:02"   # normalizat

RULES = {
    "Gopher protocol": r"gopher://",
    "CRLF injection": r"%0d%0a",
    "WinRM port": r"5985",
    "Base64 payload": r"base64",
    "bash execution": r"bash",
    "powershell execution": r"powershell",
    "cmd execution": r"cmd\.exe",
}

# ---------------------------------------- #

def log_event(text: str):
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    line = f"[{ts}] {text}\n"
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(line)
    print(line, end="")

def get_mac(ip):
    try:
        output = subprocess.check_output(["arp", "-a", ip], text=True)
        for line in output.splitlines():
            if ip in line:
                mac = line.split()[1]
                mac = mac.replace("-", ":").lower()
                return mac
    except:
        return None

# ---------------- POPUP ALERT ---------------- #

class AlertPopup(QWidget):
    def __init__(self, message):
        super().__init__()
        self.setWindowFlags(
            Qt.WindowType.FramelessWindowHint |
            Qt.WindowType.Tool |
            Qt.WindowType.WindowStaysOnTopHint
        )

        self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)

        layout = QVBoxLayout()
        label = QLabel(message)
        label.setStyleSheet("""
            background-color: #CC0000;
            color: white;
            padding: 12px;
            border-radius: 8px;
            font-size: 14px;
        """)
        layout.addWidget(label)
        self.setLayout(layout)

        self.resize(350, 80)

        screen = QApplication.primaryScreen().geometry()
        self.move(screen.width() - 380, screen.height() - 150)

        self.show()

        self.anim = QPropertyAnimation(self, b"windowOpacity")
        self.anim.setDuration(6000)
        self.anim.setStartValue(1.0)
        self.anim.setEndValue(0.0)
        self.anim.finished.connect(self.close)
        self.anim.start()

# ---------------- HTTP SERVER ---------------- #

class SecurityHandler(BaseHTTPRequestHandler):

    def _check_ip_mac(self):
        global MONITOR_IP, MONITOR_MAC

        client_ip = self.client_address[0]
        # bypass MAC check dacă este același PC
        if client_ip == MONITOR_IP:
            return True, None

        if client_ip != MONITOR_IP:
            return False, f"IP neautorizat: {client_ip}"

        mac = get_mac(client_ip)
        if mac is None:
            return False, f"MAC necunoscut pentru {client_ip}"

        if mac.lower() != MONITOR_MAC.lower():
            return False, f"MAC neautorizat: {mac} pentru IP {client_ip}"

        return True, None

    def _scan_and_block(self, source: str, data: str):
        data_lower = data.lower()
        for reason, pattern in RULES.items():
            if re.search(pattern, data_lower):
                msg = f"Blocked {source}: {reason} → {data}"
                log_event(msg)
                ALERT_QUEUE.put(msg)

                self.send_response(403)
                self.end_headers()
                self.wfile.write(f"Cerere blocată: {reason}".encode("utf-8"))
                return True
        return False

    def do_GET(self):
        ok, reason = self._check_ip_mac()
        if not ok:
            ALERT_QUEUE.put(reason)
            log_event(reason)
            self.send_response(403)
            self.end_headers()
            self.wfile.write(reason.encode())
            return

        url = unquote(self.path)

        if self._scan_and_block("GET URL", url):
            return

        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"Cerere GET permisa")
        log_event(f"Allowed GET: {url}")

    def do_POST(self):
        ok, reason = self._check_ip_mac()
        if not ok:
            ALERT_QUEUE.put(reason)
            log_event(reason)
            self.send_response(403)
            self.end_headers()
            self.wfile.write(reason.encode())
            return

        url = unquote(self.path)

        content_length = int(self.headers.get("Content-Length", 0))
        body = unquote(self.rfile.read(content_length).decode("utf-8", errors="ignore"))

        if self._scan_and_block("POST URL", url):
            return
        if self._scan_and_block("POST body", body):
            return

        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"Cerere POST permisa")
        log_event(f"Allowed POST: {url} | body: {body[:200]}")

def run_server():
    server = HTTPServer(("0.0.0.0", 8080), SecurityHandler)
    log_event("Security server running on http://0.0.0.0:8080")
    server.serve_forever()

# ---------------- DIALOG IP & MAC ---------------- #

class IpMacDialog(QDialog):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Set IP & MAC")

        layout = QVBoxLayout()

        self.ip_label = QLabel("IP Address:")
        self.ip_input = QLineEdit()
        self.ip_input.setText(MONITOR_IP)

        self.mac_label = QLabel("MAC Address:")
        self.mac_input = QLineEdit()
        self.mac_input.setText(MONITOR_MAC)

        self.save_btn = QPushButton("Save")
        self.save_btn.clicked.connect(self.save_values)

        layout.addWidget(self.ip_label)
        layout.addWidget(self.ip_input)
        layout.addWidget(self.mac_label)
        layout.addWidget(self.mac_input)
        layout.addWidget(self.save_btn)

        self.setLayout(layout)

    def save_values(self):
        global MONITOR_IP, MONITOR_MAC
        MONITOR_IP = self.ip_input.text().strip()
        MONITOR_MAC = self.mac_input.text().strip().replace("-", ":").lower()
        log_event(f"Updated IP/MAC → IP={MONITOR_IP}, MAC={MONITOR_MAC}")
        self.hide()

# ---------------- TRAY APP ---------------- #

class TrayApp:
    def __init__(self):
        self.app = QApplication(sys.argv)

        icon = QIcon("icon.svg") if os.path.exists("icon.svg") else QIcon()
        self.tray = QSystemTrayIcon(icon)
        self.tray.setToolTip("Python Security Monitor")

        menu = QMenu()

        set_ip_mac = menu.addAction("Set IP & MAC")
        set_ip_mac.triggered.connect(self.open_ip_mac_dialog)

        exit_action = menu.addAction("Exit")
        exit_action.triggered.connect(self.app.quit)

        self.tray.setContextMenu(menu)
        self.tray.show()

        self.dialog = None

        self.timer = QTimer()
        self.timer.timeout.connect(self.check_alerts)
        self.timer.start(300)

    def open_ip_mac_dialog(self):
        if self.dialog is None:
            self.dialog = IpMacDialog()
        self.dialog.show()
        self.dialog.raise_()
        self.dialog.activateWindow()

    def check_alerts(self):
        while not ALERT_QUEUE.empty():
            msg = ALERT_QUEUE.get()
            AlertPopup(msg)

    def run(self):
        self.app.exec()

# ---------------- MAIN ---------------- #

def main():
    server_thread = threading.Thread(target=run_server, daemon=True)
    server_thread.start()

    tray = TrayApp()
    tray.run()

if __name__ == "__main__":
    main()

Monday, September 8, 2025

Python 3.13.0 : Script for python modules then installs them - updated with fix.

This script scans a folder full of .py files Python scripts, identifies all the external modules they import, filters out built-in ones, writes the installable ones to a requirements.txt file, and then installs them using pip—in parallel threads for speed.
I use the copilot and some comments are into my language, but I tested and works well:
NOTE: I updated with detection python modules based "from" and another issue: check if python module is instaled and step over that python module ...
This script will try to install many python modules, I can update to be better with these issues:
...some modules are default , some scripts are from another area, see Blender 3D with bpy python modules, some packages comes with same modules, this can be soleved with defined lists with unique items.
import subprocess
import sys
import os
import shutil
import importlib.util
import re
import concurrent.futures
from typing import List, Tuple, Set

class ModuleManager:
    def __init__(self):
        self.modules: Set[str] = set()
        self.pip_path = self._get_pip_path()

    def _get_pip_path(self) -> str:
        possible_path = os.path.join(sys.exec_prefix, "Scripts", "pip.exe")
        return shutil.which("pip") or (possible_path if os.path.exists(possible_path) else None)

    def extract_imports_from_file(self, file_path: str) -> List[Tuple[str, str]]:
        imports = []
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                for line in file:
                    # Detect 'import module'
                    import_match = re.match(r'^\s*import\s+([a-zA-Z0-9_]+)(\s+as\s+.*)?$', line)
                    if import_match:
                        module = import_match.group(1)
                        imports.append((module, line.strip()))
                        continue
                    
                    # Detect 'from module import ...'
                    from_match = re.match(r'^\s*from\s+([a-zA-Z0-9_]+)\s+import\s+.*$', line)
                    if from_match:
                        module = from_match.group(1)
                        imports.append((module, line.strip()))
        except FileNotFoundError:
            print(f"❌ Fișierul {file_path} nu a fost găsit.")
        except Exception as e:
            print(f"❌ Eroare la citirea fișierului {file_path}: {e}")
        return imports

    def scan_directory_for_py_files(self, directory: str = '.') -> List[str]:
        py_files = []
        for root, _, files in os.walk(directory):
            for file in files:
                if file.endswith('.py'):
                    py_files.append(os.path.join(root, file))
        return py_files

    def collect_unique_modules(self, directory: str = '.') -> None:
        py_files = self.scan_directory_for_py_files(directory)
        all_imports = []
        with concurrent.futures.ThreadPoolExecutor() as executor:
            future_to_file = {executor.submit(self.extract_imports_from_file, file_path): file_path for file_path in py_files}
            for future in concurrent.futures.as_completed(future_to_file):
                imports = future.result()
                all_imports.extend(imports)
        
        for module, _ in all_imports:
            self.modules.add(module)

    def is_module_installed(self, module: str) -> bool:
        return importlib.util.find_spec(module) is not None

    def run_pip_install(self, module: str) -> bool:
        if not self.pip_path:
            print(f"❌ Nu am găsit pip pentru {module}.")
            return False
        try:
            subprocess.check_call([self.pip_path, "install", module])
            print(f"✅ Pachetul {module} a fost instalat cu succes.")
            return True
        except subprocess.CalledProcessError as e:
            print(f"❌ Eroare la instalarea pachetului {module}: {e}")
            return False

    def check_and_install_modules(self) -> None:
        def process_module(module):
            print(f"\n🔎 Verific dacă {module} este instalat...")
            if self.is_module_installed(module):
                print(f"✅ {module} este deja instalat.")
            else:
                print(f"📦 Instalez {module}...")
                self.run_pip_install(module)
                # Re-verifică după instalare
                if self.is_module_installed(module):
                    print(f"✅ {module} funcționează acum.")
                else:
                    print(f"❌ {module} nu funcționează după instalare.")

        with concurrent.futures.ThreadPoolExecutor() as executor:
            executor.map(process_module, self.modules)

def main():
    print("🔍 Verific pip...")
    manager = ModuleManager()
    if manager.pip_path:
        print(f"✅ Pip este disponibil la: {manager.pip_path}")
    else:
        print("⚠️ Pip nu este disponibil.")
        return

    directory = sys.argv[1] if len(sys.argv) > 1 else '.'
    print(f"\n📜 Scanez directorul {directory} pentru fișiere .py...")
    manager.collect_unique_modules(directory)
    
    if not manager.modules:
        print("⚠️ Nu s-au găsit module în importuri.")
        return
    
    print(f"\nModule unice detectate: {', '.join(manager.modules)}")
    manager.check_and_install_modules()

if __name__ == "__main__":
    main()

Tuesday, February 25, 2020

Python 3.7.6 : The new concepts of execution in python 3 - part 001.

The main goal of these tutorials series is learning to deal with python source code using the new concepts of execution in python 3.
When two or more events are concurrent it means that they are happening at the same time.
Concurrent programming is not equivalent to parallel execution.
In computing, concurrency is the execution of pieces of work or tasks by a computer at the same time.
Concurrency is a property which more than one operation can be run simultaneously.
When multiple computations or operations are carried out at the same time or in parallel with the goal of speeding up the computation process then this process is named parallelism.
Parallelism is a property which operations are actually being run simultaneously using the multiprocessing.
Multiprocessing, on the other hand, involves utilizing two or more processor units on a computer to achieve parallelism.
Multithreading is a property that refers to the ability of a CPU to execute multiple threads concurrently.
Python’s concurrency methods including threading, multiprocessing, and asyncio.
The difference between the threading and multiprocessing is this: the threading module uses threads, the multiprocessing module uses processes.
The threading is the package that provides API to create and manage threads.
With multiprocessing, Python creates new processes using an API similar to the threading module.
The asyncio is a library to write concurrent code using the async/await syntax.
The keyword async indicates that our function is a coroutine meaning they choose when to pause and let others execute and run coroutines multitask cooperatively.
The three fundamental advantages of async and await over threads are:
  • cooperative multitasking - you can reasonably have millions of concurrent tasks;
  • using await makes visible where the schedule points;
  • if a task doesn’t yield then it can accidentally block all other tasks from running;
  • tasks can support cancellation.
The next source of code show us how can deal with the execution code in Python 3 using the threading and multiprocessing python packages.
The timeit python package is used to benchmark the code write in the code_to_test variable:
Let's test the multi-threading execution with python:
[mythcat@desk ~]$ python3 
Python 3.7.6 (default, Jan 30 2020, 09:44:41) 
[GCC 9.2.1 20190827 (Red Hat 9.2.1-1)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import timeit
>>> code_to_test = """
... import threading
... 
... text = "Hello World"
... 
... def print_text(text):
...     for char in text:
...             print (char)
... 
... # multi-threading execution
... def multi_threads():
...     thread_1 = threading.Thread(target=print_text(text))
...     thread_2 = threading.Thread(target=print_text(text))
...     thread_1.start()
...     thread_2.start()
...     thread_1.join()
...     thread_2.join()
... multi_threads()
... """
>>> 
>>> elapsed_time = timeit.timeit(code_to_test, number=1)
H
e
l
l
o
 
W
o
r
l
d
H
e
l
l
o
 
W
o
r
l
d
>>> print(elapsed_time)
0.010613240000566293
Let's test the serially execution with python:
[mythcat@desk ~]$ python3 
Python 3.7.6 (default, Jan 30 2020, 09:44:41) 
[GCC 9.2.1 20190827 (Red Hat 9.2.1-1)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import timeit
>>> code_to_test = """
... import threading
... 
... text = "Hello World"
... 
... def print_text(text):
...     for char in text:
...             print (char)
... # serially execution
... def serially():
...     print_text(text)
...     print_text(text)
... serially()
... """
>>> elapsed_time = timeit.timeit(code_to_test, number=1)
H
e
l
l
o
 
W
o
r
l
d
H
e
l
l
o
 
W
o
r
l
d
>>> print(elapsed_time)
0.011771811000471644
Let's test the multiprocessing execution with python:
[mythcat@desk ~]$ python3 
Python 3.7.6 (default, Jan 30 2020, 09:44:41) 
[GCC 9.2.1 20190827 (Red Hat 9.2.1-1)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import timeit
>>> code_to_test = """
... import multiprocessing
... 
... text = "Hello World"
... 
... def print_text(text):
...     for char in text:
...             print (char)
... 
... # multiprocessing execution
... def multiprocessing_test():
...      process_1 = multiprocessing.Process(target=print_text(text))
...      process_2 = multiprocessing.Process(target=print_text(text))
...      process_1.start()
...      process_2.start()
...      process_1.join()
...      process_2.join()
... multiprocessing_test()
... """
>>> elapsed_time = timeit.timeit(code_to_test, number=1)
H
e
l
l
o
 
W
o
r
l
d
H
e
l
l
o
 
W
o
r
l
d
>>> print(elapsed_time)
0.3649730779998208
Since asyncio is a little complex, I will write about this in the next tutorial.