pagesift_api/jina/
reader_route.rs

1//! Jina Reader protocol handler (`GET /r/{*url}` and `POST /r`).
2//! Provides direct Markdown extraction and JSON responses for AI Agents.
3
4use axum::Json;
5use axum::extract::{Extension, Path, State};
6use axum::http::{HeaderMap, StatusCode, header};
7use axum::response::{IntoResponse, Response};
8use serde::Deserialize;
9use std::sync::atomic::Ordering;
10
11use super::stream_sse::{error_event, split_paragraphs, sse_parts, sse_response};
12use crate::limits::{Caller, PageHold};
13use crate::metrics::PageSource;
14use crate::server::AppState;
15use crate::types::ScrapeOptions;
16
17#[derive(Debug, Deserialize)]
18pub struct JinaPostRequest {
19    pub url: String,
20}
21
22/// Handler for `GET /r/{*target_url}`.
23pub(crate) async fn handle_jina_get(
24    State(s): State<AppState>,
25    caller: Option<Extension<Caller>>,
26    headers: HeaderMap,
27    Path(raw_url): Path<String>,
28) -> Response {
29    let clean_url = parse_target_url(&raw_url);
30    process_jina_scrape(s, caller, headers, &clean_url).await
31}
32
33/// Handler for `POST /r`.
34pub(crate) async fn handle_jina_post(
35    State(s): State<AppState>,
36    caller: Option<Extension<Caller>>,
37    headers: HeaderMap,
38    Json(payload): Json<JinaPostRequest>,
39) -> Response {
40    let clean_url = parse_target_url(&payload.url);
41    process_jina_scrape(s, caller, headers, &clean_url).await
42}
43
44fn parse_target_url(raw: &str) -> String {
45    let trimmed = raw.trim();
46    if trimmed.starts_with("http://") || trimmed.starts_with("https://") {
47        trimmed.to_string()
48    } else if let Some(idx) = trimmed.find("http:/") {
49        if trimmed[idx..].starts_with("http://") {
50            trimmed[idx..].to_string()
51        } else {
52            format!("http://{}", trimmed[idx + 6..].trim_start_matches('/'))
53        }
54    } else if let Some(idx) = trimmed.find("https:/") {
55        if trimmed[idx..].starts_with("https://") {
56            trimmed[idx..].to_string()
57        } else {
58            format!("https://{}", trimmed[idx + 7..].trim_start_matches('/'))
59        }
60    } else {
61        format!("https://{}", trimmed.trim_start_matches('/'))
62    }
63}
64
65async fn process_jina_scrape(
66    s: AppState,
67    caller: Option<Extension<Caller>>,
68    headers: HeaderMap,
69    target_url: &str,
70) -> Response {
71    if let Err(e) = crate::validate::check_url(target_url) {
72        return e.into_response();
73    }
74    let (mut opts, params) = match super::headers::parse(&headers) {
75        Ok(v) => v,
76        Err(e) => return e.into_response(),
77    };
78    opts.owner = caller
79        .as_ref()
80        .map(|Extension(c)| c.account.to_string())
81        .filter(|owner| owner != "anonymous");
82    if let Err(e) = crate::validate::check_scrape_options(&opts, s.engine.as_ref()) {
83        return e.into_response();
84    }
85    // 抓取之前先占住这一页的配额；失败、取消或提前断开时由 PageHold 退还
86    let page_hold = match &caller {
87        Some(Extension(c)) => match s.limiter.reserve_page(c).await {
88            Ok(h) => Some(h),
89            Err(e) => return e.into_response(),
90        },
91        None => None,
92    };
93    let accept = headers
94        .get(header::ACCEPT)
95        .and_then(|v| v.to_str().ok())
96        .unwrap_or("");
97    if accept.contains("text/event-stream") {
98        return stream_scrape(s, target_url.to_string(), opts, params, page_hold);
99    }
100
101    match s.engine.scrape(target_url, &opts).await {
102        Ok(doc) => {
103            record_ok(&s, page_hold).await;
104            let r = match super::respond::render(&doc, target_url, &params) {
105                Ok(r) => r,
106                Err(e) => return e.into_response(),
107            };
108            let mut warnings = doc
109                .metadata
110                .pagesift
111                .as_ref()
112                .map(|m| m.warnings.clone())
113                .unwrap_or_default();
114            warnings.extend(r.warnings.iter().cloned());
115            if accept.contains("application/json") {
116                let mut data = serde_json::json!({
117                    "title": doc.metadata.title.as_deref().unwrap_or(""),
118                    "description": doc.metadata.description.clone().unwrap_or_default(),
119                    "url": target_url,
120                    "content": r.body,
121                });
122                if let Some(chunks) = &r.chunks {
123                    data["chunks"] = serde_json::json!(chunks);
124                }
125                if !warnings.is_empty() {
126                    data["warnings"] = serde_json::json!(warnings);
127                }
128                // [协议] Jina Reader JSON 响应的固定外壳：code 200、status 20000
129                Json(serde_json::json!({ "code": 200, "status": 20000, "data": data }))
130                    .into_response()
131            } else {
132                let body = r.text;
133                let mut resp = Response::builder().header(header::CONTENT_TYPE, r.content_type);
134                // 纯文本响应没有放警告的字段：写进响应头（`x-pagesift-warnings`，多条用 `; ` 隔开）
135                if let Some(v) = (!warnings.is_empty())
136                    .then(|| warnings.join("; "))
137                    .and_then(|w| axum::http::HeaderValue::from_str(&header_safe(&w)).ok())
138                {
139                    resp = resp.header("x-pagesift-warnings", v);
140                }
141                resp.body(axum::body::Body::from(body)).unwrap_or_else(|_| {
142                    (StatusCode::INTERNAL_SERVER_ERROR, "response builder error").into_response()
143                })
144            }
145        }
146        Err(e) => {
147            s.metrics.scrape_err.fetch_add(1, Ordering::Relaxed);
148            e.into_response()
149        }
150    }
151}
152
153async fn record_ok(s: &AppState, hold: Option<PageHold>) {
154    s.metrics.scrape_ok.fetch_add(1, Ordering::Relaxed);
155    s.metrics.page(PageSource::Scrape);
156    if let Some(hold) = hold {
157        hold.commit().await;
158    }
159}
160
161/// SSE：响应头立刻发出，抓取在流里进行；整页抓完、提取完之后才逐个写出 progress、各段 content 和 done
162/// （按段分帧写出，不拼成整块），抓取期间按心跳间隔写注释保活。不边提取边推：正文识别要看完整个 DOM
163/// 才知道哪些块是正文（打分、去模板），提取本身只占请求耗时的一小部分（0.91 发布说明里有本机实测），
164/// 第一个 `content` 的时刻等于整页提取完成的时刻（0.90 的发布说明写成了"一产出就写"，不准确）。
165/// 客户端中途断开就停止抓取。响应头发出后才失败的，用 `error` 事件报告。
166fn stream_scrape(
167    s: AppState,
168    url: String,
169    opts: ScrapeOptions,
170    params: super::headers::JinaParams,
171    hold: Option<PageHold>,
172) -> Response {
173    stream_scrape_with(
174        s,
175        url,
176        opts,
177        params,
178        hold,
179        crate::firecrawl::stream::KEEPALIVE,
180    )
181}
182
183/// 同 [`stream_scrape`]，`keepalive` 是抓取期间写心跳注释（`: keepalive`）的间隔：渲染、慢站点
184/// 可能几十秒才有结果，期间不写字节，代理和客户端会按空闲超时断开（0.90 如此）。SSE 客户端忽略注释行，
185/// 事件本身的字节不变。
186fn stream_scrape_with(
187    s: AppState,
188    url: String,
189    opts: ScrapeOptions,
190    params: super::headers::JinaParams,
191    hold: Option<PageHold>,
192    keepalive: std::time::Duration,
193) -> Response {
194    let (tx, body) = crate::stream_body::channel();
195    tokio::spawn(async move {
196        let scrape = s.engine.scrape(&url, &opts);
197        tokio::pin!(scrape);
198        let mut tick = tokio::time::interval_at(tokio::time::Instant::now() + keepalive, keepalive);
199        let r = loop {
200            tokio::select! {
201                r = &mut scrape => break r,
202                () = tx.closed() => return,
203                _ = tick.tick() => {
204                    if tx.send(axum::body::Bytes::from_static(b": keepalive\n\n")).await.is_err() {
205                        return;
206                    }
207                }
208            }
209        };
210        match r {
211            Ok(doc) => {
212                record_ok(&s, hold).await;
213                // 和普通响应走同一份渲染：respond-with、保留方式、上限、分块都生效
214                let r = match super::respond::render(&doc, &url, &params) {
215                    Ok(r) => r,
216                    Err(e) => {
217                        let _ = tx.send(error_event(&e).into()).await;
218                        return;
219                    }
220                };
221                let mut warnings = doc
222                    .metadata
223                    .pagesift
224                    .as_ref()
225                    .map(|m| m.warnings.clone())
226                    .unwrap_or_default();
227                warnings.extend(r.warnings.iter().cloned());
228                let parts: Vec<String> = match r.chunks {
229                    Some(c) => c,
230                    None => split_paragraphs(&r.body),
231                };
232                let events = sse_parts(&url, doc.metadata.title.as_deref(), &parts, &warnings);
233                for event in events {
234                    if tx.send(event.into()).await.is_err() {
235                        return;
236                    }
237                }
238            }
239            Err(e) => {
240                s.metrics.scrape_err.fetch_add(1, Ordering::Relaxed);
241                let _ = tx.send(error_event(&e).into()).await;
242            }
243        }
244    });
245    sse_response(body)
246}
247
248/// 响应头只能放可见 ASCII：其余字节（比如警告里的中文文件名）写成百分号编码，不让整个头丢掉。
249fn header_safe(s: &str) -> String {
250    let mut out = String::with_capacity(s.len());
251    for b in s.bytes() {
252        if (b' '..=b'~').contains(&b) && b != b'%' {
253            out.push(b as char);
254        } else {
255            out.push_str(&format!("%{b:02X}"));
256        }
257    }
258    out
259}
pagesift_api/jina/reader_route.rs

pagesift_api/jina/
reader_route.rs
