#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import re
import argparse
import requests
import random
import time
import certifi
import json
import traceback
import os
from termcolor import colored
from typing import List, Dict, Tuple, Optional, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib3.exceptions import InsecureRequestWarning

# --- Banner ---
BANNER = """


                                                                                           
                            by: 地图API密钥全自动检测工具
"""

# --- Configuration ---

# Emojis for output enhancement
EMOJI_DETECT = "🔍"
EMOJI_PLATFORM = "🏷️"
EMOJI_KEY_FORMAT = "🔑"
EMOJI_SERVICE = "▶️"
EMOJI_SUCCESS = "✅"
EMOJI_FAIL = "❌"
EMOJI_CLOCK = "⏱️"
EMOJI_INFO = "ℹ️"
EMOJI_ERROR = "❗"
EMOJI_WARNING = "⚠️"
EMOJI_NETWORK = "🌐"
EMOJI_QUOTA = "🚦"
EMOJI_PERM = "🔒"
EMOJI_WRITE = "💾"
EMOJI_TOOL = "🛠️"


# Disable SSL warnings (use cautiously)
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# Safe coordinate bounds
SAFE_BOUNDS = {
    'global': {'lon_min': -180, 'lon_max': 180, 'lat_min': -90, 'lat_max': 90},
    'china': {'lon_min': 73.66, 'lon_max': 135.05, 'lat_min': 3.86, 'lat_max': 53.55}
}

# Test address pool
TEST_ADDRESS_POOL = [
    "北京市朝阳区望京soho",
    "上海市浦东新区陆家嘴环路1288号",
    "广州市天河区珠江新城临江大道5号",
    "深圳市福田区深南大道6001号",
    "成都市锦江区红星路三段1号"
]

# --- Helper Classes ---

class GeoGenerator:
    """Generates random geographic data."""
    @staticmethod
    def random_coord(region='china'):
        bounds = SAFE_BOUNDS[region]
        return (
            round(random.uniform(bounds['lat_min'], bounds['lat_max']), 6),
            round(random.uniform(bounds['lon_min'], bounds['lon_max']), 6)
        )

    @staticmethod
    def random_ip():
        # Generate a public IP range more likely to be locatable
        while True:
            ip = f"{random.randint(1, 223)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"
            parts = [int(p) for p in ip.split('.')]
            if parts[0] == 10: continue
            if parts[0] == 127: continue
            if parts[0] == 172 and 16 <= parts[1] <= 31: continue
            if parts[0] == 192 and parts[1] == 168: continue
            if parts[0] >= 224: continue
            return ip


class KeyValidator:
    """Validates key format and detects platform."""
    @staticmethod
    def detect_platform(key: str) -> Tuple[str, str]:
        """Detects platform and provides format description."""
        if re.match(r'^[0-9a-fA-F]{32}$', key):
            return 'amap', '高德 (32位 Hex)'
        if re.match(r'^[A-Za-z0-9]{32}$', key):
            if re.match(r'^[0-9a-fA-F]{32}$', key):
                 return 'ambiguous', '可能是高德或百度 (纯Hex字符)'
            return 'baidu', '百度 (32位 Alnum)'
        return 'unknown', '未知格式'


