AI 回答里的引用链接怎么清洗：跟踪参数、同站多域名、临时链接和失效率（附 Python）

"""把 AI 回答里引用的 URL 归一成「页面键」：同一篇内容，不管带什么参数、从哪个域名进来，键都一样。"""
import re
from urllib.parse import urlsplit, parse_qsl, urlencode

# ---------- 1. 跳转壳：真实地址就在参数里 ----------
UNWRAP = {
    "link.wtturl.cn": "target",
    "link.zhihu.com": "target",
    "link.juejin.cn": "target",
}

# ---------- 2. 同站多域名：只换 host，路径不变 ----------
HOST_ALIAS = {
    "cloud.tencent.com.cn": "cloud.tencent.com",
    "cloud.tencent.cn": "cloud.tencent.com",
    "m.sohu.com": "www.sohu.com",
    "news.sohu.com": "www.sohu.com",
    "m.163.com": "www.163.com",
    "m.jiemian.com": "www.jiemian.com",
    "toutiao.com": "www.toutiao.com",
    "m.toutiao.com": "www.toutiao.com",
    "m.toutiaocdn.com": "www.toutiao.com",
}

# ---------- 3. 路径改写：(host 正则, path 正则, 新 host, 新 path) ----------
PATH_RULES = [
    (r"developer\.cloud\.tencent\.com", r"/article/(\d+)", "cloud.tencent.com", "/developer/article/{0}"),
    (r"www\.toutiao\.com", r"/(?:article/|group/|a)(\d+)", "www.toutiao.com", "/article/{0}"),
    (r"www\.iesdouyin\.com", r"/share/video/(\d+)", "www.douyin.com", "/video/{0}"),
    (r"dy\.163\.com", r"/article/(\w+)\.html", "www.163.com", "/dy/article/{0}.html"),
    (r"view\.inews\.qq\.com", r"/[ak]/(\w+)", "news.qq.com", "/rain/a/{0}"),
    # 地方频道子域：同一 ID 在不同子域都能打开，canonical 却各指一处，只能当键用
    (r"[\w-]+\.ifeng\.com", r"/c/(\w+)", "ifeng.com", "/c/{0}"),
]
PATH_RULES = [(re.compile(h), re.compile(p + r"/?"), nh, np) for h, p, nh, np in PATH_RULES]

# ---------- 4. 参数：认识的站点用白名单，其余用黑名单 ----------
KEEP = {
    "baijiahao.baidu.com": {"id"},
    "mp.weixin.qq.com": {"__biz", "mid", "idx", "sn"},
    "www.toutiao.com": set(), "www.douyin.com": set(), "cloud.tencent.com": set(),
    "zhuanlan.zhihu.com": set(), "www.zhihu.com": set(), "blog.csdn.net": set(),
    "www.sohu.com": set(), "www.163.com": set(), "news.qq.com": set(), "ifeng.com": set(),
    "www.jiemian.com": set(), "k.sina.cn": set(), "baike.baidu.com": set(),
}
DROP = re.compile(r"(utm_\w+|spm|scm|from|frompage|source|scene|share\w*|ref|referfrom|trk|"
                  r"fbclid|gclid|msclkid|msockid|policyid|wfr|chksm|previous_page|upstream_biz|"
                  r"wid|spss|hubs_\w+|_hsenc|_hsmi)", re.I)

HOST_OK = re.compile(r"[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)+")
TRAIL = "）)】]》>。，、；;：:！!？?'\"`*"


def page_key(url):
    """返回页面键；不是合法网址返回 None。键不一定能直接打开（比如凤凰网那条），只用来去重计数"""
    url = url.strip().rstrip(TRAIL)
    try:
        s = urlsplit(url)
        host = (s.hostname or "").lower().rstrip(".")
    except ValueError:
        return None
    if s.scheme not in ("http", "https") or not HOST_OK.fullmatch(host):
        return None

    if host in UNWRAP:
        inner = dict(parse_qsl(s.query)).get(UNWRAP[host])
        if inner and inner.startswith("http"):
            return page_key(inner)

    host = HOST_ALIAS.get(host, host)
    path = re.sub(r"/{2,}", "/", s.path) or "/"
    for hp, pp, nh, np in PATH_RULES:
        m = pp.fullmatch(path) if hp.fullmatch(host) else None
        if m:
            host, path = nh, np.format(*m.groups())
            break
    if len(path) > 1:
        path = path.rstrip("/")

    q = parse_qsl(s.query, keep_blank_values=True)
    if host in KEEP:
        kept = [(k, v) for k, v in q if k in KEEP[host]]
        # 白名单里的参数一个都没有（比如带签名的临时链接），认不出是哪篇，参数原样留着
        q = kept if kept or not KEEP[host] else q
    else:
        q = [(k, v) for k, v in q if not DROP.fullmatch(k)]
    q = sorted(set(q))

    # 片段：#/ 和 #! 是前端路由，算页面的一部分；其他（#1、#respond、#toc）都扔掉
    frag = s.fragment if s.fragment.startswith(("/", "!")) else ""
    key = "https://" + host + path + ("?" + urlencode(q) if q else "")
    return key + ("#" + frag if frag else "")

