1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147# coding=utf-8
"""
URL 处理工具模块
提供 URL 标准化功能，用于去重时消除动态参数的影响：
- normalize_url: 标准化 URL，去除动态参数
"""
from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
from typing import Dict, Set
# 各平台需要移除的特定参数
# - weibo: 有 band_rank（排名）和 Refer（来源）动态参数
# - 其他平台: URL 为路径格式或简单关键词查询，无需处理
PLATFORM_PARAMS_TO_REMOVE: Dict[str, Set[str]] = {
# 微博：band_rank 是动态排名参数，Refer 是来源参数，t 是时间范围参数
# 示例：https://s.weibo.com/weibo?q=xxx&t=31&band_rank=1&Refer=top
# 保留：q（关键词）
# 移除：band_rank, Refer, t
"weibo": {"band_rank", "Refer", "t"},
}
# 通用追踪参数（适用于所有平台）
# 这些参数通常由分享链接或广告追踪添加，不影响内容识别
COMMON_TRACKING_PARAMS: Set[str] = {
# UTM 追踪参数
"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
# 常见追踪参数
"ref", "referrer", "source", "channel",
# 时间戳和随机参数
"_t", "timestamp", "_", "random",
# 分享相关
"share_token", "share_id", "share_from",
}
def normalize_url(url: str, platform_id: str = "") -> str:
"""
标准化 URL，去除动态参数
用于数据库去重，确保同一条新闻的不同 URL 变体能被正确识别为同一条。
处理规则：
1. 去除平台特定的动态参数（如微博的 band_rank）
2. 去除通用追踪参数（如 utm_*）
3. 保留核心查询参数（如搜索关键词 q=, wd=, keyword=）
4. 对查询参数按字母序排序（确保一致性）
Args:
url: 原始 URL
platform_id: 平台 ID，用于应用平台特定规则
Returns:
标准化后的 URL
Examples:
>>> normalize_url("https://s.weibo.com/weibo?q=test&band_rank=6&Refer=top", "weibo")
'https://s.weibo.com/weibo?q=test'
>>> normalize_url("https://example.com/page?id=1&utm_source=twitter", "")
'https://example.com/page?id=1'
"""
if not url:
return url
try:
# 解析 URL
parsed = urlparse(url)
# 如果没有查询参数，直接返回
if not parsed.query:
return url
# 解析查询参数
params = parse_qs(parsed.query, keep_blank_values=True)
# 收集需要移除的参数（使用小写进行比较）
params_to_remove: Set[str] = set()
# 添加通用追踪参数
params_to_remove.update(COMMON_TRACKING_PARAMS)
# 添加平台特定参数
if platform_id and platform_id in PLATFORM_PARAMS_TO_REMOVE:
params_to_remove.update(PLATFORM_PARAMS_TO_REMOVE[platform_id])
# 过滤参数（参数名转小写进行比较）
filtered_params = {
key: values
for key, values in params.items()
if key.lower() not in {p.lower() for p in params_to_remove}
}
# 如果过滤后没有参数了，返回不带查询字符串的 URL
if not filtered_params:
return urlunparse((
parsed.scheme,
parsed.netloc,
parsed.path,
parsed.params,
"", # 空查询字符串
"" # 移除 fragment
))
# 重建查询字符串（按字母序排序以确保一致性）
sorted_params = []
for key in sorted(filtered_params.keys()):
for value in filtered_params[key]:
sorted_params.append((key, value))
new_query = urlencode(sorted_params)
# 重建 URL（移除 fragment）
normalized = urlunparse((
parsed.scheme,
parsed.netloc,
parsed.path,
parsed.params,
new_query,
"" # 移除 fragment
))
return normalized
except Exception:
# 解析失败时返回原始 URL
return url
def get_url_signature(url: str, platform_id: str = "") -> str:
"""
获取 URL 的签名（用于快速比较）
基于标准化 URL 生成签名，可用于：
- 快速判断两个 URL 是否指向同一内容
- 作为缓存键
Args:
url: 原始 URL
platform_id: 平台 ID
Returns:
URL 签名字符串
"""
return normalize_url(url, platform_id)