AIyaodianzhushou/data-pipeline/crawlers/probe.py

"""
多源数据探头 — 测试各药品数据源的可用性
"""
import asyncio
import httpx
import re
from bs4 import BeautifulSoup

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}


async def probe(url, name, method="GET", json_data=None, timeout=15):
    """探测一个数据源"""
    try:
        async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as c:
            if method == "POST":
                resp = await c.post(url, headers=HEADERS, json=json_data)
            else:
                resp = await c.get(url, headers=HEADERS)

        size = len(resp.text)
        status = resp.status_code
        snippet = resp.text[:200].replace("\n", " ").strip()

        # 检查是否有验证码/反爬
        anticrawl = any(
            k in resp.text.lower()
            for k in ["captcha", "验证码", "滑块", "verify", "forbidden", "403", "blocked", "请滑动"]
        )

        flag = "⚠ 反爬" if anticrawl else "✅" if status == 200 else f"❌ {status}"
        has_drug = bool(re.search(r"阿莫西林|药品|批号|标准", resp.text)) if status == 200 else False

        print(f"{flag} {name}")
        print(f"    URL: {url}")
        print(f"    {status} | {size:,} bytes | 含药品信息: {has_drug}")
        print(f"    {snippet}\n")
        return {"ok": status == 200 and not anticrawl, "size": size, "has_drug": has_drug}
    except Exception as e:
        print(f"❌ {name}: {str(e)[:100]}\n")
        return {"ok": False, "size": 0, "has_drug": False}


async def main():
    print("═" * 60)
    print("  数据源探测报告")
    print("═" * 60 + "\n")

    sources = [
        # 1. NMPA 公告列表（静态页面）
        ("NMPA 最新公告", "https://www.nmpa.gov.cn/xxgk/ggtg/index.html"),

        # 2. NMPA 药品查询 API（可能存在的 JSON 接口）
        ("NMPA 药品查询 API (试探)", 
         "https://www.nmpa.gov.cn/datasearch/data/nmpa/search.json",
         "POST",
         {"category": "ps", "condition": "阿莫西林", "pageSize": 5, "pageNum": 1}),

        # 3. 药智网（公开信息）
        ("药智网-阿莫西林", "https://www.yaozh.com/product/103009.html"),

        # 4. 中国药典在线（公开）
        ("药典在线", "http://www.drugfuture.com/standard/"),

        # 5. 药智数据 API（公开试用）
        ("药智数据 search", "https://db.yaozh.com/drugcompatible/search"),

        # 6. 国家药品标准查询
        ("药品标准查询", "https://www.nifdc.org.cn/nifdc/xxgk/bzxx/index.html"),

        # 7. 丁香园用药助手
        ("丁香园用药助手", "https://drugs.dxy.cn/search/index.htm"),

        # 8. 药物在线（drugfuture）具体药品
        ("药物在线-阿莫西林", "http://www.drugfuture.com/standard/search.aspx?key=%B0%A2%C4%AA%CE%F7%C1%D6"),

        # 9. 百度百科（结构化药品信息）
        ("百度百科-阿莫西林", "https://baike.baidu.com/item/%E9%98%BF%E8%8E%AB%E8%A5%BF%E6%9E%97"),
    ]

    results = []
    for args in sources:
        name = args[0]
        url = args[1]
        method = args[2] if len(args) > 2 else "GET"
        json_data = args[3] if len(args) > 3 else None
        r = await probe(url, name, method, json_data)
        results.append((name, r))

    print("═" * 60)
    print("  总结")
    print("═" * 60)
    ok_sources = [(n, r) for n, r in results if r["ok"] and r["has_drug"]]
    partial = [(n, r) for n, r in results if r["ok"] and not r["has_drug"]]
    failed = [(n, r) for n, r in results if not r["ok"]]

    print(f"\n✅ 可用（含药品数据）: {len(ok_sources)}")
    for n, r in ok_sources:
        print(f"   - {n} ({r['size']:,} bytes)")

    print(f"\n🟡 可访问但无药品数据: {len(partial)}")
    for n, r in partial:
        print(f"   - {n} ({r['size']:,} bytes)")

    print(f"\n❌ 不可用: {len(failed)}")
    for n, r in failed:
        print(f"   - {n}")

    print(f"\n推荐优先采集: ", end="")
    if ok_sources:
        print(", ".join(n for n, _ in ok_sources[:3]))
    else:
        print("无可用数据源，建议手动整理药典 PDF 或购买第三方 API")


if __name__ == "__main__":
    asyncio.run(main())