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;
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    if let Some(Extension(c)) = &caller
75        && let Err(e) = s.limiter.check_quota(c)
76    {
77        return e.into_response();
78    }
79
80    let (mut opts, params) = match super::headers::parse(&headers) {
81        Ok(v) => v,
82        Err(e) => return e.into_response(),
83    };
84    opts.owner = caller
85        .as_ref()
86        .map(|Extension(c)| c.account.to_string())
87        .filter(|owner| owner != "anonymous");
88    if let Err(e) = crate::validate::check_scrape_options(&opts, s.engine.as_ref()) {
89        return e.into_response();
90    }
91    let accept = headers
92        .get(header::ACCEPT)
93        .and_then(|v| v.to_str().ok())
94        .unwrap_or("");
95    if accept.contains("text/event-stream") {
96        return stream_scrape(
97            s,
98            caller.map(|Extension(c)| c),
99            target_url.to_string(),
100            opts,
101            params,
102        );
103    }
104
105    match s.engine.scrape(target_url, &opts).await {
106        Ok(doc) => {
107            record_ok(&s, caller.as_ref().map(|Extension(c)| c)).await;
108            let r = match super::respond::render(&doc, target_url, &params) {
109                Ok(r) => r,
110                Err(e) => return e.into_response(),
111            };
112            let mut warnings = doc
113                .metadata
114                .pagesift
115                .as_ref()
116                .map(|m| m.warnings.clone())
117                .unwrap_or_default();
118            warnings.extend(r.warnings.iter().cloned());
119            if accept.contains("application/json") {
120                let mut data = serde_json::json!({
121                    "title": doc.metadata.title.as_deref().unwrap_or(""),
122                    "description": doc.metadata.description.clone().unwrap_or_default(),
123                    "url": target_url,
124                    "content": r.body,
125                });
126                if let Some(chunks) = &r.chunks {
127                    data["chunks"] = serde_json::json!(chunks);
128                }
129                if !warnings.is_empty() {
130                    data["warnings"] = serde_json::json!(warnings);
131                }
132                // [协议] Jina Reader JSON 响应的固定外壳：code 200、status 20000
133                Json(serde_json::json!({ "code": 200, "status": 20000, "data": data }))
134                    .into_response()
135            } else {
136                let body = r.text;
137                let mut resp = Response::builder().header(header::CONTENT_TYPE, r.content_type);
138                // 纯文本响应没有放警告的字段：写进响应头（`x-pagesift-warnings`，多条用 `; ` 隔开）
139                if let Some(v) = (!warnings.is_empty())
140                    .then(|| warnings.join("; "))
141                    .and_then(|w| axum::http::HeaderValue::from_str(&header_safe(&w)).ok())
142                {
143                    resp = resp.header("x-pagesift-warnings", v);
144                }
145                resp.body(axum::body::Body::from(body)).unwrap_or_else(|_| {
146                    (StatusCode::INTERNAL_SERVER_ERROR, "response builder error").into_response()
147                })
148            }
149        }
150        Err(e) => {
151            s.metrics.scrape_err.fetch_add(1, Ordering::Relaxed);
152            e.into_response()
153        }
154    }
155}
156
157async fn record_ok(s: &AppState, caller: Option<&Caller>) {
158    s.metrics.scrape_ok.fetch_add(1, Ordering::Relaxed);
159    s.metrics.page(PageSource::Scrape);
160    if let Some(c) = caller {
161        s.limiter.consume_pages(c, 1).await;
162    }
163}
164
165/// SSE：响应头立刻发出，抓取在流里进行；整页抓完、提取完之后才逐个写出 progress、各段 content 和 done
166/// （按段分帧写出，不拼成整块），抓取期间按心跳间隔写注释保活。不边提取边推：正文识别要看完整个 DOM
167/// 才知道哪些块是正文（打分、去模板），提取本身只占请求耗时的一小部分（0.91 发布说明里有本机实测），
168/// 第一个 `content` 的时刻等于整页提取完成的时刻（0.90 的发布说明写成了"一产出就写"，不准确）。
169/// 客户端中途断开就停止抓取。响应头发出后才失败的，用 `error` 事件报告。
170fn stream_scrape(
171    s: AppState,
172    caller: Option<Caller>,
173    url: String,
174    opts: ScrapeOptions,
175    params: super::headers::JinaParams,
176) -> Response {
177    stream_scrape_with(
178        s,
179        caller,
180        url,
181        opts,
182        params,
183        crate::firecrawl::stream::KEEPALIVE,
184    )
185}
186
187/// 同 [`stream_scrape`]，`keepalive` 是抓取期间写心跳注释（`: keepalive`）的间隔：渲染、慢站点
188/// 可能几十秒才有结果，期间不写字节，代理和客户端会按空闲超时断开（0.90 如此）。SSE 客户端忽略注释行，
189/// 事件本身的字节不变。
190fn stream_scrape_with(
191    s: AppState,
192    caller: Option<Caller>,
193    url: String,
194    opts: ScrapeOptions,
195    params: super::headers::JinaParams,
196    keepalive: std::time::Duration,
197) -> Response {
198    let (tx, body) = crate::stream_body::channel();
199    tokio::spawn(async move {
200        let scrape = s.engine.scrape(&url, &opts);
201        tokio::pin!(scrape);
202        let mut tick = tokio::time::interval_at(tokio::time::Instant::now() + keepalive, keepalive);
203        let r = loop {
204            tokio::select! {
205                r = &mut scrape => break r,
206                () = tx.closed() => return,
207                _ = tick.tick() => {
208                    if tx.send(axum::body::Bytes::from_static(b": keepalive\n\n")).await.is_err() {
209                        return;
210                    }
211                }
212            }
213        };
214        match r {
215            Ok(doc) => {
216                record_ok(&s, caller.as_ref()).await;
217                // 和普通响应走同一份渲染：respond-with、保留方式、上限、分块都生效
218                let r = match super::respond::render(&doc, &url, &params) {
219                    Ok(r) => r,
220                    Err(e) => {
221                        let _ = tx.send(error_event(&e).into()).await;
222                        return;
223                    }
224                };
225                let mut warnings = doc
226                    .metadata
227                    .pagesift
228                    .as_ref()
229                    .map(|m| m.warnings.clone())
230                    .unwrap_or_default();
231                warnings.extend(r.warnings.iter().cloned());
232                let parts: Vec<String> = match r.chunks {
233                    Some(c) => c,
234                    None => split_paragraphs(&r.body),
235                };
236                let events = sse_parts(&url, doc.metadata.title.as_deref(), &parts, &warnings);
237                for event in events {
238                    if tx.send(event.into()).await.is_err() {
239                        return;
240                    }
241                }
242            }
243            Err(e) => {
244                s.metrics.scrape_err.fetch_add(1, Ordering::Relaxed);
245                let _ = tx.send(error_event(&e).into()).await;
246            }
247        }
248    });
249    sse_response(body)
250}
251
252/// 响应头只能放可见 ASCII：其余字节（比如警告里的中文文件名）写成百分号编码，不让整个头丢掉。
253fn header_safe(s: &str) -> String {
254    let mut out = String::with_capacity(s.len());
255    for b in s.bytes() {
256        if (b' '..=b'~').contains(&b) && b != b'%' {
257            out.push(b as char);
258        } else {
259            out.push_str(&format!("%{b:02X}"));
260        }
261    }
262    out
263}
pagesift_api/jina/reader_route.rs

pagesift_api/jina/
reader_route.rs
