typer.py

"""
Typer — simulates keystrokes into whatever app is focused.

Typing engine: AppleScript's `keystroke` via `osascript`. This produces
real hardware-style key events so apps like Google Docs treat them as
genuine typing rather than pasted text.

Extras for looking more human in Google Docs version history:
  - Humanized timing (log-normal jitter, word/punctuation pauses, thinking pauses).
  - Optional chunked mode: type N chars, pause for S seconds, repeat.
  - Mouse jitter during long pauses (real OS-level mouse events).
  - Occasional "re-read" passes: move cursor back a few chars, pause, move forward.
"""

import random
import subprocess
import threading
import time
import tkinter as tk
from tkinter import ttk, messagebox

from pynput.keyboard import Key, Listener
from pynput.mouse import Controller as MouseController

mouse = MouseController()

# Set by the Stop button or Esc key to signal the typing thread to bail out.
stop_event = threading.Event()

# QWERTY neighbors — used to pick realistic typo keys.
NEIGHBORS = {
    'a': 'qwsz', 'b': 'vghn', 'c': 'xdfv', 'd': 'serfcx', 'e': 'wsdr',
    'f': 'drtgvc', 'g': 'ftyhbv', 'h': 'gyujnb', 'i': 'ujko', 'j': 'huiknm',
    'k': 'jiolm', 'l': 'kop', 'm': 'njk', 'n': 'bhjm', 'o': 'iklp',
    'p': 'ol', 'q': 'wa', 'r': 'edft', 's': 'awedxz', 't': 'rfgy',
    'u': 'yhji', 'v': 'cfgb', 'w': 'qase', 'x': 'zsdc', 'y': 'tghu',
    'z': 'asx',
}

# macOS virtual keycodes for keys we can't send as `keystroke "..."`.
KEYCODE_RETURN = 36
KEYCODE_TAB = 48
KEYCODE_BACKSPACE = 51
KEYCODE_LEFT = 123
KEYCODE_RIGHT = 124


def _osa(script: str):
    """Run an AppleScript one-liner. Swallow errors so typing keeps flowing."""
    try:
        subprocess.run(
            ["osascript", "-e", script],
            check=False, capture_output=True, timeout=2,
        )
    except Exception:
        pass


def type_char(char: str):
    """Send a single character as a real keystroke."""
    if char == '\n' or char == '\r':
        _osa(f'tell application "System Events" to key code {KEYCODE_RETURN}')
    elif char == '\t':
        _osa(f'tell application "System Events" to key code {KEYCODE_TAB}')
    else:
        escaped = char.replace('\\', '\\\\').replace('"', '\\"')
        _osa(f'tell application "System Events" to keystroke "{escaped}"')


def press_key(keycode: int):
    _osa(f'tell application "System Events" to key code {keycode}')


def press_backspace():
    press_key(KEYCODE_BACKSPACE)


def typo_for(char: str) -> str:
    """Return a plausible wrong key for `char`, or '' if we shouldn't typo it."""
    lower = char.lower()
    if lower in NEIGHBORS:
        wrong = random.choice(NEIGHBORS[lower])
        return wrong.upper() if char.isupper() else wrong
    return ''


def interruptible_sleep(seconds: float) -> bool:
    """Sleep, but return False immediately if stop_event is set."""
    return not stop_event.wait(max(seconds, 0))


def jitter_mouse(steps: int = 3):
    """Nudge the mouse a few pixels in a random walk. Real OS-level events."""
    try:
        x, y = mouse.position
        for _ in range(steps):
            if stop_event.is_set():
                return
            dx = random.randint(-6, 6)
            dy = random.randint(-6, 6)
            mouse.move(dx, dy)
            time.sleep(random.uniform(0.05, 0.15))
        # Return roughly to origin so we don't drift off-screen over time.
        nx, ny = mouse.position
        mouse.move(int(x - nx), int(y - ny))
    except Exception:
        pass


