fix: audit perbaikan NATS subject, worker spawn, frontend options, pipeline PDF output

Bug fixes:
1. NATS subject doubled prefix — gateway publish ke tools.tools.scan.*
   padahal workers subscribe ke tools.scan.*. Fix: pakai group() method
2. Worker subscriptions pake tokio::select! — cuma satu subscription
   yang diproses, sisanya di-cancel. Fix: ganti ke tokio::spawn + loop
3. WebSocket URL hardcoded ke localhost:3001 — gak jalan di prod.
   Fix: infer dari window.location atau env var
4. Frontend scan options hardcoded — user gak bisa atur OCR/enhance/format.
   Fix: interactive toggles, selects, slider
5. Frontend compress quality hardcoded — user gak bisa atur kualitas.
   Fix: quality range slider
6. Pipeline cuma output PNG — gak ada PDF/OCR.
   Fix: generate searchable PDF kalo tesseract feature enabled

Enhancements:
- Tool::group() method added — returns 'scan', 'image', 'pdf' dll
- Dockerfile: --features tesseract pas cargo build
- leptess OCR: proper API usage (set_image_from_mem, recognize, etc.)

Co-Authored-By: Kilo <kilo@kilo.ai>
This commit is contained in:
asepharyana
2026-07-24 18:24:22 +07:00
co-authored by Kilo
parent 37cc3a7486
commit d2bfcb83a0
9 changed files with 290 additions and 121 deletions
+7 -7
View File
@@ -36,21 +36,21 @@ pub enum Tool {
} }
impl Tool { impl Tool {
/// Returns the NATS subject prefix for this tool. /// Returns the NATS subject group (without "tools." prefix).
pub fn subject_prefix(&self) -> &'static str { pub fn group(&self) -> &'static str {
match self { match self {
Tool::Scan => "tools.scan", Tool::Scan => "scan",
Tool::ImageCompress Tool::ImageCompress
| Tool::ImageResize | Tool::ImageResize
| Tool::ImageConvert | Tool::ImageConvert
| Tool::RemoveBg => "tools.image", | Tool::RemoveBg => "image",
Tool::PdfMerge Tool::PdfMerge
| Tool::PdfSplit | Tool::PdfSplit
| Tool::ImagesToPdf | Tool::ImagesToPdf
| Tool::PdfCompress | Tool::PdfCompress
| Tool::PdfToImages => "tools.pdf", | Tool::PdfToImages => "pdf",
Tool::VideoCompress | Tool::VideoTrim | Tool::GifMaker => "tools.video", Tool::VideoCompress | Tool::VideoTrim | Tool::GifMaker => "video",
Tool::AudioExtract | Tool::AudioConvert => "tools.audio", Tool::AudioExtract | Tool::AudioConvert => "audio",
} }
} }
+2 -2
View File
@@ -16,8 +16,8 @@ impl NatsPublisher {
/// Publish a job to the appropriate NATS subject. /// Publish a job to the appropriate NATS subject.
pub async fn publish_job(nats: &Client, tool: &Tool, job: &Job) -> Result<(), NatsError> { pub async fn publish_job(nats: &Client, tool: &Tool, job: &Job) -> Result<(), NatsError> {
let prefix = tool.subject_prefix(); let group = tool.group();
let subject = nats::job_subject(prefix, &job.id.to_string()); let subject = nats::job_subject(group, &job.id.to_string());
let payload = serde_json::to_vec(job) let payload = serde_json::to_vec(job)
.map_err(|e| NatsError::Publish(e.to_string()))?; .map_err(|e| NatsError::Publish(e.to_string()))?;
+8 -10
View File
@@ -53,18 +53,16 @@ impl JobConsumer {
.await?; .await?;
tracing::info!("Subscribed to tools.scheduler.cleanup"); tracing::info!("Subscribed to tools.scheduler.cleanup");
let redis_clone = redis.clone(); // Process messages concurrently — use tokio::spawn so ALL run
let config_clone = config.clone(); tokio::spawn(Self::process_subscription(scan_sub, redis.clone(), config.clone()));
tokio::spawn(Self::process_subscription(image_sub, redis.clone(), config.clone()));
tokio::spawn(Self::process_subscription(pdf_sub, redis.clone(), config.clone()));
tokio::spawn(Self::process_cleanup(cleanup_sub, config.clone()));
// Process messages concurrently // Wait forever so the process doesn't exit
tokio::select! { loop {
_ = Self::process_subscription(scan_sub, redis.clone(), config.clone()) => {}, tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
_ = Self::process_subscription(image_sub, redis.clone(), config.clone()) => {},
_ = Self::process_subscription(pdf_sub, redis.clone(), config.clone()) => {},
_ = Self::process_cleanup(cleanup_sub, config_clone) => {},
} }
Ok(())
} }
/// Process messages from a NATS subscription. /// Process messages from a NATS subscription.
+1 -1
View File
@@ -50,7 +50,7 @@ impl ProgressReporter {
message: message.to_string(), message: message.to_string(),
}; };
let subject = format!("tools.{}.progress.{}", self.tool.subject_prefix(), self.job_id); let subject = format!("tools.{}.progress.{}", self.tool.group(), self.job_id);
if let Ok(payload) = serde_json::to_vec(&progress_msg) { if let Ok(payload) = serde_json::to_vec(&progress_msg) {
let _ = self.nats.publish(subject, payload.into()).await; let _ = self.nats.publish(subject, payload.into()).await;
} }
+67 -70
View File
@@ -1,5 +1,4 @@
use image::GrayImage; use image::GrayImage;
use tools_common::error::PipelineError; use tools_common::error::PipelineError;
/// OCR result with text and word-level bounding boxes. /// OCR result with text and word-level bounding boxes.
@@ -26,87 +25,85 @@ pub struct Bbox {
pub height: i32, pub height: i32,
} }
/// Initialize Tesseract OCR engine. /// Run OCR on a grayscale image and return extracted text.
/// Uses leptess crate which binds to libtesseract. /// Requires the "tesseract" feature. Falls back gracefully when unavailable.
/// Falls back gracefully if Tesseract is not installed.
#[cfg(feature = "tesseract")] #[cfg(feature = "tesseract")]
fn init_tesseract(lang: &str) -> Result<leptess::LepTess, PipelineError> { pub fn ocr_text(img: &GrayImage, lang: &str) -> Result<OcrResult, PipelineError> {
let tessdata_prefix = std::env::var("TESSDATA_PREFIX") let tessdata_prefix = std::env::var("TESSDATA_PREFIX")
.unwrap_or_else(|_| "/usr/share/tesseract-ocr/5/tessdata".to_string()); .unwrap_or_else(|_| "/usr/share/tesseract-ocr/5/tessdata".to_string());
let mut tess = leptess::LepTess::new(Some(&tessdata_prefix), lang) let mut tess = leptess::LepTess::new(Some(&tessdata_prefix), lang)
.map_err(|e| PipelineError::Ocr(format!("Failed to init Tesseract: {}", e)))?; .map_err(|e| PipelineError::Ocr(format!("Failed to init Tesseract: {}", e)))?;
Ok(tess) tess.set_source_resolution(300);
}
/// Run OCR on a grayscale image and return extracted text. // Set image from raw bytes
/// Uses Tesseract via leptess crate when the "tesseract" feature is enabled. let bytes = img.to_vec();
/// Falls back to a placeholder when Tesseract is unavailable. let w = img.width() as i32;
pub fn ocr_text(img: &GrayImage, lang: &str) -> Result<OcrResult, PipelineError> { let h = img.height() as i32;
#[cfg(feature = "tesseract")]
{
let mut tess = init_tesseract(lang)?;
let width = img.width() as i32; // leptess expects raw 8-bit grayscale data
let height = img.height() as i32; // Use set_image_from_mem which loads from memory buffer
if let Err(e) = tess.set_image_from_mem(&bytes) {
// Set image from memory tracing::warn!("set_image_from_mem failed: {:?}, trying set_image", e);
tess.set_image_from_mem(&img.to_vec(), width, height, 1, width) return Err(PipelineError::Ocr(format!("set_image_from_mem failed: {:?}", e)));
.map_err(|e| PipelineError::Ocr(format!("Failed to set image: {}", e)))?;
tess.set_source_resolution(300);
// Set PSM to automatic
tess.set_page_seg_mode(3);
let text = tess.get_utf8_text()
.map_err(|e| PipelineError::Ocr(format!("OCR failed: {}", e)))?;
let words = tess.get_words()
.iter()
.map(|w| OcrWord {
text: w.text.clone(),
bbox: Bbox {
x: w.x,
y: w.y,
width: w.w,
height: w.h,
},
confidence: w.confidence,
})
.collect();
let confidence = if words.is_empty() {
0.0
} else {
words.iter().map(|w| w.confidence as f32).sum::<f32>() / words.len() as f32
};
Ok(OcrResult {
full_text: text,
words,
confidence,
})
} }
#[cfg(not(feature = "tesseract"))] tess.recognize();
{
tracing::warn!("Tesseract feature not enabled, OCR returning placeholder"); let text = tess.get_utf8_text().unwrap_or_default();
Ok(OcrResult { let confidence = tess.mean_text_conf() as f32;
full_text: String::new(),
words: Vec::new(), // Get word-level bounding boxes from LSTM box text
confidence: 0.0, let words = if let Ok(box_text) = tess.get_lstm_box_text(1) {
}) parse_lstm_boxes(&box_text)
} else {
Vec::new()
};
Ok(OcrResult {
full_text: text,
words,
confidence,
})
}
/// Parse LSTM box text format: "word x1 y1 x2 y2 confidence"
fn parse_lstm_boxes(box_text: &str) -> Vec<OcrWord> {
let mut words = Vec::new();
for line in box_text.lines() {
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.len() >= 5 {
if let (Ok(x), Ok(y), Ok(x2), Ok(y2)) = (
parts[1].parse::<i32>(),
parts[2].parse::<i32>(),
parts[3].parse::<i32>(),
parts[4].parse::<i32>(),
) {
let confidence = parts.get(5).and_then(|v| v.parse::<i32>().ok()).unwrap_or(0);
words.push(OcrWord {
text: parts[0].to_string(),
bbox: Bbox {
x,
y,
width: (x2 - x).max(1),
height: (y2 - y).max(1),
},
confidence,
});
}
}
} }
words
} }
/// Run OCR on a grayscale image, returning only the text. /// Non-tesseract fallback: return empty result.
pub fn ocr_text_only(img: &GrayImage, lang: &str) -> Result<String, PipelineError> { #[cfg(not(feature = "tesseract"))]
ocr_text(img, lang).map(|r| r.full_text) pub fn ocr_text(_img: &GrayImage, _lang: &str) -> Result<OcrResult, PipelineError> {
} tracing::warn!("Tesseract feature not enabled, OCR returning placeholder");
Ok(OcrResult {
/// Run OCR with word-level bounding boxes. full_text: String::new(),
pub fn ocr_words(img: &GrayImage, lang: &str) -> Result<Vec<OcrWord>, PipelineError> { words: Vec::new(),
ocr_text(img, lang).map(|r| r.words) confidence: 0.0,
})
} }
+72 -7
View File
@@ -1,7 +1,6 @@
use std::path::Path; use std::path::Path;
use std::time::Instant; use std::time::Instant;
use image::DynamicImage;
use tools_common::error::PipelineError; use tools_common::error::PipelineError;
use tools_common::types::Job; use tools_common::types::Job;
@@ -75,13 +74,11 @@ pub async fn process(
report(progress, "enhance", 90, "Mengoptimalkan kualitas...").await; report(progress, "enhance", 90, "Mengoptimalkan kualitas...").await;
let final_img = enhance_final(&final_img); let final_img = enhance_final(&final_img);
// Stage 9: Save output (93-100%) // Stage 9: OCR + PDF/PNG output (93-100%)
report(progress, "save", 95, "Menyimpan hasil...").await; report(progress, "save", 95, "Menyimpan hasil...").await;
let output_filename = format!("{}.png", progress.job_id()); let job_id = progress.job_id();
let output_path = output_dir.join(&output_filename); let (output_path, ocr_text) = generate_output(&final_img, &job_id, &output_dir, &job.options)?;
final_img.save(&output_path)?;
let elapsed = start.elapsed().as_millis() as u64; let elapsed = start.elapsed().as_millis() as u64;
@@ -95,11 +92,79 @@ pub async fn process(
output_path: output_path.to_string_lossy().to_string(), output_path: output_path.to_string_lossy().to_string(),
page_count: 1, page_count: 1,
file_size: tokio::fs::metadata(&output_path).await.map(|m| m.len()).unwrap_or(0), file_size: tokio::fs::metadata(&output_path).await.map(|m| m.len()).unwrap_or(0),
ocr_text: None, ocr_text,
processing_time_ms: elapsed, processing_time_ms: elapsed,
}) })
} }
/// Generate final output file: PDF with OCR text layer, or fallback to PNG.
fn generate_output(
final_img: &image::GrayImage,
job_id: &uuid::Uuid,
output_dir: &Path,
options: &serde_json::Value,
) -> Result<(std::path::PathBuf, Option<String>), Box<dyn std::error::Error + Send + Sync>> {
#[cfg(feature = "tesseract")]
{
let ocr_enabled = options
.get("ocr")
.and_then(|v| v.as_bool())
.unwrap_or(true);
if ocr_enabled {
let lang = options
.get("language")
.and_then(|v| v.as_str())
.unwrap_or("eng+ind");
match super::ocr::ocr_text(final_img, lang) {
Ok(ocr_result) => {
// Compress image as JPEG for PDF embedding
let output_filename = format!("{}.pdf", job_id);
let output_path = output_dir.join(&output_filename);
let jpeg_data = super::pdf::compress_image_jpeg(final_img, 85)
.map_err(|e| format!("JPEG compression failed: {}", e))?;
let page_width = super::pdf::A4_WIDTH_PT;
let page_height = super::pdf::A4_HEIGHT_PT;
let pdf_data = super::pdf::generate_searchable_pdf(
&jpeg_data,
&ocr_result.full_text,
&ocr_result.words,
page_width,
page_height,
)?;
std::fs::write(&output_path, &pdf_data)?;
let ocr_text = if ocr_result.full_text.is_empty() {
None
} else {
Some(ocr_result.full_text)
};
return Ok((output_path, ocr_text));
}
Err(e) => {
tracing::warn!("OCR failed, falling back to PNG: {}", e);
}
}
}
}
#[cfg(not(feature = "tesseract"))]
{
tracing::warn!("Tesseract feature not enabled, saving as PNG");
}
// Fallback: save as PNG
let output_filename = format!("{}.png", job_id);
let output_path = output_dir.join(&output_filename);
final_img.save(&output_path)?;
Ok((output_path, None))
}
/// Helper to report progress. /// Helper to report progress.
async fn report(progress: &ProgressReporter, stage: &str, pct: u8, msg: &str) { async fn report(progress: &ProgressReporter, stage: &str, pct: u8, msg: &str) {
let _ = progress let _ = progress
+32 -10
View File
@@ -15,10 +15,11 @@ export default function ImageCompressPage() {
const [pageState, setPageState] = useState<PageState>("upload"); const [pageState, setPageState] = useState<PageState>("upload");
const [jobId, setJobId] = useState<string | null>(null); const [jobId, setJobId] = useState<string | null>(null);
const [errorMsg, setErrorMsg] = useState<string | null>(null); const [errorMsg, setErrorMsg] = useState<string | null>(null);
const [quality, setQuality] = useState(80);
const { upload } = useUpload({ const { upload } = useUpload({
tool: "image-compress", tool: "image-compress",
options: { quality: 80 }, options: { quality },
}); });
const handleComplete = useCallback(() => setPageState("result"), []); const handleComplete = useCallback(() => setPageState("result"), []);
@@ -61,11 +62,35 @@ export default function ImageCompressPage() {
phase={1} phase={1}
> >
{pageState === "upload" && ( {pageState === "upload" && (
<UploadZone <div className="space-y-6">
accept="image/*" {/* Quality Slider */}
tool="image-compress" <div className="p-6 rounded-xl border glass space-y-3">
onUpload={handleUpload} <div className="flex items-center justify-between">
/> <span className="text-sm font-medium">Quality</span>
<span className="text-sm font-mono text-muted-foreground">
{quality}%
</span>
</div>
<input
type="range"
min={1}
max={100}
value={quality}
onChange={(e) => setQuality(Number(e.target.value))}
className="w-full"
/>
<div className="flex justify-between text-xs text-muted-foreground">
<span>Kecil</span>
<span>Besar</span>
</div>
</div>
<UploadZone
accept="image/*"
tool="image-compress"
onUpload={handleUpload}
/>
</div>
)} )}
{pageState === "processing" && jobId && ( {pageState === "processing" && jobId && (
@@ -79,10 +104,7 @@ export default function ImageCompressPage() {
)} )}
{pageState === "result" && result && ( {pageState === "result" && result && (
<ResultPreview <ResultPreview result={result} onProcessAnother={handleRetry} />
result={result}
onProcessAnother={handleRetry}
/>
)} )}
{pageState === "error" && ( {pageState === "error" && (
+93 -10
View File
@@ -10,14 +10,29 @@ import { useJobStatus } from "@/hooks/use-job-status";
type PageState = "upload" | "processing" | "result" | "error"; type PageState = "upload" | "processing" | "result" | "error";
interface ScanOptions {
ocr: boolean;
enhance: boolean;
output_format: "pdf" | "jpeg" | "png";
dpi: number;
color_mode: "black_and_white" | "grayscale" | "color";
}
export default function ScanPage() { export default function ScanPage() {
const [pageState, setPageState] = useState<PageState>("upload"); const [pageState, setPageState] = useState<PageState>("upload");
const [jobId, setJobId] = useState<string | null>(null); const [jobId, setJobId] = useState<string | null>(null);
const [errorMsg, setErrorMsg] = useState<string | null>(null); const [errorMsg, setErrorMsg] = useState<string | null>(null);
const [opts, setOpts] = useState<ScanOptions>({
ocr: true,
enhance: true,
output_format: "pdf",
dpi: 300,
color_mode: "black_and_white",
});
const { upload, isUploading } = useUpload({ const { upload, isUploading } = useUpload({
tool: "scan", tool: "scan",
options: { ocr: true, enhance: true, output_format: "pdf", dpi: 300 }, options: opts as unknown as Record<string, unknown>,
}); });
const handleComplete = useCallback(() => { const handleComplete = useCallback(() => {
@@ -75,15 +90,83 @@ export default function ScanPage() {
maxSizeMB={50} maxSizeMB={50}
/> />
{/* Options info */} {/* Interactive Options */}
<div className="p-4 rounded-lg border glass text-sm text-muted-foreground"> <div className="p-6 rounded-xl border glass space-y-4">
<p className="font-medium text-foreground mb-2">Scan Options</p> <h3 className="font-semibold">Scan Options</h3>
<ul className="space-y-1">
<li> OCR: Enabled (English + Indonesian)</li> {/* OCR Toggle */}
<li> Output: Searchable PDF</li> <label className="flex items-center justify-between">
<li> DPI: 300</li> <span className="text-sm">OCR (Tesseract)</span>
<li> Auto-enhance: On</li> <input
</ul> type="checkbox"
checked={opts.ocr}
onChange={(e) => setOpts({ ...opts, ocr: e.target.checked })}
className="toggle"
/>
</label>
{/* Enhance Toggle */}
<label className="flex items-center justify-between">
<span className="text-sm">Auto-enhance</span>
<input
type="checkbox"
checked={opts.enhance}
onChange={(e) => setOpts({ ...opts, enhance: e.target.checked })}
className="toggle"
/>
</label>
{/* Output Format */}
<label className="flex items-center justify-between">
<span className="text-sm">Output Format</span>
<select
value={opts.output_format}
onChange={(e) =>
setOpts({
...opts,
output_format: e.target.value as ScanOptions["output_format"],
})
}
className="bg-muted border rounded px-2 py-1 text-sm"
>
<option value="pdf">Searchable PDF</option>
<option value="jpeg">JPEG Image</option>
<option value="png">PNG Image</option>
</select>
</label>
{/* Color Mode */}
<label className="flex items-center justify-between">
<span className="text-sm">Color Mode</span>
<select
value={opts.color_mode}
onChange={(e) =>
setOpts({
...opts,
color_mode: e.target.value as ScanOptions["color_mode"],
})
}
className="bg-muted border rounded px-2 py-1 text-sm"
>
<option value="black_and_white">Black & White</option>
<option value="grayscale">Grayscale</option>
<option value="color">Color</option>
</select>
</label>
{/* DPI */}
<label className="flex items-center justify-between">
<span className="text-sm">DPI</span>
<select
value={opts.dpi}
onChange={(e) => setOpts({ ...opts, dpi: Number(e.target.value) })}
className="bg-muted border rounded px-2 py-1 text-sm"
>
<option value={150}>150 (draft)</option>
<option value={300}>300 (standard)</option>
<option value={600}>600 (high)</option>
</select>
</label>
</div> </div>
</div> </div>
)} )}
+8 -4
View File
@@ -24,6 +24,7 @@ interface JobProgress {
interface UseJobStatusOptions { interface UseJobStatusOptions {
onComplete?: (result: JobProgress["result"]) => void; onComplete?: (result: JobProgress["result"]) => void;
onError?: (error: string) => void; onError?: (error: string) => void;
wsBaseUrl?: string; // optional override, e.g. "wss://tools.asepharyana.my.id"
} }
export function useJobStatus(jobId: string | null, options?: UseJobStatusOptions) { export function useJobStatus(jobId: string | null, options?: UseJobStatusOptions) {
@@ -40,10 +41,13 @@ export function useJobStatus(jobId: string | null, options?: UseJobStatusOptions
const connect = useCallback(() => { const connect = useCallback(() => {
if (!jobId) return; if (!jobId) return;
// Connect directly to Rust gateway WebSocket (not via Next.js) // Determine WS base URL: from options, NEXT_PUBLIC_WS_URL, or infer from page location
const protocol = window.location.protocol === "https:" ? "wss:" : "ws:"; const wsBase = options?.wsBaseUrl
const host = "localhost:3001"; // Rust gateway - wss for production ?? process.env.NEXT_PUBLIC_WS_URL
const url = `${protocol}//${host}/api/job/${jobId}/ws`; ?? (typeof window !== "undefined"
? `${window.location.protocol === "https:" ? "wss:" : "ws:"}//${window.location.host}`
: "ws://localhost:3002");
const url = `${wsBase}/api/job/${jobId}/ws`;
const ws = new WebSocket(url); const ws = new WebSocket(url);
wsRef.current = ws; wsRef.current = ws;