| |
| """ |
| Modding agent for Ollama. |
| |
| Gives a local model the tool loop the modding skill assumes: web search/fetch, |
| downloads, archive handling, file read/write inside allowed folders, log grep, |
| and loadable skill modules. The model does the thinking; this script does the |
| doing, and asks you before touching anything outside the workspace. |
| |
| Usage: |
| python modding_agent.py --preset qwen38 --allow "C:/Games" |
| (any game: tell it the game with /game NAME and any safe folders with /protect FOLDER, or use --game / --protect) |
| """ |
| import argparse |
| import datetime |
| import fnmatch |
| import hashlib |
| import ipaddress |
| import html |
| import json |
| import os |
| import re |
| import shutil |
| import socket |
| import sys |
| import tarfile |
| import time |
| import xml.etree.ElementTree as ET |
| import zipfile |
| from pathlib import Path |
| from urllib.parse import urlparse, urlencode, parse_qsl, urlunparse |
|
|
| import requests |
|
|
| try: |
| from ddgs import DDGS |
| except ImportError: |
| try: |
| from duckduckgo_search import DDGS |
| except ImportError: |
| DDGS = None |
| try: |
| from bs4 import BeautifulSoup |
| except ImportError: |
| BeautifulSoup = None |
|
|
| HERE = Path(__file__).resolve().parent |
| SKILL_DIR = HERE / "skill" |
| MODULE_DIR = SKILL_DIR / "modules" |
| MAX_OUT = 12000 |
| LEAN = False |
| LEAN_SKIP = {"browser_fetch", "list_archive", "save_module", "list_modules"} |
| LEAN_DROP_PARAMS = {"texture_batch": {"tint_strength", "grayscale", "invert", "sharpen", "blur", "resample", "channel_gain", "include_maps", "exclude", "pattern"}, |
| "texture_info": {"max_files"}, "deps_check": {"granularity"}, "read_file": {"start"}, "grep_file": {"max_matches", "context"}} |
| LEAN_DESC = { |
| "web_search": "Search the web (use before stating versions or links).", |
| "web_fetch": "Read a web page or JSON as text.", |
| "download_file": "Download a release file into the workspace.", |
| "extract_archive": "Unzip into workspace staging; shows top-level folders.", |
| "copy_tree": "Deploy staged folder; never overwrites a differing installed dependency (reports conflicts). exclude/replace = user decisions.", |
| "deps_check": "Read-only: compare a staged mod with installed folders; shows dependency conflicts.", |
| "tracker_dep": "Record a dependency: version, folder, provided_by, hard_for, soft_for.", |
| "list_dir": "List a folder.", |
| "read_file": "Read lines of a file (tail=true for the end).", |
| "grep_file": "Regex-search a file (logs).", |
| "write_file": "Write a text file.", |
| "tracker_stage": "Set stage: intake, research, download, extract, deploy, scan, debug, handoff.", |
| "tracker_mod": "Update a mod's status, version, source, deps, bundles (bundled dependency + hard/soft), notes.", |
| "tracker_info": "Set game, target, mod_dir, method.", |
| "tracker_note": "Record kind: decision, protected, issue (severity), resolved, conflict, conflict_resolved (resolution).", |
| "find_games": "Find games in the allowed folders.", |
| "texture_info": "Inspect loose textures: sizes, formats, data maps.", |
| "texture_batch": "Mass or fine-tuned texture edits with per-file rules. Dry run unless dry_run=false. Details: load_module texture-editing.", |
| "texture_undo": "Undo a texture job by id.", |
| "load_module": "Load a game/task module; empty name lists them.", |
| } |
| MAX_DOWNLOAD = int(os.environ.get("MODDING_MAX_DOWNLOAD_MB", "2048")) * 1024 * 1024 |
| UA = "Mozilla/5.0 (modding-agent)" |
|
|
|
|
| def ollama_url(): |
| host = os.environ.get("OLLAMA_HOST", "http://localhost:11434").rstrip("/") |
| return host if host.startswith("http") else "http://" + host |
|
|
|
|
| |
| |
| |
| class Guard: |
| def __init__(self, workspace, allow, protect, auto_yes): |
| self.workspace = Path(workspace).resolve() |
| self.workspace.mkdir(parents=True, exist_ok=True) |
| self.allow = [Path(a).resolve() for a in allow] |
| self.auto_yes = auto_yes |
| self.protect = [] |
| self._names, self._paths = set(), [] |
| for x in protect: |
| self.add_protect(x) |
|
|
| def add_protect(self, x): |
| """Protect a folder NAME (anywhere in a path) or a full folder PATH.""" |
| x = str(x).strip() |
| if not x or x in self.protect: |
| return False |
| if any(c in x for c in "/\\:"): |
| self._paths.append(Path(os.path.expandvars(os.path.expanduser(x))).resolve()) |
| else: |
| self._names.add(x.lower()) |
| self.protect.append(x) |
| return True |
|
|
| def remove_protect(self, x): |
| x = str(x).strip() |
| if x not in self.protect: |
| return False |
| self.protect.remove(x) |
| if any(c in x for c in "/\\:"): |
| tgt = Path(os.path.expandvars(os.path.expanduser(x))).resolve() |
| self._paths = [q for q in self._paths if q != tgt] |
| else: |
| self._names.discard(x.lower()) |
| return True |
|
|
| @staticmethod |
| def _inside(path, root): |
| try: |
| path.relative_to(root) |
| return True |
| except ValueError: |
| return False |
|
|
| def resolve(self, p): |
| path = Path(p) |
| if not path.is_absolute(): |
| path = self.workspace / path |
| return path.resolve() |
|
|
| def can_read(self, path): |
| return self._inside(path, self.workspace) or any(self._inside(path, r) for r in self.allow) |
|
|
| def is_protected(self, path): |
| parts = [x.lower() for x in path.parts] |
| return any(n in parts for n in self._names) or any(self._inside(path, q) for q in self._paths) |
|
|
| def confirm_write(self, path, what): |
| """Return None if allowed, or an error string.""" |
| if self._inside(path, self.workspace): |
| if self.is_protected(path): |
| pass |
| else: |
| return None |
| elif not any(self._inside(path, r) for r in self.allow): |
| return f"Blocked: {path} is outside the workspace and the allowed folders." |
| protected = self.is_protected(path) |
| if self.auto_yes and not protected: |
| return None |
| tag = "PROTECTED FOLDER - " if protected else "" |
| ans = input(f"\n[confirm] {tag}{what}\n {path}\n allow? [y/N] ").strip().lower() |
| return None if ans == "y" else "The user declined this action. Ask them what to do instead." |
|
|
|
|
| G = None |
|
|
|
|
| def clip(text, limit=None): |
| limit = limit or MAX_OUT |
| if len(text) <= limit: |
| return text |
| return text[:limit] + f"\n...[truncated, {len(text) - limit} more characters]" |
|
|
|
|
| |
| |
| |
| def _public_host(url): |
| u = urlparse(url) |
| if u.scheme not in ("http", "https") or not u.hostname: |
| return "Only http(s) URLs are allowed." |
| try: |
| for info in socket.getaddrinfo(u.hostname, None): |
| ip = ipaddress.ip_address(info[4][0]) |
| if ip.is_private or ip.is_loopback or ip.is_link_local: |
| return "Blocked: that host resolves to a private/local address." |
| except socket.gaierror: |
| return "Could not resolve that host." |
| return None |
|
|
|
|
| def _html_to_text(html): |
| if BeautifulSoup is None: |
| return re.sub(r"<[^>]+>", " ", html) |
| soup = BeautifulSoup(html, "html.parser") |
| for t in soup(["script", "style", "noscript", "svg"]): |
| t.decompose() |
| links = [] |
| for a in soup.find_all("a", href=True)[:60]: |
| label = a.get_text(" ", strip=True)[:60] |
| if label: |
| links.append(f"- {label}: {a['href']}") |
| text = re.sub(r"\n\s*\n+", "\n\n", soup.get_text("\n", strip=True)) |
| return text + ("\n\nLINKS:\n" + "\n".join(links) if links else "") |
|
|
|
|
| |
| |
| |
| def web_search(query, max_results=6): |
| if DDGS is None: |
| return "web_search unavailable: pip install ddgs" |
| try: |
| rows = list(DDGS().text(query, max_results=int(max_results))) |
| except Exception as e: |
| return f"Search failed: {e}" |
| if not rows: |
| return "No results. Try different, shorter keywords." |
| return "\n\n".join(f"{r.get('title')}\n{r.get('href')}\n{r.get('body')}" for r in rows) |
|
|
|
|
| def web_fetch(url, max_chars=None): |
| err = _public_host(url) |
| if err: |
| return err |
| try: |
| r = requests.get(url, headers={"User-Agent": UA}, timeout=30) |
| except Exception as e: |
| return f"Fetch failed: {e}" |
| ctype = r.headers.get("content-type", "") |
| if r.status_code >= 400: |
| why = r.headers.get("x-deny-reason") |
| extra = f" Network policy says: {why} (a proxy/firewall is blocking this host)." if why else \ |
| " (403 often means the site blocks scripts: try browser_fetch or another source.)" |
| return f"HTTP {r.status_code} for {url}.{extra}" |
| if "html" in ctype: |
| body = _html_to_text(r.text) |
| elif "json" in ctype or "text" in ctype: |
| body = r.text |
| else: |
| return f"Binary content ({ctype}, {len(r.content)} bytes). Use download_file instead." |
| return clip(body, int(max_chars) if max_chars else None) |
|
|
|
|
| def browser_fetch(url, js=None, wait_ms=3000): |
| """Render a JavaScript page with Playwright (optional dependency).""" |
| err = _public_host(url) |
| if err: |
| return err |
| try: |
| from playwright.sync_api import sync_playwright |
| except ImportError: |
| return "browser_fetch unavailable: pip install playwright && playwright install chromium" |
| try: |
| with sync_playwright() as p: |
| b = p.chromium.launch() |
| page = b.new_page(user_agent=UA) |
| page.goto(url, timeout=45000) |
| page.wait_for_timeout(int(wait_ms)) |
| if js: |
| out = page.evaluate(js) |
| out = out if isinstance(out, str) else json.dumps(out) |
| else: |
| out = _html_to_text(page.content()) |
| b.close() |
| return clip(out) |
| except Exception as e: |
| return f"browser_fetch failed: {e}" |
|
|
|
|
| def _with_factorio_auth(url): |
| u = urlparse(url) |
| user, token = os.environ.get("FACTORIO_USERNAME"), os.environ.get("FACTORIO_TOKEN") |
| if u.hostname == "mods.factorio.com" and u.path.startswith("/download") and user and token: |
| q = dict(parse_qsl(u.query)) |
| q.update(username=user, token=token) |
| return urlunparse(u._replace(query=urlencode(q))) |
| return url |
|
|
|
|
| def download_file(url, dest): |
| low = url.lower() |
| if re.search(r"github\.com/[^/]+/[^/]+/(archive|zipball|tarball)", low) or "zipball" in low or "tarball" in low: |
| return "Blocked: that is a source-code archive. Use a release asset instead." |
| err = _public_host(url) |
| if err: |
| return err |
| path = G.resolve(dest if Path(dest).is_absolute() else Path("downloads") / dest) |
| deny = G.confirm_write(path, f"download {url}") |
| if deny: |
| return deny |
| path.parent.mkdir(parents=True, exist_ok=True) |
| real_url = _with_factorio_auth(url) |
| try: |
| with requests.get(real_url, headers={"User-Agent": UA}, stream=True, timeout=60) as r: |
| if r.status_code >= 400: |
| return f"HTTP {r.status_code}. Needs login, a different URL, or the user must download it manually." |
| ctype = r.headers.get("content-type", "") |
| if "html" in ctype and path.suffix.lower() not in (".html", ".htm"): |
| return "That URL returned a web page, not a file (login wall or redirect). Find the real download link." |
| size = 0 |
| h = hashlib.sha256() |
| with open(path, "wb") as f: |
| for chunk in r.iter_content(1 << 20): |
| size += len(chunk) |
| if size > MAX_DOWNLOAD: |
| f.close() |
| path.unlink(missing_ok=True) |
| return "Aborted: file larger than the download limit." |
| f.write(chunk) |
| h.update(chunk) |
| except Exception as e: |
| return f"Download failed: {e}" |
| note = "" |
| if path.suffix.lower() == ".zip" and not zipfile.is_zipfile(path): |
| note = " WARNING: not a valid zip file." |
| return f"Saved {path} ({size} bytes, sha256 {h.hexdigest()[:16]}).{note}" |
|
|
|
|
| def list_dir(path, depth=2): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| if not p.exists(): |
| return f"Not found: {p}" |
| lines = [] |
|
|
| def walk(d, level): |
| try: |
| entries = sorted(d.iterdir(), key=lambda x: (x.is_file(), x.name.lower())) |
| except PermissionError: |
| return |
| for e in entries: |
| lines.append(" " * level + e.name + ("/" if e.is_dir() else f" ({e.stat().st_size}b)")) |
| if e.is_dir() and level + 1 < int(depth) and len(lines) < 400: |
| walk(e, level + 1) |
|
|
| walk(p, 0) |
| return clip("\n".join(lines) or "(empty)") |
|
|
|
|
| def read_file(path, start=1, lines=200, tail=False): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| if not p.is_file(): |
| return f"Not a file: {p}" |
| with open(p, "r", encoding="utf-8", errors="replace") as f: |
| all_lines = f.read().splitlines() |
| n = int(lines) |
| chunk = all_lines[-n:] if tail else all_lines[int(start) - 1:int(start) - 1 + n] |
| first = len(all_lines) - len(chunk) + 1 if tail else int(start) |
| body = "\n".join(f"{first + i}: {l}" for i, l in enumerate(chunk)) |
| return clip(f"[{p.name}: {len(all_lines)} lines total]\n{body}") |
|
|
|
|
| def grep_file(path, pattern, max_matches=60, context=0): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| if not p.is_file(): |
| return f"Not a file: {p}" |
| try: |
| rx = re.compile(pattern, re.I) |
| except re.error as e: |
| return f"Bad regex: {e}" |
| with open(p, "r", encoding="utf-8", errors="replace") as f: |
| data = f.read().splitlines() |
| out, c = [], int(context) |
| for i, line in enumerate(data): |
| if rx.search(line): |
| for j in range(max(0, i - c), min(len(data), i + c + 1)): |
| out.append(f"{j + 1}: {data[j][:300]}") |
| if c: |
| out.append("--") |
| if len(out) >= int(max_matches) * (1 + 2 * c): |
| break |
| return clip("\n".join(out) or "No matches.") |
|
|
|
|
| def write_file(path, content, append=False): |
| p = G.resolve(path) |
| deny = G.confirm_write(p, "write file") |
| if deny: |
| return deny |
| p.parent.mkdir(parents=True, exist_ok=True) |
| with open(p, "a" if append else "w", encoding="utf-8") as f: |
| f.write(content) |
| return f"Wrote {len(content)} characters to {p}" |
|
|
|
|
| def list_archive(path): |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| try: |
| if zipfile.is_zipfile(p): |
| with zipfile.ZipFile(p) as z: |
| names = z.namelist() |
| elif tarfile.is_tarfile(p): |
| with tarfile.open(p) as t: |
| names = t.getnames() |
| else: |
| return "Not a zip/tar archive (7z/rar need to be extracted by the user or with py7zr)." |
| except Exception as e: |
| return f"Could not read archive: {e}" |
| tops = sorted({n.split("/")[0] for n in names}) |
| return clip(f"{len(names)} entries. Top-level: {tops}\n" + "\n".join(names[:200])) |
|
|
|
|
| def extract_archive(path, dest): |
| src = G.resolve(path) |
| out = G.resolve(dest if Path(dest).is_absolute() else Path("staging") / dest) |
| if not G.can_read(src): |
| return f"Blocked: {src} is outside the workspace and allowed folders." |
| deny = G.confirm_write(out, "extract archive") |
| if deny: |
| return deny |
| out.mkdir(parents=True, exist_ok=True) |
| try: |
| if zipfile.is_zipfile(src): |
| with zipfile.ZipFile(src) as z: |
| for m in z.namelist(): |
| if not Guard._inside((out / m).resolve(), out): |
| return f"Blocked unsafe path in archive: {m}" |
| z.extractall(out) |
| elif tarfile.is_tarfile(src): |
| with tarfile.open(src) as t: |
| for m in t.getmembers(): |
| if not Guard._inside((out / m.name).resolve(), out): |
| return f"Blocked unsafe path in archive: {m.name}" |
| t.extractall(out) |
| else: |
| return "Not a zip/tar archive." |
| except Exception as e: |
| return f"Extract failed: {e}" |
| tops = sorted(x.name + ("/" if x.is_dir() else "") for x in out.iterdir())[:30] |
| return f"Extracted to {out}. Top-level: {tops}. Find the real root folder (avoid nested duplicates) before copying." |
|
|
|
|
| |
| |
| |
| |
| def _fingerprint(p): |
| """Content hash: full for files up to 64 MB, sampled (size + first/last 4 MB) above that.""" |
| h = hashlib.sha256() |
| size = p.stat().st_size |
| h.update(str(size).encode()) |
| with open(p, "rb") as f: |
| if size <= 64 * 1024 * 1024: |
| for chunk in iter(lambda: f.read(1 << 20), b""): |
| h.update(chunk) |
| else: |
| h.update(f.read(4 << 20)) |
| f.seek(max(0, size - (4 << 20))) |
| h.update(f.read(4 << 20)) |
| return h.hexdigest() |
|
|
|
|
| def _tree(root): |
| return {f.relative_to(root).as_posix(): f for f in root.rglob("*") if f.is_file()} |
|
|
|
|
| def _same_file(a, b): |
| return a.stat().st_size == b.stat().st_size and _fingerprint(a) == _fingerprint(b) |
|
|
|
|
| def _unit_version(p): |
| """Best-effort version of a mod folder: KSP-AVC .version file, info.json or fabric.mod.json. Empty if unknown.""" |
| try: |
| if p.is_file(): |
| return "" |
| cands = list(p.glob("*.version")) + list(p.glob("*/*.version")) + [x for x in (p / "info.json", p / "fabric.mod.json") if x.exists()] |
| for c in cands[:5]: |
| data = json.loads(c.read_text(encoding="utf-8-sig", errors="replace")) |
| v = data.get("VERSION") if isinstance(data, dict) else None |
| if isinstance(v, dict): |
| return ".".join(str(v.get(k, 0)) for k in ("MAJOR", "MINOR", "PATCH", "BUILD") if k in v) |
| if isinstance(v, str): |
| return v |
| if isinstance(data, dict) and isinstance(data.get("version"), str): |
| return data["version"] |
| except Exception: |
| pass |
| return "" |
|
|
|
|
| def _compare_units(a, b): |
| """Compare an incoming unit (a) with the installed one (b).""" |
| out = {"identical": False, "changed": 0, "added": 0, "removed": 0, "src_ver": _unit_version(a), "dst_ver": _unit_version(b)} |
| if a.is_file() and b.is_file(): |
| out["identical"] = _same_file(a, b) |
| out["changed"] = 0 if out["identical"] else 1 |
| return out |
| if a.is_dir() != b.is_dir(): |
| out["changed"] = 1 |
| return out |
| ta, tb = _tree(a), _tree(b) |
| out["added"] = len(set(ta) - set(tb)) |
| out["removed"] = len(set(tb) - set(ta)) |
| out["changed"] = sum(1 for k in set(ta) & set(tb) if not _same_file(ta[k], tb[k])) |
| out["identical"] = not (out["added"] or out["removed"] or out["changed"]) |
| return out |
|
|
|
|
| def _scan_units(s, d, granularity="folder"): |
| new, same, conflicts = [], [], [] |
| if granularity == "file": |
| groups = {} |
| for rel, f in _tree(s).items(): |
| t = d / rel |
| if not t.exists(): |
| new.append(rel) |
| elif _same_file(f, t): |
| same.append(rel) |
| else: |
| groups.setdefault(rel.split("/")[0], []).append(rel) |
| for k, files in sorted(groups.items()): |
| conflicts.append({"unit": k, "files": files, "changed": len(files)}) |
| return new, same, conflicts |
| for child in sorted(s.iterdir()): |
| t = d / child.name |
| if not t.exists(): |
| new.append(child.name) |
| continue |
| cmp = _compare_units(child, t) |
| (same if cmp["identical"] else conflicts).append(child.name if cmp["identical"] else {"unit": child.name, **cmp}) |
| return new, same, conflicts |
|
|
|
|
| def _fmt_conflict(c): |
| if "files" in c: |
| return f"{c['unit']}: {c['changed']} file(s) differ from installed (e.g. {', '.join(c['files'][:3])})" |
| return (f"{c['unit']}: installed {c['dst_ver'] or 'version unknown'} vs incoming {c['src_ver'] or 'version unknown'} " |
| f"({c['changed']} files differ, {c['added']} only in incoming, {c['removed']} only installed)") |
|
|
|
|
| def deps_check(src, dest, granularity="folder"): |
| """Read-only: compare a staged mod's folders with what is already installed. Never writes.""" |
| s, d = G.resolve(src), G.resolve(dest) |
| if not (G.can_read(s) and G.can_read(d)): |
| return "Blocked: both folders must be inside the workspace or the allowed folders." |
| if not s.is_dir() or not d.is_dir(): |
| return f"Both must be folders (staged: {s}, installed: {d})." |
| new, same, conflicts = _scan_units(s, d, granularity) |
| lines = [f"Checked {s} against {d} ({granularity} granularity).", |
| f"New, no clash ({len(new)}): {', '.join(new[:15])}{' ...' if len(new) > 15 else ''}", |
| f"Identical to what is installed, will be skipped ({len(same)}): {', '.join(same[:15])}{' ...' if len(same) > 15 else ''}", |
| f"DIFFERENT from what is installed ({len(conflicts)}):"] |
| lines += [f" - {_fmt_conflict(c)}" for c in conflicts[:30]] |
| if conflicts: |
| lines.append("These are potential dependency-chain breaks. For each: find out whether the incoming mod HARD-requires its bundled version " |
| "(exact) or is SOFT (works with the installed one), and whether an installed mod hard-requires the installed version. " |
| "Record with tracker_dep and tracker_note kind=conflict, then ask the user ONCE for all of them (see the dependency-chain module).") |
| return clip("\n".join(lines)) |
|
|
|
|
| def _copy_unit(src_path, target): |
| copied = skipped = 0 |
| if src_path.is_file(): |
| if G.is_protected(target): |
| return 0, 1 |
| target.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(src_path, target) |
| return 1, 0 |
| for f in src_path.rglob("*"): |
| if f.is_file(): |
| t = target / f.relative_to(src_path) |
| if G.is_protected(t): |
| skipped += 1 |
| continue |
| t.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(f, t) |
| copied += 1 |
| return copied, skipped |
|
|
|
|
| def copy_tree(src, dest, exclude="", replace="", granularity="folder"): |
| """Deploy a staged folder into a game folder WITHOUT breaking the dependency chain. |
| New folders are copied; identical ones are skipped; anything that differs from what is installed is a conflict: |
| it is skipped and reported unless the user decided (exclude = leave installed, replace = swap whole folder, with a backup).""" |
| s, d = G.resolve(src), G.resolve(dest) |
| if not G.can_read(s): |
| return f"Blocked: {s} is outside the workspace and allowed folders." |
| if not s.is_dir(): |
| return f"Not a folder: {s}" |
| ex = {x.strip().lower() for x in str(exclude).split(",") if x.strip()} |
| rep = {x.strip().lower() for x in str(replace).split(",") if x.strip()} |
| new, same, conflicts = _scan_units(s, d if d.exists() else s.parent / "__none__", granularity) if d.exists() else ( |
| [c.name for c in sorted(s.iterdir())] if granularity != "file" else list(_tree(s)), [], []) |
| deny = G.confirm_write(d, f"copy {len(new)} new item(s) from {s} into (differing items are NOT overwritten unless you chose replace)") |
| if deny: |
| return deny |
| d.mkdir(parents=True, exist_ok=True) |
| copied = prot = 0 |
| for u in new: |
| c, p_ = _copy_unit(s / u, d / u) |
| copied += c |
| prot += p_ |
| kept, replaced, open_conf = [], [], [] |
| stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") |
| for c in conflicts: |
| u = c["unit"] |
| if u.lower() in ex: |
| kept.append(u) |
| elif u.lower() in rep: |
| bdir = G.workspace / "dep_backups" / stamp |
| if granularity == "file": |
| files = c["files"] |
| for rel in files: |
| (bdir / rel).parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(d / rel, bdir / rel) |
| if not G.is_protected(d / rel): |
| shutil.copy2(s / rel, d / rel) |
| else: |
| bdir.mkdir(parents=True, exist_ok=True) |
| if G.is_protected(d / u): |
| open_conf.append(c) |
| continue |
| shutil.move(str(d / u), str(bdir / u)) |
| cc, pp = _copy_unit(s / u, d / u) |
| copied += cc |
| replaced.append(u) |
| else: |
| open_conf.append(c) |
| parts = [f"Copied {copied} files from {len(new)} new item(s); {len(same)} identical item(s) skipped" |
| + (f"; {prot} skipped (protected)" if prot else "") + "."] |
| if kept: |
| parts.append(f"Left installed version in place (your decision) for: {', '.join(kept)}.") |
| if replaced: |
| parts.append(f"Replaced as decided: {', '.join(replaced)}. Previous versions backed up in {G.workspace / 'dep_backups' / stamp}.") |
| if open_conf: |
| parts.append(f"{len(open_conf)} OPEN CONFLICT(S), nothing was overwritten:") |
| parts += [f" - {_fmt_conflict(c)}" for c in open_conf[:30]] |
| parts.append("Record each (tracker_dep, tracker_note kind=conflict), decide hard vs soft, then ask the user ONCE for all of them. " |
| "Re-run copy_tree with exclude=... (keep installed) or replace=... (swap, only if the user agrees).") |
| return clip("\n".join(parts)) |
|
|
|
|
| |
| |
| |
| STAGES = ["intake", "research", "download", "extract", "deploy", "scan", "debug", "handoff"] |
| STATUSES = ["todo", "researched", "downloaded", "extracted", "deployed", "excluded", "failed"] |
| COLORS = {"todo": "#8a8f98", "researched": "#4f8cff", "downloaded": "#a06bff", "extracted": "#d19a2a", |
| "deployed": "#2fa866", "excluded": "#8a8f98", "failed": "#e5484d"} |
|
|
|
|
| def _now(): |
| return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
|
|
|
|
| class Tracker: |
| def __init__(self, workspace): |
| self.xml_path = workspace / "tracker.xml" |
| self.html_path = workspace / "tracker.html" |
| self.d = {"game": "", "target": "", "mod_dir": "", "method": "", "started": _now(), |
| "stage": "intake", "stage_detail": "", "activity": "starting", "activity_time": _now(), |
| "protected": [], "decisions": [], "issues": [], "mods": [], "events": [], "deps": [], "conflicts": []} |
| if self.xml_path.exists(): |
| try: |
| self._load() |
| except Exception: |
| pass |
| self.save() |
|
|
| def _load(self): |
| r = ET.parse(self.xml_path).getroot() |
| t = lambda tag: (r.findtext(tag) or "") |
| d = self.d |
| d.update(game=t("game"), target=t("target"), mod_dir=t("modDir"), method=t("method"), |
| started=t("started") or d["started"]) |
| st = r.find("stage") |
| if st is not None: |
| d["stage"], d["stage_detail"] = st.get("current", "intake"), st.text or "" |
| d["protected"] = [x.text or "" for x in r.findall("protectedFolders/folder")] |
| d["decisions"] = [x.text or "" for x in r.findall("decisions/decision")] |
| d["issues"] = [{"text": x.text or "", "severity": x.get("severity", ""), "resolved": x.get("resolved") == "true"} |
| for x in r.findall("issues/issue")] |
| d["mods"] = [{k: m.get(k, "") for k in ("name", "version", "status", "source", "deps", "notes", "bundles")} |
| for m in r.findall("mods/mod")] |
| d["deps"] = [{k: x.get(k, "") for k in ("name", "version", "folder", "provided_by", "hard_for", "soft_for", "note")} |
| for x in r.findall("dependencies/dependency")] |
| d["conflicts"] = [{"text": x.text or "", "status": x.get("status", "open"), "resolution": x.get("resolution", "")} |
| for x in r.findall("conflicts/conflict")] |
| d["events"] = [{"time": x.get("time", ""), "text": x.text or ""} for x in r.findall("events/event")] |
|
|
| def event(self, text): |
| self.d["events"].append({"time": _now(), "text": text[:300]}) |
| self.d["events"] = self.d["events"][-80:] |
|
|
| def activity(self, text): |
| self.d["activity"], self.d["activity_time"] = text[:200], _now() |
|
|
| def save(self): |
| d = self.d |
| root = ET.Element("tracker", version="1", updated=_now()) |
| for tag, key in (("game", "game"), ("target", "target"), ("modDir", "mod_dir"), ("method", "method"), ("started", "started")): |
| ET.SubElement(root, tag).text = d[key] |
| ET.SubElement(root, "stage", current=d["stage"]).text = d["stage_detail"] |
| ET.SubElement(root, "activity", updated=d["activity_time"]).text = d["activity"] |
| pf = ET.SubElement(root, "protectedFolders") |
| for x in d["protected"]: |
| ET.SubElement(pf, "folder").text = x |
| ds = ET.SubElement(root, "decisions") |
| for x in d["decisions"]: |
| ET.SubElement(ds, "decision").text = x |
| iss = ET.SubElement(root, "issues") |
| for x in d["issues"]: |
| ET.SubElement(iss, "issue", severity=x["severity"], resolved=str(x["resolved"]).lower()).text = x["text"] |
| ms = ET.SubElement(root, "mods") |
| for m in d["mods"]: |
| ET.SubElement(ms, "mod", **{k: m.get(k, "") for k in ("name", "version", "status", "source", "deps", "notes", "bundles")}) |
| dp = ET.SubElement(root, "dependencies") |
| for x in d["deps"]: |
| ET.SubElement(dp, "dependency", **{k: x.get(k, "") for k in ("name", "version", "folder", "provided_by", "hard_for", "soft_for", "note")}) |
| cf = ET.SubElement(root, "conflicts") |
| for x in d["conflicts"]: |
| ET.SubElement(cf, "conflict", status=x["status"], resolution=x.get("resolution", "")).text = x["text"] |
| ev = ET.SubElement(root, "events") |
| for x in d["events"]: |
| ET.SubElement(ev, "event", time=x["time"]).text = x["text"] |
| try: |
| ET.indent(root) |
| except AttributeError: |
| pass |
| self._write(self.xml_path, ET.tostring(root, encoding="unicode", xml_declaration=False)) |
| self._write(self.html_path, self._html()) |
|
|
| @staticmethod |
| def _write(path, text): |
| tmp = path.with_suffix(path.suffix + ".tmp") |
| tmp.write_text(text, encoding="utf-8") |
| try: |
| os.replace(tmp, path) |
| except OSError: |
| path.write_text(text, encoding="utf-8") |
|
|
| def _html(self): |
| e, d = html.escape, self.d |
| idx = STAGES.index(d["stage"]) if d["stage"] in STAGES else 0 |
| steps = "".join(f'<div class="step {"done" if i < idx else "now" if i == idx else ""}">{s}</div>' |
| for i, s in enumerate(STAGES)) |
| counts = {s: sum(1 for m in d["mods"] if m["status"] == s) for s in STATUSES} |
| chips = "".join(f'<span class="chip" style="background:{COLORS[s]}">{s} {n}</span>' for s, n in counts.items() if n) |
| rows = "".join( |
| f'<tr><td>{e(m["name"])}</td><td>{e(m["version"])}</td>' |
| f'<td><span class="chip" style="background:{COLORS.get(m["status"], "#888")}">{e(m["status"])}</span></td>' |
| f'<td>{e(m["deps"])}</td><td>{e(m.get("bundles", ""))}</td><td>{e(m["notes"])}</td>' |
| f'<td><a href="{e(m["source"])}">{e(m["source"][:48])}</a></td></tr>' for m in d["mods"]) |
| deprows = "".join(f'<tr><td>{e(x["name"])}</td><td>{e(x["version"])}</td><td>{e(x["folder"])}</td><td>{e(x["provided_by"])}</td>' |
| f'<td>{e(x["hard_for"])}</td><td>{e(x["soft_for"])}</td><td>{e(x["note"])}</td></tr>' for x in d["deps"]) |
| conflicts = "".join(f'<li class="{"ok" if c["status"] == "resolved" else ""}"><b>{"RESOLVED" if c["status"] == "resolved" else "OPEN"}</b> ' |
| f'{e(c["text"])}{" -> " + e(c["resolution"]) if c.get("resolution") else ""}</li>' for c in d["conflicts"]) or "<li>none</li>" |
| issues = "".join(f'<li class="{"ok" if i["resolved"] else ""}"><b>{e(i["severity"])}</b> {e(i["text"])}' |
| f'{" (resolved)" if i["resolved"] else ""}</li>' for i in d["issues"]) |
| events = "".join(f'<div><span>{e(x["time"][11:])}</span> {e(x["text"])}</div>' for x in reversed(d["events"][-40:])) |
| return f"""<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="3"> |
| <title>Mod tracker - {e(d["game"])}</title><style> |
| :root{{color-scheme:light dark}} body{{font:14px system-ui,sans-serif;margin:20px;max-width:1100px}} |
| .steps{{display:flex;gap:6px;margin:12px 0}} .step{{flex:1;padding:6px;text-align:center;border-radius:6px;background:#8883;font-size:12px}} |
| .step.done{{background:#2fa86644}} .step.now{{background:#4f8cff;color:#fff;font-weight:700}} |
| .chip{{color:#fff;border-radius:10px;padding:1px 8px;font-size:12px;margin-right:4px}} |
| table{{border-collapse:collapse;width:100%}} td,th{{border-bottom:1px solid #8883;padding:4px 8px;text-align:left;vertical-align:top}} |
| .now-box{{border:1px solid #4f8cff;border-radius:8px;padding:8px 12px;margin:10px 0}} |
| .log div{{font:12px ui-monospace,monospace;padding:1px 0}} .log span{{opacity:.6}} li.ok{{opacity:.5}} |
| </style></head><body> |
| <h2>{e(d["game"] or "Mod install")} <small>{e(d["target"])}</small></h2> |
| <div>Mod dir: {e(d["mod_dir"])} · Method: {e(d["method"])} · Started: {e(d["started"])} · Updated: {_now()} (auto-refresh 3s)</div> |
| <div>Protected: {e(", ".join(d["protected"]) or "none set")}</div> |
| <div class="steps">{steps}</div> |
| <div class="now-box"><b>Stage: {e(d["stage"])}</b> {e(d["stage_detail"])}<br>Now: {e(d["activity"])} <small>({e(d["activity_time"][11:])})</small></div> |
| <div>{chips}</div> |
| <table><tr><th>Mod</th><th>Version</th><th>Status</th><th>Needs</th><th>Bundles (hard/soft)</th><th>Notes</th><th>Source</th></tr>{rows}</table> |
| <h3>Dependency chain</h3><table><tr><th>Dependency</th><th>Version</th><th>Folder</th><th>Provided by</th><th>Hard-required by</th><th>Soft</th><th>Note</th></tr>{deprows}</table> |
| <h3>Conflicts</h3><ul>{conflicts}</ul> |
| <h3>Issues</h3><ul>{issues or "<li>none</li>"}</ul> |
| <h3>Decisions</h3><ul>{"".join(f"<li>{e(x)}</li>" for x in d["decisions"]) or "<li>none</li>"}</ul> |
| <h3>Live log</h3><div class="log">{events}</div></body></html>""" |
|
|
|
|
| TRACKER = None |
|
|
|
|
| def tracker_stage(stage, detail=""): |
| if stage not in STAGES: |
| return f"Unknown stage. Use one of: {', '.join(STAGES)}" |
| TRACKER.d["stage"], TRACKER.d["stage_detail"] = stage, detail |
| TRACKER.event(f"STAGE {stage}: {detail}") |
| TRACKER.save() |
| return f"Stage set to {stage}." |
|
|
|
|
| def tracker_info(game=None, target=None, mod_dir=None, method=None): |
| for k, v in (("game", game), ("target", target), ("mod_dir", mod_dir), ("method", method)): |
| if v: |
| TRACKER.d[k] = v |
| TRACKER.save() |
| return "Tracker info updated." |
|
|
|
|
| def tracker_dep(name, version=None, folder=None, provided_by=None, hard_for=None, soft_for=None, note=None): |
| """Record a dependency in the chain: who provides it, which version, who HARD-requires it (exact) and who tolerates others (soft).""" |
| x = next((q for q in TRACKER.d["deps"] if q["name"].lower() == name.lower()), None) |
| if x is None: |
| x = {"name": name, "version": "", "folder": "", "provided_by": "", "hard_for": "", "soft_for": "", "note": ""} |
| TRACKER.d["deps"].append(x) |
| for k, v in (("version", version), ("folder", folder), ("provided_by", provided_by), ("note", note)): |
| if v: |
| x[k] = v |
| for k, v in (("hard_for", hard_for), ("soft_for", soft_for)): |
| if v: |
| have = [t.strip() for t in x[k].split(";") if t.strip()] |
| x[k] = "; ".join(dict.fromkeys(have + [t.strip() for t in str(v).split(";") if t.strip()])) |
| TRACKER.event(f"DEP {name} {x['version']}: hard for [{x['hard_for']}]") |
| TRACKER.save() |
| return f"Dependency {name} recorded." |
|
|
|
|
| def tracker_mod(name, status=None, version=None, source=None, deps=None, notes=None, bundles=None): |
| if status and status not in STATUSES: |
| return f"Unknown status. Use one of: {', '.join(STATUSES)}" |
| m = next((x for x in TRACKER.d["mods"] if x["name"].lower() == name.lower()), None) |
| if m is None: |
| m = {"name": name, "version": "", "status": "todo", "source": "", "deps": "", "notes": "", "bundles": ""} |
| TRACKER.d["mods"].append(m) |
| for k, v in (("status", status), ("version", version), ("source", source), ("deps", deps), ("notes", notes), ("bundles", bundles)): |
| if v: |
| m[k] = v |
| TRACKER.event(f"MOD {name}: {m['status']}" + (f" - {notes}" if notes else "")) |
| TRACKER.save() |
| return f"Tracker: {name} is {m['status']}." |
|
|
|
|
| def tracker_note(kind, text, severity="", resolution=""): |
| d = TRACKER.d |
| if kind == "decision": |
| d["decisions"].append(text) |
| elif kind == "protected": |
| d["protected"].append(text) |
| if G is not None: |
| G.add_protect(text) |
| elif kind == "issue": |
| d["issues"].append({"text": text, "severity": severity or "feature-broken", "resolved": False}) |
| elif kind == "conflict": |
| d["conflicts"].append({"text": text, "status": "open", "resolution": ""}) |
| elif kind == "conflict_resolved": |
| for c in d["conflicts"]: |
| if text.lower() in c["text"].lower(): |
| c["status"], c["resolution"] = "resolved", resolution |
| elif kind == "resolved": |
| hit = [i for i in d["issues"] if text.lower() in i["text"].lower()] |
| for i in hit: |
| i["resolved"] = True |
| TRACKER.event(f"RESOLVED {text}") |
| else: |
| return "kind must be: decision, protected, issue, resolved, conflict or conflict_resolved" |
| TRACKER.event(f"{kind.upper()}: {text}") |
| TRACKER.save() |
| return f"Tracker {kind} recorded." |
|
|
|
|
| |
| |
| |
| TEX_EXT = {".png", ".jpg", ".jpeg", ".tga", ".bmp", ".tif", ".tiff", ".webp", ".dds"} |
| MAP_RX = re.compile(r"(?:[_\-. ](?:n|nrm|norm|normal|normals|nm|bump|spec|specular|rough|roughness|metal|metallic|ao|mask|height|disp|orm|arm|gloss))$" |
| r"|normal", re.I) |
| MAX_TEX_FILES = 20000 |
| RESAMPLE = {"nearest": 0, "bilinear": 2, "bicubic": 3, "lanczos": 1} |
| COLOR_OPS = ("brightness", "contrast", "saturation", "gamma", "hue_shift", "tint", "levels", "channel_gain", |
| "grayscale", "invert", "sharpen", "blur") |
|
|
|
|
| def _pil(): |
| try: |
| from PIL import Image |
| return True |
| except ImportError: |
| return False |
|
|
|
|
| def _nums(text, n, name): |
| try: |
| vals = [float(x) for x in str(text).replace(";", ",").split(",") if x.strip() != ""] |
| except ValueError: |
| raise ValueError(f"{name} must be numbers separated by commas") |
| if len(vals) not in (n if isinstance(n, tuple) else (n,)): |
| raise ValueError(f"{name} needs {n} numbers, got {len(vals)}") |
| return vals |
|
|
|
|
| def _truthy(v): |
| return str(v).lower() in ("1", "true", "yes", "y", "on") |
|
|
|
|
| def _is_map(path): |
| return bool(MAP_RX.search(Path(path).stem)) |
|
|
|
|
| def apply_ops(img, ops, is_map=False, include_maps=False): |
| """Apply geometry then colour operations. Returns (image, notes).""" |
| from PIL import Image, ImageChops, ImageEnhance, ImageFilter, ImageOps |
| notes = [] |
| orig_mode = img.mode |
| has_alpha = "A" in orig_mode or (orig_mode == "P" and "transparency" in img.info) |
| if orig_mode not in ("RGB", "RGBA", "L", "LA"): |
| img = img.convert("RGBA" if has_alpha else "RGB") |
| |
| w, h = img.size |
| res = {"nearest": Image.NEAREST, "bilinear": Image.BILINEAR, "bicubic": Image.BICUBIC, |
| "lanczos": Image.LANCZOS}.get(str(ops.get("resample", "lanczos")).lower(), Image.LANCZOS) |
| nw, nh = w, h |
| if ops.get("scale") not in (None, ""): |
| f = float(ops["scale"]) |
| nw, nh = max(1, round(w * f)), max(1, round(h * f)) |
| if ops.get("max_size") not in (None, ""): |
| m = int(float(ops["max_size"])) |
| if max(nw, nh) > m: |
| k = m / max(nw, nh) |
| nw, nh = max(1, round(nw * k)), max(1, round(nh * k)) |
| if (nw, nh) != (w, h): |
| img = img.resize((nw, nh), res) |
| |
| wanted = [k for k in COLOR_OPS if ops.get(k) not in (None, "", False)] |
| if wanted and is_map and not include_maps: |
| notes.append("colour ops skipped (data map: " + Path(str(ops.get("_name", ""))).stem + ")") |
| wanted = [] |
| if wanted: |
| alpha = img.getchannel("A") if "A" in img.mode else None |
| gray_src = img.mode in ("L", "LA") |
| region = ops.get("region") |
| box = None |
| if region: |
| x, y, rw, rh = [int(v) for v in _nums(region, 4, "region")] |
| box = (max(0, x), max(0, y), min(img.width, x + rw), min(img.height, y + rh)) |
| rgb_full = img.convert("RGB") |
| work = rgb_full.crop(box) if box else rgb_full |
| if ops.get("levels"): |
| ib, iw, ob, ow = _nums(ops["levels"], 4, "levels") |
| span = max(1.0, iw - ib) |
| lut = [int(max(0, min(255, ob + max(0.0, min(1.0, (v - ib) / span)) * (ow - ob)))) for v in range(256)] |
| work = work.point(lut * 3) |
| if ops.get("gamma"): |
| g = float(ops["gamma"]) |
| work = work.point([int(255 * ((v / 255) ** (1 / g))) for v in range(256)] * 3) |
| if ops.get("brightness"): |
| work = ImageEnhance.Brightness(work).enhance(float(ops["brightness"])) |
| if ops.get("contrast"): |
| work = ImageEnhance.Contrast(work).enhance(float(ops["contrast"])) |
| if ops.get("saturation"): |
| work = ImageEnhance.Color(work).enhance(float(ops["saturation"])) |
| if ops.get("hue_shift"): |
| shift = int(round(float(ops["hue_shift"]) / 360 * 256)) |
| hh, ss, vv = work.convert("HSV").split() |
| hh = hh.point([(i + shift) % 256 for i in range(256)]) |
| work = Image.merge("HSV", (hh, ss, vv)).convert("RGB") |
| gain_a = 1.0 |
| if ops.get("channel_gain"): |
| vals = _nums(ops["channel_gain"], (3, 4), "channel_gain") |
| r, g_, b = vals[:3] |
| gain_a = vals[3] if len(vals) == 4 else 1.0 |
| rr, gg, bb = work.split() |
| work = Image.merge("RGB", tuple(c.point([int(min(255, v * k)) for v in range(256)]) |
| for c, k in zip((rr, gg, bb), (r, g_, b)))) |
| if ops.get("tint"): |
| col = str(ops["tint"]).lstrip("#") |
| colour = tuple(int(col[i:i + 2], 16) for i in (0, 2, 4)) |
| strength = float(ops.get("tint_strength", 0.5) or 0.5) |
| work = Image.blend(work, ImageChops.multiply(work, Image.new("RGB", work.size, colour)), strength) |
| if _truthy(ops.get("grayscale")): |
| work = work.convert("L").convert("RGB") |
| if _truthy(ops.get("invert")): |
| work = ImageOps.invert(work) |
| if ops.get("sharpen"): |
| work = work.filter(ImageFilter.UnsharpMask(radius=1.5, percent=int(float(ops["sharpen"]) * 100), threshold=2)) |
| if ops.get("blur"): |
| work = work.filter(ImageFilter.GaussianBlur(float(ops["blur"]))) |
| if box: |
| rgb_full.paste(work, box[:2]) |
| work = rgb_full |
| if alpha is not None: |
| if gain_a != 1.0: |
| alpha = alpha.point([int(min(255, v * gain_a)) for v in range(256)]) |
| work.putalpha(alpha) |
| img = work.convert("L") if gray_src and "A" not in img.mode else work |
| return img, notes |
|
|
|
|
| def _save_texture(img, path, fmt): |
| path.parent.mkdir(parents=True, exist_ok=True) |
| ext = path.suffix.lower() |
| if ext in (".jpg", ".jpeg") and img.mode in ("RGBA", "LA"): |
| img = img.convert("RGB") |
| try: |
| if ext == ".dds": |
| if img.mode not in ("RGB", "RGBA", "L", "LA"): |
| img = img.convert("RGBA") |
| img.save(path, "DDS") |
| else: |
| img.save(path) |
| return None |
| except Exception as e: |
| return f"{type(e).__name__}: {e}" + (" (DDS compression needs a newer Pillow, or use texconv/ImageMagick/nvcompress from the game's modding tools)" if ext == ".dds" else "") |
|
|
|
|
| def texture_info(path, max_files=25): |
| if not _pil(): |
| return "texture tools need Pillow: pip install pillow" |
| from PIL import Image |
| p = G.resolve(path) |
| if not G.can_read(p): |
| return f"Blocked: {p} is outside the workspace and allowed folders." |
| files = [p] if p.is_file() else [f for f in sorted(p.rglob("*")) if f.suffix.lower() in TEX_EXT] |
| if not files: |
| return "No texture files (png, jpg, tga, bmp, tif, webp, dds) found." |
| sizes, exts, modes, rows, bad = {}, {}, {}, [], 0 |
| for f in files[:MAX_TEX_FILES]: |
| try: |
| with Image.open(f) as im: |
| s, md = im.size, im.mode |
| except Exception: |
| bad += 1 |
| continue |
| sizes[f"{s[0]}x{s[1]}"] = sizes.get(f"{s[0]}x{s[1]}", 0) + 1 |
| exts[f.suffix.lower()] = exts.get(f.suffix.lower(), 0) + 1 |
| modes[md] = modes.get(md, 0) + 1 |
| if len(rows) < int(max_files): |
| rows.append(f"{f.relative_to(p) if p.is_dir() else f.name} | {s[0]}x{s[1]} | {md} | {f.stat().st_size // 1024} KB" |
| + (" | data map?" if _is_map(f) else "")) |
| top = sorted(sizes.items(), key=lambda kv: -kv[1])[:10] |
| maps = sum(1 for f in files if _is_map(f)) |
| return clip(f"{len(files)} textures ({bad} unreadable). Formats {exts}. Modes {modes}. Likely data maps (normal/spec/etc.): {maps}.\n" |
| f"Sizes (count): {top}\n" + "\n".join(rows)) |
|
|
|
|
| def texture_batch(src, dest="", pattern="**/*", exclude="", scale=None, max_size=None, brightness=None, contrast=None, |
| saturation=None, gamma=None, hue_shift=None, tint=None, tint_strength=None, levels=None, |
| channel_gain=None, grayscale=None, invert=None, sharpen=None, blur=None, region=None, |
| resample="lanczos", format=None, rules=None, include_maps=False, in_place=False, dry_run=True): |
| """Mass and fine-tuned edits of loose texture files. Dry run by default.""" |
| if not _pil(): |
| return "texture tools need Pillow: pip install pillow" |
| from PIL import Image |
| s = G.resolve(src) |
| if not G.can_read(s) or not s.exists(): |
| return f"Cannot read {s}: outside the allowed folders or missing." |
| base_ops = {k: v for k, v in dict(scale=scale, max_size=max_size, brightness=brightness, contrast=contrast, |
| saturation=saturation, gamma=gamma, hue_shift=hue_shift, tint=tint, tint_strength=tint_strength, |
| levels=levels, channel_gain=channel_gain, grayscale=grayscale, invert=invert, sharpen=sharpen, |
| blur=blur, region=region, resample=resample).items() if v not in (None, "")} |
| rule_list = [] |
| if rules: |
| try: |
| rule_list = json.loads(rules) if isinstance(rules, str) else list(rules) |
| assert isinstance(rule_list, list) |
| except Exception: |
| return 'rules must be a JSON list like [{"match":"*sky*","brightness":0.9},{"match":"*_n.*","skip":true}]' |
| root = s if s.is_dir() else s.parent |
| files = [s] if s.is_file() else [f for f in sorted(s.glob(pattern)) if f.is_file() and f.suffix.lower() in TEX_EXT] |
| excl = [x.strip().lower() for x in str(exclude).split(",") if x.strip()] |
| plan, skipped = [], {"excluded": 0, "rule-skip": 0, "protected": 0} |
| out_root = (root if in_place else (G.resolve(dest) if dest else G.workspace / "textures_out" / (root.name or "out"))) |
| for f in files[:MAX_TEX_FILES]: |
| rel = f.relative_to(root) if s.is_dir() else Path(f.name) |
| rl, nm = rel.as_posix().lower(), f.name.lower() |
| if any(fnmatch.fnmatch(rl, x) or fnmatch.fnmatch(nm, x) for x in excl): |
| skipped["excluded"] += 1 |
| continue |
| ops, inc, skip = dict(base_ops), _truthy(include_maps), False |
| for r in rule_list: |
| m = str(r.get("match", "*")).lower() |
| if fnmatch.fnmatch(rl, m) or fnmatch.fnmatch(nm, m): |
| if _truthy(r.get("skip")): |
| skip = True |
| break |
| ops.update({k: v for k, v in r.items() if k not in ("match", "skip", "include_maps")}) |
| if "include_maps" in r: |
| inc = _truthy(r["include_maps"]) |
| if skip: |
| skipped["rule-skip"] += 1 |
| continue |
| ext = ("." + str(format).lower().lstrip(".")) if format else f.suffix |
| target = out_root / rel.with_suffix(ext) |
| if G.is_protected(target): |
| skipped["protected"] += 1 |
| continue |
| plan.append((f, target, ops, inc)) |
| if not plan: |
| return f"Nothing to do (matched {len(files)} textures; skipped {skipped}). Check pattern, exclude and rules." |
| if not base_ops and not rule_list and not format: |
| return "No operation given. Set at least one of: scale, max_size, brightness, contrast, saturation, gamma, hue_shift, tint, levels, channel_gain, grayscale, invert, sharpen, blur, format, or rules." |
| maps_n = sum(1 for f, _, _, inc in plan if _is_map(f) and not inc) |
| head = (f"{'DRY RUN' if dry_run else 'RUN'}: {len(plan)} textures from {s} -> {out_root}" |
| f"{' (IN PLACE, originals backed up)' if in_place else ''}. Skipped {skipped}. " |
| f"{maps_n} look like data maps (normal/spec/etc.): colour edits are skipped for those, resizing still applies.") |
| if dry_run: |
| lines = [] |
| for f, t, ops, inc in plan[:8]: |
| try: |
| with Image.open(f) as im: |
| img, notes = apply_ops(im.copy(), dict(ops, _name=f.name), _is_map(f), inc) |
| lines.append(f" {f.name}: {im.size[0]}x{im.size[1]} -> {img.size[0]}x{img.size[1]} -> {t.name}" + (f" [{'; '.join(notes)}]" if notes else "")) |
| except Exception as e: |
| lines.append(f" {f.name}: cannot open ({e})") |
| return head + "\n" + "\n".join(lines) + "\nNo files were written. To apply, call again with the same settings and dry_run=false." |
| deny = G.confirm_write(out_root, f"write {len(plan)} edited textures" + (" IN PLACE (backups kept)" if in_place else "")) |
| if deny: |
| return deny |
| stamp = "tex-" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S") |
| job, n = stamp, 1 |
| while (G.workspace / "texture_jobs" / f"{job}.json").exists() or (G.workspace / "texture_jobs" / job).exists(): |
| n += 1 |
| job = f"{stamp}-{n}" |
| bdir = G.workspace / "texture_jobs" / job / "backup" |
| done, created, backups, errors, before, after = 0, [], {}, [], 0, 0 |
| for f, t, ops, inc in plan: |
| try: |
| if t.exists(): |
| b = bdir / (t.relative_to(out_root) if Guard._inside(t, out_root) else Path(t.name)) |
| b.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(t, b) |
| backups[str(t)] = str(b) |
| elif in_place is False: |
| created.append(str(t)) |
| with Image.open(f) as im: |
| im.load() |
| before += f.stat().st_size |
| img, _ = apply_ops(im, dict(ops, _name=f.name), _is_map(f), inc) |
| err = _save_texture(img, t, format) |
| if err: |
| errors.append(f"{f.name}: {err}") |
| continue |
| if in_place and str(t) not in backups: |
| created.append(str(t)) |
| after += t.stat().st_size |
| done += 1 |
| except Exception as e: |
| errors.append(f"{f.name}: {type(e).__name__}: {e}") |
| (G.workspace / "texture_jobs").mkdir(parents=True, exist_ok=True) |
| (G.workspace / "texture_jobs" / f"{job}.json").write_text(json.dumps( |
| {"job": job, "created": created, "backups": backups, "ops": base_ops, "rules": rule_list}, indent=1), encoding="utf-8") |
| msg = (f"Done: {done}/{len(plan)} textures written to {out_root}. Size {before // 1024} KB -> {after // 1024} KB. " |
| f"Job {job} (undo with texture_undo job='{job}').") |
| return msg + (f"\n{len(errors)} errors, first: " + "; ".join(errors[:4]) if errors else "") |
|
|
|
|
| def texture_undo(job): |
| p = G.workspace / "texture_jobs" / f"{job}.json" |
| if not p.exists(): |
| jobs = sorted(x.stem for x in (G.workspace / "texture_jobs").glob("*.json")) if (G.workspace / "texture_jobs").exists() else [] |
| return f"No job '{job}'. Jobs: {jobs or 'none'}" |
| data = json.loads(p.read_text(encoding="utf-8")) |
| anchor = Path((list(data["backups"]) + data["created"] or [str(G.workspace)])[0]).parent |
| deny = G.confirm_write(anchor, f"undo texture job {job}") |
| if deny: |
| return deny |
| restored = removed = 0 |
| for orig, bak in data["backups"].items(): |
| if Path(bak).exists(): |
| shutil.copy2(bak, orig) |
| restored += 1 |
| for c in data["created"]: |
| if c not in data["backups"] and Path(c).exists(): |
| Path(c).unlink() |
| removed += 1 |
| return f"Undid {job}: restored {restored} original files, removed {removed} created files." |
|
|
|
|
| def _module_desc(path): |
| for line in path.read_text(encoding="utf-8").splitlines(): |
| if line.startswith(">"): |
| return line[1:].strip() |
| return "" |
|
|
|
|
| def list_modules(): |
| rows = [f"{p.stem}: {_module_desc(p)}" for p in sorted(MODULE_DIR.glob("*.md"))] |
| return "\n".join(rows) or "No modules." |
|
|
|
|
| def _sections(text): |
| parts, order, head, buf = {}, [], "_intro", [] |
| for line in text.splitlines(): |
| if line.startswith("## "): |
| parts[head] = "\n".join(buf).strip() |
| order.append(head) |
| head, buf = line[3:].strip(), [] |
| else: |
| buf.append(line) |
| parts[head] = "\n".join(buf).strip() |
| order.append(head) |
| return parts, order |
|
|
|
|
| def load_module(name="", section=""): |
| if not name: |
| return list_modules() |
| p = MODULE_DIR / f"{re.sub(r'[^a-z0-9_-]', '', name.lower())}.md" |
| if not p.exists(): |
| return f"No module named '{name}'. Available:\n{list_modules()}" |
| text = p.read_text(encoding="utf-8") |
| parts, order = _sections(text) |
| heads = [k for k in order if k != "_intro"] |
| if section: |
| for k in order: |
| if k != "_intro" and (k.lower() == section.lower() or section.lower() in k.lower()): |
| return f"## {k}\n{parts[k]}" |
| return f"No section '{section}'. Sections: {', '.join(heads)}" |
| if LEAN: |
| keep = [k for k in heads if k in ("Mod folder and format", "Logs", "Method")] |
| if keep: |
| body = parts["_intro"] + "\n\n" + "\n\n".join(f"## {k}\n{parts[k]}" for k in keep) |
| else: |
| body = text[:2200] |
| return body.strip() + f"\n\n[Other sections via load_module(name, section=...): {', '.join(heads)}]" |
| return text |
|
|
|
|
| GAME_SIGS = [ |
| ("ksp", "Kerbal Space Program", ("kerbal space program",), ("KSP_x64.exe", "KSP.exe")), |
| ("factorio", "Factorio", ("factorio",), ("bin/x64/factorio.exe", "factorio.exe")), |
| ("bg3", "Baldur's Gate 3", ("baldur",), ("bin/bg3.exe", "bin/bg3_dx11.exe")), |
| ("bethesda", "Oblivion Remastered", ("oblivion remastered",), ("OblivionRemastered.exe", "OblivionRemastered")), |
| ("bethesda", "Oblivion (2006)", ("oblivion",), ("Oblivion.exe",)), |
| ("bethesda", "Fallout: New Vegas", ("new vegas",), ("FalloutNV.exe",)), |
| ("bethesda", "Fallout 3", ("fallout 3",), ("Fallout3.exe",)), |
| ("bethesda", "Fallout 4", ("fallout 4",), ("Fallout4.exe",)), |
| ("bethesda", "Skyrim", ("skyrim",), ("SkyrimSE.exe", "SkyrimVR.exe", "TESV.exe", "SkyrimSELauncher.exe")), |
| ("minecraft", "Minecraft", (".minecraft", "minecraft"), ("launcher_profiles.json",)), |
| ] |
|
|
|
|
| FOCUS = {"game": ""} |
|
|
|
|
| def module_for_game(name): |
| """Map a game name the user typed to a module, if one exists (any game is allowed; no module just means 'research it').""" |
| low = re.sub(r"[^a-z0-9 ]", "", str(name).lower()).strip() |
| slug = low.replace(" ", "-") |
| if (MODULE_DIR / f"{slug}.md").exists() and not slug.startswith("_"): |
| return slug |
| for module, label, words, files in GAME_SIGS: |
| if low and (low in label.lower() or any(w.strip(".") in low or low in w for w in words)): |
| return module |
| return None |
|
|
|
|
| def _engine_hint(d): |
| try: |
| if (d / "Engine").is_dir() or (d / "Content" / "Paks").is_dir() or any(d.glob("*/Content/Paks")): |
| return "Unreal Engine layout (mods are usually .pak files in Content/Paks, sometimes UE4SS scripts: check the game's mod page)" |
| if any(d.glob("*.pck")): |
| return "Godot (.pck)" |
| if (d / "UnityPlayer.dll").exists(): |
| return "Unity" |
| except OSError: |
| pass |
| return "" |
|
|
|
|
| def _classify(d): |
| low = d.name.lower() |
| for module, label, words, files in GAME_SIGS: |
| if any(w in low for w in words) or any((d / f).exists() for f in files): |
| note = "" |
| if module == "bethesda" and "Skyrim" in label: |
| note = ("edition SE/AE" if (d / "SkyrimSE.exe").exists() else "edition VR" if (d / "SkyrimVR.exe").exists() |
| else "edition LE" if (d / "TESV.exe").exists() else "edition unknown: ask") |
| return module, label, note |
| if (d / "UnityPlayer.dll").exists(): |
| if (d / "GameAssembly.dll").exists(): |
| kind = "IL2CPP build: needs the BepInEx IL2CPP build" |
| elif list(d.glob("*_Data/Managed/Assembly-CSharp.dll")): |
| kind = "Mono build: needs BepInEx 5.x" |
| else: |
| kind = "Unity, backend unclear: list the folder" |
| return "unity-bepinex", f"Unity game '{d.name}'", kind + ("; BepInEx already present" if (d / "BepInEx").is_dir() else "") |
| return None |
|
|
|
|
| def _candidate_dirs(root): |
| seen, out = set(), [] |
|
|
| def add(d): |
| if d not in seen and d.is_dir(): |
| seen.add(d) |
| out.append(d) |
|
|
| def kids(d, cap=200): |
| try: |
| return [c for c in sorted(d.iterdir())[:cap] if c.is_dir()] |
| except (PermissionError, OSError): |
| return [] |
|
|
| add(root) |
| for c in kids(root): |
| add(c) |
| if c.name.lower() == "steamapps": |
| for g in kids(c / "common"): |
| add(g) |
| else: |
| for g in kids(c): |
| add(g) |
| for g in kids(c / "steamapps" / "common"): |
| add(g) |
| return out |
|
|
|
|
| def find_games(root=""): |
| roots = [G.resolve(root)] if root else list(G.allow) |
| if not roots: |
| return "No allowed folders are set. Ask the user for their games root (for example C:/Games) and to restart with --allow." |
| rows, seen = [], set() |
| for r in roots: |
| if not G.can_read(r) or not r.is_dir(): |
| rows.append(f"(skipped {r}: not an allowed folder)") |
| continue |
| for d in _candidate_dirs(r): |
| hit = _classify(d) |
| if hit and d not in seen: |
| seen.add(d) |
| module, label, note = hit |
| rows.append(f"{d} | {label} | module: {module}" + (f" | {note}" if note else "")) |
| elif d not in seen and d != r: |
| try: |
| exes = [f.name for f in d.glob("*.exe") if not re.search(r"unins|crash|redist|setup|launcher|vc_|dxsetup", f.name, re.I)] |
| except OSError: |
| exes = [] |
| if exes: |
| seen.add(d) |
| hint = _engine_hint(d) |
| rows.append(f"{d} | UNRECOGNIZED game? exe: {exes[0]} | no module yet: use web_search and load_module('new-game')" |
| + (f" | {hint}" if hint else "")) |
| return "\n".join(rows[:40]) or "No known games found in the allowed folders. Ask the user which game and where it is installed." |
|
|
|
|
| def save_module(name, content): |
| slug = re.sub(r"[^a-z0-9_-]", "", name.lower()) |
| if not slug or slug.startswith("_"): |
| return "Invalid module name." |
| p = MODULE_DIR / f"{slug}.md" |
| existed = p.exists() |
| if not G.auto_yes: |
| ans = input(f"\n[confirm] {'overwrite' if existed else 'create'} module '{slug}'? [y/N] ").strip().lower() |
| if ans != "y": |
| return "The user declined." |
| p.write_text(content, encoding="utf-8") |
| return f"Saved module '{slug}'." |
|
|
|
|
| TOOLS = { |
| "web_search": (web_search, "Search the web. Use before stating any version, link, requirement or install path.", |
| {"query": "string", "max_results": "integer?"}), |
| "web_fetch": (web_fetch, "Fetch a web page or JSON as text (with a links list). Use on the best search result.", |
| {"url": "string", "max_chars": "integer?"}), |
| "browser_fetch": (browser_fetch, "Render a JavaScript-heavy page in a headless browser; optional js expression to evaluate.", |
| {"url": "string", "js": "string?", "wait_ms": "integer?"}), |
| "download_file": (download_file, "Download a file (release asset, never source code) into the workspace downloads folder.", |
| {"url": "string", "dest": "string (file name or path)"}), |
| "list_archive": (list_archive, "List the contents of a zip/tar before extracting.", {"path": "string"}), |
| "extract_archive": (extract_archive, "Extract a zip/tar into the workspace staging folder.", {"path": "string", "dest": "string"}), |
| "deps_check": (deps_check, "Read-only check of a staged mod against the installed game folder: new, identical, and DIFFERENT (possible dependency-chain breaks).", |
| {"src": "string", "dest": "string", "granularity": "string?"}), |
| "copy_tree": (copy_tree, "Deploy a staged folder into a game folder. Copies new items, skips identical ones, and NEVER overwrites a differing installed dependency: " |
| "conflicts are reported. exclude=names to keep installed, replace=names to swap (user decided; backed up). granularity folder or file.", |
| {"src": "string", "dest": "string", "exclude": "string?", "replace": "string?", "granularity": "string?"}), |
| "list_dir": (list_dir, "List a folder (workspace or allowed game folders).", {"path": "string", "depth": "integer?"}), |
| "read_file": (read_file, "Read lines of a text file; tail=true reads the end (good for crashes).", |
| {"path": "string", "start": "integer?", "lines": "integer?", "tail": "boolean?"}), |
| "grep_file": (grep_file, "Regex search a file (case-insensitive), with optional context lines. Use on logs.", |
| {"path": "string", "pattern": "string", "max_matches": "integer?", "context": "integer?"}), |
| "write_file": (write_file, "Write or append a text file (tracker, reports, scripts). Asks the user outside the workspace.", |
| {"path": "string", "content": "string", "append": "boolean?"}), |
| "texture_info": (texture_info, "Inspect loose textures (size, format, alpha, likely data maps) in a file or folder.", |
| {"path": "string", "max_files": "integer?"}), |
| "texture_batch": (texture_batch, "Mass-edit loose textures (resize, colour, levels, per-file rules). Dry run by default; set dry_run=false to apply. " |
| "Writes copies to dest; in_place=true backs up originals. Colour edits skip normal/spec maps. " |
| "rules is a JSON list of per-file overrides: [{\"match\":\"*sky*\",\"brightness\":0.9},{\"match\":\"*_n.*\",\"skip\":true}].", |
| {"src": "string", "dest": "string?", "pattern": "string?", "exclude": "string?", "scale": "number?", "max_size": "integer?", |
| "brightness": "number?", "contrast": "number?", "saturation": "number?", "gamma": "number?", "hue_shift": "number?", |
| "tint": "string?", "tint_strength": "number?", "levels": "string?", "channel_gain": "string?", "grayscale": "boolean?", |
| "invert": "boolean?", "sharpen": "number?", "blur": "number?", "region": "string?", "resample": "string?", |
| "format": "string?", "rules": "string?", "include_maps": "boolean?", "in_place": "boolean?", "dry_run": "boolean?"}), |
| "texture_undo": (texture_undo, "Undo a texture_batch job: restore backed-up originals and remove created files.", {"job": "string"}), |
| "tracker_stage": (tracker_stage, "Set the current stage (" + ", ".join(STAGES) + ") so the user's live tracker view is accurate. Call at every stage change.", |
| {"stage": "string", "detail": "string?"}), |
| "tracker_mod": (tracker_mod, "Add or update one mod in the live tracker (status: " + ", ".join(STATUSES) + "). Call after EVERY step on a mod. " |
| "bundles = dependency mods packaged inside it, with version and hard/soft, e.g. 'EVE 3.2.2 hard; ModuleManager 4.2.3 soft'.", |
| {"name": "string", "status": "string?", "version": "string?", "source": "string?", "deps": "string?", "notes": "string?", "bundles": "string?"}), |
| "tracker_dep": (tracker_dep, "Record a dependency in the chain: installed version, folder, who provided it, who HARD-requires it (exact version) and who is SOFT.", |
| {"name": "string", "version": "string?", "folder": "string?", "provided_by": "string?", "hard_for": "string?", "soft_for": "string?", "note": "string?"}), |
| "tracker_info": (tracker_info, "Set tracker header info: game, target version, mod folder, install method.", |
| {"game": "string?", "target": "string?", "mod_dir": "string?", "method": "string?"}), |
| "tracker_note": (tracker_note, "Record a decision, a protected folder, an issue (severity will-crash/feature-broken/cosmetic), a dependency conflict, or mark one resolved " |
| "(kinds: decision, protected, issue, resolved, conflict, conflict_resolved with resolution).", |
| {"kind": "string", "text": "string", "severity": "string?", "resolution": "string?"}), |
| "list_modules": (list_modules, "List available skill modules (games and tasks).", {}), |
| "find_games": (find_games, "Scan the allowed folders for installed games and say which module to load for each.", {"root": "string?"}), |
| "load_module": (load_module, "Load a skill module (game or task). Empty name lists modules; section loads one section only.", |
| {"name": "string?", "section": "string?"}), |
| "save_module": (save_module, "Save a new game/task module (follow the _template module format).", |
| {"name": "string", "content": "string"}), |
| } |
|
|
|
|
| def tool_schemas(lean=None): |
| lean = LEAN if lean is None else lean |
| schemas = [] |
| for name, (fn, desc, params) in TOOLS.items(): |
| if name == "browser_fetch": |
| try: |
| import playwright |
| except ImportError: |
| continue |
| if lean and name in LEAN_SKIP: |
| continue |
| props, req = {}, [] |
| for k, t in params.items(): |
| if lean and k in LEAN_DROP_PARAMS.get(name, ()): |
| continue |
| optional = t.endswith("?") |
| base = t.rstrip("?").split(" ")[0] |
| props[k] = {"type": base} |
| if not optional: |
| req.append(k) |
| schemas.append({"type": "function", "function": { |
| "name": name, "description": LEAN_DESC.get(name, desc[:90]) if lean else desc, |
| "parameters": {"type": "object", "properties": props, "required": req}}}) |
| return schemas |
|
|
|
|
| def run_tool(name, args): |
| if name not in TOOLS: |
| return f"Unknown tool '{name}'. Available: {', '.join(TOOLS)}" |
| if isinstance(args, str): |
| try: |
| args = json.loads(args) |
| except json.JSONDecodeError: |
| return "Tool arguments were not valid JSON." |
| auto = {"download_file": "download", "extract_archive": "extract", "list_archive": "extract", "copy_tree": "deploy", "deps_check": "scan"} |
| if not name.startswith("tracker_") and TRACKER is not None: |
| TRACKER.activity(f"{name} {json.dumps(args)[:120]}") |
| if name in auto: |
| TRACKER.d["stage"] = auto[name] |
| TRACKER.save() |
| try: |
| out = str(TOOLS[name][0](**(args or {}))) |
| if not name.startswith("tracker_") and TRACKER is not None and name in ( |
| "download_file", "extract_archive", "copy_tree", "deps_check", "write_file", "web_fetch", "web_search", "grep_file", "read_file"): |
| TRACKER.event(f"{name}: {out[:110].replace(chr(10), ' ')}") |
| TRACKER.save() |
| return out |
| except TypeError as e: |
| return f"Bad arguments for {name}: {e}" |
| except Exception as e: |
| return f"{name} failed: {e}" |
|
|
|
|
| |
| |
| |
| def text_protocol(): |
| names = [n for n in TOOLS if not (LEAN and n in LEAN_SKIP)] |
| lines = [f"- {n}({', '.join(TOOLS[n][2])})" + ("" if LEAN else f": {TOOLS[n][1]}") for n in names] |
| return ("\nTOOL CALLING FORMAT (this model has no native tool support):\n" |
| "To use a tool, reply with ONLY one block like this and nothing else:\n" |
| '<tool_call>{"name": "web_search", "arguments": {"query": "example"}}</tool_call>\n' |
| "You will receive the result in a message starting with <tool_result>. Then continue.\nTools:\n" + "\n".join(lines)) |
|
|
|
|
| def focus_lines(): |
| prot = ", ".join(G.protect) |
| return [ |
| (f"Game for this session (set by the user): {FOCUS['game']}" if FOCUS["game"] |
| else "Game: NOT SPECIFIED. Ask the user which game (or call find_games). They can also type /game NAME."), |
| (f"Protected by the user (never modify without asking): {prot}" if prot |
| else "Protected folders: NONE declared. Ask the user whether any folders are safe/protected, and tell them they can type /protect FOLDER."), |
| ] |
|
|
|
|
| def build_system(game_modules): |
| core = (SKILL_DIR / ("core-lean.md" if LEAN else "core.md")).read_text(encoding="utf-8") |
| today = datetime.date.today().isoformat() |
| mods = ", ".join(p.stem for p in sorted(MODULE_DIR.glob("*.md")) if not p.stem.startswith("_")) |
| parts = [core, f"\nToday's date: {today}. Workspace: {G.workspace}", |
| f"Allowed folders (general roots, games are found inside): {[str(a) for a in G.allow] or 'NONE: ask the user for their games root'}"] |
| parts += focus_lines() |
| parts.append("Modules: " + mods if LEAN else "\nAvailable modules:\n" + list_modules()) |
| for m in game_modules: |
| parts.append(f"\n--- MODULE: {m} ---\n" + load_module(m)) |
| return "\n".join(parts) |
|
|
|
|
| CFG = {"backend": "ollama", "base_url": "", "api_key": "", "think": "auto", |
| "thinking": "default", "effort": "", "fold_system": False} |
| THINK_RX = re.compile(r"<think>.*?(</think>|$)", re.S) |
|
|
|
|
| def chat_openai(messages, model, schemas, native): |
| """OpenAI-compatible chat completions (DeepSeek API and others).""" |
| out = [] |
| for m in messages: |
| m = dict(m) |
| if m["role"] == "tool": |
| m.pop("tool_name", None) |
| if m.get("tool_calls"): |
| fixed = [] |
| for i, tc in enumerate(m["tool_calls"]): |
| f = tc["function"] |
| a = f.get("arguments", {}) |
| fixed.append({"id": tc.get("id") or f"call_{i}", "type": "function", |
| "function": {"name": f["name"], "arguments": a if isinstance(a, str) else json.dumps(a)}}) |
| m["tool_calls"] = fixed |
| out.append(m) |
| payload = {"model": model, "messages": out, "stream": False} |
| if native: |
| payload["tools"] = schemas |
| if CFG["thinking"] in ("enabled", "disabled"): |
| payload["thinking"] = {"type": CFG["thinking"]} |
| if CFG["effort"]: |
| payload["reasoning_effort"] = CFG["effort"] |
| r = requests.post(f"{CFG['base_url']}/chat/completions", timeout=900, json=payload, |
| headers={"Authorization": f"Bearer {CFG['api_key']}", "Content-Type": "application/json"}) |
| if r.status_code >= 400: |
| hint = "" |
| if "reasoning_content" in r.text: |
| hint = " (thinking mode needs reasoning_content passed back on every assistant message; try --thinking off)" |
| raise RuntimeError(f"API error {r.status_code}: {r.text[:300]}{hint}") |
| return r.json()["choices"][0]["message"] |
|
|
|
|
| def chat(messages, model, schemas, num_ctx, native): |
| if CFG["backend"] != "ollama": |
| return chat_openai(messages, model, schemas, native) |
| payload = {"model": model, "messages": messages, "stream": False, |
| "options": {"num_ctx": num_ctx, "temperature": CFG.get("temperature", 0.3), |
| "num_predict": CFG.get("max_tokens", 4096)}} |
| if native: |
| payload["tools"] = schemas |
| if CFG["think"] in ("on", "off"): |
| payload["think"] = CFG["think"] == "on" |
| r = requests.post(f"{ollama_url()}/api/chat", json=payload, timeout=900) |
| if r.status_code == 400 and native and "tool" in r.text.lower(): |
| raise RuntimeError("This model does not support native tools. Re-run with --text-tools.") |
| if r.status_code == 400 and "think" in r.text.lower(): |
| raise RuntimeError("This model does not support the think option. Re-run with --think auto.") |
| r.raise_for_status() |
| return r.json()["message"] |
|
|
|
|
| def compact(messages, keep=8): |
| """Shrink old tool outputs so small context windows survive long jobs.""" |
| for m in messages[1:-keep]: |
| if m["role"] in ("tool", "user") and len(m.get("content", "")) > 600 and m.get("_tool"): |
| m["content"] = m["content"][:400] + "\n...[older tool output trimmed]" |
| return messages |
|
|
|
|
| BUDGET = {"ctx": 32768, "schemas": 0, "enabled": True} |
| TRIM_NOTE = "[Earlier steps were trimmed to save context. Re-read tracker.xml with read_file for the current state.]" |
|
|
|
|
| def est_tokens(msgs): |
| """Rough, deliberately pessimistic token estimate (about 3 characters per token).""" |
| return (sum(len(json.dumps(m, ensure_ascii=False)) for m in msgs)) // 3 + BUDGET["schemas"] |
|
|
|
|
| def fit(out): |
| """Keep the conversation inside ~75% of the context window (the rest is for the model's reply and thinking).""" |
| limit = int(BUDGET["ctx"] * 0.75) |
| |
| i = 1 |
| while est_tokens(out) > limit and i < len(out) - 2: |
| m = out[i] |
| is_result = m["role"] == "tool" or (m["role"] == "user" and m["content"].startswith("<tool_result")) |
| if is_result and len(m["content"]) > 300: |
| m["content"] = m["content"][:250] + "...[trimmed]" |
| i += 1 |
| |
| dropped = False |
| while est_tokens(out) > limit and len(out) > 5: |
| victim = out[2] |
| if victim.get("content") == TRIM_NOTE: |
| victim = out[3] |
| out.pop(3) |
| else: |
| out.pop(2) |
| while len(out) > 5 and out[2 if out[2].get("content") != TRIM_NOTE else 3]["role"] == "tool": |
| out.pop(2 if out[2].get("content") != TRIM_NOTE else 3) |
| dropped = True |
| if dropped and not any(m.get("content") == TRIM_NOTE for m in out): |
| out.insert(2, {"role": "user", "content": TRIM_NOTE}) |
| return out |
|
|
|
|
| def prep(messages): |
| """Clean internal keys, fit the context budget, and optionally fold the system prompt into the first user message.""" |
| out = [{k: v for k, v in m.items() if not k.startswith("_")} for m in compact(messages)] |
| if BUDGET["enabled"] and CFG["backend"] == "ollama": |
| out = fit(out) |
| if CFG.get("fold_system") and out and out[0]["role"] == "system": |
| sysmsg = out.pop(0)["content"] |
| for m in out: |
| if m["role"] == "user": |
| m["content"] = f"<instructions>\n{sysmsg}\n</instructions>\n\n" + m["content"] |
| break |
| return out |
|
|
|
|
| def agent_turn(messages, args, schemas): |
| native = not getattr(args, "text_tools", False) |
| for _ in range(args.max_steps): |
| msg = chat(prep(messages), args.model, schemas, args.num_ctx, native) |
| content = THINK_RX.sub("", msg.get("content") or "").strip() |
| tcs = msg.get("tool_calls") or [] |
| for i, tc in enumerate(tcs): |
| tc.setdefault("id", f"call_{len(messages)}_{i}") |
| entry = {"role": "assistant", "content": content} |
| if msg.get("reasoning_content") is not None: |
| entry["reasoning_content"] = msg["reasoning_content"] |
| if tcs and native: |
| entry["tool_calls"] = tcs |
| messages.append(entry) |
| if content: |
| print(f"\n{content}\n") |
| calls = [] |
| if native: |
| for tc in tcs: |
| fn = tc["function"] |
| calls.append((fn["name"], fn.get("arguments", {}), tc["id"])) |
| else: |
| for m in re.finditer(r"<tool_call>\s*(\{.*?\})\s*</tool_call>", content, re.S): |
| try: |
| j = json.loads(m.group(1)) |
| calls.append((j["name"], j.get("arguments", {}), None)) |
| except (json.JSONDecodeError, KeyError): |
| messages.append({"role": "user", "content": "<tool_result>Malformed tool_call JSON.</tool_result>", "_tool": True}) |
| if not calls: |
| if not content: |
| print("[the model returned no text. If it is a thinking model it probably spent its whole token budget thinking: " |
| "re-run with --think off, or --think on with a higher --max-tokens and --num-ctx.]") |
| return |
| for name, a, cid in calls: |
| print(f" [tool] {name}({json.dumps(a)[:160]})") |
| result = clip(run_tool(name, a)) |
| if native: |
| messages.append({"role": "tool", "tool_name": name, "tool_call_id": cid, "content": result, "_tool": True}) |
| else: |
| messages.append({"role": "user", "content": f"<tool_result name=\"{name}\">{result}</tool_result>", "_tool": True}) |
| print("[stopped: reached the step limit for this turn. Say 'continue' to keep going.]") |
|
|
|
|
| def selftest(args, schemas): |
| """Check every moving part: Ollama, the model, tool calling, web search/fetch, workspace, tracker.""" |
| results = [] |
|
|
| def check(name, fn): |
| try: |
| ok, detail = fn() |
| except Exception as e: |
| ok, detail = False, f"{type(e).__name__}: {e}" |
| results.append(ok) |
| print(f"[{'PASS' if ok else 'FAIL'}] {name}: {detail}") |
|
|
| def api_up(): |
| r = requests.get(f"{CFG['base_url']}/models", headers={"Authorization": f"Bearer {CFG['api_key']}"}, timeout=15) |
| return r.status_code == 200, f"{CFG['base_url']} answered HTTP {r.status_code} (model '{args.model}')" |
|
|
| def ollama_up(): |
| r = requests.get(f"{ollama_url()}/api/tags", timeout=10) |
| names = [m["name"] for m in r.json().get("models", [])] |
| hit = any(n == args.model or n.split(":")[0] == args.model for n in names) |
| return hit, f"server up; model '{args.model}' {'found' if hit else 'NOT found. Installed: ' + ', '.join(names[:8])}" |
|
|
| def native_tools(): |
| msg = chat([{"role": "system", "content": "You must use tools to look things up."}, |
| {"role": "user", "content": "Use web_search to look up the latest release of the BepInEx mod loader."}], |
| args.model, schemas, args.num_ctx, True) |
| calls = msg.get("tool_calls") or [] |
| names = [c["function"]["name"] for c in calls] |
| return bool(calls), f"model requested tools: {names}" if calls else \ |
| "no tool call came back (use a model tagged 'tools', add a TEMPLATE for a custom GGUF, or run with --text-tools)" |
|
|
| def search(): |
| out = web_search("BepInEx latest release GitHub", 3) |
| return ("http" in out and not out.startswith(("Search failed", "No results"))), out.splitlines()[0] if out else "empty" |
|
|
| def fetch_json(): |
| out = web_fetch("https://registry.npmjs.org/ms/latest") |
| return ('"name"' in out), out[:90].replace("\n", " ") |
|
|
| def fetch_html(): |
| out = web_fetch("https://github.com/BepInEx/BepInEx/releases") |
| return (len(out) > 200 and not out.startswith(("HTTP", "Fetch failed"))), out[:90].replace("\n", " ") |
|
|
| def workspace(): |
| p = G.workspace / "selftest.txt" |
| p.write_text("ok") |
| p.unlink() |
| return True, f"workspace writable: {G.workspace}" |
|
|
| def tracker(): |
| TRACKER.save() |
| return TRACKER.xml_path.exists() and TRACKER.html_path.exists(), f"open {TRACKER.html_path.as_uri()}" |
|
|
| check("Ollama server and model" if CFG["backend"] == "ollama" else "API reachable and key accepted", |
| ollama_up if CFG["backend"] == "ollama" else api_up) |
| check("Native tool calling", native_tools if not args.text_tools else (lambda: (True, "skipped (--text-tools)"))) |
| check("web_search (live)", search) |
| check("web_fetch JSON (npm registry API)", fetch_json) |
| check("web_fetch HTML (GitHub releases page)", fetch_html) |
| check("Texture tools (Pillow)", lambda: (_pil(), "Pillow available" if _pil() else "pip install pillow")) |
| check("Workspace", workspace) |
| check("Live tracker files", tracker) |
| print(f"\n{sum(results)}/{len(results)} checks passed.") |
| return all(results) |
|
|
|
|
| PRESET_MODELS = {"qwen38": "modding-qwen38", "qwen36": "modding-qwen36", "deepseek-r1": "modding-deepseek"} |
| _LOG = "C:/Games/MyGame/game.log" |
| EVAL_CASES = [ |
| ("Find the newest release of the BepInEx mod loader.", {"web_search", "web_fetch"}, None), |
| (f"Show me the last 40 lines of {_LOG}", {"read_file"}, lambda a: "game.log" in str(a.get("path", "")).lower()), |
| (f"In {_LOG}, find lines saying a dependency was not met.", {"grep_file"}, |
| lambda a: "game.log" in str(a.get("path", "")).lower() and re.search("depend|met", str(a.get("pattern", "")), re.I)), |
| ("Record in the tracker that SomeMod 1.0.250 is now downloaded.", {"tracker_mod"}, |
| lambda a: "somemod" in str(a.get("name", "")).lower() and str(a.get("status", "")).lower() == "downloaded"), |
| ("Download https://github.com/example/mod/releases/download/v1.0/mod.zip and save it as mod.zip.", {"download_file"}, |
| lambda a: a.get("url", "") == "https://github.com/example/mod/releases/download/v1.0/mod.zip"), |
| ("Which games are installed in my allowed folders?", {"find_games", "list_dir"}, None), |
| ("Before installing, check whether the staged mod in C:/Games/MyGame/stage/GameData clashes with what is installed in C:/Games/MyGame/GameData.", |
| {"deps_check"}, lambda a: "stage" in str(a.get("src", "")).lower() and "gamedata" in str(a.get("dest", "")).lower()), |
| ("Halve the resolution of every texture in C:/Games/MyGame/Textures. Check it first without writing anything.", {"texture_batch"}, |
| lambda a: float(a.get("scale", 0) or 0) == 0.5 and str(a.get("dry_run", True)).lower() in ("true", "1")), |
| ] |
|
|
|
|
| def compare(args, schemas): |
| """Quick head-to-head of local models on tool-call accuracy and speed, on THIS machine.""" |
| if CFG["backend"] != "ollama": |
| print("--compare only supports local Ollama models.") |
| return False |
| results = [] |
| for name in args.compare: |
| model = PRESET_MODELS.get(name, name) |
| passed, speeds, times, errors = 0, [], [], 0 |
| print(f"\n== {model} ==") |
| for prompt, expect, check in EVAL_CASES: |
| payload = {"model": model, "stream": False, "tools": schemas, |
| "messages": [{"role": "system", "content": "You are a game-modding agent. When a tool fits, call exactly one tool."}, |
| {"role": "user", "content": prompt}], |
| "options": {"num_ctx": args.num_ctx, "temperature": 0.2, "num_predict": 400}} |
| if args.think in ("on", "off"): |
| payload["think"] = args.think == "on" |
| t0 = time.time() |
| try: |
| r = requests.post(f"{ollama_url()}/api/chat", json=payload, timeout=900) |
| if r.status_code == 400 and "think" in r.text.lower(): |
| payload.pop("think", None) |
| r = requests.post(f"{ollama_url()}/api/chat", json=payload, timeout=900) |
| r.raise_for_status() |
| data = r.json() |
| except Exception as e: |
| errors += 1 |
| print(f" [ERR ] {prompt[:50]}...: {e}") |
| continue |
| times.append(time.time() - t0) |
| if data.get("eval_count") and data.get("eval_duration"): |
| speeds.append(data["eval_count"] / (data["eval_duration"] / 1e9)) |
| calls = (data.get("message") or {}).get("tool_calls") or [] |
| ok, got = False, "no tool call" |
| if calls: |
| fn = calls[0]["function"] |
| a = fn.get("arguments") or {} |
| if isinstance(a, str): |
| try: |
| a = json.loads(a) |
| except json.JSONDecodeError: |
| a = {} |
| got = fn["name"] |
| ok = fn["name"] in expect and (check is None or bool(check(a))) |
| passed += ok |
| print(f" [{'PASS' if ok else 'FAIL'}] {prompt[:56]:56s} -> {got}") |
| mean = lambda xs: sum(xs) / len(xs) if xs else 0.0 |
| results.append((model, passed, mean(speeds), mean(times), errors)) |
| print(f"\n{'model':28s} {'tool calls':>10s} {'tok/s':>8s} {'s/step':>8s}") |
| for model, passed, sp, t, errs in results: |
| print(f"{model:28s} {passed:>6d}/{len(EVAL_CASES)} {sp:>8.1f} {t:>8.1f}" + (f" ({errs} errors)" if errs else "")) |
| best = max(results, key=lambda x: (x[1], x[2])) |
| preset = next((k for k, v in PRESET_MODELS.items() if v == best[0]), None) |
| print(f"\nBest on this run: {best[0]}" + (f" -> python modding_agent.py --preset {preset}" if preset else "") |
| + "\n(Six quick checks: a screen for tool-use accuracy and speed, not a full benchmark. Run it twice if the top two are close.)") |
| return True |
|
|
|
|
| def load_config(): |
| p = HERE / "modding.config.json" |
| if p.exists(): |
| try: |
| return json.loads(p.read_text(encoding="utf-8")) |
| except Exception as e: |
| print(f"[warning] could not read {p.name}: {e}") |
| return {} |
|
|
|
|
| def _expand(path): |
| return os.path.expandvars(os.path.expanduser(path)) |
|
|
|
|
| HELP_TEXT = """Modding agent ready (any game). Tell me what to focus on: |
| /game NAME which game this session is for (any game; loads its module if I have one, else I research it) |
| /protect FOLDER a folder NAME or full PATH I must never modify without asking (repeat for more; none are protected by default) |
| /unprotect FOLDER stop protecting it /protected list what is protected |
| /allow FOLDER add a general folder I may read (and write with your OK) |
| /modules /load MODULE /save /help /quit |
| Same as command-line switches: --game NAME --protect FOLDER --allow FOLDER --session NAME (one session per game).""" |
|
|
|
|
| def main(): |
| global G, TRACKER, MAX_OUT, LEAN |
| ap = argparse.ArgumentParser(description="Modding agent for Ollama") |
| ap.add_argument("--backend", choices=["ollama", "deepseek", "openai"], default="ollama", |
| help="ollama (local), deepseek (DeepSeek API), openai (any OpenAI-compatible API)") |
| ap.add_argument("--model", default=None, help="model name (default: modding for ollama, deepseek-flash for deepseek)") |
| ap.add_argument("--base-url", default=None, help="API base URL for --backend openai (deepseek preset: https://api.deepseek.com)") |
| ap.add_argument("--api-key-env", default=None, help="env var holding the API key (deepseek preset: DEEPSEEK_API_KEY)") |
| ap.add_argument("--thinking", choices=["default", "on", "off"], default="default", help="API thinking mode (DeepSeek)") |
| ap.add_argument("--effort", choices=["low", "high", "max"], default=None, help="API reasoning effort (DeepSeek)") |
| ap.add_argument("--think", choices=["auto", "on", "off"], default=None, help="Ollama think option for thinking models (default auto; qwen presets: off)") |
| ap.add_argument("--preset", choices=["qwen38", "qwen36", "deepseek-r1"], default=None, |
| help="sensible defaults for a local model family (model name, thinking, sampling, tool mode)") |
| ap.add_argument("--compare", nargs="+", metavar="MODEL", default=None, |
| help="compare local models (names or presets, e.g. qwen38 qwen36) on tool-call accuracy and speed, then exit") |
| ap.add_argument("--fold-system", action="store_true", help="put the system prompt inside the first user message (advised for DeepSeek-R1)") |
| ap.add_argument("--temperature", type=float, default=None, help="Ollama temperature (qwen/R1 advice: 0.6)") |
| ap.add_argument("--game", action="append", default=[], |
| help="optional focus: the game for this session, ANY game by name (a matching module is loaded if one exists; " |
| "otherwise the agent researches it). Same as typing /game NAME. Repeatable") |
| ap.add_argument("--allow", action="append", default=[], |
| help="GENERAL folder the agent may read (and write with your OK), e.g. C:\\Games or a Steam library; repeatable. " |
| "Also from modding.config.json and the MODDING_ALLOW environment variable") |
| ap.add_argument("--protect", action="append", default=[], |
| help="optional: a folder NAME (anywhere) or full PATH the agent must never modify without asking. Nothing is protected " |
| "unless you say so. Same as typing /protect FOLDER. Repeatable") |
| ap.add_argument("--workspace", default=str(HERE / "workspace"), help="parent folder for per-session workspaces") |
| ap.add_argument("--session", default="default", help="session name: each session gets its own workspace, tracker and history (use one per game)") |
| ap.add_argument("--num-ctx", type=int, default=8192, help="context window in tokens (default 8192)") |
| ap.add_argument("--max-ctx", type=int, default=None, help="hard ceiling for --num-ctx (default 8192, or max_ctx in modding.config.json)") |
| ap.add_argument("--lean", choices=["auto", "on", "off"], default="auto", help="short prompt and tool set for small contexts (auto: under 12k)") |
| ap.add_argument("--max-steps", type=int, default=40) |
| ap.add_argument("--max-tokens", type=int, default=None, help="Ollama: cap on tokens per reply (default 4096, at most a third of the context)") |
| ap.add_argument("--text-tools", action="store_true", help="for models without native tool calling") |
| ap.add_argument("--selftest", action="store_true", help="test Ollama, tool calling and live web access, then exit") |
| ap.add_argument("--yes", action="store_true", help="skip confirmations (never applies to protected folders)") |
| args = ap.parse_args() |
| cfg = load_config() |
|
|
| if args.preset in ("qwen38", "qwen36"): |
| args.model = args.model or PRESET_MODELS[args.preset] |
| args.think = args.think or "off" |
| if args.temperature is None: |
| args.temperature = 0.6 |
| elif args.preset == "deepseek-r1": |
| args.model = args.model or PRESET_MODELS["deepseek-r1"] |
| args.text_tools = True |
| args.fold_system = True |
| if args.temperature is None: |
| args.temperature = 0.6 |
| args.think = args.think or "auto" |
| if args.backend == "deepseek": |
| args.model = args.model or "deepseek-flash" |
| args.base_url = args.base_url or "https://api.deepseek.com" |
| args.api_key_env = args.api_key_env or "DEEPSEEK_API_KEY" |
| elif args.backend == "openai": |
| if not args.base_url or not args.model: |
| sys.exit("--backend openai needs --base-url and --model") |
| args.api_key_env = args.api_key_env or "OPENAI_API_KEY" |
| else: |
| args.model = args.model or "modding" |
| key = os.environ.get(args.api_key_env or "", "") |
| if args.backend != "ollama" and not key: |
| sys.exit(f"Set your API key first, e.g. (PowerShell) $env:{args.api_key_env}='sk-...' or (bash) export {args.api_key_env}=sk-...") |
|
|
| |
| if args.backend == "ollama": |
| ceiling = args.max_ctx or int(cfg.get("max_ctx", 8192)) |
| if args.num_ctx > ceiling: |
| print(f"[context] --num-ctx {args.num_ctx} is above the ceiling {ceiling}; using {ceiling}. Raise --max-ctx to allow more.") |
| args.num_ctx = ceiling |
| LEAN = args.lean == "on" or (args.lean == "auto" and args.backend == "ollama" and args.num_ctx < 12000) |
| if args.max_tokens is None: |
| base = 12000 if args.think == "on" else 4096 |
| args.max_tokens = min(base, max(512, args.num_ctx // 3)) if args.backend == "ollama" else base |
| if args.backend == "ollama" and args.think == "on" and args.num_ctx <= 8192: |
| print("[warning] thinking on with an 8k window leaves little room for work; prefer --think off here.") |
|
|
| CFG.update(backend=args.backend, base_url=(args.base_url or "").rstrip("/"), api_key=key, think=args.think, |
| thinking={"on": "enabled", "off": "disabled", "default": "default"}[args.thinking], |
| effort=args.effort or "", fold_system=args.fold_system, max_tokens=args.max_tokens) |
| if args.temperature is not None: |
| CFG["temperature"] = args.temperature |
| if args.backend == "ollama": |
| BUDGET["ctx"] = args.num_ctx |
| MAX_OUT = max(1800, min(12000, int(args.num_ctx * 0.28))) |
| else: |
| BUDGET["enabled"] = False |
|
|
| |
| allow = [_expand(a) for a in cfg.get("allow", [])] |
| allow += [x for x in (_expand(a) for a in os.environ.get("MODDING_ALLOW", "").split(os.pathsep)) if x] |
| allow += [_expand(a) for a in args.allow] |
| protect = list(dict.fromkeys(list(cfg.get("protect", [])) + args.protect)) |
| safe = re.sub(r"[^A-Za-z0-9_.-]", "_", args.session) or "default" |
| ws = Path(args.workspace) / safe |
| G = Guard(ws, list(dict.fromkeys(allow)), protect, args.yes) |
| TRACKER = Tracker(G.workspace) |
| for pf in protect: |
| if pf not in TRACKER.d["protected"]: |
| TRACKER.d["protected"].append(pf) |
| TRACKER.save() |
| schemas = tool_schemas() |
| if args.selftest: |
| sys.exit(0 if selftest(args, schemas) else 1) |
| if args.compare: |
| sys.exit(0 if compare(args, schemas) else 1) |
|
|
| FOCUS["game"] = ", ".join(args.game) |
| preload = list(dict.fromkeys(x for x in (module_for_game(g) for g in args.game) if x)) |
| for g in args.game: |
| if not module_for_game(g): |
| print(f"[game] no module for '{g}' yet: the agent will research it online (and can save notes).") |
| if args.game: |
| TRACKER.d["game"] = FOCUS["game"] |
| TRACKER.save() |
|
|
| def make_system(): |
| return build_system(preload) + (text_protocol() if args.text_tools else "") |
|
|
| sess_path = G.workspace / "session.json" |
| system = make_system() |
| messages = [{"role": "system", "content": system}] |
| BUDGET["schemas"] = 0 if args.text_tools else len(json.dumps(schemas)) // 3 |
| if args.backend == "ollama": |
| fixed = len(system) // 3 + BUDGET["schemas"] |
| pct = 100 * fixed / args.num_ctx |
| print(f"Context: num_ctx={args.num_ctx} ({'lean prompt' if LEAN else 'full prompt'}); fixed prompt (instructions + tools) " |
| f"is about {fixed} tokens ({pct:.0f}%); tool outputs are capped at {MAX_OUT} characters; replies at {args.max_tokens} tokens.") |
| if pct > 45: |
| print("[warning] the fixed prompt uses a lot of this context window: preload fewer modules, or raise --max-ctx and --num-ctx.") |
| if not G.allow: |
| print("[warning] no allowed folders yet. Give a general games root, e.g. --allow C:\\Games (or /allow FOLDER, MODDING_ALLOW, modding.config.json).") |
| else: |
| print("Allowed folders: " + "; ".join(str(a) for a in G.allow)) |
| print("Protected folders: " + (", ".join(G.protect) if G.protect else "NONE (nothing is protected unless you say so)")) |
| if args.yes and not G.protect: |
| print("[warning] --yes with no protected folders: every write inside allowed folders is auto-approved. Add --protect FOLDER first if anything must stay untouched.") |
| if sess_path.exists(): |
| saved = json.loads(sess_path.read_text(encoding="utf-8")) |
| messages += [m for m in saved if m["role"] != "system"] |
| print(f"[resumed session '{args.session}' with {len(saved)} messages]") |
| if args.backend == "ollama": |
| print(f"Model: {args.model} think={args.think} tools={'text' if args.text_tools else 'native'}" |
| + (f" preset={args.preset}" if args.preset else "")) |
| if args.think == "auto" and re.search(r"qwen3|deepseek-r1|gpt-oss", args.model, re.I): |
| print("[warning] this model family thinks by default, and thinking tokens eat both the reply cap and the context window. " |
| "Use --think off (recommended for tool loops).") |
| print(HELP_TEXT) |
| print(f"Session workspace: {G.workspace}") |
| print(f"Live tracker (auto-refreshes every 3s): {TRACKER.html_path.as_uri()}") |
| print(f"Tracker data (XML): {TRACKER.xml_path}") |
| if DDGS is None: |
| print("[warning] web_search needs: pip install ddgs") |
|
|
| while True: |
| try: |
| user = input("you> ").strip() |
| except (EOFError, KeyboardInterrupt): |
| break |
| if not user: |
| continue |
| if user == "/quit": |
| break |
| if user == "/help": |
| print(HELP_TEXT) |
| continue |
| if user == "/modules": |
| print(list_modules()) |
| continue |
| if user == "/protected": |
| print("Protected: " + (", ".join(G.protect) if G.protect else "NONE")) |
| continue |
| cmd, _, rest = user.partition(" ") |
| rest = rest.strip() |
| if cmd in ("/game", "/protect", "/unprotect", "/allow") and not rest: |
| print(f"usage: {cmd} <{'folder' if cmd != '/game' else 'game name'}>") |
| continue |
| if cmd == "/game": |
| FOCUS["game"] = rest |
| mod = module_for_game(rest) |
| preload = [mod] if mod else [] |
| TRACKER.d["game"] = rest |
| TRACKER.event(f"GAME set by user: {rest}") |
| TRACKER.save() |
| messages[0]["content"] = make_system() |
| print(f"[game: {rest}; " + (f"module '{mod}' loaded]" if mod else "no module yet: the agent will research it online]")) |
| continue |
| if cmd == "/protect": |
| if G.add_protect(rest): |
| TRACKER.d["protected"].append(rest) |
| TRACKER.event(f"PROTECTED by user: {rest}") |
| TRACKER.save() |
| messages[0]["content"] = make_system() |
| print("Protected: " + ", ".join(G.protect)) |
| continue |
| if cmd == "/unprotect": |
| if G.remove_protect(rest) and rest in TRACKER.d["protected"]: |
| TRACKER.d["protected"].remove(rest) |
| TRACKER.save() |
| messages[0]["content"] = make_system() |
| print("Protected: " + (", ".join(G.protect) or "NONE")) |
| continue |
| if cmd == "/allow": |
| pth = Path(_expand(rest)).resolve() |
| if pth.is_dir() and pth not in G.allow: |
| G.allow.append(pth) |
| messages[0]["content"] = make_system() |
| print("Allowed folders: " + "; ".join(str(a) for a in G.allow) if pth.is_dir() else f"Not a folder: {pth}") |
| continue |
| if user.startswith("/load "): |
| messages.append({"role": "user", "content": "Module loaded for you:\n" + load_module(user[6:].strip())}) |
| print("[module added to the conversation]") |
| continue |
| if user == "/save": |
| pass |
| else: |
| messages.append({"role": "user", "content": user}) |
| try: |
| agent_turn(messages, args, schemas) |
| except Exception as e: |
| print(f"[error] {e}") |
| sess_path.parent.mkdir(parents=True, exist_ok=True) |
| sess_path.write_text(json.dumps([m for m in messages if m["role"] != "system"], indent=1), encoding="utf-8") |
| print("bye") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
