azcomputerguru/claudetools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: auto-sync from DESKTOP-0O8A1RL at 2026-04-21 18:46:45

Browse Source 
Author: Mike Swanson
Machine: DESKTOP-0O8A1RL
Timestamp: 2026-04-21 18:46:45
This commit is contained in:
Mike Swanson
2026-04-21 18:46:49 -07:00
parent a9bcbc2580
commit 63089c45c9
6 changed files with 1504 additions and 1324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

524
projects/dataforth-dos/datasheet-pipeline/implementation/deploy-to-ad2.py
View File

@@ -1,224 +1,300 @@
"""
Deploy staged pipeline changes to AD2:C:\\Shares\\testdatadb\\.
Backs up each existing target to <name>.bak-YYYYMMDD before overwriting.
Fails if a target file does not exist on AD2 (excluding brand-new files
declared in NEW_FILES below).
Usage:
python deploy-to-ad2.py --dry-run
python deploy-to-ad2.py
Credentials: fetched at runtime from the SOPS vault
(clients/dataforth/ad2.sops.yaml -> credentials.password). No hardcoded
password; no env-var / prompt fallback. Fails loud if the vault read fails.
"""
from __future__ import annotations
import argparse
import datetime
import os
import subprocess
import sys
import paramiko
HOST = '192.168.0.6'
USER = 'sysadmin'
VAULT_SH = 'D:/vault/scripts/vault.sh'
VAULT_ENTRY = 'clients/dataforth/ad2.sops.yaml'
VAULT_FIELD = 'credentials.password'
def get_ad2_password() -> str:
"""Fetch the AD2 sysadmin password from the SOPS vault.
Fails loud (raises) on any error: missing vault, decryption failure,
empty value. Do NOT fall back to env vars or prompts -- per CLAUDE.md
deploy scripts must not hold credentials.
"""
try:
result = subprocess.run(
['bash', VAULT_SH, 'get-field', VAULT_ENTRY, VAULT_FIELD],
capture_output=True, text=True, timeout=30, check=False,
)
except FileNotFoundError as e:
raise RuntimeError(
f'[FAIL] vault helper not runnable: {VAULT_SH} ({e})'
) from e
except subprocess.TimeoutExpired as e:
raise RuntimeError(
f'[FAIL] vault read timed out after 30s for {VAULT_ENTRY}'
) from e
if result.returncode != 0:
stderr = (result.stderr or '').strip()
raise RuntimeError(
f'[FAIL] vault read failed (rc={result.returncode}) for '
f'{VAULT_ENTRY}:{VAULT_FIELD}: {stderr}'
)
pwd = (result.stdout or '').strip()
if not pwd:
raise RuntimeError(
f'[FAIL] vault returned empty value for {VAULT_ENTRY}:{VAULT_FIELD}'
)
return pwd
REMOTE_ROOT = 'C:/Shares/testdatadb'
LOCAL_ROOT = os.path.dirname(os.path.abspath(__file__))
# ---------------------------------------------------------------------------
# Deployment file lists. Each list has different semantics:
#
# UPDATE_FILES -- file MUST already exist on AD2. Backup-then-overwrite.
# Fails loud if the remote file is missing (that's a drift
# signal -- something changed on the box we didn't expect).
#
# NEW_FILES -- file must NOT already exist on AD2. Creates it.
# Fails loud if the remote file is already present (we would
# otherwise silently clobber something we didn't back up).
# ---------------------------------------------------------------------------
# Files that already exist on AD2 and will be backed up + overwritten.
UPDATE_FILES = [
('parsers/spec-reader.js', 'parsers/spec-reader.js'),
('templates/datasheet-exact.js', 'templates/datasheet-exact.js'),
('database/import.js', 'database/import.js'),
('database/export-datasheets.js', 'database/export-datasheets.js'),
]
# Files that do NOT yet exist on AD2 and must be created fresh.
NEW_FILES = [
('parsers/vaslog-engtxt.js', 'parsers/vaslog-engtxt.js'),
]
def connect() -> paramiko.SSHClient:
pwd = get_ad2_password()
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect(
HOST, username=USER, password=pwd,
timeout=15, look_for_keys=False, allow_agent=False, banner_timeout=30,
)
return c
def remote_exists(sftp: paramiko.SFTPClient, path: str) -> bool:
try:
sftp.stat(path)
return True
except IOError:
return False
def to_remote(rel: str) -> str:
return f'{REMOTE_ROOT}/{rel}'
def backup_and_copy(sftp: paramiko.SFTPClient, ssh: paramiko.SSHClient,
local_rel: str, remote_rel: str, dry_run: bool, stamp: str) -> None:
local_path = os.path.join(LOCAL_ROOT, local_rel.replace('/', os.sep))
remote_path = to_remote(remote_rel)
backup_path = f'{remote_path}.bak-{stamp}'
if not os.path.isfile(local_path):
raise FileNotFoundError(f'[FAIL] local file missing: {local_path}')
if not remote_exists(sftp, remote_path):
raise FileNotFoundError(f'[FAIL] remote file missing on AD2: {remote_path}')
print(f'[INFO] {remote_rel}')
if dry_run:
print(f' would back up to: {backup_path}')
print(f' would upload: {local_path} -> {remote_path}')
return
# Backup via SFTP copy (read + re-upload). Paramiko has no server-side copy.
with sftp.open(remote_path, 'rb') as src:
data = src.read()
with sftp.open(backup_path, 'wb') as dst:
dst.write(data)
print(f' backup: {backup_path} ({len(data)} bytes)')
sftp.put(local_path, remote_path)
size = os.path.getsize(local_path)
print(f' uploaded: {local_path} -> {remote_path} ({size} bytes)')
def create_new(sftp: paramiko.SFTPClient, local_rel: str, remote_rel: str,
dry_run: bool) -> None:
"""Create a file that is expected to be NEW on AD2.
Fails loud if the remote file already exists -- NEW_FILES declares this
is a brand-new file, so pre-existence is a drift signal. If a previous
deploy partially ran, clean up manually or move the entry to
UPDATE_FILES.
"""
local_path = os.path.join(LOCAL_ROOT, local_rel.replace('/', os.sep))
remote_path = to_remote(remote_rel)
if not os.path.isfile(local_path):
raise FileNotFoundError(f'[FAIL] local file missing: {local_path}')
print(f'[INFO] {remote_rel} (NEW)')
if remote_exists(sftp, remote_path):
raise FileExistsError(
f'[FAIL] remote target already exists but is declared NEW: {remote_path} '
f'-- move to UPDATE_FILES or remove remote manually'
)
if dry_run:
print(f' would create: {local_path} -> {remote_path}')
return
sftp.put(local_path, remote_path)
size = os.path.getsize(local_path)
print(f' created: {remote_path} ({size} bytes)')
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument('--dry-run', action='store_true', help='print actions without writing')
args = ap.parse_args()
stamp = datetime.date.today().strftime('%Y%m%d')
print('=' * 72)
print('Deploy staged pipeline changes to AD2')
print('=' * 72)
print(f'Host: {HOST}')
print(f'Remote root: {REMOTE_ROOT}')
print(f'Local root: {LOCAL_ROOT}')
print(f'Dry run: {args.dry_run}')
print(f'Backup tag: .bak-{stamp}')
print('')
ssh = connect()
try:
sftp = ssh.open_sftp()
try:
for local_rel, remote_rel in UPDATE_FILES:
backup_and_copy(sftp, ssh, local_rel, remote_rel, args.dry_run, stamp)
for local_rel, remote_rel in NEW_FILES:
create_new(sftp, local_rel, remote_rel, args.dry_run)
finally:
sftp.close()
finally:
ssh.close()
print('')
print('[OK] done' if not args.dry_run else '[OK] dry-run complete (no changes made)')
return 0
if __name__ == '__main__':
try:
sys.exit(main())
except Exception as e:
print(f'[FAIL] {e}', file=sys.stderr)
sys.exit(1)
"""
Deploy staged pipeline changes to AD2:C:\\Shares\\testdatadb\\.
Backs up each existing target to <name>.bak-YYYYMMDD before overwriting.
Fails if a target file does not exist on AD2 (excluding brand-new files
declared in NEW_FILES below).
Usage:
python deploy-to-ad2.py --dry-run
python deploy-to-ad2.py
Credentials: fetched at runtime from the SOPS vault
(clients/dataforth/ad2.sops.yaml -> credentials.password). No hardcoded
password; no env-var / prompt fallback. Fails loud if the vault read fails.
"""
from __future__ import annotations
import argparse
import datetime
import os
import subprocess
import sys
import paramiko
HOST = '192.168.0.6'
USER = 'sysadmin'
VAULT_SH = 'D:/vault/scripts/vault.sh'
VAULT_ENTRY = 'clients/dataforth/ad2.sops.yaml'
VAULT_FIELD = 'credentials.password'
SMTP_VAULT_ENTRY = 'clients/dataforth/m365.sops.yaml'
SMTP_VAULT_FIELD = 'credentials.password'
SMTP_USER = 'sysadmin@dataforth.com'
SMTP_HOST = 'smtp.office365.com'
SMTP_PORT = 587
NOTIFY_TO = 'mike@azcomputerguru.com'
def get_ad2_password() -> str:
"""Fetch the AD2 sysadmin password from the SOPS vault.
Fails loud (raises) on any error: missing vault, decryption failure,
empty value. Do NOT fall back to env vars or prompts -- per CLAUDE.md
deploy scripts must not hold credentials.
"""
try:
result = subprocess.run(
['bash', VAULT_SH, 'get-field', VAULT_ENTRY, VAULT_FIELD],
capture_output=True, text=True, timeout=30, check=False,
)
except FileNotFoundError as e:
raise RuntimeError(
f'[FAIL] vault helper not runnable: {VAULT_SH} ({e})'
) from e
except subprocess.TimeoutExpired as e:
raise RuntimeError(
f'[FAIL] vault read timed out after 30s for {VAULT_ENTRY}'
) from e
if result.returncode != 0:
stderr = (result.stderr or '').strip()
raise RuntimeError(
f'[FAIL] vault read failed (rc={result.returncode}) for '
f'{VAULT_ENTRY}:{VAULT_FIELD}: {stderr}'
)
pwd = (result.stdout or '').strip()
if not pwd:
raise RuntimeError(
f'[FAIL] vault returned empty value for {VAULT_ENTRY}:{VAULT_FIELD}'
)
return pwd
def get_smtp_password() -> str:
try:
result = subprocess.run(
['bash', VAULT_SH, 'get-field', SMTP_VAULT_ENTRY, SMTP_VAULT_FIELD],
capture_output=True, text=True, timeout=30, check=False,
)
except (FileNotFoundError, subprocess.TimeoutExpired) as e:
raise RuntimeError(f'[FAIL] vault read failed for SMTP creds: {e}') from e
if result.returncode != 0:
raise RuntimeError(
f'[FAIL] vault read failed (rc={result.returncode}) for '
f'{SMTP_VAULT_ENTRY}:{SMTP_VAULT_FIELD}: {result.stderr.strip()}'
)
pwd = (result.stdout or '').strip().replace('\\', '')
if not pwd:
raise RuntimeError(f'[FAIL] vault returned empty SMTP password')
return pwd
REMOTE_ROOT = 'C:/Shares/testdatadb'
LOCAL_ROOT = os.path.dirname(os.path.abspath(__file__))
# ---------------------------------------------------------------------------
# Deployment file lists. Each list has different semantics:
#
# UPDATE_FILES -- file MUST already exist on AD2. Backup-then-overwrite.
# Fails loud if the remote file is missing (that's a drift
# signal -- something changed on the box we didn't expect).
#
# NEW_FILES -- file must NOT already exist on AD2. Creates it.
# Fails loud if the remote file is already present (we would
# otherwise silently clobber something we didn't back up).
# ---------------------------------------------------------------------------
# Files that already exist on AD2 and will be backed up + overwritten.
UPDATE_FILES = [
('parsers/spec-reader.js', 'parsers/spec-reader.js'),
('templates/datasheet-exact.js', 'templates/datasheet-exact.js'),
('database/import.js', 'database/import.js'),
('database/export-datasheets.js', 'database/export-datasheets.js'),
]
# Files that do NOT yet exist on AD2 and must be created fresh.
NEW_FILES = [
('parsers/vaslog-engtxt.js', 'parsers/vaslog-engtxt.js'),
('server/notify.js', 'server/notify.js'),
]
def connect() -> paramiko.SSHClient:
pwd = get_ad2_password()
c = paramiko.SSHClient()
c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
c.connect(
HOST, username=USER, password=pwd,
timeout=15, look_for_keys=False, allow_agent=False, banner_timeout=30,
)
return c
def remote_exists(sftp: paramiko.SFTPClient, path: str) -> bool:
try:
sftp.stat(path)
return True
except IOError:
return False
def to_remote(rel: str) -> str:
return f'{REMOTE_ROOT}/{rel}'
def backup_and_copy(sftp: paramiko.SFTPClient, ssh: paramiko.SSHClient,
local_rel: str, remote_rel: str, dry_run: bool, stamp: str) -> None:
local_path = os.path.join(LOCAL_ROOT, local_rel.replace('/', os.sep))
remote_path = to_remote(remote_rel)
backup_path = f'{remote_path}.bak-{stamp}'
if not os.path.isfile(local_path):
raise FileNotFoundError(f'[FAIL] local file missing: {local_path}')
if not remote_exists(sftp, remote_path):
raise FileNotFoundError(f'[FAIL] remote file missing on AD2: {remote_path}')
print(f'[INFO] {remote_rel}')
if dry_run:
print(f' would back up to: {backup_path}')
print(f' would upload: {local_path} -> {remote_path}')
return
# Backup via SFTP copy (read + re-upload). Paramiko has no server-side copy.
with sftp.open(remote_path, 'rb') as src:
data = src.read()
with sftp.open(backup_path, 'wb') as dst:
dst.write(data)
print(f' backup: {backup_path} ({len(data)} bytes)')
sftp.put(local_path, remote_path)
size = os.path.getsize(local_path)
print(f' uploaded: {local_path} -> {remote_path} ({size} bytes)')
def create_new(sftp: paramiko.SFTPClient, local_rel: str, remote_rel: str,
dry_run: bool) -> None:
"""Create a file that is expected to be NEW on AD2.
Fails loud if the remote file already exists -- NEW_FILES declares this
is a brand-new file, so pre-existence is a drift signal. If a previous
deploy partially ran, clean up manually or move the entry to
UPDATE_FILES.
"""
local_path = os.path.join(LOCAL_ROOT, local_rel.replace('/', os.sep))
remote_path = to_remote(remote_rel)
if not os.path.isfile(local_path):
raise FileNotFoundError(f'[FAIL] local file missing: {local_path}')
print(f'[INFO] {remote_rel} (NEW)')
if remote_exists(sftp, remote_path):
raise FileExistsError(
f'[FAIL] remote target already exists but is declared NEW: {remote_path} '
f'-- move to UPDATE_FILES or remove remote manually'
)
if dry_run:
print(f' would create: {local_path} -> {remote_path}')
return
sftp.put(local_path, remote_path)
size = os.path.getsize(local_path)
print(f' created: {remote_path} ({size} bytes)')
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument('--dry-run', action='store_true', help='print actions without writing')
args = ap.parse_args()
stamp = datetime.date.today().strftime('%Y%m%d')
print('=' * 72)
print('Deploy staged pipeline changes to AD2')
print('=' * 72)
print(f'Host: {HOST}')
print(f'Remote root: {REMOTE_ROOT}')
print(f'Local root: {LOCAL_ROOT}')
print(f'Dry run: {args.dry_run}')
print(f'Backup tag: .bak-{stamp}')
print('')
smtp_pass = get_smtp_password()
ssh = connect()
try:
sftp = ssh.open_sftp()
try:
for local_rel, remote_rel in UPDATE_FILES:
backup_and_copy(sftp, ssh, local_rel, remote_rel, args.dry_run, stamp)
for local_rel, remote_rel in NEW_FILES:
create_new(sftp, local_rel, remote_rel, args.dry_run)
# Write notify config (creds fetched from vault, never committed to git)
import json
notify_cfg = json.dumps({
'smtp': {
'host': SMTP_HOST,
'port': SMTP_PORT,
'user': SMTP_USER,
'pass': smtp_pass,
},
'from': SMTP_USER,
'to': NOTIFY_TO,
}, indent=2)
notify_remote = f'{REMOTE_ROOT}/config/notify.json'
print(f'[INFO] config/notify.json (SMTP creds)')
if not args.dry_run:
# Ensure config dir exists
stdin, stdout, stderr = ssh.exec_command(
f'powershell -Command "New-Item -ItemType Directory -Force -Path '
f'C:\\Shares\\testdatadb\\config | Out-Null"'
)
stdout.channel.recv_exit_status()
with sftp.open(notify_remote, 'w') as f:
f.write(notify_cfg)
print(f' written: {notify_remote}')
else:
print(f' would write: {notify_remote}')
finally:
sftp.close()
# Install nodemailer if not already present
print('[INFO] npm install nodemailer')
if not args.dry_run:
cmd = 'cd C:\\Shares\\testdatadb && npm list nodemailer --depth=0 2>nul || npm install nodemailer'
stdin, stdout, stderr = ssh.exec_command(f'cmd /c "{cmd}"')
exit_code = stdout.channel.recv_exit_status()
out = stdout.read().decode(errors='replace').strip()
if out:
print(f' {out}')
if exit_code != 0:
err = stderr.read().decode(errors='replace').strip()
raise RuntimeError(f'[FAIL] npm install nodemailer failed: {err}')
print('[OK] nodemailer ready')
else:
print(' would run: npm install nodemailer (if not already installed)')
finally:
ssh.close()
print('')
print('[OK] done' if not args.dry_run else '[OK] dry-run complete (no changes made)')
return 0
if __name__ == '__main__':
try:
sys.exit(main())
except Exception as e:
print(f'[FAIL] {e}', file=sys.stderr)
sys.exit(1)