dtgitea
/
dataops
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add workflow partner1site,dev

This commit is contained in:
root 2025-09-30 16:15:42 +08:00
parent 729092f3fa
commit 40b4f81a74
5 changed files with 313 additions and 406 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

326
dev/workflow/TK_Cust/partner1site/获取POS数据/part_summary_pos_feign.py
View File

@ -1,176 +1,150 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Partner1site 全接口抓取脚本 Partner1site 全接口抓取脚本
分页结束条件：hasNext == False 分页结束条件：hasNext == False
""" """
import random import random
import hmac import hmac
import hashlib import hashlib
import base64 import base64
import requests import requests
import json import json
import uuid import uuid
from datetime import datetime, timezone, timedelta from datetime import datetime, timezone, timedelta
from typing import Dict, Any from typing import Dict, Any
import psycopg2 import psycopg2
# ======= 配置区 ======= # ======= 配置区 =======
ACCESS_KEY = "75c4ab4d-6a67-4aed-8b1d-5bb64fd36afc" ACCESS_KEY = "75c4ab4d-6a67-4aed-8b1d-5bb64fd36afc"
SECRET_KEY = "117347a7dd066a50a4d2973c5f3d5ba9101094c5" SECRET_KEY = "117347a7dd066a50a4d2973c5f3d5ba9101094c5"
# URL 用占位符（外部替换） # URL 用占位符（外部替换）
BASE_URLS = { BASE_URLS = {
# "visits": "http://onesite.tek.cn/api/summary/pos_datas", # 客户拜访数据 "visits": "http://onesite.tek.cn/api/summary/pos_datas", # 客户拜访数据
# "reports": "http://onesite.tek.cn/api/summary/pos_datas", # 报备数据 "reports": "http://onesite.tek.cn/api/summary/pos_datas", # 报备数据
"pos_datas": "http://onesite.tek.cn/api/summary/pos_datas", # POS数据 "pos_datas": "http://onesite.tek.cn/api/summary/pos_datas", # POS数据
# "customer_and_contact_datas": "http://onesite.tek.cn/api/summary/pos_datas" # 客户及联系人数据 "customer_and_contact_datas": "http://onesite.tek.cn/api/summary/pos_datas" # 客户及联系人数据
} }
PG_DSN = dict( PG_DSN = dict(
database="dataops_db", database="dataops_db",
user="dbuser_dba", user="dbuser_dba",
password="EmBRxnmmjnE3", password="EmBRxnmmjnE3",
host="124.221.232.219", host="124.221.232.219",
port="5432" port="5432"
) )
API_ID = "f5eeff00-8454-408d-843a-d83d90f2" # 外部传入 api_id，占位符 API_ID = "f5eeff00-8454-408d-843a-d83d90f2" # 外部传入 api_id，占位符
# ====================== # ======================
class Partner1SiteClient: class Partner1SiteClient:
"""Partner1site API 客户端""" """Partner1site API 客户端"""
def __init__(self, access_key: str, secret_key: str): def __init__(self, access_key: str, secret_key: str):
self.ak = access_key self.ak = access_key
self.sk = secret_key self.sk = secret_key
@staticmethod @staticmethod
def urlsafe_b64encode(data: bytes) -> str: def urlsafe_b64encode(data: bytes) -> str:
return base64.urlsafe_b64encode(data).decode() return base64.urlsafe_b64encode(data).decode()
def gen_token(self, expire_sec: int = 600) -> str: def gen_token(self, expire_sec: int = 600) -> str:
"""生成 API Token""" """生成 API Token"""
random_num = str(random.randint(100000, 999999)) random_num = str(random.randint(100000, 999999))
deadline = int(datetime.now(timezone.utc).timestamp()) + expire_sec deadline = int(datetime.now(timezone.utc).timestamp()) + expire_sec
parm_str = f"{random_num}:{deadline}" parm_str = f"{random_num}:{deadline}"
enc_parm = self.urlsafe_b64encode(parm_str.encode()) enc_parm = self.urlsafe_b64encode(parm_str.encode())
sign = hmac.new(self.sk.encode(), enc_parm.encode(), hashlib.sha1).digest() sign = hmac.new(self.sk.encode(), enc_parm.encode(), hashlib.sha1).digest()
enc_sign = self.urlsafe_b64encode(sign.hex().encode()) enc_sign = self.urlsafe_b64encode(sign.hex().encode())
return f"{self.ak}:{enc_sign}:{enc_parm}" return f"{self.ak}:{enc_sign}:{enc_parm}"
def fetch_all_pages(self, api_name: str, params: Dict[str, Any]): def fetch_all_pages(self, api_name: str, params: Dict[str, Any]):
"""通用分页请求（结束条件：hasNext == False）""" """通用分页请求（结束条件：hasNext == False）"""
if api_name not in BASE_URLS: # if api_name not in BASE_URLS:
raise ValueError(f"未知 API 数据来源: {api_name}") # raise ValueError(f"未知 API 数据来源: {api_name}")
base_url = BASE_URLS[api_name] base_url = 'http://onesite.tek.cn/api/summary/pos_datas'
all_data = [] all_data = []
page_num = 0 page_num = 0
page_size = 50 # 固定每页大小 page_size = 1000 # 固定每页大小
while True: while True:
token = self.gen_token() token = self.gen_token()
params_with_paging = dict(params) params_with_paging = dict(params)
params_with_paging.update({ params_with_paging.update({
"token": token, "token": token,
"size": page_size, "size": page_size,
"page": page_num "page": page_num
}) })
resp = requests.get(base_url, params=params_with_paging, timeout=30) resp = requests.get(base_url, params=params_with_paging, timeout=30)
resp.raise_for_status() resp.raise_for_status()
data_json = resp.json() data_json = resp.json()
if data_json.get("code") != 100 or not data_json.get("success", False): if data_json.get("code") != 100 or not data_json.get("success", False):
raise RuntimeError(f"{api_name} API 错误: {data_json.get('message')}") raise RuntimeError(f"获取POS数据 API 错误: {data_json.get('message')}")
content = data_json.get("data", {}).get("content", []) content = data_json.get("data", {}).get("content", [])
all_data.extend(content) all_data.extend(content)
total_elements = data_json.get("data", {}).get("totalElements") total_elements = data_json.get("data", {}).get("totalElements")
has_next = data_json.get("data", {}).get("hasNext", False) has_next = data_json.get("data", {}).get("hasNext", False)
print(f"[{api_name}] 页码 {page_num} -> 本页 {len(content)} 条，累计 {len(all_data)} 条 / 总数 {total_elements}") print(f"[获取POS数据] 页码 {page_num} -> 本页 {len(content)} 条，累计 {len(all_data)} 条 / 总数 {total_elements}")
if not has_next: if not has_next:
break break
page_num += 1 page_num += 1
return all_data return all_data
def save_json_to_pg(data: list, api_id: str) -> None: def save_json_to_pg(data: list, api_id: str) -> None:
"""写入 PostgreSQL：软删历史 + 插入新数据""" """写入 PostgreSQL：软删历史 + 插入新数据"""
print(f"[save_to_pg] API={api_id} 写入 PG，记录数={len(data)}") print(f"[save_to_pg] API={api_id} 写入 PG，记录数={len(data)}")
sql = """ sql = """
UPDATE data_api.api_data UPDATE data_api.api_data
SET is_loaded = '1' SET is_loaded = '1'
WHERE api_id = %s; WHERE api_id = %s;
INSERT INTO data_api.api_data INSERT INTO data_api.api_data
(id, api_id, data, total_num, is_loaded, status, (id, api_id, data, total_num, is_loaded, status,
request_tm, execute_tm, remark) request_tm, execute_tm, remark)
VALUES (%s, %s, %s, %s, '0', '0', VALUES (%s, %s, %s, %s, '0', '0',
current_timestamp(0), current_timestamp(0), ''); current_timestamp(0), current_timestamp(0), '');
""" """
try: try:
with psycopg2.connect(**PG_DSN) as conn: with psycopg2.connect(**PG_DSN) as conn:
with conn.cursor() as cur: with conn.cursor() as cur:
cur.execute(sql, cur.execute(sql,
(api_id, (api_id,
str(uuid.uuid4()), str(uuid.uuid4()),
api_id, api_id,
json.dumps(data, ensure_ascii=False), json.dumps(data, ensure_ascii=False),
len(data))) len(data)))
conn.commit() conn.commit()
print(f"[save_to_pg] API={api_id} 写入完成") print(f"[save_to_pg] API={api_id} 写入完成")
except Exception as e: except Exception as e:
raise RuntimeError(f"PG写入错误: {e}") raise RuntimeError(f"PG写入错误: {e}")
def get_previous_date(days: int = 0) -> str: def get_previous_date(days: int = 0) -> str:
return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
def main(): def main():
client = Partner1SiteClient(ACCESS_KEY, SECRET_KEY) client = Partner1SiteClient(ACCESS_KEY, SECRET_KEY)
# ✅ 客户拜访数据（执行） data = client.fetch_all_pages(
""" api_name="visits",
visits_data = client.fetch_all_pages( params={'startPosInsertDate':'2025-8-20',} # 拉全量，若有默认时间限制可改成 {"startInsertDate":"2000-01-01","endInsertDate":get_previous_date(0)}
api_name="visits", )
params={} # 拉全量，若有默认时间限制可改成 {"startInsertDate":"2000-01-01","endInsertDate":get_previous_date(0)} save_json_to_pg(data, API_ID)
)
save_json_to_pg(visits_data, API_ID)
"""
# ❌ 报备数据（暂时注释） if __name__ == "__main__":
""" main()
reports_data = client.fetch_all_pages(
api_name="reports",
params={}
)
save_json_to_pg(reports_data, API_ID)
"""
# ❌ POS 数据（暂时注释）
pos_data = client.fetch_all_pages(
api_name="pos_datas",
params={"startPosInsertDate":get_previous_date(3)}
)
save_json_to_pg(pos_data, API_ID)
# ❌ 客户及联系人数据（暂时注释）
"""
cust_contact_data = client.fetch_all_pages(
api_name="customer_and_contact_datas",
params={}
)
save_json_to_pg(cust_contact_data, API_ID)
"""
if __name__ == "__main__":
main()

