XSearch 代码走读实录

"""
新闻 RSS 聚合爬虫：抓官方 RSS 源写入 ES "quotes" 索引（可搜索）。

RSS 是网站官方分发格式，抓取合规且无需对抗反爬。内置 3 个源（2026-09-27 实测可用）：
    news_chinanews  中国新闻网滚动   https://www.chinanews.com/rss/scroll-news.xml
                    （feed 无摘要，自动跟进文章页抓 div.left_zw 正文）
    news_ithome     IT之家科技新闻   https://www.ithome.com/rss/  （description 为正文开头）
    news_solidot    Solidot 科技    https://www.solidot.org/index.rss

已停用：news_people（人民网）。2026-09-27 实测其 /rss/*.xml 全部 feed 的 101 个
pubDate 冻结在 2025-06-03~06-05，即站方已停止更新 RSS；再采集只会用 md5(url)
反复覆盖同一批 16 个月前的旧文档，并让「新闻列表」出现"最新新闻是去年"的观感。
入库的既有 news_people 文档保留，可按需 /api/admin/db/purge/ 清理。
……
"""

# RSS 源配置：key 即写入 ES 的 source 字段
FEEDS = {
    "news_chinanews": {
        "label": "中国新闻网",
        "url": "https://www.chinanews.com/rss/scroll-news.xml",
        "follow_links": True,    # feed 无摘要，跟进文章页抓正文
    },
    "news_ithome": {
        "label": "IT之家",
        "url": "https://www.ithome.com/rss/",
        "follow_links": False,
    },
    "news_solidot": {
        "label": "Solidot",
        "url": "https://www.solidot.org/index.rss",
        "follow_links": False,
    },
}

def strip_html(text):
    """去掉 script/style 块、HTML 标签和实体，得到纯文本。

    人民网等站点的视频新闻会在 description 里嵌 <script> 播放器代码
    （showPlayer(...)），必须整块删除，否则 JS 代码会混进正文。
    """
    if not text:
        return ""
    # 成对的 script/style/noscript 块连内容一起删
    text = re.sub(r"<(script|style|noscript)\b[^>]*>.*?</\1\s*>", "", text,
                  flags=re.S | re.I)
    # 未闭合/自闭合的残留开标签也删掉
    text = re.sub(r"<(script|style|noscript)\b[^>]*/?>", "", text, flags=re.I)
    return unescape(re.sub(r"<[^>]+>", "", text)).strip()


def parse_date(text):
    """兼容两种 feed 日期格式：RFC822（IT之家/Solidot）与 YYYY-MM-DD HH:MM:SS。"""
    text = (text or "").strip()
    if not text:
        return None
    try:
        return parsedate_to_datetime(text)
    except (TypeError, ValueError):
        pass
    for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(text, fmt)
        except ValueError:
            continue
    return None

    async def start(self):
        for key in self.feed_keys:
            yield scrapy.Request(
                FEEDS[key]["url"],
                callback=self.parse_feed,
                cb_kwargs={"source_key": key},
                headers={"Accept": "application/rss+xml, application/xml, text/xml, */*"},
            )

    def parse_feed(self, response, source_key):
        feed = FEEDS[source_key]
        items = response.xpath("//channel/item")
        self.logger.info("[%s] feed 解析出 %d 条", feed["label"], len(items))
        for item in items:
            title = (item.xpath("./title/text()").get() or "").strip()
            link = (item.xpath("./link/text()").get() or "").strip()
            description = item.xpath("./description/text()").get() or ""
            pub_date = item.xpath("./pubDate/text()").get()
            if not title or not link:
                continue

            if feed["follow_links"] and CNS_ARTICLE_RE.search(link):
                yield scrapy.Request(
                    link,
                    callback=self.parse_chinanews_article,
                    cb_kwargs={"source_key": source_key, "rss_title": title,
                               "pub_date": pub_date},
                )
            else:
                content = strip_html(description) or title  # 无摘要退化为标题，保证可入库
                yield self._news_item(source_key, title, link, content, pub_date)

class BrowserFingerprintHeadersMiddleware(object):
    """对白名单域名替换成成套的真实浏览器指纹头，其余域名保留自报身份的 USER_AGENT。
    ……
    """

    def process_request(self, request):
        host = (urlparse(request.url).hostname or "").lower()
        if not self._matches(host):
            return None
        # 直接替换（而非 setdefault）：内置 UserAgentMiddleware（优先级 500）已先写入
        # USER_AGENT，setdefault 不会覆盖它，真实指纹头就不生效（曾导致 B站 接口 412）
        try:
            headers = get_headers()
        except Exception as e:
            logger.warning("指纹生成失败（%s），使用上一次/内置头", e)
            return None
        for k, v in headers.items():
            request.headers[k] = v
        return None

    @staticmethod
    def _news_item(source_key, title, url, content, pub_date):
        feed = FEEDS[source_key]
        return {
            "url_object_id": hashlib.md5(url.encode("utf-8")).hexdigest(),
            "title": title,
            "content": content[:MAX_CONTENT_CHARS],
            "author": feed["label"],
            "tags": ["新闻", feed["label"]],
            "url": url,
            "front_image_url": "",
            "source": source_key,
            "create_date": parse_date(pub_date),
        }

