business-assistant/scripts/test_qcc_proxy.py

"""Standalone QCC GET proxy diagnostic. No dependencies or company data output."""
import argparse
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import hashlib
import html
from http.client import HTTPConnection, HTTPSConnection, IncompleteRead
import json
import os
import re
from pathlib import Path
import time
from urllib.parse import parse_qsl, quote, quote_plus, urlencode, urlsplit

URL_KEYS = {'search': 'QCC_SEARCH_URL', 'basic': 'QCC_BASIC_URL', 'honor': 'QCC_HONOR_URL'}
CONFIG_KEYS = set(URL_KEYS.values()) | {'QCC_KEY', 'QCC_SECRET_KEY', 'QCC_TIMEOUT_SECONDS'}


def redact(text, secrets):
    text = html.unescape(str(text))
    for secret in sorted({s for s in secrets if s}, key=len, reverse=True):
        variants = {secret, quote(secret, safe=''), quote_plus(secret),
                    json.dumps(secret, ensure_ascii=True)[1:-1],
                    json.dumps(secret, ensure_ascii=False)[1:-1]}
        for variant in sorted(variants, key=len, reverse=True):
            text = re.sub(re.escape(variant), '[REDACTED]', text, flags=re.IGNORECASE)
    text = re.sub(r'(?i)((?:proxyToken|QCC_SECRET_KEY|QCC_KEY|Token|searchKey|keyword|key)[\"\s\x27]*[:=][\"\s\x27]*)[^&\s\"\x27<>]+', r'\1[REDACTED]', text)
    return text


def error_body(body, secrets):
    text = body.decode('utf-8', errors='replace')
    try:
        value = json.loads(text)
        def sanitize(item):
            if isinstance(item, dict):
                return {k: '[OMITTED]' if k.lower() in {'result', 'data'} else sanitize(v)
                        for k, v in item.items()}
            if isinstance(item, list):
                return [sanitize(v) for v in item]
            return item
        text = json.dumps(sanitize(value), ensure_ascii=False)
    except (ValueError, RecursionError):
        pass
    text = redact(text, secrets)
    return {'error_body_preview': text[:2000], 'error_body_truncated': len(text) > 2000}


def load_settings(path, *, env_file_only=False):
    values = {}
    if path.is_file():
        for line in path.read_text(encoding='utf-8-sig').splitlines():
            line = line.strip()
            if line.startswith('export '):
                line = line[7:]
            key, sep, value = line.partition('=')
            if not sep or key.strip() not in CONFIG_KEYS:
                continue
            value = value.strip()
            if value.startswith(('"', "'")):
                end = value.find(value[0], 1)
                if end < 0 or (value[end + 1:].strip() and not value[end + 1:].strip().startswith('#')):
                    raise ValueError('invalid_env_file')
                value = value[1:end]
            else:
                value = value.split(' #', 1)[0].strip()
            values[key.strip()] = value
    if not env_file_only:
        values.update({k: os.environ[k] for k in CONFIG_KEYS if k in os.environ})
    return values


def clock_diagnostics(timestamp, date_values, elapsed):
    """Date is an intermediary's clock, not a trusted time synchronization source."""
    result = {'request_time_utc': datetime.fromtimestamp(int(timestamp), timezone.utc).isoformat()}
    if len(date_values) != 1:
        return result
    try:
        remote = parsedate_to_datetime(date_values[0])
        if remote.tzinfo is None:
            return result
        delta = remote.timestamp() - int(timestamp)
    except (TypeError, ValueError, OverflowError):
        return result
    result.update(response_date_minus_request_seconds=round(delta, 3),
                  clock_skew_suspected=abs(delta) > elapsed + 60)
    return result


