feat: parity with rust fork — LLM auto-fallback + 6 new action types + 20+ commands
Brings the python fork in line with the new feature/pc-tools work that
landed in rust. Same UX, same trigger phrases, same env vars.
config.py:
+ LLM_AUTO_FALLBACK = True
+ LLM_AUTO_FALLBACK_MIN_CHARS = 4
Mirrors the rust config knobs. The old "скажи X" trigger still works
(VA_TBR strips it before intent matching, so what remains hits this
fallback anyway) — but now ANY unrecognized query goes to Groq, not
just "скажи" ones.
main.py:
va_respond: replaced the fuzz-on-"скажи" gate with a unified auto-
fallback that fires whenever no command matches, GROQ_TOKEN is set,
and the utterance is at least LLM_AUTO_FALLBACK_MIN_CHARS long.
+ _speak(text): recorder.stop() → tts.va_speak → start, used by the
new action handlers so the mic doesn't pick up the assistant.
+ _set_clipboard(text): subprocess to PowerShell Set-Clipboard, no
new python deps.
+ _strip_trigger(voice, triggers): shared helper for codegen/file_search
where we need the text AFTER the trigger phrase.
New action handlers (registered in run_action dispatch):
+ groq_codegen — extracts query, calls Groq with strict "code only"
system prompt at temp 0.2, strips ```lang ... ``` fences, puts the
result in the clipboard, speaks "Код в буфере, сэр. Строк: N".
+ ocr_screen — PowerShell System.Drawing screenshot to a temp PNG,
shells to tesseract.exe (-l rus+eng), speaks a 120-char preview,
clipboard. Looks for tesseract on PATH first then standard install
dirs; gives a clear "Tesseract не установлен" if missing.
+ sysinfo — CIM-based battery/cpu/ram/disk via PS one-liners, time
via python datetime (gets us localized weekday for free). Topics:
time / battery / cpu / ram / disk / all.
+ file_search — strips trigger, recursive Get-ChildItem in Desktop/
Documents/Downloads (depth 3, top 5), opens explorer /select on
the first hit, speaks count and basename.
+ audio_devices_panel — opens mmsys.cpl. Proper COM-based device
switching (like the rust IPolicyConfig version) is deferred — TODO
in backlog.
commands.yaml:
Total 33 commands (was 19). New:
+ open_notepad / open_calculator / open_explorer / open_terminal
/ open_settings / open_task_manager / lock_screen / screenshot
— all via shell action (no AHK exes needed).
+ volume_up / volume_down — five chained keys=volumeup/volumedown
presses each, like rust's _volume_helper.ps1 with Times=5.
+ media_stop — keys=stop.
+ sysinfo_time / sysinfo_battery / sysinfo_cpu / sysinfo_ram /
sysinfo_disk / sysinfo_all — sysinfo action.
+ file_search, codegen, read_screen, switch_audio_device.
Replaced (were AHK exes that Priler ships and we don't have):
~ music → keys playpause
~ music_off → keys playpause (toggle)
~ music_next → keys nexttrack
~ music_prev → keys prevtrack
Removed (Priler hardware/setup specific, can't sensibly port):
- gaming_mode_on / gaming_mode_off (display config switch)
- music_save (depends on the specific player AHK was driving)
- switch_to_headphones / switch_to_dynamics (replaced with the
generic switch_audio_device that opens mmsys.cpl)
Kept as-is (worked already, no AHK dependency):
open_browser / open_youtube / open_google (url action)
sound_off / sound_on (pycaw via system.volume_mute/unmute)
thanks / stupid / off
Smoke-tested: yaml.safe_load → 33 commands, ast.parse main.py → OK.
Portability: USERPROFILE for screenshot path, $env:SystemDrive for disk,
PATH lookup for tesseract before falling back to standard install dirs.
No hardcoded user-specific paths.
This commit is contained in:
Bossiara13
parent
53678fbdfa
commit
80a685230c
3 changed files with 463 additions and 151 deletions
238
main.py
238
main.py
|
|
@ -168,10 +168,12 @@ def va_respond(voice: str):
|
|||
if len(cmd['cmd'].strip()) <= 0:
|
||||
return False
|
||||
elif cmd['percent'] < min_percent or cmd['cmd'] not in VA_CMD_LIST.keys():
|
||||
# play("not_found")
|
||||
# tts.va_speak("Что?")
|
||||
if fuzz.ratio(voice.join(voice.split()[:1]).strip(), "скажи") > 75:
|
||||
|
||||
auto_ok = (
|
||||
config.LLM_AUTO_FALLBACK
|
||||
and config.GROQ_TOKEN
|
||||
and len(voice.strip()) >= config.LLM_AUTO_FALLBACK_MIN_CHARS
|
||||
)
|
||||
if auto_ok:
|
||||
message_log.append({"role": "user", "content": voice})
|
||||
response = gpt_answer()
|
||||
message_log.append({"role": "assistant", "content": response})
|
||||
|
|
@ -224,6 +226,214 @@ def _shutdown():
|
|||
sys.exit(0)
|
||||
|
||||
|
||||
def _speak(text: str):
|
||||
recorder.stop()
|
||||
tts.va_speak(text)
|
||||
time.sleep(0.3)
|
||||
recorder.start()
|
||||
|
||||
|
||||
def _set_clipboard(text: str):
|
||||
try:
|
||||
subprocess.run(
|
||||
['powershell', '-NoProfile', '-Command', 'Set-Clipboard -Value $input'],
|
||||
input=text, text=True, encoding='utf-8', timeout=10, check=False,
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"clipboard set failed: {e}")
|
||||
|
||||
|
||||
_CODEGEN_TRIGGERS = (
|
||||
'сгенерируй скрипт', 'напиши скрипт', 'сделай скрипт',
|
||||
'сгенерируй код', 'набросай код', 'напиши код',
|
||||
)
|
||||
|
||||
|
||||
def _strip_trigger(voice: str, triggers) -> str:
|
||||
lower = voice.lower()
|
||||
for t in triggers:
|
||||
if lower.startswith(t):
|
||||
return voice[len(t):].strip()
|
||||
return voice.strip()
|
||||
|
||||
|
||||
def do_codegen(action, voice: str):
|
||||
query = _strip_trigger(voice, _CODEGEN_TRIGGERS)
|
||||
if not query:
|
||||
_speak("Что писать, сэр?")
|
||||
return
|
||||
if not config.GROQ_TOKEN:
|
||||
_speak("Groq токен не задан.")
|
||||
return
|
||||
|
||||
sys_prompt = (
|
||||
"Ты Senior Engineer. На запрос пользователя верни ТОЛЬКО исходный код, "
|
||||
"без обрамления тройными кавычками и без объяснений. Если язык не задан "
|
||||
"— выбери самый подходящий (по умолчанию Python). Не пиши комментариев-болтовни."
|
||||
)
|
||||
try:
|
||||
resp = client.chat.completions.create(
|
||||
model=config.GROQ_MODEL,
|
||||
messages=[
|
||||
{"role": "system", "content": sys_prompt},
|
||||
{"role": "user", "content": query},
|
||||
],
|
||||
max_tokens=1024,
|
||||
temperature=0.2,
|
||||
)
|
||||
code = resp.choices[0].message.content.strip()
|
||||
except Exception as e:
|
||||
print(f"codegen api failed: {e}")
|
||||
_speak("Не получилось сгенерировать код.")
|
||||
return
|
||||
|
||||
import re
|
||||
code = re.sub(r'^```[\w+\-]*\n?', '', code)
|
||||
code = re.sub(r'\n?```\s*$', '', code).strip()
|
||||
|
||||
_set_clipboard(code)
|
||||
lines = code.count('\n') + 1 if code else 0
|
||||
_speak(f"Код в буфере, сэр. Строк: {lines}.")
|
||||
|
||||
|
||||
def _find_tesseract():
|
||||
candidates = (
|
||||
r'C:\Program Files\Tesseract-OCR\tesseract.exe',
|
||||
r'C:\Program Files (x86)\Tesseract-OCR\tesseract.exe',
|
||||
)
|
||||
on_path = subprocess.run(['where', 'tesseract.exe'], capture_output=True, text=True)
|
||||
if on_path.returncode == 0 and on_path.stdout.strip():
|
||||
return 'tesseract.exe'
|
||||
for c in candidates:
|
||||
if os.path.isfile(c):
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def do_ocr(action, voice: str):
|
||||
import tempfile
|
||||
tess = _find_tesseract()
|
||||
if not tess:
|
||||
_speak("Tesseract не установлен.")
|
||||
return
|
||||
|
||||
tmp = tempfile.NamedTemporaryFile(suffix='.png', delete=False)
|
||||
tmp.close()
|
||||
png = tmp.name
|
||||
|
||||
capture_ps = (
|
||||
"Add-Type -AssemblyName System.Windows.Forms,System.Drawing; "
|
||||
"$b=[System.Windows.Forms.Screen]::PrimaryScreen.Bounds; "
|
||||
"$bmp=New-Object Drawing.Bitmap $b.Width,$b.Height; "
|
||||
"$g=[Drawing.Graphics]::FromImage($bmp); "
|
||||
"$g.CopyFromScreen($b.Location,[Drawing.Point]::Empty,$b.Size); "
|
||||
f"$bmp.Save('{png.replace(chr(92), chr(92)+chr(92))}'); "
|
||||
"$g.Dispose(); $bmp.Dispose()"
|
||||
)
|
||||
cap = subprocess.run(
|
||||
['powershell', '-NoProfile', '-ExecutionPolicy', 'Bypass', '-Command', capture_ps],
|
||||
capture_output=True, text=True, timeout=15,
|
||||
)
|
||||
if cap.returncode != 0 or not os.path.isfile(png):
|
||||
try: os.unlink(png)
|
||||
except OSError: pass
|
||||
_speak("Не смог снять экран.")
|
||||
return
|
||||
|
||||
try:
|
||||
ocr = subprocess.run(
|
||||
[tess, png, '-', '-l', 'rus+eng'],
|
||||
capture_output=True, text=True, encoding='utf-8', timeout=20,
|
||||
)
|
||||
text = (ocr.stdout or '').strip()
|
||||
finally:
|
||||
try: os.unlink(png)
|
||||
except OSError: pass
|
||||
|
||||
if not text:
|
||||
_speak("Текста на экране не нашёл.")
|
||||
return
|
||||
|
||||
_set_clipboard(text)
|
||||
preview = ' '.join(text.split())[:120]
|
||||
_speak(f"Текст в буфере. Начало: {preview}")
|
||||
|
||||
|
||||
_FILE_SEARCH_TRIGGERS = (
|
||||
'найди документ', 'найди файл', 'поиск файла', 'ищи файл', 'где файл',
|
||||
)
|
||||
|
||||
|
||||
def do_file_search(action, voice: str):
|
||||
query = _strip_trigger(voice, _FILE_SEARCH_TRIGGERS)
|
||||
if not query:
|
||||
_speak("Что искать?")
|
||||
return
|
||||
user_profile = os.environ.get('USERPROFILE', os.path.expanduser('~'))
|
||||
roots = [
|
||||
os.path.join(user_profile, 'Desktop'),
|
||||
os.path.join(user_profile, 'Documents'),
|
||||
os.path.join(user_profile, 'Downloads'),
|
||||
]
|
||||
escaped = query.replace("'", "''")
|
||||
roots_arg = ",".join(f"'{r}'" for r in roots)
|
||||
ps = (
|
||||
f"$ErrorActionPreference='SilentlyContinue'; $r=@({roots_arg}); $q='*{escaped}*'; "
|
||||
"$hits=Get-ChildItem -Path $r -Filter $q -Recurse -Depth 3 -File | Select-Object -First 5; "
|
||||
"foreach($h in $hits){ Write-Output $h.FullName }"
|
||||
)
|
||||
res = subprocess.run(
|
||||
['powershell', '-NoProfile', '-Command', ps],
|
||||
capture_output=True, text=True, encoding='utf-8', timeout=15,
|
||||
)
|
||||
lines = [l for l in (res.stdout or '').splitlines() if l.strip()]
|
||||
if not lines:
|
||||
_speak(f"Не нашёл «{query}».")
|
||||
return
|
||||
first = lines[0]
|
||||
subprocess.Popen(['explorer.exe', '/select,', first])
|
||||
_speak(f"Найдено {len(lines)}, открываю первое: {os.path.basename(first)}.")
|
||||
|
||||
|
||||
def do_audio_devices_panel(action, voice: str):
|
||||
# Full programmatic switching needs IPolicyConfig + IMMDeviceEnumerator
|
||||
# via comtypes; out of scope for this pass. Opens Sound control panel so
|
||||
# the user can pick visually. Rust fork has the COM-based version.
|
||||
subprocess.Popen(['rundll32.exe', 'shell32.dll,Control_RunDLL', 'mmsys.cpl,,0'])
|
||||
_speak("Открываю настройки звука, сэр.")
|
||||
|
||||
|
||||
def do_sysinfo(action, voice: str):
|
||||
topic = action.get('topic', 'all')
|
||||
|
||||
if topic in ('time', 'all'):
|
||||
now = datetime.datetime.now()
|
||||
weekday = ['понедельник', 'вторник', 'среда', 'четверг', 'пятница', 'суббота', 'воскресенье'][now.weekday()]
|
||||
_speak(f"Сейчас {now.strftime('%H:%M')}, {weekday}.")
|
||||
if topic == 'time':
|
||||
return
|
||||
|
||||
ps_scripts = {
|
||||
'battery': '$b=Get-CimInstance Win32_Battery -EA SilentlyContinue; if($b){"$([int]$b.EstimatedChargeRemaining)%"}else{"батареи нет"}',
|
||||
'cpu': '$c=Get-CimInstance Win32_Processor -EA SilentlyContinue|Select-Object -First 1; $load=(Get-CimInstance Win32_PerfFormattedData_PerfOS_Processor -Filter "Name=\'_Total\'" -EA SilentlyContinue).PercentProcessorTime; if($null -eq $load){$load=$c.LoadPercentage}; "$load%"',
|
||||
'ram': '$o=Get-CimInstance Win32_OperatingSystem; $tg=[math]::Round($o.TotalVisibleMemorySize/1MB,1); $ug=[math]::Round(($o.TotalVisibleMemorySize-$o.FreePhysicalMemory)/1MB,1); "$ug из $tg гигабайт"',
|
||||
'disk': '$d=Get-CimInstance Win32_LogicalDisk -Filter "DeviceID=\'$env:SystemDrive\'"; $fg=[math]::Round($d.FreeSpace/1GB,0); $tg=[math]::Round($d.Size/1GB,0); "свободно $fg из $tg гигабайт"',
|
||||
}
|
||||
labels = {'battery': 'Питание', 'cpu': 'Процессор', 'ram': 'Память', 'disk': 'Диск'}
|
||||
order = ['battery', 'cpu', 'ram', 'disk'] if topic == 'all' else [topic]
|
||||
|
||||
for t in order:
|
||||
ps = ps_scripts.get(t)
|
||||
if not ps:
|
||||
continue
|
||||
result = subprocess.run(
|
||||
['powershell', '-NoProfile', '-Command', ps],
|
||||
capture_output=True, text=True, encoding='utf-8', timeout=10,
|
||||
)
|
||||
out = result.stdout.strip() if result.returncode == 0 else 'ошибка'
|
||||
_speak(f"{labels[t]}: {out}.")
|
||||
|
||||
|
||||
def run_action(action):
|
||||
t = action['type']
|
||||
if t == 'exe':
|
||||
|
|
@ -269,14 +479,32 @@ def run_action(action):
|
|||
elif t == 'multi':
|
||||
for step in action['steps']:
|
||||
run_action(step)
|
||||
elif t == 'groq_codegen':
|
||||
do_codegen(action, _current_voice or '')
|
||||
elif t == 'ocr_screen':
|
||||
do_ocr(action, _current_voice or '')
|
||||
elif t == 'sysinfo':
|
||||
do_sysinfo(action, _current_voice or '')
|
||||
elif t == 'file_search':
|
||||
do_file_search(action, _current_voice or '')
|
||||
elif t == 'audio_devices_panel':
|
||||
do_audio_devices_panel(action, _current_voice or '')
|
||||
|
||||
|
||||
_current_voice: str = ''
|
||||
|
||||
|
||||
def execute_cmd(cmd: str, voice: str):
|
||||
global _current_voice
|
||||
spec = VA_CMD_LIST.get(cmd)
|
||||
if not spec:
|
||||
return
|
||||
action = spec['action']
|
||||
run_action(action)
|
||||
_current_voice = voice
|
||||
try:
|
||||
run_action(action)
|
||||
finally:
|
||||
_current_voice = ''
|
||||
confirm = spec.get('confirm_sound')
|
||||
if confirm is None and action['type'] == 'exe':
|
||||
confirm = 'ok'
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue