KSADASF/gemini-rproxy

1from flask import jsonify
2import logging
3import os
4
5formatter = logging.Formatter('%(message)s')
6logger = logging.getLogger(__name__)
7logger.setLevel(logging.INFO)
8handler = logging.StreamHandler()
9handler.setFormatter(formatter)
10logger.addHandler(handler)
11
12request_counts = {}
13
14password = os.environ['password']
15
16def authenticate_request(request):
17    auth_header = request.headers.get('Authorization')
18
19    if not auth_header:
20        return False, jsonify({'error': '缺少Authorization请求头'}), 401
21
22    try:
23        auth_type, pass_word = auth_header.split(' ', 1)
24    except ValueError:
25        return False, jsonify({'error': 'Authorization请求头格式错误'}), 401
26
27    if auth_type.lower() != 'bearer':
28        return False, jsonify({'error': 'Authorization类型必须为Bearer'}), 401
29
30    if pass_word != password:
31        return False, jsonify({'error': '未授权'}), 401
32
33    return True, None, None
34
35def process_messages_for_gemini(messages, use_system_prompt=False):
36    gemini_history = []
37    errors = []
38    system_instruction_text = ""
39    is_system_phase = use_system_prompt
40    
41
42    for i, message in enumerate(messages):
43        
44        role = message.get('role')
45        content = message.get('content')
46
47        if isinstance(content, str):
48            
49            if is_system_phase and role == 'system':
50                
51                if system_instruction_text:
52                    system_instruction_text += "\n" + content
53                else:
54                    system_instruction_text = content
55                
56            else:
57                is_system_phase = False
58                
59
60                if role in ['user', 'system']:
61                    role_to_use = 'user'
62                elif role == 'assistant':
63                    role_to_use = 'model'
64                else:
65                    errors.append(f"Invalid role: {role}")
66                    logger.error(f"Invalid role: {role}")
67                    continue
68
69                if gemini_history and gemini_history[-1]['role'] == role_to_use:
70                    gemini_history[-1]['parts'].append({"text": content})
71                    
72                else:
73                    gemini_history.append({"role": role_to_use, "parts": [{"text": content}]})
74                    
75
76        elif isinstance(content, list):
77            
78            parts = []
79            for item in content:
80                
81                if item.get('type') == 'text':
82                    parts.append({"text": item.get('text')})
83                    
84                elif item.get('type') == 'image_url':
85                    image_data = item.get('image_url', {}).get('url', '')
86                    
87                    if image_data.startswith('data:image/'):
88                        try:
89                            mime_type, base64_data = image_data.split(';')[0].split(':')[1], image_data.split(',')[1]
90                            parts.append({
91                                "inline_data": {
92                                    "mime_type": mime_type,
93                                    "data": base64_data
94                                }
95                            })
96                            
97                        except (IndexError, ValueError) as e:
98                            error_message = f"Invalid data URI for image: {image_data}. Error: {e}"
99                            errors.append(error_message)
100                            logger.error(error_message)
101                    else:
102                        error_message = f"Invalid image URL format for item: {item}. URL should start with 'data:image/' for inline images."
103                        errors.append(error_message)
104                        logger.error(error_message)
105                elif item.get('type') == 'file_url':
106                    file_data = item.get('file_url', {}).get('url', '')
107                    
108                    if file_data.startswith('data:'):
109                        try:
110                            mime_type, base64_data = file_data.split(';')[0].split(':')[1], file_data.split(',')[1]
111                            parts.append({
112                                "inline_data": {
113                                    "mime_type": mime_type,
114                                    "data": base64_data
115                                }
116                            })
117                            
118                        except (IndexError, ValueError) as e:
119                            error_message = f"Invalid data URI for file: {file_data}. Error: {e}"
120                            errors.append(error_message)
121                            logger.error(error_message)
122                    else:
123                        error_message = f"Invalid file URL format for item: {item}. URL should start with 'data:' for inline files."
124                        errors.append(error_message)
125                        logger.error(error_message)
126
127            if parts:
128                if role in ['user', 'system']:
129                    role_to_use = 'user'
130                elif role == 'assistant':
131                    role_to_use = 'model'
132                else:
133                    errors.append(f"Invalid role: {role}")
134                    logger.error(f"Invalid role: {role}")
135                    continue
136                if gemini_history and gemini_history[-1]['role'] == role_to_use:
137                    gemini_history[-1]['parts'].extend(parts)
138                    
139                else:
140                    gemini_history.append({"role": role_to_use, "parts": parts})
141                    
142
143    if errors:
144        logger.warning(f"Errors encountered during message processing: {errors}")
145        return gemini_history, {"parts": [{"text": system_instruction_text}]}, (jsonify({'error': errors}), 400)
146    else:
147        
148        return gemini_history, {"parts": [{"text": system_instruction_text}]}, None