spb/wp-spboucher-ai
Public
1import { promises as fs } from "node:fs";2import { join, basename } from "node:path";3import { sql } from "@/db";4import { STORAGE_DIR } from "@/lib/env";5import { pdfInfo, pdfIntegrity, pdfText, pdfTextPages, pdfThumbnail, sha256File, storeFile, storagePath, ensureStorage } from "@/lib/pdf/process";6import { proposeMetadata, titleSimilarity } from "@/lib/pdf/extract";7import { chunkPages } from "@/lib/pdf/chunk";8import { lexvec } from "@/lib/ai/lexvec";9import { embed, embeddingsEnabled, EMBEDDING_MODEL } from "@/lib/ai/embeddings";10import { enqueue } from "./queue";11import { runBackup } from "./backup";12import { rebuildRelations } from "./relations";1314const vec = (v: number[]) => `[${v.map((x) => x.toFixed(6)).join(",")}]`;1516export async function processPdf(p: { fileId: string; paperId?: string; versionId?: string }) {17 const [f] = await sql`SELECT * FROM files WHERE id = ${p.fileId}`; if (!f) throw new Error("file not found");18 const path = storagePath(f.storage_key);19 const integ = await pdfIntegrity(path);20 const info = integ.info ?? (integ.ok ? await pdfInfo(path) : null);21 await sql`UPDATE files SET integrity_ok = ${integ.ok}, pages = ${info?.pages ?? null}, pdf_meta = ${info ? sql.json(info.raw as never) : null} WHERE id = ${p.fileId}`;22 if (!integ.ok) { await sql`UPDATE files SET text_status = 'failed' WHERE id = ${p.fileId}`; throw new Error(`integrity: ${integ.reason}`); }23 // Preview (first page) — content-addressed by sha.24 const previewBase = join(/*turbopackIgnore: true*/ STORAGE_DIR, "previews", f.sha256);25 try { await fs.access(previewBase + ".png"); } catch { await pdfThumbnail(path, previewBase, 60); }26 if (p.versionId) await sql`UPDATE paper_versions SET pages = coalesce(pages, ${info?.pages ?? null}), sha256 = ${f.sha256}, bytes = ${Number(f.bytes)} WHERE id = ${p.versionId}`;27 // Full text → chunks (only for the current version of a paper).28 if (p.paperId) {29 const pages = await pdfTextPages(path);30 const chunks = chunkPages(pages);31 await sql.begin(async (tx) => {32 await tx`DELETE FROM paper_chunks WHERE paper_id = ${p.paperId!} AND (version_id = ${p.versionId ?? null} OR version_id IS NULL)`;33 for (let i = 0; i < chunks.length; i += 200) {34 const batch = chunks.slice(i, i + 200);35 await tx`INSERT INTO paper_chunks ${tx(batch.map((c) => ({ paper_id: p.paperId!, version_id: p.versionId ?? null, chunk_index: c.index, page_from: c.pageFrom, page_to: c.pageTo, text: c.text })))}`;36 }37 });38 const [paper] = await sql`SELECT title, subtitle, abstract, keywords FROM papers WHERE id = ${p.paperId}`;39 const base = `${paper.title} ${paper.subtitle ?? ""} ${(paper.keywords ?? []).join(" ")} ${(paper.abstract as string[]).join(" ")}`;40 const lv = lexvec(base + " " + chunks.map((c) => c.text).join(" ").slice(0, 60000));41 await sql`UPDATE papers SET lexvec = ${vec(lv)}::vector WHERE id = ${p.paperId}`;42 await sql`UPDATE files SET text_status = 'done' WHERE id = ${p.fileId}`;43 if (embeddingsEnabled()) await enqueue("embed_paper", { paperId: p.paperId }, { dedupe: true });44 await enqueue("relations", {}, { dedupe: true, runAfter: new Date(Date.now() + 30_000) });45 } else await sql`UPDATE files SET text_status = 'skipped' WHERE id = ${p.fileId}`;46}4748/** Paper-level lexvec from metadata only (no manuscript yet). */49export async function lexvecFromMetadata(paperId: string) {50 const [paper] = await sql`SELECT title, subtitle, abstract, keywords, research_question, contributions, methodology, data_sections FROM papers WHERE id = ${paperId}`;51 if (!paper) return;52 const text = [paper.title, paper.subtitle, (paper.keywords ?? []).join(" "), ...(paper.abstract as string[]), paper.research_question, JSON.stringify(paper.contributions), JSON.stringify(paper.methodology), JSON.stringify(paper.data_sections)].filter(Boolean).join(" ");53 await sql`UPDATE papers SET lexvec = ${vec(lexvec(text))}::vector WHERE id = ${paperId}`;54}5556export async function embedPaper(p: { paperId: string }) {57 const [paper] = await sql`SELECT id, title, subtitle, abstract, keywords FROM papers WHERE id = ${p.paperId}`; if (!paper) return;58 const head = `${paper.title}${paper.subtitle ? ": " + paper.subtitle : ""}\n\n${(paper.abstract as string[]).join("\n\n")}\n\nKeywords: ${(paper.keywords ?? []).join(", ")}`;59 const e = await embed([head]); if (!e) return;60 await sql`UPDATE papers SET embedding = ${vec(e[0])}::vector, embedding_model = ${EMBEDDING_MODEL} WHERE id = ${p.paperId}`;61 const chunks = await sql`SELECT id, text FROM paper_chunks WHERE paper_id = ${p.paperId} AND embedding IS NULL ORDER BY chunk_index`;62 for (let i = 0; i < chunks.length; i += 64) {63 const batch = chunks.slice(i, i + 64); const vs = await embed(batch.map((c) => c.text)); if (!vs) return;64 await sql.begin(async (tx) => { for (let k = 0; k < batch.length; k++) await tx`UPDATE paper_chunks SET embedding = ${vec(vs[k])}::vector WHERE id = ${batch[k].id}`; });65 }66 await enqueue("relations", {}, { dedupe: true, runAfter: new Date(Date.now() + 30_000) });67}6869export async function dailyStats() {70 await sql`INSERT INTO paper_stats_daily (paper_id, day, views, downloads)71 SELECT paper_id, day, count(*) FILTER (WHERE kind = 'view'), count(*) FILTER (WHERE kind = 'download')72 FROM events WHERE NOT is_bot AND day >= current_date - 2 GROUP BY paper_id, day73 ON CONFLICT (paper_id, day) DO UPDATE SET views = EXCLUDED.views, downloads = EXCLUDED.downloads`;74 await sql`UPDATE papers p SET view_count = s.v, download_count = s.d FROM (SELECT paper_id, sum(views) v, sum(downloads) d FROM paper_stats_daily GROUP BY paper_id) s WHERE s.paper_id = p.id AND (p.view_count <> s.v OR p.download_count <> s.d)`;75 await sql`DELETE FROM events WHERE day < current_date - 400`;76 await sql`DELETE FROM search_log WHERE at < now() - interval '180 days'`;77 await sql`DELETE FROM jobs WHERE status IN ('done', 'failed') AND finished_at < now() - interval '30 days'`;78}7980export async function ingestScan(p: { dirs?: string[] } = {}) {81 await ensureStorage();82 const [extra] = await sql`SELECT value FROM settings WHERE key = 'ingest.extra_dirs'`;83 const dirs = [join(STORAGE_DIR, "inbox"), ...(p.dirs ?? []), ...((extra?.value as string[] | undefined) ?? [])];84 let found = 0, queued = 0;85 for (const dir of dirs) {86 let entries: string[] = [];87 try { entries = (await fs.readdir(dir)).filter((n) => n.toLowerCase().endsWith(".pdf")); } catch { continue; }88 for (const name of entries) {89 found++;90 const src = join(dir, name); const sha = await sha256File(src);91 const [known] = await sql`SELECT 1 FROM ingest_candidates WHERE sha256 = ${sha} LIMIT 1`;92 if (known) continue;93 const [existing] = await sql`SELECT f.id FROM files f WHERE f.sha256 = ${sha} LIMIT 1`;94 let fileId: string;95 if (existing) fileId = existing.id;96 else {97 const st = await fs.stat(src); const key = await storeFile(src, sha);98 const [f] = await sql`INSERT INTO files (kind, storage_key, original_name, mime, bytes, sha256) VALUES ('manuscript', ${key}, ${basename(src)}, 'application/pdf', ${st.size}, ${sha}) RETURNING id`;99 fileId = f.id;100 }101 const [c] = await sql`INSERT INTO ingest_candidates (source_path, file_id, sha256, status) VALUES (${src}, ${fileId}, ${sha}, 'queued') ON CONFLICT DO NOTHING RETURNING id`;102 if (c) { queued++; await enqueue("ingest_candidate", { candidateId: c.id }); }103 }104 }105 await sql`INSERT INTO settings (key, value) VALUES ('ingest.last_scan', ${sql.json({ at: new Date().toISOString(), dirs, found, queued } as never)}) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`;106 return { found, queued };107}108109export async function ingestCandidate(p: { candidateId: string }) {110 const [c] = await sql`SELECT c.*, f.storage_key FROM ingest_candidates c JOIN files f ON f.id = c.file_id WHERE c.id = ${p.candidateId}`; if (!c) return;111 await sql`UPDATE ingest_candidates SET status = 'processing' WHERE id = ${c.id}`;112 const path = storagePath(c.storage_key);113 const flags: { code: string; message: string }[] = [];114 const integ = await pdfIntegrity(path);115 if (!integ.ok) flags.push({ code: "integrity", message: integ.reason ?? "PDF failed integrity check" });116 const info = integ.info ?? null;117 await sql`UPDATE files SET integrity_ok = ${integ.ok}, pages = ${info?.pages ?? null}, pdf_meta = ${info ? sql.json(info.raw as never) : null} WHERE id = ${c.file_id}`;118 let proposal: Record<string, unknown> = {};119 if (integ.ok) {120 const first = await pdfText(path, { first: 1, last: 3 });121 const prop = proposeMetadata(first, info?.title, info?.author);122 proposal = { ...prop, pages: info?.pages ?? null, pdfMeta: { title: info?.title, author: info?.author, subject: info?.subject, keywords: info?.keywords, creationDate: info?.creationDate, modDate: info?.modDate, producer: info?.producer }, sourceFile: basename(c.source_path), sha256: c.sha256 };123 if (!prop.title) flags.push({ code: "title", message: "No title could be proposed" });124 if (!prop.authors.length) flags.push({ code: "authors", message: "No author line detected" });125 if (!prop.abstract) flags.push({ code: "abstract", message: "No abstract detected" });126 if (!prop.wp) flags.push({ code: "identifier", message: "No working paper number detected on the title page" });127 if (!prop.dateText) flags.push({ code: "date", message: "No release date detected; do not use the file timestamp" });128 if (first.trim().length < 200) flags.push({ code: "scanned", message: "Little or no extractable text (scanned PDF?)" });129 try { await pdfThumbnail(path, join(STORAGE_DIR, "previews", c.sha256), 60); } catch { flags.push({ code: "preview", message: "Preview generation failed" }); }130 // Duplicate detection: checksum, identifier, title similarity, author overlap.131 const dups: { paperId: string; identifier: string; title: string; reasons: string[]; score: number }[] = [];132 const papers = await sql`SELECT p.id, p.identifier, p.title, p.number, s.code, (SELECT array_agg(a.display_name) FROM paper_authors pa JOIN authors a ON a.id = pa.author_id WHERE pa.paper_id = p.id) AS authors,133 EXISTS (SELECT 1 FROM paper_versions v WHERE v.paper_id = p.id AND v.sha256 = ${c.sha256}) AS same_file FROM papers p JOIN series s ON s.id = p.series_id`;134 for (const pr of papers) {135 const reasons: string[] = []; let score = 0;136 if (pr.same_file) { reasons.push("identical checksum"); score += 1; }137 if (prop.wp && pr.code === prop.wp.seriesCode && pr.number === prop.wp.number) { reasons.push(`same identifier ${pr.identifier}`); score += 0.6; }138 const ts = prop.title ? titleSimilarity(prop.title, pr.title) : 0; if (ts >= 0.6) { reasons.push(`title similarity ${(ts * 100).toFixed(0)}%`); score += ts * 0.5; }139 const ov = prop.authors.filter((a) => (pr.authors ?? []).some((b: string) => titleSimilarity(a, b) >= 0.5)).length; if (ov && ts >= 0.3) { reasons.push(`${ov} shared author(s)`); score += 0.1 * ov; }140 if (reasons.length) dups.push({ paperId: pr.id, identifier: pr.identifier, title: pr.title, reasons, score: Math.min(1, score) });141 }142 dups.sort((a, b) => b.score - a.score);143 if (dups.length) flags.push({ code: "duplicate", message: `Possible duplicate of ${dups[0].identifier} (${dups[0].reasons.join(", ")})` });144 await sql`UPDATE ingest_candidates SET proposal = ${sql.json(proposal as never)}, flags = ${sql.json(flags as never)}, duplicates = ${sql.json(dups.slice(0, 5) as never)}, status = 'ready' WHERE id = ${c.id}`;145 } else await sql`UPDATE ingest_candidates SET flags = ${sql.json(flags as never)}, status = 'ready' WHERE id = ${c.id}`;146}147148export const handlers: Record<string, (payload: Record<string, unknown>) => Promise<unknown>> = {149 process_pdf: (p) => processPdf(p as { fileId: string; paperId?: string; versionId?: string }),150 embed_paper: (p) => embedPaper(p as { paperId: string }),151 relations: () => rebuildRelations(),152 daily_stats: () => dailyStats(),153 backup: () => runBackup(),154 ingest_scan: (p) => ingestScan(p as { dirs?: string[] }),155 ingest_candidate: (p) => ingestCandidate(p as { candidateId: string }),156};157