使用Python通过SPARQL端点将RDF数据存入三元组存储及代码咨询

你需要实现两个核心功能：一是用Python通过SPARQL端点将RDF数据存入三元组存储；二是抓取http://gnafld.net/address/?per_page=10&page=7中的地址数据，将其转换为RDF三元组后存入同一块三元组存储，方便后续查询。

import requests
from bs4 import BeautifulSoup
import pandas as pd
import numpy as np
import re

url='http://gnafld.net/address/?per_page=10&page=7'
page = requests.get(url)
response = requests.get(url)
response.raise_for_status()
results = re.findall('"Address ID: (GAACT[0-9]+)"', response.text)
address1=results[0]
a = "http://gnafld.net/address/"
new_url = a + address1
r = reques...

我会从网页数据抓取优化、RDF三元组构建、SPARQL端点数据插入三个环节帮你完善代码：

1. 优化网页数据抓取逻辑 #

你当前用正则匹配地址ID的方式不够健壮，一旦页面HTML结构微调就会失效。建议用BeautifulSoup解析页面，精准定位地址数据：

• 先解析列表页拿到所有地址详情页的链接或ID
• 再遍历每个地址详情页，提取完整的地址信息（比如街道、城市、邮编等）

2. 用RDFLib构建RDF三元组 #

Python处理RDF最常用的库是rdflib，我们可以用它来定义命名空间、创建三元组：

• 先安装依赖：pip install rdflib requests beautifulsoup4
• 定义地址相关的命名空间（比如复用标准本体如schema.org，或自定义地址本体）

3. 通过SPARQL端点插入三元组 #

你可以用SPARQL的INSERT DATA语句，或者直接用RDFLib的SPARQLUpdateStore来连接端点并插入数据。

完整示例代码 #

import requests
from bs4 import BeautifulSoup
from rdflib import Graph, Namespace, URIRef, Literal
from rdflib.namespace import RDF

# 1. 定义命名空间（复用schema.org的地址标准本体）
SCHEMA = Namespace("http://schema.org/")
ADDRESS = Namespace("http://example.org/address/")

# 2. 抓取地址列表页，获取所有地址详情链接
def get_address_detail_urls(list_url):
    response = requests.get(list_url)
    response.raise_for_status()
    soup = BeautifulSoup(response.text, "html.parser")
    # 根据页面实际结构调整选择器，这里匹配所有指向地址详情的链接
    address_links = soup.find_all("a", href=True)
    detail_urls = []
    for link in address_links:
        if link.get("href").startswith("/address/GAACT"):
            detail_urls.append(f"http://gnafld.net{link['href']}")
    return detail_urls

# 3. 抓取单个地址详情页的核心信息
def scrape_address_detail(detail_url):
    response = requests.get(detail_url)
    response.raise_for_status()
    soup = BeautifulSoup(response.text, "html.parser")
    
    # 实际使用时需根据页面元素调整选择器，这里是示例逻辑
    address_id = detail_url.split("/")[-1]
    street = soup.find("div", class_="street").text.strip() if soup.find("div", class_="street") else ""
    city = soup.find("span", class_="city").text.strip() if soup.find("span", class_="city") else ""
    postcode = soup.find("span", class_="postcode").text.strip() if soup.find("span", class_="postcode") else ""
    
    return {
        "id": address_id,
        "street": street,
        "city": city,
        "postcode": postcode
    }

# 4. 将地址数据转换为RDF三元组
def address_to_rdf(address_data):
    g = Graph()
    # 绑定命名空间，简化后续书写
    g.bind("schema", SCHEMA)
    g.bind("addr", ADDRESS)
    
    # 创建地址资源的URI
    address_uri = URIRef(f"{ADDRESS}{address_data['id']}")
    # 声明资源类型为schema.org的PostalAddress
    g.add((address_uri, RDF.type, SCHEMA.PostalAddress))
    # 添加地址属性三元组
    g.add((address_uri, SCHEMA.streetAddress, Literal(address_data['street'])))
    g.add((address_uri, SCHEMA.addressLocality, Literal(address_data['city'])))
    g.add((address_uri, SCHEMA.postalCode, Literal(address_data['postcode'])))
    
    return g

# 5. 通过SPARQL端点插入三元组
def insert_rdf_to_triplestore(graph, sparql_endpoint, username=None, password=None):
    # 构建SPARQL INSERT语句
    insert_query = f"""
    INSERT DATA {{
        {graph.serialize(format='nt').decode('utf-8')}
    }}
    """
    # 发送更新请求到端点
    headers = {"Content-Type": "application/sparql-update"}
    auth = (username, password) if username and password else None
    response = requests.post(sparql_endpoint, data=insert_query, headers=headers, auth=auth)
    response.raise_for_status()
    print(f"地址ID {list(graph.subjects())[0].split('/')[-1]} 三元组插入成功！")

if __name__ == "__main__":
    # 配置参数
    list_url = "http://gnafld.net/address/?per_page=10&page=7"
    # 替换为你的三元组存储实际SPARQL更新端点（如Blazegraph、Virtuoso等）
    sparql_endpoint = "http://your-triplestore-endpoint/sparql"
    # 若端点需要认证，取消下方注释并填写信息
    # triplestore_user = "your-username"
    # triplestore_pass = "your-password"
    
    # 执行完整流程
    detail_urls = get_address_detail_urls(list_url)
    for url in detail_urls:
        addr_data = scrape_address_detail(url)
        addr_graph = address_to_rdf(addr_data)
        # 如需认证，添加auth=(triplestore_user, triplestore_pass)参数
        insert_rdf_to_triplestore(addr_graph, sparql_endpoint)

关键注意点 #

• 页面解析适配：代码中的HTML选择器（比如class_="street"）需要根据gnafld.net页面的实际结构修改，你可以用浏览器开发者工具查看元素的类名或ID。
• SPARQL端点配置：务必替换sparql_endpoint为你的三元组存储实际更新端点地址，不同存储的端点路径可能略有差异（比如Blazegraph的端点通常是/blazegraph/sparql）。
• 错误处理增强：可以根据需求添加更多异常捕获逻辑，比如网络请求超时、页面元素缺失等场景的容错处理。
• 命名空间扩展：如果需要存储更多地址属性（如州、国家），可以参考schema.org的PostalAddress定义添加更多三元组。

内容的提问来源于stack exchange，提问作者SpongeBob