Nurfog/openccb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove AI image generation functionality from CMS and expand Docker ignore rules.

Browse Source
This commit is contained in:
Juan Enrique Allende Cifuentes
2026-03-09 10:06:26 -03:00
parent bf3f06d831
commit 7406de9a1b
11 changed files with 146 additions and 778 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/cms-service/src/handlers.rs
+48 -397
View File
@@ -1309,381 +1309,6 @@ pub async fn generate_quiz(
Ok(Json(quiz_blocks))
}
#[derive(Deserialize)]
pub struct VideoAIRequest {
pub prompt: Option<String>,
pub width: Option<u32>,
pub height: Option<u32>,
}
pub async fn generate_image(
Org(org_ctx): Org,
claims: common::auth::Claims,
State(pool): State<PgPool>,
Path(id): Path<Uuid>,
Json(payload): Json<VideoAIRequest>,
) -> Result<Json<Lesson>, StatusCode> {
if let Some(prompt) = &payload.prompt {
tracing::info!("Received prompt for video generation: {}", prompt);
}
// 1. Fetch lesson
let _lesson =
sqlx::query_as::<_, Lesson>("SELECT * FROM lessons WHERE id = $1 AND organization_id = $2")
.bind(id)
.bind(org_ctx.id)
.fetch_one(&pool)
.await
.map_err(|_| StatusCode::NOT_FOUND)?;
// 2. Set status to queued
let updated_lesson = sqlx::query_as::<_, Lesson>(
"UPDATE lessons SET video_generation_status = 'queued', video_generation_error = NULL WHERE id = $1 RETURNING *",
)
.bind(id)
.fetch_one(&pool)
.await
.map_err(|e| {
tracing::error!("Database update failed (video queued): {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
log_action(
&pool,
org_ctx.id,
claims.sub,
"VIDEO_GENERATION_QUEUED",
"Lesson",
id,
json!({ "status": "queued" }),
)
.await;
// 3. Spawn background task
let pool_clone = pool.clone();
let prompt_to_task = payload.prompt.clone();
let user_id = claims.sub;
let width = payload.width;
let height = payload.height;
tokio::spawn(async move {
if let Err(e) = run_image_generation_task(pool_clone, id, prompt_to_task, Some(user_id), false, width, height).await {
tracing::error!("Image generation task failed for lesson {}: {}", id, e);
}
});
Ok(Json(updated_lesson))
}
pub async fn generate_course_image(
Org(org_ctx): Org,
claims: common::auth::Claims,
State(pool): State<PgPool>,
Path(id): Path<Uuid>,
Json(payload): Json<VideoAIRequest>,
) -> Result<Json<Course>, StatusCode> {
// 1. Fetch course
let _course =
sqlx::query_as::<_, Course>("SELECT * FROM courses WHERE id = $1 AND organization_id = $2")
.bind(id)
.bind(org_ctx.id)
.fetch_one(&pool)
.await
.map_err(|_| StatusCode::NOT_FOUND)?;
// 2. Set status to queued
let updated_course = sqlx::query_as::<_, Course>(
"UPDATE courses SET generation_status = 'queued', generation_error = NULL WHERE id = $1 RETURNING *",
)
.bind(id)
.fetch_one(&pool)
.await
.map_err(|e| {
tracing::error!("Database update failed (course image queued): {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
log_action(
&pool,
org_ctx.id,
claims.sub,
"COURSE_IMAGE_GENERATION_QUEUED",
"Course",
id,
json!({ "status": "queued" }),
)
.await;
// 3. Spawn background task
let pool_clone = pool.clone();
let prompt_to_task = payload.prompt.clone();
let user_id = claims.sub;
let width = payload.width;
let height = payload.height;
tokio::spawn(async move {
if let Err(e) = run_image_generation_task(pool_clone, id, prompt_to_task, Some(user_id), true, width, height).await {
tracing::error!("Image generation task failed for course {}: {}", id, e);
}
});
Ok(Json(updated_course))
}
pub async fn run_image_generation_task(
pool: PgPool,
id: Uuid,
custom_prompt: Option<String>,
user_id: Option<Uuid>,
is_course: bool,
width: Option<u32>,
height: Option<u32>
) -> Result<(), String> {
let (status_col, progress_col, error_col, table_name) = if is_course {
("generation_status", "generation_progress", "generation_error", "courses")
} else {
("video_generation_status", "generation_progress", "video_generation_error", "lessons")
};
// 1. Set status to processing ONLY if it's still queued (not cancelled/idle)
let query = format!(
"UPDATE {} SET {} = 'processing' WHERE id = $1 AND {} = 'queued'",
table_name, status_col, status_col
);
let rows_affected = sqlx::query(&query)
.bind(id)
.execute(&pool)
.await
.map_err(|e| format!("Update to processing failed: {}", e))?
.rows_affected();
if rows_affected == 0 {
tracing::info!("Task {} was cancelled or is already processing. Aborting background thread.", id);
return Ok(());
}
// 2. Call Local Video Bridge (Python) with Retry Logic
let client = reqwest::Client::new();
let bridge_base_url = std::env::var("LOCAL_VIDEO_BRIDGE_URL")
.unwrap_or_else(|_| "http://t-800:8080".to_string());
let bridge_url = format!("{}/generate", bridge_base_url);
// Fallback logic for prompt: Custom Prompt > Title
let final_prompt = match custom_prompt {
Some(p) if !p.is_empty() => p,
_ => {
let title: String = sqlx::query_scalar(&format!("SELECT title FROM {} WHERE id = $1", table_name))
.bind(id)
.fetch_one(&pool)
.await
.map_err(|e| format!("Failed to fetch fallback prompt: {}", e))?;
title
}
};
let database_url = std::env::var("BRIDGE_DATABASE_URL")
.unwrap_or_else(|_| std::env::var("DATABASE_URL").unwrap_or_default());
let mut payload = serde_json::json!({
"prompt": final_prompt,
"lesson_id": id.to_string(),
"database_url": database_url,
"table_name": table_name,
"progress_column": progress_col,
});
if let Some(w) = width {
payload["width"] = serde_json::json!(w);
}
if let Some(h) = height {
payload["height"] = serde_json::json!(h);
}
let mut retry_count = 0;
let max_retries = 3; // Initial quick retries
let long_retry_delay = tokio::time::Duration::from_secs(600); // 10 minutes
loop {
let response_result = client.post(&bridge_url)
.json(&payload)
.send()
.await;
match response_result {
Ok(response) => {
if response.status().is_success() {
let result: serde_json::Value = response.json().await
.map_err(|e| format!("Failed to parse video bridge response: {}", e))?;
let bridge_content_url = result["url"].as_str()
.ok_or_else(|| "Video bridge response missing URL".to_string())?;
// Break the loop and proceed to download
return process_image_download(&client, bridge_content_url, &pool, id, is_course, user_id, &final_prompt).await;
} else {
let err_text = response.text().await.unwrap_or_default();
let _ = sqlx::query(&format!("UPDATE {} SET {} = $1, {} = 'error' WHERE id = $2", table_name, error_col, status_col))
.bind(&err_text)
.bind(id)
.execute(&pool)
.await;
return Err(format!("Video bridge error: {}", err_text));
}
}
Err(e) => {
tracing::warn!("Failed to reach AI bridge at {}: {}. Retry {}/{}", bridge_url, e, retry_count + 1, max_retries);
// Update DB to show we are waiting/retrying
let wait_msg = format!("El servidor de IA (t-800) no responde. Reintentando automáticamente en 10 min... (Error: {})", e);
let _ = sqlx::query(&format!("UPDATE {} SET {} = $1, {} = 'queued' WHERE id = $2", table_name, error_col, status_col))
.bind(&wait_msg)
.bind(id)
.execute(&pool)
.await;
// Check if task was cancelled while we were about to sleep
let current_status: String = sqlx::query_scalar(&format!("SELECT {} FROM {} WHERE id = $1", status_col, table_name))
.bind(id)
.fetch_one(&pool)
.await
.unwrap_or_else(|_| "error".to_string());
if current_status == "idle" || current_status == "error" {
tracing::info!("Task {} was cancelled or errored during retry. Aborting.", id);
return Ok(());
}
retry_count += 1;
// Wait 10 minutes before next attempt
tokio::time::sleep(long_retry_delay).await;
// After sleep, check status AGAIN to see if user cancelled during the 10min sleep
let current_status: String = sqlx::query_scalar(&format!("SELECT {} FROM {} WHERE id = $1", status_col, table_name))
.bind(id)
.fetch_one(&pool)
.await
.unwrap_or_else(|_| "error".to_string());
if current_status != "queued" {
tracing::info!("Task {} status changed to {} during sleep. Aborting retry.", id, current_status);
return Ok(());
}
// Set back to processing to "lock" it again for this attempt
let rows_affected = sqlx::query(&format!("UPDATE {} SET {} = 'processing' WHERE id = $1 AND {} = 'queued'", table_name, status_col, status_col))
.bind(id)
.execute(&pool)
.await
.map_err(|e| e.to_string())?
.rows_affected();
if rows_affected == 0 {
return Ok(());
}
}
}
}
}
async fn process_image_download(
client: &reqwest::Client,
bridge_content_url: &str,
pool: &PgPool,
id: Uuid,
is_course: bool,
user_id: Option<Uuid>,
final_prompt: &str
) -> Result<(), String> {
// --- Download image and store as Asset ---
// 1. Download from bridge
let image_bytes = client.get(bridge_content_url)
.send()
.await
.map_err(|e| format!("Failed to download generated image: {}", e))?
.bytes()
.await
.map_err(|e| format!("Failed to read image bytes: {}", e))?;
// 2. Fetch context
let (org_id, course_id, instructor_id): (Uuid, Uuid, Uuid) = if is_course {
let c = sqlx::query_as::<sqlx::Postgres, (Uuid, Uuid, Uuid)>(
"SELECT organization_id, id as course_id, instructor_id FROM courses WHERE id = $1"
)
.bind(id)
.fetch_one(pool)
.await
.map_err(|e| format!("Failed to fetch course context: {}", e))?;
c
} else {
sqlx::query_as::<sqlx::Postgres, (Uuid, Uuid, Uuid)>(
"SELECT l.organization_id, m.course_id, c.instructor_id
FROM lessons l
JOIN modules m ON l.module_id = m.id
JOIN courses c ON m.course_id = c.id
WHERE l.id = $1"
)
.bind(id)
.fetch_one(pool)
.await
.map_err(|e| format!("Failed to fetch lesson context: {}", e))?
};
let final_user_id = user_id.unwrap_or(instructor_id);
// 3. Save file locally
let asset_id = Uuid::new_v4();
let storage_filename = format!("{}.png", asset_id);
let storage_path = format!("uploads/{}", storage_filename);
let _ = tokio::fs::create_dir_all("uploads").await;
tokio::fs::write(&storage_path, &image_bytes).await.map_err(|e| e.to_string())?;
let size_bytes = image_bytes.len() as i64;
let local_url = format!("/assets/{}", storage_filename);
// 4. Register in assets table
sqlx::query(
r#"
INSERT INTO assets (id, organization_id, uploaded_by, course_id, filename, storage_path, mimetype, size_bytes)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
"#,
)
.bind(asset_id)
.bind(org_id)
.bind(final_user_id)
.bind(course_id)
.bind(format!("AI: {}", final_prompt))
.bind(&storage_path)
.bind("image/png")
.bind(size_bytes)
.execute(pool)
.await
.map_err(|e| format!("Failed to register asset: {}", e))?;
// 5. Complete task updating entity with local URL - ONLY if not cancelled (idle)
if is_course {
sqlx::query(&format!(
"UPDATE courses SET generation_status = 'completed', course_image_url = $1 WHERE id = $2 AND generation_status = 'processing'"
))
.bind(local_url)
.bind(id)
.execute(pool)
.await
.map_err(|e| format!("Failed to complete course task: {}", e))?;
} else {
sqlx::query(&format!(
"UPDATE lessons SET video_generation_status = 'completed', content_url = $1, content_type = 'image' WHERE id = $2 AND video_generation_status = 'processing'"
))
.bind(local_url)
.bind(id)
.execute(pool)
.await
.map_err(|e| format!("Failed to complete lesson task: {}", e))?;
}
Ok(())
}
pub async fn get_lesson(
Org(org_ctx): Org,
State(pool): State<PgPool>,
@@ -3921,29 +3546,55 @@ RULES:
request = request.header("Authorization", auth_header);
}
let response = request.send().await.map_err(|e| {
tracing::error!("LLM request failed: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let response_result = request.send().await;
if !response.status().is_success() {
let err_body = response.text().await.unwrap_or_default();
tracing::error!("LLM API error: {}", err_body);
return Err(StatusCode::INTERNAL_SERVER_ERROR);
}
let llm_data: serde_json::Value = response.json().await.map_err(|e| {
tracing::error!("Failed to parse LLM JSON response: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
tracing::info!("LLM Response received successfully");
let mut content_str = llm_data["choices"][0]["message"]["content"]
.as_str()
.unwrap_or("{}")
.trim()
.to_string();
let mut content_str = match response_result {
Ok(response) if response.status().is_success() => {
let llm_data: serde_json::Value = response.json().await.map_err(|e| {
tracing::error!("Failed to parse LLM JSON response: {}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
tracing::info!("LLM Response received successfully");
llm_data["choices"][0]["message"]["content"]
.as_str()
.unwrap_or("{}")
.trim()
.to_string()
}
Ok(response) => {
let err_body = response.text().await.unwrap_or_default();
tracing::error!("LLM API error (generating course): {}", err_body);
tracing::warn!("Falling back to default course template.");
r#"{
"title": "Nueva Plantilla de Curso",
"description": "El servidor de IA no respondió. Esta es una plantilla básica que puedes editar directamente.",
"modules": [
{
"title": "Módulo 1",
"lessons": [
{ "title": "Lección 1", "content_type": "text" }
]
}
]
}"#.to_string()
}
Err(e) => {
tracing::error!("LLM request failed (generating course): {}", e);
tracing::warn!("Falling back to default course template due to unreachable AI.");
r#"{
"title": "Nueva Plantilla de Curso",
"description": "El servidor de IA no está disponible en este momento. Esta es una plantilla básica.",
"modules": [
{
"title": "Módulo 1",
"lessons": [
{ "title": "Lección 1", "content_type": "text" }
]
}
]
}"#.to_string()
}
};
tracing::info!("Extracted content string (length: {})", content_str.len());
services/cms-service/src/handlers/tasks.rs
+4 -85
View File
@@ -35,34 +35,6 @@ pub async fn get_background_tasks(
JOIN modules m ON l.module_id = m.id
JOIN courses c ON m.course_id = c.id
WHERE l.transcription_status IN ('queued', 'processing', 'failed')
UNION ALL
SELECT
l.id,
l.title,
c.title as course_title,
'lesson_image' as task_type,
l.video_generation_status as status,
l.generation_progress as progress,
l.updated_at
FROM lessons l
JOIN modules m ON l.module_id = m.id
JOIN courses c ON m.course_id = c.id
WHERE l.video_generation_status IN ('queued', 'processing', 'failed')
UNION ALL
SELECT
c.id,
c.title as title,
NULL as course_title,
'course_image' as task_type,
c.generation_status as status,
c.generation_progress as progress,
c.updated_at
FROM courses c
WHERE c.generation_status IN ('queued', 'processing', 'failed')
ORDER BY updated_at DESC
"#;
@@ -83,23 +55,15 @@ pub async fn get_background_tasks(
#[derive(sqlx::FromRow)]
struct LessonStatusRow {
transcription_status: Option<String>,
video_generation_status: Option<String>,
}
#[derive(sqlx::FromRow)]
struct CourseStatusRow {
generation_status: Option<String>,
}
pub async fn retry_task(
State(pool): State<PgPool>,
Path(id): Path<Uuid>,
) -> Result<StatusCode, (StatusCode, String)> {
// We need to know WHAT to retry.
// Since we don't have task_type in the URL yet, we'll try to find the lesson/course and its current failing status.
// Check lessons for transcription or image failures
let lesson = sqlx::query_as::<_, LessonStatusRow>("SELECT transcription_status, video_generation_status FROM lessons WHERE id = $1")
// Check lessons for transcription failures
let lesson = sqlx::query_as::<_, LessonStatusRow>("SELECT transcription_status FROM lessons WHERE id = $1")
.bind(id)
.fetch_optional(&pool)
.await
@@ -114,44 +78,6 @@ pub async fn retry_task(
});
return Ok(StatusCode::ACCEPTED);
}
if l.video_generation_status.as_deref() == Some("failed") {
tokio::spawn(async move {
// For image generation, we need the worker pool.
// Wait, run_image_generation_task is in handlers.rs but it requires WorkerPool (not easily available here without complex wiring or just spawning the handler)
// Actually, the simplest way is to just set it to 'queued' and let a background worker pick it up if there was one,
// but currently we spawn them directly.
// For now, let's call the same handler logic.
// I need to import run_image_generation_task
let _ = sqlx::query("UPDATE lessons SET video_generation_status = 'queued' WHERE id = $1").bind(id).execute(&pool_clone).await;
// Note: We are missing prompt/width/height here if we want to restart exactly.
// But generally retry means restart with same params.
// We'll need to fetch them.
// TODO: Implement full image retry in a future cleanup if needed,
// for now transcription is the priority as it's the one that "fails" most often.
// Image generation usually works or the bridge is down.
});
return Ok(StatusCode::ACCEPTED);
}
}
// Check courses
let course = sqlx::query_as::<_, CourseStatusRow>("SELECT generation_status FROM courses WHERE id = $1")
.bind(id)
.fetch_optional(&pool)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
if let Some(c) = course {
if c.generation_status.as_deref() == Some("error") || c.generation_status.as_deref() == Some("failed") {
let pool_clone = pool.clone();
tokio::spawn(async move {
let _ = sqlx::query("UPDATE courses SET generation_status = 'queued' WHERE id = $1").bind(id).execute(&pool_clone).await;
// Same as above, needs a worker to pick it up.
});
return Ok(StatusCode::ACCEPTED);
}
}
Ok(StatusCode::NOT_FOUND)
@@ -161,16 +87,9 @@ pub async fn cancel_task(
State(pool): State<PgPool>,
Path(id): Path<Uuid>,
) -> Result<StatusCode, (StatusCode, String)> {
// Try to cancel in both tables
// Only cancel transcription in lessons
let _ = sqlx::query(
"UPDATE lessons SET transcription_status = 'idle', video_generation_status = 'idle' WHERE id = $1"
)
.bind(id)
.execute(&pool)
.await;
let _ = sqlx::query(
"UPDATE courses SET generation_status = 'idle' WHERE id = $1"
"UPDATE lessons SET transcription_status = 'idle' WHERE id = $1"
)
.bind(id)
.execute(&pool)
services/cms-service/src/main.rs
-60
View File
@@ -74,65 +74,7 @@ async fn main() {
}
}
// Check for queued video generations
let queued_video_lessons: Vec<sqlx::types::Uuid> = match sqlx::query_scalar(
"SELECT id FROM lessons WHERE video_generation_status = 'queued' LIMIT 5",
)
.fetch_all(&worker_pool)
.await
{
Ok(ids) => ids,
Err(e) => {
tracing::error!("Failed to fetch queued video lessons: {}", e);
tokio::time::sleep(Duration::from_secs(10)).await;
continue;
}
};
for lesson_id in queued_video_lessons {
tracing::info!("Processing video generation for lesson: {}", lesson_id);
if let Err(e) =
handlers::run_image_generation_task(worker_pool.clone(), lesson_id, None, None, false, None, None).await
{
tracing::error!("Image generation task failed for lesson {}: {}", lesson_id, e);
let _ = sqlx::query(
"UPDATE lessons SET video_generation_status = 'failed' WHERE id = $1",
)
.bind(lesson_id)
.execute(&worker_pool)
.await;
}
}
// Check for queued course image generations
let queued_course_ids: Vec<sqlx::types::Uuid> = match sqlx::query_scalar(
"SELECT id FROM courses WHERE generation_status = 'queued' LIMIT 5",
)
.fetch_all(&worker_pool)
.await
{
Ok(ids) => ids,
Err(e) => {
tracing::error!("Failed to fetch queued course images: {}", e);
tokio::time::sleep(Duration::from_secs(10)).await;
continue;
}
};
for course_id in queued_course_ids {
tracing::info!("Processing image generation for course: {}", course_id);
if let Err(e) =
handlers::run_image_generation_task(worker_pool.clone(), course_id, None, None, true, None, None).await
{
tracing::error!("Course image generation failed for {}: {}", course_id, e);
let _ = sqlx::query(
"UPDATE courses SET generation_status = 'failed' WHERE id = $1",
)
.bind(course_id)
.execute(&worker_pool)
.await;
}
}
tokio::time::sleep(Duration::from_secs(5)).await;
}
@@ -205,8 +147,6 @@ async fn main() {
.route("/lessons/{id}/vtt", get(handlers::get_lesson_vtt))
.route("/lessons/{id}/summarize", post(handlers::summarize_lesson))
.route("/lessons/{id}/generate-quiz", post(handlers::generate_quiz))
.route("/lessons/{id}/generate-image", post(handlers::generate_image))
.route("/courses/{id}/generate-image", post(handlers::generate_course_image))
.route("/courses/generate", post(handlers::generate_course))
.route("/courses/{id}/export", get(handlers::export_course))
.route("/courses/import", post(handlers::import_course))