# -*- coding: utf-8 -*- from gpt.chatgpt import get_answer_from_gpt, get_article_gpt_pydantic from gpt.gpt_check import CheckGptAnswer, CheckArticleResult from tools.new_mysql import MySQLUploader from tools.loglog import logger, log_err_e from tools.thread_pool_manager import pool_executor from common.common_data import all_exchange_words from common.split_text import split_text_to_word, get_article_words_count from pydantic import BaseModel from cachetools import TTLCache from concurrent.futures import wait from random import randint, shuffle, sample import json, time import requests from openpyxl import load_workbook from tenacity import retry, stop_after_attempt, wait_fixed import httpx import asyncio from threading import Lock from collections import defaultdict from fastapi import BackgroundTasks def get_article_difficulty(article) -> int: """获取文章的难度值""" url = "http://qbank.yunzhixue.cn/api/article/analysis" data = {"body": article, "question": ""} try: response = requests.post(url, json=data) except Exception as e: log_err_e(e, msg="获取文章难度值；") return 0 if response.status_code == 200: difficult_value = response.json()['data']['difficult'] return difficult_value else: logger.error(f"错误状态码{response.status_code}") def find_interval(number) -> int: """ 判断一个数字属于哪个难度等级区间。31级是例外情况，需要排查 :param number: 要检查的数字。 :return: 返回包含该数字的区间，如果没有找到，则返回 None。 """ intervals = [(1, 200), (201, 250), (251, 300), (301, 350), (351, 400), (401, 450), (451, 550), (551, 650), (651, 750), (751, 850), (851, 950), (951, 1100), (1101, 1250), (1251, 1400), (1401, 1550), (1551, 1700), (1701, 1900), (1901, 2100), (2101, 2300), (2301, 2600), (2601, 2900), (2901, 3200), (3201, 3500), (3501, 3900), (3901, 4300), (4301, 4700), (4701, 5100), (5101, 5500), (5501, 5900), (5901, 6500), (6501, 99999)] for index, (start, end) in enumerate(intervals, start=1): if start <= number <= end: return index logger.error(f"文章难度判断不对:{number}") return 0 def merge_and_split(list1, list2): combined = list1 + list2 import random random.shuffle(combined) two_thirds = [] one_third = [] total_length = len(combined) if total_length > 15: two_thirds = combined[:15] one_third = combined[15:] else: two_thirds = combined one_third = [] return two_thirds, one_third class GetArticle: def __init__(self): self.m = MySQLUploader() self.callback_url_dict = defaultdict(str) self.real_ip_dict = defaultdict(str) self.demo_name = defaultdict(str) self.article_result = {} self.punctuation = [",", ".", "!", "?", ":", ";", '"', "–", "_", "-", "...", "......"] all_exchange_words.update(self.punctuation) self.exchange_data: dict[str, list] = {} self.read_spring_bamboo_exchange_table() def read_spring_bamboo_exchange_table(self): """变形是键，原型是值""" wb = load_workbook(r"data/春笋单词对照变形.xlsx", read_only=True, data_only=True) ws = wb.active for row in ws.values: prototype = row[0] exchange = row[1] if prototype not in self.exchange_data: self.exchange_data[prototype] = [exchange] else: self.exchange_data[prototype].append(exchange) wb.close() def parser_insert_to_mysql(self, resp_result): try: for single_article in resp_result['articles']: article = single_article['body'] article_json = json.dumps(single_article) difficult_value = find_interval(get_article_difficulty(article)) if not difficult_value: logger.error("文章难度等级为0；") sql = "INSERT INTO spring_bamboo_article (article_json,difficult_level) VALUES (%s,%s)" self.m.execute_(sql, (article_json, difficult_value)) except Exception as e: logger.error(f"插入数据库时发生错误: {str(e)}") def submit_task(self, real_ip: str, core_words: list, take_count: int, demo_name: str, reading_level: int, article_length: int, exercise_id: int, background_tasks: BackgroundTasks): """ core_words: 词义数据组 take_count: 取文章数量 （int类型，正常是2篇，最大8篇） demo_name: 项目名称 reading_level:阅读等级 article_length:文章长度 exercise_id:学案id background_tasks: FastAPI的后台任务管理器 """ task_id = randint(10000000, 99999999) logger.info(f"reading-comprehension 生成文章id。学案id:{exercise_id},task_id:{task_id}") try: self.real_ip_dict[task_id] = real_ip self.demo_name[task_id] = demo_name resp_result = self.run_task(core_words, task_id, exercise_id, take_count, reading_level, article_length) background_tasks.add_task(self.parser_insert_to_mysql, resp_result) logger.success(f"reading-comprehension 文章2任务完成。学案id:{exercise_id},taskid:{task_id}") return resp_result except Exception as e: err_msg = f"GetArticle提交任务失败{type(e).__name__},{e}" log_err_e(e, msg="GetArticle提交任务失败;") return err_msg finally: self.real_ip_dict.pop(task_id, None) self.demo_name.pop(task_id, None) def __parse_gpt_resp(self, gpt_resp: dict, core_words: list): return_json = {"articles": []} for choice in gpt_resp["choices"]: single_article_dict = json.loads(choice["message"]["content"]) allWordAmount = 0 articleWordAmount = get_article_words_count(single_article_dict["englishArticle"]) allWordAmount += articleWordAmount for i in single_article_dict["questions"]: count_trunk = get_article_words_count(i["trunk"]) count_candidates = sum([get_article_words_count(ii["text"]) for ii in i["candidates"]]) allWordAmount += count_trunk allWordAmount += count_candidates usedMeanIds: list = single_article_dict['usedMeanIds'] article_words = split_text_to_word(single_article_dict['englishArticle']) for i in core_words: meaning_id = i.get('meaning_id', 0) if not meaning_id: continue word = i["spell"] if meaning_id not in usedMeanIds and word in self.exchange_data: words_exchanges_list = self.exchange_data[word] for exchange_word in words_exchanges_list: if exchange_word in article_words: usedMeanIds.append(meaning_id) break single_article_dict["body"] = single_article_dict.pop("englishArticle") single_article_dict["chinese"] = single_article_dict.pop("chineseArticle") for q in single_article_dict['questions']: data = q['candidates'] shuffled_candidates = sample(data, len(data)) labels = ['A', 'B', 'C', 'D'] for index, candidate in enumerate(shuffled_candidates): candidate['label'] = labels[index] q['candidates'] = shuffled_candidates return_json['articles'].append({**single_article_dict, "allWordAmount": allWordAmount, "articleWordAmount": articleWordAmount}) return return_json @retry(stop=stop_after_attempt(3), wait=wait_fixed(2), reraise=True) def get_article(self, core_words: list, task_id: int, exercise_id: int, reading_level, article_length, n) -> dict: if not article_length: if 0 < reading_level <= 10: article_length = 50 + 10 * reading_level elif 10 < reading_level <= 20: article_length = 150 + 30 * (reading_level - 10) else: article_length = 450 + 20 * (reading_level - 20) for index, (start, end) in enumerate([(1, 8), (9, 16), (17, 24), (24, 30)], start=1): if start <= reading_level <= end: difficulty_control_stage = index break else: difficulty_control_stage = 2 diffculty_control = { 1: {"grade": "小学", "desc_difficulty": "最简单最容易没有难度", "paragraph_count": "1-2", "desc2": "文章整体非常简洁，通俗易懂，适合初学者，刚入门，单词全是最常见的，语句通顺即可。", "choice_desc": "选择题难度尽可能简单，参考中国小学生水平"}, 2: {"grade": "初中", "desc_difficulty": "简单、常见、难度低", "paragraph_count": "2-3", "desc2": "文章整体难度适中，大约和中国初中生，中国CET-3，雅思4分这样的难度标准。", "choice_desc": "选择题难度适中，但是不要所有选择题让其直接在文中找到答案，参考中国初中生水平，中考标准。"}, 3: {"grade": "初中", "desc_difficulty": "简单、常见、难度低", "paragraph_count": "2-3", "desc2": "文章整体难度适中，大约和中国初中生，中国CET-3，雅思4分这样的难度标准。", "choice_desc": "选择题难度适中，但是不要所有选择题让其直接在文中找到答案，参考中国初中生水平，中考标准。"}, 4: {"grade": "高中", "desc_difficulty": "常见、高中难度的", "paragraph_count": "3-5", "desc2": "文章整体难度适中，大约和中国的高中生，中国CET-6，雅思6分这样的难度标准。", "choice_desc": "选择题难度偏难，要有迷惑性混淆性，答案不要出现直接在文中，4个选项要学生推理或逻辑判断，参考中国高中生水平，高考标准。"} } grade = diffculty_control[difficulty_control_stage]["grade"] select_diffculty = diffculty_control[difficulty_control_stage]["desc_difficulty"] select_paragraph_count = diffculty_control[difficulty_control_stage]["paragraph_count"] desc2 = diffculty_control[difficulty_control_stage]["desc2"] choice_desc = diffculty_control[difficulty_control_stage]["choice_desc"] shuffle(core_words) core_words_meaning_str = "; ".join([f"[{i['meaning_id']} {i['spell']} {i['meaning']}]" for i in core_words]) no_escape_code = r"\\n\\n" sys_prompt = "你是一个专业的英语老师，擅长根据用户提供的词汇生成对应的英语文章和中文翻译和4个配套选择题。" q = f"""下面我会为你提供一组数据，[单词组]（里面包含词义id，英语单词，中文词义），请根据这些单词的中文词义，\ 生成一篇带中文翻译的考场英语文章，英语文章和中文翻译要有[标题]。特别注意这个单词有多个词义时，生成的英语文章一定要用提供的中文词义，例如我提供单词[change 零钱]，就不要使用[变化]的词义。 要求: 1.必须用提供的这个词义的单词，其他单词使用{select_diffculty}的单词。{desc2}{choice_desc} 2.优先保证文章语句通顺，意思不要太生硬。不要为了使用特定的单词，造成文章语义前后不搭，允许不使用个别词义。 3.文章中使用提供单词，一定要和提供单词的中文词义匹配，尤其是一词多义时，务必使用提供单词的词义。必须要用提供单词的词义。如果用到的词义与提供单词词义不一致，请不要使用这个单词。 4.生成的文章要求{article_length}词左右，可以用{no_escape_code}字符分段，一般{select_paragraph_count}个段落左右。第一段是文章标题。不需要markdown格式。 5.允许不使用[单词组]的个别单词，优先保证文章整体意思通顺连贯和故事完整。 提供[单词组]：{core_words_meaning_str}; """ try: real_ip = self.real_ip_dict[task_id] demo_name = self.demo_name[task_id] gpt_resp = get_article_gpt_pydantic(q, temperature=1.2, real_ip=real_ip, demo_name=demo_name, model='gpt-4.1', check_fucn=CheckArticleResult.get_article_1, max_tokens=15000, sys_prompt=sys_prompt, n=n, task_id=task_id, exercise_id=exercise_id) multi_articles_dict = self.__parse_gpt_resp(gpt_resp=gpt_resp, core_words=core_words) return multi_articles_dict except httpx.HTTPError as e: logger.error(f"HTTP请求错误: {str(e)}") raise except json.JSONDecodeError as e: logger.error(f"JSON解析错误: {str(e)}") raise except Exception as e: log_err_e(e, f"gpt生成文章回复其他错误.") raise def run_get_article_task(self, core_words, task_id, exercise_id, take_count, reading_level, article_length) -> dict: """ :param core_words: 核心单词数据，优先级1;可能为空 :param task_id: 任务id :param take_count: 文章数量 :param reading_level:阅读等级 :param article_length:文章长度 :return: """ try: return_json = self.get_article(core_words, task_id, exercise_id, reading_level, article_length, n=take_count) return return_json except Exception as e: logger.error(f"运行文章任务时发生错误: {str(e)}") raise def run_task(self, core_words, task_id, exercise_id, take_count, reading_level, article_length): try: outside_json = self.run_get_article_task(core_words, task_id, exercise_id, take_count, reading_level, article_length) return outside_json except Exception as e: log_err_e(e, msg="外层总任务捕获错误") def cleanup(self): """清理所有资源""" pass