首页编程日常

POI (Point of Interest，兴趣点)，通常指的是在地理信息系统、地图应用程序或导航系统中的特定地点。兴趣点可以是自然景观、历史遗迹、文化地标、餐馆、商店、娱乐场所等。

前段时间，实验室的学长给我派活，内容大概是爬取一些 POI 数据，包括经纬度和文本描述信息。在此之前，我从未接触过爬虫相关的知识，所以这算是我第一次经历。虽然任务内容不多，但是我还是花费了些功夫，翻阅了许多资料，最终勉强写了个半成品。这里简单记录下这次的学习过程，也方便日后翻阅复习。

# 数据来源

# POI 文本数据

获取 POI 的地理坐标并非难点，众多来源如高德地图和百度地图提供的 API 服务均可实现。然而，获取 POI 的文本描述信息则显得相对复杂。例如，一所大学的文本描述在网络上容易找到，但对于餐饮店或商业楼宇等其他类型的地点，直接从网络中获取相关信息则较为困难。

鉴于此，我开始探索主流的应用程序作为潜在的数据来源。首先，我考察了大众点评，考虑到其在人们日常生活中的普遍使用，它似乎是一个理想的选择。然而，我发现该平台主要提供用户评价，而缺乏对大多数 POI 的详细描述。随后，我对美团和高德地图等其他平台也进行了调研，但结果同样不尽如人意。犯了难的我只好求助学长，他提醒我考虑微博的数据资源，这是我之前由于个人习惯而忽略的一个平台。于是我连忙注册账号并登录微博，进行了一番搜索，发现了这样的页面：

poi文本网页

网页上提供的简介似乎正是我寻求的 POI 文本信息。然而，在进一步考察其他地点后，我注意到，对于一般商店和楼宇，所提供的简介通常仅限于类型和地点，不够丰富。在与学长进一步确认后，我们认定微博上的文本数据虽不完美，但也具有一定的适用性。至此，POI 文本数据的来源算是最终确定下来。

# POI 地理坐标

最难的 POI 文本数据来源已经解决，我接下来转向了 POI 的地理坐标问题。在页面上，我注意到了一个地图展示功能，这通常意味着背后有服务 API 的支持。依据我的开发经验，为了实现地图坐标对特定 POI 的精确定位，经纬度信息是必不可少的。于是我点击进入了 "地图模式" 页面，并运用浏览器的开发者工具进行分析，确认页面确实包含了所需的 POI 经纬度信息。

地图页

此外，我还注意到页面左侧的搜索框。测试验证了该关键字搜索接口也能提供 POI 地理位置数据，且数据量更丰富，使用更简便。因此，我决定采用此接口来爬取经纬度数据。

关键字搜索接口


# 爬取流程

# 概述

确定了数据来源后，爬取流程也得以确定。分析包含文本数据的页面 URL，我发现 URL 中某些部分是相同的，只有最后的字符串与每个 POI 唯一关联。这个唯一的字段一般而言就是 POI 的 ID，而关键字搜索接口返回的数据中恰好包括了这个字段。因此，每个 POI 对应的页面 URL 可以表示成：

https://weibo.com/p/100101{poiid}

因此可以总结如下的爬取流程：

  • 依据需要爬取的 POI 名称，利用关键字搜索接口返回数据。
  • 保存接口返回的 POI 数据，其中包括了经纬度信息与 POIID。
  • 利用 POIID 拼接对应的页面 URL，构造请求获取页面内容。
  • 从页面内容中解析出需要的简介文本。

# 模拟登录

多数网站会对用户个人信息作检测。而上面我用到的几个页面都需要登录才放行。编写爬虫时，需要先准备网站 cookies 来模拟登录状态。最简单的方法是登录微博，查看并保存 cookies 信息，然后在构造请求时添加。

这里我还使用了 selenium 库实现另一种获取 cookies 的方法，相比前一种方法略显复杂，而且还用到了 chrome 拓展（chrome 拓展安装教程传送门）。由于这种方法比较麻烦，这里就简单地贴出代码，不再详细说明。