def validate_payload(api, payload):
    """Check the current project's QCC response contracts, never dump results."""
    if not isinstance(payload, dict):
        return {'ok': False, 'code': 'invalid_response_schema'}
    status = str(payload.get('Status', ''))
    details = {'provider_status': status if status.isascii() and status.isdigit() and len(status) <= 8 else 'unknown'}
    result = payload.get('Result')
    if api == 'search' and status == '201' and result in (None, []):
        return {'ok': True, 'code': 'no_matches', 'records': 0, **details}
    if status != '200':
        return {'ok': False, 'code': 'provider_failure', **details}
    if api == 'search':
        valid = isinstance(result, list) and all(isinstance(r, dict) and
            isinstance(r.get('Name'), str) and bool(r['Name'].strip()) and
            isinstance(r.get('KeyNo'), str) and bool(r['KeyNo'].strip()) for r in result)
        count = len(result) if valid else 0
    elif api == 'basic':
        valid = isinstance(result, dict) and bool(result) and all(
            isinstance(result.get(k), str) and bool(result[k].strip()) for k in ('Name', 'KeyNo'))
        count = int(valid)
    else:
        valid = isinstance(result, dict) and type(result.get('VerifyResult')) is int and result['VerifyResult'] in (0, 1)
        rows = result.get('Data') if valid else None
        if valid and result['VerifyResult'] == 0 and rows is None:
            rows = []
        valid = valid and isinstance(rows, list) and all(isinstance(r, dict) for r in rows)
        valid = valid and not (result['VerifyResult'] == 0 and rows)
        count = len(rows) if valid else 0
    if not valid:
        return {'ok': False, 'code': 'invalid_result', **details}
    return {'ok': True, 'code': 'usable' if count else 'no_records', 'records': count, **details}


def probe(api, values, company, timeout, mode='proxy'):
    report = {'ok': False, 'method': 'GET', 'transport': 'http.client',
              'request_connection': 'keep-alive', 'use_environment_proxy': False,
              'url_setting': URL_KEYS[api], 'mode': mode,
              'page_scope': 'first_page' if api != 'basic' else 'single_result'}
    if mode not in {'direct', 'proxy'}:
        return {**report, 'code': 'invalid_mode'}
    key, secret = values.get('QCC_KEY', ''), values.get('QCC_SECRET_KEY', '')
    if not key or not secret:
        return {**report, 'code': 'missing_qcc_credentials'}
    params = ({'searchKey': company, 'pageIndex': 1} if api == 'search' else
              {'keyword': company} if api == 'basic' else
              {'searchKey': company, 'pageSize': 20, 'pageIndex': 1})
    try:
        url = values.get(URL_KEYS[api], '')
        if not url or any(c.isspace() for c in url) or '\\' in url:
            raise ValueError
        parts = urlsplit(url)
        if (parts.scheme not in {'http', 'https'} or not parts.hostname or
                parts.username is not None or parts.password is not None or parts.fragment):
            raise ValueError
        _ = parts.port
        query = parse_qsl(parts.query, keep_blank_values=True)
    except ValueError:
        return {**report, 'code': 'invalid_proxy_url', 'hint': 'URL须单行HTTP(S)地址，不含空格、反斜杠或用户密码'}
    names = [k for k, _ in query]
    if mode == 'direct' and query:
        return {**report, 'code': 'direct_url_must_not_have_query'}
    if mode == 'proxy' and (names.count('servertype') != 1 or names.count('proxyToken') != 1
            or not dict(query).get('servertype') or not dict(query).get('proxyToken')):
        return {**report, 'code': 'missing_proxy_params', 'hint': '须使用servertype和proxyToken，区分大小写，以&分隔、以=赋值'}
    if any(k in {'key', *params} for k in names):
        return {**report, 'code': 'duplicate_business_params'}
    # Preserve the gateway query bytes and routing value; do not guess/correct its spelling.
    target = (parts.path or '/') + '?' + (parts.query + '&' if parts.query else '') + urlencode({'key': key, **params})
    timestamp = str(int(time.time()))
    token = hashlib.md5((key + timestamp + secret).encode()).hexdigest().upper()
    secrets = [key, secret, token, company, *[v for k, v in query if k != 'servertype']]
    report.update(endpoint=redact(f'{parts.scheme}://{parts.netloc}{parts.path}', secrets),
                  servertype=redact(dict(query).get('servertype', ''), secrets),
                  query_keys=sorted(set(names) | {'key', *params}), request_timespan=timestamp)

    def failure(code, body=b'', **details):
        report.update(code=code, **details)
        report.update(error_body(body, secrets))
        return report

    headers = {'Token': token, 'Timespan': timestamp, 'Accept': '*/*', 'Connection': 'keep-alive'}
    connection = None
    started = time.perf_counter()
    try:
        connection = (HTTPSConnection if parts.scheme == 'https' else HTTPConnection)(parts.hostname, parts.port, timeout=timeout)
        connection.request('GET', target, headers=headers)
        with connection.getresponse() as response:
            report.update(http_status=response.status, response_complete=False)
            report.update(clock_diagnostics(timestamp, response.headers.get_all('Date', []),
                                            time.perf_counter() - started))
            report['response_headers'] = {
                name: [redact(value, secrets)[:500] for value in response.headers.get_all(name, [])][:4]
                for name in ('Content-Type', 'Server', 'Date', 'Content-Length', 'Transfer-Encoding',
                             'Connection', 'X-Request-ID', 'X-Correlation-ID', 'X-DS-Trace-ID', 'EO-LOG-UUID')
                if response.headers.get_all(name)
            }
            try:
                body = response.read(4_000_001)
            except IncompleteRead as exc:
                report['received_bytes'] = len(exc.partial)
                return failure('response_incomplete', exc.partial, error_type='IncompleteRead')
            report['received_bytes'] = len(body)
            length = response.headers.get('Content-Length', '')
            if not response.headers.get('Transfer-Encoding') and length.isascii() and length.isdigit() and len(body) < int(length):
                return failure('response_incomplete', body)
            if len(body) > 4_000_000:
                return failure('response_too_large')
            report['response_complete'] = True
            if not 200 <= response.status < 300:
                return failure('http_error', body)
        try:
            payload = json.loads(body)
        except (ValueError, UnicodeError):
            return failure('invalid_response_json', body)
        report.update(validate_payload(api, payload))
        if not report['ok']:
            report.update(error_body(body, secrets))
            if report.get('provider_status') == '115':
                report['hint'] = ('本次Token已用运行机器当前Unix秒数实时生成。'
                    '请核对机器UTC时间、QCC_KEY与QCC_SECRET_KEY配对及环境变量覆盖，'
                    '并检查代理透传Token/Timespan/key。Date偏差仅供诊断，'
                    '可能来自代理或缓存；60秒是诊断阈值，不是企查查签名有效期。')
    except TimeoutError:
        report['code'] = 'timeout'
    except OSError as exc:
        report.update(code='connection_failed', error_type=type(exc).__name__)
    except Exception as exc:
        report.update(code='request_failed', error_type=type(exc).__name__)
    finally:
        if connection is not None:
            connection.close()
        report['elapsed_seconds'] = round(time.perf_counter() - started, 3)
    return report


