fix: история решений по отметке, пересчёт кодов и баллов карточек, вес отметки преподавателя из весов занятия

This commit is contained in:
GGlamer 2026-09-27 22:32:41 +03:00
commit db8787c6b3
4 changed files with 100 additions and 16 deletions

View file

@ -7,8 +7,10 @@
from datetime import datetime from datetime import datetime
from enum import StrEnum from enum import StrEnum
from typing import Any
from uuid import UUID
from pydantic import BaseModel from pydantic import BaseModel, model_validator
class ErrorCode(StrEnum): class ErrorCode(StrEnum):
@ -188,4 +190,24 @@ class Finding(BaseModel):
service: str | None = None service: str | None = None
#: Автор ручной отметки преподавателя (`source=instructor`). #: Автор ручной отметки преподавателя (`source=instructor`).
author: str | None = None author: str | None = None
review: FindingReview | None = None #: Ключ идемпотентности ручной отметки: повтор запроса не добавляет её снова.
client_id: UUID | None = None
#: Все решения по отметке по порядку; действует последнее. Прежние не
#: затираются: «кто, когда, почему» нужно по каждой правке, а не по итогу.
reviews: list[FindingReview] = []
@model_validator(mode="before")
@classmethod
def _single_review(cls, data: Any) -> Any:
# Разборы, сохранённые до истории решений, хранили одно поле `review`.
if isinstance(data, dict) and "review" in data:
data = dict(data)
single = data.pop("review")
if single and not data.get("reviews"):
data["reviews"] = [single]
return data
@property
def review(self) -> FindingReview | None:
"""Действующее решение — последнее."""
return self.reviews[-1] if self.reviews else None

View file

