add workflow 市场易API联系人,dev
This commit is contained in:
root
parent
82a38c022f
commit
ee4eae17c3
|
|
@ -1,195 +1,195 @@
|
|||
# coding: utf-8
|
||||
import requests
|
||||
import json
|
||||
import datetime as dt
|
||||
import psycopg2
|
||||
import uuid
|
||||
|
||||
"""
|
||||
获取指定时间段前的时间
|
||||
:param h: 时间段
|
||||
:return: 时间
|
||||
"""
|
||||
|
||||
|
||||
def formatted2_previous_hour(h, format="%Y-%m-%d %H:%M:%S"):
|
||||
if h == 0:
|
||||
return dt.datetime.now().strftime(format)
|
||||
start_of_current_hour = dt.datetime.now().replace(minute=0, second=0, microsecond=0)
|
||||
# 减去一个小时，得到前一个小时的开始时间
|
||||
start_of_previous_hour = start_of_current_hour - dt.timedelta(hours=h)
|
||||
return start_of_previous_hour.strftime(format)
|
||||
|
||||
|
||||
"""
|
||||
获取token
|
||||
:return: token
|
||||
"""
|
||||
|
||||
|
||||
def get_token():
|
||||
url = "https://open.cloud.custouch.com/platform/cdp/token"
|
||||
token_payload = {
|
||||
"grant_type": "client_credentials",
|
||||
"app_key": "e9b240eb3a9848e89a96c5e7857794da",
|
||||
"app_secret": "f8cb7069e7dd468888e360bf8c259fc6",
|
||||
"scope": "openid",
|
||||
}
|
||||
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
||||
response = requests.request("POST", url, headers=headers, data=token_payload)
|
||||
if response.status_code != 200:
|
||||
raise Exception("获取token失败")
|
||||
|
||||
res = json.loads(response.text)
|
||||
return res["access_token"]
|
||||
|
||||
|
||||
"""
|
||||
获取API数据
|
||||
:param token: token
|
||||
:return: contact_ids
|
||||
"""
|
||||
|
||||
|
||||
def fetch_data(token, index=1, pageSize=2000):
|
||||
url = "https://open.cloud.custouch.com/platform/cdp/contact/event"
|
||||
params = {
|
||||
"connectIds": [],
|
||||
"eventMeta": [
|
||||
{"id": 200001},
|
||||
{"id": 200003},
|
||||
{"id": 200004},
|
||||
{"id": 201003},
|
||||
{"id": 204001},
|
||||
{"id": 204003},
|
||||
{"id": 204004},
|
||||
{"id": 206001},
|
||||
{"id": 209002},
|
||||
{"id": 214001},
|
||||
],
|
||||
"desc": True,
|
||||
"from": formatted2_previous_hour(1440, "%Y-%m-%dT%H:%M:%SZ"),
|
||||
"index": index,
|
||||
"size": pageSize
|
||||
}
|
||||
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
print(params)
|
||||
response = requests.request("POST", url, headers=headers, data=json.dumps(params))
|
||||
if response.status_code != 200:
|
||||
print(f"获取新建联系人ID失败，响应状态码：{response}")
|
||||
raise Exception("获取新建联系人ID失败")
|
||||
|
||||
res = json.loads(response.text)
|
||||
return res
|
||||
|
||||
|
||||
def fetch_all_data(token, pageSize=2000):
|
||||
"""
|
||||
获取所有分页数据
|
||||
:param token: 授权token
|
||||
:param pageSize: 每页大小，默认2000
|
||||
:return: 所有数据列表
|
||||
"""
|
||||
all_data = []
|
||||
index = 1
|
||||
|
||||
# 先获取第一页数据以确定总页数
|
||||
first_result = fetch_data(token, index, pageSize)
|
||||
total = first_result.get('total', 0)
|
||||
total_pages = (total + pageSize - 1) // pageSize # 向上取整计算总页数
|
||||
|
||||
print(f"总共 {total} 条数据，共 {total_pages} 页")
|
||||
|
||||
while True:
|
||||
# 如果不是第一页，获取当前页数据
|
||||
if index == 1:
|
||||
result = first_result
|
||||
else:
|
||||
result = fetch_data(token, index, pageSize)
|
||||
|
||||
# 提取当前页的数据
|
||||
current_data = result.get('data', [])
|
||||
if not current_data:
|
||||
print(f"第 {index} 页无数据，结束获取")
|
||||
break
|
||||
|
||||
all_data.extend(current_data)
|
||||
print(f"已获取第 {index}/{total_pages} 页，当前页 {len(current_data)} 条数据，累计 {len(all_data)} 条数据")
|
||||
|
||||
# 检查是否还有更多数据
|
||||
if len(all_data) >= total:
|
||||
print("已获取所有数据")
|
||||
break
|
||||
|
||||
index += 1
|
||||
|
||||
return all_data
|
||||
|
||||
PG_DSN = dict(
|
||||
database="dataops_db",
|
||||
user="dbuser_dba",
|
||||
password="EmBRxnmmjnE3",
|
||||
host="124.221.232.219",
|
||||
port="5432",
|
||||
)
|
||||
|
||||
|
||||
def save_json_to_pg(data: list, api_id: str) -> None:
|
||||
"""把列表落库：先软删历史，再插入新批次"""
|
||||
print("[save_to_pg] 写入 PG...")
|
||||
sql = """
|
||||
UPDATE data_api.api_data
|
||||
SET is_loaded = '1'
|
||||
WHERE api_id = %s;
|
||||
|
||||
INSERT INTO data_api.api_data
|
||||
(id, api_id, data, total_num, is_loaded, status,
|
||||
request_tm, execute_tm, remark)
|
||||
VALUES (%s, %s, %s, %s, '0', '0',
|
||||
current_timestamp(0), current_timestamp(0), '');
|
||||
"""
|
||||
|
||||
try:
|
||||
with psycopg2.connect(**PG_DSN) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
sql,
|
||||
(
|
||||
api_id,
|
||||
str(uuid.uuid4()),
|
||||
api_id,
|
||||
json.dumps(data, ensure_ascii=False),
|
||||
len(data),
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
cur.close()
|
||||
print("[save_to_pg] 写入完成")
|
||||
except psycopg2.Error as e:
|
||||
print(f"[save_to_pg] 数据库错误: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[save_to_pg] 未知错误: {e}")
|
||||
raise
|
||||
finally:
|
||||
if "conn" in locals():
|
||||
conn.close()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""主流程"""
|
||||
# print(get_token())
|
||||
print(f"开始请求新建联系人信息:{formatted2_previous_hour(0)}")
|
||||
token = get_token()
|
||||
# print(token)
|
||||
# 获取新建联系人ID
|
||||
print(f"开始请求新建联系人ID:{formatted2_previous_hour(0)}")
|
||||
objs = fetch_all_data(token)
|
||||
# 保存联系人ID
|
||||
apiId = "a7757b4a-7038-40ef-b11e-81a2c5e0"
|
||||
save_json_to_pg(objs, apiId)
|
||||
print(f"结束请求联系人详情:{formatted2_previous_hour(0)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
# coding: utf-8
|
||||
import requests
|
||||
import json
|
||||
import datetime as dt
|
||||
import psycopg2
|
||||
import uuid
|
||||
|
||||
"""
|
||||
获取指定时间段前的时间
|
||||
:param h: 时间段
|
||||
:return: 时间
|
||||
"""
|
||||
|
||||
|
||||
def formatted2_previous_hour(h, format="%Y-%m-%d %H:%M:%S"):
|
||||
if h == 0:
|
||||
return dt.datetime.now().strftime(format)
|
||||
start_of_current_hour = dt.datetime.now().replace(minute=0, second=0, microsecond=0)
|
||||
# 减去一个小时，得到前一个小时的开始时间
|
||||
start_of_previous_hour = start_of_current_hour - dt.timedelta(hours=h)
|
||||
return start_of_previous_hour.strftime(format)
|
||||
|
||||
|
||||
"""
|
||||
获取token
|
||||
:return: token
|
||||
"""
|
||||
|
||||
|
||||
def get_token():
|
||||
url = "https://open.cloud.custouch.com/platform/cdp/token"
|
||||
token_payload = {
|
||||
"grant_type": "client_credentials",
|
||||
"app_key": "e9b240eb3a9848e89a96c5e7857794da",
|
||||
"app_secret": "f8cb7069e7dd468888e360bf8c259fc6",
|
||||
"scope": "openid",
|
||||
}
|
||||
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
||||
response = requests.request("POST", url, headers=headers, data=token_payload)
|
||||
if response.status_code != 200:
|
||||
raise Exception("获取token失败")
|
||||
|
||||
res = json.loads(response.text)
|
||||
return res["access_token"]
|
||||
|
||||
|
||||
"""
|
||||
获取API数据
|
||||
:param token: token
|
||||
:return: contact_ids
|
||||
"""
|
||||
|
||||
|
||||
def fetch_data(token, index=1, pageSize=2000):
|
||||
url = "https://open.cloud.custouch.com/platform/cdp/contact/event"
|
||||
params = {
|
||||
"connectIds": [],
|
||||
"eventMeta": [
|
||||
{"id": 200001},
|
||||
{"id": 200003},
|
||||
{"id": 200004},
|
||||
{"id": 201003},
|
||||
{"id": 204001},
|
||||
{"id": 204003},
|
||||
{"id": 204004},
|
||||
{"id": 206001},
|
||||
{"id": 209002},
|
||||
{"id": 214001},
|
||||
],
|
||||
"desc": True,
|
||||
"from": formatted2_previous_hour(1440, "%Y-%m-%dT%H:%M:%SZ"),
|
||||
"index": index,
|
||||
"size": pageSize
|
||||
}
|
||||
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
||||
print(params)
|
||||
response = requests.request("POST", url, headers=headers, data=json.dumps(params))
|
||||
if response.status_code != 200:
|
||||
print(f"获取新建联系人ID失败，响应状态码：{response}")
|
||||
raise Exception("获取新建联系人ID失败")
|
||||
|
||||
res = json.loads(response.text)
|
||||
return res
|
||||
|
||||
|
||||
def fetch_all_data(token, pageSize=2000):
|
||||
"""
|
||||
获取所有分页数据
|
||||
:param token: 授权token
|
||||
:param pageSize: 每页大小，默认2000
|
||||
:return: 所有数据列表
|
||||
"""
|
||||
all_data = []
|
||||
index = 1
|
||||
|
||||
# 先获取第一页数据以确定总页数
|
||||
first_result = fetch_data(token, index, pageSize)
|
||||
total = first_result.get('total', 0)
|
||||
total_pages = (total + pageSize - 1) // pageSize # 向上取整计算总页数
|
||||
|
||||
print(f"总共 {total} 条数据，共 {total_pages} 页")
|
||||
|
||||
while True:
|
||||
# 如果不是第一页，获取当前页数据
|
||||
if index == 1:
|
||||
result = first_result
|
||||
else:
|
||||
result = fetch_data(token, index, pageSize)
|
||||
|
||||
# 提取当前页的数据
|
||||
current_data = result.get('data', [])
|
||||
if not current_data:
|
||||
print(f"第 {index} 页无数据，结束获取")
|
||||
break
|
||||
|
||||
all_data.extend(current_data)
|
||||
print(f"已获取第 {index}/{total_pages} 页，当前页 {len(current_data)} 条数据，累计 {len(all_data)} 条数据")
|
||||
|
||||
# 检查是否还有更多数据
|
||||
if len(all_data) >= total:
|
||||
print("已获取所有数据")
|
||||
break
|
||||
|
||||
index += 1
|
||||
|
||||
return all_data
|
||||
|
||||
PG_DSN = dict(
|
||||
database="dataops_db",
|
||||
user="dbuser_dba",
|
||||
password="EmBRxnmmjnE3",
|
||||
host="124.221.232.219",
|
||||
port="5432",
|
||||
)
|
||||
|
||||
|
||||
def save_json_to_pg(data: list, api_id: str) -> None:
|
||||
"""把列表落库：先软删历史，再插入新批次"""
|
||||
print("[save_to_pg] 写入 PG...")
|
||||
sql = """
|
||||
UPDATE data_api.api_data
|
||||
SET is_loaded = '1'
|
||||
WHERE api_id = %s;
|
||||
|
||||
INSERT INTO data_api.api_data
|
||||
(id, api_id, data, total_num, is_loaded, status,
|
||||
request_tm, execute_tm, remark)
|
||||
VALUES (%s, %s, %s, %s, '0', '0',
|
||||
current_timestamp(0), current_timestamp(0), '');
|
||||
"""
|
||||
|
||||
try:
|
||||
with psycopg2.connect(**PG_DSN) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
sql,
|
||||
(
|
||||
api_id,
|
||||
str(uuid.uuid4()),
|
||||
api_id,
|
||||
json.dumps(data, ensure_ascii=False),
|
||||
len(data),
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
cur.close()
|
||||
print("[save_to_pg] 写入完成")
|
||||
except psycopg2.Error as e:
|
||||
print(f"[save_to_pg] 数据库错误: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[save_to_pg] 未知错误: {e}")
|
||||
raise
|
||||
finally:
|
||||
if "conn" in locals():
|
||||
conn.close()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""主流程"""
|
||||
# print(get_token())
|
||||
print(f"开始请求新建联系人信息:{formatted2_previous_hour(0)}")
|
||||
token = get_token()
|
||||
# print(token)
|
||||
# 获取新建联系人ID
|
||||
print(f"开始请求新建联系人ID:{formatted2_previous_hour(0)}")
|
||||
objs = fetch_all_data(token)
|
||||
# 保存联系人ID
|
||||
apiId = "a7757b4a-7038-40ef-b11e-81a2c5e0"
|
||||
save_json_to_pg(objs, apiId)
|
||||
print(f"结束请求联系人详情:{formatted2_previous_hour(0)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -16,7 +16,8 @@ insert into data_api.api_contact_events (
|
|||
, meta_id
|
||||
, meta_name
|
||||
, meta_remark
|
||||
, properties
|
||||
, properties
|
||||
, contact_id
|
||||
,etl_tx_dt
|
||||
)
|
||||
select
|
||||
|
|
@ -30,7 +31,8 @@ select
|
|||
, case when trim(both from meta_id)='' then null else meta_id::text end meta_id
|
||||
, case when trim(both from meta_name)='' then null else meta_name::text end meta_name
|
||||
, case when trim(both from meta_remark)='' then null else meta_remark::text end meta_remark
|
||||
, case when trim(both from properties)='' then null else properties::text end properties
|
||||
, case when trim(both from properties)='' then null else properties::text end properties
|
||||
, case when trim(both from contact_id)='' then null else contact_id::text end contact_id
|
||||
,etl_tx_dt
|
||||
from (
|
||||
select
|
||||
|
|
@ -44,7 +46,8 @@ select
|
|||
, (json_array_elements(data::json)::json->>'metaId') meta_id
|
||||
, (json_array_elements(data::json)::json->>'metaName') meta_name
|
||||
, (json_array_elements(data::json)::json->>'metaRemark') meta_remark
|
||||
, (json_array_elements(data::json)::json->>'properties') properties
|
||||
, (json_array_elements(data::json)::json->>'properties') properties
|
||||
, (json_array_elements(data::json)::json->>'contactId') contact_id
|
||||
,CURRENT_TIMESTAMP(0) etl_tx_dt
|
||||
from (select * from data_api.api_data
|
||||
WHERE api_id='a7757b4a-7038-40ef-b11e-81a2c5e0' and is_loaded = '0' order by request_tm desc limit 1) p )p;
|
||||
|
|
|
|||
Loading…
Reference in New Issue