"""Read-only DMS metadata diagnosis using the project's production auth/config contract.""" import argparse from contextlib import redirect_stderr import json import os from pathlib import Path import socket import ssl import sys import time from http.client import HTTPException from urllib.error import HTTPError, URLError from urllib.parse import urlencode, quote, quote_plus, urlsplit from urllib.request import Request, build_opener, ProxyHandler ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) try: from src.dms_schema import ALIASES, load_schema, names, expected_fields from src.step5_session_analysis.dms import DmsClient, DmsError, NoRedirect, clean, load_dms_config from src.step3_question_answer.dms_company import BASIC_FIELDS, HONOR_FIELDS except (OSError, ValueError, KeyError, TypeError, ImportError) as exc: print(json.dumps({'ok': False, 'stage': 'project_import', 'error_type': type(exc).__name__, 'hint': '请将文件放入现有项目scripts目录，检查项目依赖和config/dms_schema.json；不输出配置正文。'}, ensure_ascii=False)) raise SystemExit(1) from None CONFIG_KEYS = {'DMS_BASE_URL', 'DMS_LOGIN_URL', 'DMS_USERNAME', 'DMS_PASSWORD', 'DMS_TOKEN', 'DMS_RECORDS_COLUMN_ID', 'DMS_SESSIONS_COLUMN_ID', 'DMS_DEMANDS_COLUMN_ID', 'DMS_STRICT_SCOPE', 'DMS_ALLOWED_SCOPES_JSON', 'DMS_SCAN_TIMEOUT_SECONDS', 'AUTH_PORT', 'DMS_PORT'} REQUIRED = { 'records': {'c_session_id', 'c_question', 'c_answer', 'c_created_at'}, 'sessions': {'c_session_id', 'qpyszx', 'qyzt', 'rzqpyx', 'c_focus_points', 'c_settlement_intent_level', 'c_core_needs', 'jygjbm', 'summary', 'c_description', 'c_title', 'c_created_at', 'c_updated_at'}, 'demands': {'c_session_id', 'c_word', 'c_created_at'}, 'basic': {'c_' + key for key in BASIC_FIELDS} | {'c_area', 'c_revoke_info', 'c_original_name', 'c_designated_representative_list'}, 'honor': {'c_' + key for key in HONOR_FIELDS} | {'c_credit_code'}, } class DiagnosticTransport: def __init__(self, timeout=20, no_env_proxy=False): self.timeout = timeout self.calls = [] self.secrets = set() self.opener = build_opener(NoRedirect(), *([ProxyHandler({})] if no_env_proxy else [])) def __call__(self, url, form, headers): self.secrets.update(str(v) for k, v in headers.items() if k.lower() == 'token' and v) row = {'endpoint': url, 'method': 'POST', 'form_keys': sorted(form), 'token_header_present': bool(headers.get('token')), 'form': {key: '[REDACTED]' if key in {'userName', 'password'} else value for key, value in form.items()}, 'headers': {key: '[REDACTED]' if key.lower() == 'token' else value for key, value in headers.items()}, 'timeout_seconds': self.timeout} self.calls.append(row) emit('request_start', row, self.secrets) started = time.monotonic() try: request = Request(url, urlencode(form).encode(), headers) with self.opener.open(request, timeout=self.timeout) as response: row.update(http_status=response.status, content_type=response.headers.get('Content-Type', '')) body = response.read(4_000_001) row['received_bytes'] = len(body) if len(body) > 4_000_000: raise DmsError('dms_response_too_large') if getattr(response, 'length', None) not in (None, 0): raise DmsError('dms_response_incomplete') try: envelope = json.loads(body) except (ValueError, UnicodeError): raise DmsError('dms_invalid_json') from None if isinstance(envelope, dict): code = envelope.get('code') row['provider_code'] = code if type(code) is int else 'invalid_or_missing' if 'userName' in form and 'password' in form and code == 200: token = envelope.get('message') if isinstance(token, str) and token: self.secrets.add(token) return envelope except HTTPError as exc: row.update(http_status=exc.code, content_type=exc.headers.get('Content-Type', '')) exc.close() code = {401: 'dms_token_invalid', 403: 'dms_permission_denied'}.get(exc.code, 'dms_http_error') row['error'] = code raise DmsError(code) from None except (URLError, OSError, HTTPException) as exc: reason = exc.reason if isinstance(exc, URLError) else exc code = ('dns_failed' if isinstance(reason, socket.gaierror) else 'tls_failed' if isinstance(reason, ssl.SSLError) else 'timeout' if isinstance(reason, TimeoutError) else 'connection_refused' if isinstance(reason, ConnectionRefusedError) else 'response_incomplete' if isinstance(reason, HTTPException) else 'connection_failed') row.update(error=code, error_type=type(reason).__name__) if isinstance(getattr(reason, 'errno', None), int): row['errno'] = reason.errno raise DmsError('dms_' + code) from None except DmsError as exc: row['error'] = str(exc) raise finally: row['elapsed_seconds'] = round(time.monotonic() - started, 3) emit('request_end', row, self.secrets) def field_differences(schema, alias, model): raw = model.get('fieldList') if isinstance(model, dict) else None try: raw = json.loads(raw) if isinstance(raw, str) else raw if not isinstance(raw, dict): raise ValueError raw = {k: json.loads(v) if isinstance(v, str) else v for k, v in raw.items()} if not all(isinstance(v, dict) for v in raw.values()): raise ValueError except (ValueError, TypeError): raise DmsError('dms_invalid_field_list') from None spec = schema['columns'][alias] expected = expected_fields(schema, alias) rows = [] for logical, actual in spec['fields'].items(): found = raw.get(actual, {}).get('type') rows.append({'logical_field': logical, 'actual_field': actual, 'expected_type': expected[actual], 'actual_type': found if isinstance(found, str) else None, 'status': 'missing' if actual not in raw else 'type_mismatch' if found != expected[actual] else 'matched'}) missing = sorted(REQUIRED[alias] - set(spec['fields'])) return {'ok': not missing and all(row['status'] == 'matched' for row in rows), 'missing_local_mappings': missing, 'fields': rows, 'remote_field_count': len(raw), 'hint': '只核对配置字段；额外远端字段不判失败。不验证业务记录值或写权限。'} def inspect_column(client, alias): config = client.config spec = config.schema['columns'][alias] identity = {'records': config.records_id, 'sessions': config.sessions_id, 'demands': config.demands_id}.get(alias, spec['id']) report = {'ok': False, 'column_id': identity, 'expected_name': spec['name'], 'stage': 'column'} try: column = client.call('/column/getColumnById', {'id': identity}) if not isinstance(column, dict): raise DmsError('dms_invalid_column') report.update(actual_name=clean(column.get('title')), actual_column_type=column.get('type')) if clean(column.get('title')) != spec['name'] or column.get('type') != 1: raise DmsError('dms_column_mismatch') report['stage'] = 'parent' parent_id = column.get('parentId') if type(parent_id) is not int or parent_id <= 0: raise DmsError('dms_invalid_parent_id') parent = client.call('/column/getColumnById', {'id': parent_id}) report.update(parent_id=parent_id, expected_parent=config.schema['parent'], actual_parent=clean(parent.get('title')) if isinstance(parent, dict) else None) if report['actual_parent'] != config.schema['parent']: raise DmsError('dms_parent_mismatch') report['stage'] = 'model' model_id = column.get('modelId') if type(model_id) is not int or model_id <= 0: raise DmsError('dms_invalid_model_id') report['model_id'] = model_id model = client.call('/model/getModelById', {'modelId': model_id}) report['stage'] = 'fields' report.update(field_differences(config.schema, alias, model)) report['code'] = 'matched' if report['ok'] else 'dms_schema_mismatch' except DmsError as exc: report['code'] = str(exc) return report def redact_report(value, secrets): text = json.dumps(value, ensure_ascii=False, indent=2) for secret in sorted({s for s in secrets if s}, key=len, reverse=True): for variant in {secret, quote(secret, safe=''), quote_plus(secret), json.dumps(secret, ensure_ascii=False)[1:-1], json.dumps(secret)[1:-1]}: text = text.replace(variant, '[REDACTED]') return text def emit(event, details, secrets=()): print(redact_report({'event': event, **details}, secrets), file=sys.stderr, flush=True) def validate_override(value): if value is None: return try: parts = urlsplit(value) if (not value or any(c.isspace() or ord(c) < 33 for c in value) or '\\' in value or parts.scheme not in ('http', 'https') or not parts.hostname or parts.username is not None or parts.password is not None or parts.query or parts.fragment or not parts.path or parts.path == '/' or (parts.port is not None and not 1 <= parts.port <= 65535)): raise ValueError except (ValueError, TypeError): raise DmsError('invalid_endpoint_override') from None def run(root, aliases, *, env_file_only=False, timeout=20, no_env_proxy=False, api_url=None, login_url=None): wire = DiagnosticTransport(timeout, no_env_proxy) report = {'ok': False, 'env_file_exists': (root / '.env').is_file(), 'schema_file_exists': (root / 'config/dms_schema.json').is_file(), 'env_file_only': env_file_only, 'environment_override_keys': [] if env_file_only else sorted(CONFIG_KEYS & os.environ.keys()), 'environment_proxy_keys': sorted(k for k in os.environ if k.lower() in {'http_proxy', 'https_proxy', 'all_proxy', 'no_proxy'}), 'use_environment_proxy': not no_env_proxy, 'checks': {}, 'requests': wire.calls} try: report['stage'] = 'endpoint_arguments' validate_override(api_url) validate_override(login_url) emit('local_configuration', {'env_file': str(root / '.env'), 'schema_file': str(root / 'config/dms_schema.json'), 'env_file_only': env_file_only, 'selected_columns': list(aliases), 'environment_override_keys': report['environment_override_keys']}) report['stage'] = 'local_schema' schema = load_schema(root / 'config/dms_schema.json') report['stage'] = 'configuration_and_scope' config = load_dms_config(root / '.env', environ={} if env_file_only else None, schema_path=root / 'config/dms_schema.json', scope_names=names(schema, aliases)) wire.secrets.update(x for x in (config.token, config.username, config.password) if x) client = DmsClient(config, transport=wire) # Instance-only overrides: compatible with the old DmsConfig without login_url. # Both call() retries and _login() use these attributes, never write configuration. if api_url is not None: client.api = api_url.rstrip('/') if login_url is not None: client.login_url = login_url report.update(api_endpoint=client.api, login_endpoint=client.login_url, api_url_source='command_line' if api_url is not None else 'configuration', login_url_source='command_line' if login_url is not None else 'configuration', strict_scope=config.strict, token_configured=bool(config.token), login_credentials_configured=bool(config.username and config.password)) emit('connection_parameters', {'base_url': config.base_url, 'api_endpoint': client.api, 'login_endpoint': client.login_url, 'AUTH_PORT': config.auth_port, 'DMS_PORT': config.dms_port, 'api_url_source': report['api_url_source'], 'login_url_source': ('command_line' if login_url is not None else 'DMS_LOGIN_URL' if getattr(config, 'login_url', '') else 'default_route'), 'timeout_seconds': timeout, 'use_environment_proxy': not no_env_proxy, 'environment_proxy_keys': report['environment_proxy_keys'], 'strict_scope': config.strict, 'allowed_scope_names': list(config.scope_names), 'token': '[REDACTED]' if config.token else '[NOT CONFIGURED]', 'username': '[REDACTED]' if config.username else '[NOT CONFIGURED]', 'password': '[REDACTED]' if config.password else '[NOT CONFIGURED]'}, wire.secrets) report['stage'] = 'authentication' if not client.token: client._login() # Same proven login contract; no content writes. for alias in aliases: emit('column_start', {'alias': alias, 'expected_name': schema['columns'][alias]['name'], 'expected_parent': schema['parent'], 'column_id': {'records': config.records_id, 'sessions': config.sessions_id, 'demands': config.demands_id}.get(alias, schema['columns'][alias]['id'])}, wire.secrets) start = len(wire.calls) report['checks'][alias] = inspect_column(client, alias) report['checks'][alias]['request_indexes'] = list(range(start, len(wire.calls))) emit('column_end', {'alias': alias, **report['checks'][alias]}, wire.secrets) report.update(stage='finished', ok=all(row['ok'] for row in report['checks'].values())) except DmsError as exc: report['code'] = str(exc) except (ValueError, KeyError, TypeError, OSError) as exc: report.update(code='invalid_local_configuration', error_type=type(exc).__name__) return report, wire.secrets def compact_report(report, selected_count): checks = report.get('checks', {}) failed = {alias: row for alias, row in checks.items() if not row['ok']} output = {'ok': report['ok'], 'checked_columns': len(checks), 'passed_columns': len(checks) - len(failed), 'failed_columns': len(failed), 'not_checked_columns': selected_count - len(checks)} if report['ok']: return output for key in ('stage', 'code', 'error_type', 'api_endpoint', 'login_endpoint'): if key in report: output[key] = report[key] output['failed_checks'] = {} indexes = set() for alias, row in failed.items(): details = {k: v for k, v in row.items() if k not in ('fields', 'hint', 'request_indexes', 'remote_field_count')} if 'fields' in row: details['fields'] = [field for field in row['fields'] if field['status'] != 'matched'] output['failed_checks'][alias] = details if row.get('request_indexes'): indexes.add(row['request_indexes'][-1]) requests = report.get('requests', []) indexes.update(i for i, row in enumerate(requests) if row.get('error') or ('provider_code' in row and row['provider_code'] != 200)) if not checks and requests: indexes.add(len(requests) - 1) output['related_requests'] = [{'request_index': i, **requests[i]} for i in sorted(indexes)] return output def main(): parser = argparse.ArgumentParser(description='DMS只读连接、鉴权、五栏目和字段类型诊断；不读取业务记录') parser.add_argument('--env-file-only', action='store_true') parser.add_argument('--column', choices=['all', *ALIASES], default='all') parser.add_argument('--timeout', type=float, default=20) parser.add_argument('--verbose', action='store_true', help='显示全部请求过程和正常字段；默认只显示汇总及异常') parser.add_argument('--api-url', help='仅本次诊断覆盖完整DMS数据入口，不受DMS_PORT覆盖') parser.add_argument('--login-url', help='仅本次诊断覆盖完整登录地址，不受AUTH_PORT覆盖') parser.add_argument('--no-env-proxy', action='store_true', help='仅用于对比：禁用系统HTTP代理，默认与正式客户端一致') args = parser.parse_args() if not 0 < args.timeout <= 120: parser.error('timeout须在0到120秒之间') aliases = ALIASES if args.column == 'all' else (args.column,) with open(os.devnull, 'w') as sink, redirect_stderr(sys.stderr if args.verbose else sink): report, secrets = run(ROOT, aliases, env_file_only=args.env_file_only, timeout=args.timeout, no_env_proxy=args.no_env_proxy, api_url=args.api_url, login_url=args.login_url) print(redact_report(report if args.verbose else compact_report(report, len(aliases)), secrets)) return 0 if report['ok'] else 1 if __name__ == '__main__': raise SystemExit(main())