DESKTOP-L8OUL9F\gfl0034
40ce6c7dfa
补齐三页查询/开关/授权/数量布局，并附上线上 static-override 与 jar 热补部署脚本。 Co-authored-by: Cursor <cursoragent@cursor.com>
101 lines
4.0 KiB
Python
101 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
import json, os, urllib.request, urllib.error
|
|
from pathlib import Path
|
|
import paramiko
|
|
|
|
HOST = "47.236.66.9"
|
|
PEM = Path(os.path.expandvars(r"%USERPROFILE%\.ssh\wyrs_shanghai_deploy.pem"))
|
|
|
|
|
|
def load_key(pem):
|
|
for loader in (
|
|
paramiko.Ed25519Key.from_private_key_file,
|
|
paramiko.RSAKey.from_private_key_file,
|
|
paramiko.ECDSAKey.from_private_key_file,
|
|
):
|
|
try:
|
|
return loader(str(pem))
|
|
except Exception:
|
|
continue
|
|
raise SystemExit("key fail")
|
|
|
|
|
|
def main():
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
client.connect(HOST, username="root", pkey=load_key(PEM), timeout=30)
|
|
script = r'''
|
|
python3 - <<'PY'
|
|
import json, urllib.request, urllib.error
|
|
env={}
|
|
for line in open('/etc/wyrs-admin/app.env',encoding='utf-8',errors='replace'):
|
|
line=line.strip()
|
|
if not line or line.startswith('#') or '=' not in line: continue
|
|
k,v=line.split('=',1); env[k.strip()]=v.strip().strip('"').strip("'")
|
|
password=env.get('WYRS_ADMIN_PASSWORD') or ''
|
|
# try both username/password field names
|
|
for body in (
|
|
{'username':'admin','password':password},
|
|
{'email':'admin','password':password},
|
|
):
|
|
raw=json.dumps(body).encode()
|
|
req=urllib.request.Request('http://127.0.0.1:8090/api/admin/auth/login',data=raw,headers={'Content-Type':'application/json'},method='POST')
|
|
try:
|
|
with urllib.request.urlopen(req,timeout=20) as resp:
|
|
data=json.loads(resp.read().decode())
|
|
print('LOGIN_OK', body.keys(), list(data.keys()), data.get('message'), type(data.get('data')))
|
|
token=(data.get('data') or {}).get('token') if isinstance(data.get('data'), dict) else None
|
|
if token:
|
|
break
|
|
except urllib.error.HTTPError as e:
|
|
print('LOGIN_ERR', body.keys(), e.code, e.read()[:300])
|
|
token=None
|
|
if not token:
|
|
raise SystemExit('no token')
|
|
|
|
def get(url):
|
|
req=urllib.request.Request(url,headers={'Authorization':'Bearer '+token})
|
|
try:
|
|
with urllib.request.urlopen(req,timeout=20) as resp:
|
|
print('GET', url, resp.read().decode()[:700])
|
|
except urllib.error.HTTPError as e:
|
|
print('GET_ERR', url, e.code, e.read()[:500])
|
|
|
|
get('http://127.0.0.1:8090/api/admin/spark/media-pull?userId=1017')
|
|
get('http://127.0.0.1:8090/api/admin/spark/media-pull?email=spark.test.01@wyrs.test.local')
|
|
get('http://127.0.0.1:8090/api/admin/spark/contacts/summary')
|
|
get('http://127.0.0.1:8090/api/admin/spark/media/summary')
|
|
get('http://127.0.0.1:8090/api/admin/spark/media?email=spark.test.01@wyrs.test.local&kind=document&page=0&size=5')
|
|
get('http://127.0.0.1:8090/api/admin/spark/contacts?userId=1017&page=0&size=5')
|
|
|
|
# PUT toggles
|
|
payload=json.dumps({'pullContacts':False,'pullDocuments':False,'pullPhotos':False,'pullVideos':False,'enabled':False,'count':25}).encode()
|
|
req=urllib.request.Request(
|
|
'http://127.0.0.1:8090/api/admin/spark/media-pull?email=spark.test.01@wyrs.test.local',
|
|
data=payload,
|
|
headers={'Authorization':'Bearer '+token,'Content-Type':'application/json'},
|
|
method='PUT')
|
|
try:
|
|
with urllib.request.urlopen(req,timeout=20) as resp:
|
|
print('PUT', resp.read().decode()[:800])
|
|
except urllib.error.HTTPError as e:
|
|
print('PUT_ERR', e.code, e.read()[:500])
|
|
|
|
# static markers
|
|
import pathlib
|
|
s=pathlib.Path('/opt/wyrs-admin/static-override/popadmin.js').read_text(encoding='utf-8',errors='replace')
|
|
idx=pathlib.Path('/opt/wyrs-admin/static-override/popadmin.index.html').read_text(encoding='utf-8',errors='replace')
|
|
print('STATIC', 'contactsConsented' in s, '用户未授权' in s, '20261102-pullpages' in idx)
|
|
print('INDEX_JS', [line for line in idx.splitlines() if 'popadmin.js' in line][:2])
|
|
PY'''
|
|
_, stdout, stderr = client.exec_command(script, timeout=60)
|
|
print(stdout.read().decode("utf-8", "replace"))
|
|
err = stderr.read().decode("utf-8", "replace")
|
|
if err.strip():
|
|
print("STDERR", err[:2000])
|
|
client.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|