320
dev/workflow/TK_Cust/partner1site/获取客户拜访数据/part_summary_visit_feign.py
View File

@ -1,178 +1,142 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Partner1site 全接口抓取脚本 Partner1site 全接口抓取脚本
分页结束条件：hasNext == False 分页结束条件：hasNext == False
""" """
import random import random
import hmac import hmac
import hashlib import hashlib
import base64 import base64
import requests import requests
import json import json
import uuid import uuid
from datetime import datetime, timezone, timedelta from datetime import datetime, timezone, timedelta
from typing import Dict, Any from typing import Dict, Any
import psycopg2 import psycopg2
# ======= 配置区 ======= # ======= 配置区 =======
ACCESS_KEY = "75c4ab4d-6a67-4aed-8b1d-5bb64fd36afc" ACCESS_KEY = "75c4ab4d-6a67-4aed-8b1d-5bb64fd36afc"
SECRET_KEY = "117347a7dd066a50a4d2973c5f3d5ba9101094c5" SECRET_KEY = "117347a7dd066a50a4d2973c5f3d5ba9101094c5"
# URL 用占位符（外部替换） PG_DSN = dict(
BASE_URLS = { database="dataops_db",
"visits": "http://onesite.tek.cn/api/summary/visits", # 客户拜访数据 user="dbuser_dba",
"reports": "http://onesite.tek.cn/api/summary/visits", # 报备数据 password="EmBRxnmmjnE3",
"pos_datas": "http://onesite.tek.cn/api/summary/visits", # POS数据 host="124.221.232.219",
"customer_and_contact_datas": "http://onesite.tek.cn/api/summary/visits" # 客户及联系人数据 port="5432"
} )
PG_DSN = dict( API_ID = "2460976d-00c1-47d9-84b2-33e66d68" # 外部传入 api_id，占位符
database="dataops_db", # ======================
user="dbuser_dba",
password="EmBRxnmmjnE3",
host="124.221.232.219", class Partner1SiteClient:
port="5432" """Partner1site API 客户端"""
)
def __init__(self, access_key: str, secret_key: str):
API_ID = "2460976d-00c1-47d9-84b2-33e66d68" # 外部传入 api_id，占位符 self.ak = access_key
# ====================== self.sk = secret_key
@staticmethod
class Partner1SiteClient: def urlsafe_b64encode(data: bytes) -> str:
"""Partner1site API 客户端""" return base64.urlsafe_b64encode(data).decode()
def __init__(self, access_key: str, secret_key: str): def gen_token(self, expire_sec: int = 600) -> str:
self.ak = access_key """生成 API Token"""
self.sk = secret_key random_num = str(random.randint(100000, 999999))
deadline = int(datetime.now(timezone.utc).timestamp()) + expire_sec
@staticmethod parm_str = f"{random_num}:{deadline}"
def urlsafe_b64encode(data: bytes) -> str: enc_parm = self.urlsafe_b64encode(parm_str.encode())
return base64.urlsafe_b64encode(data).decode() sign = hmac.new(self.sk.encode(), enc_parm.encode(), hashlib.sha1).digest()
enc_sign = self.urlsafe_b64encode(sign.hex().encode())
def gen_token(self, expire_sec: int = 600) -> str: return f"{self.ak}:{enc_sign}:{enc_parm}"
"""生成 API Token"""
random_num = str(random.randint(100000, 999999)) def fetch_all_pages(self, api_name: str, params: Dict[str, Any]):
deadline = int(datetime.now(timezone.utc).timestamp()) + expire_sec """通用分页请求（结束条件：hasNext == False）"""
parm_str = f"{random_num}:{deadline}" # if api_name not in BASE_URLS:
enc_parm = self.urlsafe_b64encode(parm_str.encode()) # raise ValueError(f"未知 API 数据来源: {api_name}")
sign = hmac.new(self.sk.encode(), enc_parm.encode(), hashlib.sha1).digest()
enc_sign = self.urlsafe_b64encode(sign.hex().encode()) base_url = 'http://onesite.tek.cn/api/summary/visits'
return f"{self.ak}:{enc_sign}:{enc_parm}" all_data = []
page_num = 0
def fetch_all_pages(self, api_name: str, params: Dict[str, Any]): page_size = 1000 # 固定每页大小
"""通用分页请求（结束条件：hasNext == False）"""
if api_name not in BASE_URLS: while True:
raise ValueError(f"未知 API 数据来源: {api_name}") token = self.gen_token()
params_with_paging = dict(params)
base_url = BASE_URLS[api_name] params_with_paging.update({
all_data = [] "token": token,
page_num = 0 "size": page_size,
page_size = 50 # 固定每页大小 "page": page_num
})
while True:
token = self.gen_token() resp = requests.get(base_url, params=params_with_paging, timeout=30)
params_with_paging = dict(params) resp.raise_for_status()
params_with_paging.update({ data_json = resp.json()
"token": token,
"size": page_size, if data_json.get("code") != 100 or not data_json.get("success", False):
"page": page_num raise RuntimeError(f"获取客户拜访数据 API 错误: {data_json.get('message')}")
})
content = data_json.get("data", {}).get("content", [])
resp = requests.get(base_url, params=params_with_paging, timeout=30) all_data.extend(content)
resp.raise_for_status() total_elements = data_json.get("data", {}).get("totalElements")
data_json = resp.json() has_next = data_json.get("data", {}).get("hasNext", False)
if data_json.get("code") != 100 or not data_json.get("success", False): print(f"[获取客户拜访数据] 页码 {page_num} -> 本页 {len(content)} 条，累计 {len(all_data)} 条 / 总数 {total_elements}")
raise RuntimeError(f"{api_name} API 错误: {data_json.get('message')}")
if not has_next:
content = data_json.get("data", {}).get("content", []) break
all_data.extend(content)
total_elements = data_json.get("data", {}).get("totalElements") page_num += 1
has_next = data_json.get("data", {}).get("hasNext", False)
return all_data
print(f"[{api_name}] 页码 {page_num} -> 本页 {len(content)} 条，累计 {len(all_data)} 条 / 总数 {total_elements}")
if not has_next: def save_json_to_pg(data: list, api_id: str) -> None:
break """写入 PostgreSQL：软删历史 + 插入新数据"""
print(f"[save_to_pg] API={api_id} 写入 PG，记录数={len(data)}")
page_num += 1 sql = """
UPDATE data_api.api_data
return all_data SET is_loaded = '1'
WHERE api_id = %s;
def save_json_to_pg(data: list, api_id: str) -> None: INSERT INTO data_api.api_data
"""写入 PostgreSQL：软删历史 + 插入新数据""" (id, api_id, data, total_num, is_loaded, status,
print(f"[save_to_pg] API={api_id} 写入 PG，记录数={len(data)}") request_tm, execute_tm, remark)
sql = """ VALUES (%s, %s, %s, %s, '0', '0',
UPDATE data_api.api_data current_timestamp(0), current_timestamp(0), '');
SET is_loaded = '1' """
WHERE api_id = %s; try:
with psycopg2.connect(**PG_DSN) as conn:
INSERT INTO data_api.api_data with conn.cursor() as cur:
(id, api_id, data, total_num, is_loaded, status, cur.execute(sql,
request_tm, execute_tm, remark) (api_id,
VALUES (%s, %s, %s, %s, '0', '0', str(uuid.uuid4()),
current_timestamp(0), current_timestamp(0), ''); api_id,
""" json.dumps(data, ensure_ascii=False),
try: len(data)))
with psycopg2.connect(**PG_DSN) as conn: conn.commit()
with conn.cursor() as cur: print(f"[save_to_pg] API={api_id} 写入完成")
cur.execute(sql, except Exception as e:
(api_id, raise RuntimeError(f"PG写入错误: {e}")
str(uuid.uuid4()),
api_id,
json.dumps(data, ensure_ascii=False), def get_previous_date(days: int = 0) -> str:
len(data))) return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
conn.commit()
print(f"[save_to_pg] API={api_id} 写入完成")
except Exception as e: def main():
raise RuntimeError(f"PG写入错误: {e}") client = Partner1SiteClient(ACCESS_KEY, SECRET_KEY)
data = client.fetch_all_pages(
def get_previous_date(days: int = 0) -> str: api_name="visits",
return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") params={'startInsertDate':'2025-8-20',} # 拉全量，若有默认时间限制可改成 {"startInsertDate":"2000-01-01","endInsertDate":get_previous_date(0)}
)
save_json_to_pg(data, API_ID)
def main():
client = Partner1SiteClient(ACCESS_KEY, SECRET_KEY)
# ✅ 客户拜访数据（执行） if __name__ == "__main__":
visits_data = client.fetch_all_pages( main()
api_name="visits",
# params={} # 拉全量，若有默认时间限制可改成 {"startInsertDate":"2000-01-01","endInsertDate":get_previous_date(0)}
params={"startInsertDate":"2025-08-20"}
)
save_json_to_pg(visits_data, API_ID)
# ❌ 报备数据（暂时注释）
"""
reports_data = client.fetch_all_pages(
api_name="reports",
params={}
)
save_json_to_pg(reports_data, API_ID)
"""
# ❌ POS 数据（暂时注释）
"""
pos_data = client.fetch_all_pages(
api_name="pos_datas",
params={}
)
save_json_to_pg(pos_data, API_ID)
"""
# ❌ 客户及联系人数据（暂时注释）
"""
cust_contact_data = client.fetch_all_pages(
api_name="customer_and_contact_datas",
params={}
)
save_json_to_pg(cust_contact_data, API_ID)
"""
if __name__ == "__main__":
main()

