dpl/scripts/dpl-resource-downloads/migrate.py

#!/usr/bin/env python3
"""DPL-only theme download settings migration; never edits XB core or other sites."""
import argparse,datetime,hashlib,html,json,os,re,shutil,subprocess,sys,tempfile,urllib.request
from pathlib import Path
BASE='https://downloads.gemcook.net/downloads/software/'
PAGE='/theme/Stellar/resource-downloads.html'
FILES={
 'windows':('clash-verge-rev','clash-verge-rev-windows-x64.exe'),
 'macos_arm64':('clash-verge-rev','clash-verge-rev-macos-arm64.dmg'),
 'macos_x64':('clash-verge-rev','clash-verge-rev-macos-x64.dmg'),
 'android':('flclash','flclash-android-arm64.apk'),
}
CHOICES=[('windows','Windows ARM64','clash-verge-rev','clash-verge-rev-windows-arm64.exe'),('macos','macOS Apple 芯片','clash-verge-rev','clash-verge-rev-macos-arm64.dmg'),('macos','macOS Intel 芯片','clash-verge-rev','clash-verge-rev-macos-x64.dmg'),('android','Android ARM64（多数手机）','flclash','flclash-android-arm64.apk'),('android','Android ARMv7（旧款手机）','flclash','flclash-android-armv7.apk')]+[('linux',f'Linux {arch.upper()} · {ext.upper()}','clash-verge-rev',f'clash-verge-rev-linux-{arch}.{ext}') for arch in ['x64','arm64'] for ext in ['deb','rpm']]
def url(software,name):return BASE+software+'/'+name
def digest(b):return hashlib.sha256(b).hexdigest()
def prepare(root):
 env=root/'storage/theme/Stellar/env.js';s=env.read_text()
 pat=r'([\"\']?client_downloads[\"\']?\s*:\s*\{)(.*?)(\})'
 match=re.search(pat,s,re.S);assert match,'Missing client_downloads'
 block=match[2];before=block
 settings={k:url(*v) for k,v in FILES.items()};settings.update(macos=PAGE+'#macos',linux=PAGE+'#linux')
 for key,value in settings.items():
  block,n=re.subn(r'([\"\']?'+key+r'[\"\']?\s*:\s*)([\"\'])([^\"\']*)\2',lambda m:m[1]+m[2]+value+m[2],block)
  assert n==1,'Missing or duplicated download field: '+key
 # Verify unrelated download options and every other setting are byte-identical.
 assert all(re.search(r'[\"\']?'+k+r'[\"\']?\s*:\s*[\"\']([^\"\']*)',before)[1]==re.search(r'[\"\']?'+k+r'[\"\']?\s*:\s*[\"\']([^\"\']*)',block)[1] for k in ['ios','router'])
 replacement=s[:match.start(2)]+block+s[match.end(2):]
 page='''<!doctype html><html lang="zh-CN"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>DPL 客户端下载</title><style>body{font:16px/1.65 system-ui,sans-serif;max-width:760px;margin:40px auto;padding:0 24px;color:#1f2937;background:#f8fafc}h1{font-size:28px}section{background:white;padding:18px 24px;margin:20px 0;border:1px solid #e2e8f0;border-radius:14px}a{color:#2563eb}li{margin:12px 0}p{color:#475569}nav a{margin-right:20px}</style><h1>DPL 客户端下载</h1><p>按设备架构选择安装包。下载链接固定，软件更新后无需更换地址。</p><nav><a href="#windows">Windows</a><a href="#macos">macOS</a><a href="#android">Android</a><a href="#linux">Linux</a></nav>'''
 for group,label in [('windows','Windows'),('macos','macOS'),('android','Android'),('linux','Linux')]:
  page+=f'<section id="{group}"><h2>{label}</h2>'
  if group=='linux':page+='<p>Ubuntu / Debian / Deepin 选择 DEB，Fedora / RHEL 系选择 RPM。运行 <code>uname -m</code>：x86_64 选择 X64，aarch64 选择 ARM64。Clash Verge Rev 需要图形桌面。</p>'
  page+='<ul>'
  if group=='windows':page+=f'<li><a href="{url(*FILES["windows"])}">Windows X64</a></li>'
  for g,title,sw,name in CHOICES:
   if group==g:page+=f'<li><a href="{html.escape(url(sw,name),quote=True)}">{html.escape(title)}</a></li>'
  page+='</ul></section>'
 page+='<p><a href="/">返回 DPL</a></p></html>\n'
 changes={env:replacement.encode(),root/'storage/theme/Stellar/resource-downloads.html':page.encode()}
 public=root/'public/theme/Stellar'
 assert public.is_dir(),'Public Stellar directory missing'
 public_env=public/'env.js';ps=public_env.read_text();pm=re.search(pat,ps,re.S);assert pm
 pb=pm[2]
 for key,value in settings.items():
  pb,n=re.subn(r'([\"\']?'+key+r'[\"\']?\s*:\s*)([\"\'])([^\"\']*)\2',lambda m:m[1]+m[2]+value+m[2],pb)
  assert n==1,'Public theme field missing: '+key
 changes[public_env]=(ps[:pm.start(2)]+pb+ps[pm.end(2):]).encode()
 changes[public/'resource-downloads.html']=page.encode()
 edits={
 'windows/clash-verge': [('https://github.com/clash-verge-rev/clash-verge-rev/releases',PAGE+'#windows'),('官方发布页','客户端下载页')],
 'macos/clash-verge': [('https://github.com/clash-verge-rev/clash-verge-rev/releases',PAGE+'#macos'),('官方发布页','客户端下载页')],
 'android/flclash': [('https://github.com/chen08209/FlClash/releases',PAGE+'#android'),('官方发布页','客户端下载页')],
 'linux/clash-verge': [('打开 Clash Verge Rev 官方发布页。在页面的 <strong>Assets</strong> 中选择安装包。','打开客户端下载页，按发行版和架构选择安装包。'),('，<code>armv7l</code> 选 armhf 或 armhfp','；当前 Clash Verge Rev 安装包仅提供 x64 与 ARM64'),('<strong>2.5.7、x86_64</strong>','<strong>x64</strong>'),('Clash.Verge_2.5.7_amd64.deb','clash-verge-rev-linux-x64.deb'),('Clash.Verge-2.5.7-1.x86_64.rpm','clash-verge-rev-linux-x64.rpm')]
 }
 for suffix,pairs in edits.items():
  p=root/'resources/views/tutorials'/f'{suffix}.blade.php';old=p.read_text();new=old
  for a,b in pairs:
   assert a in new or b in new,'Tutorial content changed; review required: '+suffix
   new=new.replace(a,b)
  changes[p]=new.encode()
 return {p:b for p,b in changes.items() if not p.exists() or p.read_bytes()!=b}
