From 2c815edbcb0ec1c9f288a2801a99e908e34e73c2 Mon Sep 17 00:00:00 2001 From: ivis-kosaka Date: Wed, 4 Feb 2026 05:15:48 +0000 Subject: [PATCH 01/47] feat: add bulk user upload --- src/app/components/bulk/BulkResultStep.vue | 251 +++++++ src/app/components/bulk/BulkUploadStep.vue | 200 ++++++ src/app/components/bulk/BulkUserTable.vue | 142 ++++ .../components/bulk/BulkValidationStep.vue | 400 +++++++++++ src/app/composables/useBulk.ts | 284 ++++++++ src/app/i18n/locales/en.json | 28 + src/app/i18n/locales/ja.json | 25 + src/app/pages/bulk/[id]/index.vue | 79 +++ src/app/pages/bulk/index.vue | 64 ++ src/app/types/bulks.ts | 92 +++ src/server/api/bulk.py | 219 ++++++ src/server/api/schemas.py | 59 ++ src/server/db/history.py | 26 +- src/server/entities/bulk.py | 147 ++++ src/server/services/bulks.py | 666 ++++++++++++++++++ src/server/services/history_table.py | 118 ++++ 16 files changed, 2789 insertions(+), 11 deletions(-) create mode 100644 src/app/components/bulk/BulkResultStep.vue create mode 100644 src/app/components/bulk/BulkUploadStep.vue create mode 100644 src/app/components/bulk/BulkUserTable.vue create mode 100644 src/app/components/bulk/BulkValidationStep.vue create mode 100644 src/app/composables/useBulk.ts create mode 100644 src/app/pages/bulk/[id]/index.vue create mode 100644 src/app/pages/bulk/index.vue create mode 100644 src/app/types/bulks.ts create mode 100644 src/server/api/bulk.py create mode 100644 src/server/entities/bulk.py create mode 100644 src/server/services/bulks.py create mode 100644 src/server/services/history_table.py diff --git a/src/app/components/bulk/BulkResultStep.vue b/src/app/components/bulk/BulkResultStep.vue new file mode 100644 index 00000000..cde78791 --- /dev/null +++ b/src/app/components/bulk/BulkResultStep.vue @@ -0,0 +1,251 @@ + + + diff --git a/src/app/components/bulk/BulkUploadStep.vue b/src/app/components/bulk/BulkUploadStep.vue new file mode 100644 index 00000000..6649cd8a --- /dev/null +++ b/src/app/components/bulk/BulkUploadStep.vue @@ -0,0 +1,200 @@ + + + diff --git a/src/app/components/bulk/BulkUserTable.vue b/src/app/components/bulk/BulkUserTable.vue new file mode 100644 index 00000000..61b72584 --- /dev/null +++ b/src/app/components/bulk/BulkUserTable.vue @@ -0,0 +1,142 @@ + + + diff --git a/src/app/components/bulk/BulkValidationStep.vue b/src/app/components/bulk/BulkValidationStep.vue new file mode 100644 index 00000000..782360cd --- /dev/null +++ b/src/app/components/bulk/BulkValidationStep.vue @@ -0,0 +1,400 @@ + + + diff --git a/src/app/composables/useBulk.ts b/src/app/composables/useBulk.ts new file mode 100644 index 00000000..19c6a2f6 --- /dev/null +++ b/src/app/composables/useBulk.ts @@ -0,0 +1,284 @@ +import type { ImportResultResponse, MissingUser, StatusType, ValidationResult } from "~/types/bulks" + +export const useUserUpload = () => { + const selectedFile = useState('userUpload:selectedFile', () => {}) + const selectedRepository = useState('userUpload:selectedRepository', () => {}) + const validationResults = useState('userUpload:validationResults', () => []) + const MissingUsers = useState('userUpload:MissingUsers', () => []) + const selectedMissingUsers = useState('userUpload:selectedMissingUsers', () => []) + const temporaryFileId = useState('userUpload:tempFileId', () => {}) + const summary = useState('userUpload:summary', () => ({ + total: 0, + status: { create: 0, update: 0, delete: 0, skip: 0, error: 0 }, + })) + const isProcessing = useState('userUpload:isProcessing', () => false) + const taskId = useState('userUpload:taskId', () => {}) + + const importResult = useState('userUpload:importResult', () => {}) + + function mapBackendOperation(backendStatus: string): StatusType { + const operationMap: Record = { + create: 'create', + update: 'update', + delete: 'delete', + skip: 'skip', + error: 'error', + } + return operationMap[backendStatus] || 'skip' + } + + async function fetchValidationResults(taskIdValue: string, queryParameters?: string) { + try { + const url = queryParameters + ? `/api/bulk/validate/result/${taskIdValue}?${queryParameters}` + : `/api/bulk/validate/result/${taskIdValue}` + + const result = await $fetch<{ + results: Array<{ + id: string + eppn: string[] + emails: string[] + userName: string + groups: string[] + status: string + code: string + }> + summary: { + create: number + update: number + delete: number + skip: number + error: number + } + missingUser: Array<{ + id: string + eppns: string[] + userName: string + emails: string[] + preferredLanguage: string + isSystemAdmin: boolean + repositories: Array<{ + id: string + displayName: string + serviceUrl: string | null + spConnecterId: string | null + userRoles: any | null + }> + groups: Array<{ + id: string + displayName: string | null + public: boolean | null + memberListVisibility: string | null + usersCount: number | null + }> + created: string + lastModified: string + }> + }>(url) + + const parameters = new URLSearchParams(queryParameters || '') + const pageIndex = Number.parseInt(parameters.get('p') || '0', 10) + const pageSize = Number.parseInt(parameters.get('l') || '10', 10) + + validationResults.value = result.results.map((item, index) => { + return { + row: pageIndex * pageSize + index + 1, + id: item.id, + status: item.status, + userName: item.userName, + email: item.emails || '', + eppn: item.eppn || '', + groups: item.groups, + code: item.code, + } + }) + + MissingUsers.value = (result.missingUser || []).map(user => ({ + id: user.id, + name: user.userName, + eppn: user.eppns[0] || '', + groups: user.groups.map(g => g.displayName || g.id), + })) + + summary.value = { + total: result.summary.create + result.summary.update + result.summary.delete + result.summary.skip + result.summary.error, + status: { + create: result.summary.create, + update: result.summary.update, + delete: result.summary.delete, + skip: result.summary.skip, + error: result.summary.error, + }, + } + + return result + } + catch (error) { + console.error('Failed to fetch validation results:', error) + throw error + } + } + + async function validateFile(taskIdValue: string) { + taskId.value = taskIdValue + await fetchValidationResults(taskIdValue) + } + + async function executeUpload() { + if (!taskId.value || !selectedRepository.value) { + throw new Error('Missing required data') + } + + isProcessing.value = true + + try { + const result = await $fetch<{ + history_id: string + task_id: string + temp_file_id?: string + }>(`/api/bulk/execute`, { + method: 'POST', + body: { + task_id: taskId.value, + temp_file_id: temporaryFileId.value, + repository_id: selectedRepository.value, + delete_users: selectedMissingUsers.value, + }, + }) + + const uploadTaskId = result.task_id + const uploadHistoryId = result.history_id + + await pollExecuteStatus(uploadTaskId) + + await fetchUploadtResult(uploadHistoryId) + + return { task_id: uploadTaskId, + history_id: uploadHistoryId, + } + } + finally { + isProcessing.value = false + } + } + + async function pollExecuteStatus(uploadTaskId: string) { + const maxAttempts = 100 + const interval = 2000 + + for (let index = 0; index < maxAttempts; index++) { + const res = await $fetch(`/api/bulk/execute/status/${uploadTaskId}`) + const st = (res.status) as string | undefined + + if (st === 'SUCCESS') return + if (st === 'FAILURE') throw new Error(res.error ?? 'Validation task failed') + + await new Promise(r => setTimeout(r, interval)) + } + + throw new Error('Validation timeout') + } + + async function fetchUploadtResult(historyIdValue: string, queryParameters?: string) { + try { + const url = queryParameters + ? `/api/bulk/result/${historyIdValue}?${queryParameters}` + : `/api/bulk/result/${historyIdValue}` + + const result = await $fetch<{ + results: Array<{ + id: string + eppn: string[] + emails: string[] + userName: string + groups: string[] + operation: string + status: string + message?: string + code?: string + }> + summary: { + create: number + update: number + delete: number + skip: number + error: number + } + fileInfo?: { + fileName: string + startedAt: string + completedAt: string + executedBy: string + } + }>(url) + + const parameters = new URLSearchParams(queryParameters || '') + const pageIndex = Number.parseInt(parameters.get('p') || '0', 10) + const pageSize = Number.parseInt(parameters.get('l') || '10', 10) + + importResult.value = { + results: result.results.map((item, index) => ({ + id: item.id, + row: pageIndex * pageSize + index + 1, + eppn: item.eppn, + emails: item.emails, + userName: item.userName, + groups: item.groups, + operation: item.operation as 'create' | 'update' | 'delete' | 'skip', + status: item.status as 'success' | 'failed', + message: item.message, + code: item.code, + })), + summary: { + total: result.summary.create + result.summary.update + result.summary.delete + result.summary.skip + result.summary.error, + success: result.summary.create + result.summary.update + result.summary.delete, + failed: result.summary.error, + create: result.summary.create, + update: result.summary.update, + delete: result.summary.delete, + skip: result.summary.skip, + }, + fileInfo: result.fileInfo, + } + + return importResult.value + } + catch (error) { + console.error('Failed to fetch import result:', error) + throw error + } + } + + function resetUpload() { + selectedFile.value = undefined + selectedRepository.value = undefined + validationResults.value = [] + MissingUsers.value = [] + selectedMissingUsers.value = [] + taskId.value = undefined + importResult.value = undefined + temporaryFileId.value = undefined + summary.value = { + total: 0, + status: { create: 0, update: 0, delete: 0, skip: 0, error: 0 }, + } + } + + return { + selectedFile, + selectedRepository, + validationResults, + MissingUsers, + selectedMissingUsers, + summary, + taskId, + tempFileId: temporaryFileId, + isProcessing, + importResult, + validateFile, + executeUpload, + fetchValidationResults, + fetchUploadtResult, + resetUpload, + } +} diff --git a/src/app/i18n/locales/en.json b/src/app/i18n/locales/en.json index 9bedee88..f8f567e0 100644 --- a/src/app/i18n/locales/en.json +++ b/src/app/i18n/locales/en.json @@ -1,8 +1,33 @@ { + "bulk": { + "about": "About bulk operations", + "about-description": "Users included in the file are newly registered or updated", + "about-description2": "Existing users are matched by id and updated if they match", + "about-description3": "You can choose to delete users who are not included in the file", + "about-description4": "Supported formats: TSV, CSV, Excel", + "description": "You can perform bulk operations on users belonging to a repository by uploading files.", + "empty": "sky", + "file-format-error": "The file format is not supported. \nChoose TSV, CSV, or Excel file.", + "missing_user": "User not included in file", + "select-repository": "Please select a repository", + "status": { + "create": "create", + "delete": "delete", + "error": "error", + "skip": "No updates", + "update": "update" + }, + "target-repository": "Operation target repository", + "title": "Bulk user operations", + "upload-field": "Select or drag files", + "upload-file": "upload file", + "view-all": "Show all" + }, "button": { "add": "Add", "cancel": "Cancel", "create-new": "Create new", + "next": "to the next", "delete": "Delete", "reload": "Reload", "save": "Save", @@ -341,6 +366,9 @@ } } }, + "table": { + "count": "subject" + }, "users": { "button": { "selected-users-actions": "Actions for selected users", diff --git a/src/app/i18n/locales/ja.json b/src/app/i18n/locales/ja.json index a752f5ee..2730b985 100644 --- a/src/app/i18n/locales/ja.json +++ b/src/app/i18n/locales/ja.json @@ -1,9 +1,34 @@ { + "bulk": { + "about": "一括操作について", + "about-description": "ファイルに含まれるユーザーは新規登録または更新されます。", + "about-description2": "既存ユーザーはidで照合し、一致する場合は更新されます", + "about-description3": "ファイルに含まれていないユーザーの削除を選択できます", + "about-description4": "対応形式: TSV, CSV, Excel", + "description": "ファイルのアップロードによりリポジトリに所属するユーザーを一括操作することができます。", + "empty": "空", + "file-format-error": "対応していないファイル形式です。TSV、CSV、またはExcelファイルを選択してください。", + "missing_user": "ファイルに含まれていないユーザー", + "select-repository": "リポジトリを選択してください", + "status": { + "create": "作成", + "delete": "削除", + "error": "エラー", + "skip": "更新なし", + "update": "更新" + }, + "target-repository": "操作対象リポジトリ", + "title": "ユーザー一括操作", + "upload-field": "ファイルを選択またはドラッグ&ドロップ", + "upload-file": "アップロードファイル", + "view-all": "すべて表示" + }, "button": { "add": "追加", "cancel": "取り消し", "create-new": "新規作成", "delete": "削除", + "next": "次へ", "reload": "再読み込み", "save": "保存", "update": "更新", diff --git a/src/app/pages/bulk/[id]/index.vue b/src/app/pages/bulk/[id]/index.vue new file mode 100644 index 00000000..4f3f1f96 --- /dev/null +++ b/src/app/pages/bulk/[id]/index.vue @@ -0,0 +1,79 @@ + + + diff --git a/src/app/pages/bulk/index.vue b/src/app/pages/bulk/index.vue new file mode 100644 index 00000000..b7a38b5c --- /dev/null +++ b/src/app/pages/bulk/index.vue @@ -0,0 +1,64 @@ + + + diff --git a/src/app/types/bulks.ts b/src/app/types/bulks.ts new file mode 100644 index 00000000..8a6e1b5b --- /dev/null +++ b/src/app/types/bulks.ts @@ -0,0 +1,92 @@ + type StatusType = 'create' | 'update' | 'delete' | 'skip' | 'error' + + interface ValidationResult { + row: number + id: string + status: StatusType + userName: string + eppn: string[] + emails: string[] + groups: string[] + code?: string +} + + interface MissingUser { + id: string + name: string + eppn: string + groups: string[] +} + + interface UploadResult { + row: number + id: string + status: StatusType + name: string + eppn: string + emails: string + groups: string[] + message?: string +} + + interface Summary { + total: number + status: { + create: number + update: number + delete: number + skip: number + error: number + } +} + + interface ImportResult { + id: string + row?: number + eppn: string | string[] + userName: string + emails: string[] + groups: string[] + status: StatusType + code?: string +} + + interface ImportResultResponse { + results: ImportResult[] + summary: { + total: number + success: number + failed: number + create: number + update: number + delete: number + skip: number + } + fileInfo?: { + fileName: string + startedAt: string + completedAt: string + executedBy: string + } +} + + interface ImportResultResponse { + results: ImportResult[] + summary: { + total: number + success: number + failed: number + create: number + update: number + delete: number + skip: number + } + fileInfo?: { + fileName: string + startedAt: string + completedAt: string + executedBy: string + } +} + +export type { StatusType, ValidationResult, MissingUser, UploadResult, Summary, ImportResult, ImportResultResponse } \ No newline at end of file diff --git a/src/server/api/bulk.py b/src/server/api/bulk.py new file mode 100644 index 00000000..12722074 --- /dev/null +++ b/src/server/api/bulk.py @@ -0,0 +1,219 @@ +# +# Copyright (C) 2025 National Institute of Informatics. +# + +"""API router for bulk endpoints.""" + +import traceback + +from pathlib import Path +from uuid import UUID, uuid7 + +from flask import Blueprint, current_app +from flask_pydantic import validate +from redis import exceptions + +from server.api.helpers import validate_files +from server.api.schemas import ( + BulkBody, + ErrorResponse, + TagetRepository, + UploadBody, + UploadFiles, + UploadQuery, +) + +# from server.config import config +from server.entities.bulk import ResultSummary, ValidateSummary +from server.services import bulks, history_table + + +config_temp_file_dir = "/code/uploads" + +bp = Blueprint("bulk", __name__) + + +STATUS_MAP = {0: "create", 1: "delete", 2: "error", 3: "skip", 4: "update"} + + +@bp.post("/upload-file") +@validate_files +@validate(response_by_alias=True) +def upload_file( + form: TagetRepository, files: UploadFiles +) -> tuple[BulkBody | ErrorResponse, int]: + """Upload a file for bulk processing. + + Args: + form (TagetRepository): Target repository ID for upload. + files (UploadFiles): File to upload. + + Returns: + BulkBody: The response containing task ID + ErrorResponse: The response containing task ID or error message. + """ + temp_id = uuid7() + temp_dir = Path(config_temp_file_dir) + current_app.logger.info("files %s", files) + original_filename = files.bulk_file.filename or "upload_file" + operator_id = "test-user" # current_user.id + operator_name = "test user" # current_user.user_name + new_filename = f"{temp_id}_{Path(original_filename).name}" + file_path = temp_dir / new_filename + try: + files.bulk_file.save(str(file_path)) + file_content = {"repositories": [{"id": form.repository_id}]} + history_table.create_file( + file_id=temp_id, file_path=str(file_path), file_content=file_content + ) + + bulks.delete_temporary_file.apply_async((str(temp_id),), countdown=3600) # pyright: ignore[reportCallIssue] + async_result = bulks.validate_upload_data.delay( + temp_file_id=temp_id, operator_id=operator_id, operator_name=operator_name + ) # pyright: ignore[reportCallIssue] + return BulkBody(task_id=async_result.id, temp_file_id=temp_id), 200 + except Exception as e: + traceback.print_exc() + return ErrorResponse(code="", message=str(e)), 400 + + +@bp.get("/validate/status/") +@validate(response_by_alias=True) +def validate_status(task_id: str) -> tuple[BulkBody | ErrorResponse, int]: + """Get the status of a validation task. + + Args: + task_id (str): The ID of the validation task. + + Returns: + BulkBody: The response containing task status + ErrorResponse: The response containing task status or error message + """ + try: + res = current_app.extensions["celery"].AsyncResult(task_id) + except exceptions.ConnectionError: + return ErrorResponse(code="", message=""), 500 + if not res: + return ErrorResponse(code="", message=f"{task_id} not found."), 404 + return BulkBody(status=res.state), 200 + + +@bp.get("/validate/result/") +@validate(response_by_alias=True) +def validate_result( + query: UploadQuery, + task_id: str, +) -> tuple[ValidateSummary | ErrorResponse, int]: + """Get the result of a validation task. + + Args: + query (UploadQuery): Query parameters for filtering results. + task_id (str): The ID of the validation task. + + Returns: + ValidateSummary: The response containing validation result + ErrorResponse: The response containing validation result or error message. + """ + try: + res = current_app.extensions["celery"].AsyncResult(task_id) + except exceptions.ConnectionError: + return ErrorResponse(code="", message=""), 500 + if not res: + return ErrorResponse(code="", message=f"{task_id} not found."), 404 + if not res.successful(): + return ErrorResponse(code="", message="Task not successful."), 400 + history_id = res.result + status_filters = [STATUS_MAP[c] for c in query.f] if query.f else [] + offset = query.p or 1 + size = query.l or 20 + return bulks.get_validate_result( + history_id=history_id, status_filter=status_filters, offset=offset, size=size + ), 200 + + +@bp.get("/missing-user-get/") +@validate(response_by_alias=True) +def missing_user_get(task_id: str) -> tuple[UploadBody | ErrorResponse, int]: + """Get the list of users not included in a validation task. + + Args: + task_id (str): The ID of the validation task. + + Returns: + list[UserDetail]: The response containing list of users not included + ErrorResponse: The response containing error message + """ + try: + res = current_app.extensions["celery"].AsyncResult(task_id) + except exceptions.ConnectionError: + return ErrorResponse(code="", message=""), 500 + if not res: + return ErrorResponse(code="", message=f"{task_id} not found."), 404 + if not res.successful(): + return ErrorResponse(code="", message="Task not successful."), 400 + history_id = res.result + return UploadBody(delete_users=bulks.get_missing_users(history_id)), 200 + + +@bp.post("/execute") +@validate(response_by_alias=True) +def execute(body: UploadBody) -> tuple[BulkBody | ErrorResponse, int]: + """Execute a bulk upload. + + Args: + body (UploadBody): The request body containing temporary ID, repository ID, task ID, and users to delete. + + Returns: + BulkBody: The response containing task ID + ErrorResponse: The response containing task ID or error message + """ + async_result = bulks.update_users.delay( # pyright: ignore[reportFunctionMemberAccess] + task_id=body.task_id, + temp_file_id=body.temp_file_id, + delete_users=body.delete_users, + ) + history_id = ( + history_table.get_history_by_file_id(body.temp_file_id).id + if body.temp_file_id + else None + ) + return BulkBody(task_id=async_result.id, history_id=history_id), 200 + + +@bp.get("/execute/status/") +@validate() +def execute_status(task_id: str) -> tuple[BulkBody | ErrorResponse, int]: + """Get the status of an execution task. + + Args: + task_id (str): The ID of the execution task. + + Returns: + str: The response containing task status + ErrorResponse: The response containing task status or error message + """ + res = current_app.extensions["celery"].AsyncResult(task_id) + return BulkBody(status=res.state), 200 + + +@bp.get("/result/") +@validate(response_by_alias=True) +def result( + history_id: UUID, query: UploadQuery +) -> tuple[ResultSummary | ErrorResponse, int]: + """Get the result of a bulk upload. + + Args: + history_id (UUID):ID of the history to get. + query(UPloadQuery): Query parameters for filtering results. + + Returns: + ResultSummary: Summary of displayed history If the get is successful + ErrorResponse: If the get is failde + """ + status_filter = [STATUS_MAP[c] for c in query.f] if query.f else [] + offset = query.p or 1 + size = query.l or 10 + return bulks.get_upload_result( + history_id=history_id, status_filter=status_filter, offset=offset, size=size + ), 200 diff --git a/src/server/api/schemas.py b/src/server/api/schemas.py index 4a5c938b..fbca875f 100644 --- a/src/server/api/schemas.py +++ b/src/server/api/schemas.py @@ -10,10 +10,13 @@ import typing as t from datetime import date +from uuid import UUID from pydantic import BaseModel, ConfigDict +from werkzeug.datastructures import FileStorage from server.entities.common import camel_case_config +from server.entities.user_detail import UserDetail ignore_extra_config = ConfigDict( @@ -196,3 +199,59 @@ class HistoryPublic(BaseModel): public: bool """Public status.""" + + +class TagetRepository(BaseModel): + repository_id: str + """ID of the target repository.""" + + model_config = camel_case_config + """Configure to use camelCase aliasing.""" + + +class UploadFiles(BaseModel): + """Schema for upload files.""" + + bulk_file: FileStorage + + model_config = ConfigDict(arbitrary_types_allowed=True) + + +class BulkBody(BaseModel): + """Body schema for bulk upload response.""" + + temp_file_id: UUID | None = None + """Temporary ID for the bulk upload session.""" + history_id: UUID | None = None + """History ID associated with the bulk upload.""" + task_id: str | None = None + """Task ID associated with the bulk upload.""" + status: str | None = None + """Status of the bulk upload.""" + + +class UploadBody(BaseModel): + """Body schema for upload requests.""" + + temp_file_id: UUID | None = None + """Temporary ID for the upload session.""" + repository_id: str | None = None + """ID of the target repository.""" + task_id: str | None = None + """Task ID associated with the upload.""" + delete_users: list[UserDetail] | None = None + """List of users whose files are to be deleted.""" + + +class UploadQuery(BaseModel): + """Query parameters for upload history data.""" + + f: list[int] | None = None + """Filter by status. + 0:create, 1:update, 2:delete, 3:skip, 4:error""" + + p: int | None = None + """Page number for pagination.""" + + l: int | None = None + """Number of users per page for pagination.""" diff --git a/src/server/db/history.py b/src/server/db/history.py index eb68acd0..92914a33 100644 --- a/src/server/db/history.py +++ b/src/server/db/history.py @@ -7,7 +7,7 @@ from __future__ import annotations import typing as t -import uuid # noqa: TC003 +import uuid from datetime import datetime # noqa: TC003 @@ -25,7 +25,7 @@ ) from sqlalchemy.dialects import postgresql from sqlalchemy.ext.mutable import MutableDict -from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy.orm import Mapped, mapped_column, relationship from .base import db @@ -43,6 +43,7 @@ class Files(db.Model): id: Mapped[uuid.UUID] = mapped_column( UUID, + default=uuid.uuid7, primary_key=True, ) """File identifier (UUID).""" @@ -57,10 +58,7 @@ class Files(db.Model): MutableDict.as_mutable(JSON().with_variant(postgresql.JSONB, "postgresql")), ) """Repositories, groups, and users contained in the file.""" - - __table_args__ = ( - Index("ix_files_file_content", "file_content", postgresql_using="gin"), - ) + __table_args__ = (Index(None, "file_content", postgresql_using="gin"),) class DownloadHistory(db.Model): @@ -70,6 +68,7 @@ class DownloadHistory(db.Model): id: Mapped[uuid.UUID] = mapped_column( UUID, + default=uuid.uuid7, primary_key=True, ) """History record ID (uuid.UUID).""" @@ -84,7 +83,7 @@ class DownloadHistory(db.Model): file_id: Mapped[uuid.UUID] = mapped_column( UUID, - ForeignKey("files.id"), + ForeignKey(Files.id), nullable=False, ) """Foreign key to files.id (downloaded file ID).""" @@ -115,6 +114,7 @@ class DownloadHistory(db.Model): nullable=True, ) """Self-referencing FK to the first download record (nullable).""" + file = relationship("Files") class UploadHistory(db.Model): @@ -122,15 +122,17 @@ class UploadHistory(db.Model): __tablename__ = "upload_history" - type Status = t.Literal["S", "P", "F"] + type Status = t.Literal["S", "P", "F", "C"] """Allowed status values for the upload history. - 'S': Success - 'P': In Progress - 'F': Failed + - 'C': Cancel """ id: Mapped[uuid.UUID] = mapped_column( UUID, + default=uuid.uuid7, primary_key=True, ) """History record ID (UUID).""" @@ -151,7 +153,7 @@ class UploadHistory(db.Model): file_id: Mapped[uuid.UUID] = mapped_column( UUID, - ForeignKey("files.id"), + ForeignKey(Files.id), nullable=False, ) """Foreign key to files.id (uploaded file ID).""" @@ -178,9 +180,10 @@ class UploadHistory(db.Model): status: Mapped[Status] = mapped_column( String(1), + default="C", nullable=False, ) - """Status: 'S' (success), 'F' (failure), 'P' (in progress).""" + """Status: 'C' (cancel) 'S' (success), 'F' (failure), 'P' (in progress).""" results: Mapped[dict[str, t.Any]] = mapped_column( MutableDict.as_mutable(JSON().with_variant(postgresql.JSONB, "postgresql")), @@ -188,10 +191,11 @@ class UploadHistory(db.Model): ) """upload result data.""" + file = relationship("Files") __table_args__ = ( CheckConstraint( status.in_(t.get_args(Status.__value__)), name="status", ), - Index("ix_upload_history_results", "results", postgresql_using="gin"), + Index(None, "results", postgresql_using="gin"), ) diff --git a/src/server/entities/bulk.py b/src/server/entities/bulk.py new file mode 100644 index 00000000..c5af827f --- /dev/null +++ b/src/server/entities/bulk.py @@ -0,0 +1,147 @@ +# +# Copyright (C) 2025 National Institute of Informatics. +# + +"""Models for Bulk entity for client side.""" + +import typing as t + +from datetime import datetime +from uuid import UUID + +from pydantic import BaseModel, EmailStr, RootModel + +from server.entities.common import camel_case_config, forbid_extra_config +from server.entities.user_detail import UserDetail + + +CSV_TO_FIELDS = { + "user_name": "user_name", + "groups[].id": "groups_ids", + "groups[].name": "groups_names", + "edu_person_principal_names[]": "eppns", + "emails[]": "emails", + "preferred_language": "preferred_language", +} + + +class RepositoryMember(BaseModel): + """Model for members of a repository.""" + + groups: set[str] + """The groups belonging to the repository.""" + + users: set[str] + """The users belonging to the repository.""" + + +class ValidateSummary(BaseModel): + """Model for summary of bulk validation result.""" + + results: list[CheckResult] + """The list of validation results for each user.""" + + summary: HistorySummary + + missing_user: list[UserDetail] = [] + + model_config = camel_case_config | forbid_extra_config + """Configure camelCase aliasing and forbid extra fields.""" + + +class HistorySummary(BaseModel): + """Summary of the history operation.""" + + create: int + """Number of created items.""" + update: int + """Number of updated items.""" + delete: int + """Number of deleted items.""" + skip: int + """Number of skipped items.""" + error: int + """Number of error items.""" + + model_config = camel_case_config | forbid_extra_config + """Configure camelCase aliasing and forbid extra fields.""" + + +class CheckResult(BaseModel): + """Model for result of validation check for each user.""" + + id: str + """The unique identifier for the user.""" + + eppn: list[str] + """The eduPersonPrincipalNames of the user.""" + + email: list[EmailStr] + """The e-mail of the user.""" + + user_name: str + """The username of the user.""" + + groups: set[str] + """The groups of the user.""" + + status: t.Literal["create", "update", "delete", "skip", "error"] + """The status of the validation check.""" + + code: str | None + """The code representing the result of the validation check.""" + + model_config = camel_case_config | forbid_extra_config + """Configure camelCase aliasing and forbid extra fields.""" + + +class ResultSummary(BaseModel): + """Model for summary of bulk upload result.""" + + results: list[CheckResult] + """The list of upload results for each user.""" + + summary: HistorySummary + + file_id: UUID + """The ID of the uploaded file.""" + + file_name: str + """The name of the uploaded file.""" + + operator: str + """The operator who performed the upload.""" + + start_timestamp: datetime + """The timestamp when the upload started.""" + + end_timestamp: datetime | None = None + """The timestamp when the upload ended.""" + + model_config = camel_case_config | forbid_extra_config + """Configure camelCase aliasing and forbid extra fields.""" + + +class UserAggregated(RootModel): + root: list[UserDetail] + + +class Aggregated(t.TypedDict): + summary: dict[str, int] + results: list[CheckResult] + missing_user: list[UserDetail] + + +class FileContent(t.TypedDict): + repositories: dict[str, str] + groups: dict[str, str] + users: dict[str, str] + + +class FileUserDict(t.TypedDict, total=False): + user_name: list[str] + groups_ids: list[str] + groups_names: list[str] + eppns: list[str] + emails: list[str] + preferred_language: list[str] diff --git a/src/server/services/bulks.py b/src/server/services/bulks.py new file mode 100644 index 00000000..af35bd8c --- /dev/null +++ b/src/server/services/bulks.py @@ -0,0 +1,666 @@ +# +# Copyright (C) 2025 National Institute of Informatics. +# + +"""Services for managing bulk.""" + +import csv +import re +import typing as t + +from datetime import UTC, datetime +from http import HTTPStatus +from itertools import zip_longest +from pathlib import Path +from uuid import UUID + +import openpyxl +import requests + +from celery import shared_task +from entities.map_user import EPPN, Email, Group, MapUser +from flask import current_app +from pydantic import ValidationError +from services import users + +from server.clients import bulks +from server.config import config +from server.entities.bulk import ( + CheckResult, + HistorySummary, + RepositoryMember, + ResultSummary, + UserAggregated, + ValidateSummary, +) +from server.entities.bulk_request import BulkOperation +from server.entities.map_error import MapError +from server.entities.patch_request import RemoveOperation +from server.entities.summaries import GroupSummary +from server.entities.user_detail import UserDetail +from server.exc import ( + OAuthTokenError, + ResourceInvalid, + ResourceNotFound, + UnexpectedResponseError, +) +from server.services import groups, history_table, utils +from server.services.token import get_access_token, get_client_secret + + +@shared_task() +def validate_upload_data( + operator_id: str, operator_name: str, temp_file_id: UUID +) -> UUID: + """Validate the uploaded file data for bulk operation. + + Args: + temp_file_id (UUID): The ID of the temporary file. + + Returns: + UUID: The ID of the upload history record. + + Raises: + ResourceNotFound:if + """ + record = history_table.get_file_by_id(temp_file_id) + file_path = record.file_path + repository_id = record.file_content["repositories"][0]["id"] + repository_member = get_repository_member(repository_id) + + file_users: list[UserDetail] = build_user_from_file(file_path).root + file_users_id = {u.id for u in file_users} + missing_user_ids = list(repository_member.users - file_users_id) + update_users_ids = list(repository_member.users & file_users_id) + + user_list = users.search( + utils.make_criteria_object("users", i=missing_user_ids), raw=True + ).resources + missing_users = [UserDetail.from_map_user(u) for u in user_list] + user_list = users.search( + utils.make_criteria_object("users", i=update_users_ids), raw=True + ).resources + repo_users = [UserDetail.from_map_user(u) for u in user_list] + repo_user_by_id: dict[str, UserDetail] = {cu.id: cu for cu in repo_users} + + count_create = 0 + count_update = 0 + count_delete = 0 + count_skip = 0 + count_error = 0 + check_results: list[CheckResult] = [] + for u in file_users: + code = None + user_group_ids = {g.id for g in u.groups} if u.groups else set() + + if not user_group_ids.issubset(repository_member.groups): + code = "E001" + check_results.append( + CheckResult( + id=u.id, + eppn=u.eppns or [], + user_name=u.user_name, + email=u.emails or [], + groups=user_group_ids, + status="error", + code=code, + ) + ) + count_error += 1 + continue + + if not check_value(u): + code = "E002" + check_results.append( + CheckResult( + id=u.id, + eppn=u.eppns or [], + user_name=u.user_name, + groups=user_group_ids, + email=u.emails or [], + status="error", + code=code, + ) + ) + count_error += 1 + continue + + repo_user = repo_user_by_id.get(u.id) + if repo_user is None: + check_results.append( + CheckResult( + id=u.id, + eppn=u.eppns or [], + user_name=u.user_name, + groups=user_group_ids, + email=u.emails or [], + status="create", + code=code, + ) + ) + count_create += 1 + continue + + repo_group_ids = {g.id for g in repo_user.groups} if repo_user.groups else set() + if repo_group_ids != user_group_ids: + status = "update" + count_update += 1 + else: + status = "skip" + count_skip += 1 + code = is_immutable_attribute(repo_user, u) + check_results.append( + CheckResult( + id=u.id, + eppn=repo_user.eppns or [], + user_name=repo_user.user_name, + email=u.emails or [], + groups=user_group_ids, + status=status, + code=code, + ) + ) + summary = HistorySummary( + create=count_create, + update=count_update, + delete=count_delete, + skip=count_skip, + error=count_error, + ) + results = ValidateSummary( + results=check_results, summary=summary, missing_user=missing_users + ) + return history_table.create_upload( + operator_id=operator_id, + operator_name=operator_name, + file_id=temp_file_id, + results=results.model_dump(mode="json"), + ) + + +def get_repository_member(repository_id: str) -> RepositoryMember: + """Get the members of the specified repository from mAP Core API. + + Args: + repository_id (str): The ID of the repository. + + Returns: + RepositoryMember: The members of the repository. + """ + result = groups.search( + utils.make_criteria_object("groups", r=[repository_id]), raw=True + ).resources + group_ids = {g.id for g in result} + user_ids = { + m.value for g in result if g.members for m in g.members if m.type == "User" + } + + return RepositoryMember(groups=group_ids, users=user_ids) + + +def read_file(file_path: str) -> t.Generator: + """Read a user data file and return a list of `UserDetail` instances. + + Args: + file_path (str): Path to the inpatch file containing user data. + + + Raises: + ResourceNotFound: If the file does not exist. + ResourceInvalid : If the format is unsupported or parsing fails. + """ + path = Path(file_path) + if not path.exists(): + error = f"{path}: File not found." + raise ResourceNotFound(error) + + iterator = None + suffix = path.suffix.lower() + if suffix in {".csv", ".tsv"}: + delimiter = "," if suffix == ".csv" else "\t" + with path.open("r", encoding="utf-8-sig", newline="") as f: + iterator = csv.reader(f, delimiter=delimiter) + yield iterator + elif suffix == ".xlsx": + wb = openpyxl.load_workbook(path, read_only=True, data_only=True) + ws = wb.active + if ws: + iterator = ws.iter_rows(values_only=True) + if iterator is None: + error = f"{path.suffix}: Unsupported file format." + raise ResourceInvalid(error) + + yield iterator + + +def build_user_from_file(file_path) -> UserAggregated: + try: + gen = read_file(file_path) + except Exception as e: + current_app.logger.error(e) + raise ResourceNotFound + it = next(gen) + header = [("" if h is None else str(h).strip()) for h in next(it)] + access_row = next(it) + id_idx = header.index("id") + idx_of = {name: i for i, name in enumerate(header)} + + data = {} + for row in it: + if row is None: + continue + r = list(row) + + rid = r[id_idx] + + bucket = data.setdefault(rid, {}) + for col, j in idx_of.items(): + if col != "id": + bucket.setdefault(col, []).append(r[j]) + return build_user_detail_from_dict(data) + + +def build_user_detail_from_dict( + data: dict[str, dict[str, list[str]]], +) -> UserAggregated: + """""" + users: list[UserDetail] = [] + + for raw_uid, columns in data.items(): + uid = str(raw_uid).strip() if raw_uid else "" + if not uid: + continue + + user_name: str = columns.get("user_name")[0] + preferred_language: str = columns.get("preferred_language")[0] + + eppns: set[str] = set(columns.get("edu_person_principal_names[]")) + emails: set[str] = set(columns.get("emails[]")) + + gid_list: list[str] = columns.get("groups[].id") + gname_list: list[str] = columns.get("groups[].name") + groups: list[GroupSummary] = [] + seen = set() + for gid, gname in zip_longest(gid_list, gname_list, fillvalue=None): + if not gid or gid in seen: + continue + seen.add(gid) + groups.append( + GroupSummary.model_construct( + id=gid, + display_name=gname, + ) + ) + + users.append( + UserDetail.model_construct( + id=uid, + eppns=eppns, + user_name=user_name, + emails=emails, + preferred_language=preferred_language, + groups=groups, + ) + ) + + return UserAggregated(root=users) + + +def check_value(user: UserDetail) -> bool: + """""" + if not re.compile(r"^[A-Za-z0-9]{1,50}$").fullmatch(user.id): + return False + return len(user.user_name) <= 50 + + +def is_immutable_attribute( + original: UserDetail, update_user: UserDetail +) -> ( + t.Literal["emails", "eppns", "preferred_language", "last_modified", "created"] + | None +): + """""" + if original.id != update_user.id: + raise ValueError + if original.emails != update_user.emails: + return "emails" + if original.eppns != update_user.eppns: + return "eppns" + if original.preferred_language != update_user.preferred_language: + return "preferred_language" + if original.last_modified != update_user.last_modified: + return "last_modified" + if original.created != update_user.created: + return "created" + return None + + +def get_validate_result( + history_id: UUID, status_filter: list[str], offset: int, size: int +) -> ValidateSummary: + """Get the validation result summary for the specified upload history ID. + + Args: + history_id (UUID): The ID of the upload history. + status_filter (list[str]): Filter for Displayed Status + offset (int): the offset for pagination + size (int): page size + + Returns: + ValidateSummary: The summary of the validation result. + """ + results = history_table.get_paginated_upload_results( + history_id, offset, size, status_filter + ) + current_app.logger.info("results %s", results) + check_results = [CheckResult.model_validate(it) for it in results] + + summary = history_table.get_upload_results(history_id, "summary") + missing_user = history_table.get_upload_results(history_id, "missingUser") + current_app.logger.info("results: %s", check_results) + current_app.logger.info("summary: %s", summary) + current_app.logger.info("missingUser: %s", missing_user) + return ValidateSummary.model_validate({ + "results": check_results, + "summary": summary, + "missingUser": missing_user or [], + }) + + +def get_missing_users(history_id: UUID) -> list[UserDetail]: + """Get the list of users not included in the bulk operation. + + Args: + history_id (UUID): The ID of the upload history. + + Returns: + list[UserDetail]: The list of users not included in the bulk operation. + """ + missing_user = history_table.get_upload_results(history_id, "missingUser") + if not isinstance(missing_user, list): + return [] + + return [UserDetail.model_validate(user_data) for user_data in missing_user] + + +@shared_task +def update_users( + task_id: str, temp_file_id: UUID, delete_users: list[UserDetail] +) -> UUID: + """Perform bulk update of users based on the validation results. + + Args: + task_id (str): The ID of the validation task. + temp_file_id (UUID): The ID of the temporary file. + delete_users (list[UserDetail]): The list of user detail to be deleted. + + Returns: + UUID: The ID of the upload history record. + + Raises: + ResourceNotFound: If the upload history does not exist. + requests.RequestException: If there is an error communicating with mAP Core API. + ValidationError: If there is an error parsing the response from mAP Core API. + OAuthTokenError: If there is an issue with the access token. + UnexpectedResponseError: If there is an unexpected response from mAP Core API. + ResourceInvalid: If there is an invalid resource error from mAP Core API. + ValueError: + """ + history_id = current_app.extensions["celery"].AsyncResult(task_id).result + upload_data = history_table.get_upload_by_id(history_id) + if not upload_data: + error = f"History not found: {history_id}" + raise ResourceNotFound(error) + repository_id = upload_data.file.file_content["repositories"][0]["id"] + check_results: list[CheckResult] = upload_data.results.get("results", []) + summary = upload_data.results.get("summary", {}) + if summary.get("error") > 0: + raise ValueError + repository_member = get_repository_member(repository_id) + + update_users_ids = list( + repository_member.users & {user.id for user in check_results} + ) + user_list = users.search( + utils.make_criteria_object("users", i=update_users_ids), raw=True + ).resources + repo_users = [UserDetail.from_map_user(u) for u in user_list] + repo_user_by_id: dict[str, UserDetail] = {cu.id: cu for cu in repo_users} + bulk_ops: list[BulkOperation] = [] + count_delete = summary.get("delete", 0) + for user in check_results: + match user.status: + case "create": + bulk_ops.append( + BulkOperation( + method="POST", + path="/Users", + data=build_map_user_from_check_result(user), + ) + ) + case "update": + original = repo_user_by_id.get(user.id) + if not original: + raise ValueError + update_user = build_user_detail_from_check_result(user) + pathc_ops = utils.build_patch_operations(original, update_user) + for patch_op in pathc_ops: + bulk_op = BulkOperation( + method="PATCH", + path=f"/Users/{user.id}", + data=patch_op, + ) + bulk_ops.append(bulk_op) + + for user in delete_users: + bulk_op = build_remove_user_path(user, repository_id) + bulk_ops.append(bulk_op) + check_results.append( + CheckResult( + id=user.id, + eppn=user.eppns or [], + user_name=user.user_name, + email=user.emails or [], + groups={g.id for g in user.groups} if user.groups else set(), + status="delete", + code=None, + ) + ) + count_delete += 1 + summary.update({"delete": count_delete}) + file_id = save_file(temp_file_id) + history_table.update_upload_status( + history_id=history_id, + status="P", + file_id=file_id, + new_results={"results": check_results, "summary": summary}, + ) + history_table.delete_file_by_id(temp_file_id) + try: + access_token = get_access_token() + client_secret = get_client_secret() + result = bulks.post( + bulk_ops, access_token=access_token, client_secret=client_secret + ) + except ( + requests.RequestException, + ValidationError, + OAuthTokenError, + UnexpectedResponseError, + ) as exc: + history_table.update_upload_status(history_id=history_id, status="F") + current_app.logger.error(exc) + raise + + if isinstance(result, MapError): + current_app.logger.info(result.detail) + raise ResourceInvalid(result.detail) + count_error = 0 + for i, operation in enumerate(result.operations): + if ( + operation.status + and int(operation.status) >= HTTPStatus.BAD_REQUEST + and i < len(check_results) + ): + check_results[i].status = "error" + count_error += 1 + + summary.update({"error": count_error}) + if count_error > 0: + history_table.update_upload_status( + history_id=history_id, + status="F", + new_results={"results": check_results, "summary": summary}, + ) + else: + history_table.update_upload_status( + history_id=history_id, + status="S", + ) + + return history_id + + +def save_file(temp_file_id: UUID) -> UUID: + """Save the temporary file as a permanent file for bulk operation. + + Args: + temp_file_id (UUID): The ID of the temporary file. + + Returns: + UUID: The ID of the saved permanent file. + + Raises: + ResourceNotFound: If the temporary file does not exist. + ResourceInvalid: If the file format is invalid. + """ + try: + files = history_table.get_file_by_id(temp_file_id) + repository_id = files.file_content["repositories"][0]["id"] + except Exception as e: + current_app.logger.error("Failed to retrieve temporary file: %s", e) + raise + + file_path = Path(files.file_path) + if file_path.parent != Path(config.temp_file_dir): + return files.id + if not file_path.exists(): + raise ResourceNotFound("") + if file_path.suffix not in {".csv", ".tsv", ".xlsx"}: + error = "not supported file format." + raise ResourceInvalid(error) + + target_dir = Path(config.file_dir) / datetime.now(UTC).strftime("%Y/%m") + target_dir.mkdir(parents=True, exist_ok=True) + target_path = target_dir / file_path.name + file_path.rename(target_path) + return history_table.create_file( + file_path=str(target_path), + file_content={"repositories": [{"id": repository_id}]}, + ) + + +def build_map_user_from_check_result(user: CheckResult) -> MapUser: + user_emails = [Email(value=email) for email in user.email] + user_eppns = [EPPN(value=eppn) for eppn in user.eppn] + user_groups = [Group(value=group_id) for group_id in user.groups] + return MapUser( + id=user.id, + user_name=user.user_name, + emails=user_emails, + edu_person_principal_names=user_eppns, + groups=user_groups, + ) + + +def build_user_detail_from_check_result(user: CheckResult) -> UserDetail: + user_groups = [GroupSummary(id=group_id) for group_id in user.groups] + return UserDetail( + id=user.id, + user_name=user.user_name, + emails=user.email, + eppns=user.eppn, + groups=user_groups, + ) + + +def build_remove_user_path(user: UserDetail, repository_id: str) -> BulkOperation: + """Remove the user from the group belong to the repository. + + Args: + user (UserDetail): + The user to be removed from the repository. + repository_id (str): + The ID of the repository from which the user will be removed. + + Returns: + BulkOperation: + request body representing the bulk remove operation. + """ + user_groups = [g.id for g in user.groups] if user.groups else [] + affiliations = utils.detect_affiliations(user_groups) + group_list = affiliations.groups + + condition = " or ".join( + f"(value eq '{g.group_id}')" + for g in group_list + if g.repository_id == repository_id + ) + + path = f"groups[{condition}]" + remove_op = RemoveOperation(path=path) + return BulkOperation(method="PATCH", path=f"Users/{user.id}", data=remove_op) + + +def get_upload_result( + history_id: UUID, status_filter: list[str], offset: int, size: int +) -> ResultSummary: + """Get the bulk operation result summary with filtering and pagination. + + Args: + history_id (UUID): The ID of the upload history. + status_filter (list[str]): The list of status filters to apply. + size (int): The number of items to return. + offset (int): The offset for pagination. + + Returns: + ResultSummary: The summary of the bulk operation result. + + Raises: + ResourceNotFound: If the upload history with the given ID does not exist. + """ + upload = history_table.get_upload_by_id(history_id) + if not upload: + raise ResourceNotFound(f"upload history not found: {history_id}") + + raw_results: list[dict] = history_table.get_paginated_upload_results( + history_id, offset, size, status_filter or [] + ) + + summary = history_table.get_upload_results(history_id, "summary") + + payload = { + "results": raw_results, + "summary": summary, + "fileId": upload.file_id, + "fileName": Path(upload.file.file_path).name, + "operator": upload.operator_name or upload.operator_id, + "startTimestamp": upload.timestamp, + "endTimestamp": upload.end_timestamp, + } + current_app.logger.info("payload: %s", payload) + return ResultSummary.model_validate(payload) + + +@shared_task() +def delete_temporary_file(temp_id: str) -> None: + """Delete the temporary file with the given ID. + + Args: + temp_id(str): delete file id + """ + temp_file_id = UUID(temp_id) + file_path = history_table.get_file_by_id(temp_file_id).file_path + if file_path and Path(file_path).exists(): + Path(file_path).unlink() + history_table.delete_file_by_id(temp_file_id) diff --git a/src/server/services/history_table.py b/src/server/services/history_table.py new file mode 100644 index 00000000..de2b6176 --- /dev/null +++ b/src/server/services/history_table.py @@ -0,0 +1,118 @@ +# +# Copyright (C) 2025 National Institute of Informatics. +# + +"""Services for managing history table.""" + +import typing as t + +from datetime import UTC, datetime +from uuid import UUID # noqa: TC003 + +from sqlalchemy import func +from sqlalchemy.orm import selectinload + +from server.db import db +from server.db.history import Files, UploadHistory + + +def get_upload_by_id(history_id: UUID): + return db.session.get( + UploadHistory, history_id, options=[selectinload(UploadHistory.file)] + ) + + +def get_upload_results(history_id: UUID, attribute: str): + result = ( + db.session + .query(UploadHistory.results[attribute]) + .filter(UploadHistory.id == history_id) + .first() + ) + return result[0] if result else {} + + +def get_paginated_upload_results( + history_id: UUID, offset: int, size: int, status_filter: list[str] +): + if offset < 1 or size < 1: + raise ValueError("Invalid offset or size") + + elements = func.jsonb_array_elements( + UploadHistory.results["results"] + ).column_valued("item") + query = ( + db.session + .query(elements) + .select_from(UploadHistory) + .filter(UploadHistory.id == history_id) + ) + + if status_filter: + query = query.filter(elements.op("->>")("status").in_(status_filter)) + + offset_val = (offset - 1) * size + + raw_results = query.limit(size).offset(offset_val).all() + + return [r[0] for r in raw_results] + + +def create_upload(file_id: UUID, results: dict, operator_id: str, operator_name: str): + history_record = UploadHistory() + history_record.file_id = file_id + history_record.results = results + history_record.operator_id = operator_id + history_record.operator_name = operator_name + db.session.add(history_record) + db.session.commit() + return history_record.id + + +def update_upload_status( + history_id: UUID, + status: t.Literal["P", "S", "F"], + new_results: dict | None = None, + file_id: UUID | None = None, +): + obj = db.session.get(UploadHistory, history_id) + if obj is None: + return + if new_results: + obj.results = new_results + + obj.status = status + now = datetime.now(UTC) + if status == "P": + obj.timestamp = now + else: + obj.end_timestamp = now + + if file_id: + obj.file_id = file_id + + db.session.commit() + + +def get_history_by_file_id(file_id: UUID): + return db.session.query(UploadHistory).filter_by(file_id=file_id).one() + + +def get_file_by_id(file_id: UUID): + return db.session.query(Files).filter_by(id=file_id).one() + + +def delete_file_by_id(file_id: UUID): + Files.query.filter(Files.id == file_id).delete() + db.session.commit() + + +def create_file(file_path: str, file_content: dict, file_id: UUID | None = None): + file_record = Files() + if file_id: + file_record.id = file_id + file_record.file_path = str(file_path) + file_record.file_content = file_content + db.session.add(file_record) + db.session.commit() + return file_record.id From 7a01ae038b43408e3ac5c791a3c47448b16144ec Mon Sep 17 00:00:00 2001 From: ivis-kosaka Date: Thu, 5 Feb 2026 02:43:50 +0000 Subject: [PATCH 02/47] fix: bulk view translation and add openpyxl library --- pyproject.toml | 1 + src/app/components/bulk/BulkResultStep.vue | 40 ++++-------- src/app/components/bulk/BulkUploadStep.vue | 15 +++-- .../components/bulk/BulkValidationStep.vue | 65 ++++++++++++------- src/app/composables/useBulk.ts | 26 ++++---- src/app/i18n/locales/en.json | 51 +++++++++++++-- src/app/i18n/locales/ja.json | 44 +++++++++++++ src/app/pages/bulk/[id]/index.vue | 21 +++--- src/app/pages/bulk/index.vue | 13 ++-- src/app/types/bulks.ts | 62 +++++------------- uv.lock | 23 +++++++ 11 files changed, 222 insertions(+), 139 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 0534d358..9e903a33 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "flask-login>=0.6.3", "flask-pydantic>=0.14.0", "flask-sqlalchemy>=3.1.1", + "openpyxl>=3.1.5", "psycopg[binary]>=3.3.2", "pydantic-settings>=2.12.0", "pydantic[email]>=2.12.5", diff --git a/src/app/components/bulk/BulkResultStep.vue b/src/app/components/bulk/BulkResultStep.vue index cde78791..374aaccc 100644 --- a/src/app/components/bulk/BulkResultStep.vue +++ b/src/app/components/bulk/BulkResultStep.vue @@ -35,7 +35,7 @@ const STATUS_CONFIG: Record[] = [ +const resultColumns: TableColumn[] = [ { accessorKey: 'row', header: $t('bulk.column.row'), @@ -78,9 +78,9 @@ const resultColumns: TableColumn[] = [ }, { accessorKey: 'status', - header: $t('bulk.status'), + header: $t('bulk.column.status'), cell: ({ row }) => { - const data = row.original as ImportResult + const data = row.original as UploadResult const status = data.status const message = data.code @@ -110,7 +110,7 @@ const filteredResults = computed(() => { return importResult.value.results } - return importResult.value.results.filter((result: ImportResult) => { + return importResult.value.results.filter((result: UploadResult) => { return selectedFilters.value.includes(result.status) }) }) @@ -123,41 +123,23 @@ const paginatedResults = computed(() => { const resultSummary = computed(() => ({ total: filteredResults.value.length, - success: importResult.value?.summary?.success || 0, - failed: importResult.value?.summary?.failed || 0, - create: importResult.value?.summary?.create || 0, - update: importResult.value?.summary?.update || 0, - delete: importResult.value?.summary?.delete || 0, - skip: importResult.value?.summary?.skip || 0, + create: importResult.value?.summary?.status.create || 0, + update: importResult.value?.summary?.status.update || 0, + delete: importResult.value?.summary?.status.delete || 0, + skip: importResult.value?.summary?.status.skip || 0, })) async function reloadResults(queryParameters?: string) { - if (!props.historyId) return + if (!properties.historyId) return try { - await fetchUploadtResult(props.historyId, queryParameters) + await fetchUploadtResult(properties.historyId, queryParameters) } catch (error) { console.error('Failed to reload import results:', error) } } -function changeResultPageSize(size: number) { - resultPagination.value.pageSize = size - resultPagination.value.pageIndex = 0 -} - -function toggleFilter(filter: string) { - const index = selectedFilters.value.indexOf(filter) - if (index === -1) { - selectedFilters.value.push(filter) - } - else { - selectedFilters.value.splice(index, 1) - } - resultPagination.value.pageIndex = 0 -} - function clearFilters() { selectedFilters.value = [] resultPagination.value.pageIndex = 0 @@ -167,7 +149,7 @@ function handleRestart() { emit('restart') } -const fileInfo = computed(() => importResult.value?.fileInfo || {}) +const fileInfo = computed(() => importResult.value?.fileInfo)