def api_search(request):
    query = request.GET.get("q", "").strip()
    source = request.GET.get("source", "").strip()  # 可选：按来源过滤
    try:
        page = max(int(request.GET.get("p", "1")), 1)
    except ValueError:
        page = 1
    sort_param = request.GET.get("sort", "relevance")   # relevance/time/hot
    days = request.GET.get("days", "").strip()          # 近 N 天过滤（7/30/90）
    if not query:
        return JsonResponse({"total": 0, "page": 1, "page_nums": 0, "results": []})
    _record_search_keyword(query)

    # ---- ES 库内检索 ----
    # 活跃维度的筛选（来源/时间）放 post_filter：分面聚合统计不受自身筛选影响，
    # 每个分面的计数始终是"其他筛选条件下"的分布，供用户直接切换
    post_filters = []
    if source:
        post_filters.append({"term": {"source": source}})
    if days in _DAY_VALUES:
        post_filters.append({"range": {"create_date": {"gte": "now-{}d".format(days)}}})
    sort = _SORTS.get(sort_param)

    def _run_search(match_text, page_num, fuzzy=False):
        mm = {"query": match_text,
              "fields": ["title^3", "tags^2", "author", "content"],
              # 超过 2 个词元的查询要求 80% 词元命中，防止低质命中刷屏
              "minimum_should_match": "2<80%"}
        if fuzzy:
            # 模糊重搜：编辑距离容错（如"猪申克"→"肖申克"），
            # 并放宽词元命中率让部分错词不拖垮整个查询
            mm["fuzziness"] = "AUTO"
            mm["minimum_should_match"] = "2<50%"
        body = {
            "index": INDEX,
            "query": {"bool": {
                "must": {"multi_match": mm},
                "must_not": [{"terms": {"source": list(SEARCH_EXCLUDED_SOURCES)}}],
            }},
            "from_": (page_num - 1) * PAGE_SIZE,
            "size": PAGE_SIZE,
            "highlight": _HIGHLIGHT,
            "aggs": _FACET_AGGS,
        }
        if post_filters:
            body["post_filter"] = {"bool": {"must": post_filters}}
        if sort:
            body["sort"] = sort
        return client.search(**body)