class APITester:
    """Executes API test calls."""
    def __init__(self):
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36',
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:90.0) Gecko/20100101 Firefox/90.0'
        ]

    def _generate_service_configs(self, platform: str) -> Dict:
         """Generates service configs dynamically for fresh random data."""
         coord = GeoGenerator.random_coord()
         random_addr = random.choice(TEST_ADDRESS_POOL)
         random_ip = GeoGenerator.random_ip()

         if platform == 'amap':
            return {
                "静态地图": {
                    "url": "https://restapi.amap.com/v3/staticmap",
                    "params": {"location": f"{coord[1]},{coord[0]}", "zoom": 10, "size": "400*300", "markers": f"mid,,A:{coord[1]},{coord[0]}"},
                    "is_static": True
                },
                "地理编码": {
                    "url": "https://restapi.amap.com/v3/geocode/geo",
                    "params": {"address": random_addr}
                },
                "逆地理编码": {
                    "url": "https://restapi.amap.com/v3/geocode/regeo",
                    "params": {"location": f"{coord[1]},{coord[0]}"}
                },
                "路径规划": {
                    "url": "https://restapi.amap.com/v3/direction/driving",
                    "params": {"origin": "116.481028,39.989643", "destination": "116.434446,39.90816"}
                },
                "IP定位": {
                    "url": "https://restapi.amap.com/v3/ip",
                    "params": {"ip": random_ip}
                }
            }
         elif platform == 'baidu':
            return {
                "静态地图": {
                    "url": "https://api.map.baidu.com/staticimage/v2",
                    "params": {"center": f"{coord[1]},{coord[0]}", "zoom": 10, "width": 400, "height": 300, "markers": f"{coord[1]},{coord[0]}"},
                    "is_static": True
                },
                "地理编码": {
                    "url": "https://api.map.baidu.com/geocoding/v3/",
                    "params": {"address": random_addr, "output": "json"}
                },
                "逆地理编码": {
                    "url": "https://api.map.baidu.com/reverse_geocoding/v3/",
                    "params": {"location": f"{coord[0]},{coord[1]}", "output": "json"}
                },
                "路径规划": {
                    "url": "https://api.map.baidu.com/direction/v2/driving",
                    "params": {"origin": "40.01116,116.339303", "destination": "39.936404,116.452562"}
                },
                "IP定位": {
                    "url": "https://api.map.baidu.com/location/ip",
                    "params": {"ip": random_ip, "coor": "bd09ll"}
                }
            }
         return {}

    def _call_api(self, url: str, params: dict, platform: str, service_name: str, is_static: bool = False) -> Tuple[bool, dict]:
        """Executes a single API request with retries."""
        current_params = params.copy()
        retries = current_params.pop('retry', 2) + 1
        detail = {'params': params, 'latency': None, 'error': None, 'error_info': None, 'response': None}

        for attempt in range(retries):
            response = None
            try:
                headers = {'User-Agent': random.choice(self.user_agents)}
                start_time = time.time()
                response = requests.get(
                    url,
                    params=current_params,
                    headers=headers,
                    timeout=15,
                    verify=certifi.where(),
                    stream=is_static
                )
                latency = round((time.time() - start_time) * 1000, 2)
                detail['latency'] = latency

                content_type = response.headers.get('Content-Type', '').lower()

                if is_static and response.status_code == 200 and content_type.startswith('image/'):
                    detail['response'] = {'content_type': content_type, 'status_code': 200}
                    response.close()
                    return True, detail

                if response.status_code != 200:
                    error_msg = f"HTTP {response.status_code}"
                    try:
                        error_data = response.json()
                        error_msg += f". API Msg: {json.dumps(error_data, ensure_ascii=False)}"
                    except (json.JSONDecodeError, requests.exceptions.RequestException):
                        error_msg += f". Response: {response.text[:200]}..."
                    detail['error'] = error_msg
                    raise requests.HTTPError(error_msg, response=response)

                data = None
                try:
                    if 'application/json' in content_type or 'text/javascript' in content_type:
                         data = response.json()
                    elif platform == 'amap' and service_name == "IP定位" and 'text/html' in content_type and response.text.startswith('{') and response.text.endswith('}'):
                         data = json.loads(response.text)
                    else:
                         raise ValueError(f"非预期响应类型: {content_type}")

                    detail['response'] = data

                    success = False
                    error_info = ""
                    if platform == 'amap':
                        success = str(data.get('status')) == '1'
                        if not success: error_info = data.get('info', '未知高德错误') + f" (infocode: {data.get('infocode', '')})"
                    elif platform == 'baidu':
                        success = data.get('status') == 0
                        if not success: error_info = data.get('message', '未知百度错误') + f" (status: {data.get('status', '')})"
                    else: error_info = "未知平台错误"

                    detail['error_info'] = error_info if not success else None
                    return success, detail

                except json.JSONDecodeError as e:
                    detail['error'] = f"JSON解析失败 ({content_type}): {str(e)} | 响应: {response.text[:200]}..."
                    return False, detail
                except ValueError as e:
                    detail['error'] = f"{str(e)} | 响应: {response.text[:200]}..."
                    return False, detail

            except requests.exceptions.Timeout:
                detail['error'] = "请求超时"
                if attempt == retries - 1: return False, detail
                time.sleep(1 + 1.5 * attempt)
            except requests.exceptions.RequestException as e:
                detail['error'] = f"请求错误: {type(e).__name__}" + (f" (Status: {response.status_code})" if response else "")
                if attempt == retries - 1: return False, detail
                time.sleep(1 + 1.5 * attempt)
            except Exception as e:
                 detail['error'] = f"未知处理错误: {type(e).__name__}: {str(e)}"
                 print(f"{EMOJI_ERROR} {colored('内部脚本错误', 'red')}: {traceback.format_exc()[:500]}...")
                 return False, detail
            finally:
                if response:
                    response.close()

        detail['error'] = f"超过最大重试次数 ({retries}). 最后错误: {detail['error']}"
        return False, detail

    def test_service(self, platform: str, service_name: str, key: str, service_config: dict) -> Tuple[bool, dict]:
        """Tests a single service with its config."""
        params = service_config['params'].copy()
        params['key' if platform == 'amap' else 'ak'] = key
        is_static = service_config.get('is_static', False)
        return self._call_api(service_config['url'], params, platform, service_name, is_static)