def verify_resources():
 base=Path('/www/wwwroot/resource-software/app/storage/app/private/software/current');pairs=set(FILES.values())|{(s,n) for _,_,s,n in CHOICES}
 for sw,name in sorted(pairs):
  cat=json.loads((base/sw/'catalog.json').read_text());asset=next(a for a in cat['assets'] if a['filename']==name)
  p=base/sw/name;assert p.is_file() and p.stat().st_size==asset['size']
  h=hashlib.sha256()
  with p.open('rb') as f:
   for chunk in iter(lambda:f.read(1024*1024),b''):h.update(chunk)
  assert h.hexdigest()==asset['sha256']
  req=urllib.request.Request(url(sw,name),headers={'Range':'bytes=0-31','User-Agent':'DPL-Resource-Acceptance/1.0'})
  with urllib.request.urlopen(req,timeout=30) as response:
   assert response.status==206 and response.geturl()==url(sw,name)
   with p.open('rb') as source:assert response.read()==source.read(32)
  print('verified',sw,name,cat['version'])
def apply(root,backup):
 changes=prepare(root)
 if not changes:print('No changes required');return
 snapshots={p:p.read_bytes() if p.exists() else None for p in changes}
 if root==Path('/www/wwwroot/dp-link.net'):verify_resources()
 for p,b in snapshots.items():
  assert (p.read_bytes() if p.exists() else None)==b,'Concurrent file change: '+str(p)
 backup.mkdir(parents=True,exist_ok=False);os.chmod(backup,0o700);manifest=[]
 for i,(p,b) in enumerate(changes.items()):
  old=snapshots[p];entry={'path':str(p),'before':digest(old) if old is not None else None,'after':digest(b),'backup':str(i)+'.bak' if old is not None else None}
  if old is not None:(backup/entry['backup']).write_bytes(old)
  manifest.append(entry)
 (backup/'manifest.json').write_text(json.dumps(manifest,indent=2))
 try:
  for p,b in changes.items():
   assert (p.read_bytes() if p.exists() else None)==snapshots[p],'Concurrent file change'
   stat=p.stat() if p.exists() else p.parent.stat();fd,tmp=tempfile.mkstemp(dir=p.parent,prefix='.resource-download-')
   with os.fdopen(fd,'wb') as f:f.write(b)
   os.chmod(tmp,stat.st_mode&0o777 if p.exists() else 0o644)
   if os.geteuid()==0:os.chown(tmp,stat.st_uid,stat.st_gid)
   os.replace(tmp,p)
  assert not prepare(root)
 except Exception:
  rollback(backup);raise
 print('Applied',len(changes),'files; backup',backup)
def rollback(backup):
 rows=json.loads((backup/'manifest.json').read_text())
 for row in rows:
  p=Path(row['path']);assert p.exists() and digest(p.read_bytes())==row['after'],'File changed since deployment; manual merge required'
 for row in rows:
  p=Path(row['path'])
  if row['backup']:p.write_bytes((backup/row['backup']).read_bytes())
  else:p.unlink()
 print('Rollback completed')
if __name__=='__main__':
 parser=argparse.ArgumentParser();parser.add_argument('mode',choices=['plan','apply','rollback']);parser.add_argument('--root',default='/www/wwwroot/dp-link.net');parser.add_argument('--backup');args=parser.parse_args();root=Path(args.root).resolve()
 assert root in [Path('/www/wwwroot/dp-link.net'),Path('/Users/hellcat/Documents/GemCodes/xb')],'DPL roots only'
 if args.mode=='rollback':rollback(Path(args.backup))
 elif args.mode=='plan':print('\n'.join(str(p) for p in prepare(root)))
 else:apply(root,Path(args.backup) if args.backup else Path('/root/codex-backups/dpl/resource-downloads-'+datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%SZ')))