feat(adapters): RSS and PT site adapters ported from MoviePilot v3 + mocks (W1-3b)
- rss: port of MP RssHelper._parse_impl/get_rss_link semantics with safe XML (entities off, DTD/ENTITY rejected, size/item caps), ETag/Last-Modified, backoff + Retry-After, circuit breaker with manual reset, passkey redaction, overlap-window GUID+digest dedupe - sites/spider: TS port of MP SiteSpider (indexer schema, search URL builder, field DSL, filters, Jinja templates via nunjucks, login-page detection); audiences definition hand-written as JSON; raw category id preserved; shared per-site token bucket - sites/mteam: port of MP MTorrentSpider (params, result mapping, promo factors, genDlToken flow) without embedding the API key in enclosures; business code/message checks - testkit: mock RSS server, mock NexusPHP site, mock M-Team API - docs: W1-3b migration mapping appended to porting ledger Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
rootandClaude Fable 5.1 committed
1 parent
ef2e504a94
commit
a6d049db81
47 files changed
+5414
No files matched your search
@@ -0,0 +1,229 @@
|
||||
/**
|
||||
* RSS 抓取。迁移自 MP `app/application/rss.py RssHelper.parse/_parse_impl`(GET + UA + 自定义头,非 200 视为失败,
|
||||
* 大小上限后再解析);条件请求、退避、Retry-After、熔断、脱敏为任务书新增要求(MP 无)。
|
||||
*/
|
||||
import { parseFeed } from "./parse";
|
||||
import { redactText, redactUrl } from "./redact";
|
||||
import { FeedError, type FeedFetchResult, type FeedFetchState } from "./types";
|
||||
|
||||
export interface RssFetcherOptions {
|
||||
fetch?: typeof globalThis.fetch;
|
||||
now?: () => Date;
|
||||
/** 单次请求超时,默认 30s(MP `timeout or 30`) */
|
||||
timeoutMs?: number;
|
||||
/** 解压后响应体上限(字节),默认 50 MiB(MP MAX_RSS_SIZE);同时约束 gzip/br 解压量 */
|
||||
maxBytes?: number;
|
||||
/** 条目上限,默认 1000(MP MAX_RSS_ITEMS) */
|
||||
maxItems?: number;
|
||||
userAgent?: string;
|
||||
/** 指数退避:base * 2^(n-1),上限 max。默认 60s / 6h */
|
||||
backoff?: { baseMs: number; maxMs: number };
|
||||
/** 连续失败达到该次数进入熔断(open),默认 5 */
|
||||
circuitThreshold?: number;
|
||||
/** 熔断冷却后允许一次半开探测;null(默认)= 只能手动 resetFeedCircuit 恢复 */
|
||||
circuitCooldownMs?: number | null;
|
||||
}
|
||||
|
||||
export interface FetchFeedOptions {
|
||||
signal?: AbortSignal;
|
||||
/** 手动刷新:忽略退避窗口(不忽略熔断) */
|
||||
ignoreBackoff?: boolean;
|
||||
}
|
||||
|
||||
export function initialFeedState(): FeedFetchState {
|
||||
return { consecutiveFailures: 0, circuit: "closed" };
|
||||
}
|
||||
|
||||
/** 手动恢复:关闭熔断并清空退避,保留 ETag/Last-Modified。 */
|
||||
export function resetFeedCircuit(state: FeedFetchState): FeedFetchState {
|
||||
const { nextAttemptAt: _n, circuitOpenedAt: _c, ...rest } = state;
|
||||
return { ...rest, consecutiveFailures: 0, circuit: "closed" };
|
||||
}
|
||||
|
||||
export function computeBackoffMs(failures: number, baseMs: number, maxMs: number): number {
|
||||
if (failures <= 0) return 0;
|
||||
return Math.min(maxMs, baseMs * 2 ** Math.min(failures - 1, 30));
|
||||
}
|
||||
|
||||
/** Retry-After:秒数或 HTTP-date;返回毫秒,无效返回 undefined。 */
|
||||
export function parseRetryAfter(value: string | null | undefined, now: Date): number | undefined {
|
||||
if (!value) return undefined;
|
||||
const v = value.trim();
|
||||
if (/^\d+$/.test(v)) return Number(v) * 1000;
|
||||
const t = Date.parse(v);
|
||||
if (Number.isNaN(t)) return undefined;
|
||||
return Math.max(0, t - now.getTime());
|
||||
}
|
||||
|
||||
async function readCapped(res: Response, maxBytes: number): Promise<Uint8Array> {
|
||||
const len = Number(res.headers.get("content-length") ?? "");
|
||||
// content-length 是压缩后长度,只做早拒;真正的上限在流式读取里按解压后字节执行
|
||||
if (Number.isFinite(len) && len > maxBytes) {
|
||||
await res.body?.cancel().catch(() => {});
|
||||
throw new FeedError("too_large", `响应声明长度 ${len} 超过上限 ${maxBytes}`);
|
||||
}
|
||||
if (!res.body) return new Uint8Array();
|
||||
const reader = res.body.getReader();
|
||||
const chunks: Uint8Array[] = [];
|
||||
let total = 0;
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
total += value.byteLength;
|
||||
if (total > maxBytes) {
|
||||
await reader.cancel().catch(() => {});
|
||||
throw new FeedError("too_large", `响应体(解压后)超过上限 ${maxBytes} 字节`);
|
||||
}
|
||||
chunks.push(value);
|
||||
}
|
||||
const out = new Uint8Array(total);
|
||||
let off = 0;
|
||||
for (const c of chunks) {
|
||||
out.set(c, off);
|
||||
off += c.byteLength;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function decodeBody(bytes: Uint8Array, contentType: string | null): string {
|
||||
const head = new TextDecoder("latin1").decode(bytes.subarray(0, 256));
|
||||
const label =
|
||||
/charset=([\w-]+)/i.exec(contentType ?? "")?.[1] ??
|
||||
/<\?xml[^>]*encoding=["']([\w-]+)["']/i.exec(head)?.[1] ??
|
||||
"utf-8";
|
||||
try {
|
||||
return new TextDecoder(label).decode(bytes);
|
||||
} catch {
|
||||
return new TextDecoder("utf-8").decode(bytes);
|
||||
}
|
||||
}
|
||||
|
||||
export interface RssFetcher {
|
||||
fetchFeed(url: string, state?: FeedFetchState, opts?: FetchFeedOptions): Promise<FeedFetchResult>;
|
||||
}
|
||||
|
||||
export function createRssFetcher(options: RssFetcherOptions = {}): RssFetcher {
|
||||
const doFetch = options.fetch ?? globalThis.fetch;
|
||||
const now = options.now ?? (() => new Date());
|
||||
const timeoutMs = options.timeoutMs ?? 30_000;
|
||||
const maxBytes = options.maxBytes ?? 50 * 1024 * 1024;
|
||||
const backoff = options.backoff ?? { baseMs: 60_000, maxMs: 6 * 3600_000 };
|
||||
const threshold = options.circuitThreshold ?? 5;
|
||||
const cooldown = options.circuitCooldownMs ?? null;
|
||||
|
||||
function fail(state: FeedFetchState, error: FeedError): FeedFetchResult {
|
||||
const at = now();
|
||||
const failures = state.consecutiveFailures + 1;
|
||||
const wait = Math.max(computeBackoffMs(failures, backoff.baseMs, backoff.maxMs), error.retryAfterMs ?? 0);
|
||||
const next: FeedFetchState = {
|
||||
...state,
|
||||
consecutiveFailures: failures,
|
||||
nextAttemptAt: new Date(at.getTime() + wait).toISOString(),
|
||||
lastError: {
|
||||
kind: error.kind,
|
||||
message: error.message,
|
||||
at: at.toISOString(),
|
||||
...(error.httpStatus !== undefined ? { httpStatus: error.httpStatus } : {}),
|
||||
},
|
||||
};
|
||||
if (failures >= threshold) {
|
||||
next.circuit = "open";
|
||||
next.circuitOpenedAt = at.toISOString();
|
||||
}
|
||||
return { kind: "error", error, state: next };
|
||||
}
|
||||
|
||||
async function fetchFeed(
|
||||
url: string,
|
||||
state: FeedFetchState = initialFeedState(),
|
||||
opts: FetchFeedOptions = {},
|
||||
): Promise<FeedFetchResult> {
|
||||
const at = now();
|
||||
if (state.circuit === "open") {
|
||||
const openedAt = state.circuitOpenedAt ? Date.parse(state.circuitOpenedAt) : 0;
|
||||
const halfOpen = cooldown !== null && at.getTime() - openedAt >= cooldown;
|
||||
if (!halfOpen) {
|
||||
return {
|
||||
kind: "skipped",
|
||||
reason: "circuit_open",
|
||||
...(cooldown !== null ? { retryAt: new Date(openedAt + cooldown).toISOString() } : {}),
|
||||
state,
|
||||
};
|
||||
}
|
||||
} else if (!opts.ignoreBackoff && state.nextAttemptAt && Date.parse(state.nextAttemptAt) > at.getTime()) {
|
||||
return { kind: "skipped", reason: "backoff", retryAt: state.nextAttemptAt, state };
|
||||
}
|
||||
|
||||
const safeUrl = redactUrl(url);
|
||||
const headers: Record<string, string> = {
|
||||
Accept: "application/rss+xml, application/atom+xml, application/xml;q=0.9, text/xml;q=0.8, */*;q=0.1",
|
||||
};
|
||||
if (options.userAgent) headers["User-Agent"] = options.userAgent;
|
||||
if (state.etag) headers["If-None-Match"] = state.etag;
|
||||
if (state.lastModified) headers["If-Modified-Since"] = state.lastModified;
|
||||
|
||||
const timeout = AbortSignal.timeout(timeoutMs);
|
||||
const signal = opts.signal ? AbortSignal.any([opts.signal, timeout]) : timeout;
|
||||
let res: Response;
|
||||
try {
|
||||
res = await doFetch(url, { headers, signal, redirect: "follow" });
|
||||
} catch (e) {
|
||||
if (opts.signal?.aborted) throw e;
|
||||
const err = e as Error;
|
||||
const kind = timeout.aborted ? "timeout" : "network";
|
||||
return fail(state, new FeedError(kind, redactText(`请求 ${safeUrl} 失败:${err.message}`)));
|
||||
}
|
||||
|
||||
if (res.status === 304) {
|
||||
await res.body?.cancel().catch(() => {});
|
||||
const next: FeedFetchState = {
|
||||
...state,
|
||||
consecutiveFailures: 0,
|
||||
circuit: "closed",
|
||||
lastSuccessAt: at.toISOString(),
|
||||
};
|
||||
delete next.nextAttemptAt;
|
||||
delete next.circuitOpenedAt;
|
||||
return { kind: "not_modified", state: next };
|
||||
}
|
||||
if (!res.ok) {
|
||||
await res.body?.cancel().catch(() => {});
|
||||
const retryAfterMs = parseRetryAfter(res.headers.get("retry-after"), at);
|
||||
const kind =
|
||||
res.status === 429 || (res.status === 503 && retryAfterMs !== undefined) ? "rate_limited" : "http_error";
|
||||
return fail(
|
||||
state,
|
||||
new FeedError(kind, `请求 ${safeUrl} 返回 HTTP ${res.status}`, {
|
||||
httpStatus: res.status,
|
||||
...(retryAfterMs !== undefined ? { retryAfterMs } : {}),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
const bytes = await readCapped(res, maxBytes);
|
||||
const text = decodeBody(bytes, res.headers.get("content-type"));
|
||||
const feed = parseFeed(text, { maxItems: options.maxItems ?? 1000, maxChars: maxBytes, fetchedAt: at });
|
||||
const next: FeedFetchState = {
|
||||
...state,
|
||||
consecutiveFailures: 0,
|
||||
circuit: "closed",
|
||||
lastSuccessAt: at.toISOString(),
|
||||
};
|
||||
delete next.nextAttemptAt;
|
||||
delete next.circuitOpenedAt;
|
||||
const etag = res.headers.get("etag");
|
||||
const lastModified = res.headers.get("last-modified");
|
||||
if (etag) next.etag = etag;
|
||||
if (lastModified) next.lastModified = lastModified;
|
||||
return { kind: "ok", feed, state: next, httpStatus: res.status };
|
||||
} catch (e) {
|
||||
if (opts.signal?.aborted) throw e;
|
||||
if (e instanceof FeedError) return fail(state, new FeedError(e.kind, redactText(e.message)));
|
||||
const kind = timeout.aborted ? "timeout" : "network";
|
||||
return fail(state, new FeedError(kind, redactText(`读取 ${safeUrl} 失败:${(e as Error).message}`)));
|
||||
}
|
||||
}
|
||||
|
||||
return { fetchFeed };
|
||||
}
|
||||
Reference in new issue
Block a user