from collections import defaultdict
from urlnorm import page_key

urls = """
https://blog.csdn.net/someone/article/details/123456789#1#1
https://blog.csdn.net/someone/article/details/123456789
https://zhuanlan.zhihu.com/p/1000000000000000001?utm_medium=openapi_platform&utm_source=abc
https://zhuanlan.zhihu.com/p/1000000000000000001#1
https://baijiahao.baidu.com/s?id=1800000000000000001&wfr=spider&for=pc
https://mp.weixin.qq.com/s?__biz=MzAxMTEx==&mid=2650000001&idx=2&sn=abc123&chksm=ff00aa&scene=27
http://mp.weixin.qq.com/s?mid=2650000001&__biz=MzAxMTEx==&sn=abc123&idx=2
http://mp.weixin.qq.com/s?src=11&timestamp=1790000000&ver=6990&signature=xyz
https://cloud.tencent.com.cn/developer/article/2700001?policyId=1003#1#1
https://cloud.tencent.cn/developer/article/2700001?from=15425&frompage=seopage
https://developer.cloud.tencent.com/article/2700001
http://m.toutiao.com/group/7600000000000000001/
https://www.toutiao.com/a7600000000000000001
https://www.toutiao.com/article/7600000000000000001/?upstream_biz=toutiao_pc&wid=1789
https://www.iesdouyin.com/share/video/7600000000000000002
http://i.ifeng.com/c/8abcdEFGhij
https://nb.ifeng.com/c/8abcdEFGhij
https://link.wtturl.cn/?target=https%3A%2F%2Fwww.example.org%2F%3Fp%3D42%26utm_source%3Dx&scene=im
https://www.example.org/?p=42）。
https://app.example.org/#/docs/rate
https://`，或在换行处截断
""".split()

groups = defaultdict(list)
for u in urls:
    groups[page_key(u)].append(u)
for k, us in groups.items():
    print(f"{len(us)}  {k}")
print(f"{len(urls)} 种写法 -> {len([k for k in groups if k])} 个页面，{len(groups.get(None, []))} 条不是网址")

2  https://blog.csdn.net/someone/article/details/123456789
2  https://zhuanlan.zhihu.com/p/1000000000000000001
1  https://baijiahao.baidu.com/s?id=1800000000000000001
2  https://mp.weixin.qq.com/s?__biz=MzAxMTEx%3D%3D&idx=2&mid=2650000001&sn=abc123
1  https://mp.weixin.qq.com/s?signature=xyz&src=11&timestamp=1790000000&ver=6990
3  https://cloud.tencent.com/developer/article/2700001
3  https://www.toutiao.com/article/7600000000000000001
1  https://www.douyin.com/video/7600000000000000002
2  https://ifeng.com/c/8abcdEFGhij
2  https://www.example.org/?p=42
1  https://app.example.org/#/docs/rate
1  None
21 种写法 -> 11 个页面，1 条不是网址

"""逐个打开引用链接，判断还能不能看：ok / gone / soft404 / login / blocked / shell / error"""
import concurrent.futures as cf
import json, re, sys, urllib.error, urllib.request
from collections import Counter

UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0 Safari/537.36"
SOFT_404 = ["链接已过期", "已被发布者删除", "因违规无法查看", "内容已删除", "内容不存在", "文章不存在",
            "页面不存在", "页面找不到", "该文章已被删除", "page not found", "404 not found"]
LOGIN_URL = re.compile(r"(^|[./])(login|signin|passport|sso)([./?]|$)", re.I)
LOGIN_TEXT = ["登录后查看", "登录后才能查看", "请先登录", "扫码登录"]
BLOCK_TEXT = ["安全验证", "访问过于频繁", "人机验证", "captcha", "just a moment"]
SCRIPT = re.compile(r"<(script|style)[^>]*>.*?</\1>", re.S | re.I)
TAG = re.compile(r"<[^>]+>")


def visible_text(html):
    return re.sub(r"\s+", " ", TAG.sub(" ", SCRIPT.sub(" ", html))).strip().lower()


