算能 SophNet 免费 DeepSeek R1 满血 API

zezhiya

今天又发现一家免费的第三方 DeepSeek R1 API，注册就有免费额度 50¥（现在缩减到 20¥了，而且多了个邀请码机制，不过被邀请的没有额外额度
https://sophnet.com/#/model/overview
官网标参数如下：

实测速度非常快，比华为云和腾讯云 LKEAP 都快，推理信息放在 reasoning_content 字段。
懒得注册的自己玩，附上脚本

1. import requests
   
2. import time
   
3. from concurrent.futures import ThreadPoolExecutor
   
4. import logging
   
5. from datetime import datetime
   
6. 
   
7. # 配置日志
   
8. logging.basicConfig(
   
9.     level=logging.INFO,
   
10.     format='%(asctime)s - %(threadName)s - %(message)s'
    
11. )
    
12. 
    
13. # 定义常量
    
14. TOKEN = ''
    
15. PROJECT_ID = '887100'
    
16. PROJECT_TYPE = '1'
    
17. 
    
18. # 获取手机号码
    
19. 
    
20. def get_mobile():
    
21.     print('请求获取手机号码...')
    
22.     response = requests.post('http://h5.do889.com:81/api/get_mobile', data={
    
23.         'token': TOKEN + ';project_id=' + PROJECT_ID + ';project_type=' + PROJECT_TYPE + ';operator=0;loop=1',
    
24.     })
    
25.     print(f'获取手机号码请求的响应: {response.status_code}, 内容: {response.text}')
    
26.     mobile_data = response.json()
    
27.     return mobile_data.get('mobile')
    
28. 
    
29. # 发送验证码请求
    
30. 
    
31. def send_code(phone_number):
    
32.     print('请求发送验证码...')
    
33.     code_response = requests.post('https://sophnet.com/api/sys/code', json={
    
34.         'phoneNumber': phone_number
    
35.     })
    
36.     print(f'发送验证码请求的响应: {code_response.status_code}, 内容: {code_response.text}')
    
37.     return code_response
    
38. 
    
39. # 轮询获取短信
    
40. 
    
41. def poll_for_sms(phone_number):
    
42.     print('开始轮询获取短信...')
    
43.     start_time = time.time()  # 记录开始时间
    
44.     while True:
    
45.         message_response = requests.post('http://h5.do889.com:81/api/get_message', data={
    
46.             'token': TOKEN + ';phone_num=' + phone_number + ';project_type=' + PROJECT_TYPE + ';project_id=' + PROJECT_ID
    
47.         })
    
48.         print(f'获取短信请求的响应: {message_response.status_code}, 内容: {message_response.text}')
    
49.         message_data = message_response.json()
    
50.         if message_data['message'] == '短信还未到达,请继续获取':
    
51.             print('短信还未到达，继续轮询...')
    
52.             if time.time() - start_time > 30:  # 检查是否超过 30 秒
    
53.                 print('超过 30 秒未收到短信，跳过此次...')
    
54.                 return None
    
55.             time.sleep(5)  # 等待 5 秒后重试
    
56.         else:
    
57.             code = message_data['data'][0]['modle'].split('验证码：')[1].split('，')[0]
    
58.             print(f'获取到的验证码: {code}')
    
59.             return code
    
60. 
    
61. # 登录并获取 access token
    
62. 
    
63. def login(phone_number, code):
    
64.     print('请求登录并获取 access token...')
    
65.     login_response = requests.post('https://sophnet.com/api/sys/login/code', json={
    
66.         'phoneNumber': phone_number,
    
67.         'code': code,
    
68.         'password': ''
    
69.     })
    
70.     print(f'登录请求的响应: {login_response.status_code}, 内容: {login_response.text}')
    
71.     return login_response.json()['result']['token']
    
72. 
    
73. # 获取组织ID
    
74. 
    
75. def get_org_id(access_token):
    
76.     print('获取组织ID...')
    
77.     org_response = requests.get('https://sophnet.com/api/orgs/joined-org', headers={
    
78.         'Authorization': f'Bearer {access_token}'
    
79.     })
    
80.     print(f'获取组织ID响应: {org_response.status_code}, 内容: {org_response.text}')
    
81.     return org_response.json()['result']['list'][0]['id']
    
82. 
    
83. # 获取项目ID
    
84. 
    
85. def get_project_id(org_id, access_token):
    
86.     print('获取项目ID...')
    
87.     project_response = requests.get(f'https://sophnet.com/api/orgs/{org_id}/projects', params={
    
88.         'pageNum': 1,
    
89.         'pageSize': 99999
    
90.     }, headers={
    
91.         'Authorization': f'Bearer {access_token}'
    
92.     })
    
93.     print(f'获取项目ID响应: {project_response.status_code}, 内容: {project_response.text}')
    
94.     return project_response.json()['result']['list'][0]['id']
    
95. 
    
96. # 请求服务
    
97. 
    
98. def request_service(project_id, access_token, display_model_id):
    
99.     print('请求服务...')
    
