193 lines
9.4 KiB
Python
193 lines
9.4 KiB
Python
"""Разбор по отметкам: преподаватель подтверждает, снимает и добавляет отметку.
|
||
|
||
Баллы по-прежнему считает код. Преподаватель меняет входы — какие отметки
|
||
действуют, — а не формулу: балл пересчитывается теми же метриками и весами.
|
||
Связь «отметка → метрика» — методика `METRIC_MAP` в пределах карточки и службы;
|
||
если генератор знает метрику точно, она записана в `Finding.metric_key`.
|
||
|
||
Функции чистые и работают со словарём сохранённой оценки: живое занятие,
|
||
строка `Score` и архивный разбор меняются одним и тем же путём, как правка
|
||
итога в `apply_score_override`.
|
||
"""
|
||
|
||
from app.domain.events import Metric
|
||
from app.domain.taxonomy import Finding, FindingDecision, FindingReview, FindingSource
|
||
from app.scoring.competency import radar
|
||
from app.scoring.gost import GostResult
|
||
from app.scoring.taxonomy import INSTRUCTOR_METRIC, METRIC_MAP
|
||
|
||
__all__ = ["INSTRUCTOR_METRIC", "add_finding", "has_finding", "linked", "review_finding",
|
||
"reviewed_metrics"]
|
||
|
||
|
||
def _in_scope(finding: Finding, metric: Metric) -> bool:
|
||
if finding.card != metric.card:
|
||
return False
|
||
return finding.service is None or metric.service is None or finding.service == metric.service
|
||
|
||
|
||
def linked(finding: Finding, metric: Metric) -> bool:
|
||
"""Отметка объясняет провал этой метрики."""
|
||
if not _in_scope(finding, metric):
|
||
return False
|
||
if finding.metric_key is not None:
|
||
return finding.metric_key == metric.key
|
||
mapping = METRIC_MAP.get(metric.key)
|
||
return mapping is not None and mapping[0] is finding.code
|
||
|
||
|
||
def _dismissed(finding: Finding) -> bool:
|
||
return finding.review is not None and finding.review.decision is FindingDecision.DISMISSED
|
||
|
||
|
||
def _credit(metric: Metric) -> float:
|
||
return metric.credit if metric.credit is not None else float(metric.passed)
|
||
|
||
|
||
def reviewed_metrics(metrics: list[Metric], findings: list[Finding],
|
||
instructor_weight: float = 1.0) -> list[Metric]:
|
||
"""Метрики с учётом решений преподавателя.
|
||
|
||
Проваленная метрика засчитывается, только если все её автоматические
|
||
отметки сняты. Ручная отметка проваливает связанные метрики; если их нет,
|
||
появляется отдельная метрика преподавателя. Исходный вердикт остаётся в
|
||
`passed_auto`/`credit_auto`, повторный пересчёт начинается с него.
|
||
"""
|
||
base = [
|
||
metric.model_copy(update={
|
||
"passed": metric.passed_auto if metric.passed_auto is not None else metric.passed,
|
||
"credit": metric.credit_auto if metric.passed_auto is not None else metric.credit,
|
||
"passed_auto": None,
|
||
"credit_auto": None,
|
||
})
|
||
for metric in metrics
|
||
if metric.key != INSTRUCTOR_METRIC
|
||
]
|
||
result: list[Metric] = []
|
||
for metric in base:
|
||
links = [finding for finding in findings if linked(finding, metric)]
|
||
manual = [finding for finding in links
|
||
if finding.source is FindingSource.INSTRUCTOR and not _dismissed(finding)]
|
||
automatic = [finding for finding in links if finding.source is not FindingSource.INSTRUCTOR]
|
||
if manual and _credit(metric) > 0:
|
||
metric = metric.model_copy(update={
|
||
"passed": False, "credit": 0.0,
|
||
"passed_auto": metric.passed, "credit_auto": metric.credit,
|
||
})
|
||
elif (not manual and _credit(metric) < 1 and automatic
|
||
and all(_dismissed(finding) for finding in automatic)):
|
||
metric = metric.model_copy(update={
|
||
"passed": True, "credit": 1.0,
|
||
"passed_auto": metric.passed, "credit_auto": metric.credit,
|
||
})
|
||
result.append(metric)
|
||
|
||
for finding in findings:
|
||
if (finding.source is FindingSource.INSTRUCTOR and not _dismissed(finding)
|
||
and not any(linked(finding, metric) for metric in base)):
|
||
result.append(Metric(
|
||
key=INSTRUCTOR_METRIC,
|
||
title=f"Отметка преподавателя: {finding.summary}",
|
||
fact=finding.fact,
|
||
norm=finding.norm or "по решению преподавателя",
|
||
ref=finding.ref,
|
||
passed=False,
|
||
weight=instructor_weight,
|
||
card=finding.card,
|
||
service=finding.service,
|
||
))
|
||
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:
|
||
metrics = reviewed_metrics(
|
||
[Metric.model_validate(item) for item in report.get("metrics", [])], findings,
|
||
_instructor_weight(report),
|
||
)
|
||
score = GostResult(metrics=metrics).score
|
||
fields = {
|
||
"findings": [finding.model_dump(mode="json") for finding in findings],
|
||
"metrics": [metric.model_dump(mode="json") for metric in metrics],
|
||
"competencies": [item.model_dump() for item in radar(metrics)],
|
||
"score_reviewed": score,
|
||
}
|
||
# Правка итога — последнее слово преподавателя; решение по отметке её не отменяет.
|
||
if not report.get("overridden_by"):
|
||
fields["score_final"] = score
|
||
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")
|
||
if isinstance(archived, dict):
|
||
failed = sum(not metric.passed and metric.weight > 0 for metric in metrics)
|
||
allowed = (archived.get("criteria") or {}).get("allowed_errors", 0)
|
||
updated["full_report"] = {
|
||
**archived, **fields,
|
||
"card_results": _cards(archived.get("card_results") or [], metrics, findings),
|
||
"failed_metrics": failed,
|
||
"passed": failed <= allowed,
|
||
"score_final": updated.get("score_final", archived.get("score_final")),
|
||
}
|
||
return updated
|
||
|
||
|
||
def _findings(report: dict) -> list[Finding]:
|
||
return [Finding.model_validate(item) for item in report.get("findings", [])]
|
||
|
||
|
||
def review_finding(report: dict, index: int, review: FindingReview) -> dict:
|
||
"""Подтвердить или снять отметку. Решение дописывается, действует последнее."""
|
||
findings = _findings(report)
|
||
if not 0 <= index < len(findings):
|
||
raise LookupError(f"нет отметки {index}")
|
||
findings[index] = findings[index].model_copy(
|
||
update={"reviews": [*findings[index].reviews, review]})
|
||
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:
|
||
"""Добавить отметку преподавателя; она штрафует связанную метрику."""
|
||
if finding.source is not FindingSource.INSTRUCTOR:
|
||
raise ValueError("вручную добавляется только отметка преподавателя")
|
||
# Повтор (ретрай сети, вторая вкладка) не штрафует второй раз.
|
||
if has_finding(report, finding.client_id):
|
||
return report
|
||
return _rescore(report, [*_findings(report), finding])
|