def classify(status, final_url, html):
    if status in (404, 410):
        return "gone"
    path = final_url.split("?")[0]
    if LOGIN_URL.search(path):
        return "login"
    if "captcha" in path.lower():
        return "blocked"
    if status in (401, 403, 412, 418, 429, 503, 521):
        return "blocked"
    if status >= 400:
        return "error"
    text = visible_text(html)
    if len(text) < 2000:            # 错误页、登录墙的正文都很短，长正文里出现这些词多半是文章内容
        for kind, words in (("soft404", SOFT_404), ("login", LOGIN_TEXT), ("blocked", BLOCK_TEXT)):
            if any(w in text for w in words):
                return kind
    if len(text) < 50:              # 正文全靠 JS 渲染（抖音、头条都是），静态请求看不出死活
        return "shell"
    return "ok"


def fetch(url, timeout=15):
    req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:      # 默认跟随跳转，短链也在这一步展开
            return r.status, r.geturl(), r.read(500_000).decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, e.geturl() or url, e.read(200_000).decode("utf-8", "replace")


def check(url, retries=1):
    for attempt in range(retries + 1):
        try:
            status, final, html = fetch(url)
            m = re.search(r'<link[^>]+rel=["\']canonical["\'][^>]*href=["\']([^"\']+)', html, re.I)
            return {"url": url, "status": status, "final": final,
                    "canonical": m.group(1) if m else None, "kind": classify(status, final, html)}
        except Exception as e:      # 超时、DNS、TLS、连接被重置
            err = f"{type(e).__name__}: {e}"[:120]
    return {"url": url, "status": None, "final": None, "canonical": None, "kind": "error", "error": err}


def run(urls, workers=16):
    with cf.ThreadPoolExecutor(workers) as ex:
        return list(ex.map(check, urls))


if __name__ == "__main__":
    urls = [l.strip() for l in open(sys.argv[1], encoding="utf-8") if l.strip()]
    results = run(urls)
    with open(sys.argv[2], "w", encoding="utf-8") as f:
        for r in results:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    c = Counter(r["kind"] for r in results)
    for k in ("ok", "gone", "soft404", "login", "blocked", "shell", "error"):
        print(f"{k:<8}{c[k]:>5}  {c[k] / len(results):6.1%}")

"""本机起一个假站点，把几种失效情况各做一个页面，跑一遍 linkcheck"""
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from linkcheck import run

ARTICLE = "<html><head><link rel='canonical' href='https://news.example/a/1'></head><body>" + "正文。" * 800 + "</body></html>"
PAGES = {
    "/a/1": (200, {}, ARTICLE),
    "/s/xyz": (302, {"Location": "/a/1"}, ""),                                  # 短链
    "/a/2": (404, {}, "not found"),
    "/s?src=11&signature=abc": (200, {}, "<p>链接已过期</p>"),                  # 临时链接过期，状态码却是 200
    "/p/3": (302, {"Location": "/passport/login?next=/p/3"}, ""),
    "/passport/login?next=/p/3": (200, {}, "<p>请先登录</p>"),
    "/p/4": (403, {}, "forbidden"),
    "/video/5": (200, {}, "<div id='root'></div><script>render()</script>"),   # 前端渲染的空壳
}


class H(BaseHTTPRequestHandler):
    def do_GET(self):
        status, headers, body = PAGES.get(self.path, (404, {}, ""))
        self.send_response(status)
        for k, v in headers.items():
            self.send_header(k, v)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        self.wfile.write(body.encode())

    def log_message(self, *a):
        pass


srv = ThreadingHTTPServer(("127.0.0.1", 8765), H)
threading.Thread(target=srv.serve_forever, daemon=True).start()
base = "http://127.0.0.1:8765"
urls = [base + p for p in ["/a/1", "/s/xyz", "/a/2", "/s?src=11&signature=abc", "/p/3", "/p/4", "/video/5"]]
urls.append("http://127.0.0.1:9/nothing")                                     # 端口没开，连接失败
for r in run(urls, workers=4):
    final = (r["final"] or "-").replace(base, "")
    print(f'{r["kind"]:<8}{str(r["status"]):<5}{r["url"].replace(base, ""):<28}-> {final:<28}{r["canonical"] or ""}')
srv.shutdown()

ok      200  /a/1                        -> /a/1                        https://news.example/a/1
ok      200  /s/xyz                      -> /a/1                        https://news.example/a/1
gone    404  /a/2                        -> /a/2                        
soft404 200  /s?src=11&signature=abc     -> /s?src=11&signature=abc     
login   200  /p/3                        -> /passport/login?next=/p/3   
blocked 403  /p/4                        -> /p/4                        
shell   200  /video/5                    -> /video/5                    
error   None http://127.0.0.1:9/nothing  -> -