from selenium import webdriver  # 导入 Selenium 的 webdriver 模块，用于浏览器自动化控制
import time
import json
if __name__ == "__main__":
    browser = webdriver.Chrome()  # 初始化一个 Chrome 浏览器实例
    log_url = 'https://passport.weibo.com/sso/signin?entry=miniblog&source=miniblog&disp=popup&url=https%3A%2F%2Fweibo.com%2Fnewlogin%3Ftabtype%3Dweibo%26gid%3D102803%26openLoginLayer%3D0%26url%3D'  # 定义微博登录页面的 URL
    browser.get(log_url)  # 使用 webdriver 打开微博登录页面
    time.sleep(45)  # 等待 45 秒，以便手动登录
    cookies_list = browser.get_cookies()  # 获取当前浏览器会话的所有 cookies
    cookies_dict = {cookie['name']: cookie['value'] for cookie in cookies_list}  # 将 cookies 列表转换成字典格式
    try:
        with open('config\weibo_cookies.txt', 'w') as f:
            json.dump(cookies_dict, f)  # 将 cookies 字典以 JSON 格式写入文件
        print('cookies保存成功！')
    except Exception as e:
        print("save cookies failed: ", e)
    
    browser.quit() # 关闭浏览器
将 cookie 信息用 JSON 格式保存下来后，就可以在  request  请求时使用了。


with open('config/weibo_cookies.txt', 'r', encoding='utf-8') as f:
    cookies_dict = json.loads(f.read())  # 用户 cookies
response = requests.get(search_url, cookies=cookies_dict) # example
# 遍历 POI 地点


如何高效地遍历并获取所有 POI 名称几乎是整个项目中最麻烦的部分，甚至于到现在都没能取得比较满意的效果。之前我已确定了爬取 POI 所有数据的来源，但是如何遍历获取 POI 名称反而成了最棘手的问题。


刚开始，我尝试通过地名细分进行关键字搜索，但很快发现微博的接口似乎是以 POI 名称作为匹配依据的，而大多数 POI 名称并不包含地名信息。后来我又分析，发现 POIID 的前半部分似乎是相同的，这让我猜测 POIID 可能根据地区进行区分。基于这一假设，我尝试通过遍历 POIID 来获取北京市的数据。然而，经过多次测试，我发现 POIID 并不连续，且存在大量空 ID 的情况，不同地区的 ID 也可能相互混杂，这严重影响了遍历的效果。之后我考虑了微博开放平台提供的 API，遇到了收费 800 元的博主。最终，我在这个 GitHub 仓库 中找到了启发。


综合考量了经济，效率等条件，我选择了一种与上述仓库类似的遍历策略。分析微博页面 https://m.weibo.cn/p/index?containerid=2304410027_&extparam=8008611000000000000 ，我很快找到了返回数据的接口，用于爬取 POI 名称。我发现地标和美食两个板块的数据接口只有一个参数值不同，取值规则也与仓库中提到的一致。后续再利用关键字搜索去获取更多的数据，形成完整的爬取思路。


遍历POI


基于此，我将遍历的过程封装成  get_poi_list  函数，代码如下：


with open('config/landmark_params.txt', 'r', encoding='utf-8') as f:
    landmark_params = json.loads(f.read())  	# poi 遍历地标接口
with open('config/restaurant_params.txt', 'r', encoding='utf-8') as f:
    restaurant_params = json.loads(f.read())    # poi 遍历美食接口
poi_list_url = "https://m.weibo.cn/api/container/getIndex"  # 微博 poi 遍历接口
def get_poi_list(category=1, page=1):
    """
        category=1: 地标
        category=2: 美食
    """
    if category == 1:
        landmark_params['page'] = page
        response = requests.get(poi_list_url, params=landmark_params)
    elif category == 2:
        restaurant_params['page'] = page
        response = requests.get(poi_list_url, params=restaurant_params)
    else:
        return []
    if json.loads(response.text)['ok'] == 1:
        groups = json.loads(response.text)["data"]["cards"][-1]["card_group"]
        lst = [group['title_sub'] for group in groups]
        return lst
    else:
        return []
但是这种遍历方法的缺点在于，这些页面提供的 POI 数据量并不大，比如北京只有 5000 条左右，加上关键字搜索提供的额外数据，也只有一万多条数据。因此，如何优化这一部分，是该项目后续改进的重点方向之一。


# 收集经纬度与 ID 信息


通过微博的关键词搜索接口，可以获取 POI 的 ID、名称以及相应的地理坐标（经纬度）等信息。这些信息被保存在请求响应的 "pois" 字段中。但是我测试时发现，仅使用  keyword  参数和 cookies 调用此接口无法成功，还需额外添加请求头信息。于是我将请求头从浏览器中复制下来保存到本地文件，在代码运行开始时读取。


