app/crowd_test_data.py - master - artyuhovs/AIFeedBackTrainingBot

app/crowd_test_data.py
565 строк26 KB
Codex
Завершить архитектурное оздоровление и исправить итоги
15 июл 2026, 11:05
15 июл 2026, 11:05c65f9a6
from __future__ import annotations

import argparse
import re
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path

from app.config import load_settings
from app.db.repositories import Repositories, text_sha256
from app.db.repository_parts.common import inserted_id
from app.devtools import (
    PRODUCTION_CONFIRMATION,
    authorize_test_data_write,
    record_devtool_event,
)
from app.test_story import prepare_test_story
from app.utils.telegram_users import normalize_username

DEFAULT_REAL_USERS = ("@sergeart", "@annikart")

WORKS = (
    (
        "Можно дам обратную связь по дейли? За последнюю неделю ты трижды подключался "
        "на 7–10 минут позже. Команда возвращалась к уже обсуждённым вопросам. "
        "Давай приходить к началу или заранее писать в чат о задержке."
    ),
    "Ты постоянно опаздываешь на встречи и всех подводишь. Пожалуйста, стань ответственнее.",
    (
        "На вчерашнем дейли тебя не было первые восемь минут, поэтому статус по интеграции "
        "пришлось уточнять отдельно. Предлагаю предупреждать о задержке и оставлять короткий "
        "статус в чате до начала встречи."
    ),
    (
        "Мне кажется, тебе не очень интересна командная работа. Когда ты опаздываешь, "
        "это выглядит неуважительно. Постарайся больше так не делать."
    ),
    (
        "Хочу обсудить начало дейли. В понедельник, среду и пятницу ты подключился после "
        "10:07, хотя встреча начинается в 10:00. Из-за повторов мы заканчивали позже. "
        "Что поможет тебе быть вовремя?"
    ),
    (
        "Ты хороший разработчик, но с пунктуальностью есть проблема. Было бы здорово "
        "исправить её как-нибудь в ближайшее время."
    ),
    (
        "Когда ты сегодня вошёл в 10:09, мы уже распределили блокеры и вернулись к ним ещё раз. "
        "Давай со следующей недели подключаться к 10:00; если не получается — сообщать до встречи."
    ),
    "Не опаздывай на Daily Scrum. Это обязательная встреча.",
    (
        "Можно поделиться наблюдением? На четырёх дейли за две недели ты присоединился "
        "после начала. Я переживаю, что мы теряем общий контекст. Давай проверим через "
        "неделю договорённость о предупреждении в чате."
    ),
    "Все уже заметили твои опоздания. Команда раздражена, потому что приходится ждать. Исправь, пожалуйста.",
    (
        "Вчера дейли начался без тебя, а вопрос по релизу завис до твоего подключения. "
        "Если утро неудобно, давай обсудим перенос встречи; пока прошу заранее оставлять статус."
    ),
    (
        "На этой неделе ты два раза опоздал на дейли. Из-за этого обсуждение заняло примерно "
        "на пять минут больше. Можешь поставить напоминание за десять минут до встречи?"
    ),
)

REVIEWS = (
    (
        "Сильная конкретика: названы даты, время и влияние на команду. Следующий шаг понятен. "
        "Я бы только сначала спросил разрешение на обратную связь."
    ),
    (
        "Есть наблюдение и просьба, но фраза звучит обвиняюще. Лучше убрать обобщение "
        "«постоянно» и сослаться на конкретные встречи."
    ),
    (
        "Хорошо отделены факты от интерпретаций. Особенно полезно, что предложен запасной "
        "вариант — сообщение в чат при задержке."
    ),
    (
        "Понятно, что нужно изменить, однако не хватает последствий для команды. Добавьте, "
        "зачем пунктуальность важна именно в этой ситуации."
    ),
    (
        "Тон спокойный и уважительный, есть приглашение к диалогу. Вопрос в конце помогает "
        "вместе найти решение, а не навязать его."
    ),
    "Рецензия короткая: всё нормально, можно отправлять.",
    (
        "Факты приведены точно, но просьба может восприниматься как ультиматум. Я бы добавил "
        "пространство для ответа и уточнил возможные препятствия."
    ),
    (
        "Начало удачное, но слова о неуважении приписывают человеку мотивы. Безопаснее "
        "описать наблюдаемое поведение и его эффект."
    ),
    (
        "Есть и конкретный пример, и реалистичная договорённость. Для проверки результата "
        "можно обозначить срок: например, вернуться к теме через неделю."
    ),
    "Мне понравилось. Всё хорошо написано.",
    (
        "Полезно, что автор не оценивает личность и не говорит от имени всей команды. "
        "Можно усилить формулировку, указав точное число опозданий."
    ),
    (
        "Следующий шаг измеримый, тон нейтральный. При этом совет с напоминанием дан слишком "
        "рано — сначала стоило спросить, что мешает приходить вовремя."
    ),
)


