在数字经济与智慧旅游深度融合的背景下,网络平台已成为公众获取旅游信息、规划行程和预订服务的核心渠道。作为国内领先的在线旅行服务商,携程不仅提供机票、酒店等基础服务,还沉淀了海量结构化的景区数据,涵盖景点位置、用户评分、门票价格、标签分类、热度指数等多维度信息。这些实时、细粒度的数据为城市旅游空间研究提供了宝贵的一手资料,尤其适用于解析景区的空间分布格局、游客偏好特征以及区域文旅发展水平。本文以杭州市为典型案例,基于从携程公开接口获取的景区数据,结合地理信息系统(GIS)进行可视化与空间分析,系统揭示其旅游景观的组织逻辑与功能结构。

数据获取是本研究的关键前提。通过分析携程网页端的网络请求,我们定位到其景区列表接口,发现请求参数中需传入城市唯一编码(districtId)。而该编码并未公开提供,而是隐含在携程攻略频道的网页链接中,据此,我们设计了两阶段数据采集流程:第一阶段,爬取全国城市列表,提取“城市名称—城市ID”对应关系并保存为 cities.csv;第二阶段,遍历该文件中的每个城市ID,调用API批量获取其前300个景点的详细信息(每城最多约3000条记录),并引入重试机制与随机延时策略,以提升爬虫的稳定性与鲁棒性,最终构建覆盖全国的地级及以上城市的景区数据库。

在获得原始数据后,我们对杭州市的景点进行专项分析。考虑到平台返回的地理坐标存在地图偏移问题,需先进行坐标纠偏处理,以确保在GIS平台中的空间定位准确。随后,结合景点的等级、评分、是否免费、所属行政区及功能标签等属性,对杭州市景区进行空间可视化与结构解读。分析结果清晰呈现出“西湖核心区高度集聚、沿江河廊道延伸、外围组团多点支撑”的三级空间格局,不仅反映了杭州“因水而兴、文景交融”的城市特质,也为优化旅游资源配置、引导客流疏解和推动全域旅游发展提供了实证依据与决策参考。

携程景区查询地址:中国旅游景点攻略_中国打卡/必去景点大全/排名/推荐【携程攻略】

首先,我们找到网点数据的存储位置,然后看3个关键部分标头、负载、 预览;

标头:通常包括URL的连接,也就是目标资源的位置;

负载:对于POST请求:负载通常包含了传递的参数,因为所有参数都通过URL传递,这里我们可以看到城市编码、每页景区数量还有一些其他参数,没有进行加密;

预览:指的是对响应内容的快速查看或摘要显示,可以帮助用户快速了解返回的数据结构或内容片段,我们可以看到数据在card里,这里返回是有关西湖风景名胜区的相关内容;

接下来就是数据获取部分,先讲一下方法思路,一共三个步骤;

方法思路

  1. 找到对应城市的编码数据存储位置并保存到本地csv;
  2. 我们通过遍历城市编码来查询所有城市的前300个景区数据;
  3. 把景区地理坐标进行可视化,并完成坐标转换;

经过测试发现,我们只要选择到获取到城市名称对应的城市编码,就可以获取当前城市所有的景区数据,但是携程只能让我们获取前300页,且城市名称和编码是存储在 <a href="http://you.ctrip.com/place/xxx数字.html">某某旅游攻略</a> 这种结构中,那么我们先通过脚本抓取所有城市的ID编码;

第一步:利用requests库发送HTTP请求所有城市的ID编码数据,并根据标签进行保存,另存为csv;

完整代码#运行环境 Python 3.11

import requests
from lxml import etree
import re
import csv

# 1. 请求网页
url = "https://you.ctrip.com/countrysightlist/china110000/p1.html"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36"
}

print("正在请求网页...")
response = requests.get(url, headers=headers)
response.encoding = 'utf-8'

# 2. 解析 HTML
tree = etree.HTML(response.text)