with open('config/headers.txt', 'r', encoding='utf-8') as f:
    search_headers = json.loads(f.read())
接下来，定义一个  search_by_keyword  函数，该函数封装了使用关键词进行搜索的逻辑：


def search_by_keyword(keyword=''):
    try:
        response = requests.get(search_url, params={'keyword': keyword}, headers=search_headers, cookies=cookies_dict)
        data = json.loads(response.text)
        if len(data) > 0:
            print(f'关键字\'{keyword}\'搜索到{len(data["pois"])}条数据')
            return data['pois']
        else:
            print(f'No results found for keyword \'{keyword}\'.')
            return []
    except Exception as e:
        print(f'An error occurred during the search for keyword \'{keyword}\': {e}')
        return []
此函数首先使用给定的关键词发起 GET 请求，并将返回的数据解析为 JSON 格式。如果请求成功，则返回这些 POI 信息。如果搜索没有返回任何结果，或者在请求过程中发生异常，则返回一个空列表。


# 收集文本描述数据


利用关键词搜索接口获取的数据包含了 POIID、经纬度等关键信息。通过 POIID 可以构造出对应的微博页面 URL。经过一番探索，我发现 POI 微博页面的简介信息是通过 HTTP 接口以 HTML、CSS 和 JavaScript 资源的形式返回的。因此，可以应用正则表达式对获取的数据进行匹配，以截取并拼接出完整的文本描述。


def get_poi_text(poi_id):
    target_url = poi_text_base_url + str(poi_id)  # 构造目标网页 URL
    pattern = re.compile(r'<p class="p_txt">(.*?)</p>', re.DOTALL)
    matches = []
    try:
        response = requests.get(target_url, cookies=cookies_dict)
        matches = pattern.findall(response.text)
    except Exception as e:
        print(f'Error retrieving text for POI ID {poi_id}: {e}')
    return '\n'.join(text_matches)
本来这个函数这么写就已经完成了，然而，在程序运行一段时间后，我发现一大部分数据缺失了文本描述，但程序仍在继续执行。经分析，这可能是由于微博对频繁访问的 IP 实施了访问限制。每次通过接口获取 POIID 后，对页面的访问相当于一次页面请求，因此容易被微博识别并限制。


解决此问题的方法之一是使用 IP 代理池，从代理服务获取不同的 IP 地址，并在访问时携带这些代理 IP。通过定期更换代理 IP，可以降低被服务器识别和限制的风险。即使某个 IP 被限制，也可以迅速切换到另一个。


想法是很美好的，但是现实却是，在我调研学习了编写代理池的方法后，尝试了几个免费的代理服务商提供的 IP，全部都被微博拦截，导致根本请求不到任何数据。无奈之下，我选择了最简单的等待策略：当检测到 IP 被限制时，程序将等待一段时间后再次尝试，通常等待约 5 分钟后 IP 会被解封。


for times in range(1, 6):
    try:
        response = requests.get(target_url, cookies=cookies_dict)
        matches = pattern.findall(response.text)
        if matches:
            return '\n'.join(matches)
        else:
            print(f"No text found for POI ID {poi_id} on attempt {times}.")
    except Exception as e:
        print(f"Attempt {times} failed for POI ID {poi_id}: {e}")
        time.sleep(80 * times)
# 保存数据


上述方法已经可以完整地获取所需的 POI 数据，现在只需要在主体函数中组合上述方法形成完整的爬取流程即可。


因为遍历 POI 时用到的页面包括地标和美食两个板块，用到的参数有所不同，所以爬取过程也是按照这两块内容分别爬取，但爬取的逻辑是一致的，可以代码复用。


除此之外，接口调用需要传入页面参数，所以维护一个页面数据结构来跟踪当前的页面状态。


page = {
     "landmark": landmark_params["page"],
     "restaurant": restaurant_params["page"]
 }
page  参数会随程序运行而动态更新，但是程序运行时间可能很长，还有可能因各种情况意外终止。为了避免每次运行程序都从第一页开始爬取，这个数据需要时不时更新到文件中。