def reread_pass():
    """
    Simulate a writer glancing back at what they just wrote: tap Left a few
    times, pause, then tap Right the same number of times to return.
    """
    n = random.randint(2, 5)
    for _ in range(n):
        if stop_event.is_set():
            return
        press_key(KEYCODE_LEFT)
        time.sleep(random.uniform(0.04, 0.1))
    if not interruptible_sleep(random.uniform(0.4, 1.2)):
        return
    for _ in range(n):
        if stop_event.is_set():
            return
        press_key(KEYCODE_RIGHT)
        time.sleep(random.uniform(0.04, 0.1))


def human_delay(base_delay: float, char: str, next_char: str) -> float:
    """Return a realistic delay *after* typing `char`."""
    jitter = random.lognormvariate(0, 0.25)
    delay = base_delay * jitter

    if char == ' ':
        delay += base_delay * random.uniform(1.5, 2.5)
    elif char == ',':
        delay += base_delay * random.uniform(3, 5)
    elif char in '.!?':
        delay += base_delay * random.uniform(6, 10)
    elif char == '\n':
        delay += base_delay * random.uniform(4, 8)

    if next_char and next_char.isupper():
        delay += base_delay * 0.3

    return delay


def split_sentences(text: str):
    """Split text into sentences, preserving punctuation and trailing whitespace."""
    sentences = []
    current = []
    i = 0
    while i < len(text):
        c = text[i]
        current.append(c)
        if c in '.!?':
            # Swallow any whitespace that follows, so sentence concat == original text.
            j = i + 1
            while j < len(text) and text[j] in ' \t\n':
                current.append(text[j])
                j += 1
            sentences.append(''.join(current))
            current = []
            i = j
        else:
            i += 1
    if current:
        sentences.append(''.join(current))
    return sentences


# When chunked mode is on, this is the per-sentence probability of a
# "false start" — typing part of the sentence, realizing it's wrong,
# backspacing it, and rewriting. Only applies to sentences ≥ 30 chars.
FALSE_START_RATE = 0.25


def backspace_many(count: int):
    """Quickly backspace `count` characters, with small human-ish jitter."""
    for _ in range(count):
        if stop_event.is_set():
            return
        press_backspace()
        time.sleep(random.uniform(0.03, 0.08))


def _type_run(chars, state, base_delay, osa_overhead, error_rate,
              chunked, chunk_chars, chunk_pause, total, status_callback):
    """
    Type a run of characters. `state` is a dict carrying counters across
    multiple calls (thinking pauses, rereads, chunk progress, global index).
    Returns True normally, False if stopped.
    """
    for idx, char in enumerate(chars):
        if stop_event.is_set():
            status_callback("Stopped.")
            return False

        next_char = chars[idx + 1] if idx + 1 < len(chars) else ''
        delay = max(human_delay(base_delay, char, next_char) - osa_overhead, 0)

        if random.random() < error_rate:
            wrong = typo_for(char)
            if wrong:
                type_char(wrong)
                if not interruptible_sleep(delay * random.uniform(1.5, 2.5)):
                    status_callback("Stopped.")
                    return False
                press_backspace()
                if not interruptible_sleep(delay * 0.5):
                    status_callback("Stopped.")
                    return False

        type_char(char)
        state['chars_this_chunk'] += 1
        state['global_i'] += 1
        if not interruptible_sleep(delay):
            status_callback("Stopped.")
            return False

        # Thinking pause between words.
        state['next_thinking_pause'] -= 1
        if state['next_thinking_pause'] <= 0 and char == ' ':
            jitter_mouse(2)
            if not interruptible_sleep(random.uniform(0.5, 1.5)):
                status_callback("Stopped.")
                return False
            state['next_thinking_pause'] = random.randint(15, 40)

        # Occasional re-read.
        state['next_reread'] -= 1
        if state['next_reread'] <= 0 and char == ' ':
            reread_pass()
            state['next_reread'] = random.randint(150, 300)

        # Chunked mode: long idle gap at word boundaries.
        if (chunked and state['chars_this_chunk'] >= chunk_chars
                and char == ' ' and state['global_i'] < total):
            pause_sec = chunk_pause * random.uniform(0.8, 1.2)
            status_callback(f"Chunk pause ({pause_sec:.0f}s)… {state['global_i']}/{total}")
            bursts = max(1, int(pause_sec / 6))
            per_burst = pause_sec / bursts
            for _ in range(bursts):
                if stop_event.is_set():
                    status_callback("Stopped.")
                    return False
                jitter_mouse(random.randint(2, 5))
                if not interruptible_sleep(per_burst - 0.3):
                    status_callback("Stopped.")
                    return False
            state['chars_this_chunk'] = 0

        if state['global_i'] % 10 == 0:
            status_callback(f"Typing… {state['global_i']}/{total}")

    return True


