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