"""Generation status update helper for content briefs."""
from datetime import datetime, timezone
from typing import Optional
from .storage import get_db
def _now() -> str:
"""Current UTC timestamp as ISO string."""
return datetime.now(timezone.utc).isoformat()
def update_generation_status(
brief_id: str,
status: str,
content_output: Optional[str] = None,
struggle_score: Optional[float] = None,
generation_job_id: Optional[str] = None,
) -> None:
"""
Update brief status during/after generation.
Args:
brief_id: The brief UUID
status: 'generating', 'completed', or 'failed'
content_output: Generated markdown content (for completed)
struggle_score: Calculated struggle score (for completed)
generation_job_id: Job ID tracking
"""
conn = get_db()
try:
cursor = conn.cursor()
# Build update fields
fields = ["status = ?", "updated_at = ?"]
values = [status, _now()]
if content_output is not None:
fields.append("content_output = ?")
values.append(content_output)
if struggle_score is not None:
fields.append("struggle_score = ?")
values.append(struggle_score)
if generation_job_id is not None:
fields.append("generation_job_id = ?")
values.append(generation_job_id)
# Add brief_id for WHERE clause
values.append(brief_id)
sql = f"UPDATE content_briefs_v2 SET {', '.join(fields)} WHERE id = ?"
cursor.execute(sql, values)
conn.commit()
finally:
conn.close()