Files
wechat_crawler/test_api.py

142 lines
3.9 KiB
Python
Raw Permalink Normal View History

# -*- coding: utf-8 -*-
"""
测试百炼API连接和岗位信息提取功能
"""
import json
import requests
# API配置
BAILIAN_API_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
API_KEY = "sk-46cb053d75eb4ad88713917ba0f1c81a"
# 测试消息样例
TEST_MESSAGES = [
"""
招聘Python后端开发工程师
公司北京某互联网科技有限公司
地点北京市海淀区中关村
薪资20K-35K
岗位职责
1. 负责公司核心业务系统的后端开发
2. 参与系统架构设计和优化
要求3年以上Python开发经验熟悉Django/Flask
联系人李经理
微信tech_hr_2024
""",
"""
急招前端开发
坐标上海浦东
月薪15-25k
我们公司是做电商的需要熟悉Vue3和React的前端
有意者联系张总 18612345678
""",
"""
今天天气真好大家周末愉快
"""
]
def test_extract(message_content):
"""测试提取功能"""
print("=" * 80)
print("测试消息:")
print(message_content.strip())
print("-" * 80)
prompt = f"""请从以下消息中提取招聘岗位信息并以JSON格式返回。如果消息不包含招聘信息返回空对象。
要提取的字段
- job_name: 工作名称
- job_description: 工作描述
- job_location: 工作地点
- salary_min: 月薪最低数字单位
- salary_max: 月薪最高数字单位
- company_name: 公司名称
- contact_person: 联系人
- contact_info: 联系方式电话/微信等
消息内容
{message_content}
请直接返回JSON格式不要包含其他说明文字"""
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
payload = {
"model": "qwen-plus",
"input": {
"messages": [
{
"role": "system",
"content": "你是一个专业的招聘信息提取助手,擅长从文本中提取结构化的岗位信息。"
},
{
"role": "user",
"content": prompt
}
]
},
"parameters": {
"result_format": "message"
}
}
try:
print("正在调用API...")
response = requests.post(BAILIAN_API_URL, headers=headers, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
if "output" in result and "choices" in result["output"]:
content = result["output"]["choices"][0]["message"]["content"]
print("API返回:")
print(content)
print("-" * 80)
# 尝试解析JSON
try:
job_data = json.loads(content)
if job_data and any(job_data.values()):
print("✓ 提取成功:")
print(json.dumps(job_data, ensure_ascii=False, indent=2))
else:
print("× 未提取到有效岗位信息")
except json.JSONDecodeError:
print(f"× JSON解析失败")
else:
print(f"× API返回格式异常: {result}")
except requests.exceptions.RequestException as e:
print(f"× API请求失败: {e}")
except Exception as e:
print(f"× 发生错误: {e}")
print()
def main():
print("百炼API岗位信息提取测试")
print("=" * 80)
print(f"API地址: {BAILIAN_API_URL}")
print(f"API密钥: {API_KEY[:20]}...")
print()
for i, msg in enumerate(TEST_MESSAGES, 1):
print(f"\n测试用例 {i}/{len(TEST_MESSAGES)}")
test_extract(msg)
if i < len(TEST_MESSAGES):
input("按回车继续下一个测试...")
print("=" * 80)
print("测试完成")
if __name__ == "__main__":
main()