class ReportGenerator:
    """Generates console output for test results."""
    @staticmethod
    def print_result(service: str, platform: str, success: bool, detail: dict):
        status_icon = EMOJI_SUCCESS if success else EMOJI_FAIL
        status_color = 'green' if success else 'red'
        service_status = colored("成功" if success else "失败", status_color)

        print(f"{EMOJI_SERVICE} {service.ljust(7)} {status_icon} {service_status}")

        indent = "  │"

        latency = detail.get('latency')
        latency_str = f"{latency:.2f} ms" if latency is not None else "N/A"
        print(f"{indent} {EMOJI_CLOCK} 耗时: {latency_str}")

        params_str = " | ".join([f"{k}={v:.4f}" if isinstance(v, float) else f"{k}={str(v)[:30]}{'...' if len(str(v))>30 else ''}"
                                 for k, v in detail.get('params', {}).items() if k not in ['key', 'ak']])
        if params_str:
             print(f"{indent} {EMOJI_INFO} 参数: {colored(params_str, 'cyan')}")

        if success:
            success_info_str = ReportGenerator._extract_success_info(service, platform, detail.get('response'))
            if success_info_str:
                print(f"{indent} {EMOJI_INFO} 结果: {colored(success_info_str, 'green')}")

        error_msg = detail.get('error')
        error_info = detail.get('error_info')

        if error_info:
            print(f"{indent} {EMOJI_WARNING} 原因: {colored(error_info, 'yellow')}")
            ReportGenerator._print_error_analysis(error_info)
        elif error_msg:
            print(f"{indent} {EMOJI_ERROR} 原因: {colored(error_msg, 'red')}")
            ReportGenerator._print_error_analysis(error_msg)
        elif not success:
            print(f"{indent} {EMOJI_ERROR} 原因: {colored('未知原因失败', 'red')}")
        print( "  └" + "─" * 30)


    @staticmethod
    def _extract_success_info(service: str, platform: str, response_data: Optional[Any]) -> str:
        """Extracts a brief summary from successful API response data."""
        if response_data is None: return ""
        try:
            if isinstance(response_data, dict) and response_data.get('content_type', '').startswith('image/'):
                return f"成功获取图片 ({response_data['content_type']})"
            elif isinstance(response_data, dict):
                if service == "IP定位":
                    city = "未知"
                    if platform == 'amap': city = response_data.get('city') if isinstance(response_data.get('city'), str) and response_data.get('city') else response_data.get('adcode', 'N/A')
                    elif platform == 'baidu': city = response_data.get('content', {}).get('address_detail', {}).get('city', 'N/A')
                    return f"定位城市: {city}" if city and city != 'N/A' else "定位成功 (无城市信息)"
                elif service == "地理编码":
                    loc = "未知"
                    if platform == 'amap': loc = response_data.get('geocodes', [{}])[0].get('location', 'N/A')
                    elif platform == 'baidu': loc_dict = response_data.get('result', {}).get('location', {}); loc = f"{loc_dict.get('lat', 'N/A')},{loc_dict.get('lng', 'N/A')}" if loc_dict else 'N/A'
                    return f"获取坐标: {loc}"
                elif service == "逆地理编码":
                    addr = "未知"
                    if platform == 'amap': addr = response_data.get('regeocode', {}).get('formatted_address', 'N/A')
                    elif platform == 'baidu': addr = response_data.get('result', {}).get('formatted_address', 'N/A')
                    return f"获取地址: {addr[:40]}{'...' if len(addr)>40 else ''}"
                elif service == "路径规划":
                    dist, dur = "N/A", "N/A"
                    try:
                        if platform == 'amap': path = response_data.get('route', {}).get('paths', [{}])[0]; dist, dur = path.get('distance'), path.get('duration')
                        elif platform == 'baidu': route = response_data.get('result', {}).get('routes', [{}])[0]; dist, dur = route.get('distance'), route.get('duration')
                        return f"距离: {dist}米, 时间: {dur}秒"
                    except (IndexError, KeyError, TypeError): return "路径规划成功"
                else: return "调用成功"
            return ""
        except Exception as e:
            return colored(f"结果解析出错: {e}", 'magenta')

    @staticmethod
    def _print_error_analysis(error_text: str):
        """Prints specific warning icons based on error text keywords."""
        indent = "  │  "
        error_lower = error_text.lower()
        analysis_printed = False
        if 'quota' in error_lower or '配额' in error_lower or '并发' in error_lower or 'qps' in error_lower or 'limit' in error_lower:
            print(f"{indent}{EMOJI_QUOTA} {colored('[疑似配额/并发限制]', 'yellow')}")
            analysis_printed = True
        if 'invalid user key' in error_lower or 'key不正确' in error_lower or 'ak不存在' in error_lower or '权限校验失败' in error_lower or 'permission denied' in error_lower or 'key status' in error_lower or '无效' in error_lower :
            print(f"{indent}{EMOJI_PERM} {colored('[密钥无效或服务未开通/权限不足]', 'red')}")
            analysis_printed = True
        if 'domain' in error_lower or 'ip白名单' in error_lower or 'referer' in error_lower or 'sn校验失败' in error_lower:
             print(f"{indent}{EMOJI_PERM} {colored('[IP/域名/Referer/SN白名单校验失败]', 'magenta')}")
             analysis_printed = True
        if 'timeout' in error_lower or '超时' in error_lower:
            print(f"{indent}{EMOJI_NETWORK} {colored('[请求超时]', 'blue')}")
            analysis_printed = True
        if 'json解析失败' in error_lower or '非预期响应类型' in error_lower:
            print(f"{indent}{EMOJI_NETWORK} {colored('[响应格式错误或非预期]', 'magenta')}")
            analysis_printed = True
        if '请求错误' in error_text or 'http' in error_lower or 'connection' in error_lower or 'ssl' in error_lower:
            print(f"{indent}{EMOJI_NETWORK} {colored('[网络/连接/HTTP错误]', 'blue')}")
            analysis_printed = True