def save_params(landmark_page=1, restaurant_page=1):
    landmark_params["page"] = landmark_page
    restaurant_params["page"] = restaurant_page
    with open('config/restaurant_params.txt', 'r', encoding='utf-8') as file:
        json.dump(restaurant_params, file)
    with open('config/landmark_params.txt', 'r', encoding='utf-8') as file:
        json.dump(landmark_params, file)
在准备工作完成后，接着编写爬虫的主体逻辑。首先使用  get_poi_list  函数获取 POI 的名称列表，然后遍历调用  search_by_keyword  函数获取基本的数据，接着调用  get_poi_text  函数获取文本，最后将数据保存到文件中。


考虑到数据存在重复的可能，比如关键字 A 搜索出来的结果包括兴趣点 P，而关键字 B 搜索出来的结果也包含 P，此时没有必要获取两次 P 的文本数据，因此每次调用  get_poi_text  函数前需要先判断 POI 数据是否已经在文件中。此外，如果每次获取一个 POI 的数据就写入文件，会导致程序 I/O 过多，效率降低。因此我选择将一个关键字搜索出来了一系列数据一次性写入文件。


综上可以封装以下两个函数：


    

  • write_csv  接受数据并去重，然后调用  get_poi_text  补充文本数据，最后写入文件中；
    • 

  • save_by_keywords  将关键字搜索，文本数据爬取和文件写入的过程再进行一次封装以方便  main  函数调用。
    • 



unique_rows = set()  # 数据文件唯一标识集合
with open(file_name, 'r', encoding='gbk') as file:
    reader = csv.reader(file)
    for row in reader:
        unique_rows.add(row[0])
def write_csv(data):
    if not data:
        return
    with open(file_name, 'a', newline='', encoding='gbk') as file:
        fields = ['id', 'title', 'longitude', 'latitude', 'address', 'text']
        writer = csv.DictWriter(file, fieldnames=fields)
        for item in data:
            if item['poiid'] not in unique_rows:
                try:
                    writer.writerow({
                        'id': item['poiid'],
                        'title': item['title'],
                        'longitude': item['lon'],
                        'latitude': item['lat'],
                        'address': item['address'],
                        'text': get_poi_text(item['poiid'])
                    })
                except Exception as e:
                    print("Write to CSV ERROR", e)
                    continue
def save_by_keywords(keywords=None):
    if keywords is None:
        return
    for keyword in keywords:
        data = search_by_keyword(keyword)
        write_csv(data)
在  main  函数中，循环获取 POI 列表，调用  save_by_keywords  爬取数据并保存。


def main():
    page = {
        "landmark": landmark_params["page"],
        "restaurant": restaurant_params["page"]
    }
    categories = [(1, 'landmark'), (2, 'restaurant')]
    for category, name in categories:
        res = get_poi_list(category, page[name])
        while res:
            save_by_keywords(res)
            page[name] += 1
            res = get_poi_list(category, page[name])
            save_params(page['landmark'], page['restaurant'])
# 数据清洗


由于一开始爬取数据时我没有考虑到 IP 会被 ban 的情况，导致中间爬取的很多数据都缺少了文本描述。因此我编写了  clean.py  文件来处理这些不完整的数据项。


这部分的逻辑相对比较简单，即读取 CSV 文件中的每一列，检出不完整有的部分，重现调用上面的函数再做一次爬取。但是由于文件不好在特定位置写入，于是我干脆将原先的  data.csv  数据全部读出并做数据清洗，将新的完整的数据写入  new_data.csv  文件中。


