oscarz
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
import requests
|
|
import time
|
|
import json
|
|
import sys
|
|
|
|
# 企业微信相关信息
|
|
CORP_ID = 'ww5d7d350d9b8c0be3'
|
|
SECRET = 'YhagYQpaNIK9j1ATopgKNQhw3D13mpGZ64YVr23Je-A'
|
|
AGENT_ID = '1000003'
|
|
|
|
# 获取 access_token
|
|
def get_access_token():
|
|
url = f'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={SECRET}'
|
|
response = requests.get(url)
|
|
result = response.json()
|
|
if result.get('errcode') == 0:
|
|
return result.get('access_token')
|
|
else:
|
|
print(f"获取 access_token 失败: {result.get('errmsg')}")
|
|
return None
|
|
|
|
# 发送消息到企业微信
|
|
def send_message(access_token, message, touser=None, toparty=None, totag=None):
|
|
url = f'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}'
|
|
data = {
|
|
"msgtype": "text",
|
|
"agentid": AGENT_ID,
|
|
"text": {
|
|
"content": message
|
|
},
|
|
"safe": 0
|
|
}
|
|
if touser:
|
|
data["touser"] = touser
|
|
if toparty:
|
|
data["toparty"] = toparty
|
|
if totag:
|
|
data["totag"] = totag
|
|
|
|
response = requests.post(url, json=data)
|
|
result = response.json()
|
|
if result.get('errcode') == 0:
|
|
print("消息发送成功")
|
|
else:
|
|
print(f"消息发送失败: {result.get('errmsg')}")
|
|
|
|
def pretty_print_json(data, n=10, indent=4, sort_keys=False):
|
|
"""
|
|
以美化格式打印数组的前n个元素，其他元素用"..."表示
|
|
|
|
参数:
|
|
- data: 要打印的数据（应为数组）
|
|
- n: 要显示的元素数量
|
|
- indent: 缩进空格数
|
|
- sort_keys: 是否按键排序
|
|
"""
|
|
try:
|
|
# 处理非数组数据
|
|
if not isinstance(data, list):
|
|
formatted = json.dumps(data, indent=indent, ensure_ascii=False, sort_keys=sort_keys)
|
|
return formatted
|
|
|
|
# 复制原始数据，避免修改原数组
|
|
data_copy = data.copy()
|
|
|
|
# 切片取前n个元素
|
|
first_n_elements = data_copy[:n]
|
|
|
|
# 如果数组长度超过n，添加"..."标记
|
|
if len(data) > n:
|
|
result = first_n_elements + ["... ({} more elements)".format(len(data) - n)]
|
|
else:
|
|
result = first_n_elements
|
|
|
|
# 格式化输出
|
|
formatted = json.dumps(result, indent=indent, ensure_ascii=False, sort_keys=sort_keys)
|
|
return formatted
|
|
|
|
except TypeError as e:
|
|
print(f"错误：无法格式化数据。详情：{e}")
|
|
return str(data)
|
|
except Exception as e:
|
|
print(f"格式化时发生意外错误：{e}")
|
|
return str(data)
|
|
|
|
def is_json(s):
|
|
"""判断字符串是否可以解析为JSON"""
|
|
try:
|
|
json.loads(s)
|
|
return True
|
|
except json.JSONDecodeError:
|
|
return False
|
|
|
|
# 主函数
|
|
def main(report_content=None):
|
|
# 模拟数据报表内容
|
|
if report_content is None:
|
|
report_content = "这是第一行\n这是第二行\n这是第三行"
|
|
else:
|
|
# 处理转义字符
|
|
report_content = report_content.encode().decode('unicode_escape')
|
|
|
|
# 判断是否为JSON并格式化
|
|
if is_json(report_content):
|
|
try:
|
|
parsed_data = json.loads(report_content)
|
|
report_content = pretty_print_json(parsed_data)
|
|
except Exception as e:
|
|
print(f"JSON解析或格式化失败: {e}")
|
|
# 解析失败时保持原始内容
|
|
|
|
# 获取 access_token
|
|
access_token = get_access_token()
|
|
if access_token:
|
|
# 示例：发送给特定人员
|
|
send_message(access_token, report_content, touser='oscar')
|
|
# 示例：发送给特定部门
|
|
# send_message(access_token, report_content, toparty='department1|department2')
|
|
# 示例：发送给特定标签
|
|
# send_message(access_token, report_content, totag='tag1|tag2')
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1:
|
|
message = sys.argv[1]
|
|
main(message)
|
|
else:
|
|
main()
|