import base64 from django.utils import timezone from rest_framework.parsers import MultiPartParser, FormParser from rest_framework.permissions import AllowAny from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi from celery.result import AsyncResult from chunyu_project.common_schemas import success_response, error_response, unauthorized_response, not_found_response from .info.baidu_lang_info import ( languages, auto_lang, cuid, mac, p_code, p_lang_type, p_target_lang_type, y_speech_type, pcm_y_lang_type, y_lang_type, y_target_lang_type, pcm_y_target_lang_type, text_languages_flat ) from .info.baidu_fanyi_appid import appid, appkey, endpoint from .baidu_response_codes import ( BaiduResponseCode, create_baidu_standardized_response, create_baidu_error_response ) from ..tasks import ( baidu_translate_task, baidu_recognize_lang_task, baidu_picture_translate_task, baidu_speech_recognize_task, record_translate_usage, ) from utils.safe_task import submit_task def get_client_ip(request): """获取客户端真实IP地址""" x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: return x_forwarded_for.split(',')[0].strip() return request.META.get('REMOTE_ADDR') class BaiduFanyiView(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='提交翻译任务', operation_description='提交文本翻译任务，返回task_id用于查询翻译结果', request_body=openapi.Schema( type=openapi.TYPE_OBJECT, required=['from_lang', 'to_lang', 'q'], properties={ 'from_lang': openapi.Schema(type=openapi.TYPE_STRING, description='源语言代码'), 'to_lang': openapi.Schema(type=openapi.TYPE_STRING, description='目标语言代码'), 'q': openapi.Schema(type=openapi.TYPE_STRING, description='待翻译文本（最多3000字符）'), } ), responses={202: success_response, 400: error_response} ) def post(self, request): from_lang = request.data.get('from_lang') to_lang = request.data.get('to_lang') if (from_lang not in languages) or (to_lang not in languages) or (to_lang == "auto"): return Response( {"message": "翻译语种异常"}, status=status.HTTP_400_BAD_REQUEST ) if from_lang == to_lang: return Response( {"message": "翻译数据异常"}, status=status.HTTP_400_BAD_REQUEST ) query = request.data.get('q') if not query: return Response( {"message": "翻译内容不能为空"}, status=status.HTTP_400_BAD_REQUEST ) if len(query) > 3000: return Response( create_baidu_error_response( code=BaiduResponseCode.TEXT_TOO_LONG, status_code=status.HTTP_400_BAD_REQUEST ), status=status.HTTP_400_BAD_REQUEST ) # 优先同步调用百度 API，成功后直接返回翻译结果，不依赖 Celery worker try: sync_result = baidu_translate_task(query, from_lang, to_lang) except Exception as e: sync_result = {'success': False, 'error_code': 'INNER_ERROR', 'error_msg': str(e)} if sync_result.get('success'): baidu_data = sync_result.get('data') or {} user_id = request.user.id if request.user.is_authenticated else None record_translate_usage.delay('text', from_lang, to_lang, len(query), user_id=user_id, ip_address=get_client_ip(request)) return Response({ "code": 0, "data": { "status": "SUCCESS", "ready": True, "trans_result": baidu_data.get('trans_result', []) } }, status=status.HTTP_200_OK) # 同步失败降级为异步（若 Celery worker 可用） try: task = submit_task(baidu_translate_task, query, from_lang, to_lang) task_id = task.id if task is not None else None user_id = request.user.id if request.user.is_authenticated else None record_translate_usage.delay('text', from_lang, to_lang, len(query), user_id=user_id, ip_address=get_client_ip(request)) return Response( { "task_id": task_id, "status": "pending", "message": "翻译任务已提交，请通过 task_id 查询结果" }, status=status.HTTP_202_ACCEPTED ) except Exception as e: return Response({ "code": 1, "message": sync_result.get('error_msg') or str(e) }, status=status.HTTP_502_BAD_GATEWAY) class AutoLangTypeViews(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='获取自动识别语言类型', operation_description='返回支持自动识别的语言类型列表', responses={200: success_response} ) def get(self, request): return Response(auto_lang, status=status.HTTP_200_OK) class AllLangTypeViews(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='获取所有语言类型', operation_description='返回所有支持的翻译语言类型列表', responses={200: success_response} ) def get(self, request): return Response(text_languages_flat, status=status.HTTP_200_OK) class PictureLangTypeViews(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='获取图片翻译语言类型', operation_description='返回图片翻译支持的源语言和目标语言类型列表', responses={200: success_response} ) def get(self, request): return Response({ "source_langs": p_lang_type, "target_langs": p_target_lang_type }, status=status.HTTP_200_OK) class SpeechLangTypeViews(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='获取语音翻译语言类型', operation_description='返回语音翻译支持的源语言和目标语言类型列表', manual_parameters=[ openapi.Parameter('speech_type', openapi.IN_QUERY, description='语音类型(mp3/wav/pcm)', type=openapi.TYPE_STRING), ], responses={200: success_response, 400: error_response} ) def get(self, request): speech_type = request.GET.get("speech_type", "mp3") if speech_type not in y_speech_type: return Response({"message": "不支持的语音类型"}, status=status.HTTP_400_BAD_REQUEST) if speech_type == "pcm": return Response({ "source_langs": pcm_y_lang_type, "target_langs": pcm_y_target_lang_type }, status=status.HTTP_200_OK) else: return Response({ "source_langs": y_lang_type, "target_langs": y_target_lang_type }, status=status.HTTP_200_OK) class RecognizeLangTypeViews(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='提交语种识别任务', operation_description='提交文本语种识别任务，返回task_id用于查询结果', request_body=openapi.Schema( type=openapi.TYPE_OBJECT, required=['q'], properties={ 'q': openapi.Schema(type=openapi.TYPE_STRING, description='待识别文本（最多3000字符）'), } ), responses={202: success_response, 400: error_response} ) def post(self, request): query = request.data.get("q") if not query: return Response({"message": "查询内容不能为空"}, status=status.HTTP_400_BAD_REQUEST) if len(query) > 3000: return Response({"message": "请求数据长度过长已经超过3000."}, status=status.HTTP_400_BAD_REQUEST) # 优先同步调用百度 API，避免依赖 Celery worker try: sync_result = baidu_recognize_lang_task(query) except Exception as e: sync_result = {'success': False, 'error_code': 'INNER_ERROR', 'error_msg': str(e)} if sync_result.get('success'): baidu_data = sync_result.get('data') or {} # 百度识别接口返回的字段可能是： # {'error_code':0, 'result': {'src': 'en'}} 或 {'error_code':0, 'data':[{'src':'en'}]} src_lang = None result = baidu_data.get('result') or baidu_data.get('data') if isinstance(result, dict): src_lang = result.get('src') elif isinstance(result, list) and result: src_lang = result[0].get('src') if isinstance(result[0], dict) else None if not src_lang and 'src' in baidu_data: src_lang = baidu_data.get('src') trans_result = [{'src': query, 'dst': src_lang}] if src_lang else [] user_id = request.user.id if request.user.is_authenticated else None record_translate_usage.delay('detect', 'auto', '', len(query), user_id=user_id, ip_address=get_client_ip(request)) return Response({ "code": 0, "data": { "status": "SUCCESS", "ready": True, "trans_result": trans_result, } }, status=status.HTTP_200_OK) try: task = submit_task(baidu_recognize_lang_task, query) task_id = task.id if task is not None else None user_id = request.user.id if request.user.is_authenticated else None record_translate_usage.delay('detect', 'auto', '', len(query), user_id=user_id, ip_address=get_client_ip(request)) return Response( { "task_id": task_id, "status": "pending", "message": "语种识别任务已提交，请通过 task_id 查询结果" }, status=status.HTTP_202_ACCEPTED ) except Exception as e: return Response({ "code": 1, "message": sync_result.get('error_msg') or str(e) }, status=status.HTTP_502_BAD_GATEWAY) class PictureRecognizeViews(APIView): permission_classes = [AllowAny] parser_classes = [MultiPartParser, FormParser] @swagger_auto_schema( tags=['API'], operation_summary='提交图片翻译任务', operation_description='上传图片进行翻译，返回task_id用于查询结果', manual_parameters=[ openapi.Parameter('from_lang', openapi.IN_QUERY, description='源语言代码', type=openapi.TYPE_STRING, required=True), openapi.Parameter('to_lang', openapi.IN_QUERY, description='目标语言代码', type=openapi.TYPE_STRING, required=True), openapi.Parameter('picture', openapi.IN_QUERY, description='图片类型', type=openapi.TYPE_STRING), openapi.Parameter('file', openapi.IN_FORM, description='待翻译的图片文件', type=openapi.TYPE_FILE, required=True), ], responses={202: success_response, 400: error_response} ) def post(self, request): file_data = request.FILES['file'].read() file_data_base64 = base64.b64encode(file_data).decode('utf-8') from_lang = request.GET.get("from_lang") to_lang = request.GET.get("to_lang") picture_type = request.GET.get("picture") if (from_lang == to_lang) or (to_lang == "auto"): return Response({"message": "翻译数据错误."}, status=status.HTTP_400_BAD_REQUEST) if (from_lang not in p_lang_type.keys()) or (to_lang not in p_lang_type.keys()): return Response({"message": "翻译语种异常."}, status=status.HTTP_400_BAD_REQUEST) try: task = submit_task(baidu_picture_translate_task, file_data_base64, from_lang, to_lang, picture_type) except Exception as e: return Response( {"code": 1, "message": "图片翻译任务提交失败，请稍后重试"}, status=status.HTTP_502_BAD_GATEWAY ) task_id = task.id if task is not None else None user_id = request.user.id if request.user.is_authenticated else None record_translate_usage.delay('image', from_lang, to_lang, len(file_data), user_id=user_id, ip_address=get_client_ip(request)) return Response( { "task_id": task_id, "status": "pending", "message": "图片翻译任务已提交，请通过 task_id 查询结果" }, status=status.HTTP_202_ACCEPTED ) class SpeechRecognitionView(APIView): permission_classes = [AllowAny] parser_classes = [MultiPartParser, FormParser] @swagger_auto_schema( tags=['API'], operation_summary='提交语音识别任务', operation_description='上传语音进行识别翻译，返回task_id用于查询结果', manual_parameters=[ openapi.Parameter('speech_type', openapi.IN_QUERY, description='语音类型', type=openapi.TYPE_STRING, required=True), openapi.Parameter('from_lang', openapi.IN_QUERY, description='源语言代码', type=openapi.TYPE_STRING, required=True), openapi.Parameter('to_lang', openapi.IN_QUERY, description='目标语言代码', type=openapi.TYPE_STRING, required=True), openapi.Parameter('voice', openapi.IN_FORM, description='待识别的语音文件', type=openapi.TYPE_FILE, required=True), ], responses={202: success_response, 400: error_response} ) def post(self, request): speech_type = request.GET.get("speech_type") if speech_type not in y_speech_type: return Response({"message": "不支持的语音类型."}, status=status.HTTP_400_BAD_REQUEST) from_lang = request.GET.get('from_lang') to_lang = request.GET.get('to_lang') if (from_lang == to_lang) or (to_lang == "auto"): return Response({"message": "翻译数据错误."}, status=status.HTTP_400_BAD_REQUEST) if (from_lang not in (pcm_y_lang_type.keys() if speech_type == "pcm" else y_lang_type.keys())) or (to_lang not in (pcm_y_lang_type.keys() if speech_type == "pcm" else y_lang_type.keys())): return Response({"message": "翻译语种异常."}, status=status.HTTP_400_BAD_REQUEST) voice = request.FILES.get("voice").read() audio_data_base64 = base64.b64encode(voice).decode('utf-8') try: task = submit_task(baidu_speech_recognize_task, audio_data_base64, from_lang, to_lang, speech_type) except Exception as e: return Response( {"code": 1, "message": "语音识别任务提交失败，请稍后重试"}, status=status.HTTP_502_BAD_GATEWAY ) task_id = task.id if task is not None else None user_id = request.user.id if request.user.is_authenticated else None record_translate_usage.delay('audio', from_lang, to_lang, len(voice), user_id=user_id, ip_address=get_client_ip(request)) return Response( { "task_id": task_id, "status": "pending", "message": "语音识别任务已提交，请通过 task_id 查询结果" }, status=status.HTTP_202_ACCEPTED ) class TaskResultView(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='查询翻译任务结果', operation_description='通过 task_id 查询异步翻译任务的执行结果', responses={200: '成功返回翻译结果'} ) def get(self, request, task_id): try: task_result = AsyncResult(task_id) except Exception: return Response( {"code": 1, "message": "任务查询失败"}, status=status.HTTP_400_BAD_REQUEST ) if not task_result.ready(): return Response({ "code": 0, "data": { "status": "PENDING", "ready": False, "trans_result": [] } }) if task_result.failed(): return Response({ "code": 1, "message": str(task_result.result) if task_result.result else "任务执行失败" }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) result_data = task_result.result if isinstance(result_data, dict) and result_data.get('success') is False: return Response({ "code": 1, "message": result_data.get('error_msg') or result_data.get('error_info') or "翻译失败" }, status=status.HTTP_400_BAD_REQUEST) baidu_result = result_data.get('data', {}) if isinstance(result_data, dict) else {} return Response({ "code": 0, "data": { "status": "SUCCESS", "ready": True, "trans_result": baidu_result.get('trans_result', []) } }) class TranslateUsageView(APIView): permission_classes = [AllowAny] @swagger_auto_schema( tags=['API'], operation_summary='获取翻译使用统计', operation_description='返回当前用户/IP的今日使用量、总量以及各类型使用统计', responses={200: success_response} ) def get(self, request): from ..models import TranslateUsage if request.user.is_authenticated: queryset = TranslateUsage.objects.filter(user=request.user) else: ip_address = request.META.get('REMOTE_ADDR') queryset = TranslateUsage.objects.filter(ip_address=ip_address) today_start = timezone.localdate() today_count = queryset.filter(created_at__date=today_start).count() total_count = queryset.count() by_type = {} for type_key, _ in TranslateUsage.TRANSLATE_TYPE_CHOICES: by_type[type_key] = queryset.filter(translate_type=type_key).count() return Response({ "code": 0, "data": { "today_count": today_count, "total_count": total_count, "by_type": by_type, } })