更新

2026-09-03 08:38:17 +08:00
parent 6cd4f1b1db
commit 842990b0e7
1853 changed files with 278406 additions and 361 deletions
@@ -0,0 +1,392 @@
+#!/usr/bin/env python3
+"""Scoped release of the labelled 100-user fixture batch on the existing BaoTa host.
+
+Run with the panel Python as root. DB credentials stay in memory; backups are
+root-only. No admin login, fabricated sessions, or OAuth migration is performed.
+"""
+import argparse
+import gzip
+import hashlib
+import json
+import os
+import pwd
+import re
+import shlex
+import shutil
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+import zipfile
+from pathlib import Path
+
+import pymysql
+
+DOMAIN = 'https://im.bchongw.com'
+BATCH = 'cn-adults-20260831-v1'
+BT = Path('/www/server/xingyu-im/bt')
+WEB = Path('/www/wwwroot/im.bchongw.com')
+MEDIA = Path('/www/wwwroot/xingyu-data/uploads')
+OLD_BINARY = '23ab263e16593b6f588e42bf480e08f364f8f46a12cf8c143077b767860b8b30'
+
+
+def digest(path):
+    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def env_values():
+    values = {}
+    for line in Path('/etc/xingyu-im-bt.env').read_text().splitlines():
+        line = line[7:] if line.startswith('export ') else line
+        if not line.strip() or line.lstrip().startswith('#'):
+            continue
+        key, value = line.split('=', 1)
+        parts = shlex.split(value)
+        assert len(parts) <= 1
+        values[key] = parts[0] if parts else ''
+    assert values['IM_ENV'] == 'production'
+    assert values['IM_HOST'] == '127.0.0.1' and values['IM_PORT'] == '18888'
+    return values
+
+
+def database(values):
+    match = re.fullmatch(r'([^:]+):(.*?)@tcp\(([^:]+):(\d+)\)/([^?]+)\?.*', values['IM_DB_DSN'])
+    assert match and match[3] == '127.0.0.1' and match[4] == '3307' and match[5] == 'im'
+    options = dict(host=match[3], port=int(match[4]), user=match[1], password=match[2], database=match[5], charset='utf8mb4', autocommit=True)
+    return pymysql.connect(**options), options
+
+
+def query(conn, sql, args=()):
+    with conn.cursor() as cursor:
+        cursor.execute(sql, args)
+        return cursor.fetchall()
+
+
+def fetch(path):
+    with urllib.request.urlopen(DOMAIN + path, timeout=20) as response:
+        return response.status, response.headers, response.read()
+
+
+def health():
+    for _ in range(100):
+        try:
+            status, _, data = fetch('/healthz')
+            if status == 200 and json.loads(data)['data']['status'] == 'ok':
+                return
+        except Exception:
+            pass
+        time.sleep(0.5)
+    raise RuntimeError('backend failed health check')
+
+
+def restart():
+    os.chdir('/www/server/panel')
+    sys.path.insert(0, '/www/server/panel')
+    sys.path.insert(0, '/www/server/panel/class')
+    import public
+    from projectModel.goModel import main as GoProject
+    request = public.dict_obj()
+    request.project_name = 'xingyu_im'
+    manager = GoProject()
+    project = manager.get_project_find('xingyu_im')
+    assert project and project['project_config']['is_power_on'] == 1
+    result = manager.restart_project(request)
+    if not result.get('status'):
+        recovered = manager.start_project(request)
+        if not recovered.get('status'):
+            raise RuntimeError('BaoTa could not start project')
+    health()
+
+
+def extract(archive, directory):
+    directory.mkdir(mode=0o755, parents=True, exist_ok=False)
+    with zipfile.ZipFile(str(archive)) as package:
+        seen = set()
+        for item in package.infolist():
+            name = item.filename.replace('\\', '/')
+            target = (directory / name).resolve()
+            assert name not in seen and name and ':' not in name
+            assert not name.startswith('/') and '..' not in Path(name).parts
+            assert os.path.commonpath([str(directory.resolve()), str(target)]) == str(directory.resolve())
+            assert (item.external_attr >> 16) & 0o170000 != 0o120000
+            seen.add(name)
+            if item.is_dir() or name.endswith('/'):
+                target.mkdir(parents=True, exist_ok=True)
+            else:
+                target.parent.mkdir(parents=True, exist_ok=True)
+                with package.open(item) as source, target.open('xb') as output:
+                    shutil.copyfileobj(source, output)
+
+
+def switch(entry, target, stamp):
+    next_link = entry.with_name(entry.name + '.next-' + stamp)
+    assert not next_link.exists() and not next_link.is_symlink()
+    os.symlink(str(target), str(next_link))
+    os.replace(str(next_link), str(entry))
+
+
+def prepare_static(archive, kind, stamp):
+    entry = WEB / ('admin' if kind == 'admin' else 'app')
+    releases = Path('/www/wwwroot/xingyu-' + kind + '/releases')
+    assert entry.is_symlink()
+    previous = entry.resolve()
+    assert previous.parent == releases
+    release = releases / ('test-users-' + stamp)
+    extract(archive, release)
+    index = (release / 'index.html').read_text()
+    assert ('/admin/jse/' if kind == 'admin' else '/app/assets/') in index
+    if kind == 'admin':
+        assert '"/admin/v1"' in (release / '_app.config.js').read_text()
+    # Keep hashed resources used by tabs opened before this release.
+    for folder in (['js', 'jse', 'css'] if kind == 'admin' else ['assets']):
+        if not (previous / folder).is_dir():
+            continue
+        for source in (previous / folder).rglob('*'):
+            if source.is_file():
+                target = release / source.relative_to(previous)
+                if not target.exists():
+                    target.parent.mkdir(parents=True, exist_ok=True)
+                    shutil.copy2(str(source), str(target))
+    www = pwd.getpwnam('www')
+    for path in [release] + list(release.rglob('*')):
+        assert not path.is_symlink()
+        os.chown(str(path), www.pw_uid, www.pw_gid)
+        os.chmod(str(path), 0o755 if path.is_dir() else 0o644)
+    return entry, previous, release
+
+
+def verify(conn):
+    rows = query(conn, "SELECT p.gender,COUNT(*) FROM users u JOIN user_profiles p ON p.user_id=u.id WHERE u.is_test=1 AND u.test_batch=%s AND u.deleted_at IS NULL GROUP BY p.gender ORDER BY p.gender", (BATCH,))
+    assert rows == ((1, 50), (2, 50)), rows
+    count = query(conn, "SELECT COUNT(*) FROM users WHERE is_test=1 AND test_batch=%s AND phone_hash IS NULL AND phone_cipher IS NULL AND password_hash='!TEST_PROFILE_NO_LOGIN'", (BATCH,))[0][0]
+    assert count == 100
+    avatars = query(conn, "SELECT DISTINCT p.avatar_url FROM users u JOIN user_profiles p ON p.user_id=u.id WHERE u.is_test=1 AND u.test_batch=%s", (BATCH,))
+    assert len(avatars) == 10
+    for (url,) in avatars:
+        assert url.startswith(DOMAIN + '/uploads/')
+        status, headers, body = fetch(url[len(DOMAIN):])
+        assert status == 200 and headers.get_content_type() == 'image/png' and body.startswith(b'\x89PNG\r\n\x1a\n')
+    for public_path in ['/admin/', '/app/']:
+        status, headers, data = fetch(public_path)
+        cache_control = ','.join(headers.get_all('Cache-Control', []))
+        assert status == 200 and b'<html' in data.lower() and 'no-store' in cache_control
+    for private_path in ['/admin/v1/users?userType=test', '/api/v1/discover/recommendations', '/ws']:
+        try:
+            fetch(private_path)
+        except urllib.error.HTTPError as error:
+            assert error.code == 401
+        else:
+            raise RuntimeError('private endpoint did not require authentication')
+    health()
+    return {'batch': BATCH, 'total': 100, 'male': 50, 'female': 50, 'avatarsHttp200': len(avatars), 'privateEndpointsRequireAuth': True}
+
+
+def publish(archive, expected):
+    assert os.geteuid() == 0
+    archive = Path(archive).resolve()
+    assert archive.parent == Path('/tmp') and re.fullmatch('[a-f0-9]{64}', expected)
+    assert digest(archive) == expected
+    values = env_values()
+    conn, db_options = database(values)
+    assert digest(BT / 'xingyu-api') == OLD_BINARY, 'online binary changed; re-audit before deployment'
+    assert query(conn, "SELECT COUNT(*) FROM schema_migrations WHERE version='027_app_oauth_login.sql'")[0][0] == 0, 'OAuth release changed; do not replace it'
+    assert query(conn, "SELECT COUNT(*) FROM users WHERE public_id LIKE 'TESTCN%%'")[0][0] == 0, 'test IDs already exist; inspect before retry'
+    assert query(conn, "SELECT config_value FROM system_configs WHERE config_key='storage.local.directory'")[0][0] == './uploads'
+    pid = int(Path('/var/tmp/gopids/xingyu_im.pid').read_text().strip())
+    assert Path('/proc/{}/cwd'.format(pid)).resolve() == BT
+    assert not (BT / 'uploads').exists() and not (BT / 'uploads').is_symlink()
+    assert MEDIA.is_dir()
+    health()
+    stamp = time.strftime('%Y%m%d-%H%M%S')
+    backup = Path('/www/backup/xingyu-test-users-' + stamp)
+    backup.mkdir(mode=0o700)
+    state = {
+        'stamp': stamp, 'archiveSha256': expected, 'previousBackendSha256': OLD_BINARY,
+        'usersBefore': query(conn, 'SELECT COUNT(*) FROM users')[0][0],
+        'previousAdmin': str((WEB / 'admin').resolve()), 'previousH5': str((WEB / 'app').resolve()),
+        'unchangedCounts': {table: query(conn, 'SELECT COUNT(*) FROM ' + table)[0][0] for table in ['orders', 'posts', 'im_messages', 'subscriptions', 'user_sessions', 'admin_users']},
+    }
+    # Dump only this application database using the matching MySQL 8 client.
+    dump_env = dict(os.environ, MYSQL_PWD=db_options['password'])
+    dump = subprocess.Popen(['/opt/mysql-8.4.11/bin/mysqldump', '-h', '127.0.0.1', '-P', '3307', '-u', db_options['user'], '--single-transaction', '--no-tablespaces', '--set-gtid-purged=OFF', '--routines', '--triggers', 'im'], env=dump_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    with gzip.open(str(backup / 'im.sql.gz'), 'wb') as output:
+        shutil.copyfileobj(dump.stdout, output)
+    _, errors = dump.communicate()
+    if dump.returncode:
+        raise RuntimeError('database backup failed: ' + errors.decode('utf-8', errors='replace'))
+    os.chmod(str(backup / 'im.sql.gz'), 0o600)
+    shutil.copy2(str(BT / 'xingyu-api'), str(backup / 'xingyu-api'))
+    shutil.copy2('/etc/xingyu-im-bt.env', str(backup / 'environment.env'))
+    os.chmod(str(backup / 'environment.env'), 0o600)
+    (backup / 'state.json').write_text(json.dumps(state, indent=2))
+    print('BACKUP=' + str(backup), flush=True)
+    stage = Path('/www/server/xingyu-im/releases/test-users-' + stamp)
+    extract(archive, stage)
+    os.chmod(str(stage), 0o700)
+    # Validate all deployed migrations, then add only the test-user migration.
+    recorded = dict(query(conn, 'SELECT version,checksum FROM schema_migrations'))
+    for file in (stage / 'migrations').glob('*.sql'):
+        if file.name == '028_test_users.sql':
+            continue
+        assert recorded.get(file.name) == digest(file), 'migration history mismatch: ' + file.name
+    migration = stage / 'migrations/028_test_users.sql'
+    assert migration.is_file() and '028_test_users.sql' not in recorded
+    sql = '\n'.join(line for line in migration.read_text().splitlines() if not line.lstrip().startswith('--'))
+    for statement in sql.split(';'):
+        if statement.strip():
+            query(conn, statement)
+    query(conn, 'INSERT INTO schema_migrations (version,checksum) VALUES (%s,%s)', (migration.name, digest(migration)))
+    shutil.copy2(str(migration), str(BT / 'migrations' / migration.name))
+    www = pwd.getpwnam('www')
+    os.chown(str(BT / 'migrations' / migration.name), www.pw_uid, www.pw_gid)
+    os.chmod(str(BT / 'migrations' / migration.name), 0o640)
+    # The existing local-storage config is relative to the BaoTa working dir.
+    # Preserve that config and make it point at the established persistent data dir.
+    os.symlink(str(MEDIA), str(BT / 'uploads'))
+    admin = prepare_static(stage / 'admin.zip', 'admin', stamp)
+    h5 = prepare_static(stage / 'h5.zip', 'h5', stamp)
+    switched = []
+    try:
+        new_binary = BT / ('xingyu-api.next-' + stamp)
+        shutil.copy2(str(stage / 'xingyu-api'), str(new_binary))
+        os.chown(str(new_binary), www.pw_uid, www.pw_gid)
+        os.chmod(str(new_binary), 0o750)
+        os.replace(str(new_binary), str(BT / 'xingyu-api'))
+        restart()
+        for entry, previous, release in [admin, h5]:
+            switch(entry, release, stamp)
+            switched.append((entry, previous))
+        for _, _, release in [admin, h5]:
+            prefix = '/admin/' if release == admin[2] else '/app/'
+            assert fetch(prefix)[2] == (release / 'index.html').read_bytes()
+        seeder = stage / 'seed-test-users'
+        os.chmod(str(seeder), 0o700)
+        seed_env = dict(os.environ, **values)
+        command = [str(seeder), '--apply', '--allow-production-test-data', '--confirm-database', 'im', '--public-base', DOMAIN + '/uploads', '--avatars-dir', str(stage / 'avatars'), '--media-dir', str(MEDIA)]
+        result = subprocess.run(command, env=seed_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
+        if result.returncode:
+            raise RuntimeError('seed failed: ' + result.stderr)
+        imported = json.loads(result.stdout)
+        assert imported['created'] == 100 and imported['male'] == 50 and imported['female'] == 50
+        report = verify(conn)
+        assert query(conn, 'SELECT COUNT(*) FROM users WHERE is_test=0')[0][0] == state['usersBefore']
+        for table, count in state['unchangedCounts'].items():
+            assert query(conn, 'SELECT COUNT(*) FROM ' + table)[0][0] == count, 'unexpected change to ' + table
+        repeat = subprocess.run(command, env=seed_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
+        assert repeat.returncode == 0
+        repeated = json.loads(repeat.stdout)
+        assert repeated['created'] == 0 and repeated['skipped'] == 100
+        state.update(report)
+        state.update({'backendSha256': digest(BT / 'xingyu-api'), 'admin': str(admin[2]), 'h5': str(h5[2]), 'idempotencySkipped': 100, 'seedResult': imported, 'backup': str(backup), 'stage': str(stage)})
+        (backup / 'result.json').write_text(json.dumps(state, ensure_ascii=False, indent=2))
+        print(json.dumps(state, ensure_ascii=False), flush=True)
+    except Exception:
+        # Keep additive schema and fixture data for diagnosis; never restore a
+        # full DB dump over users who may have registered during deployment.
+        for entry, previous in reversed(switched):
+            switch(entry, previous, stamp + '-rollback')
+        restore = BT / ('xingyu-api.rollback-' + stamp)
+        shutil.copy2(str(backup / 'xingyu-api'), str(restore))
+        os.chown(str(restore), www.pw_uid, www.pw_gid)
+        os.chmod(str(restore), 0o750)
+        os.replace(str(restore), str(BT / 'xingyu-api'))
+        restart()
+        print('APPLICATION_ROLLED_BACK; additive schema/data retained; backup=' + str(backup), flush=True)
+        raise
+    finally:
+        conn.close()
+
+
+def resume(backup_path):
+    """Resume this exact backed-up release after an application-only rollback.
+
+    No migration, initial import, environment change, or DB restoration occurs.
+    The original preflight deliberately rejects retries after a partial release.
+    """
+    assert os.geteuid() == 0
+    backup = Path(backup_path).resolve()
+    assert backup.parent == Path('/www/backup')
+    assert re.fullmatch(r'xingyu-test-users-\d{8}-\d{6}', backup.name)
+    state = json.loads((backup / 'state.json').read_text())
+    stamp = state['stamp']
+    assert backup.name == 'xingyu-test-users-' + stamp
+    stage = Path('/www/server/xingyu-im/releases/test-users-' + stamp)
+    archive = Path('/tmp/xingyu-test-users-20260831.zip')
+    assert digest(archive) == state['archiveSha256']
+    with zipfile.ZipFile(str(archive)) as package:
+        for name in ['xingyu-api', 'seed-test-users', 'admin.zip', 'h5.zip', 'migrations/028_test_users.sql']:
+            matching = [item for item in package.infolist() if item.filename.replace('\\', '/') == name]
+            assert len(matching) == 1
+            assert digest(stage / name) == hashlib.sha256(package.read(matching[0])).hexdigest()
+    assert digest(BT / 'xingyu-api') == OLD_BINARY == state['previousBackendSha256']
+    assert (BT / 'uploads').is_symlink() and (BT / 'uploads').resolve() == MEDIA
+    admin = (WEB / 'admin', Path(state['previousAdmin']), Path('/www/wwwroot/xingyu-admin/releases/test-users-' + stamp))
+    h5 = (WEB / 'app', Path(state['previousH5']), Path('/www/wwwroot/xingyu-h5/releases/test-users-' + stamp))
+    for entry, previous, release in [admin, h5]:
+        assert entry.is_symlink() and entry.resolve() == previous
+        assert previous.parent == release.parent
+        assert (release / 'index.html').is_file()
+    values = env_values()
+    conn, _ = database(values)
+    recorded = dict(query(conn, 'SELECT version,checksum FROM schema_migrations'))
+    assert '027_app_oauth_login.sql' not in recorded
+    assert recorded['028_test_users.sql'] == digest(stage / 'migrations/028_test_users.sql')
+    health()
+    # The complete fixture import must already exist. The seeder's transactional
+    # repeat check is a no-op and refuses partial/modified/conflicting batches.
+    assert query(conn, 'SELECT COUNT(*) FROM users WHERE is_test=1 AND test_batch=%s', (BATCH,))[0][0] == 100
+    command = [str(stage / 'seed-test-users'), '--apply', '--allow-production-test-data', '--confirm-database', 'im', '--public-base', DOMAIN + '/uploads', '--avatars-dir', str(stage / 'avatars'), '--media-dir', str(MEDIA)]
+    repeated = subprocess.run(command, env=dict(os.environ, **values), stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
+    assert repeated.returncode == 0, 'existing fixture batch failed repeat verification'
+    imported = json.loads(repeated.stdout)
+    assert imported['created'] == 0 and imported['skipped'] == 100
+    switched = []
+    www = pwd.getpwnam('www')
+    try:
+        new_binary = BT / ('xingyu-api.resume-' + stamp)
+        assert not new_binary.exists()
+        shutil.copy2(str(stage / 'xingyu-api'), str(new_binary))
+        os.chown(str(new_binary), www.pw_uid, www.pw_gid)
+        os.chmod(str(new_binary), 0o750)
+        os.replace(str(new_binary), str(BT / 'xingyu-api'))
+        restart()
+        for entry, previous, release in [admin, h5]:
+            switch(entry, release, stamp + '-resume')
+            switched.append((entry, previous))
+        assert fetch('/admin/')[2] == (admin[2] / 'index.html').read_bytes()
+        assert fetch('/app/')[2] == (h5[2] / 'index.html').read_bytes()
+        state.update(verify(conn))
+        assert query(conn, 'SELECT COUNT(*) FROM users WHERE is_test=0')[0][0] == state['usersBefore']
+        for table, count in state['unchangedCounts'].items():
+            assert query(conn, 'SELECT COUNT(*) FROM ' + table)[0][0] == count, 'unexpected change to ' + table
+        state.update({'backendSha256': digest(BT / 'xingyu-api'), 'admin': str(admin[2]), 'h5': str(h5[2]), 'idempotencySkipped': 100, 'repeatSeedResult': imported, 'backup': str(backup), 'stage': str(stage), 'resumedAfterCacheHeaderCheckFix': True})
+        (backup / 'result.json').write_text(json.dumps(state, ensure_ascii=False, indent=2))
+        print(json.dumps(state, ensure_ascii=False), flush=True)
+    except Exception:
+        for entry, previous in reversed(switched):
+            switch(entry, previous, stamp + '-resume-rollback')
+        restore = BT / ('xingyu-api.resume-rollback-' + stamp)
+        shutil.copy2(str(backup / 'xingyu-api'), str(restore))
+        os.chown(str(restore), www.pw_uid, www.pw_gid)
+        os.chmod(str(restore), 0o750)
+        os.replace(str(restore), str(BT / 'xingyu-api'))
+        restart()
+        print('APPLICATION_ROLLED_BACK; imported fixtures retained', flush=True)
+        raise
+    finally:
+        conn.close()
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('archive', nargs='?')
+    parser.add_argument('sha256', nargs='?')
+    parser.add_argument('--resume-backup')
+    args = parser.parse_args()
+    if args.resume_backup:
+        assert not args.archive and not args.sha256
+        resume(args.resume_backup)
+    else:
+        assert args.archive and args.sha256
+        publish(args.archive, args.sha256)