# 3. 提取所有符合条件的链接
links = tree.xpath('//a[contains(@href, "/place/") and contains(@href, ".html") and contains(text(), "旅游攻略")]')

# 使用字典自动去重:key=城市名,value=城市ID
city_dict = {}

for link in links:
    text = link.text.strip()
    href = link.get('href', '')

    city_name = text.replace("旅游攻略", "").strip()
    match = re.search(r'(\d+)\.html$', href)

    if city_name and match:
        city_id = match.group(1)
        # 如果城市名尚未存在,则添加(避免重复)
        if city_name not in city_dict:
            city_dict[city_name] = city_id
            print(f"{city_name} → {city_id}")
        else:
            print(f"跳过重复城市: {city_name}")

# 4. 保存去重后的结果
if city_dict:
    with open("cities.csv", "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["城市名称", "城市ID"])
        for name, cid in city_dict.items():
            writer.writerow([name, cid])
    print(f"\n共提取 {len(city_dict)} 个不重复城市,已保存到 cities.csv")
else:
    print("未找到任何符合条件的城市链接。")

数据会以csv表格的形式,保存在运行脚本的目录下,数据标签包括:城市名称、城市ID;

第二步:利用requests库发送HTTP请求遍历所有获取到的城市ID编码,并获取所有城市前300个景区数据,并根据标签进行保存,另存为csv;

完整代码#运行环境 Python 3.11(单城市版)

import requests
import json
import csv
import time
import random

# === 配置 ===
url = "https://m.ctrip.com/restapi/soa2/18109/json/getAttractionList"

# 注意:这些 trace 参数可能有时效性,若失效需重新抓包
params_template = {
    "_fxpcqlniredt": "09031097417836601222",
    "x-traceID": "09031097417836601222-1772693012613-6871235"
}

headers = {
    "Content-Type": "application/json",
    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 Ctrip/9.99.9"
}

fields = [
    'poiId', 'poiName', 'districtName', 'zoneName', 'sightLevelStr',
    'commentScore', 'commentCount', 'price', 'marketPrice', 'isFree',
    'heatScore', 'latitude', 'longitude', 'coverImageUrl',
    'tagNameList', 'bookingTags', 'shortFeatures', 'sightCategoryInfo'
]

def fetch_page(index, max_retries=3):
    """获取单页数据,带重试机制,返回 attractionList、空列表或 None"""
    payload = {
        "head": {
            "cid": "09031097417836601222",
            "ctok": "",
            "cver": "1.0",
            "lang": "01",
            "sid": "8888",
            "syscode": "999",
            "auth": "",
            "xsid": "",
            "extension": []
        },
        "scene": "online",
        "districtId": 110000,
        "index": index,
        "sortType": 1,
        "count": 10,
        "filter": {"filterItems": ["4;14"]},
        "returnModuleType": "product"
    }

    for attempt in range(max_retries + 1):
        try:
            resp = requests.post(
                url,
                params=params_template,
                headers=headers,
                json=payload,
                timeout=12  # 稍微放宽超时
            )
            if resp.status_code == 200:
                data = resp.json()
                attraction_list = data.get("attractionList", [])
                return attraction_list  # 成功,直接返回(可能是空列表)
            else:
                print(f"第 {index} 页请求失败,状态码: {resp.status_code}")
        except Exception as e:
            print(f"第 {index} 页请求异常 (尝试 {attempt + 1}/{max_retries + 1}): {e}")

        # 如果不是最后一次尝试,就等待后重试
        if attempt < max_retries:
            delay = (2 ** attempt) + random.uniform(0, 1)  # 指数退避 + 随机抖动
            print(f"  → {delay:.1f} 秒后重试...")
            time.sleep(delay)
        else:
            print(f"  → 第 {index} 页最终失败,跳过。")
            return None  # 所有重试都失败

    return None  # 理论上不会走到这里

def process_item(card):
    """将 card 转为 CSV 行字典"""
    coord = card.get("coordinate", {})
    tag_names = ';'.join(card.get("tagNameList", []))
    booking_tags = ';'.join([
        tag["name"] for tag in card.get("otherTagList", [])
        if tag.get("type") == "booking"
    ])
    features = ';'.join(card.get("shortFeatures", []))

    return {
        'poiId': card.get('poiId'),
        'poiName': card.get('poiName'),
        'districtName': card.get('districtName'),
        'zoneName': card.get('zoneName'),
        'sightLevelStr': card.get('sightLevelStr'),
        'commentScore': card.get('commentScore'),
        'commentCount': card.get('commentCount'),
        'price': card.get('price'),
        'marketPrice': card.get('marketPrice'),
        'isFree': card.get('isFree'),
        'heatScore': card.get('heatScore'),
        'latitude': coord.get('latitude'),
        'longitude': coord.get('longitude'),
        'coverImageUrl': card.get('coverImageUrl'),
        'tagNameList': tag_names,
        'bookingTags': booking_tags,
        'shortFeatures': features,
        'sightCategoryInfo': card.get('sightCategoryInfo')
    }

# === 主流程 ===
print("开始采集景点数据(带重试机制)...")
with open('ctrip_attractions.csv', 'w', encoding='utf-8-sig', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=fields)
    writer.writeheader()

    total_saved = 0
    for page in range(1, 301):  # 页码从 1 到 301
        print(f"\n正在请求第 {page} 页...")

        result = fetch_page(page)
        if result is None:
            # 所有重试失败,跳过本页
            print(f"跳过第 {page} 页(多次重试后仍失败)")
        elif len(result) == 0:
            # 无数据,提前结束
            print("无更多数据,提前结束采集。")
            break
        else:
            # 成功获取,写入 CSV
            for item in result:
                card = item.get("card", {})
                if card:
                    row = process_item(card)
                    writer.writerow(row)
                    total_saved += 1

        # 随机延时 1~3 秒,模拟人工操作
        delay = random.uniform(1.0, 3.0)
        print(f"等待 {delay:.1f} 秒...")
        time.sleep(delay)

print(f"\n采集完成!共保存 {total_saved} 条景点数据到 ctrip_attractions.csv")

这里有一个tips:"filter": {"filterItems": ["4;14"]}, 这里通过修改对应城市ID就可以换其他城市了,也就是把这里的“14”换成其他编码,另外这里的delay = random.uniform(1.0, 3.0)随延时也可以改,现在暂时是1到3秒,可以根据需求调整;

完整代码#运行环境 Python 3.11(全量城市版)

import requests
import csv
import time
import random
import os
from functools import wraps


# ====== 重试装饰器 ======
def retry(max_retries=3, delay=1, backoff=2):
    """
    装饰器:对函数进行重试
    :param max_retries: 最大重试次数
    :param delay: 初始延迟(秒)
    :param backoff: 每次延迟倍数(指数退避)
    """

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries:
                        print(f"最终失败: {e}")
                        raise
                    else:
                        print(f"第 {attempt + 1} 次尝试失败: {e},{current_delay}秒后重试...")
                        time.sleep(current_delay)
                        current_delay *= backoff
            return None

        return wrapper

    return decorator


# ====== 配置 ======
API_URL = "https://m.ctrip.com/restapi/soa2/18109/json/getAttractionList"

PARAMS_TEMPLATE = {
    "_fxpcqlniredt": "09031097417836601222",
    "x-traceID": "09031097417836601222-1772693012613-6871235"
}

HEADERS = {
    "Content-Type": "application/json",
    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 Ctrip/9.99.9"
}

FIELDS = [
    'cityName', 'poiId', 'poiName', 'districtName', 'zoneName', 'sightLevelStr',
    'commentScore', 'commentCount', 'price', 'marketPrice', 'isFree',
    'heatScore', 'latitude', 'longitude', 'coverImageUrl',
    'tagNameList', 'bookingTags', 'shortFeatures', 'sightCategoryInfo'
]


# ====== 带重试的单页请求函数 ======
@retry(max_retries=3, delay=1, backoff=2)
def fetch_single_page(city_name, city_id, page):
    """带重试的单页请求,返回 attractionList 或抛出异常"""
    payload = {
        "head": {
            "cid": "09031097417836601222",
            "ctok": "",
            "cver": "1.0",
            "lang": "01",
            "sid": "8888",
            "syscode": "999",
            "auth": "",
            "xsid": "",
            "extension": []
        },
        "scene": "online",
        "districtId": int(city_id),
        "index": page,
        "sortType": 1,
        "count": 10,
        "filter": {"filterItems": ["4;14"]},
        "returnModuleType": "product"
    }

    resp = requests.post(
        API_URL,
        params=PARAMS_TEMPLATE,
        headers=HEADERS,
        json=payload,
        timeout=12  # 稍微放宽超时
    )

    if resp.status_code != 200:
        raise requests.HTTPError(f"HTTP {resp.status_code}")

    data = resp.json()
    return data.get("attractionList", [])


# ====== 主采集函数 ======
def fetch_attractions_by_city(city_name, city_id, max_pages=301):
    """爬取单个城市的所有景点,自动翻页"""
    all_items = []
    for page in range(1, max_pages + 1):
        print(f"  → 第 {page} 页...", end="", flush=True)

        try:
            attraction_list = fetch_single_page(city_name, city_id, page)
        except Exception as e:
            print(f" 请求失败,跳过本页: {e}")
            continue  # 单页失败不影响后续页

        if not attraction_list:
            print(" 无数据,停止翻页。")
            break

        # 处理数据
        for item in attraction_list:
            card = item.get("card", {})
            if not card:
                continue
            coord = card.get("coordinate", {})
            tag_names = ';'.join(card.get("tagNameList", []))
            booking_tags = ';'.join([
                tag["name"] for tag in card.get("otherTagList", [])
                if tag.get("type") == "booking"
            ])
            features = ';'.join(card.get("shortFeatures", []))

            row = {
                'cityName': city_name,
                'poiId': card.get('poiId'),
                'poiName': card.get('poiName'),
                'districtName': card.get('districtName'),
                'zoneName': card.get('zoneName'),
                'sightLevelStr': card.get('sightLevelStr'),
                'commentScore': card.get('commentScore'),
                'commentCount': card.get('commentCount'),
                'price': card.get('price'),
                'marketPrice': card.get('marketPrice'),
                'isFree': card.get('isFree'),
                'heatScore': card.get('heatScore'),
                'latitude': coord.get('latitude'),
                'longitude': coord.get('longitude'),
                'coverImageUrl': card.get('coverImageUrl'),
                'tagNameList': tag_names,
                'bookingTags': booking_tags,
                'shortFeatures': features,
                'sightCategoryInfo': card.get('sightCategoryInfo')
            }
            all_items.append(row)

        print(f" 获取 {len(attraction_list)} 条")
        time.sleep(random.uniform(1.0, 2.5))

    return all_items


# ====== 主流程 ======
def main():
    if not os.path.exists("cities.csv"):
        print("错误:未找到 cities.csv 文件!")
        return

    cities = []
    with open("cities.csv", "r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            cities.append((row["城市名称"], row["城市ID"]))

    print(f"共加载 {len(cities)} 个城市,开始逐个采集景点...")

    output_file = "all_cities_attractions.csv"
    file_exists = os.path.exists(output_file)

    with open(output_file, "a" if file_exists else "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        if not file_exists:
            writer.writeheader()

        total_saved = 0
        for idx, (city_name, city_id) in enumerate(cities, 1):
            print(f"\n[{idx}/{len(cities)}] 正在采集城市: {city_name} (ID: {city_id})")

            items = fetch_attractions_by_city(city_name, city_id, max_pages=301)

            if items:
                writer.writerows(items)
                total_saved += len(items)
                print(f"本城共获取 {len(items)} 条景点")
            else:
                print("无景点数据")

            time.sleep(random.uniform(2.0, 4.0))

    print(f"\n全部完成!共采集 {total_saved} 条景点数据,已保存到 {output_file}")


if __name__ == "__main__":
    main()

获取数据标签如下:poiId(景点唯一ID)、poiName(景点名称)、districtName(所属行政区)、zoneName(所属城市)、 sightLevelStr(景区等级)、commentScore(用户评分)、commentCount(评论数量)、price(实际销售价格) marketPrice(原价)、isFree(是否免费)、heatScore(热度分)、lat&lon(地理坐标)、tagNameList(标签列表)、bookingTags(预订相关标签)、shortFeatures(简短特色描述)、sightCategoryInfo(景点分类信息),其他一些非关键标签,这里省略;

第三步:坐标系转换,由于携程景区数据使用的是百度坐标系(BD-09),为了在ArcGIS上准确展示而不发生偏移,我们需要将景区的坐标从BD-09转换为WGS-84坐标系。我们可以利用coord-convert库中的bd2wgs(lng, lat)函数,也可以用免费这个网站:批量转换工具:地图坐标系批量转换 - 免费在线工具;

对CSV文件中的门店坐标列进行转换,完成坐标转换后,再将数据导入ArcGIS进行可视化;

接下来,我们以杭州市为例进行看图说话:

杭州市景区分布以西湖为核心,形成高度集聚的旅游引力中心。 西湖风景名胜区及其周边5公里范围内集中了全市绝大多数高等级景点,包括“西湖十景”(如断桥残雪、雷峰夕照、三潭印月)、千年古刹灵隐寺、岳王庙、净慈寺,以及中国茶叶博物馆、浙江省博物馆、南宋御街、河坊街等文化与商业地标。这一区域不仅是世界文化遗产地,更通过环湖绿道、夜游演艺等主题活动实现全时段、全季节运营,成为国内外游客到访杭州的“必游之地”。然而,高密度客流也带来交通拥堵、环境承载压力等问题,促使管理部门实施预约限流、智慧导览和分流引导等精细化治理措施。

景区沿水系与历史廊道呈带状延伸,构建多条特色旅游轴线。 钱塘江生态文化带串联六和塔、白塔公园、城市阳台、奥体中心“大莲花”及萧山湘湖旅游度假区,融合历史遗迹、现代地标与滨水休闲功能;京杭大运河(杭州段)则活化利用拱宸桥、小河直街、桥西直街等历史街区,展现“因河而兴”的市井文化;西溪国家湿地公园作为全国首个国家湿地公园,以“曲水寻梅”“秋芦飞雪”等生态场景,提供与西湖互补的静谧自然体验。这些廊道不仅拓展了城市旅游空间,也强化了“水城共生”的杭州地域特色,推动旅游从点状观光向线性深度体验升级。

城市外围依托自然与文化遗产,形成多元化的次级旅游组团。 余杭区的良渚古城遗址实证中华五千年文明起源,成为考古研学与文化朝圣的重要目的地;临安区的天目山、大明山以原始森林、峡谷瀑布和避暑气候吸引生态与户外爱好者;富阳区依托《富春山居图》文化IP,打造黄公望隐居地、鹳山公园等人文景观;此外,超山梅花、径山禅寺、湘湖跨湖桥遗址等也各具特色。这些外围节点虽密度较低,但类型丰富、主题鲜明,有效支撑杭州从“西湖依赖”向“全域协同、多极联动”的全域旅游格局转型,为高质量文旅发展提供纵深空间。

文章仅用于分享个人学习成果与个人存档之用,分享知识,如有侵权,请联系作者进行删除。所有信息均基于作者的个人理解和经验,不代表任何官方立场或权威解读。

Logo

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容,并连接云服务、办公空间等稀缺资源,助你专注创造,无忧运营。

更多推荐