# --- Main Execution ---

def parse_args():
    """Parses command line arguments."""
    parser = argparse.ArgumentParser(
        description=f"{EMOJI_TOOL} 地图API密钥全自动检测工具 (v3.6)", # Version bump for banner
        formatter_class=argparse.RawTextHelpFormatter
        )
    parser.add_argument('-k', '--keys', nargs='+', help="直接在命令行指定一个或多个密钥")
    parser.add_argument('-f', '--file', help="包含密钥的文件路径 (每行一个密钥)")
    parser.add_argument('-t', '--threads', type=int, default=5, help="并发测试线程数 (默认: 5)")
    parser.add_argument('-o', '--output', help="将成功的检测结果输出到指定的JSON文件")
    parser.add_argument('--skip-static', action='store_true', help="跳过静态地图服务的检测")

    temp_args, _ = parser.parse_known_args()
    if not temp_args.keys and not temp_args.file:
         # Print banner before showing help on error
         print(colored(BANNER, 'cyan'))
         parser.print_help()
         print(colored("\n错误: 必须通过 -k 或 -f 提供至少一个密钥。", "red"))
         exit(1)

    return parser.parse_args()

def main(keys_to_test: List[str], num_threads: int, output_file: Optional[str], skip_static: bool):
    """Main detection workflow."""
    tester = APITester()
    futures_map = {}
    results_for_output = []

    print(f"{EMOJI_DETECT} 开始检测 {len(keys_to_test)} 个密钥，使用 {num_threads} 个线程...")
    if skip_static: print(f"{EMOJI_INFO} 已设置跳过静态地图检测。")

    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        for key in keys_to_test:
            key = key.strip()
            if not key: continue

            platform, key_format_desc = KeyValidator.detect_platform(key)
            masked_key = f"{key[:6]}...{key[-4:]}" if len(key) > 10 else key

            if platform == 'unknown':
                print(colored(f"\n{EMOJI_FAIL} 无效密钥格式: {key}", "red"))
                continue
            if platform == 'ambiguous':
                 print(colored(f"\n{EMOJI_WARNING} 检测到模糊密钥: {masked_key}", "yellow"))
                 print(colored(f"  格式 ({key_format_desc}) 可能属于高德或百度，将尝试两者。", "yellow"))
                 platforms_to_try = ['amap', 'baidu']
            else:
                 platforms_to_try = [platform]

            for p in platforms_to_try:
                print(colored(f"\n{EMOJI_DETECT} 正在检测 {p.upper()} 服务 - 密钥: {masked_key}", "blue", attrs=['bold']))
                print(f"  {EMOJI_PLATFORM} 平台识别: {p.upper()}")
                print(f"  {EMOJI_KEY_FORMAT} Key格式: {key_format_desc}")

                platform_services = tester._generate_service_configs(p)

                for service_name, service_config in platform_services.items():
                    if skip_static and service_config.get('is_static', False):
                        continue

                    future = executor.submit(tester.test_service, p, service_name, key, service_config)
                    futures_map[future] = {'key': key, 'platform': p, 'service_name': service_name}

        print(colored("\n--- 检测结果 ---", attrs=['bold']))
        processed_count = 0
        total_tasks = len(futures_map)

        for future in as_completed(futures_map):
            processed_count += 1
            context = futures_map[future]
            key = context['key']
            platform = context['platform']
            service_name = context['service_name']
            masked_key = f"{key[:6]}...{key[-4:]}" if len(key) > 10 else key

            try:
                success, detail = future.result()
                ReportGenerator.print_result(service_name, platform, success, detail)

                if success and output_file:
                    result_item = {
                        'key': key,
                        'platform': platform,
                        'service_name': service_name,
                        'request_params': detail.get('params'),
                        'response_summary': ReportGenerator._extract_success_info(service_name, platform, detail.get('response')),
                        'latency_ms': detail.get('latency'),
                    }
                    results_for_output.append(result_item)

            except Exception as exc:
                 print(colored(f"{EMOJI_FAIL} {service_name.ljust(7)}", 'red'))
                 print(f"  └─ {colored(f'执行 {service_name} (平台: {platform}, 密钥: {masked_key}) 时发生内部错误: {exc}', 'red')}")

    if output_file:
        print(colored(f"\n{EMOJI_WRITE} 正在将 {len(results_for_output)} 条成功结果写入到: {output_file}", "blue"))
        try:
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(results_for_output, f, ensure_ascii=False, indent=4)
            print(colored(f"{EMOJI_SUCCESS} 成功写入 JSON 文件。", "green"))
        except IOError as e:
            print(colored(f"{EMOJI_ERROR} 写入文件失败: {e}", "red"))
        except TypeError as e:
             print(colored(f"{EMOJI_ERROR} 序列化结果为JSON时失败 (可能包含无法序列化的数据): {e}", "red"))


if __name__ == "__main__":
    # --- Print Banner First ---
    print(colored(BANNER, 'cyan')) # Choose a color for the banner

    args = parse_args() # Parse arguments after printing banner
    keys = []

    if args.keys:
        keys.extend(args.keys)
    if args.file:
        try:
            with open(args.file, 'r', encoding='utf-8') as f:
                keys.extend([line.strip() for line in f if line.strip()])
        except FileNotFoundError:
             print(colored(f"{EMOJI_ERROR} 错误：密钥文件 '{args.file}' 未找到。", "red"))
             exit(1)
        except Exception as e:
             print(colored(f"{EMOJI_ERROR} 错误：读取密钥文件 '{args.file}' 时出错: {e}", "red"))
             exit(1)

    unique_keys = sorted(list(set(keys)))

    # Input validation is handled within parse_args now

    main(unique_keys, args.threads, args.output, args.skip_static)
    print(colored(f"\n{EMOJI_TOOL} 检测完成。", attrs=['bold']))