高德地图商家电话导出工具 百度高德腾讯实体门店采集教程

pip install requests

import csv
import os
import time
from dataclasses import dataclass, field
from typing import List, Optional
from urllib.parse import quote

import requests

OUTPUT_CSV = "poi_contacts.csv"
REQUEST_INTERVAL = 0.5  # 秒，自行根据配额调整


@dataclass
class PoiContact:
    platform: str
    name: str
    address: str
    phone: str
    category: str = ""
    poi_id: str = ""


# ---------------- 高德 ----------------

def search_amap(key: str, keywords: str, city: str, page: int = 1) -> List[PoiContact]:
    url = "https://restapi.amap.com/v3/place/text"
    params = {
        "key": key,
        "keywords": keywords,
        "city": city,
        "citylimit": "true",
        "offset": 20,
        "page": page,
        "extensions": "all",  # 不加这个拿不到 tel
        "output": "json",
    }
    results = []
    try:
        resp = requests.get(url, params=params, timeout=10)
        data = resp.json()
    except Exception as e:
        print(f"[高德] 请求异常：{e}")
        return results

    if data.get("status") != "1":
        print(f"[高德] 失败：{data.get('info')}")
        return results

    for poi in data.get("pois", []):
        tel = poi.get("tel", "") or ""
        # 高德 tel 可能是空的，也可能用分号分隔多个号码
        results.append(PoiContact(
            platform="高德",
            name=poi.get("name", ""),
            address=poi.get("address", "") if isinstance(poi.get("address"), str) else "",
            phone=tel,
            category=poi.get("type", ""),
            poi_id=poi.get("id", ""),
        ))
    return results


# ---------------- 百度 ----------------

def search_baidu(key: str, keywords: str, region: str, page: int = 0) -> List[PoiContact]:
    url = "https://api.map.baidu.com/place/v2/search"
    params = {
        "ak": key,
        "query": keywords,
        "region": region,
        "output": "json",
        "page_size": 20,
        "page_num": page,
        "scope": 2,  # 必须为 2，才会返回 detail_info，其中包含 telephone
    }
    results = []
    try:
        resp = requests.get(url, params=params, timeout=10)
        data = resp.json()
    except Exception as e:
        print(f"[百度] 请求异常：{e}")
        return results

    if data.get("status") != 0:
        print(f"[百度] 失败：{data.get('message')}")
        return results

    for item in data.get("results", []):
        detail = item.get("detail_info") or {}
        results.append(PoiContact(
            platform="百度",
            name=item.get("name", ""),
            address=item.get("address", ""),
            phone=detail.get("telephone", "") or "",
            category=detail.get("tag", "") or "",
            poi_id=item.get("uid", ""),
        ))
    return results


# ---------------- 腾讯 ----------------

def search_tencent(key: str, keywords: str, region: str, page: int = 1) -> List[PoiContact]:
    url = "https://apis.map.qq.com/ws/place/v1/search"
    # filter=tel<>null 用于筛选有电话的地点，参数值需要 URL 编码
    params = {
        "key": key,
        "keyword": keywords,
        "boundary": f"region({region},0)",
        "page_size": 20,
        "page_index": page,
        "filter": quote("tel<>null"),
        "output": "json",
    }
    results = []
    try:
        resp = requests.get(url, params=params, timeout=10)
        data = resp.json()
    except Exception as e:
        print(f"[腾讯] 请求异常：{e}")
        return results

    if data.get("status") != 0:
        print(f"[腾讯] 失败：{data.get('message')}")
        return results

    for item in data.get("data", []):
        results.append(PoiContact(
            platform="腾讯",
            name=item.get("title", ""),
            address=item.get("address", ""),
            phone=item.get("tel", "") or "",
            category=item.get("category", ""),
            poi_id=item.get("id", ""),
        ))
    return results


# ---------------- 汇总与导出 ----------------

def collect_all(
    amap_key: Optional[str],
    baidu_key: Optional[str],
    tencent_key: Optional[str],
    keywords: str,
    city: str,
    max_pages: int = 3,
) -> List[PoiContact]:
    all_contacts: List[PoiContact] = []

    if amap_key:
        for p in range(1, max_pages + 1):
            batch = search_amap(amap_key, keywords, city, page=p)
            all_contacts.extend(batch)
            time.sleep(REQUEST_INTERVAL)
            if len(batch) < 20:
                break

    if baidu_key:
        for p in range(max_pages):
            batch = search_baidu(baidu_key, keywords, city, page=p)
            all_contacts.extend(batch)
            time.sleep(REQUEST_INTERVAL)
            if len(batch) < 20:
                break

    if tencent_key:
        for p in range(1, max_pages + 1):
            batch = search_tencent(tencent_key, keywords, city, page=p)
            all_contacts.extend(batch)
            time.sleep(REQUEST_INTERVAL)
            if len(batch) < 20:
                break

    return all_contacts


def dedupe_and_export(contacts: List[PoiContact], output: str = OUTPUT_CSV):
    # 只保留有电话的记录，并按“电话 + 名称”去重（同一门店可能在多个平台出现）
    with_phone = [c for c in contacts if c.phone.strip()]
    seen = set()
    unique = []
    for c in with_phone:
        key = (c.phone.strip(), c.name.strip())
        if key not in seen:
            seen.add(key)
            unique.append(c)

    os.makedirs(os.path.dirname(output) or ".", exist_ok=True)
    with open(output, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["平台", "名称", "地址", "电话", "分类", "POI ID"])
        for c in unique:
            writer.writerow([c.platform, c.name, c.address, c.phone, c.category, c.poi_id])

    print(f"[完成] 有电话记录 {len(unique)} 条 → {output}")


if __name__ == "__main__":
    # 在环境变量中设置 Key，不要硬编码到代码里
    amap_key = os.getenv("AMAP_KEY")
    baidu_key = os.getenv("BAIDU_KEY")
    tencent_key = os.getenv("TENCENT_KEY")

    contacts = collect_all(
        amap_key=amap_key,
        baidu_key=baidu_key,
        tencent_key=tencent_key,
        keywords="咖啡馆",
        city="北京",
        max_pages=2,  # 先小批量验证
    )
    dedupe_and_export(contacts)