business-assistant/scripts/test_dms_connection.py

"""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())