"""HTTP surface for security awareness (安全宣贯). Two prefixes, so the auth gate is readable at the route layer: * ``/api/v1/security-awareness`` — the employee learning centre. ANY authenticated user. Published materials only. * ``/api/v1/admin/security-awareness`` — the material library console. Reads gate on ``can_view_security`` (so auditor / management can see what the workforce is being told without being able to change it), writes on ``can_manage_security`` (global_admin / auditor / security_officer — running awareness IS the security desk's job). **Language is a query parameter, not a guess.** The frontend passes its active i18next language; absent that we fall back to ``Accept-Language`` and then to the baseline. One decision, made server-side, so the block that renders is the block the impression ledger records. """ from __future__ import annotations import csv import io import uuid from fastapi import APIRouter, Depends, Header, Query, Request, status from fastapi.responses import StreamingResponse from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUser from app.core.coordinators.awareness_points_coordinator import ( award_material_quiz_points, ) from app.core.database import get_db from app.core.dependencies import get_current_user from app.core.exceptions import ForbiddenError from app.core.rbac import can_manage_security, can_view_security from app.core.response import success_response from app.domains.security_awareness.ambient import ( SETTING_DAILY_NEW, SETTING_ENABLED, SETTING_MAX_REPEATS, AmbientService, ) from app.domains.security_awareness.ambient import ( load_settings as load_bubble_settings, ) from app.domains.security_awareness.campaign_service import CampaignService from app.domains.security_awareness.catalog import ( ACK_WEB, MATERIAL_CATEGORIES, MATERIAL_KINDS, MATERIAL_LANGS, MATERIAL_STATUSES, STATUS_DRAFT, SURFACE_DETAIL, ) from app.domains.security_awareness.compliance import ( EVIDENCE_COLUMNS, coverage_report, evidence_rows, participation_report, ) from app.domains.security_awareness.quiz import MaterialQuizService from app.domains.security_awareness.schemas import ( BubbleSettingsUpdate, ImpressionCreate, MaterialCreate, MaterialReorder, MaterialStatusUpdate, MaterialTranslateRequest, MaterialUpdate, QuizQuestionSet, QuizSubmission, TranslateMissingRequest, TranslationReviewed, ) from app.domains.security_awareness.service import ( ADMIN_MAX_PAGE_SIZE, SecurityAwarenessService, normalize_lang, ) from app.domains.security_awareness.translation import ( BACKFILL_STATUSES, MaterialTranslator, ) from app.integrations.service import PlatformSettingService # ════════════════════════════════════════════════════════════════════ # Employee surface # ════════════════════════════════════════════════════════════════════ self_router = APIRouter( prefix="/api/v1/security-awareness", tags=["security-awareness"], ) def _lang(explicit: str | None, header: str | None) -> str: """Helix's own language switcher wins over the browser header — it is what the reader actually chose, and it is what they are looking at.""" return normalize_lang(explicit or header) @self_router.get("/facets", summary="Filter options for the learning centre") async def learning_facets( request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Categories and tags that actually occur in the published library. Derived from the **whole** published set, never from a filtered result: filter chips computed from the rows a filter already returned disappear as soon as you use them, which reads as the app losing options. Categories are returned as keys (the frontend renders ``t('securityAwareness.category.')``) plus a live count, so an empty category can be hidden rather than offered as a dead end. """ svc = SecurityAwarenessService(db) data = await svc.facets( person_id=user.id, lang=_lang(lang, accept_language), ) return success_response(data=data, request=request) @self_router.get("/materials", summary="Employee learning centre") async def list_materials( request: Request, category: str | None = Query(default=None, max_length=32), tag: str | None = Query(default=None, max_length=50), q: str | None = Query(default=None, max_length=200), lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): svc = SecurityAwarenessService(db) rows = await svc.list_library( person_id=user.id, lang=_lang(lang, accept_language), category=category, tag=tag, q=q, ) return success_response(data=rows, request=request) @self_router.get("/materials/{material_id}", summary="Read one material") async def get_material( material_id: uuid.UUID, request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Hydrate one material AND record the exposure. The write lives here rather than in a client-issued follow-up call on purpose: "did the workforce see this" is a compliance claim, and a claim that depends on the browser remembering to report is a claim that quietly under-counts. The ledger write is idempotent inside the dedupe window, so a reader refreshing the page ten times still produces one exposure — which is what makes a state-changing GET acceptable here (platform rule: every state change is idempotent). """ resolved = _lang(lang, accept_language) svc = SecurityAwarenessService(db) campaigns = CampaignService(db) # Resolved BEFORE the read, because it decides whether the read is # allowed at all. A material that has been retired from the library # (archived, or pushed back to draft) 404s for everyone — except the # people a live campaign is still asking to read it. Without this, # archiving a material mid-campaign strands every straggler: their # todo list still names it, the reminder cron keeps chasing them # about it, and the link goes to a 404. Retiring content is a # decision about the LIBRARY; it does not retract an obligation # somebody was already given. (Retracting it is what closing or # cancelling the campaign does — and that also removes this row, # so the material goes back to 404 for them too.) pending = await campaigns.pending_for_material(user.id, material_id) detail = await svc.get_material_detail( person_id=user.id, material_id=material_id, lang=resolved, include_unpublished=pending is not None, ) await svc.record_impression( person_id=user.id, material_id=material_id, surface=SURFACE_DETAIL, lang=resolved, ) # Campaign accounting, when this material was pushed to them. # ``mark_opened`` is first-open-only, and ``pending_ack`` is what the # page renders its "我已阅读并理解" control from — the acknowledgement # lives ON the material, not on a campaign-shaped intermediate page # that would sit between the push and the content. await campaigns.mark_opened(person_id=user.id, material_id=material_id) detail["pending_ack"] = pending # When a material carries a quiz, the quiz REPLACES the confirm # button — two controls for one claim would let somebody assert # understanding without demonstrating it. The page needs to know # which control to render before it renders anything. quiz = MaterialQuizService(db) detail["has_quiz"] = await quiz.has_quiz(material_id) detail["quiz_passed"] = ( await quiz.has_passed(user.id, material_id) if detail["has_quiz"] else False ) return success_response(data=detail, request=request) @self_router.get("/bubble", summary="Ambient surface feed") async def bubble_feed( request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """What THIS person's ambient bubble should rotate through. Rationed per person (see ``ambient.py``): a small set, stable within the day, capped both on how many new materials somebody meets per day and on how many times one material may be shown to them. Anything they have already opened is dropped. Reading this endpoint records NOTHING — the client reports what it actually rendered through ``POST /impressions``. That split is deliberate and is the opposite of the material-detail route: there, a human opening the page IS the event; here, the browser fetching a rotation list is not evidence that any of it reached anyone's eyes. An empty ``items`` with ``enabled: true`` is a normal state — this person is up to date — and the client falls back to the legacy tips rather than showing an empty bubble. """ data = await AmbientService(db).bubble_feed( person_id=user.id, lang=_lang(lang, accept_language), ) return success_response(data=data, request=request) @self_router.get( "/materials/{material_id}/quiz", summary="This material's quiz", ) async def get_quiz( material_id: uuid.UUID, request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The questions, WITHOUT the answer key. Shipping ``correct_key`` to the browser would make the whole exercise theatre, and it is the kind of thing only noticed after somebody reads the network tab. The key is revealed per question, on a miss, by the submit response. """ svc = MaterialQuizService(db) # Published only — this route is addressable on its own, so it cannot # inherit the library's publication check from the page that links to # it. A draft's questions are unreleased content. The exception is a # reader who owes an acknowledgement on a live campaign for it: # archiving the material does not retract that. pending = await CampaignService(db).pending_for_material( user.id, material_id, ) await svc.assert_visible(material_id, allow_retired=pending is not None) return success_response( data={ # Same language contract as the material itself: an English # article that asked its quiz in Chinese would be a worse bug # than the one this domain's translation backfill fixed. "questions": await svc.questions_for( material_id, lang=_lang(lang, accept_language), ), "passed": await svc.has_passed(user.id, material_id), }, request=request, ) @self_router.post( "/materials/{material_id}/quiz", summary="Answer this material's quiz", ) async def submit_quiz( material_id: uuid.UUID, body: QuizSubmission, request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Grade an attempt; on a pass, that IS the acknowledgement. When a material carries a quiz the confirm button is replaced by it — two buttons for one claim would let somebody assert understanding without demonstrating it, which is the gap the quiz exists to close. So a passing attempt acknowledges every pending campaign for this material, and only then awards the (small, one-off) points. Retries are unlimited and a miss returns the explanation. The bar is "all correct" rather than a percentage: two or three questions written for one short piece do not support a meaningful partial score, and "2 of 3 counts" is a weaker claim than the plain acknowledgement it replaces. """ svc = MaterialQuizService(db) campaigns = CampaignService(db) # Resolved first: it is what allows a straggler on a retired material # to still answer, and it is the set the pass settles below. pending = await campaigns.pending_all_for_material(user.id, material_id) result = await svc.submit( person_id=user.id, material_id=material_id, answers=body.answers, allow_retired=bool(pending), # Only decides the language of the explanation a miss returns — # grading is on option keys, which every language shares. lang=_lang(lang, accept_language), ) result["acknowledged_campaign_id"] = None result["acknowledged_campaign_ids"] = [] result["points_awarded"] = 0 if result["passed"]: # EVERY pending campaign for this material, not just the most # urgent. One passing attempt is one demonstration of # understanding of one piece of content; settling only the first # would strand the rest, because the quiz component will not # re-offer a quiz somebody has already passed and so there would # be no control left anywhere to clear them with. for owed in pending: ack = await campaigns.acknowledge( person_id=user.id, campaign_id=uuid.UUID(owed["campaign_id"]), source=ACK_WEB, trace_id=getattr(request.state, "trace_id", ""), # This IS the demonstration the gate asks for — the # attempt was graded a pass a few lines ago. quiz_verified=True, ) result["acknowledged_campaign_ids"].append(owed["campaign_id"]) # Singular kept for the clients already reading it; it names # the most urgent one, which is the one the banner showed. if result["acknowledged_campaign_id"] is None: result["acknowledged_campaign_id"] = owed["campaign_id"] result["acknowledged_at"] = ack["acknowledged_at"] # Best-effort and LAST: a wallet hiccup must not cost somebody the # compliance record they just earned. result["points_awarded"] = await award_material_quiz_points( db, person_id=user.id, material_id=material_id, ) return success_response(data=result, request=request) @self_router.post( "/impressions", status_code=status.HTTP_202_ACCEPTED, summary="Report an ambient exposure", ) async def report_impression( body: ImpressionCreate, request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """For surfaces the server cannot observe — the ambient mascot bubble is the reason this exists. The caller says *what it rendered*; the server decides whether that is a new exposure and stamps the rest. Always 202: whether the row was written or collapsed into the dedupe window is not the client's business, and telling it would invite a retry loop. """ svc = SecurityAwarenessService(db) written = await svc.record_impression( person_id=user.id, material_id=body.material_id, surface=body.surface, lang=_lang(lang, accept_language), ) return success_response(data={"recorded": written}, request=request) # ════════════════════════════════════════════════════════════════════ # Admin surface # ════════════════════════════════════════════════════════════════════ admin_router = APIRouter( prefix="/api/v1/admin/security-awareness", tags=["security-awareness-admin"], ) def _require_read(user: AuthUser) -> None: if not can_view_security(user.role_names): raise ForbiddenError("security awareness console requires security read") def _require_write(user: AuthUser) -> None: if not can_manage_security(user.role_names): raise ForbiddenError("security awareness changes require security write") @admin_router.get("/meta", summary="Catalogs + the caller's own authority") async def admin_meta( request: Request, user: AuthUser = Depends(get_current_user), ): """Closed-list catalogs plus ``can_manage``. ``can_manage`` is what the page uses to decide whether to render the write affordances at all — an auditor opening the console should see the library, not a row of buttons that 403. The gate is still enforced per-endpoint; this only stops the UI from lying. """ _require_read(user) return success_response( data={ "kinds": sorted(MATERIAL_KINDS), "categories": list(MATERIAL_CATEGORIES), "statuses": sorted(MATERIAL_STATUSES), "langs": list(MATERIAL_LANGS), "can_manage": can_manage_security(user.role_names), }, request=request, ) @admin_router.get("/materials", summary="Material library console") async def admin_list_materials( request: Request, status_filter: str | None = Query(default=None, alias="status", max_length=16), category: str | None = Query(default=None, max_length=32), kind: str | None = Query(default=None, max_length=16), q: str | None = Query(default=None, max_length=200), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=ADMIN_MAX_PAGE_SIZE), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_read(user) svc = SecurityAwarenessService(db) rows, total = await svc.list_admin( status=status_filter, category=category, kind=kind, q=q, page=page, page_size=page_size, ) return success_response( data=rows, request=request, meta={"total": total, "page": page, "page_size": page_size}, ) @admin_router.get( "/materials/{material_id}/preview", summary="See a draft the way an employee will", ) async def admin_preview_material( material_id: uuid.UUID, request: Request, lang: str | None = Query(default=None, max_length=16), accept_language: str | None = Header(default=None, alias="Accept-Language"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The employee renderer, pointed at an unpublished row. Deliberately the SAME service method the employee route calls (``include_unpublished=True``) rather than a parallel preview implementation — a preview that renders through different code is a preview that eventually disagrees with production. """ _require_read(user) svc = SecurityAwarenessService(db) detail = await svc.get_material_detail( person_id=user.id, material_id=material_id, lang=_lang(lang, accept_language), include_unpublished=True, ) return success_response(data=detail, request=request) @admin_router.post( "/materials", status_code=status.HTTP_201_CREATED, summary="Create a material", ) async def admin_create_material( body: MaterialCreate, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_write(user) svc = SecurityAwarenessService(db) row = await svc.create( body, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response(data={"id": str(row.id)}, request=request) @admin_router.put("/materials/{material_id}", summary="Edit a material") async def admin_update_material( material_id: uuid.UUID, body: MaterialUpdate, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_write(user) svc = SecurityAwarenessService(db) row = await svc.update( material_id, body, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response( data={"id": str(row.id), "version": row.version}, request=request, ) @admin_router.post( "/materials/{material_id}/status", summary="Publish / unpublish / archive", ) async def admin_set_status( material_id: uuid.UUID, body: MaterialStatusUpdate, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_write(user) svc = SecurityAwarenessService(db) row = await svc.set_status( material_id, body.status, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response( data={"id": str(row.id), "status": row.status}, request=request, ) @admin_router.post( "/materials/{material_id}/duplicate", summary="Copy into a new draft", ) async def admin_duplicate_material( material_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_write(user) svc = SecurityAwarenessService(db) row = await svc.duplicate( material_id, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response(data={"id": str(row.id)}, request=request) @admin_router.get( "/translation-gaps", summary="How much of the library en / ja cannot read", ) async def admin_translation_gaps( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The number behind the "AI 补齐翻译" button. Cheap — no model calls — so the console can show "12 条素材缺英文" next to the control that fixes it rather than making an admin count language chips down a paginated table. Also reports whether the shared AI/LLM connector has a key, because a button that 503s is a worse answer than a button that explains itself. """ _require_read(user) return success_response( data=await MaterialTranslator(db).gap_summary(), request=request, ) @admin_router.post( "/materials/translate-missing", summary="AI-fill every material missing a language", ) async def admin_translate_missing( body: TranslateMissingRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Bulk backfill, capped per call. Same code path the hourly cron runs — one implementation, so the button and the schedule cannot drift. Idempotent: a language that already has content is skipped, so clicking twice costs nothing and changes nothing. """ _require_write(user) statuses = ( (*BACKFILL_STATUSES, STATUS_DRAFT) if body.include_drafts else BACKFILL_STATUSES ) summary = await MaterialTranslator(db).fill_missing( limit=body.limit, statuses=statuses, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response(data=summary, request=request) @admin_router.post( "/materials/{material_id}/translate", summary="AI-translate one material into its missing languages", ) async def admin_translate_material( material_id: uuid.UUID, body: MaterialTranslateRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Translate the ``zh`` block into ``targets`` (default: the missing ones). Never touches the baseline, never changes the status, and never translates artwork — a comic's pages are carried over unchanged, and the generated block is marked so every reader surface can say so. """ _require_write(user) outcome = await MaterialTranslator(db).translate_material( material_id, targets=body.targets, overwrite=body.overwrite, persist=body.persist, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response(data=outcome.as_dict(), request=request) @admin_router.post( "/materials/{material_id}/translation-reviewed", summary="Clear the machine-translation badge for one language", ) async def admin_translation_reviewed( material_id: uuid.UUID, body: TranslationReviewed, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """"I read it, it is fine." Editing the block clears the badge automatically; this is the path for the common review that produced no edit, so the badge is not permanent on exactly the translations that came out well. """ _require_write(user) svc = SecurityAwarenessService(db) row = await svc.mark_translation_reviewed( material_id, body.lang, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response( data={"id": str(row.id), "mt_langs": list(row.mt_langs or [])}, request=request, ) @admin_router.post("/materials/reorder", summary="Bulk reorder") async def admin_reorder_materials( body: MaterialReorder, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_write(user) svc = SecurityAwarenessService(db) await svc.reorder( body.ordered_ids, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return success_response(data={"ok": True}, request=request) @admin_router.delete( "/materials/{material_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a never-shown material", ) async def admin_delete_material( material_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Only while nothing has ever been shown — see ``SecurityAwarenessService.delete``. Anything that reached an employee is archived instead, because the exposure record is the reason this domain exists.""" _require_write(user) svc = SecurityAwarenessService(db) await svc.delete( material_id, actor_id=user.id, trace_id=getattr(request.state, "trace_id", ""), ) return None @admin_router.get("/bubble-settings", summary="Ambient-surface frequency caps") async def get_bubble_settings( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The three knobs that ration the ambient bubble. Exposed HERE, next to the feature they govern, rather than only in the generic platform-settings page: a threshold whose only home is a flat key-value table is a threshold nobody adjusts, and this one is the difference between an ambient nudge and a nuisance. """ _require_read(user) settings = await load_bubble_settings(db) return success_response( data={ "enabled": settings.enabled, "daily_new_materials": settings.daily_new_materials, "max_repeats_per_material": settings.max_repeats_per_material, "can_manage": can_manage_security(user.role_names), }, request=request, ) @admin_router.put("/bubble-settings", summary="Change the frequency caps") async def update_bubble_settings( body: BubbleSettingsUpdate, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_write(user) svc = PlatformSettingService(db) trace_id = getattr(request.state, "trace_id", "") changes: dict[str, object] = {} if body.enabled is not None: changes[SETTING_ENABLED] = bool(body.enabled) if body.daily_new_materials is not None: changes[SETTING_DAILY_NEW] = int(body.daily_new_materials) if body.max_repeats_per_material is not None: changes[SETTING_MAX_REPEATS] = int(body.max_repeats_per_material) for key, value in changes.items(): # ``description=None`` keeps whatever the migration seeded — the # descriptions ARE the operator documentation for these knobs and # a settings write from this console must not blank them. await svc.upsert( key, value, None, trace_id=str(trace_id or ""), actor_user_id=user.id, ) settings = await load_bubble_settings(db) return success_response( data={ "enabled": settings.enabled, "daily_new_materials": settings.daily_new_materials, "max_repeats_per_material": settings.max_repeats_per_material, }, request=request, ) # ════════════════════════════════════════════════════════════════════ # Compliance: coverage + per-person evidence # ════════════════════════════════════════════════════════════════════ #: Formula prefixes a spreadsheet would execute on open. Same guard the #: other CSV exports on this platform use — an evidence file is the LAST #: thing that should be able to run something. _CSV_INJECTION_PREFIXES = ("=", "+", "-", "@", "\t", "\r") def _csv_safe(value: object) -> object: if isinstance(value, str) and value.startswith(_CSV_INJECTION_PREFIXES): return "'" + value return value @admin_router.get("/coverage", summary="Awareness coverage") async def coverage( request: Request, window_days: int = Query(default=365, ge=1, le=1825), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Per-campaign coverage plus a people-weighted roll-up. Required, acknowledgement-mode campaigns only — a voluntary notice asks nothing of anybody, and letting a comic nobody had to read carry the number would make it meaningless in the one direction that matters. The roll-up weights by PEOPLE, not by campaign: averaging percentages lets one sent to 3 people move the number as much as one sent to 480. """ _require_read(user) data = await coverage_report(db, window_days=window_days) return success_response(data=data, request=request) @admin_router.get( "/campaigns/{campaign_id}/evidence.csv", summary="Per-person evidence for one campaign", ) async def evidence_csv( campaign_id: uuid.UUID, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> StreamingResponse: """The auditable artefact: one row per person, with the timestamps. Reads the target ledger's SNAPSHOT columns, not live person rows — somebody who transferred out last month was reached as a member of their old team, and re-deriving their department here would quietly restate a past department's coverage. """ _require_read(user) rows = await evidence_rows(db, campaign_id) buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow(list(EVIDENCE_COLUMNS)) for row in rows: writer.writerow([_csv_safe(row[c]) for c in EVIDENCE_COLUMNS]) # BOM so Excel opens the UTF-8 names correctly rather than as mojibake. payload = "\ufeff" + buffer.getvalue() return StreamingResponse( iter([payload]), media_type="text/csv; charset=utf-8", headers={ "Content-Disposition": ( f'attachment; filename="awareness-evidence-{campaign_id}.csv"' ), }, ) @admin_router.get( "/materials/{material_id}/quiz", summary="The quiz, with the answer key", ) async def admin_get_quiz( material_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): _require_read(user) svc = MaterialQuizService(db) return success_response( data={ "questions": await svc.questions_for( material_id, include_answers=True, ), # Per-question FIRST-attempt miss rate. The reason failed # attempts are kept: a question most people get wrong the # first time says a passage of the material is unclear, and # that is the input to the next edit. "difficulty": await svc.question_difficulty(material_id), }, request=request, ) @admin_router.put( "/materials/{material_id}/quiz", summary="Set the material's questions", ) async def admin_set_quiz( material_id: uuid.UUID, body: QuizQuestionSet, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Replace the question set, in the order given. Past attempts are untouched — they carry their own graded results, so changing the quiz never re-grades somebody's old answer. """ _require_write(user) count = await MaterialQuizService(db).set_questions( material_id, body.question_ids, actor_id=user.id, ) return success_response(data={"questions": count}, request=request) @admin_router.get("/participation", summary="Is anybody reading it") async def participation( request: Request, window_days: int = Query(default=90, ge=1, le=730), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Engagement with the WHOLE library, required or not. Kept separate from coverage on purpose: coverage is a compliance claim about required reading, and folding voluntary reading into it is how a comic nobody had to read ends up carrying an audit statement. This is the number that says whether the library and the ambient surface are worth running. """ _require_read(user) data = await participation_report(db, window_days=window_days) return success_response(data=data, request=request) @admin_router.get("/question-bank", summary="Questions available to attach") async def question_bank( request: Request, q: str | None = Query(default=None, max_length=200), limit: int = Query(default=100, ge=1, le=300), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Active questions from the shared security-quiz bank, for the attach picker. A narrow, read-only view served from THIS domain rather than reusing ``/security-quiz/admin/questions``: that endpoint gates on ``global_admin`` / ``auditor`` and does not admit ``security_officer`` — the role that actually runs awareness — and it also carries the bank's full CRUD surface. The awareness console needs "which questions may I attach", not "let me manage the daily quiz". Widening the other domain's gate to get a picker would hand the desk write access to the daily-quiz bank as a side effect. """ _require_read(user) from app.domains.security_quiz.models import SecQuizQuestion stmt = select(SecQuizQuestion).where(SecQuizQuestion.status == "active") # Filter in SQL, not after a capped fetch. Loading the newest N rows # and searching those in Python gives a search box a silent horizon: # once the bank passes N questions, typing a word that only occurs in # an older one returns nothing, which reads as "we do not have that # question" rather than as "we did not look". AND over tokens, so # typing more words narrows. for token in [t for t in (q or "").strip().lower().split() if t][:8]: needle = f"%{token}%" stmt = stmt.where( func.lower(SecQuizQuestion.prompt).like(needle) | func.lower(func.coalesce(SecQuizQuestion.category, "")).like( needle, ) ) rows = list((await db.execute( stmt.order_by(SecQuizQuestion.created_at.desc()).limit(limit) )).scalars().all()) return success_response( data=[ { "question_id": str(r.id), "prompt": r.prompt, "category": r.category, "qtype": r.qtype, "difficulty": r.difficulty, } for r in rows ], request=request, )