def main():
    parser = argparse.ArgumentParser(description='企查查三接口代理GET诊断，不输出密钥或公司正文')
    parser.add_argument('--company', required=True, help='要测试的准确企业全称')
    parser.add_argument('--api', choices=['all', *URL_KEYS], default='all')
    parser.add_argument('--mode', choices=['proxy', 'direct'], default='proxy',
                        help='默认proxy要求代理参数；direct使用无query的直连URL，传输和签名不变')
    parser.add_argument('--timeout', type=float, help='每次请求超时秒数，默认QCC_TIMEOUT_SECONDS或20')
    parser.add_argument('--env-file-only', action='store_true', help='仅读取项目.env，忽略同名环境变量')
    args = parser.parse_args()
    if not args.company.strip():
        parser.error('--company不能为空')
    path = Path(__file__).resolve().parents[1] / '.env'
    try:
        settings = load_settings(path, env_file_only=args.env_file_only)
        timeout = args.timeout if args.timeout is not None else float(settings.get('QCC_TIMEOUT_SECONDS', '20'))
        if not 0 < timeout <= 120:
            raise ValueError
    except (OSError, ValueError):
        print(json.dumps({'ok': False, 'code': 'invalid_env_or_timeout'}))
        return 1
    checks = {api: probe(api, settings, args.company.strip(), timeout, mode=args.mode)
              for api in (URL_KEYS if args.api == 'all' else [args.api])}
    ok = all(check['ok'] for check in checks.values())
    print(json.dumps({'ok': ok, 'env_file_exists': path.is_file(),
                      'env_file_only': args.env_file_only,
                      'environment_override_keys': [] if args.env_file_only else sorted(CONFIG_KEYS & os.environ.keys()),
                      'checks': checks}, ensure_ascii=False, indent=2))
    return 0 if ok else 1


if __name__ == '__main__':
    raise SystemExit(main())