def pick_wrong_sentence(current_idx: int, sentences):
    """
    Pick a *different* sentence from the text to use as a mistaken first draft.
    Returns the stripped sentence text, or '' if no suitable alternative exists.
    """
    candidates = [
        s.strip().rstrip('.!?').strip()
        for j, s in enumerate(sentences)
        if j != current_idx and len(s.strip()) >= 20
    ]
    if not candidates:
        return ''
    return random.choice(candidates)


def type_text(text, cpm, error_rate, chunked, chunk_chars, chunk_pause, status_callback):
    """Type `text` with humanization, optional chunked pauses, and — when
    chunked is on — occasional sentence-level false starts: type a chunk of a
    *different* sentence (a real mistake), pause, backspace it, type the
    correct sentence."""
    base_delay = 60.0 / max(cpm, 1)
    osa_overhead = 0.03
    total = len(text)

    state = {
        'next_thinking_pause': random.randint(15, 40),
        'next_reread': random.randint(150, 300),
        'chars_this_chunk': 0,
        'global_i': 0,
    }

    sentences = split_sentences(text)

    for s_idx, sentence in enumerate(sentences):
        if stop_event.is_set():
            status_callback("Stopped.")
            return

        # Decide whether to do a false start on this sentence.
        # Only when chunked mode is on, only on sentences with real substance.
        do_false_start = (
            chunked
            and len(sentence.strip()) >= 30
            and random.random() < FALSE_START_RATE
        )

        if do_false_start:
            wrong_sentence = pick_wrong_sentence(s_idx, sentences)
            if wrong_sentence:
                # Type a chunk of the wrong sentence — usually 30–70% of it,
                # capped so we don't type a whole alternate paragraph.
                max_wrong = min(len(wrong_sentence), max(25, int(len(sentence) * 0.7)))
                min_wrong = min(max_wrong, max(20, int(len(sentence) * 0.3)))
                wrong_len = random.randint(min_wrong, max_wrong)
                wrong_chunk = wrong_sentence[:wrong_len]

                # Type the wrong content (counts toward chunk/global progress
                # like real typing — we'll undo the counters after backspace).
                if not _type_run(wrong_chunk, state, base_delay, osa_overhead, error_rate,
                                 chunked, chunk_chars, chunk_pause, total, status_callback):
                    return

                # "Wait, this isn't right" pause — longer than a partial redo,
                # because the typist notices the whole thought is off.
                status_callback("Realizing the mistake…")
                jitter_mouse(4)
                if not interruptible_sleep(random.uniform(1.0, 2.5)):
                    status_callback("Stopped.")
                    return

                # Backspace the whole wrong chunk.
                backspace_many(wrong_len)
                state['chars_this_chunk'] = max(0, state['chars_this_chunk'] - wrong_len)
                state['global_i'] = max(0, state['global_i'] - wrong_len)

                # Brief settle before retyping.
                if not interruptible_sleep(random.uniform(0.4, 1.0)):
                    status_callback("Stopped.")
                    return

        # Type the correct sentence (either after a false start, or fresh).
        if not _type_run(sentence, state, base_delay, osa_overhead, error_rate,
                         chunked, chunk_chars, chunk_pause, total, status_callback):
            return

    status_callback("Done.")


