756 lines
26 KiB
Rust
756 lines
26 KiB
Rust
//! Persistence for AI page-analysis results.
|
||
//!
|
||
//! [`persist_analysis`] is the single transactional writer: it replaces a
|
||
//! page's OCR text, global auto-tags, and content warnings, upserts the
|
||
//! `page_analysis` row, and computes the kind-weighted `search_doc`
|
||
//! tsvector — all atomically, so re-analysis (delete+reinsert) is
|
||
//! idempotent and never leaves a half-written page. It deliberately does
|
||
//! NOT touch the per-user `page_tags` table.
|
||
//!
|
||
//! Auto-tag names resolve to the shared `tags` vocabulary via
|
||
//! [`crate::repo::tag::upsert_by_name`], the same path manga tags and
|
||
//! personal page tags use.
|
||
|
||
use sqlx::PgPool;
|
||
use uuid::Uuid;
|
||
|
||
use crate::crawler::jobs::{self, JobPayload};
|
||
use crate::domain::page_analysis::{
|
||
AnalysisHistoryRow, AnalysisMetrics, ChapterCoverage, ContentWarning, MangaCoverage,
|
||
ModelDuration, OcrKind, OcrLine, PageAnalysis, PageAnalysisDetail, PageSearchItem,
|
||
PageStatusItem, VisionAnalysis,
|
||
};
|
||
use chrono::{DateTime, Utc};
|
||
use crate::error::AppResult;
|
||
|
||
/// Filter set for [`page_search`]. `tags` are AND-ed (a page must carry
|
||
/// every one, satisfiable by EITHER the user's page tag or a global auto
|
||
/// tag); `text` ranks via the weighted tsvector; `cw_include` requires all
|
||
/// listed warnings; `cw_exclude` rejects any. All string values arrive
|
||
/// pre-normalized (tags lowercased, warnings validated) from the handler.
|
||
#[derive(Debug, Default, Clone)]
|
||
pub struct PageSearchQuery {
|
||
pub user_id: uuid::Uuid,
|
||
pub tags: Vec<String>,
|
||
pub text: Option<String>,
|
||
pub cw_include: Vec<String>,
|
||
pub cw_exclude: Vec<String>,
|
||
pub limit: i64,
|
||
pub offset: i64,
|
||
}
|
||
|
||
/// Longest tag the shared `tags` table accepts (`upsert_by_name` enforces
|
||
/// the same bound). Auto-tags over this are dropped here rather than
|
||
/// aborting the whole page's analysis on one bad model output.
|
||
const MAX_TAG_CHARS: usize = 64;
|
||
|
||
/// Enqueue an `analyze_page` job for `page_id`. `force` re-analyzes a page
|
||
/// that is already `done`. Enqueue is idempotent at the job level only in
|
||
/// that duplicate pending jobs are harmless — processing is idempotent.
|
||
pub async fn enqueue_for_page(pool: &PgPool, page_id: Uuid, force: bool) -> AppResult<()> {
|
||
jobs::enqueue(pool, &JobPayload::AnalyzePage { page_id, force }).await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// How wide a bulk re-enqueue reaches.
|
||
#[derive(Debug, Clone, Copy)]
|
||
pub enum ReenqueueScope {
|
||
/// Every page in the library.
|
||
All,
|
||
/// Every page across all chapters of one manga.
|
||
Manga(uuid::Uuid),
|
||
/// Every page of one chapter.
|
||
Chapter(uuid::Uuid),
|
||
}
|
||
|
||
/// Bulk-enqueue `analyze_page` jobs for existing pages within `scope` — the
|
||
/// admin backfill / re-analyze path.
|
||
///
|
||
/// When `only_unanalyzed` is true, pages that already have a `done`
|
||
/// analysis row are skipped and the enqueued jobs carry `force = false`.
|
||
/// When false, ALL in-scope pages are enqueued with `force = true` so the
|
||
/// worker re-analyzes even already-done pages (otherwise the worker's
|
||
/// skip-if-done net would no-op them). Pages with a pending/running
|
||
/// `analyze_page` job are always skipped so repeated calls don't pile up
|
||
/// duplicates. Returns the number of jobs enqueued.
|
||
pub async fn enqueue_pages(
|
||
pool: &PgPool,
|
||
scope: ReenqueueScope,
|
||
only_unanalyzed: bool,
|
||
) -> AppResult<u64> {
|
||
// Scope predicate; the bound uuid (when present) is always $2.
|
||
let scope_clause = match scope {
|
||
ReenqueueScope::All => "",
|
||
ReenqueueScope::Manga(_) => {
|
||
"AND p.chapter_id IN (SELECT id FROM chapters WHERE manga_id = $2)"
|
||
}
|
||
ReenqueueScope::Chapter(_) => "AND p.chapter_id = $2",
|
||
};
|
||
let sql = format!(
|
||
r#"
|
||
INSERT INTO crawler_jobs (payload)
|
||
SELECT jsonb_build_object('kind', 'analyze_page', 'page_id', p.id, 'force', NOT $1)
|
||
FROM pages p
|
||
WHERE ($1 = false OR NOT EXISTS (
|
||
SELECT 1 FROM page_analysis pa
|
||
WHERE pa.page_id = p.id AND pa.status = 'done'))
|
||
{scope_clause}
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM crawler_jobs j
|
||
WHERE j.payload->>'kind' = 'analyze_page'
|
||
AND j.payload->>'page_id' = p.id::text
|
||
AND j.state IN ('pending', 'running'))
|
||
"#
|
||
);
|
||
let query = sqlx::query(&sql).bind(only_unanalyzed);
|
||
let query = match scope {
|
||
ReenqueueScope::All => query,
|
||
ReenqueueScope::Manga(id) | ReenqueueScope::Chapter(id) => query.bind(id),
|
||
};
|
||
Ok(query.execute(pool).await?.rows_affected())
|
||
}
|
||
|
||
/// Per-manga analysis coverage for the admin overview. Only mangas that
|
||
/// have at least one page appear (nothing to analyze otherwise). `search`
|
||
/// filters by title (case-insensitive substring). Ordered by title.
|
||
/// Returns the page plus the total matching-manga count for pagination.
|
||
pub async fn manga_coverage(
|
||
pool: &PgPool,
|
||
search: Option<&str>,
|
||
limit: i64,
|
||
offset: i64,
|
||
) -> AppResult<(Vec<MangaCoverage>, i64)> {
|
||
let rows = sqlx::query_as::<_, MangaCoverage>(
|
||
r#"
|
||
SELECT m.id AS manga_id, m.title,
|
||
count(p.id) AS total_pages,
|
||
count(*) FILTER (WHERE pa.status = 'done') AS analyzed_pages
|
||
FROM mangas m
|
||
JOIN chapters c ON c.manga_id = m.id
|
||
JOIN pages p ON p.chapter_id = c.id
|
||
LEFT JOIN page_analysis pa ON pa.page_id = p.id
|
||
WHERE ($1::text IS NULL OR m.title ILIKE '%' || $1 || '%')
|
||
GROUP BY m.id, m.title
|
||
ORDER BY lower(m.title), m.id
|
||
LIMIT $2 OFFSET $3
|
||
"#,
|
||
)
|
||
.bind(search)
|
||
.bind(limit)
|
||
.bind(offset)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
let (total,): (i64,) = sqlx::query_as(
|
||
r#"
|
||
SELECT count(*) FROM (
|
||
SELECT m.id
|
||
FROM mangas m
|
||
JOIN chapters c ON c.manga_id = m.id
|
||
JOIN pages p ON p.chapter_id = c.id
|
||
WHERE ($1::text IS NULL OR m.title ILIKE '%' || $1 || '%')
|
||
GROUP BY m.id
|
||
) x
|
||
"#,
|
||
)
|
||
.bind(search)
|
||
.fetch_one(pool)
|
||
.await?;
|
||
Ok((rows, total))
|
||
}
|
||
|
||
/// Library-wide analysis coverage `(analyzed_pages, total_pages)` for the
|
||
/// admin overview. `total_pages` is every page; `analyzed_pages` are those
|
||
/// with a `done` analysis row.
|
||
pub async fn library_coverage(pool: &PgPool) -> AppResult<(i64, i64)> {
|
||
let row: (i64, i64) = sqlx::query_as(
|
||
r#"
|
||
SELECT
|
||
count(*) FILTER (WHERE pa.status = 'done')::bigint AS analyzed,
|
||
count(p.id)::bigint AS total
|
||
FROM pages p
|
||
LEFT JOIN page_analysis pa ON pa.page_id = p.id
|
||
"#,
|
||
)
|
||
.fetch_one(pool)
|
||
.await?;
|
||
Ok(row)
|
||
}
|
||
|
||
/// Per-chapter analysis coverage for one manga, ordered by chapter number.
|
||
pub async fn chapter_coverage(
|
||
pool: &PgPool,
|
||
manga_id: uuid::Uuid,
|
||
) -> AppResult<Vec<ChapterCoverage>> {
|
||
let rows = sqlx::query_as::<_, ChapterCoverage>(
|
||
r#"
|
||
SELECT c.id AS chapter_id, c.number, c.title,
|
||
count(p.id) AS total_pages,
|
||
count(*) FILTER (WHERE pa.status = 'done') AS analyzed_pages
|
||
FROM chapters c
|
||
JOIN pages p ON p.chapter_id = c.id
|
||
LEFT JOIN page_analysis pa ON pa.page_id = p.id
|
||
WHERE c.manga_id = $1
|
||
GROUP BY c.id, c.number, c.title
|
||
ORDER BY c.number, c.id
|
||
"#,
|
||
)
|
||
.bind(manga_id)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
Ok(rows)
|
||
}
|
||
|
||
/// Per-page status grid for a chapter. `status` resolves to the analysis
|
||
/// row's state (`done`/`failed`), else `queued` when a pending/running
|
||
/// analyze_page job exists, else `none`.
|
||
pub async fn chapter_page_status(
|
||
pool: &PgPool,
|
||
chapter_id: uuid::Uuid,
|
||
) -> AppResult<Vec<PageStatusItem>> {
|
||
let rows = sqlx::query_as::<_, PageStatusItem>(
|
||
r#"
|
||
SELECT p.id AS page_id, p.page_number,
|
||
COALESCE(
|
||
pa.status,
|
||
CASE WHEN EXISTS (
|
||
SELECT 1 FROM crawler_jobs j
|
||
WHERE j.payload->>'kind' = 'analyze_page'
|
||
AND j.payload->>'page_id' = p.id::text
|
||
AND j.state IN ('pending', 'running')
|
||
) THEN 'queued' ELSE 'none' END
|
||
) AS status
|
||
FROM pages p
|
||
LEFT JOIN page_analysis pa ON pa.page_id = p.id
|
||
WHERE p.chapter_id = $1
|
||
ORDER BY p.page_number
|
||
"#,
|
||
)
|
||
.bind(chapter_id)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
Ok(rows)
|
||
}
|
||
|
||
/// Filters for [`list_history`]. `None` widens the scope.
|
||
#[derive(Debug, Default, Clone)]
|
||
pub struct AnalysisHistoryFilter<'a> {
|
||
/// Exact analysis status (`done` or `failed`).
|
||
pub status: Option<&'a str>,
|
||
/// Only NSFW-flagged pages.
|
||
pub nsfw_only: bool,
|
||
/// Case-insensitive manga-title substring.
|
||
pub search: Option<&'a str>,
|
||
}
|
||
|
||
/// Paginated, newest-first analysis history — every completed/failed page
|
||
/// analysis resolved to its manga/chapter/page context. Reads the
|
||
/// persistent `page_analysis` table, so unlike the crawler job history it
|
||
/// is not bounded by job reaping. Returns the page slice plus the filtered
|
||
/// total. `pending` rows are excluded — history is terminal outcomes only.
|
||
pub async fn list_history(
|
||
pool: &PgPool,
|
||
filter: AnalysisHistoryFilter<'_>,
|
||
limit: i64,
|
||
offset: i64,
|
||
) -> AppResult<(Vec<AnalysisHistoryRow>, i64)> {
|
||
let search_pat = filter
|
||
.search
|
||
.map(|s| format!("%{}%", s.trim()))
|
||
.filter(|p| p.len() > 2);
|
||
|
||
let items = sqlx::query_as::<_, AnalysisHistoryRow>(
|
||
r#"
|
||
SELECT
|
||
pa.page_id,
|
||
p.page_number,
|
||
p.chapter_id,
|
||
c.number AS chapter_number,
|
||
c.manga_id,
|
||
m.title AS manga_title,
|
||
pa.status,
|
||
pa.is_nsfw,
|
||
pa.model,
|
||
pa.error,
|
||
pa.analyzed_at,
|
||
pa.duration_ms
|
||
FROM page_analysis pa
|
||
JOIN pages p ON p.id = pa.page_id
|
||
JOIN chapters c ON c.id = p.chapter_id
|
||
JOIN mangas m ON m.id = c.manga_id
|
||
WHERE pa.status <> 'pending'
|
||
AND ($1::text IS NULL OR pa.status = $1)
|
||
AND ($2::bool IS FALSE OR pa.is_nsfw)
|
||
AND ($3::text IS NULL OR m.title ILIKE $3)
|
||
ORDER BY pa.analyzed_at DESC NULLS LAST, pa.page_id
|
||
LIMIT $4 OFFSET $5
|
||
"#,
|
||
)
|
||
.bind(filter.status)
|
||
.bind(filter.nsfw_only)
|
||
.bind(&search_pat)
|
||
.bind(limit)
|
||
.bind(offset)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
let (total,): (i64,) = sqlx::query_as(
|
||
r#"
|
||
SELECT count(*)
|
||
FROM page_analysis pa
|
||
JOIN pages p ON p.id = pa.page_id
|
||
JOIN chapters c ON c.id = p.chapter_id
|
||
JOIN mangas m ON m.id = c.manga_id
|
||
WHERE pa.status <> 'pending'
|
||
AND ($1::text IS NULL OR pa.status = $1)
|
||
AND ($2::bool IS FALSE OR pa.is_nsfw)
|
||
AND ($3::text IS NULL OR m.title ILIKE $3)
|
||
"#,
|
||
)
|
||
.bind(filter.status)
|
||
.bind(filter.nsfw_only)
|
||
.bind(&search_pat)
|
||
.fetch_one(pool)
|
||
.await?;
|
||
|
||
Ok((items, total))
|
||
}
|
||
|
||
/// Full analysis detail for one page. `None` when the page itself doesn't
|
||
/// exist; an existing-but-unanalyzed page returns `status = "none"` with
|
||
/// empty OCR/tags/warnings so the UI can show "not analyzed yet".
|
||
pub async fn page_detail(
|
||
pool: &PgPool,
|
||
page_id: uuid::Uuid,
|
||
) -> AppResult<Option<PageAnalysisDetail>> {
|
||
let Some((page_number, chapter_id, manga_id)): Option<(i32, uuid::Uuid, uuid::Uuid)> =
|
||
sqlx::query_as(
|
||
"SELECT p.page_number, p.chapter_id, c.manga_id \
|
||
FROM pages p JOIN chapters c ON c.id = p.chapter_id WHERE p.id = $1",
|
||
)
|
||
.bind(page_id)
|
||
.fetch_optional(pool)
|
||
.await?
|
||
else {
|
||
return Ok(None);
|
||
};
|
||
|
||
let analysis = load(pool, page_id).await?;
|
||
|
||
let ocr = sqlx::query_as::<_, OcrLine>(
|
||
"SELECT kind, text FROM page_ocr_text WHERE page_id = $1 ORDER BY ord, id",
|
||
)
|
||
.bind(page_id)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
let tags: Vec<String> = sqlx::query_scalar(
|
||
"SELECT t.name FROM page_auto_tags pat JOIN tags t ON t.id = pat.tag_id \
|
||
WHERE pat.page_id = $1 ORDER BY t.name",
|
||
)
|
||
.bind(page_id)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
let content_warnings = sqlx::query_scalar::<_, ContentWarning>(
|
||
"SELECT warning FROM page_content_warnings WHERE page_id = $1 ORDER BY warning",
|
||
)
|
||
.bind(page_id)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
Ok(Some(PageAnalysisDetail {
|
||
page_id,
|
||
page_number,
|
||
chapter_id,
|
||
manga_id,
|
||
status: analysis
|
||
.as_ref()
|
||
.map(|a| format!("{:?}", a.status).to_lowercase())
|
||
.unwrap_or_else(|| "none".to_string()),
|
||
is_nsfw: analysis.as_ref().map(|a| a.is_nsfw).unwrap_or(false),
|
||
scene_description: analysis.as_ref().and_then(|a| a.scene_description.clone()),
|
||
model: analysis.as_ref().and_then(|a| a.model.clone()),
|
||
error: analysis.as_ref().and_then(|a| a.error.clone()),
|
||
analyzed_at: analysis.as_ref().and_then(|a| a.analyzed_at),
|
||
ocr,
|
||
tags,
|
||
content_warnings,
|
||
}))
|
||
}
|
||
|
||
/// Deduplicated union of content warnings across all of a manga's pages,
|
||
/// ordered alphabetically. Powers the manga-detail content-warning banner.
|
||
pub async fn warnings_for_manga(
|
||
pool: &PgPool,
|
||
manga_id: uuid::Uuid,
|
||
) -> AppResult<Vec<ContentWarning>> {
|
||
let rows = sqlx::query_scalar::<_, ContentWarning>(
|
||
r#"
|
||
SELECT DISTINCT pw.warning
|
||
FROM page_content_warnings pw
|
||
JOIN pages p ON p.id = pw.page_id
|
||
JOIN chapters c ON c.id = p.chapter_id
|
||
WHERE c.manga_id = $1
|
||
ORDER BY pw.warning
|
||
"#,
|
||
)
|
||
.bind(manga_id)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
Ok(rows)
|
||
}
|
||
|
||
/// Load a page's analysis row, if it has one.
|
||
pub async fn load(pool: &PgPool, page_id: Uuid) -> AppResult<Option<PageAnalysis>> {
|
||
let row = sqlx::query_as::<_, PageAnalysis>(
|
||
r#"
|
||
SELECT page_id, status, scene_description, is_nsfw, model, error, analyzed_at
|
||
FROM page_analysis
|
||
WHERE page_id = $1
|
||
"#,
|
||
)
|
||
.bind(page_id)
|
||
.fetch_optional(pool)
|
||
.await?;
|
||
Ok(row)
|
||
}
|
||
|
||
/// Record that analysis failed terminally for a page (retries exhausted /
|
||
/// dead-lettered). Leaves a `failed` row so the page's state is
|
||
/// observable; it simply contributes nothing to search.
|
||
pub async fn mark_failed(pool: &PgPool, page_id: Uuid, error: &str) -> AppResult<()> {
|
||
sqlx::query(
|
||
r#"
|
||
INSERT INTO page_analysis (page_id, status, error)
|
||
VALUES ($1, 'failed', $2)
|
||
ON CONFLICT (page_id) DO UPDATE
|
||
SET status = 'failed', error = EXCLUDED.error, updated_at = now()
|
||
"#,
|
||
)
|
||
.bind(page_id)
|
||
.bind(error)
|
||
.execute(pool)
|
||
.await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Record the wall-clock the worker spent on a page. Best-effort: updates the
|
||
/// existing `page_analysis` row (written by `persist_analysis`/`mark_failed`);
|
||
/// a transient failure with no row yet simply updates nothing.
|
||
pub async fn record_duration(
|
||
pool: &PgPool,
|
||
page_id: Uuid,
|
||
duration_ms: i64,
|
||
) -> AppResult<()> {
|
||
sqlx::query("UPDATE page_analysis SET duration_ms = $2 WHERE page_id = $1")
|
||
.bind(page_id)
|
||
.bind(duration_ms)
|
||
.execute(pool)
|
||
.await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Aggregate analysis timing + outcome roll-up over an optional time window
|
||
/// (`since = None` → all time). Mean duration ignores NULL `duration_ms`
|
||
/// (pre-tracking rows); `by_model` breaks the mean out per model.
|
||
pub async fn analysis_metrics(
|
||
pool: &PgPool,
|
||
since: Option<DateTime<Utc>>,
|
||
) -> AppResult<AnalysisMetrics> {
|
||
let (n, ok, failed, avg_ms): (i64, i64, i64, Option<f64>) = sqlx::query_as(
|
||
r#"
|
||
SELECT
|
||
COUNT(*) AS n,
|
||
COUNT(*) FILTER (WHERE status = 'done') AS ok,
|
||
COUNT(*) FILTER (WHERE status = 'failed') AS failed,
|
||
-- Mean over successful pages only, so it reconciles with the
|
||
-- by-model breakdown below (which is done-only). Failed /
|
||
-- timed-out durations would otherwise skew the headline up.
|
||
AVG(duration_ms) FILTER (WHERE status = 'done')::float8 AS avg_ms
|
||
FROM page_analysis
|
||
WHERE status <> 'pending'
|
||
AND ($1::timestamptz IS NULL OR analyzed_at >= $1)
|
||
"#,
|
||
)
|
||
.bind(since)
|
||
.fetch_one(pool)
|
||
.await?;
|
||
|
||
let by_model = sqlx::query_as::<_, ModelDuration>(
|
||
r#"
|
||
SELECT model, AVG(duration_ms)::float8 AS avg_ms, COUNT(*) AS n
|
||
FROM page_analysis
|
||
WHERE status = 'done'
|
||
AND duration_ms IS NOT NULL
|
||
AND ($1::timestamptz IS NULL OR analyzed_at >= $1)
|
||
GROUP BY model
|
||
ORDER BY n DESC
|
||
"#,
|
||
)
|
||
.bind(since)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
Ok(AnalysisMetrics { n, ok, failed, avg_ms, by_model })
|
||
}
|
||
|
||
/// Persist a completed analysis for a page, replacing any previous result.
|
||
///
|
||
/// The model's free-form `kind` / `content_type` strings are mapped onto
|
||
/// the closed vocabularies here ([`OcrKind::from_model_str`] /
|
||
/// [`ContentWarning::from_model_str`]); unrecognized warnings are dropped,
|
||
/// unknown OCR kinds fall back to `narration`. Empty / over-long tags and
|
||
/// empty OCR text are skipped so one bad item never aborts the page.
|
||
pub async fn persist_analysis(
|
||
pool: &PgPool,
|
||
page_id: Uuid,
|
||
analysis: &VisionAnalysis,
|
||
model: &str,
|
||
) -> AppResult<()> {
|
||
// Kind-weighted text buckets for the tsvector: A=speech/title,
|
||
// B=narration/thought/caption, C=scene, D=sfx.
|
||
let mut bucket_a = String::new();
|
||
let mut bucket_b = String::new();
|
||
let bucket_c = analysis.scene_description.trim().to_string();
|
||
let mut bucket_d = String::new();
|
||
|
||
let mut ocr_rows: Vec<(OcrKind, String)> = Vec::new();
|
||
for r in &analysis.ocr_results {
|
||
let text = r.text.trim();
|
||
if text.is_empty() {
|
||
continue;
|
||
}
|
||
let kind = OcrKind::from_model_str(&r.kind);
|
||
match kind.weight() {
|
||
'A' => push_token(&mut bucket_a, text),
|
||
'B' => push_token(&mut bucket_b, text),
|
||
'D' => push_token(&mut bucket_d, text),
|
||
_ => {}
|
||
}
|
||
ocr_rows.push((kind, text.to_string()));
|
||
}
|
||
|
||
// Dedup tags case-insensitively, preserving first-seen order; drop
|
||
// empties and over-long names rather than failing the transaction.
|
||
let mut seen_tags = std::collections::HashSet::new();
|
||
let mut tags: Vec<String> = Vec::new();
|
||
for raw in &analysis.tagging_results {
|
||
let t = raw.trim();
|
||
if t.is_empty() || t.chars().count() > MAX_TAG_CHARS {
|
||
continue;
|
||
}
|
||
if seen_tags.insert(t.to_lowercase()) {
|
||
tags.push(t.to_string());
|
||
}
|
||
}
|
||
|
||
// Map + dedup warnings.
|
||
let mut seen_warn = std::collections::HashSet::new();
|
||
let mut warnings: Vec<ContentWarning> = Vec::new();
|
||
for raw in &analysis.safety_flag.content_type {
|
||
if let Some(w) = ContentWarning::from_model_str(raw) {
|
||
if seen_warn.insert(w) {
|
||
warnings.push(w);
|
||
}
|
||
}
|
||
}
|
||
|
||
let mut tx = pool.begin().await?;
|
||
|
||
sqlx::query("DELETE FROM page_ocr_text WHERE page_id = $1")
|
||
.bind(page_id)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
sqlx::query("DELETE FROM page_auto_tags WHERE page_id = $1")
|
||
.bind(page_id)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
sqlx::query("DELETE FROM page_content_warnings WHERE page_id = $1")
|
||
.bind(page_id)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
|
||
for (ord, (kind, text)) in ocr_rows.iter().enumerate() {
|
||
sqlx::query(
|
||
"INSERT INTO page_ocr_text (page_id, kind, text, ord) VALUES ($1, $2, $3, $4)",
|
||
)
|
||
.bind(page_id)
|
||
.bind(kind)
|
||
.bind(text)
|
||
.bind(ord as i32)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
}
|
||
|
||
for tag in &tags {
|
||
let tag_row = crate::repo::tag::upsert_by_name(&mut *tx, tag).await?;
|
||
sqlx::query(
|
||
"INSERT INTO page_auto_tags (page_id, tag_id) VALUES ($1, $2) \
|
||
ON CONFLICT (page_id, tag_id) DO NOTHING",
|
||
)
|
||
.bind(page_id)
|
||
.bind(tag_row.id)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
}
|
||
|
||
for w in &warnings {
|
||
sqlx::query(
|
||
"INSERT INTO page_content_warnings (page_id, warning) VALUES ($1, $2) \
|
||
ON CONFLICT (page_id, warning) DO NOTHING",
|
||
)
|
||
.bind(page_id)
|
||
.bind(w)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
}
|
||
|
||
let scene = if bucket_c.is_empty() {
|
||
None
|
||
} else {
|
||
Some(bucket_c.as_str())
|
||
};
|
||
sqlx::query(
|
||
r#"
|
||
INSERT INTO page_analysis
|
||
(page_id, status, scene_description, is_nsfw, model, analyzed_at, search_doc)
|
||
VALUES
|
||
($1, 'done', $2, $3, $4, now(),
|
||
setweight(to_tsvector('simple', $5), 'A') ||
|
||
setweight(to_tsvector('simple', $6), 'B') ||
|
||
setweight(to_tsvector('simple', $7), 'C') ||
|
||
setweight(to_tsvector('simple', $8), 'D'))
|
||
ON CONFLICT (page_id) DO UPDATE SET
|
||
status = 'done',
|
||
scene_description = EXCLUDED.scene_description,
|
||
is_nsfw = EXCLUDED.is_nsfw,
|
||
model = EXCLUDED.model,
|
||
analyzed_at = now(),
|
||
error = NULL,
|
||
search_doc = EXCLUDED.search_doc,
|
||
updated_at = now()
|
||
"#,
|
||
)
|
||
.bind(page_id)
|
||
.bind(scene)
|
||
.bind(analysis.safety_flag.is_nsfw)
|
||
.bind(model)
|
||
.bind(&bucket_a)
|
||
.bind(&bucket_b)
|
||
.bind(&bucket_c)
|
||
.bind(&bucket_d)
|
||
.execute(&mut *tx)
|
||
.await?;
|
||
|
||
tx.commit().await?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Content search over pages: multi-tag AND across (user page tags ∪
|
||
/// global auto tags), optional weighted text search over the OCR/scene
|
||
/// document, and content-warning include/exclude. Returns one row per
|
||
/// matching page plus the total for pagination.
|
||
///
|
||
/// Empty `tags` makes the tag clause vacuously true (text- or
|
||
/// warning-only search), mirroring the manga search's `unnest` idiom. The
|
||
/// caller is responsible for requiring at least one positive filter so
|
||
/// this never degenerates into "every page".
|
||
pub async fn page_search(
|
||
pool: &PgPool,
|
||
q: &PageSearchQuery,
|
||
) -> AppResult<(Vec<PageSearchItem>, i64)> {
|
||
// `text` participates in three places ($3): the rank, the match
|
||
// predicate, and the order key. Empty string disables text filtering.
|
||
let text = q.text.as_deref().unwrap_or("").trim();
|
||
|
||
const WHERE: &str = r#"
|
||
NOT EXISTS (
|
||
SELECT 1 FROM unnest($2::text[]) AS req(name)
|
||
WHERE NOT EXISTS (
|
||
SELECT 1 FROM page_tags ut
|
||
JOIN tags t ON t.id = ut.tag_id
|
||
WHERE ut.page_id = p.id AND ut.user_id = $1 AND lower(t.name) = req.name
|
||
UNION ALL
|
||
SELECT 1 FROM page_auto_tags at
|
||
JOIN tags t ON t.id = at.tag_id
|
||
WHERE at.page_id = p.id AND lower(t.name) = req.name
|
||
)
|
||
)
|
||
AND ($3 = '' OR pa.search_doc @@ plainto_tsquery('simple', $3))
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM unnest($4::text[]) AS req(w)
|
||
WHERE NOT EXISTS (
|
||
SELECT 1 FROM page_content_warnings pw
|
||
WHERE pw.page_id = p.id AND pw.warning = req.w
|
||
)
|
||
)
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM page_content_warnings pw
|
||
WHERE pw.page_id = p.id AND pw.warning = ANY($5::text[])
|
||
)
|
||
"#;
|
||
|
||
let rows_sql = format!(
|
||
r#"
|
||
SELECT
|
||
p.id AS page_id,
|
||
p.chapter_id AS chapter_id,
|
||
ch.manga_id AS manga_id,
|
||
p.page_number AS page_number,
|
||
ch.number AS chapter_number,
|
||
ch.title AS chapter_title,
|
||
m.title AS manga_title,
|
||
p.storage_key AS storage_key,
|
||
COALESCE(pa.is_nsfw, false) AS is_nsfw,
|
||
COALESCE(
|
||
(SELECT array_agg(pw.warning ORDER BY pw.warning)
|
||
FROM page_content_warnings pw WHERE pw.page_id = p.id),
|
||
ARRAY[]::text[]
|
||
) AS content_warnings,
|
||
COALESCE(ts_rank(pa.search_doc, plainto_tsquery('simple', $3)), 0)::real AS rank
|
||
FROM pages p
|
||
JOIN chapters ch ON ch.id = p.chapter_id
|
||
JOIN mangas m ON m.id = ch.manga_id
|
||
LEFT JOIN page_analysis pa ON pa.page_id = p.id
|
||
WHERE {WHERE}
|
||
ORDER BY (CASE WHEN $3 = '' THEN 0 ELSE 1 END) DESC, rank DESC, p.id
|
||
LIMIT $6 OFFSET $7
|
||
"#
|
||
);
|
||
let rows = sqlx::query_as::<_, PageSearchItem>(&rows_sql)
|
||
.bind(q.user_id)
|
||
.bind(&q.tags)
|
||
.bind(text)
|
||
.bind(&q.cw_include)
|
||
.bind(&q.cw_exclude)
|
||
.bind(q.limit)
|
||
.bind(q.offset)
|
||
.fetch_all(pool)
|
||
.await?;
|
||
|
||
let count_sql = format!(
|
||
"SELECT count(*) FROM pages p \
|
||
JOIN chapters ch ON ch.id = p.chapter_id \
|
||
JOIN mangas m ON m.id = ch.manga_id \
|
||
LEFT JOIN page_analysis pa ON pa.page_id = p.id \
|
||
WHERE {WHERE}"
|
||
);
|
||
let (total,): (i64,) = sqlx::query_as(&count_sql)
|
||
.bind(q.user_id)
|
||
.bind(&q.tags)
|
||
.bind(text)
|
||
.bind(&q.cw_include)
|
||
.bind(&q.cw_exclude)
|
||
.fetch_one(pool)
|
||
.await?;
|
||
|
||
Ok((rows, total))
|
||
}
|
||
|
||
/// Append a token to a tsvector text bucket with a trailing space.
|
||
fn push_token(bucket: &mut String, text: &str) {
|
||
bucket.push_str(text);
|
||
bucket.push(' ');
|
||
}
|