@dataclass(frozen=True)
class SeedResult:
    stream_id: int
    publication_id: int
    synthetic_users: int
    works_for_each_real_user: int
    reviews_for_each_real_user: int
    stage: str


def _key(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_") or "stream"


def _real_user(db: sqlite3.Connection, username: str) -> sqlite3.Row:
    normalized = normalize_username(username)
    row = db.execute(
        "SELECT * FROM bot_users WHERE lower(username)=? ORDER BY telegram_user_id>0 DESC LIMIT 1",
        (normalized,),
    ).fetchone()
    if row is None or int(row["telegram_user_id"]) <= 0:
        raise ValueError(f"Живой пользователь @{normalized} ещё не зарегистрирован в боте")
    return row


def seed_crowd_test_data(
    repository: Repositories,
    *,
    stream_title: str,
    real_usernames: tuple[str, ...] = DEFAULT_REAL_USERS,
    synthetic_count: int = 24,
    works_per_real_user: int = 12,
    reviews_per_real_user: int = 12,
    stage: str = "review",
) -> SeedResult:
    if synthetic_count < max(works_per_real_user, reviews_per_real_user):
        raise ValueError("synthetic_count должен быть не меньше числа работ и рецензий на пользователя")
    if stage not in {"review", "quality"}:
        raise ValueError("stage должен быть review или quality")

    story = prepare_test_story(
        repository,
        start_date=repository.database.clock.now().date().isoformat(),
        activate=True,
        stream_title=stream_title,
        participant_usernames=real_usernames,
    )
    timestamp = repository.database.now_iso()
    now = datetime.fromisoformat(timestamp)
    prefix = f"seed_{_key(stream_title)}_"
    tool = "crowd_test_data"

    with repository.database.transaction(immediate=True) as db:
        real_users = [_real_user(db, username) for username in real_usernames]
        publication = db.execute(
            """
            SELECT stream_publications.id
            FROM stream_publications
            JOIN course_publications ON course_publications.id=stream_publications.course_publication_id
            WHERE stream_publications.stream_id=? AND course_publications.crowd_training_enabled=1
            ORDER BY stream_publications.id LIMIT 1
            """,
            (story.stream_id,),
        ).fetchone()
        if publication is None:
            raise RuntimeError("В тестовом курсе не найдено crowd-задание")
        publication_id = int(publication["id"])
        batch_key = str(story.stream_id)
        repository.devtool_ownership.clear_batch(
            db,
            tool=tool,
            batch_key=batch_key,
        )

        real_ids = [int(user["telegram_user_id"]) for user in real_users]

        synthetic_ids: list[int] = []
        for index in range(synthetic_count):
            user_id = -8_000_000 - story.stream_id * 10_000 - index
            username = f"{prefix}{index + 1:03d}"
            synthetic_ids.append(user_id)
            if (
                db.execute(
                    "SELECT 1 FROM bot_users WHERE telegram_user_id=? OR lower(username)=?",
                    (user_id, username.lower()),
                ).fetchone()
                is not None
            ):
                raise RuntimeError("Synthetic identity collides with a record not owned by this batch")
            db.execute(
                """
                INSERT INTO bot_users
                    (telegram_user_id, identity_kind, username, first_name, last_name, is_admin,
                     current_stream_id, current_assignment_id, current_assignment_kind,
                     free_training_pending, created_at, updated_at)
                VALUES (?, 'synthetic', ?, 'Тестовый', ?, 0, ?, NULL, 'original', 0, ?, ?)
                """,
                (user_id, username, f"Участник {index + 1}", story.stream_id, timestamp, timestamp),
            )
            repository.devtool_ownership.claim(
                db,
                tool=tool,
                batch_key=batch_key,
                entity_type="bot_users",
                entity_id=user_id,
            )
            membership = db.execute(
                """
                INSERT INTO stream_memberships
                    (stream_id, telegram_user_id, status, requested_at, decided_at, updated_at)
                VALUES (?, ?, 'approved', ?, ?, ?)
                """,
                (story.stream_id, user_id, timestamp, timestamp, timestamp),
            )
            membership_id = inserted_id(membership)
            repository.devtool_ownership.claim(
                db,
                tool=tool,
                batch_key=batch_key,
                entity_type="stream_memberships",
                entity_id=membership_id,
            )
            db.execute(
                "INSERT INTO stream_membership_roles (membership_id, role, created_at) VALUES (?, 'learner', ?)",
                (membership_id, timestamp),
            )
            assignment = db.execute(
                """
                INSERT INTO participant_assignments
                    (stream_publication_id, telegram_user_id, status, submitted_at,
                     self_assessment_status, created_at, updated_at)
                VALUES (?, ?, 'submitted', ?, 'skipped', ?, ?)
                """,
                (publication_id, user_id, timestamp, timestamp, timestamp),
            )
            assignment_id = inserted_id(assignment)
            repository.devtool_ownership.claim(
                db,
                tool=tool,
                batch_key=batch_key,
                entity_type="participant_assignments",
                entity_id=assignment_id,
            )
            work = WORKS[index % len(WORKS)]
            message = db.execute(
                """
                INSERT INTO messages
                    (stream_id, participant_assignment_id, submission_kind, telegram_user_id,
                     telegram_chat_id, telegram_message_id, telegram_file_id, media_type,
                     mime_type, file_size_bytes, status, transcript_text, transcript_sha256,
                     created_at, updated_at, transcribed_at)
                VALUES (?, ?, 'original', ?, ?, ?, ?, 'text', 'text/plain', ?,
                        'transcribed', ?, ?, ?, ?, ?)
                """,
                (
                    story.stream_id,
                    assignment_id,
                    user_id,
                    user_id,
                    index + 1,
                    f"seed:{story.stream_id}:{index + 1}",
                    len(work.encode()),
                    work,
                    text_sha256(work),
                    timestamp,
                    timestamp,
                    timestamp,
                ),
            )
            repository.devtool_ownership.claim(
                db,
                tool=tool,
                batch_key=batch_key,
                entity_type="messages",
                entity_id=inserted_id(message),
            )

        synthetic_assignments = {
            int(row["telegram_user_id"]): int(row["id"])
            for row in db.execute(
                """
                SELECT id, telegram_user_id
                FROM participant_assignments
                WHERE stream_publication_id=? AND telegram_user_id<0
                """,
                (publication_id,),
            )
            if int(row["telegram_user_id"]) in synthetic_ids
        }
        review_deadline = (now + timedelta(days=2)).isoformat()
        for real_index, real_user_id in enumerate(real_ids):
            for offset in range(works_per_real_user):
                fake_id = synthetic_ids[(real_index * works_per_real_user + offset) % synthetic_count]
                target_assignment_id = synthetic_assignments[fake_id]
                if (
                    db.execute(
                        """
                    SELECT 1 FROM peer_review_assignments
                    WHERE target_assignment_id=? AND reviewer_user_id=?
                    """,
                        (target_assignment_id, real_user_id),
                    ).fetchone()
                    is not None
                ):
                    continue
                review_assignment = db.execute(
                    """
                    INSERT INTO peer_review_assignments
                        (target_assignment_id, reviewer_user_id, status, assigned_at,
                         deadline_at)
                    VALUES (?, ?, 'assigned', ?, ?)
                    """,
                    (target_assignment_id, real_user_id, timestamp, review_deadline),
                )
                repository.devtool_ownership.claim(
                    db,
                    tool=tool,
                    batch_key=batch_key,
                    entity_type="peer_review_assignments",
                    entity_id=inserted_id(review_assignment),
                )

            real_assignment = db.execute(
                """
                SELECT id, status, submitted_at, self_assessment_status, updated_at
                FROM participant_assignments
                WHERE stream_publication_id=? AND telegram_user_id=?
                """,
                (publication_id, real_user_id),
            ).fetchone()
            if real_assignment is None:
                raise RuntimeError("Не найдено задание живого участника")
            # If there is no live answer yet, add an explicit fixture answer so received reviews are testable now.
            has_answer = db.execute(
                "SELECT 1 FROM messages WHERE participant_assignment_id=? AND submission_kind='original' LIMIT 1",
                (real_assignment["id"],),
            ).fetchone()
            if has_answer is None:
                work = WORKS[(real_index * 5) % len(WORKS)]
                message = db.execute(
                    """
                    INSERT INTO messages
                        (stream_id, participant_assignment_id, submission_kind, telegram_user_id,
                         telegram_chat_id, telegram_message_id, telegram_file_id, media_type,
                         mime_type, file_size_bytes, status, transcript_text, transcript_sha256,
                         created_at, updated_at, transcribed_at)
                    VALUES (?, ?, 'original', ?, ?, ?, ?, 'text', 'text/plain', ?,
                            'transcribed', ?, ?, ?, ?, ?)
                    """,
                    (
                        story.stream_id,
                        real_assignment["id"],
                        real_user_id,
                        real_user_id,
                        -story.stream_id,
                        f"seed:real:{story.stream_id}:{real_user_id}",
                        len(work.encode()),
                        work,
                        text_sha256(work),
                        timestamp,
                        timestamp,
                        timestamp,
                    ),
                )
                repository.devtool_ownership.claim(
                    db,
                    tool=tool,
                    batch_key=batch_key,
                    entity_type="messages",
                    entity_id=inserted_id(message),
                )
                repository.devtool_ownership.claim(
                    db,
                    tool=tool,
                    batch_key=batch_key,
                    entity_type="participant_assignment_state",
                    entity_id=int(real_assignment["id"]),
                    snapshot={
                        "before": {
                            "status": real_assignment["status"],
                            "submitted_at": real_assignment["submitted_at"],
                            "self_assessment_status": real_assignment["self_assessment_status"],
                            "updated_at": real_assignment["updated_at"],
                        },
                        "after": {
                            "status": "submitted",
                            "submitted_at": timestamp,
                            "self_assessment_status": "skipped",
                            "updated_at": timestamp,
                        },
                    },
                )
                db.execute(
                    """
                    UPDATE participant_assignments
                    SET status='submitted', submitted_at=?,
                        self_assessment_status='skipped', updated_at=?
                    WHERE id=?
                    """,
                    (timestamp, timestamp, real_assignment["id"]),
                )
            for offset in range(reviews_per_real_user):
                fake_id = synthetic_ids[(real_index * reviews_per_real_user + offset) % synthetic_count]
                review_assignment = db.execute(
                    """
                    INSERT INTO peer_review_assignments
                        (target_assignment_id, reviewer_user_id, status, assigned_at,
                         completed_at, deadline_at)
                    VALUES (?, ?, 'completed', ?, ?, ?)
                    """,
                    (
                        real_assignment["id"],
                        fake_id,
                        timestamp,
                        timestamp,
                        review_deadline,
                    ),
                )
                review_assignment_id = inserted_id(review_assignment)
                repository.devtool_ownership.claim(
                    db,
                    tool=tool,
                    batch_key=batch_key,
                    entity_type="peer_review_assignments",
                    entity_id=review_assignment_id,
                )
                comment = REVIEWS[(real_index * 3 + offset) % len(REVIEWS)]
                score = float(4 + (offset * 5 + real_index * 3) % 7)
                review = db.execute(
                    """
                    INSERT INTO peer_reviews
                        (peer_review_assignment_id, score, max_score, comment, created_at, updated_at)
                    VALUES (?, ?, 10, ?, ?, ?)
                    """,
                    (review_assignment_id, score, comment, timestamp, timestamp),
                )
                repository.devtool_ownership.claim(
                    db,
                    tool=tool,
                    batch_key=batch_key,
                    entity_type="peer_reviews",
                    entity_id=inserted_id(review),
                )

        review_closed_at = timestamp if stage == "quality" else None
        quality_deadline = (now + timedelta(days=2)).isoformat() if stage == "quality" else None
        db.execute(
            """
            UPDATE stream_publications
            SET scheduled_at=?, deadline_at=?, crowd_distributed_at=?,
                review_deadline_at=?, review_closed_at=?, review_quality_deadline_at=?,
                review_quality_closed_at=NULL, updated_at=?
            WHERE id=?
            """,
            (
                (now - timedelta(minutes=10)).isoformat(),
                (now - timedelta(minutes=5)).isoformat(),
                timestamp,
                review_deadline,
                review_closed_at,
                quality_deadline,
                timestamp,
                publication_id,
            ),
        )
        if stage == "quality":
            db.execute(
                """
                UPDATE peer_review_assignments
                SET status='skipped', completed_at=COALESCE(completed_at, ?)
                WHERE status='assigned'
                  AND id IN (
                    SELECT entity_id FROM devtool_owned_records
                    WHERE tool=? AND batch_key=?
                      AND entity_type='peer_review_assignments'
                  )
                """,
                (timestamp, tool, batch_key),
            )
        # Prepared crowd data must stay invisible until an admin explicitly starts the stream.
        db.execute(
            "UPDATE streams SET status='draft', updated_at=? WHERE id=?",
            (timestamp, story.stream_id),
        )

    result = SeedResult(
        stream_id=story.stream_id,
        publication_id=publication_id,
        synthetic_users=synthetic_count,
        works_for_each_real_user=works_per_real_user,
        reviews_for_each_real_user=reviews_per_real_user,
        stage=stage,
    )
    record_devtool_event(
        repository,
        tool=tool,
        operation="seed",
        result="completed",
        details={
            "stream_id": result.stream_id,
            "publication_id": result.publication_id,
            "synthetic_count": synthetic_count,
            "works_per_real_user": works_per_real_user,
            "reviews_per_real_user": reviews_per_real_user,
            "stage": stage,
        },
    )
    return result


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate reusable crowd-training test data")
    parser.add_argument("--database", type=Path, default=None)
    parser.add_argument("--stream", required=True, help="Stream title, for example тестпоток2")
    parser.add_argument("--real-user", action="append", dest="real_users")
    parser.add_argument("--synthetic-count", type=int, default=24)
    parser.add_argument("--works-per-user", type=int, default=12)
    parser.add_argument("--reviews-per-user", type=int, default=12)
    parser.add_argument("--stage", choices=("review", "quality"), default="review")
    parser.add_argument("--allow-production", action="store_true")
    parser.add_argument("--confirm-production", metavar=PRODUCTION_CONFIRMATION)
    args = parser.parse_args()
    settings = load_settings()
    database = args.database.resolve() if args.database else settings.database_path
    repository = Repositories(database)
    repository.schema.init()
    authorize_test_data_write(
        settings,
        repository,
        tool="crowd_test_data",
        allow_production=args.allow_production,
        confirmation=args.confirm_production,
    )
    result = seed_crowd_test_data(
        repository,
        stream_title=args.stream,
        real_usernames=tuple(args.real_users or DEFAULT_REAL_USERS),
        synthetic_count=args.synthetic_count,
        works_per_real_user=args.works_per_user,
        reviews_per_real_user=args.reviews_per_user,
        stage=args.stage,
    )
    print(
        f"Seeded stream_id={result.stream_id}, publication_id={result.publication_id}, "
        f"synthetic_users={result.synthetic_users}, works_per_real_user={result.works_for_each_real_user}, "
        f"reviews_per_real_user={result.reviews_for_each_real_user}, stage={result.stage}"
    )


if __name__ == "__main__":
    main()