class App:
    def __init__(self, root):
        root.title("Typer")
        root.geometry("580x700")

        frm = ttk.Frame(root, padding=14)
        frm.pack(fill="both", expand=True)

        ttk.Label(frm, text="Paste text to type:").pack(anchor="w")
        self.text = tk.Text(frm, height=10, wrap="word")
        self.text.pack(fill="both", expand=True, pady=(4, 10))

        # Speed slider
        self.speed_var = tk.IntVar(value=300)
        self.speed_label = self._slider_row(frm, "Speed (chars/min):", self.speed_var,
                                            60, 700, lambda v: str(int(float(v))))

        # Error rate slider
        self.err_var = tk.DoubleVar(value=2.0)
        self.err_label = self._slider_row(frm, "Error rate (%):", self.err_var,
                                          0, 15, lambda v: f"{float(v):.1f}")

        # Chunked mode controls
        chunked_row = ttk.Frame(frm)
        chunked_row.pack(fill="x", pady=(8, 2))
        self.chunked_var = tk.BooleanVar(value=False)
        ttk.Checkbutton(chunked_row, text="Chunked mode (pause periodically to create version-history breaks)",
                        variable=self.chunked_var).pack(anchor="w")

        self.chunk_size_var = tk.IntVar(value=120)
        self.chunk_size_label = self._slider_row(frm, "Chunk size (chars):", self.chunk_size_var,
                                                 30, 400, lambda v: str(int(float(v))))

        self.chunk_pause_var = tk.IntVar(value=45)
        self.chunk_pause_label = self._slider_row(frm, "Chunk pause (sec):", self.chunk_pause_var,
                                                  10, 180, lambda v: f"{int(float(v))}s")

        # Buttons row
        btn_row = ttk.Frame(frm)
        btn_row.pack(pady=10)
        self.start_btn = ttk.Button(btn_row, text="Start (5s countdown)", command=self.on_start)
        self.start_btn.pack(side="left", padx=4)
        self.stop_btn = ttk.Button(btn_row, text="Stop", command=self.on_stop, state="disabled")
        self.stop_btn.pack(side="left", padx=4)

        self.status = ttk.Label(frm, text="Ready. Click Start, then focus the target app. (Esc also stops.)")
        self.status.pack(anchor="w")

        self._esc_listener = Listener(on_press=self._on_global_key)
        self._esc_listener.daemon = True
        self._esc_listener.start()

    def _slider_row(self, parent, label_text, var, lo, hi, fmt):
        row = ttk.Frame(parent)
        row.pack(fill="x", pady=4)
        ttk.Label(row, text=label_text).pack(side="left")
        value_label = ttk.Label(row, text=fmt(var.get()))
        value_label.pack(side="right")
        scale = ttk.Scale(
            row, from_=lo, to=hi, orient="horizontal", variable=var,
            command=lambda v: value_label.config(text=fmt(v)),
        )
        scale.pack(side="left", fill="x", expand=True, padx=10)
        return value_label

    def _on_global_key(self, key):
        if key == Key.esc and not self.stop_btn.instate(["disabled"]):
            self.on_stop()

    def set_status(self, msg: str):
        self.status.after(0, lambda: self.status.config(text=msg))
        if msg in ("Done.", "Stopped.") or msg.startswith("Error:"):
            self.status.after(0, self._reset_buttons)

    def _reset_buttons(self):
        self.start_btn.config(state="normal")
        self.stop_btn.config(state="disabled")

    def on_start(self):
        text = self.text.get("1.0", "end-1c")
        if not text.strip():
            messagebox.showwarning("Typer", "Paste some text first.")
            return

        cpm = self.speed_var.get()
        error_rate = self.err_var.get() / 100.0
        chunked = self.chunked_var.get()
        chunk_chars = self.chunk_size_var.get()
        chunk_pause = self.chunk_pause_var.get()

        stop_event.clear()
        self.start_btn.config(state="disabled")
        self.stop_btn.config(state="normal")

        def countdown_then_type():
            for s in range(5, 0, -1):
                if stop_event.is_set():
                    self.set_status("Stopped.")
                    return
                self.set_status(f"Starting in {s}… focus your target app!")
                time.sleep(1)
            try:
                type_text(text, cpm, error_rate, chunked, chunk_chars, chunk_pause, self.set_status)
            except Exception as e:
                self.set_status(f"Error: {e}")

        threading.Thread(target=countdown_then_type, daemon=True).start()

    def on_stop(self):
        stop_event.set()
        self.set_status("Stopping…")


if __name__ == "__main__":
    root = tk.Tk()
    App(root)
    root.mainloop()