@ -14,12 +14,10 @@ from app.domain.events import Metric
from app.domain.taxonomy import Finding, FindingDecision, FindingReview, FindingSource from app.domain.taxonomy import Finding, FindingDecision, FindingReview, FindingSource
from app.scoring.competency import radar from app.scoring.competency import radar
from app.scoring.gost import GostResult from app.scoring.gost import GostResult
from app.scoring.taxonomy import METRIC_MAP from app.scoring.taxonomy import INSTRUCTOR_METRIC, METRIC_MAP
#: Метрика для ручной отметки, которой нечего провалить: у кода нет метрики __all__ = ["INSTRUCTOR_METRIC", "add_finding", "has_finding", "linked", "review_finding",
#: в METRIC_MAP (D4) или в карточке она не считалась. Вес — как у метрики по "reviewed_metrics"]
#: умолчанию, чтобы отметка преподавателя не весила больше автоматической.
INSTRUCTOR_METRIC = "instructor_finding"
def _in_scope(finding: Finding, metric: Metric) -> bool: def _in_scope(finding: Finding, metric: Metric) -> bool:
@ -46,7 +44,8 @@ def _credit(metric: Metric) -> float:
return metric.credit if metric.credit is not None else float(metric.passed) return metric.credit if metric.credit is not None else float(metric.passed)
def reviewed_metrics(metrics: list[Metric], findings: list[Finding]) -> list[Metric]: def reviewed_metrics(metrics: list[Metric], findings: list[Finding],
instructor_weight: float = 1.0) -> list[Metric]:
"""Метрики с учётом решений преподавателя. """Метрики с учётом решений преподавателя.
Проваленная метрика засчитывается, только если все её автоматические Проваленная метрика засчитывается, только если все её автоматические
@ -93,15 +92,47 @@ def reviewed_metrics(metrics: list[Metric], findings: list[Finding]) -> list[Met
norm=finding.norm or "по решению преподавателя", norm=finding.norm or "по решению преподавателя",
ref=finding.ref, ref=finding.ref,
passed=False, passed=False,
weight=instructor_weight,
card=finding.card, card=finding.card,
service=finding.service, service=finding.service,
)) ))
return result return result
def _instructor_weight(report: dict) -> float:
"""Вес метрики преподавателя из весов занятия: веса настраивает преподаватель."""
criteria = (report.get("full_report") or {}).get("criteria") or {}
return float((criteria.get("score_weights") or {}).get(INSTRUCTOR_METRIC, 1.0))
def _codes(findings: list[Finding]) -> dict[str, int]:
"""Коды действующих отметок: по ним строятся частые ошибки группы и профиль."""
codes: dict[str, int] = {}
for finding in findings:
if not _dismissed(finding):
codes[finding.code.value] = codes.get(finding.code.value, 0) + 1
return codes
def _cards(cards: list[dict], metrics: list[Metric], findings: list[Finding]) -> list[dict]:
"""Карточки очереди ДДС с тем же разбором: их метрики — часть метрик занятия."""
updated = []
for number, card in enumerate(cards, 1):
own = [metric for metric in metrics if metric.card == number]
updated.append({
**card,
"metrics": [metric.model_dump(mode="json") for metric in own],
"findings": [finding.model_dump(mode="json")
for finding in findings if finding.card == number],
"score_reviewed": GostResult(metrics=own).score,
})
return updated
def _rescore(report: dict, findings: list[Finding]) -> dict: def _rescore(report: dict, findings: list[Finding]) -> dict:
metrics = reviewed_metrics( metrics = reviewed_metrics(
[Metric.model_validate(item) for item in report.get("metrics", [])], findings, [Metric.model_validate(item) for item in report.get("metrics", [])], findings,
_instructor_weight(report),
) )
score = GostResult(metrics=metrics).score score = GostResult(metrics=metrics).score
fields = { fields = {
@ -114,12 +145,17 @@ def _rescore(report: dict, findings: list[Finding]) -> dict:
if not report.get("overridden_by"): if not report.get("overridden_by"):
fields["score_final"] = score fields["score_final"] = score
updated = {**report, **fields} updated = {**report, **fields}
if "summary" in report:
updated["summary"] = {**(report.get("summary") or {}), "codes": _codes(findings)}
if report.get("card_results"):
updated["card_results"] = _cards(report["card_results"], metrics, findings)
archived = report.get("full_report") archived = report.get("full_report")
if isinstance(archived, dict): if isinstance(archived, dict):
failed = sum(not metric.passed and metric.weight > 0 for metric in metrics) failed = sum(not metric.passed and metric.weight > 0 for metric in metrics)
allowed = (archived.get("criteria") or {}).get("allowed_errors", 0) allowed = (archived.get("criteria") or {}).get("allowed_errors", 0)
updated["full_report"] = { updated["full_report"] = {
**archived, **fields, **archived, **fields,
"card_results": _cards(archived.get("card_results") or [], metrics, findings),
"failed_metrics": failed, "failed_metrics": failed,
"passed": failed <= allowed, "passed": failed <= allowed,
"score_final": updated.get("score_final", archived.get("score_final")), "score_final": updated.get("score_final", archived.get("score_final")),
@ -132,16 +168,26 @@ def _findings(report: dict) -> list[Finding]:
def review_finding(report: dict, index: int, review: FindingReview) -> dict: def review_finding(report: dict, index: int, review: FindingReview) -> dict:
"""Подтвердить или снять отметку. Повторное решение заменяет прежнее.""" """Подтвердить или снять отметку. Решение дописывается, действует последнее."""
findings = _findings(report) findings = _findings(report)
if not 0 <= index < len(findings): if not 0 <= index < len(findings):
raise LookupError(f"нет отметки {index}") raise LookupError(f"нет отметки {index}")
findings[index] = findings[index].model_copy(update={"review": review}) findings[index] = findings[index].model_copy(
update={"reviews": [*findings[index].reviews, review]})
return _rescore(report, findings) return _rescore(report, findings)
def has_finding(report: dict, client_id) -> bool:
"""Ручная отметка с этим ключом уже в разборе — повтор запроса."""
return client_id is not None and any(
finding.client_id == client_id for finding in _findings(report))
def add_finding(report: dict, finding: Finding) -> dict: def add_finding(report: dict, finding: Finding) -> dict:
"""Добавить отметку преподавателя; она штрафует связанную метрику.""" """Добавить отметку преподавателя; она штрафует связанную метрику."""
if finding.source is not FindingSource.INSTRUCTOR: if finding.source is not FindingSource.INSTRUCTOR:
raise ValueError("вручную добавляется только отметка преподавателя") raise ValueError("вручную добавляется только отметка преподавателя")
# Повтор (ретрай сети, вторая вкладка) не штрафует второй раз.
if has_finding(report, finding.client_id):
return report
return _rescore(report, [*_findings(report), finding]) return _rescore(report, [*_findings(report), finding])

View file

@ -35,6 +35,11 @@ METRIC_MAP: dict[str, tuple[ErrorCode, Competency]] = {
"description_grammar": (ErrorCode.E4, Competency.COMMUNICATION), "description_grammar": (ErrorCode.E4, Competency.COMMUNICATION),
} }
#: Метрика ручной отметки, которой нечего провалить (у кода нет метрики в
#: METRIC_MAP или в карточке она не считалась). Кода у неё нет: код несёт сама
#: отметка. Вес задаёт преподаватель в весах занятия, по умолчанию 1.
INSTRUCTOR_METRIC = "instructor_finding"
#: Вес метрики в детерминированной оценке. #: Вес метрики в детерминированной оценке.
#: #:
#: **Предварительные значения, требуют утверждения методистом.** Без весов все #: **Предварительные значения, требуют утверждения методистом.** Без весов все

View file

@ -47,6 +47,7 @@ from app.session.store import (
apply_finding_change, apply_finding_change,
apply_score_override, apply_score_override,
) )
from app.scoring.review import has_finding
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
LEASE_SECONDS = 15 LEASE_SECONDS = 15
@ -249,18 +250,28 @@ class PostgresSessionStore:
) )
if score is None: if score is None:
raise LookupError(f"нет оценки занятия {session_id}") raise LookupError(f"нет оценки занятия {session_id}")
# Повтор ручной отметки под блокировкой: ни изменения, ни второй строки аудита.
if (isinstance(record, FindingAdded)
and has_finding(score.report or {}, record.finding.client_id)):
return
before = score.score_final before = score.score_final
findings = (score.report or {}).get("findings", [])
score.report = apply_finding_change(dict(score.report or {}), record) score.report = apply_finding_change(dict(score.report or {}), record)
score.score_final = score.report.get("score_final", score.score_final) score.score_final = score.report.get("score_final", score.score_final)
if isinstance(record, FindingReviewed): if isinstance(record, FindingReviewed):
code = (findings[record.index].get("code", "?")
if 0 <= record.index < len(findings) else "?")
actor, action = record.review.author, "finding.review" actor, action = record.review.author, "finding.review"
what = (f"#{record.index} {record.review.decision.value}; " what = (f"#{record.index} {code} {record.review.decision.value}; "
f"reason_chars={len(record.review.reason)}") f"причина: {record.review.reason}")
else: else:
actor, action = record.finding.author or "instructor", "finding.add" finding = record.finding
what = f"{record.finding.code.value}; fact_chars={len(record.finding.fact)}" actor, action = finding.author or "instructor", "finding.add"
# Как у правки итога: текст решения живёт в разборе, в аудите — where = f", карточка {finding.card}" if finding.card is not None else ""
# что изменено, кем и как сдвинулся балл. what = (f"{finding.code.value}{where}; факт: {finding.fact}; "
f"норма: {finding.norm or ''}")
# Решение по отметке меняет чужой балл: аудит сам по себе должен
# ответить «кто, когда, что и почему», даже если разбор потом правили.
db.add(AuditLog( db.add(AuditLog(
actor=actor, role=record.role, action=action, actor=actor, role=record.role, action=action,
object_id=str(session_id), object_id=str(session_id),