scripts/import_issues.py - master - Snikius/kanban

scripts/import_issues.py
210 строк7 KB
ssamushenkov
Import only tickets assigned to mapped Scalecar users
16 июл 2026, 15:10
16 июл 2026, 15:10221077e
#!/usr/bin/env python3
import argparse
import re
import secrets
import sys
import xml.etree.ElementTree as ET
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path

from sqlalchemy import func

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from app import app  # noqa: E402
from models import Column, Task, User, db  # noqa: E402


STATUS_COLUMNS = {
    'Беклог': 'To Do',
    'Selected for Development': 'To Do',
    'В работе': 'In Progress',
    'Ready for Testing': 'Ready To Test',
    'Тестируется': 'Testing',
    'Done': 'Done',
    'Invalid': 'Done',
}

# Legacy Jira login -> active Scalecar moderator login.  Only issues assigned
# to a key in this map may be imported.  This prevents legacy/removed users
# from appearing in the new SSO-only board.
ASSIGNEE_MAPPING = {
    's.samushenkov': ('Stas', 'Stas Samushenkov'),
    'a.bocharov': ('alexbo', 'Alex Bocharov'),
    'm.kharlamov': ('Desperado', 'Michael Kharlamov'),
    'y.bondarev': ('y.bondarev', 'y.bondarev'),
    'a.brovkina': ('a.brovkina', 'Anna Brovkina'),
    'n.rudenko': ('n.rudenko', 'n.rudenko'),
}

COLUMN_COLORS = {
    'To Do': '#6b7280',
    'In Progress': '#f59e0b',
    'Ready To Test': '#3b82f6',
    'Testing': '#8b5cf6',
    'Done': '#10b981',
}

RUSSIAN_MONTHS = {
    'января': 1, 'февраля': 2, 'марта': 3, 'апреля': 4,
    'мая': 5, 'июня': 6, 'июля': 7, 'августа': 8,
    'сентября': 9, 'октября': 10, 'ноября': 11, 'декабря': 12,
}


def parse_assignee(value):
    value = (value or '').strip()
    match = re.fullmatch(r'(.+?)\(([^()]+)\)', value)
    if not match:
        return (value, value) if value else (None, None)
    return match.group(1).strip(), match.group(2).strip()


def parse_updated(value):
    match = re.fullmatch(
        r'(\d{1,2})\s+(\S+)\s+(\d{4})\s+в\s+(\d{1,2}):(\d{2})',
        (value or '').strip(),
    )
    if not match:
        return datetime.now(timezone.utc)
    day, month_name, year, hour, minute = match.groups()
    return datetime(
        int(year), RUSSIAN_MONTHS[month_name], int(day),
        int(hour), int(minute), tzinfo=timezone.utc,
    )


def import_issues(
    xml_path,
    creator_login,
    dry_run=False,
):
    issues = ET.parse(xml_path).getroot().findall('issue')
    columns = {column.title: column for column in Column.query.all()}
    missing = sorted(set(STATUS_COLUMNS.values()) - set(columns))
    if missing:
        next_position = max((column.position for column in columns.values()), default=-1) + 1
        for title in COLUMN_COLORS:
            if title in missing:
                column = Column(
                    title=title,
                    color=COLUMN_COLORS[title],
                    position=next_position,
                )
                db.session.add(column)
                db.session.flush()
                columns[title] = column
                next_position += 1

    users = {user.login: user for user in User.query.all()}
    created_users = 0
    created_tasks = 0
    updated_tasks = 0
    skipped_tasks = 0
    skipped_assignees = defaultdict(int)

    # Local user rows are only projections of Scalecar moderators. They are
    # created up front so assignees and the required task creator are valid;
    # they cannot be used to log in because password login is disabled.
    for target_login, target_name in ASSIGNEE_MAPPING.values():
        if target_login not in users:
            user = User(login=target_login, name=target_name)
            user.set_password(secrets.token_urlsafe(18))
            db.session.add(user)
            db.session.flush()
            users[target_login] = user
            created_users += 1

    creator = users.get(creator_login)
    if not creator:
        raise RuntimeError(
            f'Creator {creator_login!r} is not in the approved assignee mapping'
        )

    positions = defaultdict(int)
    for column in columns.values():
        positions[column.id] = (
            db.session.query(func.max(Task.position))
            .filter_by(column_id=column.id).scalar() or -1
        ) + 1

    for issue in issues:
        _, assignee_login = parse_assignee(issue.findtext('assignee'))
        target = ASSIGNEE_MAPPING.get(assignee_login)
        if not target:
            skipped_tasks += 1
            skipped_assignees[assignee_login or '(unassigned)'] += 1
            continue
        assignee = users[target[0]]

        key = (issue.findtext('key') or '').strip()
        status = (issue.findtext('status') or '').strip()
        column = columns[STATUS_COLUMNS[status]]
        updated_at = parse_updated(issue.findtext('updated'))
        task = Task.query.filter_by(external_key=key).first()
        if not task:
            task = Task(
                external_key=key,
                creator_id=creator.id,
                position=positions[column.id],
                created_at=updated_at,
            )
            positions[column.id] += 1
            db.session.add(task)
            created_tasks += 1
        else:
            updated_tasks += 1

        summary = (issue.findtext('summary') or key).strip()
        task.title = summary
        if not task.description:
            task.description = summary
        task.priority = (issue.findtext('priority') or '').strip() or None
        task.issue_type = (issue.findtext('type') or '').strip() or None
        task.source_status = status
        task.parent_key = (issue.findtext('parent') or '').strip() or None
        task.assignee_id = assignee.id if assignee else None
        task.column_id = column.id
        task.updated_at = updated_at
        if not task.created_at:
            task.created_at = updated_at

    result = {
        'issues': len(issues),
        'skipped_tasks': skipped_tasks,
        'skipped_assignees': dict(sorted(skipped_assignees.items())),
        'created_tasks': created_tasks,
        'updated_tasks': updated_tasks,
        'created_users': created_users,
    }
    if dry_run:
        db.session.rollback()
    else:
        db.session.commit()
    return result


def main():
    parser = argparse.ArgumentParser(description='Import legacy board issues XML')
    parser.add_argument('xml_path')
    parser.add_argument('--creator-login', required=True)
    parser.add_argument(
        '--dry-run', action='store_true',
        help='Validate and count the import, then roll back all local changes.',
    )
    args = parser.parse_args()
    with app.app_context():
        result = import_issues(
            args.xml_path,
            args.creator_login,
            args.dry_run,
        )
    print(' '.join(
        f'{key}={value}' for key, value in result.items() if key != 'skipped_assignees'
    ))
    print(f"skipped_assignees={result['skipped_assignees']}")


if __name__ == '__main__':
    main()