Hugging Face's logo

Buckets:
Elmerfudge
/
ModdingAssistant-bucket

Files
xet
Elmerfudge/ModdingAssistant-bucket / modding_agent.py
Elmerfudge's picture
Elmerfudge
download
raw
103 kB
#!/usr/bin/env python3
"""
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 # max characters returned from any tool
LEAN = False # lean prompt/tool set for small context windows (set in main)
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"}} # still accepted, just not advertised
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
# --------------------------------------------------------------------------- #
# Safety: where the agent may read/write
# --------------------------------------------------------------------------- #
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 = [] # what the user declared (display); NOTHING is protected unless the user says so
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 # still ask below
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 # set in main()
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]"
# --------------------------------------------------------------------------- #
# Network helpers
# --------------------------------------------------------------------------- #
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 "")
# --------------------------------------------------------------------------- #
# Tools
# --------------------------------------------------------------------------- #
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."
# --------------------------------------------------------------------------- #
# Dependency chain protection: never let one mod's bundled copy of a dependency
# overwrite (or mix with) a different version that is already installed.
# --------------------------------------------------------------------------- #
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)) # swap the WHOLE unit so two versions are never mixed
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))
# --------------------------------------------------------------------------- #
# Live tracker: tracker.xml (source of truth) + tracker.html (auto-refreshing view)
# --------------------------------------------------------------------------- #
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"])} &middot; Method: {e(d["method"])} &middot; Started: {e(d["started"])} &middot; 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."
# --------------------------------------------------------------------------- #
# Loose texture editing: mass operations with per-file rules, dry run, backups, undo
# --------------------------------------------------------------------------- #
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} # PIL constants, resolved lazily
COLOR_OPS = ("brightness", "contrast", "saturation", "gamma", "hue_shift", "tint", "levels", "channel_gain",
"grayscale", "invert", "sharpen", "blur")
def _pil():
try:
from PIL import Image # noqa: F401
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")
# --- geometry (always applies, even to data maps)
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)
# --- colour operations
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}" # unique even for several jobs in the same second
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 = [ # (module, label, folder-name keywords, signature files/folders); most specific first
("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 # noqa: F401
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: # keep the loop alive
return f"{name} failed: {e}"
# --------------------------------------------------------------------------- #
# Model loop
# --------------------------------------------------------------------------- #
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)
# 1) shrink older tool results first
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
# 2) then drop the oldest steps (keep the system prompt, the first user message and the last 3 messages)
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) # never leave orphan tool results
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() # never feed <think> text back
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: # DeepSeek thinking mode requires this to be sent back
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 = [ # (prompt, acceptable first tool names, optional check on the arguments)
("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" # Qwen3.x thinks by default and can burn thousands of tokens per step
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-...")
# context restriction (local models)
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))) # smaller windows get smaller tool outputs
else:
BUDGET["enabled"] = False
# general folders: config file + environment + command line
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)) # nothing is protected unless the user says so
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()

Xet Storage Details

Size:
103 kB
·
Xet hash:
b2b69f343c3606af0ccdab7e080ca19f63be2ecb5c57dbb25ee2460fd8dc8a32

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.