9
dev/workflow/TK_Cust/partner1site/获取客户拜访数据/part_summary_visit_load.sql
View File

@ -32,7 +32,8 @@ insert into data_api.partner_summary_visit (
, sub_industry , sub_industry
, tsm_names_by_alias , tsm_names_by_alias
, visit_remark , visit_remark
, visitor , visitor
, id
,etl_tx_dt ,etl_tx_dt
) )
select select
@ -62,7 +63,8 @@ select
, case when trim(both from sub_industry)='' then null else sub_industry::text end sub_industry , case when trim(both from sub_industry)='' then null else sub_industry::text end sub_industry
, case when trim(both from tsm_names_by_alias)='' then null else tsm_names_by_alias::text end tsm_names_by_alias , case when trim(both from tsm_names_by_alias)='' then null else tsm_names_by_alias::text end tsm_names_by_alias
, case when trim(both from visit_remark)='' then null else visit_remark::text end visit_remark , case when trim(both from visit_remark)='' then null else visit_remark::text end visit_remark
, case when trim(both from visitor)='' then null else visitor::text end visitor , case when trim(both from visitor)='' then null else visitor::text end visitor
, case when trim(both from id)='' then null else id::text end id
,etl_tx_dt ,etl_tx_dt
from ( from (
select select
@ -92,7 +94,8 @@ select
, (json_array_elements(data::json)::json->>'subIndustry') sub_industry , (json_array_elements(data::json)::json->>'subIndustry') sub_industry
, (json_array_elements(data::json)::json->>'tsmNamesByAlias') tsm_names_by_alias , (json_array_elements(data::json)::json->>'tsmNamesByAlias') tsm_names_by_alias
, (json_array_elements(data::json)::json->>'visitRemark') visit_remark , (json_array_elements(data::json)::json->>'visitRemark') visit_remark
, (json_array_elements(data::json)::json->>'visitor') visitor , (json_array_elements(data::json)::json->>'visitor') visitor
, (json_array_elements(data::json)::json->>'id') id
,CURRENT_TIMESTAMP(0) etl_tx_dt ,CURRENT_TIMESTAMP(0) etl_tx_dt
from (select * from data_api.api_data from (select * from data_api.api_data
WHERE api_id='2460976d-00c1-47d9-84b2-33e66d68' and is_loaded = '0' order by request_tm desc limit 1) p )p; WHERE api_id='2460976d-00c1-47d9-84b2-33e66d68' and is_loaded = '0' order by request_tm desc limit 1) p )p;

