fix(windows): sweep remaining bare read_text/write_text sites + linter rule

AST-driven pass over every Path.read_text()/write_text() without an
explicit encoding= across non-test code: 71 sites in 34 files
(skills_hub, hermes_cli/main+profiles+service_manager+container_boot,
mem0/hindsight/honcho plugins, achievements dashboard, release/CI
scripts, productivity+comfyui skill helpers, agent/*). Verified zero
positional-encoding collisions before insertion; per-file compile()
check after.

Adds a check-windows-footguns rule flagging bare single-line
read_text/write_text (multi-line forms stay covered by the AST guard
test from #38985). Together with the salvaged contributor commits this
retires the ~169-site bare file-I/O class (#37423's long tail).
This commit is contained in:
teknium1
2026-07-24 17:10:39 -07:00
committed by Teknium
parent adecb0d1a9
commit 75e0d52034
36 changed files with 103 additions and 75 deletions
+1 -1
View File
@@ -164,7 +164,7 @@ def _remove_env_source(provider: str, removed) -> RemovalResult:
if env_path.exists(): if env_path.exists():
env_in_dotenv = any( env_in_dotenv = any(
line.strip().startswith(f"{env_var}=") line.strip().startswith(f"{env_var}=")
for line in env_path.read_text(errors="replace").splitlines() for line in env_path.read_text(errors="replace", encoding="utf-8").splitlines()
) )
except OSError: except OSError:
pass pass
+4 -4
View File
@@ -3066,7 +3066,7 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
if _refuse_temp_home_service_write(new_unit, "systemd unit"): if _refuse_temp_home_service_write(new_unit, "systemd unit"):
return False return False
unit_path.write_text(new_unit, encoding="utf-8", encoding="utf-8") unit_path.write_text(new_unit, encoding="utf-8")
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30) _run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
print( print(
f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install" f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install"
@@ -3248,7 +3248,7 @@ def systemd_install(
if _refuse_temp_home_service_write(new_unit, "systemd unit"): if _refuse_temp_home_service_write(new_unit, "systemd unit"):
return return
print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}") print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}")
unit_path.write_text(new_unit, encoding="utf-8", encoding="utf-8") unit_path.write_text(new_unit, encoding="utf-8")
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30) _run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
if enable_on_startup: if enable_on_startup:
@@ -4082,7 +4082,7 @@ def refresh_launchd_plist_if_needed() -> bool:
if _refuse_temp_home_service_write(new_plist, "launchd plist"): if _refuse_temp_home_service_write(new_plist, "launchd plist"):
return False return False
plist_path.write_text(new_plist, encoding="utf-8", encoding="utf-8") plist_path.write_text(new_plist, encoding="utf-8")
label = get_launchd_label() label = get_launchd_label()
domain = _launchd_domain() domain = _launchd_domain()
target = f"{domain}/{label}" target = f"{domain}/{label}"
@@ -4301,7 +4301,7 @@ def launchd_start():
sys.exit(1) sys.exit(1)
print("↻ launchd plist missing; regenerating service definition") print("↻ launchd plist missing; regenerating service definition")
plist_path.parent.mkdir(parents=True, exist_ok=True) plist_path.parent.mkdir(parents=True, exist_ok=True)
plist_path.write_text(new_plist, encoding="utf-8", encoding="utf-8") plist_path.write_text(new_plist, encoding="utf-8")
try: try:
_launchctl_bootstrap(_launchd_domain(), plist_path, label, timeout=30) _launchctl_bootstrap(_launchd_domain(), plist_path, label, timeout=30)
subprocess.run( subprocess.run(
+2 -2
View File
@@ -9959,7 +9959,7 @@ def _ensure_fhs_path_guard() -> None:
if not cfg.is_file(): if not cfg.is_file():
continue continue
try: try:
existing = cfg.read_text(errors="replace") existing = cfg.read_text(errors="replace", encoding="utf-8")
except OSError: except OSError:
continue continue
# Idempotency: skip if any uncommented PATH= line already references # Idempotency: skip if any uncommented PATH= line already references
@@ -12535,7 +12535,7 @@ def _cmd_update_impl(args, gateway_mode: bool):
if gateway_mode: if gateway_mode:
_exit_code_path = get_hermes_home() / ".update_exit_code" _exit_code_path = get_hermes_home() / ".update_exit_code"
try: try:
_exit_code_path.write_text("1") _exit_code_path.write_text("1", encoding="utf-8")
except OSError: except OSError:
pass pass
_warn_incomplete_gateway_fleet_restart(failed_or_stale_units) _warn_incomplete_gateway_fleet_restart(failed_or_stale_units)
+1 -1
View File
@@ -1601,7 +1601,7 @@ class PluginManager:
init_file = plugin_dir / "__init__.py" init_file = plugin_dir / "__init__.py"
if init_file.exists(): if init_file.exists():
try: try:
source_text = init_file.read_text(errors="replace")[:8192] source_text = init_file.read_text(errors="replace", encoding="utf-8")[:8192]
if ( if (
"register_memory_provider" in source_text "register_memory_provider" in source_text
or "MemoryProvider" in source_text or "MemoryProvider" in source_text
+1 -1
View File
@@ -78,7 +78,7 @@ def _warn_profile_fallback_once() -> None:
try: try:
fallback_home = _get_platform_default_hermes_home() fallback_home = _get_platform_default_hermes_home()
active_path = fallback_home / "active_profile" active_path = fallback_home / "active_profile"
active = active_path.read_text().strip() if active_path.exists() else "" active = active_path.read_text(encoding="utf-8").strip() if active_path.exists() else ""
except (UnicodeDecodeError, OSError): except (UnicodeDecodeError, OSError):
active = "" active = ""
if active and active != "default": if active and active != "default":
@@ -68,7 +68,7 @@ ASSETS_DIR = Path(__file__).resolve().parent.parent / "assets"
def load_template(name: str) -> str: def load_template(name: str) -> str:
return (ASSETS_DIR / name).read_text() return (ASSETS_DIR / name).read_text(encoding="utf-8")
PROFILE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") PROFILE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
@@ -471,7 +471,7 @@ def main():
help="Write TEAM.md alongside (default: skipped)") help="Write TEAM.md alongside (default: skipped)")
args = ap.parse_args() args = ap.parse_args()
plan = json.loads(Path(args.plan_json).read_text()) plan = json.loads(Path(args.plan_json).read_text(encoding="utf-8"))
errors = validate_plan(plan) errors = validate_plan(plan)
if errors: if errors:
print("Plan validation failed:", file=sys.stderr) print("Plan validation failed:", file=sys.stderr)
@@ -483,15 +483,15 @@ def main():
team = render_team_md(plan) team = render_team_md(plan)
setup = render_setup_sh(plan, brief, team) setup = render_setup_sh(plan, brief, team)
Path(args.out).write_text(setup) Path(args.out).write_text(setup, encoding="utf-8")
os.chmod(args.out, 0o755) os.chmod(args.out, 0o755)
print(f"Wrote {args.out}") print(f"Wrote {args.out}")
if args.brief_out: if args.brief_out:
Path(args.brief_out).write_text(brief) Path(args.brief_out).write_text(brief, encoding="utf-8")
print(f"Wrote {args.brief_out}") print(f"Wrote {args.brief_out}")
if args.team_out: if args.team_out:
Path(args.team_out).write_text(team) Path(args.team_out).write_text(team, encoding="utf-8")
print(f"Wrote {args.team_out}") print(f"Wrote {args.team_out}")
@@ -141,7 +141,7 @@ def build_findings(
# 3. Timing-based findings. # 3. Timing-based findings.
if timing_path and Path(timing_path).exists(): if timing_path and Path(timing_path).exists():
timing = json.loads(Path(timing_path).read_text()) timing = json.loads(Path(timing_path).read_text(encoding="utf-8"))
for r in timing.get("results", []): for r in timing.get("results", []):
if not r.get("significant"): if not r.get("significant"):
continue continue
@@ -190,7 +190,7 @@ def build_findings(
}, },
"findings": findings, "findings": findings,
} }
Path(out_path).write_text(json.dumps(payload, indent=2)) Path(out_path).write_text(json.dumps(payload, indent=2), encoding="utf-8")
return payload return payload
@@ -198,7 +198,7 @@ def analyze(
"results": results, "results": results,
} }
Path(out_path).write_text(json.dumps(payload, indent=2)) Path(out_path).write_text(json.dumps(payload, indent=2), encoding="utf-8")
return payload return payload
+1 -1
View File
@@ -86,7 +86,7 @@ def _is_cron_provider_dir(path: Path) -> bool:
if not init_file.exists(): if not init_file.exists():
return False return False
try: try:
source = init_file.read_text(errors="replace")[:8192] source = init_file.read_text(errors="replace", encoding="utf-8")[:8192]
return "register_cron_scheduler" in source or "CronScheduler" in source return "register_cron_scheduler" in source or "CronScheduler" in source
except Exception: except Exception:
return False return False
+3 -3
View File
@@ -110,12 +110,12 @@ def load_tracked() -> List[Dict[str, Any]]:
return [] return []
try: try:
return json.loads(tf.read_text()) return json.loads(tf.read_text(encoding="utf-8"))
except (json.JSONDecodeError, ValueError): except (json.JSONDecodeError, ValueError):
bak = tf.with_suffix(".json.bak") bak = tf.with_suffix(".json.bak")
if bak.exists(): if bak.exists():
try: try:
data = json.loads(bak.read_text()) data = json.loads(bak.read_text(encoding="utf-8"))
_log("WARN: tracked.json corrupted — restored from .bak") _log("WARN: tracked.json corrupted — restored from .bak")
return data return data
except Exception: except Exception:
@@ -129,7 +129,7 @@ def save_tracked(tracked: List[Dict[str, Any]]) -> None:
tf = get_tracked_file() tf = get_tracked_file()
tf.parent.mkdir(parents=True, exist_ok=True) tf.parent.mkdir(parents=True, exist_ok=True)
tmp = tf.with_suffix(".json.tmp") tmp = tf.with_suffix(".json.tmp")
tmp.write_text(json.dumps(tracked, indent=2)) tmp.write_text(json.dumps(tracked, indent=2), encoding="utf-8")
if tf.exists(): if tf.exists():
shutil.copy2(tf, tf.with_suffix(".json.bak")) shutil.copy2(tf, tf.with_suffix(".json.bak"))
tmp.replace(tf) tmp.replace(tf)
@@ -262,7 +262,7 @@ class RealtimeSpeaker:
if not self.queue_path.exists(): if not self.queue_path.exists():
return [] return []
out: list[dict] = [] out: list[dict] = []
for line in self.queue_path.read_text().splitlines(): for line in self.queue_path.read_text(encoding="utf-8").splitlines():
line = line.strip() line = line.strip()
if not line: if not line:
continue continue
@@ -281,10 +281,10 @@ class RealtimeSpeaker:
if not remaining: if not remaining:
# Keep the file but empty — consumers may be watching for # Keep the file but empty — consumers may be watching for
# new writes via mtime, and delete-then-recreate is a race. # new writes via mtime, and delete-then-recreate is a race.
self.queue_path.write_text("") self.queue_path.write_text("", encoding="utf-8")
return return
self.queue_path.write_text( self.queue_path.write_text(
"\n".join(json.dumps(e) for e in remaining) + "\n" "\n".join(json.dumps(e) for e in remaining) + "\n", encoding="utf-8"
) )
def _append_processed(self, entry: dict, result: dict) -> None: def _append_processed(self, entry: dict, result: dict) -> None:
@@ -159,7 +159,7 @@ def load_state() -> Dict[str, Any]:
if not path.exists(): if not path.exists():
return {"unlocks": {}} return {"unlocks": {}}
try: try:
return json.loads(path.read_text()) return json.loads(path.read_text(encoding="utf-8"))
except Exception: except Exception:
return {"unlocks": {}} return {"unlocks": {}}
@@ -167,7 +167,7 @@ def load_state() -> Dict[str, Any]:
def save_state(state: Dict[str, Any]) -> None: def save_state(state: Dict[str, Any]) -> None:
path = state_path() path = state_path()
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(state, indent=2, sort_keys=True)) path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8")
def _json_safe(value: Any) -> Any: def _json_safe(value: Any) -> Any:
@@ -185,7 +185,7 @@ def load_snapshot() -> Optional[Dict[str, Any]]:
if not path.exists(): if not path.exists():
return None return None
try: try:
data = json.loads(path.read_text()) data = json.loads(path.read_text(encoding="utf-8"))
if isinstance(data, dict): if isinstance(data, dict):
return data return data
except Exception: except Exception:
@@ -196,7 +196,7 @@ def load_snapshot() -> Optional[Dict[str, Any]]:
def save_snapshot(data: Dict[str, Any]) -> None: def save_snapshot(data: Dict[str, Any]) -> None:
path = snapshot_path() path = snapshot_path()
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(_json_safe(data), indent=2, sort_keys=True)) path.write_text(json.dumps(_json_safe(data), indent=2, sort_keys=True), encoding="utf-8")
def load_checkpoint() -> Dict[str, Any]: def load_checkpoint() -> Dict[str, Any]:
@@ -204,7 +204,7 @@ def load_checkpoint() -> Dict[str, Any]:
if not path.exists(): if not path.exists():
return {"schema_version": 1, "generated_at": 0, "sessions": {}} return {"schema_version": 1, "generated_at": 0, "sessions": {}}
try: try:
data = json.loads(path.read_text()) data = json.loads(path.read_text(encoding="utf-8"))
if isinstance(data, dict): if isinstance(data, dict):
data.setdefault("schema_version", 1) data.setdefault("schema_version", 1)
data.setdefault("generated_at", 0) data.setdefault("generated_at", 0)
@@ -219,7 +219,7 @@ def load_checkpoint() -> Dict[str, Any]:
def save_checkpoint(data: Dict[str, Any]) -> None: def save_checkpoint(data: Dict[str, Any]) -> None:
path = checkpoint_path() path = checkpoint_path()
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(_json_safe(data), indent=2, sort_keys=True)) path.write_text(json.dumps(_json_safe(data), indent=2, sort_keys=True), encoding="utf-8")
def session_fingerprint(meta: Dict[str, Any]) -> Dict[str, Any]: def session_fingerprint(meta: Dict[str, Any]) -> Dict[str, Any]:
+1 -1
View File
@@ -81,7 +81,7 @@ def _is_memory_provider_dir(path: Path) -> bool:
if not init_file.exists(): if not init_file.exists():
return False return False
try: try:
source = init_file.read_text(errors="replace")[:8192] source = init_file.read_text(errors="replace", encoding="utf-8")[:8192]
return "register_memory_provider" in source or "MemoryProvider" in source return "register_memory_provider" in source or "MemoryProvider" in source
except Exception: except Exception:
return False return False
+3 -3
View File
@@ -228,7 +228,7 @@ def _save_mem0_json(hermes_home: str, data: dict) -> None:
except Exception: except Exception:
pass pass
existing.update(data) existing.update(data)
config_path.write_text(json.dumps(existing, indent=2) + "\n") config_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
def _setup_platform(hermes_home: str, config: dict, flags: dict[str, str]) -> None: def _setup_platform(hermes_home: str, config: dict, flags: dict[str, str]) -> None:
@@ -248,7 +248,7 @@ def _setup_platform(hermes_home: str, config: dict, flags: dict[str, str]) -> No
config_path = Path(hermes_home) / "mem0.json" config_path = Path(hermes_home) / "mem0.json"
if config_path.exists(): if config_path.exists():
try: try:
existing_config = json.loads(config_path.read_text()) existing_config = json.loads(config_path.read_text(encoding="utf-8"))
except Exception: except Exception:
pass pass
@@ -369,7 +369,7 @@ def _setup_selfhosted(hermes_home: str, config: dict, flags: dict[str, str]) ->
config_path = Path(hermes_home) / "mem0.json" config_path = Path(hermes_home) / "mem0.json"
if config_path.exists(): if config_path.exists():
try: try:
existing_config = json.loads(config_path.read_text()) existing_config = json.loads(config_path.read_text(encoding="utf-8"))
except Exception: except Exception:
pass pass
+28
View File
@@ -373,6 +373,34 @@ FOOTGUNS: list[Footgun] = [
and _is_likely_subprocess_call(line) and _is_likely_subprocess_call(line)
), ),
), ),
Footgun(
name="bare Path.read_text()/write_text() without encoding=",
# Match ``.read_text(`` / ``.write_text(`` when the same line does
# not pass ``encoding=``. Multi-line calls where encoding= sits on
# a later line are handled by the post_filter's lookahead-free
# heuristic accepting a small false-negative rate — the AST guard
# test in tests/gateway/test_gateway_utf8_encoding.py catches the
# gateway/adapters exactly, and this rule catches the common
# single-line form everywhere else.
pattern=re.compile(r"\.(read_text|write_text)\s*\("),
message=(
"Path.read_text()/write_text() without encoding= uses "
"locale.getpreferredencoding() — cp936/cp1252 on Windows — "
"so UTF-8 content (config JSON, session state, skills) "
"crashes with UnicodeDecodeError or writes mojibake. "
"See issue #37423 and the #71014 / read_text campaign."
),
fix='path.read_text(encoding="utf-8") / path.write_text(data, encoding="utf-8")',
post_filter=lambda m, line: (
"encoding=" not in line
and "encoding =" not in line
and not _looks_like_string_literal(line, m)
# Skip calls that continue onto the next line — the closing
# paren isn't on this line, so encoding= may follow. AST-level
# enforcement for those lives in the gateway guard test.
and line.rstrip().endswith(")")
),
),
] ]
+2 -2
View File
@@ -181,7 +181,7 @@ def main() -> int:
if any(skip.rstrip("/") in parts for skip in SKIP_DIRS): if any(skip.rstrip("/") in parts for skip in SKIP_DIRS):
continue continue
content = py_file.read_text() content = py_file.read_text(encoding="utf-8")
violations = find_subprocess_calls(content, rel) violations = find_subprocess_calls(content, rel)
all_violations.extend(violations) all_violations.extend(violations)
@@ -205,7 +205,7 @@ def main() -> int:
continue continue
try: try:
content = py_file.read_text() content = py_file.read_text(encoding="utf-8")
except Exception: except Exception:
continue continue
violations = find_subprocess_calls(content, rel) violations = find_subprocess_calls(content, rel)
+1 -1
View File
@@ -419,7 +419,7 @@ def main() -> int:
pending_jobs=pending, pending_jobs=pending,
) )
args.output.write_text(body) args.output.write_text(body, encoding="utf-8")
print(f"Wrote {len(body)} chars to {args.output}") print(f"Wrote {len(body)} chars to {args.output}")
return 0 return 0
+2 -2
View File
@@ -30,7 +30,7 @@ def _load_json(path: Path | None) -> list[dict]:
if path is None or not path.exists() or path.stat().st_size == 0: if path is None or not path.exists() or path.stat().st_size == 0:
return [] return []
try: try:
data = json.loads(path.read_text()) data = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc: except json.JSONDecodeError as exc:
print(f"warning: could not parse {path}: {exc}", file=sys.stderr) print(f"warning: could not parse {path}: {exc}", file=sys.stderr)
return [] return []
@@ -197,7 +197,7 @@ def main() -> int:
summary = "\n".join(buf) summary = "\n".join(buf)
if args.output: if args.output:
args.output.write_text(summary) args.output.write_text(summary, encoding="utf-8")
else: else:
print(summary) print(summary)
return 0 return 0
+1 -1
View File
@@ -516,7 +516,7 @@ def main() -> int:
if not path.exists(): if not path.exists():
print(f"\n⚠ no baseline at {path} — run with --save {args.compare} first") print(f"\n⚠ no baseline at {path} — run with --save {args.compare} first")
else: else:
before = json.loads(path.read_text()) before = json.loads(path.read_text(encoding="utf-8"))
print(f"\n═══ A/B diff vs /tmp/perf-{args.compare}.json ═══") print(f"\n═══ A/B diff vs /tmp/perf-{args.compare}.json ═══")
print(format_diff(before, metrics)) print(format_diff(before, metrics))
+5 -5
View File
@@ -2138,7 +2138,7 @@ def next_available_tag(base_tag: str) -> tuple[str, str]:
def get_current_version(): def get_current_version():
"""Read current semver from __init__.py.""" """Read current semver from __init__.py."""
content = VERSION_FILE.read_text() content = VERSION_FILE.read_text(encoding="utf-8")
match = re.search(r'__version__\s*=\s*"([^"]+)"', content) match = re.search(r'__version__\s*=\s*"([^"]+)"', content)
return match.group(1) if match else "0.0.0" return match.group(1) if match else "0.0.0"
@@ -2168,7 +2168,7 @@ def bump_version(current: str, part: str) -> str:
def update_version_files(semver: str, calver_date: str): def update_version_files(semver: str, calver_date: str):
"""Update version strings in source files.""" """Update version strings in source files."""
# Update __init__.py # Update __init__.py
content = VERSION_FILE.read_text() content = VERSION_FILE.read_text(encoding="utf-8")
content = re.sub( content = re.sub(
r'__version__\s*=\s*"[^"]+"', r'__version__\s*=\s*"[^"]+"',
f'__version__ = "{semver}"', f'__version__ = "{semver}"',
@@ -2179,17 +2179,17 @@ def update_version_files(semver: str, calver_date: str):
f'__release_date__ = "{calver_date}"', f'__release_date__ = "{calver_date}"',
content, content,
) )
VERSION_FILE.write_text(content) VERSION_FILE.write_text(content, encoding="utf-8")
# Update pyproject.toml # Update pyproject.toml
pyproject = PYPROJECT_FILE.read_text() pyproject = PYPROJECT_FILE.read_text(encoding="utf-8")
pyproject = re.sub( pyproject = re.sub(
r'^version\s*=\s*"[^"]+"', r'^version\s*=\s*"[^"]+"',
f'version = "{semver}"', f'version = "{semver}"',
pyproject, pyproject,
flags=re.MULTILINE, flags=re.MULTILINE,
) )
PYPROJECT_FILE.write_text(pyproject) PYPROJECT_FILE.write_text(pyproject, encoding="utf-8")
# Keep the desktop Electron app's package.json version in lockstep with the # Keep the desktop Electron app's package.json version in lockstep with the
# Python package version. The desktop About panel reads the live Hermes # Python package version. The desktop About panel reads the live Hermes
+2 -2
View File
@@ -533,7 +533,7 @@ def _load_durations(repo_root: Path) -> dict[str, float]:
if not path.is_file(): if not path.is_file():
return {} return {}
try: try:
return json.loads(path.read_text()) return json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e: except (json.JSONDecodeError, OSError) as e:
print("[ERROR] Failed to load json durations file! {e}") print("[ERROR] Failed to load json durations file! {e}")
return {} return {}
@@ -555,7 +555,7 @@ def _save_durations(
key = _format_file(f, repo_root) key = _format_file(f, repo_root)
data[key] = round(t, 3) data[key] = round(t, 3)
path = repo_root / _DURATIONS_FILE path = repo_root / _DURATIONS_FILE
path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n") path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def _compute_lpt_slices( def _compute_lpt_slices(
@@ -158,7 +158,7 @@ def main(argv: list[str] | None = None) -> int:
sources: dict[str, str] = {} sources: dict[str, str] = {}
if args.models_from_file: if args.models_from_file:
try: try:
sources = json.loads(Path(args.models_from_file).read_text()) sources = json.loads(Path(args.models_from_file).read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as e: except (OSError, json.JSONDecodeError) as e:
log(f"Could not read --models-from-file: {e}") log(f"Could not read --models-from-file: {e}")
@@ -303,7 +303,7 @@ def main(argv: list[str] | None = None) -> int:
out = json.dumps(schema, indent=2, default=str) out = json.dumps(schema, indent=2, default=str)
if args.output: if args.output:
Path(args.output).write_text(out) Path(args.output).write_text(out, encoding="utf-8")
print(f"Schema written to {args.output}", file=sys.stderr) print(f"Schema written to {args.output}", file=sys.stderr)
else: else:
print(out) print(out)
@@ -620,7 +620,7 @@ def main(argv: list[str] | None = None) -> int:
args_str = args.args args_str = args.args
if args_str.startswith("@"): if args_str.startswith("@"):
try: try:
args_str = Path(args_str[1:]).read_text() args_str = Path(args_str[1:]).read_text(encoding="utf-8")
except OSError as e: except OSError as e:
emit_json({"error": f"Cannot read args file: {e}"}) emit_json({"error": f"Cannot read args file: {e}"})
return 1 return 1
+3 -3
View File
@@ -22,17 +22,17 @@ sys.path.insert(0, str(SCRIPTS))
@pytest.fixture @pytest.fixture
def sd15_workflow() -> dict: def sd15_workflow() -> dict:
return json.loads((WORKFLOWS / "sd15_txt2img.json").read_text()) return json.loads((WORKFLOWS / "sd15_txt2img.json").read_text(encoding="utf-8"))
@pytest.fixture @pytest.fixture
def flux_workflow() -> dict: def flux_workflow() -> dict:
return json.loads((WORKFLOWS / "flux_dev_txt2img.json").read_text()) return json.loads((WORKFLOWS / "flux_dev_txt2img.json").read_text(encoding="utf-8"))
@pytest.fixture @pytest.fixture
def video_workflow() -> dict: def video_workflow() -> dict:
return json.loads((WORKFLOWS / "wan_video_t2v.json").read_text()) return json.loads((WORKFLOWS / "wan_video_t2v.json").read_text(encoding="utf-8"))
@pytest.fixture @pytest.fixture
+1 -1
View File
@@ -436,7 +436,7 @@ class TestVideoWorkflow:
def test_animatediff_workflow(self, workflows_dir): def test_animatediff_workflow(self, workflows_dir):
import json import json
wf = json.loads((workflows_dir / "animatediff_video.json").read_text()) wf = json.loads((workflows_dir / "animatediff_video.json").read_text(encoding="utf-8"))
assert looks_like_video_workflow(wf) is True assert looks_like_video_workflow(wf) is True
def test_wan_workflow(self, video_workflow): def test_wan_workflow(self, video_workflow):
@@ -16,20 +16,20 @@ from run_workflow import (
class TestParseInputImageArg: class TestParseInputImageArg:
def test_with_name(self, tmp_path): def test_with_name(self, tmp_path):
f = tmp_path / "x.png" f = tmp_path / "x.png"
f.write_text("x") f.write_text("x", encoding="utf-8")
n, p = parse_input_image_arg(f"image={f}") n, p = parse_input_image_arg(f"image={f}")
assert n == "image" assert n == "image"
assert p == f assert p == f
def test_without_name_defaults(self, tmp_path): def test_without_name_defaults(self, tmp_path):
f = tmp_path / "x.png" f = tmp_path / "x.png"
f.write_text("x") f.write_text("x", encoding="utf-8")
n, p = parse_input_image_arg(str(f)) n, p = parse_input_image_arg(str(f))
assert n == "image" assert n == "image"
def test_custom_name(self, tmp_path): def test_custom_name(self, tmp_path):
f = tmp_path / "x.png" f = tmp_path / "x.png"
f.write_text("x") f.write_text("x", encoding="utf-8")
n, p = parse_input_image_arg(f"mask_image={f}") n, p = parse_input_image_arg(f"mask_image={f}")
assert n == "mask_image" assert n == "mask_image"
@@ -92,7 +92,7 @@ def _setup_libreoffice_macro() -> bool:
macro_dir = Path(MACRO_DIR) macro_dir = Path(MACRO_DIR)
macro_file = macro_dir / "Module1.xba" macro_file = macro_dir / "Module1.xba"
if macro_file.exists() and "AcceptAllTrackedChanges" in macro_file.read_text(): if macro_file.exists() and "AcceptAllTrackedChanges" in macro_file.read_text(encoding="utf-8"):
return True return True
if not macro_dir.exists(): if not macro_dir.exists():
@@ -111,7 +111,7 @@ def _setup_libreoffice_macro() -> bool:
macro_dir.mkdir(parents=True, exist_ok=True) macro_dir.mkdir(parents=True, exist_ok=True)
try: try:
macro_file.write_text(ACCEPT_CHANGES_MACRO) macro_file.write_text(ACCEPT_CHANGES_MACRO, encoding="utf-8")
return True return True
except Exception as e: except Exception as e:
logger.warning(f"Failed to setup LibreOffice macro: {e}") logger.warning(f"Failed to setup LibreOffice macro: {e}")
@@ -64,7 +64,7 @@ def _ensure_shim() -> Path:
return _SHIM_SO return _SHIM_SO
src = Path(tempfile.gettempdir()) / "lo_socket_shim.c" src = Path(tempfile.gettempdir()) / "lo_socket_shim.c"
src.write_text(_SHIM_SOURCE) src.write_text(_SHIM_SOURCE, encoding="utf-8")
subprocess.run( subprocess.run(
["gcc", "-shared", "-fPIC", "-o", str(_SHIM_SO), str(src), "-ldl"], ["gcc", "-shared", "-fPIC", "-o", str(_SHIM_SO), str(src), "-ldl"],
check=True, check=True,
@@ -70,7 +70,7 @@ def _ensure_authenticated():
def _stored_token_scopes() -> list[str]: def _stored_token_scopes() -> list[str]:
try: try:
data = json.loads(TOKEN_PATH.read_text()) data = json.loads(TOKEN_PATH.read_text(encoding="utf-8"))
except Exception: except Exception:
return list(SCOPES) return list(SCOPES)
scopes = data.get("scopes") scopes = data.get("scopes")
@@ -192,7 +192,7 @@ def get_credentials():
json.dumps( json.dumps(
_normalize_authorized_user_payload(json.loads(creds.to_json())), _normalize_authorized_user_payload(json.loads(creds.to_json())),
indent=2, indent=2,
) ), encoding="utf-8"
) )
if not creds.valid: if not creds.valid:
print("Token is invalid. Re-run setup.", file=sys.stderr) print("Token is invalid. Re-run setup.", file=sys.stderr)
@@ -69,7 +69,7 @@ def refresh_token(token_data: dict) -> dict:
).isoformat() ).isoformat()
get_token_path().write_text( get_token_path().write_text(
json.dumps(_normalize_authorized_user_payload(token_data), indent=2) json.dumps(_normalize_authorized_user_payload(token_data), indent=2), encoding="utf-8"
) )
return token_data return token_data
@@ -81,7 +81,7 @@ def get_valid_token() -> str:
print("ERROR: No Google token found. Run setup.py --auth-url first.", file=sys.stderr) print("ERROR: No Google token found. Run setup.py --auth-url first.", file=sys.stderr)
sys.exit(1) sys.exit(1)
token_data = json.loads(token_path.read_text()) token_data = json.loads(token_path.read_text(encoding="utf-8"))
expiry = token_data.get("expiry", "") expiry = token_data.get("expiry", "")
if expiry: if expiry:
@@ -71,7 +71,7 @@ def _normalize_authorized_user_payload(payload: dict) -> dict:
def _load_token_payload(path: Path = TOKEN_PATH) -> dict: def _load_token_payload(path: Path = TOKEN_PATH) -> dict:
try: try:
return json.loads(path.read_text()) return json.loads(path.read_text(encoding="utf-8"))
except Exception: except Exception:
return {} return {}
@@ -220,7 +220,7 @@ def check_auth(quiet: bool = False):
json.dumps( json.dumps(
_normalize_authorized_user_payload(json.loads(creds.to_json())), _normalize_authorized_user_payload(json.loads(creds.to_json())),
indent=2, indent=2,
) ), encoding="utf-8"
) )
missing_scopes = _missing_scopes_from_payload(_load_token_payload(TOKEN_PATH)) missing_scopes = _missing_scopes_from_payload(_load_token_payload(TOKEN_PATH))
if missing_scopes: if missing_scopes:
@@ -260,7 +260,7 @@ def store_client_secret(path: str):
sys.exit(1) sys.exit(1)
try: try:
data = json.loads(src.read_text()) data = json.loads(src.read_text(encoding="utf-8"))
except json.JSONDecodeError: except json.JSONDecodeError:
print("ERROR: File is not valid JSON.") print("ERROR: File is not valid JSON.")
sys.exit(1) sys.exit(1)
@@ -270,7 +270,7 @@ def store_client_secret(path: str):
print("Download the correct file from: https://console.cloud.google.com/apis/credentials") print("Download the correct file from: https://console.cloud.google.com/apis/credentials")
sys.exit(1) sys.exit(1)
CLIENT_SECRET_PATH.write_text(json.dumps(data, indent=2)) CLIENT_SECRET_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8")
print(f"OK: Client secret saved to {CLIENT_SECRET_PATH}") print(f"OK: Client secret saved to {CLIENT_SECRET_PATH}")
@@ -284,7 +284,7 @@ def _save_pending_auth(*, state: str, code_verifier: str):
"redirect_uri": REDIRECT_URI, "redirect_uri": REDIRECT_URI,
}, },
indent=2, indent=2,
) ), encoding="utf-8"
) )
@@ -295,7 +295,7 @@ def _load_pending_auth() -> dict:
sys.exit(1) sys.exit(1)
try: try:
data = json.loads(PENDING_AUTH_PATH.read_text()) data = json.loads(PENDING_AUTH_PATH.read_text(encoding="utf-8"))
except Exception as e: except Exception as e:
print(f"ERROR: Could not read pending OAuth session: {e}") print(f"ERROR: Could not read pending OAuth session: {e}")
print("Run --auth-url again to start a fresh OAuth session.") print("Run --auth-url again to start a fresh OAuth session.")
@@ -410,7 +410,7 @@ def exchange_auth_code(code: str):
print(f"WARNING: Token missing some Google Workspace scopes: {', '.join(missing_scopes)}") print(f"WARNING: Token missing some Google Workspace scopes: {', '.join(missing_scopes)}")
print("Some services may not be available.") print("Some services may not be available.")
TOKEN_PATH.write_text(json.dumps(token_payload, indent=2)) TOKEN_PATH.write_text(json.dumps(token_payload, indent=2), encoding="utf-8")
PENDING_AUTH_PATH.unlink(missing_ok=True) PENDING_AUTH_PATH.unlink(missing_ok=True)
print(f"OK: Authenticated. Token saved to {TOKEN_PATH}") print(f"OK: Authenticated. Token saved to {TOKEN_PATH}")
print(f"Profile-scoped token location: {display_hermes_home()}/google_token.json") print(f"Profile-scoped token location: {display_hermes_home()}/google_token.json")
@@ -64,7 +64,7 @@ def _ensure_shim() -> Path:
return _SHIM_SO return _SHIM_SO
src = Path(tempfile.gettempdir()) / "lo_socket_shim.c" src = Path(tempfile.gettempdir()) / "lo_socket_shim.c"
src.write_text(_SHIM_SOURCE) src.write_text(_SHIM_SOURCE, encoding="utf-8")
subprocess.run( subprocess.run(
["gcc", "-shared", "-fPIC", "-o", str(_SHIM_SO), str(src), "-ldl"], ["gcc", "-shared", "-fPIC", "-o", str(_SHIM_SO), str(src), "-ldl"],
check=True, check=True,
@@ -64,7 +64,7 @@ def _ensure_shim() -> Path:
return _SHIM_SO return _SHIM_SO
src = Path(tempfile.gettempdir()) / "lo_socket_shim.c" src = Path(tempfile.gettempdir()) / "lo_socket_shim.c"
src.write_text(_SHIM_SOURCE) src.write_text(_SHIM_SOURCE, encoding="utf-8")
subprocess.run( subprocess.run(
["gcc", "-shared", "-fPIC", "-o", str(_SHIM_SO), str(src), "-ldl"], ["gcc", "-shared", "-fPIC", "-o", str(_SHIM_SO), str(src), "-ldl"],
check=True, check=True,
+1 -1
View File
@@ -71,7 +71,7 @@ def setup_libreoffice_macro(profile_dir: Path, timeout=30):
return None, "LibreOffice did not create a usable profile; formulas were NOT recalculated" return None, "LibreOffice did not create a usable profile; formulas were NOT recalculated"
try: try:
(macro_dir / MACRO_FILENAME).write_text(RECALCULATE_MACRO) (macro_dir / MACRO_FILENAME).write_text(RECALCULATE_MACRO, encoding="utf-8")
except OSError as e: except OSError as e:
return None, f"Could not install the recalculation macro: {e}" return None, f"Could not install the recalculation macro: {e}"
+1 -1
View File
@@ -37,7 +37,7 @@ def _read_nous_provider_state() -> Optional[dict]:
path = auth_json_path() path = auth_json_path()
if not path.is_file(): if not path.is_file():
return None return None
data = json.loads(path.read_text()) data = json.loads(path.read_text(encoding="utf-8"))
providers = data.get("providers", {}) providers = data.get("providers", {})
if not isinstance(providers, dict): if not isinstance(providers, dict):
return None return None