| 1 | from datetime import datetime |
| 2 | from datetime import timedelta |
| 3 | from difflib import SequenceMatcher |
| 4 | from typing import List |
| 5 | from typing import Optional |
| 6 | |
| 7 | from fastapi import HTTPException |
| 8 | from loguru import logger |
| 9 | from sqlalchemy import case |
| 10 | from sqlalchemy import func |
| 11 | from sqlalchemy.ext.asyncio import AsyncSession |
| 12 | from sqlalchemy.future import select |
| 13 | from sqlalchemy.orm import selectinload |
| 14 | |
| 15 | from app.ai_analyst.schema.ai_analyst import AlertWithReportResponse |
| 16 | from app.ai_analyst.schema.ai_analyst import CreateJobRequest |
| 17 | from app.ai_analyst.schema.ai_analyst import CreateJobResponse |
| 18 | from app.ai_analyst.schema.ai_analyst import IocResponse |
| 19 | from app.ai_analyst.schema.ai_analyst import IocReviewResponse |
| 20 | from app.ai_analyst.schema.ai_analyst import JobResponse |
| 21 | from app.ai_analyst.schema.ai_analyst import MyReviewResponse |
| 22 | from app.ai_analyst.schema.ai_analyst import PalaceConsolidationDuplicatePair |
| 23 | from app.ai_analyst.schema.ai_analyst import PalaceConsolidationLesson |
| 24 | from app.ai_analyst.schema.ai_analyst import PalaceConsolidationResponse |
| 25 | from app.ai_analyst.schema.ai_analyst import PalaceConsolidationRoomGroup |
| 26 | from app.ai_analyst.schema.ai_analyst import PalaceLessonResponse |
| 27 | from app.ai_analyst.schema.ai_analyst import QueuePalaceLessonRequest |
| 28 | from app.ai_analyst.schema.ai_analyst import QueuePalaceLessonResponse |
| 29 | from app.ai_analyst.schema.ai_analyst import ReportResponse |
| 30 | from app.ai_analyst.schema.ai_analyst import ReviewResponse |
| 31 | from app.ai_analyst.schema.ai_analyst import ReviewStatsIocAccuracy |
| 32 | from app.ai_analyst.schema.ai_analyst import ReviewStatsResponse |
| 33 | from app.ai_analyst.schema.ai_analyst import ReviewStatsTemplate |
| 34 | from app.ai_analyst.schema.ai_analyst import SubmitIocsRequest |
| 35 | from app.ai_analyst.schema.ai_analyst import SubmitIocsResponse |
| 36 | from app.ai_analyst.schema.ai_analyst import SubmitReportRequest |
| 37 | from app.ai_analyst.schema.ai_analyst import SubmitReportResponse |
| 38 | from app.ai_analyst.schema.ai_analyst import SubmitReviewRequest |
| 39 | from app.ai_analyst.schema.ai_analyst import SubmitReviewResponse |
| 40 | from app.ai_analyst.schema.ai_analyst import UpdateJobRequest |
| 41 | from app.ai_analyst.schema.ai_analyst import UpdateJobResponse |
| 42 | from app.db.universal_models import AiAnalystIoc |
| 43 | from app.db.universal_models import AiAnalystIocReview |
| 44 | from app.db.universal_models import AiAnalystJob |
| 45 | from app.db.universal_models import AiAnalystPalaceLesson |
| 46 | from app.db.universal_models import AiAnalystReport |
| 47 | from app.db.universal_models import AiAnalystReview |
| 48 | from app.incidents.models import Alert |
| 49 | |
| 50 | |
| 51 | def _job_to_response(job: AiAnalystJob) -> JobResponse: |
| 52 | return JobResponse( |
| 53 | id=job.id, |
| 54 | alert_id=job.alert_id, |
| 55 | customer_code=job.customer_code, |
| 56 | status=job.status, |
| 57 | alert_type=job.alert_type, |
| 58 | triggered_by=job.triggered_by, |
| 59 | template_used=job.template_used, |
| 60 | created_at=job.created_at, |
| 61 | started_at=job.started_at, |
| 62 | completed_at=job.completed_at, |
| 63 | error_message=job.error_message, |
| 64 | ) |
| 65 | |
| 66 | |
| 67 | def _report_to_response(report: AiAnalystReport) -> ReportResponse: |
| 68 | return ReportResponse( |
| 69 | id=report.id, |
| 70 | job_id=report.job_id, |
| 71 | alert_id=report.alert_id, |
| 72 | customer_code=report.customer_code, |
| 73 | severity_assessment=report.severity_assessment, |
| 74 | summary=report.summary, |
| 75 | report_markdown=report.report_markdown, |
| 76 | recommended_actions=report.recommended_actions, |
| 77 | created_at=report.created_at, |
| 78 | ) |
| 79 | |
| 80 | |
| 81 | def _ioc_to_response(ioc: AiAnalystIoc) -> IocResponse: |
| 82 | return IocResponse( |
| 83 | id=ioc.id, |
| 84 | report_id=ioc.report_id, |
| 85 | alert_id=ioc.alert_id, |
| 86 | customer_code=ioc.customer_code, |
| 87 | ioc_value=ioc.ioc_value, |
| 88 | ioc_type=ioc.ioc_type, |
| 89 | vt_verdict=ioc.vt_verdict, |
| 90 | vt_score=ioc.vt_score, |
| 91 | details=ioc.details, |
| 92 | created_at=ioc.created_at, |
| 93 | ) |
| 94 | |
| 95 | |
| 96 | # --- Job operations --- |
| 97 | |
| 98 | |
| 99 | async def create_job(request: CreateJobRequest, session: AsyncSession) -> CreateJobResponse: |
| 100 | logger.info(f"Creating AI analyst job {request.id} for alert {request.alert_id}") |
| 101 | |
| 102 | # Check if job already exists (may have been auto-created by an early status update) |
| 103 | existing = await session.get(AiAnalystJob, request.id) |
| 104 | if existing: |
| 105 | logger.info(f"Job {request.id} already exists, updating fields") |
| 106 | if request.alert_type is not None: |
| 107 | existing.alert_type = request.alert_type |
| 108 | if request.template_used is not None: |
| 109 | existing.template_used = request.template_used |
| 110 | existing.triggered_by = request.triggered_by.value |
| 111 | session.add(existing) |
| 112 | await session.commit() |
| 113 | await session.refresh(existing) |
| 114 | return CreateJobResponse(success=True, message="Job updated", job=_job_to_response(existing)) |
| 115 | |
| 116 | job = AiAnalystJob( |
| 117 | id=request.id, |
| 118 | alert_id=request.alert_id, |
| 119 | customer_code=request.customer_code, |
| 120 | status="pending", |
| 121 | alert_type=request.alert_type, |
| 122 | triggered_by=request.triggered_by.value, |
| 123 | template_used=request.template_used, |
| 124 | created_at=datetime.utcnow(), |
| 125 | ) |
| 126 | session.add(job) |
| 127 | await session.commit() |
| 128 | await session.refresh(job) |
| 129 | |
| 130 | logger.info(f"AI analyst job {job.id} created successfully") |
| 131 | return CreateJobResponse(success=True, message="Job created", job=_job_to_response(job)) |
| 132 | |
| 133 | |
| 134 | async def _auto_create_job_from_id(job_id: str, session: AsyncSession) -> Optional[AiAnalystJob]: |
| 135 | """ |
| 136 | Parse a job ID in the format 'copilot-inv-{alert_id}-{timestamp}' and |
| 137 | auto-create the job record by looking up the alert for customer_code. |
| 138 | Returns the created job, or None if parsing/lookup fails. |
| 139 | """ |
| 140 | import re |
| 141 | |
| 142 | match = re.match(r"^copilot-inv-(\d+)-\d+$", job_id) |
| 143 | if not match: |
| 144 | logger.warning(f"Cannot parse alert_id from job ID: {job_id}") |
| 145 | return None |
| 146 | |
| 147 | alert_id = int(match.group(1)) |
| 148 | alert = await session.get(Alert, alert_id) |
| 149 | if not alert: |
| 150 | logger.warning(f"Alert {alert_id} not found, cannot auto-create job {job_id}") |
| 151 | return None |
| 152 | |
| 153 | job = AiAnalystJob( |
| 154 | id=job_id, |
| 155 | alert_id=alert_id, |
| 156 | customer_code=alert.customer_code, |
| 157 | status="pending", |
| 158 | triggered_by="webhook", |
| 159 | created_at=datetime.utcnow(), |
| 160 | ) |
| 161 | session.add(job) |
| 162 | await session.commit() |
| 163 | await session.refresh(job) |
| 164 | logger.info(f"Auto-created AI analyst job {job_id} for alert {alert_id}") |
| 165 | return job |
| 166 | |
| 167 | |
| 168 | async def update_job(job_id: str, request: UpdateJobRequest, session: AsyncSession) -> UpdateJobResponse: |
| 169 | logger.info(f"Updating AI analyst job {job_id} to status {request.status}") |
| 170 | |
| 171 | job = await session.get(AiAnalystJob, job_id) |
| 172 | if not job: |
| 173 | # Auto-create the job if it doesn't exist yet (handles race condition |
| 174 | # where Talon sends a status update before the create request arrives) |
| 175 | logger.info(f"Job {job_id} not found, attempting to auto-create from job ID") |
| 176 | job = await _auto_create_job_from_id(job_id, session) |
| 177 | if not job: |
| 178 | raise HTTPException(status_code=404, detail=f"Job {job_id} not found and could not be auto-created") |
| 179 | |
| 180 | job.status = request.status.value |
| 181 | |
| 182 | if request.alert_type is not None: |
| 183 | job.alert_type = request.alert_type |
| 184 | if request.template_used is not None: |
| 185 | job.template_used = request.template_used |
| 186 | if request.error_message is not None: |
| 187 | job.error_message = request.error_message |
| 188 | |
| 189 | if request.status.value == "running" and job.started_at is None: |
| 190 | job.started_at = datetime.utcnow() |
| 191 | elif request.status.value in ("completed", "failed"): |
| 192 | job.completed_at = datetime.utcnow() |
| 193 | |
| 194 | session.add(job) |
| 195 | await session.commit() |
| 196 | await session.refresh(job) |
| 197 | |
| 198 | logger.info(f"AI analyst job {job_id} updated to {request.status}") |
| 199 | return UpdateJobResponse(success=True, message="Job updated", job=_job_to_response(job)) |
| 200 | |
| 201 | |
| 202 | async def get_job(job_id: str, session: AsyncSession) -> JobResponse: |
| 203 | job = await session.get(AiAnalystJob, job_id) |
| 204 | if not job: |
| 205 | raise HTTPException(status_code=404, detail=f"Job {job_id} not found") |
| 206 | return _job_to_response(job) |
| 207 | |
| 208 | |
| 209 | async def list_jobs_by_alert(alert_id: int, session: AsyncSession) -> List[JobResponse]: |
| 210 | result = await session.execute( |
| 211 | select(AiAnalystJob).where(AiAnalystJob.alert_id == alert_id).order_by(AiAnalystJob.created_at.desc()), |
| 212 | ) |
| 213 | jobs = result.scalars().all() |
| 214 | return [_job_to_response(j) for j in jobs] |
| 215 | |
| 216 | |
| 217 | async def list_jobs_by_customer(customer_code: str, session: AsyncSession) -> List[JobResponse]: |
| 218 | result = await session.execute( |
| 219 | select(AiAnalystJob).where(AiAnalystJob.customer_code == customer_code).order_by(AiAnalystJob.created_at.desc()), |
| 220 | ) |
| 221 | jobs = result.scalars().all() |
| 222 | return [_job_to_response(j) for j in jobs] |
| 223 | |
| 224 | |
| 225 | # --- Report operations --- |
| 226 | |
| 227 | |
| 228 | async def submit_report(request: SubmitReportRequest, session: AsyncSession) -> SubmitReportResponse: |
| 229 | logger.info(f"Submitting AI analyst report for job {request.job_id}, alert {request.alert_id}") |
| 230 | |
| 231 | # Verify the job exists |
| 232 | job = await session.get(AiAnalystJob, request.job_id) |
| 233 | if not job: |
| 234 | raise HTTPException(status_code=404, detail=f"Job {request.job_id} not found") |
| 235 | |
| 236 | report = AiAnalystReport( |
| 237 | job_id=request.job_id, |
| 238 | alert_id=request.alert_id, |
| 239 | customer_code=request.customer_code, |
| 240 | severity_assessment=request.severity_assessment.value if request.severity_assessment else None, |
| 241 | summary=request.summary, |
| 242 | report_markdown=request.report_markdown, |
| 243 | recommended_actions=request.recommended_actions, |
| 244 | created_at=datetime.utcnow(), |
| 245 | ) |
| 246 | session.add(report) |
| 247 | await session.commit() |
| 248 | await session.refresh(report) |
| 249 | |
| 250 | logger.info(f"AI analyst report {report.id} created for job {request.job_id}") |
| 251 | return SubmitReportResponse(success=True, message="Report submitted", report=_report_to_response(report)) |
| 252 | |
| 253 | |
| 254 | async def get_report_by_job(job_id: str, session: AsyncSession) -> Optional[ReportResponse]: |
| 255 | result = await session.execute( |
| 256 | select(AiAnalystReport).where(AiAnalystReport.job_id == job_id), |
| 257 | ) |
| 258 | report = result.scalars().first() |
| 259 | if not report: |
| 260 | return None |
| 261 | return _report_to_response(report) |
| 262 | |
| 263 | |
| 264 | async def list_reports_by_alert(alert_id: int, session: AsyncSession) -> List[ReportResponse]: |
| 265 | result = await session.execute( |
| 266 | select(AiAnalystReport).where(AiAnalystReport.alert_id == alert_id).order_by(AiAnalystReport.created_at.desc()), |
| 267 | ) |
| 268 | reports = result.scalars().all() |
| 269 | return [_report_to_response(r) for r in reports] |
| 270 | |
| 271 | |
| 272 | # --- IOC operations --- |
| 273 | |
| 274 | |
| 275 | async def submit_iocs(request: SubmitIocsRequest, session: AsyncSession) -> SubmitIocsResponse: |
| 276 | logger.info(f"Submitting {len(request.iocs)} IOCs for report {request.report_id}, alert {request.alert_id}") |
| 277 | |
| 278 | # Verify the report exists |
| 279 | report = await session.get(AiAnalystReport, request.report_id) |
| 280 | if not report: |
| 281 | raise HTTPException(status_code=404, detail=f"Report {request.report_id} not found") |
| 282 | |
| 283 | created_iocs = [] |
| 284 | for ioc_data in request.iocs: |
| 285 | ioc = AiAnalystIoc( |
| 286 | report_id=request.report_id, |
| 287 | alert_id=request.alert_id, |
| 288 | customer_code=request.customer_code, |
| 289 | ioc_value=ioc_data.ioc_value, |
| 290 | ioc_type=ioc_data.ioc_type.value, |
| 291 | vt_verdict=ioc_data.vt_verdict.value, |
| 292 | vt_score=ioc_data.vt_score, |
| 293 | details=ioc_data.details, |
| 294 | created_at=datetime.utcnow(), |
| 295 | ) |
| 296 | session.add(ioc) |
| 297 | created_iocs.append(ioc) |
| 298 | |
| 299 | await session.commit() |
| 300 | for ioc in created_iocs: |
| 301 | await session.refresh(ioc) |
| 302 | |
| 303 | logger.info(f"{len(created_iocs)} IOCs created for report {request.report_id}") |
| 304 | return SubmitIocsResponse( |
| 305 | success=True, |
| 306 | message=f"{len(created_iocs)} IOCs submitted", |
| 307 | iocs_created=len(created_iocs), |
| 308 | iocs=[_ioc_to_response(i) for i in created_iocs], |
| 309 | ) |
| 310 | |
| 311 | |
| 312 | async def list_iocs_by_report(report_id: int, session: AsyncSession) -> List[IocResponse]: |
| 313 | result = await session.execute( |
| 314 | select(AiAnalystIoc).where(AiAnalystIoc.report_id == report_id).order_by(AiAnalystIoc.created_at.desc()), |
| 315 | ) |
| 316 | iocs = result.scalars().all() |
| 317 | return [_ioc_to_response(i) for i in iocs] |
| 318 | |
| 319 | |
| 320 | async def list_iocs_by_alert(alert_id: int, session: AsyncSession) -> List[IocResponse]: |
| 321 | result = await session.execute( |
| 322 | select(AiAnalystIoc).where(AiAnalystIoc.alert_id == alert_id).order_by(AiAnalystIoc.created_at.desc()), |
| 323 | ) |
| 324 | iocs = result.scalars().all() |
| 325 | return [_ioc_to_response(i) for i in iocs] |
| 326 | |
| 327 | |
| 328 | async def list_iocs_by_customer( |
| 329 | customer_code: str, |
| 330 | session: AsyncSession, |
| 331 | vt_verdict: Optional[str] = None, |
| 332 | ) -> List[IocResponse]: |
| 333 | query = select(AiAnalystIoc).where(AiAnalystIoc.customer_code == customer_code) |
| 334 | if vt_verdict: |
| 335 | query = query.where(AiAnalystIoc.vt_verdict == vt_verdict) |
| 336 | query = query.order_by(AiAnalystIoc.created_at.desc()) |
| 337 | result = await session.execute(query) |
| 338 | iocs = result.scalars().all() |
| 339 | return [_ioc_to_response(i) for i in iocs] |
| 340 | |
| 341 | |
| 342 | # --- Alerts with reports --- |
| 343 | |
| 344 | |
| 345 | async def list_alerts_with_reports( |
| 346 | session: AsyncSession, |
| 347 | customer_code: Optional[str] = None, |
| 348 | ) -> List[AlertWithReportResponse]: |
| 349 | """Return all alerts that have at least one AI analyst report.""" |
| 350 | query = ( |
| 351 | select(Alert, AiAnalystReport) |
| 352 | .join(AiAnalystReport, AiAnalystReport.alert_id == Alert.id) |
| 353 | .order_by(AiAnalystReport.created_at.desc()) |
| 354 | ) |
| 355 | if customer_code: |
| 356 | query = query.where(Alert.customer_code == customer_code) |
| 357 | |
| 358 | result = await session.execute(query) |
| 359 | rows = result.all() |
| 360 | |
| 361 | return [ |
| 362 | AlertWithReportResponse( |
| 363 | alert_id=alert.id, |
| 364 | alert_name=alert.alert_name, |
| 365 | customer_code=alert.customer_code, |
| 366 | status=alert.status, |
| 367 | source=alert.source, |
| 368 | assigned_to=alert.assigned_to, |
| 369 | alert_creation_time=alert.alert_creation_time, |
| 370 | report=_report_to_response(report), |
| 371 | ) |
| 372 | for alert, report in rows |
| 373 | ] |
| 374 | |
| 375 | |
| 376 | # --- Combined alert analysis lookup --- |
| 377 | |
| 378 | |
| 379 | async def get_alert_analysis(alert_id: int, session: AsyncSession): |
| 380 | """Get the latest job, report, and IOCs for a given alert.""" |
| 381 | # Get latest job |
| 382 | job_result = await session.execute( |
| 383 | select(AiAnalystJob).where(AiAnalystJob.alert_id == alert_id).order_by(AiAnalystJob.created_at.desc()), |
| 384 | ) |
| 385 | job = job_result.scalars().first() |
| 386 | if not job: |
| 387 | return None, None, [] |
| 388 | |
| 389 | # Get report for this job |
| 390 | report_result = await session.execute( |
| 391 | select(AiAnalystReport).where(AiAnalystReport.job_id == job.id), |
| 392 | ) |
| 393 | report = report_result.scalars().first() |
| 394 | |
| 395 | # Get IOCs for this report |
| 396 | iocs = [] |
| 397 | if report: |
| 398 | ioc_result = await session.execute( |
| 399 | select(AiAnalystIoc).where(AiAnalystIoc.report_id == report.id), |
| 400 | ) |
| 401 | iocs = ioc_result.scalars().all() |
| 402 | |
| 403 | return ( |
| 404 | _job_to_response(job), |
| 405 | _report_to_response(report) if report else None, |
| 406 | [_ioc_to_response(i) for i in iocs], |
| 407 | ) |
| 408 | |
| 409 | |
| 410 | # --- Review / Palace lesson helpers --- |
| 411 | |
| 412 | |
| 413 | def _ioc_review_to_response(ir: AiAnalystIocReview) -> IocReviewResponse: |
| 414 | return IocReviewResponse( |
| 415 | id=ir.id, |
| 416 | review_id=ir.review_id, |
| 417 | ioc_id=ir.ioc_id, |
| 418 | verdict_correct=ir.verdict_correct, |
| 419 | note=ir.note, |
| 420 | created_at=ir.created_at, |
| 421 | ) |
| 422 | |
| 423 | |
| 424 | def _review_to_response(review: AiAnalystReview) -> ReviewResponse: |
| 425 | return ReviewResponse( |
| 426 | id=review.id, |
| 427 | report_id=review.report_id, |
| 428 | alert_id=review.alert_id, |
| 429 | customer_code=review.customer_code, |
| 430 | reviewer_user_id=review.reviewer_user_id, |
| 431 | overall_verdict=review.overall_verdict, |
| 432 | template_choice=review.template_choice, |
| 433 | template_used=review.template_used, |
| 434 | rating_instructions=review.rating_instructions, |
| 435 | rating_artifacts=review.rating_artifacts, |
| 436 | rating_severity=review.rating_severity, |
| 437 | missing_steps=review.missing_steps, |
| 438 | suggested_edits=review.suggested_edits, |
| 439 | created_at=review.created_at, |
| 440 | updated_at=review.updated_at, |
| 441 | ioc_reviews=[_ioc_review_to_response(ir) for ir in (review.ioc_reviews or [])], |
| 442 | ) |
| 443 | |
| 444 | |
| 445 | def _palace_lesson_to_response(lesson: AiAnalystPalaceLesson) -> PalaceLessonResponse: |
| 446 | return PalaceLessonResponse( |
| 447 | id=lesson.id, |
| 448 | review_id=lesson.review_id, |
| 449 | customer_code=lesson.customer_code, |
| 450 | lesson_type=lesson.lesson_type, |
| 451 | lesson_text=lesson.lesson_text, |
| 452 | durability=lesson.durability, |
| 453 | status=lesson.status, |
| 454 | ingested_at=lesson.ingested_at, |
| 455 | created_at=lesson.created_at, |
| 456 | ) |
| 457 | |
| 458 | |
| 459 | async def submit_review( |
| 460 | report_id: int, |
| 461 | request: SubmitReviewRequest, |
| 462 | reviewer_user_id: int, |
| 463 | session: AsyncSession, |
| 464 | ) -> SubmitReviewResponse: |
| 465 | """ |
| 466 | Upsert an analyst review of an AI investigation report. |
| 467 | |
| 468 | Enforces one review per (report_id, reviewer_user_id) pair. If the user |
| 469 | has already reviewed this report, their existing row is updated in place |
| 470 | (and `updated_at` is set) and their per-IOC corrections are replaced |
| 471 | wholesale. Otherwise a new row is inserted. |
| 472 | """ |
| 473 | logger.info(f"Submitting review for report {report_id} by user {reviewer_user_id}") |
| 474 | |
| 475 | report = await session.get(AiAnalystReport, report_id) |
| 476 | if not report: |
| 477 | raise HTTPException(status_code=404, detail=f"Report {report_id} not found") |
| 478 | |
| 479 | # If template_used wasn't supplied, inherit from the job for auditability |
| 480 | template_used = request.template_used |
| 481 | if template_used is None: |
| 482 | job = await session.get(AiAnalystJob, report.job_id) |
| 483 | if job is not None: |
| 484 | template_used = job.template_used |
| 485 | |
| 486 | # Validate every referenced IOC before mutating anything. A 400 here means |
| 487 | # we don't half-write and then bail. |
| 488 | for correction in request.ioc_reviews: |
| 489 | ioc = await session.get(AiAnalystIoc, correction.ioc_id) |
| 490 | if ioc is None or ioc.report_id != report_id: |
| 491 | raise HTTPException( |
| 492 | status_code=400, |
| 493 | detail=f"IOC {correction.ioc_id} not found or does not belong to report {report_id}", |
| 494 | ) |
| 495 | |
| 496 | # Look up existing review by the unique (report_id, reviewer_user_id) key |
| 497 | existing_result = await session.execute( |
| 498 | select(AiAnalystReview) |
| 499 | .where(AiAnalystReview.report_id == report_id) |
| 500 | .where(AiAnalystReview.reviewer_user_id == reviewer_user_id) |
| 501 | .options(selectinload(AiAnalystReview.ioc_reviews)), |
| 502 | ) |
| 503 | review = existing_result.scalars().first() |
| 504 | |
| 505 | is_edit = review is not None |
| 506 | |
| 507 | if is_edit: |
| 508 | # Update existing review in place |
| 509 | review.overall_verdict = request.overall_verdict.value if request.overall_verdict else None |
| 510 | review.template_choice = request.template_choice.value if request.template_choice else None |
| 511 | review.template_used = template_used |
| 512 | review.rating_instructions = request.rating_instructions |
| 513 | review.rating_artifacts = request.rating_artifacts |
| 514 | review.rating_severity = request.rating_severity |
| 515 | review.missing_steps = request.missing_steps |
| 516 | review.suggested_edits = request.suggested_edits |
| 517 | review.updated_at = datetime.utcnow() |
| 518 | session.add(review) |
| 519 | |
| 520 | # Replace per-IOC corrections wholesale — simpler than diffing and the |
| 521 | # edit UI always submits the full set anyway. |
| 522 | for old_ir in list(review.ioc_reviews or []): |
| 523 | await session.delete(old_ir) |
| 524 | await session.flush() |
| 525 | else: |
| 526 | review = AiAnalystReview( |
| 527 | report_id=report_id, |
| 528 | alert_id=report.alert_id, |
| 529 | customer_code=report.customer_code, |
| 530 | reviewer_user_id=reviewer_user_id, |
| 531 | overall_verdict=request.overall_verdict.value if request.overall_verdict else None, |
| 532 | template_choice=request.template_choice.value if request.template_choice else None, |
| 533 | template_used=template_used, |
| 534 | rating_instructions=request.rating_instructions, |
| 535 | rating_artifacts=request.rating_artifacts, |
| 536 | rating_severity=request.rating_severity, |
| 537 | missing_steps=request.missing_steps, |
| 538 | suggested_edits=request.suggested_edits, |
| 539 | created_at=datetime.utcnow(), |
| 540 | ) |
| 541 | session.add(review) |
| 542 | await session.flush() # get review.id before inserting child rows |
| 543 | |
| 544 | # Insert the new per-IOC corrections |
| 545 | for correction in request.ioc_reviews: |
| 546 | session.add( |
| 547 | AiAnalystIocReview( |
| 548 | review_id=review.id, |
| 549 | ioc_id=correction.ioc_id, |
| 550 | verdict_correct=correction.verdict_correct, |
| 551 | note=correction.note, |
| 552 | created_at=datetime.utcnow(), |
| 553 | ), |
| 554 | ) |
| 555 | |
| 556 | await session.commit() |
| 557 | |
| 558 | # Re-fetch with ioc_reviews eagerly loaded for the response |
| 559 | result = await session.execute( |
| 560 | select(AiAnalystReview).where(AiAnalystReview.id == review.id).options(selectinload(AiAnalystReview.ioc_reviews)), |
| 561 | ) |
| 562 | review_loaded = result.scalars().first() |
| 563 | |
| 564 | action = "updated" if is_edit else "created" |
| 565 | logger.info(f"Review {review.id} {action} for report {report_id}") |
| 566 | return SubmitReviewResponse( |
| 567 | success=True, |
| 568 | message=f"Review {action}", |
| 569 | review=_review_to_response(review_loaded), |
| 570 | ) |
| 571 | |
| 572 | |
| 573 | async def get_my_review( |
| 574 | report_id: int, |
| 575 | reviewer_user_id: int, |
| 576 | session: AsyncSession, |
| 577 | ) -> MyReviewResponse: |
| 578 | """ |
| 579 | Look up the current user's existing review for a report. |
| 580 | |
| 581 | Used by the UI to decide whether to render the rubric in 'create' mode or |
| 582 | 'edit existing' mode. Returns success=True with review=None when no review |
| 583 | exists yet (not an error). |
| 584 | """ |
| 585 | # Ensure the report itself exists so the UI gets a real 404 for bad ids |
| 586 | report = await session.get(AiAnalystReport, report_id) |
| 587 | if not report: |
| 588 | raise HTTPException(status_code=404, detail=f"Report {report_id} not found") |
| 589 | |
| 590 | result = await session.execute( |
| 591 | select(AiAnalystReview) |
| 592 | .where(AiAnalystReview.report_id == report_id) |
| 593 | .where(AiAnalystReview.reviewer_user_id == reviewer_user_id) |
| 594 | .options(selectinload(AiAnalystReview.ioc_reviews)), |
| 595 | ) |
| 596 | review = result.scalars().first() |
| 597 | |
| 598 | if review is None: |
| 599 | return MyReviewResponse( |
| 600 | success=True, |
| 601 | message="No existing review for this user on this report", |
| 602 | review=None, |
| 603 | ) |
| 604 | |
| 605 | return MyReviewResponse( |
| 606 | success=True, |
| 607 | message="Existing review retrieved", |
| 608 | review=_review_to_response(review), |
| 609 | ) |
| 610 | |
| 611 | |
| 612 | async def queue_palace_lesson( |
| 613 | request: QueuePalaceLessonRequest, |
| 614 | session: AsyncSession, |
| 615 | ) -> QueuePalaceLessonResponse: |
| 616 | """ |
| 617 | Queue a MemPalace lesson for async drainer pickup. Does NOT call Talon |
| 618 | directly — the drainer (roadmap item 17) reads status='pending' rows and |
| 619 | POSTs to NanoClaw's /palace/lesson endpoint. |
| 620 | """ |
| 621 | logger.info(f"Queuing palace lesson for {request.customer_code})") |
| 622 | |
| 623 | # If review_id supplied, validate it exists |
| 624 | if request.review_id is not None: |
| 625 | review = await session.get(AiAnalystReview, request.review_id) |
| 626 | if review is None: |
| 627 | raise HTTPException( |
| 628 | status_code=404, |
| 629 | detail=f"Review {request.review_id} not found", |
| 630 | ) |
| 631 | |
| 632 | lesson = AiAnalystPalaceLesson( |
| 633 | review_id=request.review_id, |
| 634 | customer_code=request.customer_code, |
| 635 | lesson_type=request.lesson_type.value, |
| 636 | lesson_text=request.lesson_text, |
| 637 | durability=request.durability.value, |
| 638 | status="pending", |
| 639 | created_at=datetime.utcnow(), |
| 640 | ) |
| 641 | session.add(lesson) |
| 642 | await session.commit() |
| 643 | await session.refresh(lesson) |
| 644 | |
| 645 | logger.info(f"Palace lesson {lesson.id} queued (status=pending)") |
| 646 | return QueuePalaceLessonResponse( |
| 647 | success=True, |
| 648 | message="Palace lesson queued for ingestion", |
| 649 | lesson=_palace_lesson_to_response(lesson), |
| 650 | ) |
| 651 | |
| 652 | |
| 653 | async def list_reviews_by_customer( |
| 654 | customer_code: str, |
| 655 | session: AsyncSession, |
| 656 | ) -> List[ReviewResponse]: |
| 657 | """Dashboard feed — reviews for a customer, newest first, with nested IOC reviews.""" |
| 658 | result = await session.execute( |
| 659 | select(AiAnalystReview) |
| 660 | .where(AiAnalystReview.customer_code == customer_code) |
| 661 | .options(selectinload(AiAnalystReview.ioc_reviews)) |
| 662 | .order_by(AiAnalystReview.created_at.desc()), |
| 663 | ) |
| 664 | reviews = result.scalars().all() |
| 665 | return [_review_to_response(r) for r in reviews] |
| 666 | |
| 667 | |
| 668 | def _pct(numerator: int, denominator: int) -> Optional[float]: |
| 669 | """Percentage helper — None when the denominator is 0 so the UI can |
| 670 | render a dash instead of a misleading '0%'.""" |
| 671 | if denominator <= 0: |
| 672 | return None |
| 673 | return round((numerator / denominator) * 100, 2) |
| 674 | |
| 675 | |
| 676 | def _round_float(v) -> Optional[float]: |
| 677 | """Avg helper — SQLAlchemy returns Decimal on some dialects; normalize to |
| 678 | a 2-decimal float for JSON. Returns None if the source was NULL (no rows).""" |
| 679 | if v is None: |
| 680 | return None |
| 681 | return round(float(v), 2) |
| 682 | |
| 683 | |
| 684 | async def get_review_stats( |
| 685 | customer_code: str, |
| 686 | session: AsyncSession, |
| 687 | recent_limit: int = 10, |
| 688 | ) -> ReviewStatsResponse: |
| 689 | """Aggregate feedback metrics for the customer's review dashboard. |
| 690 | |
| 691 | Uses SQL-side COUNT/AVG/CASE aggregates so this scales with review count |
| 692 | rather than pulling every row through Python — per the scale-first |
| 693 | design call on Step 20. |
| 694 | """ |
| 695 | # Main rollup: totals, verdict counts, template-choice counts, avg ratings. |
| 696 | main_q = select( |
| 697 | func.count(AiAnalystReview.id).label("total"), |
| 698 | func.sum(case((AiAnalystReview.overall_verdict == "up", 1), else_=0)).label("thumbs_up"), |
| 699 | func.sum(case((AiAnalystReview.overall_verdict == "down", 1), else_=0)).label("thumbs_down"), |
| 700 | func.sum(case((AiAnalystReview.template_choice == "correct", 1), else_=0)).label("tpl_correct"), |
| 701 | func.sum(case((AiAnalystReview.template_choice == "partial", 1), else_=0)).label("tpl_partial"), |
| 702 | func.sum(case((AiAnalystReview.template_choice == "wrong", 1), else_=0)).label("tpl_wrong"), |
| 703 | func.avg(AiAnalystReview.rating_instructions).label("avg_instr"), |
| 704 | func.avg(AiAnalystReview.rating_artifacts).label("avg_artifacts"), |
| 705 | func.avg(AiAnalystReview.rating_severity).label("avg_severity"), |
| 706 | ).where(AiAnalystReview.customer_code == customer_code) |
| 707 | main_row = (await session.execute(main_q)).one() |
| 708 | |
| 709 | total = int(main_row.total or 0) |
| 710 | thumbs_up = int(main_row.thumbs_up or 0) |
| 711 | thumbs_down = int(main_row.thumbs_down or 0) |
| 712 | # Non-null denominator for the up% gauge — reviews that actually picked a |
| 713 | # thumb. Skips reviews that left overall_verdict null. |
| 714 | verdict_total = thumbs_up + thumbs_down |
| 715 | |
| 716 | # Per-template rollup, grouped by template_used (which may be NULL). |
| 717 | per_template_q = ( |
| 718 | select( |
| 719 | AiAnalystReview.template_used.label("template_used"), |
| 720 | func.count(AiAnalystReview.id).label("total"), |
| 721 | func.sum(case((AiAnalystReview.overall_verdict == "up", 1), else_=0)).label("thumbs_up"), |
| 722 | func.sum(case((AiAnalystReview.overall_verdict == "down", 1), else_=0)).label("thumbs_down"), |
| 723 | func.sum(case((AiAnalystReview.template_choice == "correct", 1), else_=0)).label("correct"), |
| 724 | func.sum(case((AiAnalystReview.template_choice == "partial", 1), else_=0)).label("partial"), |
| 725 | func.sum(case((AiAnalystReview.template_choice == "wrong", 1), else_=0)).label("wrong"), |
| 726 | func.avg(AiAnalystReview.rating_instructions).label("avg_instr"), |
| 727 | func.avg(AiAnalystReview.rating_artifacts).label("avg_artifacts"), |
| 728 | func.avg(AiAnalystReview.rating_severity).label("avg_severity"), |
| 729 | ) |
| 730 | .where(AiAnalystReview.customer_code == customer_code) |
| 731 | .group_by(AiAnalystReview.template_used) |
| 732 | .order_by(func.count(AiAnalystReview.id).desc()) |
| 733 | ) |
| 734 | per_template_rows = (await session.execute(per_template_q)).all() |
| 735 | |
| 736 | # IOC verdict accuracy — join IocReview rows back to the customer's reviews |
| 737 | # so we only count corrections attached to this customer's reports. |
| 738 | ioc_q = ( |
| 739 | select( |
| 740 | func.count(AiAnalystIocReview.id).label("total"), |
| 741 | func.sum(case((AiAnalystIocReview.verdict_correct.is_(True), 1), else_=0)).label("correct"), |
| 742 | func.sum(case((AiAnalystIocReview.verdict_correct.is_(False), 1), else_=0)).label("incorrect"), |
| 743 | ) |
| 744 | .join(AiAnalystReview, AiAnalystReview.id == AiAnalystIocReview.review_id) |
| 745 | .where(AiAnalystReview.customer_code == customer_code) |
| 746 | ) |
| 747 | ioc_row = (await session.execute(ioc_q)).one() |
| 748 | ioc_total = int(ioc_row.total or 0) |
| 749 | ioc_correct = int(ioc_row.correct or 0) |
| 750 | ioc_incorrect = int(ioc_row.incorrect or 0) |
| 751 | |
| 752 | # Recent reviews (hydrated with ioc_reviews for drill-in). |
| 753 | recent_q = ( |
| 754 | select(AiAnalystReview) |
| 755 | .where(AiAnalystReview.customer_code == customer_code) |
| 756 | .options(selectinload(AiAnalystReview.ioc_reviews)) |
| 757 | .order_by(AiAnalystReview.created_at.desc()) |
| 758 | .limit(recent_limit) |
| 759 | ) |
| 760 | recent_reviews = (await session.execute(recent_q)).scalars().all() |
| 761 | |
| 762 | per_template: List[ReviewStatsTemplate] = [ |
| 763 | ReviewStatsTemplate( |
| 764 | template_used=row.template_used, |
| 765 | total=int(row.total or 0), |
| 766 | thumbs_up=int(row.thumbs_up or 0), |
| 767 | thumbs_down=int(row.thumbs_down or 0), |
| 768 | correct=int(row.correct or 0), |
| 769 | partial=int(row.partial or 0), |
| 770 | wrong=int(row.wrong or 0), |
| 771 | avg_rating_instructions=_round_float(row.avg_instr), |
| 772 | avg_rating_artifacts=_round_float(row.avg_artifacts), |
| 773 | avg_rating_severity=_round_float(row.avg_severity), |
| 774 | ) |
| 775 | for row in per_template_rows |
| 776 | ] |
| 777 | |
| 778 | return ReviewStatsResponse( |
| 779 | success=True, |
| 780 | message=f"Review stats for {customer_code}", |
| 781 | customer_code=customer_code, |
| 782 | total_reviews=total, |
| 783 | thumbs_up=thumbs_up, |
| 784 | thumbs_down=thumbs_down, |
| 785 | thumbs_up_pct=_pct(thumbs_up, verdict_total), |
| 786 | template_choice_correct=int(main_row.tpl_correct or 0), |
| 787 | template_choice_partial=int(main_row.tpl_partial or 0), |
| 788 | template_choice_wrong=int(main_row.tpl_wrong or 0), |
| 789 | avg_rating_instructions=_round_float(main_row.avg_instr), |
| 790 | avg_rating_artifacts=_round_float(main_row.avg_artifacts), |
| 791 | avg_rating_severity=_round_float(main_row.avg_severity), |
| 792 | ioc_accuracy=ReviewStatsIocAccuracy( |
| 793 | total=ioc_total, |
| 794 | correct=ioc_correct, |
| 795 | incorrect=ioc_incorrect, |
| 796 | accuracy_pct=_pct(ioc_correct, ioc_total), |
| 797 | ), |
| 798 | per_template=per_template, |
| 799 | recent_reviews=[_review_to_response(r) for r in recent_reviews], |
| 800 | ) |
| 801 | |
| 802 | |
| 803 | # --- Palace consolidation (Step 21.B) --- |
| 804 | |
| 805 | # Keep in sync with invoke_palace_lesson_sweeper.ONE_OFF_EXPIRY_DAYS — |
| 806 | # duplicated here rather than imported to avoid pulling a scheduler |
| 807 | # service into the synchronous route path at import time. |
| 808 | _ONE_OFF_EXPIRY_DAYS = 7 |
| 809 | # Lessons within this many days of expiring are surfaced to the reviewer |
| 810 | # as "about to be swept" — gives a window to promote to durable. |
| 811 | _EXPIRY_SOON_WINDOW_DAYS = 2 |
| 812 | # Similarity threshold for near-duplicate detection. 0.70 picks up |
| 813 | # paraphrases without flooding the reviewer with every shared phrase. |
| 814 | # difflib's SequenceMatcher on short strings is cheap — we can afford |
| 815 | # O(n²) pairs per room. |
| 816 | _DUPLICATE_SIMILARITY_THRESHOLD = 0.70 |
| 817 | |
| 818 | |
| 819 | def _lesson_to_consolidation( |
| 820 | lesson: AiAnalystPalaceLesson, |
| 821 | now: datetime, |
| 822 | ) -> PalaceConsolidationLesson: |
| 823 | days_until_expiry: Optional[int] = None |
| 824 | if lesson.durability == "one_off" and lesson.ingested_at is not None: |
| 825 | expiry = lesson.ingested_at + timedelta(days=_ONE_OFF_EXPIRY_DAYS) |
| 826 | days_until_expiry = (expiry - now).days |
| 827 | return PalaceConsolidationLesson( |
| 828 | id=lesson.id, |
| 829 | lesson_type=lesson.lesson_type, |
| 830 | lesson_text=lesson.lesson_text, |
| 831 | durability=lesson.durability, |
| 832 | status=lesson.status, |
| 833 | drawer_id=lesson.drawer_id, |
| 834 | created_at=lesson.created_at, |
| 835 | ingested_at=lesson.ingested_at, |
| 836 | days_until_expiry=days_until_expiry, |
| 837 | ) |
| 838 | |
| 839 | |
| 840 | def _find_duplicate_pairs( |
| 841 | lessons: List[PalaceConsolidationLesson], |
| 842 | ) -> List[PalaceConsolidationDuplicatePair]: |
| 843 | """Pairwise SequenceMatcher within the same room. Only returns |
| 844 | pairs above the threshold, sorted by similarity descending.""" |
| 845 | pairs: List[PalaceConsolidationDuplicatePair] = [] |
| 846 | # Group by room first so we only compare within-room. |
| 847 | by_room: dict[str, List[PalaceConsolidationLesson]] = {} |
| 848 | for lesson in lessons: |
| 849 | by_room.setdefault(lesson.lesson_type, []).append(lesson) |
| 850 | |
| 851 | for room, room_lessons in by_room.items(): |
| 852 | # Normalize once up-front so SequenceMatcher has stable inputs. |
| 853 | normalized = [(ls, ls.lesson_text.strip().lower()) for ls in room_lessons] |
| 854 | for i in range(len(normalized)): |
| 855 | a_lesson, a_text = normalized[i] |
| 856 | if not a_text: |
| 857 | continue |
| 858 | for j in range(i + 1, len(normalized)): |
| 859 | b_lesson, b_text = normalized[j] |
| 860 | if not b_text: |
| 861 | continue |
| 862 | ratio = SequenceMatcher(None, a_text, b_text).ratio() |
| 863 | if ratio >= _DUPLICATE_SIMILARITY_THRESHOLD: |
| 864 | pairs.append( |
| 865 | PalaceConsolidationDuplicatePair( |
| 866 | room=room, |
| 867 | lesson_a_id=a_lesson.id, |
| 868 | lesson_b_id=b_lesson.id, |
| 869 | lesson_a_text=a_lesson.lesson_text, |
| 870 | lesson_b_text=b_lesson.lesson_text, |
| 871 | similarity=round(ratio, 3), |
| 872 | ), |
| 873 | ) |
| 874 | pairs.sort(key=lambda p: p.similarity, reverse=True) |
| 875 | return pairs |
| 876 | |
| 877 | |
| 878 | def _render_consolidation_markdown( |
| 879 | customer_code: str, |
| 880 | generated_at: datetime, |
| 881 | total_lessons: int, |
| 882 | total_durable: int, |
| 883 | total_one_off: int, |
| 884 | rooms: List[PalaceConsolidationRoomGroup], |
| 885 | duplicates: List[PalaceConsolidationDuplicatePair], |
| 886 | upcoming: List[PalaceConsolidationLesson], |
| 887 | ) -> str: |
| 888 | """Pre-render the digest as markdown so the drawer can offer a |
| 889 | one-click copy/export. Kept intentionally terse — headings + bullets.""" |
| 890 | lines: List[str] = [] |
| 891 | lines.append(f"# Palace consolidation — {customer_code}") |
| 892 | lines.append(f"_Generated {generated_at.isoformat()} UTC_") |
| 893 | lines.append("") |
| 894 | lines.append("## Summary") |
| 895 | lines.append(f"- Total active lessons: **{total_lessons}**") |
| 896 | lines.append(f"- Durable: {total_durable} | One-off: {total_one_off}") |
| 897 | if upcoming: |
| 898 | lines.append( |
| 899 | f"- **{len(upcoming)} one-off lesson(s) expiring within " |
| 900 | f"{_EXPIRY_SOON_WINDOW_DAYS} day(s)** — consider promoting to durable.", |
| 901 | ) |
| 902 | if duplicates: |
| 903 | lines.append(f"- **{len(duplicates)} near-duplicate pair(s)** flagged for review.") |
| 904 | lines.append("") |
| 905 | |
| 906 | if upcoming: |
| 907 | lines.append("## Upcoming expirations") |
| 908 | for ls in upcoming: |
| 909 | due = ls.days_until_expiry if ls.days_until_expiry is not None else "?" |
| 910 | lines.append(f"- _{ls.lesson_type}_ (id {ls.id}, in {due}d): {ls.lesson_text}") |
| 911 | lines.append("") |
| 912 | |
| 913 | if duplicates: |
| 914 | lines.append("## Near-duplicate candidates") |
| 915 | for pair in duplicates: |
| 916 | pct = int(pair.similarity * 100) |
| 917 | lines.append(f"- **{pair.room}** — {pct}% similar") |
| 918 | lines.append(f" - #{pair.lesson_a_id}: {pair.lesson_a_text}") |
| 919 | lines.append(f" - #{pair.lesson_b_id}: {pair.lesson_b_text}") |
| 920 | lines.append("") |
| 921 | |
| 922 | lines.append("## Rooms") |
| 923 | for group in rooms: |
| 924 | lines.append( |
| 925 | f"### {group.room} ({group.total} total — " f"{group.durable} durable, {group.one_off} one-off)", |
| 926 | ) |
| 927 | for ls in group.lessons: |
| 928 | tag = "🧷" if ls.durability == "durable" else "⏳" |
| 929 | suffix = "" |
| 930 | if ls.durability == "one_off" and ls.days_until_expiry is not None: |
| 931 | suffix = f" _(expires in {ls.days_until_expiry}d)_" |
| 932 | lines.append(f"- {tag} #{ls.id}{suffix}: {ls.lesson_text}") |
| 933 | lines.append("") |
| 934 | |
| 935 | return "\n".join(lines).rstrip() + "\n" |
| 936 | |
| 937 | |
| 938 | async def get_palace_consolidation( |
| 939 | customer_code: str, |
| 940 | session: AsyncSession, |
| 941 | ) -> PalaceConsolidationResponse: |
| 942 | """Build a point-in-time digest of a customer's active MemPalace |
| 943 | lessons. Pure read-only, pure Python — no Talon round-trip. Used by |
| 944 | the manual "Consolidate Lessons" button in the Feedback dashboard.""" |
| 945 | logger.info(f"Building palace consolidation digest for customer {customer_code}") |
| 946 | |
| 947 | # Exclude expired (already swept) and failed (never reached the palace) |
| 948 | # rows — consolidation is about what's actually live right now. |
| 949 | stmt = ( |
| 950 | select(AiAnalystPalaceLesson) |
| 951 | .where(AiAnalystPalaceLesson.customer_code == customer_code) |
| 952 | .where(AiAnalystPalaceLesson.status.in_(["pending", "ingested"])) |
| 953 | .order_by( |
| 954 | AiAnalystPalaceLesson.lesson_type.asc(), |
| 955 | AiAnalystPalaceLesson.created_at.desc(), |
| 956 | ) |
| 957 | ) |
| 958 | result = await session.execute(stmt) |
| 959 | raw_lessons = result.scalars().all() |
| 960 | |
| 961 | now = datetime.utcnow() |
| 962 | lessons = [_lesson_to_consolidation(ls, now) for ls in raw_lessons] |
| 963 | |
| 964 | total_lessons = len(lessons) |
| 965 | total_durable = sum(1 for ls in lessons if ls.durability == "durable") |
| 966 | total_one_off = sum(1 for ls in lessons if ls.durability == "one_off") |
| 967 | total_pending = sum(1 for ls in lessons if ls.status == "pending") |
| 968 | total_ingested = sum(1 for ls in lessons if ls.status == "ingested") |
| 969 | |
| 970 | upcoming = sorted( |
| 971 | [ |
| 972 | ls |
| 973 | for ls in lessons |
| 974 | if ls.durability == "one_off" and ls.days_until_expiry is not None and ls.days_until_expiry <= _EXPIRY_SOON_WINDOW_DAYS |
| 975 | ], |
| 976 | key=lambda ls: ls.days_until_expiry if ls.days_until_expiry is not None else 0, |
| 977 | ) |
| 978 | |
| 979 | # Build per-room groups in deterministic room order. |
| 980 | by_room: dict[str, List[PalaceConsolidationLesson]] = {} |
| 981 | for ls in lessons: |
| 982 | by_room.setdefault(ls.lesson_type, []).append(ls) |
| 983 | rooms: List[PalaceConsolidationRoomGroup] = [] |
| 984 | for room in sorted(by_room.keys()): |
| 985 | room_lessons = by_room[room] |
| 986 | rooms.append( |
| 987 | PalaceConsolidationRoomGroup( |
| 988 | room=room, |
| 989 | total=len(room_lessons), |
| 990 | durable=sum(1 for ls in room_lessons if ls.durability == "durable"), |
| 991 | one_off=sum(1 for ls in room_lessons if ls.durability == "one_off"), |
| 992 | lessons=room_lessons, |
| 993 | ), |
| 994 | ) |
| 995 | |
| 996 | duplicates = _find_duplicate_pairs(lessons) |
| 997 | |
| 998 | markdown = _render_consolidation_markdown( |
| 999 | customer_code=customer_code, |
| 1000 | generated_at=now, |
| 1001 | total_lessons=total_lessons, |
| 1002 | total_durable=total_durable, |
| 1003 | total_one_off=total_one_off, |
| 1004 | rooms=rooms, |
| 1005 | duplicates=duplicates, |
| 1006 | upcoming=upcoming, |
| 1007 | ) |
| 1008 | |
| 1009 | return PalaceConsolidationResponse( |
| 1010 | success=True, |
| 1011 | message=( |
| 1012 | f"Palace consolidation for {customer_code}: " |
| 1013 | f"{total_lessons} active lesson(s), " |
| 1014 | f"{len(duplicates)} duplicate pair(s), " |
| 1015 | f"{len(upcoming)} expiring soon" |
| 1016 | ), |
| 1017 | customer_code=customer_code, |
| 1018 | generated_at=now, |
| 1019 | total_lessons=total_lessons, |
| 1020 | total_durable=total_durable, |
| 1021 | total_one_off=total_one_off, |
| 1022 | total_pending=total_pending, |
| 1023 | total_ingested=total_ingested, |
| 1024 | upcoming_expirations=upcoming, |
| 1025 | rooms=rooms, |
| 1026 | duplicate_candidates=duplicates, |
| 1027 | markdown=markdown, |
| 1028 | ) |