-
Notifications
You must be signed in to change notification settings - Fork 24
Expand file tree
/
Copy pathcraftbot.py
More file actions
1596 lines (1347 loc) · 56 KB
/
craftbot.py
File metadata and controls
1596 lines (1347 loc) · 56 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""
CraftBot Service Manager
Run CraftBot as a background process that survives terminal closure,
and optionally register it to auto-start when your system boots.
Commands:
python craftbot.py start [options] Start CraftBot in background
python craftbot.py stop Stop CraftBot
python craftbot.py restart [options] Stop then start
python craftbot.py status Show if CraftBot is running
python craftbot.py logs [-n N] Show last N log lines (default: 50)
python craftbot.py install [options] Register for auto-start on boot/login
python craftbot.py uninstall Remove auto-start registration
python craftbot.py repair [options] Re-copy current EXE over the installed
copy and restart (frozen EXE only)
python craftbot.py wizard Open the GUI wizard
When running as a frozen EXE (CraftBot.exe with no args), the wizard opens
automatically. Subcommands work the same as in source mode.
Options passed to 'start' / 'install':
--tui Run in TUI mode instead of browser
--cli Run in CLI mode
--no-open-browser Don't open browser automatically (default for service)
--frontend-port PORT Frontend port (default: 7925)
--backend-port PORT Backend port (default: 7926)
--conda Use conda environment
--no-conda Don't use conda
Examples:
python craftbot.py start # Start in background (browser mode)
python craftbot.py start --tui # Start in background (TUI mode)
python craftbot.py install # Auto-start on login (browser mode)
python craftbot.py install --no-open-browser # Auto-start without opening browser
python craftbot.py stop
python craftbot.py logs -n 100
"""
import sys
# In windowed (console=False) PyInstaller builds, sys.stdout and sys.stderr
# are None. Lots of libraries (including run.py at import time) call
# sys.stdout.isatty() unconditionally and crash with AttributeError. Install
# a dummy file-like before ANY other import that might transitively hit
# stdout. Must be the very first thing this module does.
if sys.stdout is None or sys.stderr is None:
import io
class _NullIO(io.TextIOBase):
def isatty(self) -> bool:
return False
def write(self, s: str) -> int:
return len(s)
def flush(self) -> None:
pass
if sys.stdout is None:
sys.stdout = _NullIO()
if sys.stderr is None:
sys.stderr = _NullIO()
import os
import shutil
import signal
import subprocess
import threading
import time
import webbrowser
from typing import Callable, List, Optional
from installer import helpers as _helpers
from installer import metadata as _metadata
from installer import payload as _payload
# Store platform once so static analysers don't short-circuit platform branches
_PLATFORM: str = sys.platform
# ─── Frozen-mode detection ────────────────────────────────────────────────────
# True when running as a PyInstaller-bundled EXE. The frozen EXE is the
# *installer* — a small Tkinter wizard that downloads the agent payload from
# GitHub Releases and installs it into a chosen location. The installer does
# NOT contain the agent itself (no run.py, no openai/anthropic/etc bundled).
IS_FROZEN: bool = bool(getattr(sys, "frozen", False))
EXE_PATH: Optional[str] = sys.executable if IS_FROZEN else None
# Agent payload (download/extract/version) lives in craftbot_payload.
# These re-exports keep external callers (e.g. CraftBotInstaller.spec docstring,
# any future tooling) working with the legacy `craftbot.GITHUB_OWNER` etc.
GITHUB_OWNER = _payload.GITHUB_OWNER
GITHUB_REPO = _payload.GITHUB_REPO
# ─── Paths ────────────────────────────────────────────────────────────────────
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
def _read_bundled_version() -> str:
return _payload.read_bundled_version(BASE_DIR)
def _user_data_dir() -> str:
"""Return the per-user persistent data directory for CraftBot.
Used for PID file, log file, and install metadata when running as a
frozen EXE (BASE_DIR points at the temp-extracted bundle, which is
wiped between runs). For source installs we keep the legacy behaviour
of writing alongside craftbot.py, since BASE_DIR is already stable.
"""
if not IS_FROZEN:
return BASE_DIR
if _PLATFORM == "win32":
root = os.environ.get("LOCALAPPDATA") or os.path.expanduser(r"~\AppData\Local")
path = os.path.join(root, "CraftBot")
elif _PLATFORM == "darwin":
path = os.path.expanduser("~/Library/Application Support/CraftBot")
else:
root = os.environ.get("XDG_DATA_HOME") or os.path.expanduser("~/.local/share")
path = os.path.join(root, "craftbot")
os.makedirs(path, exist_ok=True)
return path
PID_FILE = os.path.join(_user_data_dir(), "craftbot.pid")
LOG_FILE = os.path.join(_user_data_dir(), "craftbot.log")
INSTALL_METADATA_FILE = os.path.join(_user_data_dir(), "install.json")
# Source-mode only — relative to craftbot.py. The frozen installer never
# uses this; it spawns the installed agent EXE directly.
RUN_SCRIPT = os.path.join(BASE_DIR, "run.py")
def download_agent_zip(
progress_cb: Optional[Callable[[int, Optional[int]], None]] = None,
) -> str:
return _payload.download_agent_zip(BASE_DIR, EXE_PATH, progress_cb=progress_cb)
def extract_agent_zip(zip_path: str, target_dir: str) -> str:
return _payload.extract_agent_zip(zip_path, target_dir)
def default_install_location() -> str:
"""Return the default install directory used by the wizard's location chooser.
Picks a per-user path that does NOT require admin elevation. The wizard
lets the user override this via filedialog.askdirectory().
"""
if _PLATFORM == "win32":
root = os.environ.get("LOCALAPPDATA") or os.path.expanduser(r"~\AppData\Local")
return os.path.join(root, "Programs", "CraftBot")
if _PLATFORM == "darwin":
return os.path.expanduser("~/Applications/CraftBot")
return os.path.expanduser("~/.local/share/craftbot")
# Install metadata helpers — pure-function impls live in craftbot_metadata.
# We keep these no-arg wrappers so the wizard (and any external caller) can
# use the legacy `craftbot.installed_exe_path()` API without threading the
# metadata-file path through every call site.
def read_install_metadata() -> Optional[dict]:
return _metadata.read(INSTALL_METADATA_FILE)
def write_install_metadata(installed_path: str, mode: str) -> None:
_metadata.write(INSTALL_METADATA_FILE, installed_path, mode)
def clear_install_metadata() -> None:
_metadata.clear(INSTALL_METADATA_FILE)
def installed_exe_path() -> Optional[str]:
return _metadata.installed_exe_path(INSTALL_METADATA_FILE)
TASK_NAME = "CraftBot" # Windows Task Scheduler task name
SYSTEMD_SERVICE = "craftbot" # Linux systemd service name
LAUNCHD_LABEL = "com.craftbot.agent" # macOS launchd label
BROWSER_URL = "http://localhost:7925"
SHORTCUT_NAME = "CraftBot.lnk"
# Bundled icons live in sys._MEIPASS in frozen mode (PyInstaller's runtime
# extract dir) and alongside craftbot.py in source mode. _ensure_ico() copies
# the bundled icon to the persistent user data dir during install so the
# desktop shortcut keeps a stable path after _MEIPASS is wiped.
_BUNDLE_DIR = getattr(sys, "_MEIPASS", BASE_DIR)
LOGO_PNG = os.path.join(_BUNDLE_DIR, "craftbot_logo_1.png")
LOGO_ICO = os.path.join(_BUNDLE_DIR, "craftbot_logo_1.ico")
# Wordmark logo for the wizard header.
LOGO_TEXT_WHITE_PNG = os.path.join(
_BUNDLE_DIR, "assets", "craftbot_logo_text_no_border_dark.png"
)
# ─── Terminal colors (orange/white brand palette) ─────────────────────────────
def _enable_windows_vtp() -> None:
if sys.platform != "win32":
return
try:
import ctypes
k32 = ctypes.windll.kernel32
h = k32.GetStdHandle(-11)
m = ctypes.c_ulong()
k32.GetConsoleMode(h, ctypes.byref(m))
k32.SetConsoleMode(h, m.value | 0x0004)
except Exception:
pass
_enable_windows_vtp()
# Force UTF-8 output on Windows so Unicode box-drawing characters don't crash
if sys.platform == "win32":
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
_USE_COLOR = sys.stdout.isatty()
def _c(code: str) -> str:
return code if _USE_COLOR else ""
ORANGE = _c("\033[38;2;255;79;24m") # #FF4F18
WHITE = _c("\033[38;2;255;255;255m") # #FFFFFF
BOLD = _c("\033[1m")
DIM = _c("\033[38;2;80;80;80m") # dark gray
GREEN = _c("\033[38;2;80;220;100m")
RED = _c("\033[91m")
RESET = _c("\033[0m")
def _retro_step(num: int, total: int, desc: str) -> None:
"""Print a retro-style step header box."""
W = 62
visible = f" ▸ STEP {num}/{total} ░░ {desc.upper()}"
pad = max(0, W - len(visible))
content = (
f" {ORANGE}▸ STEP {num}/{total}{RESET}"
f" {DIM}░░{RESET}"
f" {WHITE}{desc.upper()}{RESET}" + " " * pad
)
print(f"\n{ORANGE}╔{'═' * W}╗{RESET}")
print(f"{ORANGE}║{RESET}{content}{ORANGE}║{RESET}")
print(f"{ORANGE}╚{'═' * W}╝{RESET}")
# ─── Helpers ──────────────────────────────────────────────────────────────────
def _to_short_path(path: str) -> str:
"""Return the Windows 8.3 short path form to avoid Unicode / long-path
issues when embedding paths inside command strings (e.g. schtasks /tr)."""
if _PLATFORM != "win32":
return path
try:
import ctypes
buf = ctypes.create_unicode_buffer(1024)
if ctypes.windll.kernel32.GetShortPathNameW(path, buf, len(buf)):
return buf.value
except Exception:
pass
return path
def _warn_path_issues() -> None:
"""Print warnings if BASE_DIR is too long or contains non-ASCII characters."""
if len(BASE_DIR) > 200:
print(f"WARNING: Installation path is very long ({len(BASE_DIR)} chars).")
print(" Windows MAX_PATH limit may cause failures.")
print(" Consider moving CraftBot to a shorter path.\n")
try:
BASE_DIR.encode("ascii")
except UnicodeEncodeError:
print(
"WARNING: Installation path contains non-ASCII characters (e.g. Japanese)."
)
print(
" Some commands may fail. Short paths will be used where possible.\n"
)
def _python_exe() -> str:
"""Return the Python executable to use for the service process."""
# On Windows prefer pythonw.exe (no console window) when not in TUI/CLI mode
if _PLATFORM == "win32":
pythonw = os.path.join(os.path.dirname(sys.executable), "pythonw.exe")
if os.path.isfile(pythonw):
return pythonw
return sys.executable
def _read_pid() -> Optional[int]:
"""Read PID from the PID file. Returns None if file missing or invalid."""
try:
with open(PID_FILE) as f:
return int(f.read().strip())
except (FileNotFoundError, ValueError):
return None
def _write_pid(pid: int) -> None:
with open(PID_FILE, "w") as f:
f.write(str(pid))
def _remove_pid() -> None:
try:
os.remove(PID_FILE)
except FileNotFoundError:
pass
def _is_running(pid: int) -> bool:
"""Return True if a process with the given PID is currently alive."""
if _PLATFORM == "win32":
try:
result = subprocess.run(
["tasklist", "/FI", f"PID eq {pid}", "/NH"],
capture_output=True,
text=True,
timeout=5,
)
return str(pid) in result.stdout
except Exception:
return False
else:
try:
os.kill(pid, 0)
return True
except (ProcessLookupError, PermissionError):
return False
def _stop_running_agent_if_alive(grace_s: float = 1.0) -> bool:
"""If an agent is currently running, stop it and pause briefly so the OS
releases file handles before we try to overwrite the agent EXE. Used by
install/repair so reinstall over a running agent doesn't fail with
"Permission denied" on Windows. Returns True if a process was stopped.
"""
pid = _read_pid()
if pid and _is_running(pid):
cmd_stop()
time.sleep(grace_s)
return True
return False
def _build_run_args(extra: List[str], service_mode: bool = True) -> List[str]:
"""Build the argument list for run.py.
Adds --no-open-browser by default in service mode (auto-start at boot
should not pop open a browser without the user asking).
"""
args = list(extra)
# TUI/CLI modes don't use the browser flag
if service_mode and "--tui" not in args and "--cli" not in args:
if "--no-open-browser" not in args:
args.append("--no-open-browser")
return args
# ─── Core operations ──────────────────────────────────────────────────────────
def _open_browser_when_ready(url: str, pid_check_fn, delay: float = 4.0) -> None:
"""Wait for the server to start, then open the browser."""
time.sleep(delay)
if not pid_check_fn():
print("\nWarning: CraftBot process exited before browser could open.")
print("Check logs: python craftbot.py logs")
return
webbrowser.open(url)
def _open_browser_detached(url: str) -> None:
"""Poll the server URL and open the browser once it responds.
In frozen mode (installer EXE) sys.executable is CraftBotInstaller.exe,
not python — so spawning `sys.executable -c <script>` would just re-launch
the wizard. We run the poll in-process on a daemon thread instead. In
source mode we keep the detached-subprocess path so `python craftbot.py
start` returns immediately even on slow agent boots.
"""
if IS_FROZEN:
def _poll_and_open() -> None:
from urllib.request import urlopen
deadline = time.time() + 120
while time.time() < deadline:
try:
urlopen(url, timeout=1).close()
break
except Exception:
time.sleep(1)
webbrowser.open(url)
threading.Thread(target=_poll_and_open, daemon=True).start()
return
poll_script = (
"import sys, time, webbrowser\n"
"try:\n"
" from urllib.request import urlopen\n"
" deadline = time.time() + 120\n"
" while time.time() < deadline:\n"
" try:\n"
f" urlopen('{url}', timeout=1).close()\n"
" break\n"
" except Exception:\n"
" time.sleep(1)\n"
"except Exception:\n"
" pass\n"
f"webbrowser.open('{url}')\n"
)
python = sys.executable
if _PLATFORM == "win32":
pythonw = python.replace("python.exe", "pythonw.exe")
if os.path.isfile(pythonw):
python = pythonw
kwargs: dict = dict(
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
**_helpers.detached_popen_flags(),
)
subprocess.Popen([python, "-c", poll_script], **kwargs)
def cmd_start(extra_args: List[str]) -> None:
"""Start CraftBot as a detached background process."""
pid = _read_pid()
if pid and _is_running(pid):
cmd_stop()
# service_mode=False — don't suppress the browser; we open it ourselves below
run_args = _build_run_args(extra_args, service_mode=False)
# Always pass --no-open-browser to run.py; craftbot.py handles opening the browser
if "--tui" not in run_args and "--cli" not in run_args:
if "--no-open-browser" not in run_args:
run_args.append("--no-open-browser")
if IS_FROZEN:
# In frozen-installer mode, spawn the installed agent EXE (downloaded
# by the install flow). The agent EXE is its own self-contained
# PyInstaller binary and runs run.py's __main__ block directly.
installed = installed_exe_path()
if not installed:
print("Error: no installed agent found — run install first.")
return
cmd = [installed] + run_args
else:
python = _python_exe()
# Use plain python.exe for TUI/CLI because pythonw has no console
if "--tui" in run_args or "--cli" in run_args:
python = sys.executable
cmd = [python, RUN_SCRIPT] + run_args
# UTF-8 with replace so the agent's Unicode banner / box-drawing chars
# don't crash on Windows where the default file encoding is cp1252.
log_fh = open(LOG_FILE, "a", encoding="utf-8", errors="replace")
log_fh.write(f"\n{'=' * 60}\n")
log_fh.write(f"CraftBot service started at {_timestamp()}\n")
log_fh.write(f"Command: {' '.join(cmd)}\n")
log_fh.write(f"{'=' * 60}\n")
log_fh.flush()
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
env["PYTHONUTF8"] = "1"
kwargs = dict(
cwd=BASE_DIR,
stdout=log_fh,
stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
env=env,
**_helpers.detached_popen_flags(new_process_group=True),
)
try:
proc = subprocess.Popen(cmd, **kwargs)
except FileNotFoundError as e:
log_fh.close()
print(f" {RED}✗{RESET} {WHITE}Could not launch CraftBot — {e}{RESET}")
return
# Parent closes its copy — the child process (run.py) keeps the fd open
log_fh.close()
_write_pid(proc.pid)
print(
f" {GREEN}▸{RESET} {WHITE}CRAFTBOT STARTED{RESET} {DIM}PID {proc.pid}{RESET}"
)
# Create a desktop shortcut so the user can reopen the browser anytime
if "--tui" not in run_args and "--cli" not in run_args:
if _PLATFORM == "win32":
_create_desktop_shortcut_windows()
else:
_create_desktop_shortcut_unix()
open_browser = (
"--tui" not in run_args
and "--cli" not in run_args
and "--no-open-browser" not in extra_args
)
if open_browser:
print(f" {DIM}░░{RESET} {ORANGE}{BROWSER_URL}{RESET}")
_open_browser_detached(BROWSER_URL)
def cmd_stop() -> None:
"""Stop the running CraftBot service."""
pid = _read_pid()
if pid is None:
print("CraftBot does not appear to be running (no PID file found).")
return
if not _is_running(pid):
print(f" {DIM}▸ PID {pid} not running — cleaning up stale PID file{RESET}")
_remove_pid()
return
print(f" {ORANGE}▸{RESET} {WHITE}STOPPING CRAFTBOT{RESET} {DIM}PID {pid}{RESET}")
if _PLATFORM == "win32":
try:
subprocess.run(
["taskkill", "/PID", str(pid), "/F", "/T"],
capture_output=True,
timeout=15,
)
except Exception as e:
print(f"Warning: taskkill failed — {e}")
else:
try:
# Kill the entire process group so child processes also die
pgid = os.getpgid(pid)
os.killpg(pgid, signal.SIGTERM)
# Give it a moment to exit gracefully
for _ in range(10):
time.sleep(0.5)
if not _is_running(pid):
break
else:
os.killpg(pgid, signal.SIGKILL)
except ProcessLookupError:
pass
except Exception as e:
print(f"Warning: {e}")
_remove_pid()
print(f" {GREEN}▸{RESET} {WHITE}CRAFTBOT STOPPED{RESET}")
def cmd_status() -> None:
"""Print whether CraftBot is currently running and whether auto-start is installed."""
W = 50
pid = _read_pid()
print(f"\n{ORANGE}╔{'═' * W}╗{RESET}")
if pid and _is_running(pid):
print(
f"{ORANGE}║{RESET} {GREEN}▸ RUNNING{RESET} {DIM}PID {pid}{RESET}{' ' * (W - 14 - len(str(pid)))}{ORANGE}║{RESET}"
)
print(
f"{ORANGE}║{RESET} {DIM}░░ LOG: {LOG_FILE[: W - 8]}{RESET}{' ' * max(0, W - 8 - len(LOG_FILE[: W - 8]))}{ORANGE}║{RESET}"
)
else:
if pid:
_remove_pid()
print(
f"{ORANGE}║{RESET} {RED}▸ NOT RUNNING{RESET}{' ' * (W - 14)}{ORANGE}║{RESET}"
)
print(f"{ORANGE}║{' ' * W}║{RESET}")
if _is_installed():
print(
f"{ORANGE}║{RESET} {GREEN}▸ AUTO-START: INSTALLED{RESET}{' ' * (W - 23)}{ORANGE}║{RESET}"
)
else:
print(
f"{ORANGE}║{RESET} {DIM}▸ AUTO-START: NOT INSTALLED{RESET}{' ' * (W - 27)}{ORANGE}║{RESET}"
)
print(f"{ORANGE}╚{'═' * W}╝{RESET}\n")
def cmd_logs(n: int = 50) -> None:
"""Print the last N lines of the CraftBot log."""
if not os.path.isfile(LOG_FILE):
print(f"No log file found at {LOG_FILE}")
return
try:
with open(LOG_FILE, "r", errors="replace") as f:
lines = f.readlines()
tail = lines[-n:] if len(lines) > n else lines
print(f"\n{ORANGE}╔{'═' * 60}╗{RESET}")
print(
f"{ORANGE}║{RESET} {WHITE}CRAFTBOT LOG{RESET} {DIM}last {len(tail)} lines{RESET}{' ' * (44 - len(str(len(tail))))}{ORANGE}║{RESET}"
)
print(f"{ORANGE}╚{'═' * 60}╝{RESET}\n")
print("".join(tail), end="")
except Exception as e:
print(f" {RED}✗{RESET} {WHITE}Error reading log: {e}{RESET}")
def cmd_restart(extra_args: List[str]) -> None:
cmd_stop()
time.sleep(1)
cmd_start(extra_args)
# ─── Desktop shortcut ─────────────────────────────────────────────────────────
def _find_desktop() -> Optional[str]:
"""Return the path to the user's Desktop folder.
Works for all users regardless of language, OneDrive config, or custom paths.
On Windows reads the Shell Folders registry key directly (no subprocess).
"""
if _PLATFORM == "win32":
# Method 1: Read the registry directly — fast, no subprocess
try:
import winreg
key = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Explorer\User Shell Folders",
0,
winreg.KEY_READ,
)
raw, _ = winreg.QueryValueEx(key, "Desktop")
winreg.CloseKey(key)
path = os.path.expandvars(raw)
if path and os.path.isdir(path):
return path
except Exception:
pass
# Method 2: ctypes SHGetFolderPath (CSIDL_DESKTOPDIRECTORY = 0x0010)
try:
import ctypes
buf = ctypes.create_unicode_buffer(260)
ctypes.windll.shell32.SHGetFolderPathW(None, 0x0010, None, 0, buf)
path = buf.value
if path and os.path.isdir(path):
return path
except Exception:
pass
# Fallback: check common paths
for candidate in [
os.path.join(os.path.expanduser("~"), "Desktop"),
os.path.join(os.path.expanduser("~"), "OneDrive", "Desktop"),
]:
if os.path.isdir(candidate):
return candidate
return None
def _ensure_ico() -> Optional[str]:
"""Return a path to a .ico file the desktop shortcut can reference long-term.
In frozen mode prefer the persistent copy in _user_data_dir() (the bundled
_MEIPASS path is wiped when the EXE exits, which would break the shortcut's
icon after the first run).
"""
persistent_ico = os.path.join(_user_data_dir(), "craftbot_logo_1.ico")
if os.path.isfile(persistent_ico):
return persistent_ico
if os.path.isfile(LOGO_ICO):
return LOGO_ICO
if not os.path.isfile(LOGO_PNG):
return None
try:
from PIL import Image
img = Image.open(LOGO_PNG)
# Write the converted .ico into the persistent dir so it survives
# process exit when running as a frozen EXE.
os.makedirs(os.path.dirname(persistent_ico), exist_ok=True)
img.save(
persistent_ico,
format="ICO",
sizes=[(256, 256), (48, 48), (32, 32), (16, 16)],
)
return persistent_ico
except Exception:
return None
def _create_desktop_shortcut_windows() -> None:
"""Create a .lnk shortcut on the Windows Desktop with the CraftBot icon."""
desktop = _find_desktop()
if not desktop:
return
shortcut_path = os.path.join(desktop, SHORTCUT_NAME)
if os.path.exists(shortcut_path):
return # already exists, don't recreate
ico_path = _ensure_ico()
try:
# Write the PS script to a temp file with UTF-8-BOM so PowerShell
# handles non-ASCII paths (e.g. Japanese Desktop folder) correctly.
import tempfile
ps_lines = [
"$ws = New-Object -ComObject WScript.Shell",
f'$s = $ws.CreateShortcut("{shortcut_path}")',
'$s.TargetPath = "cmd.exe"',
f'$s.Arguments = "/c start {BROWSER_URL}"',
"$s.WindowStyle = 7", # minimized (hides the cmd flash)
]
if ico_path:
ps_lines.append(f'$s.IconLocation = "{ico_path},0"')
ps_lines += [
'$s.Description = "Open CraftBot in your browser"',
"$s.Save()",
]
with tempfile.NamedTemporaryFile(
mode="w", suffix=".ps1", delete=False, encoding="utf-8-sig"
) as tf:
tf.write("\n".join(ps_lines))
tmp_ps1 = tf.name
try:
subprocess.run(
[
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
tmp_ps1,
],
capture_output=True,
timeout=15,
)
finally:
try:
os.remove(tmp_ps1)
except Exception:
pass
if os.path.exists(shortcut_path):
print(f" Desktop shortcut created: {shortcut_path}")
print(" Double-click it anytime to open CraftBot in your browser.")
else:
print(f" (Shortcut creation may have failed — check {desktop})")
except Exception as e:
print(f" (Could not create desktop shortcut: {e})")
def _create_desktop_shortcut_unix() -> None:
"""Create a desktop shortcut on Linux or macOS."""
desktop = _find_desktop()
if not desktop:
return
try:
if _PLATFORM == "darwin":
# macOS does not support XDG .desktop files — create a double-clickable .command script
shortcut_path = os.path.join(desktop, "CraftBot.command")
content = f"#!/bin/sh\nopen '{BROWSER_URL}'\n"
with open(shortcut_path, "w") as f:
f.write(content)
os.chmod(shortcut_path, 0o755)
else:
# Linux: standard XDG .desktop entry
shortcut_path = os.path.join(desktop, "CraftBot.desktop")
open_cmd = (
"xdg-open"
if os.path.isfile("/usr/bin/xdg-open")
or os.path.isfile("/usr/local/bin/xdg-open")
else "sensible-browser"
)
content = (
"[Desktop Entry]\n"
"Type=Application\n"
"Name=CraftBot\n"
f"Exec={open_cmd} {BROWSER_URL}\n"
"Icon=web-browser\n"
"Terminal=false\n"
)
with open(shortcut_path, "w") as f:
f.write(content)
os.chmod(shortcut_path, 0o755)
print(f" Desktop shortcut created: {shortcut_path}")
print(" Double-click it anytime to open CraftBot in your browser.")
except Exception as e:
print(f" (Could not create desktop shortcut: {e})")
# ─── Auto-start: Windows Task Scheduler ───────────────────────────────────────
def _install_windows_registry(action: str) -> bool:
"""Fallback: register auto-start via HKCU Run registry key (no admin needed)."""
try:
import winreg
key = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Run",
0,
winreg.KEY_SET_VALUE,
)
winreg.SetValueEx(key, TASK_NAME, 0, winreg.REG_SZ, action)
winreg.CloseKey(key)
return True
except Exception:
return False
def _install_windows(run_args: List[str]) -> None:
if IS_FROZEN:
# Frozen mode: register the extracted agent EXE for auto-start.
target = installed_exe_path()
if not target:
print(
f" {RED}✗{RESET} {WHITE}No installed agent found — run install first.{RESET}"
)
return
target_s = _to_short_path(target)
action = f'"{target_s}" {" ".join(run_args)}'.strip()
else:
python = _python_exe()
# Use 8.3 short paths to avoid Unicode/long-path failures in schtasks /tr
python_s = _to_short_path(python)
script_s = _to_short_path(RUN_SCRIPT)
action = f'"{python_s}" "{script_s}" {" ".join(run_args)}'
# Try Task Scheduler first; silently fall back to Registry on failure
registered = False
try:
result = subprocess.run(
[
"schtasks",
"/create",
"/tn",
TASK_NAME,
"/tr",
action,
"/sc",
"ONLOGON",
"/f",
],
capture_output=True,
text=True,
timeout=30,
)
registered = result.returncode == 0
except Exception:
pass
if not registered:
registered = _install_windows_registry(action)
if registered:
print("Auto-start registered. CraftBot will start automatically on login.")
print(f"Open CraftBot: {BROWSER_URL}")
_create_desktop_shortcut_windows()
else:
print(
"Could not register auto-start. Use 'python craftbot.py start' to start manually."
)
def _uninstall_windows() -> None:
removed_any = False
# Remove from Task Scheduler
try:
result = subprocess.run(
["schtasks", "/delete", "/tn", TASK_NAME, "/f"],
capture_output=True,
text=True,
timeout=15,
)
if result.returncode == 0:
print(f"Auto-start removed (task '{TASK_NAME}' deleted).")
removed_any = True
except Exception as e:
print(f"Warning: Could not query Task Scheduler — {e}")
# Remove from Registry (HKCU\...\Run) — the fallback auto-start method
try:
import winreg
key = winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Run",
0,
winreg.KEY_SET_VALUE,
)
try:
winreg.DeleteValue(key, TASK_NAME)
print(f"Auto-start removed (registry entry '{TASK_NAME}' deleted).")
removed_any = True
except FileNotFoundError:
pass # Entry didn't exist in registry — that's fine
finally:
winreg.CloseKey(key)
except Exception as e:
print(f"Warning: Could not clean registry — {e}")
if not removed_any:
print("No auto-start registration found (already uninstalled?).")
# ─── Auto-start: Linux systemd (user service) ─────────────────────────────────
def _install_linux(run_args: List[str]) -> None:
service_dir = os.path.expanduser("~/.config/systemd/user")
os.makedirs(service_dir, exist_ok=True)
service_file = os.path.join(service_dir, f"{SYSTEMD_SERVICE}.service")
if IS_FROZEN:
target = installed_exe_path()
if not target:
print("Error: no installed agent found — run install first.")
return
exec_start = f"{target} {' '.join(run_args)}".strip()
else:
python = sys.executable
exec_start = f"{python} {RUN_SCRIPT} {' '.join(run_args)}"
content = f"""[Unit]
Description=CraftBot AI Agent
After=network.target
[Service]
Type=simple
ExecStart={exec_start}
WorkingDirectory={BASE_DIR}
Restart=on-failure
RestartSec=5
Environment=PYTHONUNBUFFERED=1
[Install]
WantedBy=default.target
"""
with open(service_file, "w") as f:
f.write(content)
try:
subprocess.run(["systemctl", "--user", "daemon-reload"], check=True, timeout=10)
subprocess.run(
["systemctl", "--user", "enable", SYSTEMD_SERVICE], check=True, timeout=10
)
print(f"Auto-start registered as systemd user service '{SYSTEMD_SERVICE}'.")
print("CraftBot will start automatically when you log in.")
print(f"\nOpen CraftBot: {BROWSER_URL}")
print(f" Tip: Bookmark {BROWSER_URL} so you never have to remember it!")
_create_desktop_shortcut_unix()
print(f"\nTo start it now: systemctl --user start {SYSTEMD_SERVICE}")
print(f"To view logs: journalctl --user -u {SYSTEMD_SERVICE} -f")
except subprocess.CalledProcessError as e:
print(f"Error enabling systemd service: {e}")
print(f"Service file written to: {service_file}")
print(
"Try manually: systemctl --user daemon-reload && systemctl --user enable craftbot"
)
except FileNotFoundError:
print("systemctl not found. Is systemd running on this system?")
def _uninstall_linux() -> None:
service_file = os.path.expanduser(
f"~/.config/systemd/user/{SYSTEMD_SERVICE}.service"
)
try:
subprocess.run(
["systemctl", "--user", "disable", SYSTEMD_SERVICE],
capture_output=True,
timeout=10,
)
subprocess.run(
["systemctl", "--user", "stop", SYSTEMD_SERVICE],
capture_output=True,
timeout=10,
)
except Exception:
pass
if os.path.isfile(service_file):
os.remove(service_file)
print("Auto-start removed (service file deleted).")
else:
print("No systemd service file found.")