_HIGHLIGHT = {"pre_tags": ['<span class="kw">'], "post_tags": ["</span>"],
              "fields": {"title": {}, "content": {}}}

    def _extract(resp):
        results = []
        for hit in resp["hits"]["hits"]:
            src = hit["_source"]
            hl = hit.get("highlight") or {}
            results.append({
                # ES 文档 _id：详情页按稳定 id 取文，不再用"标题再搜一遍取第一条"
                "id": hit.get("_id", ""),
                "title": "".join(hl["title"]) if hl.get("title") else src.get("title", ""),
                "content": "".join(hl["content"]) if hl.get("content") else (src.get("content") or "")[:200],
                "url": src.get("url", ""),
                "author": src.get("author", ""),
                "source": src.get("source", ""),
                "rating": src.get("rating"),
                "rank": src.get("rank"),
                "front_image_url": src.get("front_image_url", ""),
                "create_date": _display_date(src.get("create_date")),
                "praise_nums": src.get("praise_nums"),
                "view_nums": src.get("view_nums"),
                "reply_nums": src.get("reply_nums"),
                "danmaku_nums": src.get("danmaku_nums"),
            })
        aggs = resp.get("aggregations") or {}
        facets = {
            "sources": [{"key": b["key"], "count": b["doc_count"]}
                        for b in (aggs.get("facet_sources") or {}).get("buckets", [])],
            "days": [{"key": b["key"], "count": b["doc_count"]}
                     for b in (aggs.get("facet_days") or {}).get("buckets", [])],
        }
        return resp["hits"]["total"]["value"], results, facets

    try:
        resp = _run_search(query, page)
    except Exception as e:
        return _es_error(e)
    total, results, facets = _extract(resp)

    # 容错纠错（两级）：0 结果时先用短语建议器找相近词重搜；
    # 仍无结果则用编辑距离模糊匹配重搜（fuzziness AUTO）
    did_you_mean = None
    corrected = None
    fuzzy_match = False
    if total == 0 and page == 1:
        did_you_mean = _did_you_mean(query)
        if did_you_mean and did_you_mean != query:
            try:
                total2, results2, facets2 = _extract(_run_search(did_you_mean, 1))
                if results2:
                    corrected, total, results, facets = did_you_mean, total2, results2, facets2
            except Exception:
                pass
        if not results:
            try:
                total2, results2, facets2 = _extract(_run_search(query, 1, fuzzy=True))
                if results2:
                    fuzzy_match, total, results, facets = True, total2, results2, facets2
            except Exception:
                pass

    # 分页 bug 修复：ES 的 total 偶尔会高于实际命中数（估算偏差），
    # 导致出现空页。如果当前页为空且不是第 1 页，自动回退到最后一页。
    if not results and page > 1:
        for fallback_page in range(page - 1, 0, -1):
            try:
                resp2 = _run_search(query, fallback_page)
            except Exception:
                continue
            hits2 = resp2["hits"]["hits"]
            if hits2:
                _, results, _ = _extract(resp2)
                page = fallback_page
                total = (page - 1) * PAGE_SIZE + len(results)
                break
    page_nums = math.ceil(total / PAGE_SIZE) if total else 0
    if page > 1 and len(results) < PAGE_SIZE:
        page_nums = page

    def start(self, spider="douyin_hot", pages=2, js=False, resume_job=None):
        """启动爬虫任务。

        :param resume_job: 要恢复的任务 ID（JOBDIR 名称），None 表示新任务。
                           恢复时使用同一个 JOBDIR，Scrapy 会跳过已爬取的 URL。
        """
        spec = SPIDERS.get(spider)
        if spec is None:
            return {"started": False, "reason": "未知爬虫: {}".format(spider)}
        with self._lock:
            # 上一轮可能已经结束但还没被收（盯梢线程尚未跑完 / 后端重启），
            # 先收割一次，避免旧任务永久停在 running 且新任务把它的日志算进来
            self._check_and_update_current()
            if self._proc is not None and self._proc.poll() is None:
                return {"started": False, "reason": "已有采集任务在运行",
                        "pid": self._proc.pid, "started_at": self._started_at}
            if not os.path.isdir(SPIDER_DIR):
                return {"started": False, "reason": "找不到 XSearch 项目目录"}

            # JOBDIR：持久化爬虫状态（已爬取 URL、请求队列），支持中断恢复
            if resume_job:
                # 只接受 jobs/ 的直接子目录名：拒绝绝对路径与 "../" 穿越，
                # 否则等于让调用方把 JOBDIR（爬虫可写状态的地方）到任意目录
                name = resume_job.strip().rstrip("/")
                if (name != os.path.basename(name) or name in (".", "..")
                        or not JOB_NAME_RE.fullmatch(name)):
                    logger.warning("拒绝非法 resume_job 参数: %r", resume_job)
                    return {"started": False, "reason": "非法的任务名称"}
                job_dir = os.path.join(JOBS_DIR, name)
                if not os.path.isdir(job_dir):
                    return {"started": False, "reason": "找不到要恢复的任务: {}".format(name)}
            else:
                job_dir = os.path.join(JOBS_DIR, "{}_{}".format(spider, int(time.time())))

            cmd = [sys.executable, "-m", "scrapy", "crawl", spec["scrapy_name"]]
            cmd += spec.get("extra_args", [])
            # 添加 JOBDIR 支持中断恢复
            cmd += ["-s", "JOBDIR={}".format(job_dir)]
            # 打开日志文件传给子进程，子进程继承 fd 后父进程立即关闭自己的句柄，
            # 避免依赖"下一次 start 时关闭"导致的句柄泄漏
            log_file = open(self._log_path, "w")
            self._proc = subprocess.Popen(cmd, cwd=SPIDER_DIR,
                                          stdout=log_file, stderr=subprocess.STDOUT)
            log_file.close()
            self._started_at = datetime.now().isoformat(timespec="seconds")
            self._current_job_id = "job_{}".format(int(time.time() * 1000))
            self._record_history(self._current_job_id, spider, "running",
                                 pid=self._proc.pid, log_path=self._log_path,
                                 job_dir=os.path.basename(job_dir),
                                 resumed=bool(resume_job))
            # 盯梢线程：进程退出就立刻回写终态并按需告警。
            # 放在 _current_job_id 赋值之后启动，否则 expected_job_id 对不上会被自己忽略。
            threading.Thread(target=self._watch_exit,
                             args=(self._proc, self._current_job_id),
                             daemon=True, name="xsearch-crawl-watch").start()
            return {"started": True, "pid": self._proc.pid,
                    "started_at": self._started_at, "spider": spider, "label": spec["label"],
                    "job_dir": os.path.basename(job_dir), "resumed": bool(resume_job)}