100.     service_response = requests.post(f'https://sophnet.com/api/projects/{project_id}/service', json={
     
101.         "displayModelId": display_model_id,
     
102.         "llmType": "ChatCompletion",
     
103.         "quota": 200
     
104.     }, headers={
     
105.         'Authorization': f'Bearer {access_token}'
     
106.     })
     
107.     print(f'请求服务响应: {service_response.status_code}, 内容: {service_response.text}')
     
108.     return service_response
     
109. 
     
110. # 请求 API Key
     
111. 
     
112. def request_api_key(project_id, access_token):
     
113.     print('请求服务以获取 API Key...')
     
114.     api_key_response = requests.post(f'https://sophnet.com/api/projects/{project_id}/apikey', headers={
     
115.         'Authorization': f'Bearer {access_token}'
     
116.     }, json={
     
117.         'name': 'API Key'
     
118.     })
     
119.     print(f'获取 API Key 请求的响应: {api_key_response.status_code}, 内容: {api_key_response.text}')
     
120.     return api_key_response.json()['result']['apikey']
     
121. 
     
122. def process_single_flow():
     
123.     """执行单个完整的流程"""
     
124.     try:
     
125.         phone_number = get_mobile()
     
126.         if not phone_number:
     
127.             logging.info("获取手机号失败")
     
128.             return
     
129. 
     
130.         code_response = send_code(phone_number)
     
131.         if not code_response:
     
132.             logging.info(f"发送验证码失败: {phone_number}")
     
133.             return
     
134. 
     
135.         code = poll_for_sms(phone_number)
     
136.         if not code:
     
137.             logging.info(f"获取验证码失败: {phone_number}")
     
138.             return
     
139. 
     
140.         access_token = login(phone_number, code)
     
141.         if not access_token:
     
142.             logging.info(f"登录失败: {phone_number}")
     
143.             return
     
144. 
     
145.         org_id = get_org_id(access_token)
     
146.         if not org_id:
     
147.             logging.info(f"获取组织ID失败: {phone_number}")
     
148.             return
     
149. 
     
150.         project_id = get_project_id(org_id, access_token)
     
151.         if not project_id:
     
152.             logging.info(f"获取项目ID失败: {phone_number}")
     
153.             return
     
154. 
     
155.         # 并发执行服务请求和API密钥获取
     
156.         with ThreadPoolExecutor() as executor:
     
157.             # 先执行服务请求
     
158.             service_futures = [
     
159.                 executor.submit(request_service, project_id, access_token, 11),
     
160.                 executor.submit(request_service, project_id, access_token, 6)
     
161.             ]
     
162.             # 等待服务请求完成
     
163.             for future in service_futures:
     
164.                 service_response = future.result()
     
165.                 if not service_response:
     
166.                     logging.error("服务请求失败")
     
167.                     return False
     
168.             
     
169.             # 获取API密钥
     
170.             api_key_future = executor.submit(request_api_key, project_id, access_token)
     
171.             api_key = api_key_future.result()
     
172.             if api_key:
     
173.                 with open('result.txt', 'a') as f:
     
174.                     f.write(api_key + '\n')  # 保存 API 密钥
     
175.                 logging.info(f"API Key 已保存: {phone_number}")
     
176.             else:
     
177.                 logging.error("获取 API Key 失败")
     
178.                 return False
     
179. 
     
180.         logging.info(f"流程完成: {phone_number}")
     
181.         return True
     
182. 
     
183.     except Exception as e:
     
184.         logging.error(f"流程出错: {str(e)}")
     
185.         return False
     
186. 
     
187. def main(concurrent_flows=5):
     
188.     """
     
189.     主函数，控制并发流程数量，当一个流程完成后立即启动新的流程
     
190.     :param concurrent_flows: 最大并发流程数量
     
191.     """
     
192.     logging.info(f"开始运行，最大并发数: {concurrent_flows}")
     
193.    
     
194.     # 创建一个长期运行的线程池
     
195.     with ThreadPoolExecutor(max_workers=concurrent_flows) as executor:
     
196.         running_futures = set()
     
197.         
     
198.         while True:
     
199.             # 检查并移除已完成的任务
     
200.             done_futures = {f for f in running_futures if f.done()}
     
201.             for future in done_futures:
     
202.                 try:
     
203.                     result = future.result()
     
204.                     if result:
     
205.                         logging.info("一个流程成功完成")
     
206.                     else:
     
207.                         logging.info("一个流程失败")
     
208.                 except Exception as e:
     
209.                     logging.error(f"流程执行出错: {str(e)}")
     
210.                 running_futures.remove(future)
     
211.             
     
212.             # 当有空闲线程时，立即启动新的流程
     
213.             while len(running_futures) < concurrent_flows:
     
214.                 new_future = executor.submit(process_single_flow)
     
215.                 running_futures.add(new_future)
     
216.                 logging.info("启动新的流程")
     
217.             
     
218.             # 短暂休眠以避免CPU过度使用
     
219.             time.sleep(0.1)
     
220. 
     
221. if __name__ == "__main__":
     
222.     # 设置并发数量为5，您可以根据需要调整这个数字
     
223.     main(concurrent_flows=20)

复制代码

nicexi

备案的，高级模型还只有v3和r1感觉没必要