245 lines
7.1 KiB
Python
245 lines
7.1 KiB
Python
import requests
|
||
import http
|
||
import json
|
||
from typing import Dict, List, Optional
|
||
|
||
# 配置参数
|
||
CONFIG = {
|
||
"BASE_URL": "https://boss-api.shadow-rpa.net/boss/api/v1",
|
||
"TOKEN": "234e69f9-939d-44e0-8b8f-1560ba77e375",
|
||
"newDutyUuid": "a16edf00-c99f-44ae-bbb9-56d3e3629386",
|
||
}
|
||
|
||
|
||
def get_headers(token: Optional[str] = None) -> Dict:
|
||
"""
|
||
生成请求头
|
||
Args:
|
||
token: 认证token，如果为None则使用默认配置
|
||
Returns:
|
||
包含认证信息的请求头字典
|
||
"""
|
||
return {
|
||
"accept": "application/json",
|
||
"authorization": f"Bearer {token or CONFIG['TOKEN']}",
|
||
"content-type": "application/json;charset=UTF-8",
|
||
"origin": "https://boss.shadow-rpa.net",
|
||
"referer": "https://boss.shadow-rpa.net/",
|
||
}
|
||
|
||
|
||
def api_request(method: str, endpoint: str, **kwargs) -> Dict:
|
||
"""
|
||
统一的API请求处理
|
||
Args:
|
||
method: 请求方法（GET/POST）
|
||
endpoint: API端点
|
||
**kwargs: 请求参数
|
||
Returns:
|
||
API响应数据
|
||
"""
|
||
url = f"{CONFIG['BASE_URL']}/{endpoint}"
|
||
response = requests.request(method, url, **kwargs)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
|
||
def get_all_work_orders(page_size: int = 20, status: str = "new") -> List:
|
||
"""
|
||
获取所有工单列表
|
||
Args:
|
||
page_size: 每页数量
|
||
status: 工单状态
|
||
Returns:
|
||
工单列表
|
||
"""
|
||
all_orders = []
|
||
page = 1
|
||
|
||
while True:
|
||
params = {"page": page, "size": page_size, "status": status}
|
||
res = api_request(
|
||
"GET", "wo/order/woList", headers=get_headers(), params=params
|
||
)
|
||
|
||
data = res.get("data", {})
|
||
print(data)
|
||
pages = res.get("page", {}).get("pages", 0)
|
||
|
||
if not data:
|
||
break
|
||
|
||
all_orders.extend(data)
|
||
if page >= pages:
|
||
break
|
||
page += 1
|
||
# print(all_orders)
|
||
|
||
return all_orders
|
||
|
||
|
||
def process_work_order(order: Dict) -> Dict:
|
||
"""
|
||
处理单个工单信息，获取提交者信息
|
||
Args:
|
||
order: 工单数据
|
||
Returns:
|
||
处理后的工单信息
|
||
"""
|
||
result = {
|
||
"uuid": order["uuid"],
|
||
"title": order.get("title", "").strip(),
|
||
"content": order.get("content", "").strip(),
|
||
"submitter": order.get("submitterId"),
|
||
"enterprise": order.get("enterpriseName"),
|
||
"phone": None,
|
||
}
|
||
|
||
# 获取提交者信息
|
||
if result["submitter"]:
|
||
user_info = query_user_info(result["submitter"])
|
||
user_data = user_info.get("data", {}).get("dataSource", {}).get("data", [])
|
||
|
||
# 查找匹配的用户数据
|
||
matched_user = next(
|
||
(x for x in user_data if x.get("userName") == result["submitter"]), None
|
||
)
|
||
|
||
# 解密电话号码
|
||
if matched_user and matched_user.get("phone__l"):
|
||
encrypt_phone = matched_user["phone__l"].get("encrypte")
|
||
if encrypt_phone:
|
||
decrypted = decrypt_sensitive_data(encrypt_phone)
|
||
result["phone"] = decrypted.get("data", {}).get("decryptValue")
|
||
|
||
return result
|
||
|
||
|
||
def claim_work_order(work_order_uuid):
|
||
url = "https://boss-api.shadow-rpa.net/boss/api/v1/wo/order/claim"
|
||
headers = get_headers()
|
||
data = {"workOrderUuid": work_order_uuid}
|
||
response = requests.post(url, headers=headers, json=data)
|
||
return response.json()
|
||
|
||
|
||
def transfer_work_order(work_order_uuid, new_duty_uuid, remark=""):
|
||
url = "https://boss-api.shadow-rpa.net/boss/api/v1/wo/order/transfer"
|
||
headers = get_headers()
|
||
data = {
|
||
"workOrderUuid": work_order_uuid,
|
||
"remark": remark,
|
||
"newDutyUuid": new_duty_uuid,
|
||
}
|
||
response = requests.post(url, headers=headers, json=data)
|
||
return response.json()
|
||
|
||
|
||
def query_user_info(
|
||
fieldValues, table_code="userInfoList", page=1, size=20, filters=None
|
||
):
|
||
if filters is None:
|
||
filters = [
|
||
{"fieldName": "userName", "fieldValues": fieldValues, "filterType": "WD"}
|
||
]
|
||
|
||
url = "https://boss-api.shadow-rpa.net/boss/api/v1/sys/universal/component/table"
|
||
headers = get_headers()
|
||
data = {
|
||
"source": "",
|
||
"tableCode": table_code,
|
||
"pageQuery": {"page": page, "size": size, "filters": filters, "orders": []},
|
||
}
|
||
response = requests.post(url, headers=headers, json=data)
|
||
return response.json()
|
||
|
||
|
||
def decrypt_sensitive_data(
|
||
encrypt_value, code="userInfoList", code_type="table", field_name="phone"
|
||
):
|
||
url = (
|
||
f"https://boss-api.shadow-rpa.net/boss/api/v1/sys/operateLog/sensitiveDataQuery"
|
||
)
|
||
headers = get_headers()
|
||
params = {
|
||
"encryptValue": encrypt_value,
|
||
"code": code,
|
||
"codeType": code_type,
|
||
"fieldName": field_name,
|
||
}
|
||
response = requests.get(url, headers=headers, params=params)
|
||
return response.json()
|
||
|
||
|
||
def feishu_notify(text):
|
||
|
||
webhook_url = "/open-apis/bot/v2/hook/d6551d8c-d792-46e4-929d-ad0d1b031027"
|
||
Keys = "lPQwjE2TqMD0Q4q0zDlUY"
|
||
|
||
conn = http.client.HTTPSConnection("open.feishu.cn")
|
||
payload = json.dumps({"msg_type": "text", "content": {"text": text}})
|
||
headers = {"Content-Type": "application/json"}
|
||
conn.request("POST", webhook_url, payload, headers)
|
||
res = conn.getresponse()
|
||
data = res.read()
|
||
|
||
|
||
def main():
|
||
"""
|
||
主程序入口
|
||
"""
|
||
messages = []
|
||
try:
|
||
# 获取所有工单
|
||
work_orders = get_all_work_orders()
|
||
if not work_orders:
|
||
print("未找到工单")
|
||
|
||
# 处理每个工单
|
||
print(f"共找到 {len(work_orders)} 个工单")
|
||
|
||
for order in work_orders:
|
||
processed_order = process_work_order(order)
|
||
|
||
# 打印工单信息
|
||
print(f"\n{'='*40}")
|
||
print(f"工单ID: {processed_order['uuid']}")
|
||
print(f"标题: {processed_order['title']}")
|
||
print(f"内容: {processed_order['content']}")
|
||
print(f"提交者: {processed_order['submitter']}")
|
||
print(f"电话: {processed_order['phone'] or '未找到'}")
|
||
print(f"企业: {processed_order['enterprise'] or '未找到'}")
|
||
print(
|
||
f"邀约: 我是影刀的小伙伴, 想跟你沟通反馈的需求: {processed_order['content']}"
|
||
)
|
||
print(f"{'='*40}\n")
|
||
|
||
# 通知提交者
|
||
|
||
messages.append(
|
||
f"工单ID: {processed_order['uuid']}\n"
|
||
f"提交者: {processed_order['submitter']}\n"
|
||
f"电话: {processed_order['phone']}\n"
|
||
f"企业: {processed_order['enterprise'] or '未找到'}\n"
|
||
f"邀约: 我是影刀的小伙伴, 想跟你沟通反馈的需求: {processed_order['content']}\n\n"
|
||
)
|
||
|
||
transfer_work_order(
|
||
processed_order["uuid"],
|
||
CONFIG.get("newDutyUuid"),
|
||
)
|
||
# break
|
||
messages = "\n".join(messages)
|
||
if messages.strip():
|
||
print("发送飞书通知")
|
||
feishu_notify(text=messages)
|
||
|
||
except requests.RequestException as e:
|
||
print(f"请求发生错误: {e}")
|
||
except Exception as e:
|
||
print(f"程序发生错误: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|