整理整个代码流程可以分成如下步骤


    

  1. 

    定义  write_to_new_csv  函数负责将数据追加写入新数据文件。这部分内容与之前爬取数据存入是类似的
    

    def write_to_new_csv(data):
        try:
            with open(new_file_name, 'a', newline='', encoding='gbk') as file2:
                writer = csv.writer(file2)
                for item in data:
                    writer.writerow(item)
        except Exception as e:
            print("Write to CSV ERROR", e)
    2. 

  2. 

    定义主体函数  clean  ，读取新数据文件，收集已处理行的唯一标识符。这一步主要是为了防止程序意外终止，后续清理时重复写入数据，增加程序适用性。
    

    unique_rows = set()
    try:
        with open(new_file_name, 'r', encoding='gbk') as file:
            reader = csv.reader(file)
            for row in reader:
                unique_rows.add(row[0])  # 唯一标识是 poiid
    except Exception as e:
        print("read new_data.csv failed: ", e)
    3. 

  3. 

    接着读取原始数据文件，对于未处理的行，如果其 text 字段为空，则尝试重新调用  get_poi_text  函数获取文本信息并更新。
    

    with open(file_name, 'r', encoding='gbk') as file1:
        reader = csv.reader(file1)
        origin_rows = list(reader)
        updated_rows = []
        for index, row in enumerate(origin_rows):
            if row[0] not in unique_rows:
                if row[-1] == '':
                    try:
                        row[-1] = get_poi_text(row[0])
                        print(f"get the new poi text of {row[1]}, the poi text is: {row[-1]}")
                    except Exception as e:
                        print(f"Error fetching text for {row[1]}: {e}")
                        continue
                unique_rows.add(row[0])
                updated_rows.append(row)
    4. 

  4. 

    考虑到每次更新的数据都会被添加到列表中，这会直接占用内存，如果数据量一大很可能导致内存泄露。因此有必要即时地将新数据存储到指定文件中。另一方面，我为了防止程序异常终止导致列表中的数据没有及时存入文件中，因此我设置的是每处理 30 行或处理完所有行后写入新数据文件。
    

    if len(updated_rows) >= 30 or index == len(origin_rows) - 1:
        write_to_new_csv(updated_rows)
        updated_rows = []
    5. 



 

# 结果展示


爬取过程截图如下：


爬取流程截图


最终获取的 csv 文件内容预览：


数据截图


 

# 补充


# 不足


目前这个爬虫程序在数据收集方面取得了一定的进展，但仍存在一些明显的不足之处：


    

  1. POI 遍历方法的局限性：目前所采用的 POI 遍历方法尚未达到理想的效率。尽管已经能够获取到一定量级的数据，但与预期目标相比，数据量仍显不足，仅达到万级规模。不过更好的遍历方法我现在也尚未发现。
    2. 

  2. 代理池的缺失：由于微博严格的反爬机制，本项目并没有使用代理池。这样就导致数据收集速度受限。例如，我在服务器上运行爬虫一夜，也仅完成了对北京市数据的收集，且数量仅有一万多条。
    3. 



但这篇博客的目的在于记录和学习，所以我就简单地总结本项目中我在代理池上的实践和收获。这里附上有关代理池基础知识的两篇参考博客 [1]、[2]。


# 代理池


我在  proxy.py  文件里封装了一系列函数。具体的代码实现如下：


# 引入依赖库


首先，脚本引入了所需的 Python 库：


import requests
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
import random
import datetime
# 代理有效性检查


check_proxy  函数用于测试代理 IP 的有效性。通过向百度发起一个 GET 请求，如果响应状态码为 200，认为代理有效。


def check_proxy(proxy):
    try:
        response = requests.get('https://www.baidu.com', proxies={'http': proxy, 'https': proxy}, timeout=3)
        return response.status_code == 200
    except requests.RequestException:
        return False
# 从服务商中获取代理 IP


get_ihuan_ip 、 get_proxy_list  函数分别从 ihuan 网站和 proxy-list 网站获取代理 IP。


将不同代理网站爬取 IP 的逻辑写到对应的函数中，这样可以很方便地扩充新的代理。


def get_ihuan_ip():
    ...
def get_proxy_list():
    ...
用列表存储所有代理网站的接口，便于管理和调用。


proxy_funcs = [get_ihuan_ip, get_proxy_list]
proxy_func_names = ["ihuan", "proxy-list"]
# 测试代理可用性


封装  test  函数遍历所有代理获取函数，测试并打印出哪些代理源是有用的。


def test():
    print("------ try test proxy func ---------")
    for index in range(len(proxy_funcs)):
        try:
            lst = proxy_funcs[index]()
            if lst:
                print(proxy_func_names[index], "is useful!")
            else:
                print(proxy_func_names[index], "is useless... for the reason that there is no useful ip")
        except Exception as e:
            print(proxy_func_names[index], "is useless... for the reason that", e)
# 获取可用代理 IP


最后是主函数  get_proxys ，该函数尝试从所有代理源获取代理 IP，直到返回第一个成功的结果。


def get_proxys():
    for fun in proxy_funcs:
        lst = fun()
        if lst:
            return lst
    return []
# 开源


目前本项目已全部开源（传送门)。