# -*- coding: UTF-8 -*- import requests import sys import json import random import string import os from csv import reader from datetime import datetime from user_test import get_user, get_all_users, delete_user from group_test import get_group_by_name, get_group, create_group, add_user_to_group from exam_test import create_exam, create_question, get_question, add_question_to_exam from category_test import create_material, get_material, create_category, get_category, add_material_to_category, update_category, create_question_category, get_question_category, questions_in_category, add_question_to_question_category, update_question_category, show_root_categories from learningmaterial_test import add_comment USERNAME_RANDOM_LENGTH = 10 from constants import WEBSITE_PREFIX override = False want_material = False want_user = False want_question = False if len(sys.argv) > 1: for arg in sys.argv: if arg == 'override': override = True if arg == 'material': want_material = True if arg == 'user': want_user = True if arg == 'question': want_question = True def _random_string(length): letters = string.ascii_letters return ''.join(random.choice(letters) for i in range(length)) def _text(resp): return json.loads(resp.text) # 创建 10 个 前缀为namePrefix 密码为password的 用户 def add_dummy_users(namePrefix = 'testuser-', password = 'test', number = 10): groups = add_groups() users = [] nicknames = ['小强', '小张', '小李', '大王'] firstNames = ['一', '二', '三', '四', '五'] lastNames = ['张', '王', '李', '杜', '靳'] for i in range(number): user_name = namePrefix + _random_string(USERNAME_RANDOM_LENGTH) # register_user(user_name, password) new_user = { 'userName': user_name, 'password': password, 'nickName': nicknames[i % len(nicknames)], 'studyId': 'U0000' + str(i), 'email': 'test@gmail.com', 'firstName': firstNames[i % len(firstNames)], 'lastName': lastNames[i % len(lastNames)], 'phone': '13060901234', 'birthday': '1999-01-01', } register_user(new_user) user = get_user(user_name) random_group = groups[random.randint(0, len(groups)-1)] add_user_to_group(random_group, _text(user)) users.append(_text(user)['userName']) print('created ' + str(len(users)) + ' users: ' + ','.join(users)) global all_users all_users = users return users # 创建 3个二级的group # orgtest-level1 # orgtest-level2-a orgtest-level2-b orgtest-level2-c def add_groups(namePrefix = 'orgtest-', number=3): # creating level1 group level1_name = namePrefix + 'level1-' + _random_string(3) level1_group_body = { 'name': level1_name } create_group(level1_group_body) level1_group = _text(get_group_by_name(level1_name)) groups = [] for i in range(number): level2_group_body = { 'name': namePrefix + 'level2-' + _random_string(3), 'parentGroupId': level1_group['id'] } resp = create_group(level2_group_body) groupId = _text(resp)['id'] groups.append(groupId) print ('created ' + str(len(groups)) + ' groups: ' + ','.join(groups)) return groups def randomly_pick_user(): global all_users rand_index = random.randint(0, len(all_users) - 1) return all_users[rand_index] def create_exams(namePrefix='测试考试-', number=2, tags=[]): exams = [] for i in range(number): new_exam = { 'name': namePrefix + _random_string(3), 'description': '可以回家做', 'duration': random.randint(0, 863) * 100, # seconds 'tags': tags, 'startTime': '2021-05-01T10:00:00' } resp = create_exam(new_exam) exams.append(_text(resp)['id']) print ('created ' + str(len(exams)) + ' exams: ' + ','.join(exams)) return exams def create_questions(namePrefix='qtest-', number=10, from_file=False): exams = [] if not from_file: exams = create_exams() questions = [] question_categories = create_question_categories() if from_file: files = os.listdir('./demo-questions/') for file_name in files: with open('./demo-questions/' + file_name, encoding="UTF-8") as read_obj: print('loading ' + file_name + ' to db') exam_tag = file_name.split('-')[1] exam_prefix = file_name.split('-')[0] exams = create_exams(namePrefix=exam_prefix + '-' + exam_tag, number=6, tags=[exam_tag]) csv_reader = reader(read_obj) for row in csv_reader: qNumber = row[0] if qNumber == '题号' or len(qNumber) < 1: continue content = row[1] # required temp_type = row[2] # required qType = 'DanXuan' if temp_type == '2': qType = 'DuoXuan' if temp_type == '3': qType = 'PanDuan' answers = [] # required for i in [3, 4, 5, 6]: if len(row[i]) > 0: answers.append(row[i]) finalAnswer = row[7] # required answerAnalysis = row[8] keyword = row[9]; level = row[10]; tags = [] if len(row[11]) > 0: if row[11] == '线路工': tags.append('XianLuGong') elif row[11] == '桥隧工': tags.append('QiaoSuiGong') elif row[11] == '大型线路机械司机': tags.append('DaXingXianLuJiXieSiJi') elif row[11] == '测量工': tags.append('CeLiangGong') if len(content) == 0 or len(qType) == 0 or len(answers) == 0 or len(finalAnswer) == 0: print('Skip due to required info missing') continue question = { 'userId': randomly_pick_user(), 'content': content, 'answers': answers, 'finalAnswer': [finalAnswer], 'keyword': keyword, 'type': qType, 'questionLevel': level, 'tags': tags, 'answerAnalysis': answerAnalysis, } resp = create_question(question) random_exam_id = exams[random.randint(0, len(exams)-1)] exam_to_question = { 'questionId': _text(resp)['id'] } # if random number equals to magic number, then add tag 今日必练 if random.randint(0, 100) == 3: add_tag(datetime.today().strftime('%Y%m%d'), questionId=_text(resp)['id']) # add newly created question to a random exam add_question_to_exam(random_exam_id, [exam_to_question]) questions.append(_text(resp)['id']) else: for i in range(number): question = { 'userId': randomly_pick_user(), 'content': namePrefix + '天上什么星星最亮?-' + str(i), 'answers': ['太阳', '小天狼星', '北极星'], 'finalAnswer': ['小天狼星'], 'type': 'DanXuan' # 单选题 } resp = create_question(question) random_exam_id = exams[random.randint(0, len(exams)-1)] # add newly created question to a random exam add_question_to_exam(random_exam_id, [_text(resp)]) questions.append(_text(resp)['id']) tiankong_question = { 'userId': randomly_pick_user(), 'content': namePrefix + '天上$PH$最亮?-' + str(i), 'finalAnswer': ['小天狼星'], 'type': 'TianKong' # 填空题 } resp = create_question(tiankong_question) random_exam_id = exams[random.randint(0, len(exams)-1)] # add newly created question to a random exam add_question_to_exam(random_exam_id, [_text(resp)]) random_category_id = question_categories[random.randint(0, len(question_categories)-1)] add_question_to_question_category(random_category_id, _text(resp)) questions.append(_text(resp)['id']) print ('created ' + str(len(questions)) + ' questions') return questions def create_question_categories(namePrefix= '知识类-', number=3): resp = create_question_category(namePrefix + _random_string(3)) question_root_id = _text(resp)['id'] categories = [] names = ['科普类-', '技术类-', '历史类-', '天文类-', '兴趣类-'] for i in range(number): child = create_question_category(names[i % len(names)] + _random_string(3)) child_id = _text(child)['id'] update_question_category(child_id, { 'parentId': question_root_id }) categories.append(child_id) print ('created ' + str(len(categories)) + ' question categories: ' + ','.join(categories)) return categories def create_categories(namePrefix= 'testCategory-', number=3): resp = create_category(namePrefix + 'root-' + _random_string(3)) root_id = _text(resp)['id'] categories = [] for i in range(number): child = create_category(namePrefix + 'child-' + _random_string(3)) child_id = _text(child)['id'] update_category(child_id, { 'parentId': root_id }) categories.append(child_id) print ('created ' + str(len(categories)) + ' categories: ' + ','.join(categories)) return categories def create_2_levels_categories(root_level, second_levels): resp = create_category(root_level) parent_id = _text(resp)['id'] children = [] for category_name in second_levels: child = create_category(category_name) child_id = _text(child)['id'] update_category(child_id, { 'parentId': parent_id }) children.append(child_id) return children def create_specific_categories(): categories = [] resp = show_root_categories() if len(_text(resp)) == 0: categories.extend(create_2_levels_categories('思政', ['党规党纪', '思想理论', '党史故事'])) categories.extend(create_2_levels_categories('安全', ['劳动安全', '作业安全', '事故案例'])) categories.extend(create_2_levels_categories('理论', ['基础理论', '专业理论'])) categories.extend(create_2_levels_categories('实作', ['作业指导书', '作业技巧', '应急处置'])) categories.extend(create_2_levels_categories('教材', ['文字', '音像', '课件'])) categories.extend(create_2_levels_categories('法纪', ['国家法律', '法纪法规', '交通规则'])) categories.extend(create_2_levels_categories('健康', ['健康知识', '健康养生', '心理辅导'])) categories.extend(create_2_levels_categories('集萃', ['旅游', '摄影','诗词', '名篇'])) else: print('Skip creating specific categories') root_categories = _text(resp) for root_category in root_categories: categories.append(root_category['id']) return categories def read_material_from_file(file_name): file = open('./demo-materials/' + file_name, encoding="UTF-8") file_lines = file.readlines() count = 0 content = '' title = '' for line in file_lines: line = line.strip() if len(line) < 1: continue # whitespaces if count == 0: title = line else: content += line + '\n' count+=1 return title, content def create_learning_materials(namePrefix='测试资料-', number=20, from_file=False): categories = [] if not from_file: categories = create_categories() materials = [] if from_file: files = os.listdir('./demo-materials/') categories = create_specific_categories() for file_name in files: title, content = read_material_from_file(file_name) material = create_material({ 'name': title, 'userId': randomly_pick_user(), 'type': 'ARTICLE', 'contents': content, }) materialId = _text(material)['id'] # if random number equals to magic number, then add tag 今日必练 if random.randint(0, 10) == 3: add_tag(datetime.today().strftime('%Y%m%d'), materialId=materialId) materials.append(materialId) random_category = categories[random.randint(0, len(categories)-1)] add_material_to_category(random_category, _text(material)) create_comments(materialId) else: for i in range(number): user = randomly_pick_user() material = create_material({ 'name': namePrefix + _random_string(5), 'userId': randomly_pick_user(), 'description': '好看的资料', 'type': 'ARTICLE', 'contents': '十年高考三年模拟-' + _random_string(20), 'links': [_random_string(3), _random_string(3), _random_string(3)] }) materialId = _text(material)['id'] materials.append(materialId) random_category = categories[random.randint(0, len(categories)-1)] add_material_to_category(random_category, _text(material)) create_comments(materialId) print ('created ' + str(len(materials)) + ' materials: ' + ','.join(materials)) return materials def create_comments(materialId, namePrefix='[自动生成]', number=5): dummy_comments = ['我去，这也太酷了！', '我太爱学这个了，老牛了', '这太有意思了，收藏了', '好文，顶一个', '我踩'] for i in range(number): comment = { 'userId': randomly_pick_user(), 'content': dummy_comments[i % len(dummy_comments)] } add_comment(materialId, comment) print('created ' + str(number) + ' comments to material: ' + materialId) def register_user(userData): url = WEBSITE_PREFIX + '/user/register' return requests.post(url, json=userData) def add_tag(tag, materialId=None, questionId=None, examId=None): url = WEBSITE_PREFIX + '/tags/' + tag + '?' if materialId: url += 'material=' + materialId if questionId: url += 'question=' + questionId if examId: url += 'exam=' + examId return requests.post(url) def remove_tag(tag, materialId=None, questionId=None, examId=None): url = WEBSITE_PREFIX + '/tags/' + tag + '?' if materialId: url += 'material=' + materialId if questionId: url += 'question=' + questionId if examId: url += 'exam=' + examId return requests.delete(url) def get_tag(tag): url = WEBSITE_PREFIX + '/tags/' + tag return requests.get(url) def delete_all_exams(): url = WEBSITE_PREFIX + '/admin/exams' return requests.delete(url) def delete_all_questions(): url = WEBSITE_PREFIX + '/admin/questions' return requests.delete(url) def delete_all_categories(): url = WEBSITE_PREFIX + '/admin/categories' return requests.delete(url) def delete_all_tags(): url = WEBSITE_PREFIX + '/admin/tags' return requests.delete(url) def delete_all_materials(): url = WEBSITE_PREFIX + '/admin/materials' return requests.delete(url) def delete_all_comments(): url = WEBSITE_PREFIX + '/admin/comments' return requests.delete(url) if override: all_users = json.loads(get_all_users().text) for user in all_users: delete_user(user['userName']) print('deleted all users') delete_all_exams() print('deleted all exams') delete_all_questions() print('deleted all questions') delete_all_categories() print('deleted all categories') delete_all_tags() print('deleted all tags') delete_all_materials() print('deleted all materials') delete_all_comments() print('deleted all comments') if want_user: add_dummy_users() else: print('Skip creating users') if want_question: create_questions(from_file=True) else: print('Skip creating questions') if want_material: create_learning_materials(from_file=True) else: print('Skip creating materials')