52
dev/workflow/TK_Cust/partner1site/获取报备数据/part_summary_report_feign.py
View File

@ -20,14 +20,6 @@ import psycopg2
ACCESS_KEY = "75c4ab4d-6a67-4aed-8b1d-5bb64fd36afc" ACCESS_KEY = "75c4ab4d-6a67-4aed-8b1d-5bb64fd36afc"
SECRET_KEY = "117347a7dd066a50a4d2973c5f3d5ba9101094c5" SECRET_KEY = "117347a7dd066a50a4d2973c5f3d5ba9101094c5"
# URL 用占位符（外部替换）
BASE_URLS = {
# "visits": "http://onesite.tek.cn/api/summary/reports", # 客户拜访数据
"reports": "http://onesite.tek.cn/api/summary/reports" # 报备数据
# "pos_datas": "http://onesite.tek.cn/api/summary/reports", # POS数据
# "customer_and_contact_datas": "http://onesite.tek.cn/api/summary/reports" # 客户及联系人数据
}
PG_DSN = dict( PG_DSN = dict(
database="dataops_db", database="dataops_db",
user="dbuser_dba", user="dbuser_dba",
@ -63,13 +55,13 @@ class Partner1SiteClient:
def fetch_all_pages(self, api_name: str, params: Dict[str, Any]): def fetch_all_pages(self, api_name: str, params: Dict[str, Any]):
"""通用分页请求（结束条件：hasNext == False）""" """通用分页请求（结束条件：hasNext == False）"""
if api_name not in BASE_URLS: # if api_name not in BASE_URLS:
raise ValueError(f"未知 API 数据来源: {api_name}") # raise ValueError(f"未知 API 数据来源: {api_name}")
base_url = BASE_URLS[api_name] base_url = 'http://onesite.tek.cn/api/summary/reports'
all_data = [] all_data = []
page_num = 0 page_num = 0
page_size = 50 # 固定每页大小 page_size = 1000 # 固定每页大小
while True: while True:
token = self.gen_token() token = self.gen_token()
@ -85,14 +77,14 @@ class Partner1SiteClient:
data_json = resp.json() data_json = resp.json()
if data_json.get("code") != 100 or not data_json.get("success", False): if data_json.get("code") != 100 or not data_json.get("success", False):
raise RuntimeError(f"{api_name} API 错误: {data_json.get('message')}") raise RuntimeError(f"获取报备数据 API 错误: {data_json.get('message')}")
content = data_json.get("data", {}).get("content", []) content = data_json.get("data", {}).get("content", [])
all_data.extend(content) all_data.extend(content)
total_elements = data_json.get("data", {}).get("totalElements") total_elements = data_json.get("data", {}).get("totalElements")
has_next = data_json.get("data", {}).get("hasNext", False) has_next = data_json.get("data", {}).get("hasNext", False)
print(f"[{api_name}] 页码 {page_num} -> 本页 {len(content)} 条，累计 {len(all_data)} 条 / 总数 {total_elements}") print(f"[获取报备数据] 页码 {page_num} -> 本页 {len(content)} 条，累计 {len(all_data)} 条 / 总数 {total_elements}")
if not has_next: if not has_next:
break break
@ -138,40 +130,14 @@ def get_previous_date(days: int = 0) -> str:
def main(): def main():
client = Partner1SiteClient(ACCESS_KEY, SECRET_KEY) client = Partner1SiteClient(ACCESS_KEY, SECRET_KEY)
# ✅ 报备数据（执行） data = client.fetch_all_pages(
reports_data = client.fetch_all_pages(
api_name="reports", api_name="reports",
params={"startInsertDate":get_previous_date(3)} params={"startInsertDate":"2000-01-01"} # 拉全量，若有默认时间限制可改成 {"startInsertDate":"2000-01-01","endInsertDate":get_previous_date(0)}
) )
save_json_to_pg(reports_data, API_ID)
# ❌ 客户拜访数据（暂时注释）
"""
visits_data = client.fetch_all_pages(
api_name="visits",
params={} # 拉全量，若有默认时间限制可改成 {"startInsertDate":"2000-01-01","endInsertDate":get_previous_date(0)}
)
save_json_to_pg(visits_data, API_ID)
"""
# ❌ POS 数据（暂时注释） save_json_to_pg(data, API_ID)
"""
pos_data = client.fetch_all_pages(
api_name="pos_datas",
params={}
)
save_json_to_pg(pos_data, API_ID)
"""
# ❌ 客户及联系人数据（暂时注释）
"""
cust_contact_data = client.fetch_all_pages(
api_name="customer_and_contact_datas",
params={}
)
save_json_to_pg(cust_contact_data, API_ID)
"""
if __name__ == "__main__": if __name__ == "__main__":

12
dev/workflow/TK_Cust/partner1site/获取报备数据/part_summary_report_load.sql
View File

@ -6,8 +6,7 @@
DELETE FROM data_api.partner_summary_report; DELETE FROM data_api.partner_summary_report;
insert into data_api.partner_summary_report ( insert into data_api.partner_summary_report (
accepted address
, address
, advantage , advantage
, alias , alias
, apply_admin_name , apply_admin_name
@ -48,6 +47,7 @@ insert into data_api.partner_summary_report (
, product_type , product_type
, progress , progress
, province , province
, record_id
, region_name , region_name
, remark , remark
, report_num , report_num
@ -66,8 +66,7 @@ insert into data_api.partner_summary_report (
,etl_tx_dt ,etl_tx_dt
) )
select select
case when trim(both from accepted)='' then null else accepted::text end accepted case when trim(both from address)='' then null else address::text end address
, case when trim(both from address)='' then null else address::text end address
, case when trim(both from advantage)='' then null else advantage::text end advantage , case when trim(both from advantage)='' then null else advantage::text end advantage
, case when trim(both from alias)='' then null else alias::text end alias , case when trim(both from alias)='' then null else alias::text end alias
, case when trim(both from apply_admin_name)='' then null else apply_admin_name::text end apply_admin_name , case when trim(both from apply_admin_name)='' then null else apply_admin_name::text end apply_admin_name
@ -108,6 +107,7 @@ select
, case when trim(both from product_type)='' then null else product_type::text end product_type , case when trim(both from product_type)='' then null else product_type::text end product_type
, case when trim(both from progress)='' then null else progress::text end progress , case when trim(both from progress)='' then null else progress::text end progress
, case when trim(both from province)='' then null else province::text end province , case when trim(both from province)='' then null else province::text end province
, case when trim(both from record_id)='' then null else record_id::text end record_id
, case when trim(both from region_name)='' then null else region_name::text end region_name , case when trim(both from region_name)='' then null else region_name::text end region_name
, case when trim(both from remark)='' then null else remark::text end remark , case when trim(both from remark)='' then null else remark::text end remark
, case when trim(both from report_num)='' then null else report_num::text end report_num , case when trim(both from report_num)='' then null else report_num::text end report_num
@ -126,8 +126,7 @@ select
,etl_tx_dt ,etl_tx_dt
from ( from (
select select
(json_array_elements(data::json)::json->>'accepted') accepted (json_array_elements(data::json)::json->>'address') address
, (json_array_elements(data::json)::json->>'address') address
, (json_array_elements(data::json)::json->>'advantage') advantage , (json_array_elements(data::json)::json->>'advantage') advantage
, (json_array_elements(data::json)::json->>'alias') alias , (json_array_elements(data::json)::json->>'alias') alias
, (json_array_elements(data::json)::json->>'applyAdminName') apply_admin_name , (json_array_elements(data::json)::json->>'applyAdminName') apply_admin_name
@ -168,6 +167,7 @@ select
, (json_array_elements(data::json)::json->>'productType') product_type , (json_array_elements(data::json)::json->>'productType') product_type
, (json_array_elements(data::json)::json->>'progress') progress , (json_array_elements(data::json)::json->>'progress') progress
, (json_array_elements(data::json)::json->>'province') province , (json_array_elements(data::json)::json->>'province') province
, (json_array_elements(data::json)::json->>'recordId') record_id
, (json_array_elements(data::json)::json->>'regionName') region_name , (json_array_elements(data::json)::json->>'regionName') region_name
, (json_array_elements(data::json)::json->>'remark') remark , (json_array_elements(data::json)::json->>'remark') remark
, (json_array_elements(data::json)::json->>'reportNum') report_num , (json_array_elements(data::json)::json->>'reportNum') report_num