#!/usr/bin/env python3 """Explicit project-local Ozon client lifecycle. Python 3.10+, Linux/macOS/WSL.""" import argparse import contextlib import fcntl import getpass import hashlib import json import os from pathlib import Path import re import shutil import stat import subprocess import sys import tempfile import urllib.error import urllib.request import uuid ORIGIN = 'https://x-mp.ru' RELEASE_PATH = '/docs/ozon/downloads/release.json' LEGACY_FILES = {'install.py', 'xmp_agent_mcp.py', 'xmp_updater.py', 'xmp_ozon_mcp.py', 'xmp-ozon-cabinet/SKILL.md'} SKILLS = ('xmp-ozon-cabinet', 'xmp-ozon-analytics', 'xmp-ozon-memory', 'xmp-ozon-repricer', 'xmp-ozon-bidder', 'xmp-ozon-communications', 'xmp-ozon-cards', 'xmp-support') FILES = LEGACY_FILES | {name + '/SKILL.md' for name in SKILLS} PREVIOUS_FILES = FILES - {'xmp-support/SKILL.md'} class InstallError(Exception): pass class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def fetch(path, token=None): headers = {'Accept': 'application/json' if path.endswith('.json') or token else 'application/octet-stream', 'Cache-Control': 'no-cache'} if token: headers['Authorization'] = 'Bearer ' + token req = urllib.request.Request(ORIGIN + path, headers=headers) try: with urllib.request.build_opener(NoRedirect).open(req, timeout=25) as response: body = response.read(2 * 1024 * 1024 + 1) except (OSError, urllib.error.URLError): raise InstallError('Сервер недоступен или отказал в доступе; настройки сохранены.') from None if len(body) > 2 * 1024 * 1024: raise InstallError('Превышен размер загрузки.') return body def validate(manifest): if manifest.get('product') != 'xmp-ozon-client' or manifest.get('schema_version') != 1: raise InstallError('Неизвестный формат выпуска.') if not re.fullmatch(r'\d+\.\d+\.\d+', manifest.get('version', '')): raise InstallError('Некорректная версия.') allowed = FILES if tuple(map(int, manifest['version'].split('.'))) >= (0, 6, 1) else PREVIOUS_FILES if manifest['version'] == '0.6.0' else LEGACY_FILES if set(manifest.get('files', {})) != allowed: raise InstallError('Неожиданный состав выпуска.') for name, item in manifest['files'].items(): if item.get('path') != '/docs/ozon/downloads/' + manifest['version'] + '/' + name: raise InstallError('Некорректный путь загрузки.') if not re.fullmatch('[0-9a-f]{64}', item.get('sha256', '')): raise InstallError('Некорректная контрольная сумма.') if type(item.get('size')) is not int or not 0 < item['size'] <= 2 * 1024 * 1024: raise InstallError('Некорректный размер файла.') return manifest def digest(body): return hashlib.sha256(body).hexdigest() def release_id(manifest): return manifest['version'] + '-' + digest(json.dumps(manifest, sort_keys=True).encode())[:16] def regular(path): if path.is_symlink() or (path.exists() and not path.is_file()): raise InstallError('Ожидался обычный файл: ' + str(path)) def atomic(path, body): regular(path) fd, tmp = tempfile.mkstemp(prefix='.xmp-write-', dir=path.parent) try: with os.fdopen(fd, 'wb') as out: out.write(body) out.flush() os.fsync(out.fileno()) os.replace(tmp, path) finally: if os.path.exists(tmp): os.unlink(tmp) def directory(path): if path.is_symlink() or (path.exists() and not path.is_dir()): raise InstallError('Ожидался обычный каталог: ' + str(path)) path.mkdir(mode=0o700, parents=True, exist_ok=True) def read_json(path): regular(path) return json.loads(path.read_text()) if path.exists() else {} def verified_release(state, target): if not re.fullmatch(r'releases/\d+\.\d+\.\d+-[a-f0-9]{16}', target): raise InstallError('Некорректная ссылка выпуска.') folder = state / target if folder.is_symlink(): raise InstallError('Каталог выпуска не должен быть ссылкой.') manifest = validate(read_json(folder / 'manifest.json')) if target != 'releases/' + release_id(manifest): raise InstallError('Выпуск не соответствует манифесту.') for name, item in manifest['files'].items(): path = folder / name if path.parent.is_symlink(): raise InstallError('Изменён путь файла выпуска.') regular(path) body = path.read_bytes() if len(body) != item['size'] or digest(body) != item['sha256']: raise InstallError('Изменён файл установленного выпуска: ' + name) return manifest def current_target(state): current = state / 'current' if not current.is_symlink(): if current.exists(): raise InstallError('current должен быть управляемой ссылкой.') return None target = os.readlink(current) verified_release(state, target) return target def switch(state, target): tmp = state / ('.current-' + uuid.uuid4().hex) tmp.symlink_to(target) os.replace(tmp, state / 'current') def check_token(token, account): if not re.fullmatch(r'xmp_(live|special)_[A-Za-z0-9_-]{16,512}', token): raise InstallError('Нужен токен агента X-MP, а не ключ Ozon.') identity = json.loads(fetch('/cabinet/api/agent/me', token)) context = json.loads(fetch('/cabinet/api/agent/context', token)) if identity.get('marketplace_account_id') != account or context.get('marketplace_account_id') != account: raise InstallError('Токен принадлежит другому магазину.') if context.get('cabinet', {}).get('platform') != 'ozon': raise InstallError('Токен не принадлежит магазину Ozon.') def get_token(args, state): path = Path(args.key_file).expanduser() if args.key_file else state / 'secrets/agent.key' if path.exists() or path.is_symlink(): regular(path) if stat.S_IMODE(path.stat().st_mode) & 0o077: raise InstallError('Файл токена должен иметь права 0600.') return path.read_text().strip() if args.key_file: raise InstallError('Файл токена не найден.') if not sys.stdin.isatty(): raise InstallError('Укажите --key-file или используйте prepare без токена.') return getpass.getpass('Токен агента X-MP (ввод скрыт): ').strip() def config_content(root, state, account): directory(root / '.codex') path = root / '.codex/config.toml' regular(path) old = path.read_text() if path.exists() else '' start, end = '# BEGIN X-MP OZON MANAGED', '# END X-MP OZON MANAGED' if old.count(start) != old.count(end) or old.count(start) > 1: raise InstallError('Повреждены границы настроек Ozon.') clean = re.sub(re.escape(start) + r'.*?' + re.escape(end) + r'\n?', '', old, flags=re.S) if re.search(r'\[mcp_servers\.[^\n]*xmp-ozon-', clean): raise InstallError('Найдена прежняя настройка Ozon; сначала перенесите её в отдельный проект.') server = 'xmp-ozon-' + account[:8] block = f'''{start} [mcp_servers.{server}] command = {json.dumps(sys.executable)} args = [{json.dumps(str(state / 'current/xmp_ozon_mcp.py'))}] cwd = {json.dumps(str(root))} startup_timeout_sec = 30 [mcp_servers.{server}.env] XMP_AGENT_KEY_FILE = {json.dumps(str(state / 'secrets/agent.key'))} XMP_EXPECTED_ACCOUNT_ID = {json.dumps(account)} XMP_ACCOUNT_BASE_URL = "https://x-mp.ru" {end} ''' new = clean.rstrip() + '\n\n' + block try: import tomllib except ImportError: try: import tomli as tomllib except ImportError: tomllib = None if tomllib: tomllib.loads(new) return path, old.encode(), new.encode() def skill_links(root, state, manifest): names = sorted(name.split('/')[0] for name in manifest['files'] if name.endswith('/SKILL.md')) directory(root / '.agents') directory(root / '.agents/skills') links = [] for name in names: link = root / '.agents/skills' / name target = str(state / 'current' / name) if (link.exists() or link.is_symlink()) and (not link.is_symlink() or os.readlink(link) != target): raise InstallError('Существующий навык изменять автоматически нельзя: ' + name) links.append((link, target)) return links def doctor(root, state, metadata, current): if not current or not metadata.get('connected_config'): raise InstallError('Сначала выполните install с токеном магазина.') manifest = verified_release(state, current) token_path = state / 'secrets/agent.key' regular(token_path) if stat.S_IMODE(token_path.stat().st_mode) & 0o077: raise InstallError('Файл токена должен иметь права 0600.') calls = [('agent.me', {}), ('ozon.status', {}), ('ozon.records', {'table':'products','limit':1}), ('tasks.search', {'query':'__xmp_connection_probe__','limit':1}), ('ozon.config.get', {}), ('ozon.schedule.get', {})] messages = [{'jsonrpc':'2.0','id':0,'method':'tools/list'}] + [ {'jsonrpc':'2.0','id':i,'method':'tools/call','params':{'name':name,'arguments':args}} for i,(name,args) in enumerate(calls,1)] env = {**os.environ, 'XMP_AGENT_KEY_FILE':str(token_path), 'XMP_EXPECTED_ACCOUNT_ID':metadata['account_id'], 'XMP_ACCOUNT_BASE_URL':ORIGIN} try: run = subprocess.run([sys.executable,str(state/'current/xmp_ozon_mcp.py')], input='\n'.join(map(json.dumps,messages))+'\n', text=True, capture_output=True,env=env,timeout=240) replies = [json.loads(line) for line in run.stdout.splitlines()] if run.returncode or len(replies)!=len(messages): raise InstallError('Проверка MCP не завершилась; проверьте доступ к серверу.') except (subprocess.TimeoutExpired, ValueError): raise InstallError('Проверка MCP не завершилась; проверьте доступ к серверу.') from None checks = [] for (name,_), reply in zip(calls,replies[1:]): checks.append({'tool':name,'ok':not bool(reply.get('error') or reply.get('result',{}).get('isError'))}) identity = replies[1].get('result',{}).get('structuredContent',{}) same_account = identity.get('marketplace_account_id')==metadata['account_id'] and identity.get('platform')=='ozon' scopes = {grant['scope'] for grant in identity.get('grants',[]) if isinstance(grant,dict) and 'scope' in grant} available_skills = [name for name in SKILLS if (root/'.agents/skills'/name/'SKILL.md').is_file()] return {'status':'connected' if same_account and all(x['ok'] for x in checks) else 'needs-attention', 'account_id':metadata['account_id'],'version':manifest['version'], 'tools':len(replies[0].get('result',{}).get('tools',[])), 'skills':available_skills, 'data_status':replies[2].get('result',{}).get('structuredContent',{}).get('status'), 'checks':checks,'can_prepare_plans':'operations.plan' in scopes, 'can_save_memory':'tasks.write' in scopes, 'hint':'Для настройки и сохранения задач выберите токен «Чтение и подготовка планов».' if 'operations.plan' not in scopes else 'Настройки доступны; запись на Ozon требует подтверждения конкретного плана.', 'marketplace_writes':0,'business_cards_created':0} def execute(args): root = Path(args.project_root).expanduser().resolve() directory(root) state = root / '.xmp-ozon' directory(state) lock_path = state / '.lock' regular(lock_path) with lock_path.open('a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) directory(state / 'releases') metadata = read_json(state / 'state.json') current = current_target(state) account = str(uuid.UUID(args.account_id)) if args.account_id else metadata.get('account_id') if metadata.get('account_id') and account != metadata['account_id']: raise InstallError('Этот проект уже связан с другим магазином. Используйте другой каталог.') if args.action == 'doctor': return doctor(root, state, metadata, current) if args.action == 'rotate-token': if not metadata.get('connected_config') or not account: raise InstallError('Сначала выполните install с токеном магазина.') if not args.key_file: raise InstallError('Для замены токена укажите --key-file с новым токеном.') token = get_token(args, state) check_token(token, account) directory(state / 'secrets') atomic(state / 'secrets/agent.key', token.encode()) return {'status': 'token-replaced', 'old_token_revocation': 'Отозвать старый токен в личном кабинете.', 'restart_required': True} if args.action == 'rollback': target = metadata.get('previous') if not target: raise InstallError('Нет предыдущего выпуска для отката.') rollback_manifest = verified_release(state, target) rollback_links = skill_links(root, state, rollback_manifest) if metadata.get('connected_config') else [] switch(state, target) for link, target_path in rollback_links: if not link.is_symlink(): link.symlink_to(target_path, target_is_directory=True) for name in SKILLS: link = root / '.agents/skills' / name if name + '/SKILL.md' not in rollback_manifest['files'] and link.is_symlink() and os.readlink(link) == str(state/'current'/name): link.unlink() metadata['previous'] = current atomic(state / 'state.json', json.dumps(metadata).encode()) return {'status': 'rolled-back', 'release': target, 'restart_required': True} manifest = validate(json.loads(fetch(RELEASE_PATH))) target = 'releases/' + release_id(manifest) installed = verified_release(state, current) if current else None older = installed and tuple(map(int, manifest['version'].split('.'))) < tuple(map(int, installed['version'].split('.'))) if args.action == 'check': return {'status': 'server-older' if older else 'update-available' if target != current else 'up-to-date', 'installed': installed['version'] if installed else None, 'available': manifest['version']} if older: raise InstallError('Сервер предлагает старую версию; используйте явный rollback.') if args.action == 'update' and not current: raise InstallError('Сначала выполните prepare или install.') token = None config = None links = skill_links(root, state, manifest) if args.action == 'install' or metadata.get('connected_config') else [] if args.action == 'install': if not account: raise InstallError('Нужен --account-id магазина Ozon из X-MP.') token = get_token(args, state) check_token(token, account) config = config_content(root, state, account) directory(state / 'secrets') regular(state / 'secrets/agent.key') destination = state / target if destination.exists() or destination.is_symlink(): verified_release(state, target) else: temp = Path(tempfile.mkdtemp(prefix='stage-', dir=state)) try: for name, item in manifest['files'].items(): body = fetch(item['path']) if len(body) != item['size'] or digest(body) != item['sha256']: raise InstallError('Контрольная сумма загрузки не совпадает: ' + name) path = temp / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(body) path.chmod(0o600) (temp / 'manifest.json').write_text(json.dumps(manifest)) os.replace(temp, destination) finally: if temp.exists(): shutil.rmtree(temp) atomic(state / '.gitignore', b'*\n!.gitignore\n') if config: path, old, new = config # Persist a recoverable backup before activation; never include the token. if old and old != new: atomic(state / ('config-backup-' + uuid.uuid4().hex + '.toml'), old) atomic(state / 'secrets/agent.key', token.encode()) if current != target: switch(state, target) metadata['previous'] = current for link, target_path in links: if not link.is_symlink(): link.symlink_to(target_path, target_is_directory=True) if config: atomic(config[0], config[2]) metadata.update(account_id=account, connected_config=True) atomic(state / 'state.json', json.dumps(metadata).encode()) return {'status': 'client-configured' if config else 'client-files-ready', 'version': manifest['version'], 'restart_required': bool(config or current != target), 'connection': 'Проверены принадлежность токена и Ozon; данные требуют readiness.' if config else 'Токен не запрашивался; подключение магазина не проверялось.', 'capabilities': 'Чтение Ozon, планы и проверка готовности блоков. Точные операции требуют активации на сервере и подтверждения владельца.'} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('action', choices=['prepare', 'install', 'check', 'update', 'rollback', 'rotate-token', 'doctor']) parser.add_argument('--project-root', default='.') parser.add_argument('--account-id') parser.add_argument('--key-file', help='Private 0600 file; never pass the token value in the command') args = parser.parse_args() try: print(json.dumps(execute(args), ensure_ascii=False, indent=2)) except (InstallError, OSError, ValueError, KeyError) as exc: # Avoid printing remote response bodies, token values or full tracebacks. message = str(exc) if isinstance(exc, InstallError) else 'Ошибка файлов, конфигурации или формата ответа; изменения проверьте перед повтором.' raise SystemExit(message) from None if __name__ == '__main__': main()