feat(analysis): page-analysis schema, domain types, and analyze_page job kind
Adds the persistence foundation for the AI content-analysis worker:
- Migration 0025: page_analysis (status + scene + is_nsfw + weighted
search_doc tsvector), page_ocr_text (kind-tagged pieces), global
page_auto_tags (shared tags vocabulary, separate from per-user
page_tags), and page_content_warnings (closed vocabulary).
- domain::page_analysis: AnalysisStatus/OcrKind/ContentWarning enums with
kind->tsvector-weight mapping and lenient model-string parsing, plus the
VisionAnalysis response DTOs.
- JobPayload::AnalyzePage { page_id, force } + KIND_ANALYZE_PAGE, reusing
the existing crawler_jobs queue.
- repo::page_analysis: enqueue_for_page, load, mark_failed, and the single
transactional persist_analysis (delete+reinsert; idempotent; never
touches user page_tags; computes the A/B/C/D-weighted search_doc).
Tests: repo integration (persist/idempotency/user-tags-untouched/enqueue/
mark_failed/speech>sfx ranking) + unit (weights, parsing, DTO, job serde).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -9,6 +9,7 @@ pub mod crawler;
|
||||
pub mod genre;
|
||||
pub mod manga;
|
||||
pub mod page;
|
||||
pub mod page_analysis;
|
||||
pub mod page_tag;
|
||||
pub mod read_progress;
|
||||
pub mod session;
|
||||
|
||||
224
backend/src/repo/page_analysis.rs
Normal file
224
backend/src/repo/page_analysis.rs
Normal file
@@ -0,0 +1,224 @@
|
||||
//! 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::{ContentWarning, OcrKind, PageAnalysis, VisionAnalysis};
|
||||
use crate::error::AppResult;
|
||||
|
||||
/// 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(())
|
||||
}
|
||||
|
||||
/// 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(())
|
||||
}
|
||||
|
||||
/// 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(())
|
||||
}
|
||||
|
||||
/// 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(' ');
|
||||
}
|
||||
Reference in New Issue
Block a user