Spaces:
Sleeping
Sleeping
| import uuid | |
| import json | |
| import gradio as gr | |
| from datetime import datetime | |
| import urllib.parse | |
| import pytz | |
| class AssignmentService: | |
| def __init__(self, gcs_service): | |
| self.bucket_name = "ai_assignment_submission" | |
| self.gcs_service = gcs_service | |
| def create_assignment_id(self): | |
| while True: | |
| assignment_id = "assign_" + str(uuid.uuid4()) | |
| file_name = f"assignments/{assignment_id}.json" | |
| if not self.gcs_service.check_file_exists(self.bucket_name, file_name): | |
| return assignment_id | |
| def create_assignment_metadata(self, assignment_type, grade, topic, introduction, description, attach_materials, submission_deadline): | |
| taipei_tz = pytz.timezone('Asia/Taipei') | |
| if submission_deadline: | |
| submission_deadline = datetime.fromtimestamp(submission_deadline, tz=taipei_tz).isoformat() | |
| else: | |
| submission_deadline = None | |
| metadata = { | |
| "assignment_type": assignment_type, | |
| "grade": grade, | |
| "topic": topic, | |
| "introduction": introduction, | |
| "description": description, | |
| "attach_materials": attach_materials if attach_materials else None, | |
| "submission_deadline": submission_deadline | |
| } | |
| return metadata | |
| def create_assignment(self, user_data, user_email, user_nickname,assignment_type, metadata): | |
| assignment_id = self.create_assignment_id() | |
| timestamp_now = datetime.now(pytz.utc).astimezone(pytz.timezone('Asia/Taipei')).strftime("%Y-%m-%d %H:%M:%S") | |
| host_url = "https://www.junyiacademy.org/event/jutor_write_2024/?__theme=light" | |
| language_mapping = { | |
| "中文寫作 AI 批改": "chinese", | |
| "英文寫作 AI 批改": "english" | |
| } | |
| language = language_mapping.get(assignment_type, "") | |
| assignment_url = f"{host_url}&language={language}&assignment={assignment_id}" | |
| new_assignment = { | |
| "assignment_id": assignment_id, | |
| "assigner_data": user_data, | |
| "assigner_email": user_email, | |
| "assigner_nickname": user_nickname, | |
| "assignment_type": assignment_type, | |
| "metadata": metadata, | |
| "assignment_url": assignment_url, | |
| "submission_ids": [], | |
| "timestamp": timestamp_now | |
| } | |
| self.save_assignment_to_gcs(new_assignment) | |
| self.save_user_assignment_to_gcs(user_data, assignment_id) | |
| return new_assignment | |
| def get_assignment_url(self, assignment_id): | |
| assignment_data = self.get_assignment(assignment_id) | |
| if isinstance(assignment_data, str): | |
| print(f"错误：无法获取作业数据。返回的数据：{assignment_data}") | |
| return None | |
| assignment_url = assignment_data.get("assignment_url") | |
| if not assignment_url: | |
| print(f"警告：作业 {assignment_id} 没有 URL") | |
| return None | |
| return assignment_url | |
| def update_assignment(self, assignment_id, assignment_data): | |
| self.gcs_service.upload_json_string(self.bucket_name, f"assignments/{assignment_id}.json", json.dumps(assignment_data)) | |
| def save_assignment_to_gcs(self, assignment): | |
| file_name = f"assignments/{assignment['assignment_id']}.json" | |
| self.gcs_service.upload_json_string(self.bucket_name, file_name, json.dumps(assignment)) | |
| def save_user_assignment_to_gcs(self, user_id, assignment_id): | |
| user_assignments = self.get_user_assignments(user_id) | |
| assignment_data = self.get_assignment(assignment_id) | |
| user_assignments[assignment_id] = { | |
| "assignment_name": assignment_data["metadata"]["topic"], | |
| "assignment_type": assignment_data["assignment_type"], | |
| "timestamp": assignment_data["timestamp"] | |
| } | |
| encoded_user_id_url = urllib.parse.quote(user_id, safe='') | |
| self.gcs_service.upload_json_string(self.bucket_name, f"users/{encoded_user_id_url}/assignments.json", json.dumps(user_assignments)) | |
| def get_assignment(self, assignment_id): | |
| try: | |
| file_name = f"assignments/{assignment_id}.json" | |
| assignment_json = self.gcs_service.download_as_string(self.bucket_name, file_name) | |
| assignment_data = json.loads(assignment_json) | |
| except Exception as e: | |
| print(f"Error: {e}") | |
| return {} # 返回空字典而不是字串 | |
| return assignment_data | |
| def get_assignment_submissions(self, assignment_id): | |
| assignment_data = self.get_assignment(assignment_id) | |
| return assignment_data["submission_ids"] | |
| def get_user_assignments(self, user_id): | |
| if user_id is None or user_id == "": | |
| raise gr.Error("出現錯誤，請重新整理畫面") | |
| try: | |
| encoded_user_id_url = urllib.parse.quote(user_id, safe='') | |
| user_assignments_json = self.gcs_service.download_as_string(self.bucket_name, f"users/{encoded_user_id_url}/assignments.json") | |
| user_assignments = json.loads(user_assignments_json) | |
| except Exception as e: | |
| print(f"Error: {e}") | |
| user_assignments = {} | |
| return user_assignments | |
| def get_all_assignments(self): | |
| try: | |
| # 列出所有 'assignments/' 開頭的檔案 | |
| objects = self.gcs_service.list_files(self.bucket_name, prefix="assignments/") | |
| assignments = {} | |
| for obj in objects: | |
| if obj.endswith(".json"): | |
| assignment_id = obj.split("/")[-1].replace(".json", "") | |
| assignment_json = self.gcs_service.download_as_string(self.bucket_name, obj) | |
| assignments[assignment_id] = json.loads(assignment_json) | |
| except Exception as e: | |
| print(f"Error: {e}") | |
| assignments = {} | |
| return assignments | |
| def update_assignment_list(self, assigner_data): | |
| choices = [] | |
| if assigner_data == "admin": | |
| user_assignments = self.get_all_assignments() | |
| else: | |
| user_assignments = self.get_user_assignments(assigner_data) | |
| for assignment_id, assignment_info in user_assignments.items(): | |
| assignment_data = self.get_assignment(assignment_id) | |
| metadata = assignment_data.get("metadata", {}) | |
| topic = metadata.get("topic", "未命名作業") if metadata else "未命名作業" | |
| assignment_type = assignment_data.get("assignment_type", "未知類型") | |
| timestamp = assignment_data.get("timestamp", "未知日期") | |
| # 格式化日期 | |
| if timestamp != "未知日期": | |
| try: | |
| date = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S").strftime("%Y-%m-%d") | |
| except ValueError: | |
| date = "未知日期" | |
| else: | |
| date = "未知日期" | |
| # 創建格式化的選項 | |
| formatted_choice = f"{date}|{assignment_type}|{topic}" | |
| choice = (formatted_choice, assignment_id) | |
| choices.append(choice) | |
| # 按日期降序排序，將"未知日期"放在最後 | |
| choices.sort(key=lambda x: (x[0].split("|")[0] == "未知日期", x[0].split("|")[0]), reverse=True) | |
| return gr.update(choices=choices) | |
| def delete_assignment(self, assignment_id, user_data): | |
| try: | |
| # 首先檢查作業是否存在 | |
| file_name = f"assignments/{assignment_id}.json" | |
| if not self.gcs_service.check_file_exists(self.bucket_name, file_name): | |
| print(f"警告：作業文件 {file_name} 不存在") | |
| else: | |
| # 如果文件存在，嘗試刪除 | |
| if not self.gcs_service.delete_file(self.bucket_name, file_name): | |
| print(f"警告：無法刪除作業文件 {file_name}") | |
| # 從用戶的作業列表中刪除 | |
| user_assignments = self.get_user_assignments(user_data) | |
| if assignment_id in user_assignments: | |
| del user_assignments[assignment_id] | |
| encoded_user_id_url = urllib.parse.quote(user_data, safe='') | |
| self.gcs_service.upload_json_string(self.bucket_name, f"users/{encoded_user_id_url}/assignments.json", json.dumps(user_assignments)) | |
| print(f"已從用戶 {user_data} 的作業列表中刪除作業 {assignment_id}") | |
| else: | |
| print(f"警告：作業 {assignment_id} 不在用戶 {user_data} 的作業列表中") | |
| return True | |
| except Exception as e: | |
| print(f"刪除作業時發生錯誤：{e}") | |
| return False | |