Skip to content

S
spider
Commit fb663d1b by Peng
Options

no message

parent a8e938f4
Showing with 74 additions and 1513 deletions
py_spider/amazon_every_day_spider/secure_db_client.py deleted 100644 → 0
#
#
#
# '''旧版'''
#
# import json
#
# import pandas as pd
# import numpy as np
# import orjson, requests, time
# from typing import List
#
# # -------- 映射字典 --------
# site_name_db_dict = {
# "us": "selection",
# "uk": "selection_uk",
# "de": "selection_de",
# "es": "selection_es",
# "fr": "selection_fr",
# "it": "selection_it",
# }
#
# db_type_alias_map = {
# "mysql": "mysql", # 阿里云mysql
# "postgresql_14": "postgresql_14", # pg14爬虫库-内网
# "postgresql_14_outer": "postgresql_14", # pg14爬虫库-外网
# "postgresql_15": "postgresql_15", # pg15正式库-内网
# "postgresql_15_outer": "postgresql_15_outer", # pg15正式库-外网
# "postgresql_cluster": "postgresql_cluster", # pg集群-内网
# "postgresql_cluster_outer": "postgresql_cluster_outer", # pg集群-外网
# "doris": "doris", # doris集群-内网
# }
#
# DEFAULT_SERVERS = [
# # "http://192.168.200.210:7777", # 内网
# # "http://192.168.10.217:7777", # 内网-h7
# # "http://113.100.143.162:7777", # 外网
# # "http://113.100.143.162:7779", # 外网
# # "http://61.145.136.61:7777", # 外网
# # "http://61.145.136.61:7779", # 外网
# "http://192.168.200.210:7780"
#
#
# ]
# # ---------------------------
#
# def df_to_json_records(df: pd.DataFrame) -> list:
# """保证 DataFrame 可安全序列化为 JSON records（处理 NaN / ±Inf）"""
# df_clean = df.copy()
#
# # 1️⃣ 替换 ±Inf -> NaN
# num_cols = df_clean.select_dtypes(include=[np.number]).columns
# if len(num_cols):
# df_clean[num_cols] = df_clean[num_cols].replace([np.inf, -np.inf], np.nan)
#
# # 2️⃣ 替换 NaN -> None（注意：有时 astype(object) 不彻底，需用 applymap）
# df_clean = df_clean.applymap(lambda x: None if pd.isna(x) else x)
#
# # 3️⃣ 转为 dict records
# return df_clean.to_dict("records")
#
#
# def clean_json_field_for_orjson(v):
# """清洗单个 JSON 字段的值，使其符合 orjson 要求并避免空字典入库"""
# if v is None or pd.isna(v):
# return None
#
# # 1️⃣ 如果是空字典对象，返回 None
# if isinstance(v, dict) and not v:
# return None
#
# # 2️⃣ 如果是空字符串或仅为 "{}"，返回 None
# if isinstance(v, str):
# stripped = v.strip()
# if not stripped or stripped == "{}":
# return None
# try:
# parsed = json.loads(stripped)
# if isinstance(parsed, dict) and not parsed:
# return None
# return json.dumps(parsed, ensure_ascii=False)
# except Exception:
# return v # 非 JSON 字符串则原样保留
#
# return v
#
#
# def fully_clean_for_orjson(df: pd.DataFrame) -> pd.DataFrame:
# # """全面清洗 DataFrame 以符合 orjson 要求"""
# # df = df.replace([np.inf, -np.inf], np.nan)
# # df = df.applymap(lambda x: None if pd.isna(x) else x)
# #
# # # 找出所有可能为 JSON 字符串的字段
# # json_like_cols = [col for col in df.columns if col.endswith('_json')]
# #
# # # 针对每个 JSON-like 字段，应用清洗函数
# # for col in json_like_cols:
# # df[col] = df[col].apply(clean_json_field_for_orjson)
# #
# # return df
# """全面清洗 DataFrame 以符合 orjson 要求"""
# df = df.replace([np.inf, -np.inf], np.nan)
#
# # NaN → None （比 applymap 高效且不出错）
# df = df.where(pd.notna(df), None)
#
# # 找出所有可能为 JSON 字符串的字段
# json_like_cols = [col for col in df.columns if col.endswith('_json')]
#
# # 针对每个 JSON-like 字段，应用清洗函数
# for col in json_like_cols:
# df[col] = df[col].apply(clean_json_field_for_orjson)
#
# return df
#
#
# class RemoteTransaction:
#
# def __init__(self, db: str, database: str,
# session: requests.Session, urls: List[str]):
# self.db = db
# self.database = database
# self.session = session
# self.urls = urls
# self.sql_queue = []
#
# # def execute(self, sql: str):
# # self.sql_queue.append(sql)
# def execute(self, sql: str, params=None):
# """
# params 可取：
# • None → 纯文本 SQL
# • dict → 单条参数化 e.g. {"id":1,"name":"a"}
# • list/tuple → 批量 executemany
# - list[dict] ↔ INSERT .. VALUES (:id,:name)
# - list[tuple] ↔ INSERT .. VALUES (%s,%s)
# """
# self.sql_queue.append({"sql": sql, "params": params})
#
# def __enter__(self): return self
#
# def __exit__(self, exc_type, exc, tb):
# for url in self.urls:
# try:
# self.session.post(
# url + "/transaction",
# json={"db": self.db,
# "sql_list": self.sql_queue,
# "site_name": self.database}, # site_name not needed on server, kept for clarity
# timeout=3000,
# ).raise_for_status()
# return
# except Exception as e:
# print(f"[WARN] 事务失败 {url}: {e}")
# raise RuntimeError("All servers failed for transaction")
#
#
# class RemoteEngine:
# def __init__(self, db: str, database: str,
# server_urls: List[str], retries: int = 2):
# self.db = db
# self.database = database
# self.urls = [u.rstrip("/") for u in server_urls]
# self.session = requests.Session()
# self.retries = retries
#
# def _request(self, endpoint: str, payload):
# for url in self.urls:
# for _ in range(self.retries):
# try:
# json_bytes = orjson.dumps(payload)
# r = self.session.post(f"{url}/{endpoint}",
# data=json_bytes,
# headers={"Content-Type": "application/json"},
# timeout=3000)
#
# # r = self.session.post(f"{url}/{endpoint}",
# # json=payload, timeout=10)
# r.raise_for_status()
# return r.json()
# except Exception as e:
# print(f"[WARN] {endpoint} fail @ {url}: {e}")
# time.sleep(1)
# raise RuntimeError(f"All servers failed for {endpoint}")
# # def _request(self, endpoint: str, payload):
# # # 用 orjson，“allow_nan” 会把 NaN/Inf 写成 null
# # # json_bytes = orjson.dumps(payload,
# # # option=orjson.OPT_NON_STR_KEYS | orjson.OPT_NAIVE_UTC | orjson.OPT_OMIT_MICROSECOND | orjson.OPT_ALLOW_INF_AND_NAN)
# # json_bytes = orjson.dumps(
# # payload,
# # option=orjson.OPT_NON_STR_KEYS | orjson.OPT_NAIVE_UTC | orjson.OPT_ALLOW_INF_AND_NAN
# # )
# #
# # headers = {"Content-Type": "application/json"}
# #
# # for url in self.urls:
# # for _ in range(self.retries):
# # try:
# # r = self.session.post(f"{url}/{endpoint}",
# # data=json_bytes, headers=headers,
# # timeout=15)
# # r.raise_for_status()
# # return r.json()
# # except Exception as e:
# # print(f"[WARN] {endpoint} fail @ {url}: {e}")
# # time.sleep(1)
# # raise RuntimeError(f"All servers failed for {endpoint}")
#
# # ---------- 公共 API ----------
# def read_sql(self, sql: str) -> pd.DataFrame:
# data = self._request("query",
# {"db": self.db,
# "sql": sql,
# "site_name": self.database})
# return pd.DataFrame(data["result"])
#
# def to_sql(self, df: pd.DataFrame, table: str, if_exists="append"):
#
# return self._request("insert",
# {"db": self.db,
# "table": table,
# "if_exists": if_exists,
# "data": fully_clean_for_orjson(df=df).to_dict("records"),
# # "data": df_to_json_records(df), # ← 清洗后的 records
# "site_name": self.database})
#
# def read_then_update(
# self,
# select_sql: str,
# update_table: str,
# set_values: dict,
# where_keys: List[str],
# error_if_empty: bool = False,
# ):
# """
# 动态生成 UPDATE：把 select_sql 读到的行，按 where_keys 精准更新 set_values
# 返回 (DataFrame, rows_updated)
# """
# payload = {
# "db": self.db,
# "site_name": self.database,
# "select_sql": select_sql,
# "update_table": update_table,
# "set_values": set_values,
# "where_keys": where_keys,
# "error_if_empty": error_if_empty,
# }
# resp = self._request("read_then_update", payload)
# df = pd.DataFrame(resp["read_result"])
# rows_updated = resp.get("rows_updated", 0)
# return df
#
# def begin(self):
# return RemoteTransaction(self.db, self.database,
# self.session, self.urls)
# # ---------------------------------
#
#
# def get_remote_engine(site_name: str, db_type: str,
# servers: List[str] = None) -> RemoteEngine:
# if site_name not in site_name_db_dict:
# raise ValueError(f"Unknown site_name: {site_name}")
# if db_type not in db_type_alias_map:
# raise ValueError(f"Unknown db_type: {db_type}")
# return RemoteEngine(
# db=db_type_alias_map[db_type],
# database=site_name,
# server_urls=servers or DEFAULT_SERVERS,
# )
import json
import sys,os
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
import pandas as pd
import numpy as np
import orjson, requests, time
from typing import List
# -------- 映射字典 --------
site_name_db_dict = {
"us": "selection",
"uk": "selection_uk",
"de": "selection_de",
"es": "selection_es",
"fr": "selection_fr",
"it": "selection_it",
}
db_type_alias_map = {
"mysql": "mysql", # 阿里云mysql
"postgresql_14": "postgresql_14", # pg14爬虫库-内网
"postgresql_14_outer": "postgresql_14_outer", # pg14爬虫库-外网
"postgresql_15": "postgresql_15", # pg15正式库-内网
"postgresql_15_outer": "postgresql_15_outer", # pg15正式库-外网
"postgresql_cluster": "postgresql_cluster", # pg集群-内网
"postgresql_cluster_outer": "postgresql_cluster_outer", # pg集群-外网
"doris": "doris", # doris集群-内网
"doris_adv": "doris_adv",
}
DEFAULT_SERVERS = [
# "http://192.168.200.210:7777", # 内网
# "http://192.168.10.217:7777", # 内网-h7
# "http://61.145.136.61:7777", # 外网
# "http://61.145.136.61:7779", # 外网
"http://61.145.136.61:7780"
]
# ---------------------------
def df_to_json_records(df: pd.DataFrame) -> list:
"""保证 DataFrame 可安全序列化为 JSON records（处理 NaN / ±Inf）"""
df_clean = df.copy()
# 1️⃣ 替换 ±Inf -> NaN
num_cols = df_clean.select_dtypes(include=[np.number]).columns
if len(num_cols):
df_clean[num_cols] = df_clean[num_cols].replace([np.inf, -np.inf], np.nan)
# 2️⃣ 替换 NaN -> None（兼容 pandas 1.x 和 2.x）
df_clean = df_clean.where(pd.notna(df_clean), None)
# 3️⃣ 转为 dict records
return df_clean.to_dict("records")
def clean_json_field_for_orjson(v):
"""清洗单个 JSON 字段的值，使其符合 orjson 要求并避免空字典入库"""
if v is None or pd.isna(v):
return None
# 1️⃣ 如果是空字典对象，返回 None
if isinstance(v, dict) and not v:
return None
# 2️⃣ 如果是空字符串或仅为 "{}"，返回 None
if isinstance(v, str):
stripped = v.strip()
if not stripped or stripped == "{}":
return None
try:
parsed = json.loads(stripped)
if isinstance(parsed, dict) and not parsed:
return None
return json.dumps(parsed, ensure_ascii=False)
except Exception:
return v # 非 JSON 字符串则原样保留
return v
def fully_clean_for_orjson(df: pd.DataFrame) -> pd.DataFrame:
"""全面清洗 DataFrame 以符合 orjson 要求"""
df = df.replace([np.inf, -np.inf], np.nan)
df = df.where(pd.notna(df), None)
# 找出所有可能为 JSON 字符串的字段
json_like_cols = [col for col in df.columns if col.endswith('_json')]
# 针对每个 JSON-like 字段，应用清洗函数
for col in json_like_cols:
df[col] = df[col].apply(clean_json_field_for_orjson)
return df
class RemoteTransaction:
def __init__(self, db: str, database: str,
session: requests.Session, urls: List[str]):
self.db = db
self.database = database
self.session = session
self.urls = urls
self.sql_queue = []
# def execute(self, sql: str):
# self.sql_queue.append(sql)
def execute(self, sql: str, params=None):
"""
params 可取：
• None → 纯文本 SQL
• dict → 单条参数化 e.g. {"id":1,"name":"a"}
• list/tuple → 批量 executemany
- list[dict] ↔ INSERT .. VALUES (:id,:name)
- list[tuple] ↔ INSERT .. VALUES (%s,%s)
"""
self.sql_queue.append({"sql": sql, "params": params})
def __enter__(self): return self
def __exit__(self, exc_type, exc, tb):
for url in self.urls:
try:
self.session.post(
url + "/transaction",
json={"db_type": self.db,
"sql_list": self.sql_queue,
"database": self.database}, # site_name not needed on server, kept for clarity
timeout=3000,
).raise_for_status()
return
except Exception as e:
print(f"[WARN] 事务失败 {url}: {e}")
raise RuntimeError("All servers failed for transaction")
class RemoteEngine:
def __init__(self, db: str, database: str,
server_urls: List[str], retries: int = 20):
"""
:param db: db_type--数据库类型
:param database: 数据库名称
:param server_urls:
:param retries:
"""
self.db = db
self.database = database
self.urls = [u.rstrip("/") for u in server_urls]
self.session = requests.Session()
self.retries = retries
def _request(self, endpoint: str, payload):
for url in self.urls:
for _ in range(self.retries):
try:
json_bytes = orjson.dumps(payload)
r = self.session.post(f"{url}/{endpoint}",
data=json_bytes,
headers={"Content-Type": "application/json"},
timeout=3600)
# r = self.session.post(f"{url}/{endpoint}",
# json=payload, timeout=10)
r.raise_for_status()
return r.json()
except Exception as e:
print(f"[WARN] {endpoint} fail @ {url}: {e}")
time.sleep(1)
raise RuntimeError(f"All servers failed for {endpoint}")
# ---------- 公共 API ----------
def read_sql(self, sql: str) -> pd.DataFrame:
data = self._request("query",
{"db_type": self.db,
"sql": sql,
"database": self.database})
return pd.DataFrame(data["result"])
def to_sql(self, df: pd.DataFrame, table: str, if_exists="append"):
return self._request("insert",
{"db_type": self.db,
"table": table,
"if_exists": if_exists,
"data": fully_clean_for_orjson(df=df).to_dict("records"),
# "data": df_to_json_records(df), # ← 清洗后的 records
"database": self.database})
def read_then_update(
self,
select_sql: str,
update_table: str,
set_values: dict,
where_keys: List[str],
error_if_empty: bool = False,
):
"""
动态生成 UPDATE：把 select_sql 读到的行，按 where_keys 精准更新 set_values
返回 (DataFrame, rows_updated)
"""
payload = {
"db_type": self.db,
"database": self.database,
"select_sql": select_sql,
"update_table": update_table,
"set_values": set_values,
"where_keys": where_keys,
"error_if_empty": error_if_empty,
}
resp = self._request("read_then_update", payload)
df = pd.DataFrame(resp["read_result"])
rows_updated = resp.get("rows_updated", 0)
return df
def begin(self):
return RemoteTransaction(self.db, self.database,
self.session, self.urls)
# ---------------------------------
# db -- 数据库类型
# database -- 站点
def get_remote_engine(site_name: str, db_type: str, database: str = None,
servers: List[str] = None) -> RemoteEngine:
"""
:param site_name: 站点
:param db_type: 数据库类型
:param database: 数据库名称-默认为None， 否则通过站点来匹配
:param servers: 服务器url地址
:return: 数据库连接对象
"""
if not database:
database = site_name_db_dict[site_name]
if site_name not in site_name_db_dict:
raise ValueError(f"Unknown site_name: {site_name}")
if db_type not in db_type_alias_map:
raise ValueError(f"Unknown db_type: {db_type}")
print(f"db_type: {db_type_alias_map[db_type]}, database: {database}")
return RemoteEngine(
db=db_type_alias_map[db_type],
database=database,
server_urls=servers or DEFAULT_SERVERS,
)
def stream_load_df(df: pd.DataFrame, database: str='advertising_manager', table: str='', server_url: str = None):
server = (server_url or DEFAULT_SERVERS[0]).rstrip("/")
payload = {
"database": database,
"table": table,
"data": fully_clean_for_orjson(df).to_dict("records"),
}
json_bytes = orjson.dumps(payload)
r = requests.post(
f"{server}/stream_load_df",
data=json_bytes,
headers={"Content-Type": "application/json"},
timeout=3000,
)
r.raise_for_status()
return r.json()
if __name__ == '__main__':
engine_mysql = get_remote_engine(
site_name='us', # -> database "selection"
db_type="doris_adv", # -> 服务端 alias "mysql"
database="advertising_manager", # -> 服务端 alias "mysql"
)
# e = engine_mysql.read_sql('SELECT max(created_time),MIN(created_time) FROM us_sp_search_term_rank WHERE time_batch="2026-03-09-15"')
# e = engine_mysql.read_sql('SELECT * FROM us_sp_search_term_rank WHERE time_batch="2026-03-09-12" and page_row=68')
# e = engine_mysql.read_sql('SELECT * FROM us_sp_search_term_rank WHERE id=50564433')
e = engine_mysql.read_sql('SELECT count(id),time_batch FROM us_sp_search_term_rank GROUP BY time_batch ORDER BY time_batch ASC')
# e = engine_mysql.read_sql('SELECT count(id),time_batch FROM us_sp_search_term_rank GROUP BY time_batch ORDER BY time_batch ASC')
# e = engine_mysql.read_sql('SELECT count(id) FROM sp_keyword_position_keyword')
# e = engine_mysql.read_sql('select * from us_sp_search_term_syn limit 100')
# dog bandanas spring
# 2026-04-03-11
# e = engine_mysql.read_sql('select * from us_sp_search_term_syn where search_term = "dog bandanas spring"')
# e = engine_mysql.read_sql('select * from us_sp_search_term_rank where search_term="dog bandanas spring" and time_batch="2026-04-03-11" and cate_type="sp" and asin="B0GCN7CTCM"')
# e = engine_mysql.read_sql('select * from us_sp_search_term_rank where time_batch="2026-03-17-16" and cate_type="sp" limit 100')
# print(e)
print(e.columns)
print(e.values.tolist())
py_spider/amazon_spider/secure_db_client.py deleted 100644 → 0
#
#
#
# '''旧版'''
#
# import json
#
# import pandas as pd
# import numpy as np
# import orjson, requests, time
# from typing import List
#
# # -------- 映射字典 --------
# site_name_db_dict = {
# "us": "selection",
# "uk": "selection_uk",
# "de": "selection_de",
# "es": "selection_es",
# "fr": "selection_fr",
# "it": "selection_it",
# }
#
# db_type_alias_map = {
# "mysql": "mysql", # 阿里云mysql
# "postgresql_14": "postgresql_14", # pg14爬虫库-内网
# "postgresql_14_outer": "postgresql_14", # pg14爬虫库-外网
# "postgresql_15": "postgresql_15", # pg15正式库-内网
# "postgresql_15_outer": "postgresql_15_outer", # pg15正式库-外网
# "postgresql_cluster": "postgresql_cluster", # pg集群-内网
# "postgresql_cluster_outer": "postgresql_cluster_outer", # pg集群-外网
# "doris": "doris", # doris集群-内网
# }
#
# DEFAULT_SERVERS = [
# # "http://192.168.200.210:7777", # 内网
# # "http://192.168.10.217:7777", # 内网-h7
# # "http://113.100.143.162:7777", # 外网
# # "http://113.100.143.162:7779", # 外网
# # "http://61.145.136.61:7777", # 外网
# # "http://61.145.136.61:7779", # 外网
# "http://192.168.200.210:7780"
#
#
# ]
# # ---------------------------
#
# def df_to_json_records(df: pd.DataFrame) -> list:
# """保证 DataFrame 可安全序列化为 JSON records（处理 NaN / ±Inf）"""
# df_clean = df.copy()
#
# # 1️⃣ 替换 ±Inf -> NaN
# num_cols = df_clean.select_dtypes(include=[np.number]).columns
# if len(num_cols):
# df_clean[num_cols] = df_clean[num_cols].replace([np.inf, -np.inf], np.nan)
#
# # 2️⃣ 替换 NaN -> None（注意：有时 astype(object) 不彻底，需用 applymap）
# df_clean = df_clean.applymap(lambda x: None if pd.isna(x) else x)
#
# # 3️⃣ 转为 dict records
# return df_clean.to_dict("records")
#
#
# def clean_json_field_for_orjson(v):
# """清洗单个 JSON 字段的值，使其符合 orjson 要求并避免空字典入库"""
# if v is None or pd.isna(v):
# return None
#
# # 1️⃣ 如果是空字典对象，返回 None
# if isinstance(v, dict) and not v:
# return None
#
# # 2️⃣ 如果是空字符串或仅为 "{}"，返回 None
# if isinstance(v, str):
# stripped = v.strip()
# if not stripped or stripped == "{}":
# return None
# try:
# parsed = json.loads(stripped)
# if isinstance(parsed, dict) and not parsed:
# return None
# return json.dumps(parsed, ensure_ascii=False)
# except Exception:
# return v # 非 JSON 字符串则原样保留
#
# return v
#
#
# def fully_clean_for_orjson(df: pd.DataFrame) -> pd.DataFrame:
# # """全面清洗 DataFrame 以符合 orjson 要求"""
# # df = df.replace([np.inf, -np.inf], np.nan)
# # df = df.applymap(lambda x: None if pd.isna(x) else x)
# #
# # # 找出所有可能为 JSON 字符串的字段
# # json_like_cols = [col for col in df.columns if col.endswith('_json')]
# #
# # # 针对每个 JSON-like 字段，应用清洗函数
# # for col in json_like_cols:
# # df[col] = df[col].apply(clean_json_field_for_orjson)
# #
# # return df
# """全面清洗 DataFrame 以符合 orjson 要求"""
# df = df.replace([np.inf, -np.inf], np.nan)
#
# # NaN → None （比 applymap 高效且不出错）
# df = df.where(pd.notna(df), None)
#
# # 找出所有可能为 JSON 字符串的字段
# json_like_cols = [col for col in df.columns if col.endswith('_json')]
#
# # 针对每个 JSON-like 字段，应用清洗函数
# for col in json_like_cols:
# df[col] = df[col].apply(clean_json_field_for_orjson)
#
# return df
#
#
# class RemoteTransaction:
#
# def __init__(self, db: str, database: str,
# session: requests.Session, urls: List[str]):
# self.db = db
# self.database = database
# self.session = session
# self.urls = urls
# self.sql_queue = []
#
# # def execute(self, sql: str):
# # self.sql_queue.append(sql)
# def execute(self, sql: str, params=None):
# """
# params 可取：
# • None → 纯文本 SQL
# • dict → 单条参数化 e.g. {"id":1,"name":"a"}
# • list/tuple → 批量 executemany
# - list[dict] ↔ INSERT .. VALUES (:id,:name)
# - list[tuple] ↔ INSERT .. VALUES (%s,%s)
# """
# self.sql_queue.append({"sql": sql, "params": params})
#
# def __enter__(self): return self
#
# def __exit__(self, exc_type, exc, tb):
# for url in self.urls:
# try:
# self.session.post(
# url + "/transaction",
# json={"db": self.db,
# "sql_list": self.sql_queue,
# "site_name": self.database}, # site_name not needed on server, kept for clarity
# timeout=3000,
# ).raise_for_status()
# return
# except Exception as e:
# print(f"[WARN] 事务失败 {url}: {e}")
# raise RuntimeError("All servers failed for transaction")
#
#
# class RemoteEngine:
# def __init__(self, db: str, database: str,
# server_urls: List[str], retries: int = 2):
# self.db = db
# self.database = database
# self.urls = [u.rstrip("/") for u in server_urls]
# self.session = requests.Session()
# self.retries = retries
#
# def _request(self, endpoint: str, payload):
# for url in self.urls:
# for _ in range(self.retries):
# try:
# json_bytes = orjson.dumps(payload)
# r = self.session.post(f"{url}/{endpoint}",
# data=json_bytes,
# headers={"Content-Type": "application/json"},
# timeout=3000)
#
# # r = self.session.post(f"{url}/{endpoint}",
# # json=payload, timeout=10)
# r.raise_for_status()
# return r.json()
# except Exception as e:
# print(f"[WARN] {endpoint} fail @ {url}: {e}")
# time.sleep(1)
# raise RuntimeError(f"All servers failed for {endpoint}")
# # def _request(self, endpoint: str, payload):
# # # 用 orjson，“allow_nan” 会把 NaN/Inf 写成 null
# # # json_bytes = orjson.dumps(payload,
# # # option=orjson.OPT_NON_STR_KEYS | orjson.OPT_NAIVE_UTC | orjson.OPT_OMIT_MICROSECOND | orjson.OPT_ALLOW_INF_AND_NAN)
# # json_bytes = orjson.dumps(
# # payload,
# # option=orjson.OPT_NON_STR_KEYS | orjson.OPT_NAIVE_UTC | orjson.OPT_ALLOW_INF_AND_NAN
# # )
# #
# # headers = {"Content-Type": "application/json"}
# #
# # for url in self.urls:
# # for _ in range(self.retries):
# # try:
# # r = self.session.post(f"{url}/{endpoint}",
# # data=json_bytes, headers=headers,
# # timeout=15)
# # r.raise_for_status()
# # return r.json()
# # except Exception as e:
# # print(f"[WARN] {endpoint} fail @ {url}: {e}")
# # time.sleep(1)
# # raise RuntimeError(f"All servers failed for {endpoint}")
#
# # ---------- 公共 API ----------
# def read_sql(self, sql: str) -> pd.DataFrame:
# data = self._request("query",
# {"db": self.db,
# "sql": sql,
# "site_name": self.database})
# return pd.DataFrame(data["result"])
#
# def to_sql(self, df: pd.DataFrame, table: str, if_exists="append"):
#
# return self._request("insert",
# {"db": self.db,
# "table": table,
# "if_exists": if_exists,
# "data": fully_clean_for_orjson(df=df).to_dict("records"),
# # "data": df_to_json_records(df), # ← 清洗后的 records
# "site_name": self.database})
#
# def read_then_update(
# self,
# select_sql: str,
# update_table: str,
# set_values: dict,
# where_keys: List[str],
# error_if_empty: bool = False,
# ):
# """
# 动态生成 UPDATE：把 select_sql 读到的行，按 where_keys 精准更新 set_values
# 返回 (DataFrame, rows_updated)
# """
# payload = {
# "db": self.db,
# "site_name": self.database,
# "select_sql": select_sql,
# "update_table": update_table,
# "set_values": set_values,
# "where_keys": where_keys,
# "error_if_empty": error_if_empty,
# }
# resp = self._request("read_then_update", payload)
# df = pd.DataFrame(resp["read_result"])
# rows_updated = resp.get("rows_updated", 0)
# return df
#
# def begin(self):
# return RemoteTransaction(self.db, self.database,
# self.session, self.urls)
# # ---------------------------------
#
#
# def get_remote_engine(site_name: str, db_type: str,
# servers: List[str] = None) -> RemoteEngine:
# if site_name not in site_name_db_dict:
# raise ValueError(f"Unknown site_name: {site_name}")
# if db_type not in db_type_alias_map:
# raise ValueError(f"Unknown db_type: {db_type}")
# return RemoteEngine(
# db=db_type_alias_map[db_type],
# database=site_name,
# server_urls=servers or DEFAULT_SERVERS,
# )
import json
import sys,os
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
import pandas as pd
import numpy as np
import orjson, requests, time
from typing import List
# -------- 映射字典 --------
site_name_db_dict = {
"us": "selection",
"uk": "selection_uk",
"de": "selection_de",
"es": "selection_es",
"fr": "selection_fr",
"it": "selection_it",
}
db_type_alias_map = {
"mysql": "mysql", # 阿里云mysql
"postgresql_14": "postgresql_14", # pg14爬虫库-内网
"postgresql_14_outer": "postgresql_14_outer", # pg14爬虫库-外网
"postgresql_15": "postgresql_15", # pg15正式库-内网
"postgresql_15_outer": "postgresql_15_outer", # pg15正式库-外网
"postgresql_cluster": "postgresql_cluster", # pg集群-内网
"postgresql_cluster_outer": "postgresql_cluster_outer", # pg集群-外网
"doris": "doris", # doris集群-内网
"doris_adv": "doris_adv",
}
DEFAULT_SERVERS = [
# "http://192.168.200.210:7777", # 内网
# "http://192.168.10.217:7777", # 内网-h7
# "http://61.145.136.61:7777", # 外网
# "http://61.145.136.61:7779", # 外网
"http://61.145.136.61:7780"
]
# ---------------------------
def df_to_json_records(df: pd.DataFrame) -> list:
"""保证 DataFrame 可安全序列化为 JSON records（处理 NaN / ±Inf）"""
df_clean = df.copy()
# 1️⃣ 替换 ±Inf -> NaN
num_cols = df_clean.select_dtypes(include=[np.number]).columns
if len(num_cols):
df_clean[num_cols] = df_clean[num_cols].replace([np.inf, -np.inf], np.nan)
# 2️⃣ 替换 NaN -> None（兼容 pandas 1.x 和 2.x）
df_clean = df_clean.where(pd.notna(df_clean), None)
# 3️⃣ 转为 dict records
return df_clean.to_dict("records")
def clean_json_field_for_orjson(v):
"""清洗单个 JSON 字段的值，使其符合 orjson 要求并避免空字典入库"""
if v is None or pd.isna(v):
return None
# 1️⃣ 如果是空字典对象，返回 None
if isinstance(v, dict) and not v:
return None
# 2️⃣ 如果是空字符串或仅为 "{}"，返回 None
if isinstance(v, str):
stripped = v.strip()
if not stripped or stripped == "{}":
return None
try:
parsed = json.loads(stripped)
if isinstance(parsed, dict) and not parsed:
return None
return json.dumps(parsed, ensure_ascii=False)
except Exception:
return v # 非 JSON 字符串则原样保留
return v
def fully_clean_for_orjson(df: pd.DataFrame) -> pd.DataFrame:
"""全面清洗 DataFrame 以符合 orjson 要求"""
df = df.replace([np.inf, -np.inf], np.nan)
df = df.where(pd.notna(df), None)
# 找出所有可能为 JSON 字符串的字段
json_like_cols = [col for col in df.columns if col.endswith('_json')]
# 针对每个 JSON-like 字段，应用清洗函数
for col in json_like_cols:
df[col] = df[col].apply(clean_json_field_for_orjson)
return df
class RemoteTransaction:
def __init__(self, db: str, database: str,
session: requests.Session, urls: List[str]):
self.db = db
self.database = database
self.session = session
self.urls = urls
self.sql_queue = []
# def execute(self, sql: str):
# self.sql_queue.append(sql)
def execute(self, sql: str, params=None):
"""
params 可取：
• None → 纯文本 SQL
• dict → 单条参数化 e.g. {"id":1,"name":"a"}
• list/tuple → 批量 executemany
- list[dict] ↔ INSERT .. VALUES (:id,:name)
- list[tuple] ↔ INSERT .. VALUES (%s,%s)
"""
self.sql_queue.append({"sql": sql, "params": params})
def __enter__(self): return self
def __exit__(self, exc_type, exc, tb):
for url in self.urls:
try:
self.session.post(
url + "/transaction",
json={"db_type": self.db,
"sql_list": self.sql_queue,
"database": self.database}, # site_name not needed on server, kept for clarity
timeout=3000,
).raise_for_status()
return
except Exception as e:
print(f"[WARN] 事务失败 {url}: {e}")
raise RuntimeError("All servers failed for transaction")
class RemoteEngine:
def __init__(self, db: str, database: str,
server_urls: List[str], retries: int = 20):
"""
:param db: db_type--数据库类型
:param database: 数据库名称
:param server_urls:
:param retries:
"""
self.db = db
self.database = database
self.urls = [u.rstrip("/") for u in server_urls]
self.session = requests.Session()
self.retries = retries
def _request(self, endpoint: str, payload):
for url in self.urls:
for _ in range(self.retries):
try:
json_bytes = orjson.dumps(payload)
r = self.session.post(f"{url}/{endpoint}",
data=json_bytes,
headers={"Content-Type": "application/json"},
timeout=3600)
# r = self.session.post(f"{url}/{endpoint}",
# json=payload, timeout=10)
r.raise_for_status()
return r.json()
except Exception as e:
print(f"[WARN] {endpoint} fail @ {url}: {e}")
time.sleep(1)
raise RuntimeError(f"All servers failed for {endpoint}")
# ---------- 公共 API ----------
def read_sql(self, sql: str) -> pd.DataFrame:
data = self._request("query",
{"db_type": self.db,
"sql": sql,
"database": self.database})
return pd.DataFrame(data["result"])
def to_sql(self, df: pd.DataFrame, table: str, if_exists="append"):
return self._request("insert",
{"db_type": self.db,
"table": table,
"if_exists": if_exists,
"data": fully_clean_for_orjson(df=df).to_dict("records"),
# "data": df_to_json_records(df), # ← 清洗后的 records
"database": self.database})
def read_then_update(
self,
select_sql: str,
update_table: str,
set_values: dict,
where_keys: List[str],
error_if_empty: bool = False,
):
"""
动态生成 UPDATE：把 select_sql 读到的行，按 where_keys 精准更新 set_values
返回 (DataFrame, rows_updated)
"""
payload = {
"db_type": self.db,
"database": self.database,
"select_sql": select_sql,
"update_table": update_table,
"set_values": set_values,
"where_keys": where_keys,
"error_if_empty": error_if_empty,
}
resp = self._request("read_then_update", payload)
df = pd.DataFrame(resp["read_result"])
rows_updated = resp.get("rows_updated", 0)
return df
def begin(self):
return RemoteTransaction(self.db, self.database,
self.session, self.urls)
# ---------------------------------
# db -- 数据库类型
# database -- 站点
def get_remote_engine(site_name: str, db_type: str, database: str = None,
servers: List[str] = None) -> RemoteEngine:
"""
:param site_name: 站点
:param db_type: 数据库类型
:param database: 数据库名称-默认为None， 否则通过站点来匹配
:param servers: 服务器url地址
:return: 数据库连接对象
"""
if not database:
database = site_name_db_dict[site_name]
if site_name not in site_name_db_dict:
raise ValueError(f"Unknown site_name: {site_name}")
if db_type not in db_type_alias_map:
raise ValueError(f"Unknown db_type: {db_type}")
print(f"db_type: {db_type_alias_map[db_type]}, database: {database}")
return RemoteEngine(
db=db_type_alias_map[db_type],
database=database,
server_urls=servers or DEFAULT_SERVERS,
)
def stream_load_df(df: pd.DataFrame, database: str='advertising_manager', table: str='', server_url: str = None):
server = (server_url or DEFAULT_SERVERS[0]).rstrip("/")
payload = {
"database": database,
"table": table,
"data": fully_clean_for_orjson(df).to_dict("records"),
}
json_bytes = orjson.dumps(payload)
r = requests.post(
f"{server}/stream_load_df",
data=json_bytes,
headers={"Content-Type": "application/json"},
timeout=3000,
)
r.raise_for_status()
return r.json()
if __name__ == '__main__':
engine_mysql = get_remote_engine(
site_name='us', # -> database "selection"
db_type="doris_adv", # -> 服务端 alias "mysql"
database="advertising_manager", # -> 服务端 alias "mysql"
)
# e = engine_mysql.read_sql('SELECT max(created_time),MIN(created_time) FROM us_sp_search_term_rank WHERE time_batch="2026-03-09-15"')
# e = engine_mysql.read_sql('SELECT * FROM us_sp_search_term_rank WHERE time_batch="2026-03-09-12" and page_row=68')
# e = engine_mysql.read_sql('SELECT * FROM us_sp_search_term_rank WHERE id=50564433')
e = engine_mysql.read_sql('SELECT count(id),time_batch FROM us_sp_search_term_rank GROUP BY time_batch ORDER BY time_batch ASC')
# e = engine_mysql.read_sql('SELECT count(id),time_batch FROM us_sp_search_term_rank GROUP BY time_batch ORDER BY time_batch ASC')
# e = engine_mysql.read_sql('SELECT count(id) FROM sp_keyword_position_keyword')
# e = engine_mysql.read_sql('select * from us_sp_search_term_syn limit 100')
# dog bandanas spring
# 2026-04-03-11
# e = engine_mysql.read_sql('select * from us_sp_search_term_syn where search_term = "dog bandanas spring"')
# e = engine_mysql.read_sql('select * from us_sp_search_term_rank where search_term="dog bandanas spring" and time_batch="2026-04-03-11" and cate_type="sp" and asin="B0GCN7CTCM"')
# e = engine_mysql.read_sql('select * from us_sp_search_term_rank where time_batch="2026-03-17-16" and cate_type="sp" limit 100')
# print(e)
print(e.columns)
print(e.values.tolist())
wangjing_projects/projects/tiktok/TK_video_data/tk_video_ch_CC.py
......@@ -102,6 +102,9 @@ class TkVideo():
'xpath://div[.//*[contains(text(),"Upgrade your account")]]//button[.//svg][1]',
# 新增：通用兜底，任何 dialog 的右上角第一个无文字 svg 按钮
'xpath:(//div[@role="dialog"]//button[.//svg and not(.//span)])[1]',
# —— 新增：TikTok "商业工具已解锁" 弹窗 ——
'xpath://button[.//div[text()="开始使用"]]',
'xpath://button[.//div[text()="Get started"]]',
]
for xp in candidates:
try:
......@@ -124,6 +127,14 @@ class TkVideo():
time.sleep(2)
self.dismiss_popups() # 二次复查，覆盖延迟出现的"升级你的账号"弹窗
# 如果弹窗导致页面跳转，重新回到视频数据页
if 'insight/video' not in self.page_chrome.url:
print('⚠️ 弹窗导致页面跳转，重新回到视频数据页')
self.page_chrome.get("https://www.tiktok.com/business-suite/insight/video")
time.sleep(random.randint(6, 10))
self.dismiss_popups()
time.sleep(2)
export_orders = self.page_chrome.ele('xpath://span[text()="自定义"]', timeout=13)
export_orders.click()
print('点击自定义')
......@@ -192,15 +203,66 @@ class TkVideo():
timeout=15
).text
print(f'已获取店铺名: {self.shop_name}')
self.key = self.page_chrome.ele(
'xpath://img[contains(@class,"rounded-full")]/following-sibling::div//div[contains(@class,"text-P3-Regular") and contains(@class,"text-color-TextSecondary")]',
timeout=15
).text.lstrip('@')
self.key = self._get_account_key(self.page_chrome)
print(f'已获取账号用户名: {self.key}')
sleep(randint(5, 10))
self.save_to_redis()
time.sleep(5)
def _get_account_key(self, page):
"""多策略获取TikTok账号用户名，带交叉验证"""
results = {}
# 策略1: class XPath
try:
val = page.ele(
'xpath://img[contains(@class,"rounded-full")]/following-sibling::div'
'//div[contains(@class,"text-P3-Regular") and contains(@class,"text-color-TextSecondary")]',
timeout=10
).text.strip()
if val.startswith('@') and len(val) > 1:
results['class_xpath'] = val.lstrip('@')
else:
print(f' [策略1] 命中元素但内容非用户名: [{val}]')
except Exception as e:
print(f' [策略1] class XPath 失败: {e}')
# 策略2: 头像容器内找 @ 开头文本
try:
val = page.ele(
'xpath://img[contains(@class,"rounded-full")]/parent::div//*[starts-with(text(),"@")]',
timeout=10
).text.strip()
if val.startswith('@') and len(val) > 1:
results['at_prefix'] = val.lstrip('@')
except Exception as e:
print(f' [策略2] @ 前缀搜索失败: {e}')
# 策略3: "粉丝" 文字反向定位同行第一个 div
try:
val = page.ele(
'xpath://span[text()="粉丝"]/parent::div/parent::div/div[1]',
timeout=5
).text.strip()
if val.startswith('@') and len(val) > 1:
results['fans_anchor'] = val.lstrip('@')
except Exception as e:
print(f' [策略3] 粉丝锚点失败: {e}')
if not results:
raise Exception('所有策略均无法获取账号用户名，TikTok页面可能改版')
unique = set(results.values())
if len(unique) == 1:
key = unique.pop()
print(f' [校验通过] {len(results)}个策略一致: {list(results.keys())}')
return key
print(f' [校验警告] 策略结果不一致: {results}')
if 'at_prefix' in results:
return results['at_prefix']
return list(results.values())[0]
def connect_redis(self):
"""建立 Redis 连接"""
self.r = redis.StrictRedis(**self.REDIS_CONFIG)
......@@ -227,8 +289,7 @@ class TkVideo():
'添加到收藏': 'favorites'
}
# 强制所有列为字符串类型，防止科学计数法
df = pd.read_excel(file_path, dtype=str)
df = pd.read_excel(file_path)
# 替换列名为英文
df.rename(columns=column_mapping, inplace=True)
......@@ -337,13 +398,15 @@ class TkVideo():
print('删除数据失败')
def send_error_notification_via_wechat(self,error_message):
webhook_url = 'http://47.112.96.71:8082/selection/sendMessage' # 替换为你的企业微信机器人的Webhook URL
self.receiver_name = 'pengyanbing,zouliqing'
host = "http://120.79.147.190:8080"
webhook_url = f'{host}/soundasia_selection/dolphinScheduler/sendMessage'
data = {
"account": self.receiver_name,
'title':'【TK视频数据下载异常提醒】',
'content':f'账号：{self.key}，错误信息：{error_message}, 时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
'account': self.receiver_name,
'title': '【TK视频数据下载异常提醒】',
'content': f'账号：{self.key}，错误信息：{error_message}, 时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}',
'msgtype': 'textcard'
}
response = requests.post(url=webhook_url, data=data,timeout=15)
if response.status_code == 200:
print("已成功发送错误通知到企业微信")
......
wangjing_projects/projects/tiktok/TK_video_data/tk_video_ch_TS.py deleted 100644 → 0
# 打包exe 命令 D:\Python3.7\python.exe -m PyInstaller -F E:\Git_new\spider\wangjing_projects\projects\tiktok\TK_video_data\tk_video_ch_CX.py
import os
os.environ['NO_PROXY'] = 'stackoverflow.com'
import logging
logging.captureWarnings(True)
from DrissionPage import ChromiumPage, ChromiumOptions
import time
from datetime import datetime, timedelta
from time import sleep
from random import randint
import requests
import math
import pandas as pd
import redis
import json
from pathlib import Path
import re
import random
class TkVideo():
def __init__(self):
# 修改请求头
self.headers = {
'accept': '*/*',
'accept-language': 'en-US,en;q=0.9', # 'en-US,en;q=0.9'
'cache-control': 'no-cache',
'content-type': 'application/json; charset=UTF-8',
'origin': 'https://www.tiktok.com',
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': 'https://www.tiktok.com/',
'sec-ch-ua': '"Google Chrome";v="135", "Not-A.Brand";v="8", "Chromium";v="135"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'cross-site',
'sec-fetch-storage-access': 'active',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36',
}
self.key = "Tessco Shop"
self.download_folder = r"D:\Downloads"
self.receiver_name = 'pengyanbing'
# Redis 配置信息
self.REDIS_CONFIG = {
'host': '120.79.147.190',
'port': 6379,
'password': 'fG7#vT6kQ1pX',
'db': 13,
'decode_responses': True
}
# self.page_chrome = ChromiumPage()
# 配置 Chrome 浏览器 - 端口 9222
chrome_options = ChromiumOptions()
chrome_options.set_browser_path(r'C:\Program Files\Google\Chrome\Application\chrome.exe')
chrome_options.set_local_port(9333) # 设置 Chrome 的调试端口
# 锁定 profile 到 .py 跑时使用的目录，保持登录态（解决 exe 打包后未登录问题）
chrome_options.set_user_data_path(os.path.join(os.environ.get('LOCALAPPDATA', ''), 'Temp', 'DrissionPage', f'userData_{9333}'))
# 屏蔽浏览器层弹窗/通知
chrome_options.set_pref('profile.default_content_setting_values.notifications', 2)
chrome_options.set_argument('--disable-popup-blocking')
chrome_options.set_argument('--disable-notifications')
chrome_options.set_argument('--disable-infobars')
self.page_chrome = ChromiumPage(addr_or_opts=chrome_options)
print(f"Chrome 浏览器运行在端口: {9333}")
def get_datetime(self):
"""获取当前日期，并计算前2天的完整日期（年-月-日），并按照指定格式输出"""
# 获取今天的日期
today = datetime.today().date()
# 开始日期：去年的今天
self.start_date = today.replace(year=today.year - 1)
self.start_year = self.start_date.year
self.start_month = self.start_date.month
self.start_day = self.start_date.day
# 结束日期：今天的前一天
self.end_date = today - timedelta(days=3)
self.end_year = self.end_date.year
self.end_month = self.end_date.month
self.end_day = self.end_date.day
def dismiss_popups(self):
"""只关模态层弹窗，不点业务卡片。仅在页面初次加载后调用一次。"""
candidates = [
'xpath://button[.//span[text()="同意" or text()="Accept all" or text()="Accept"]]',
'xpath://button[.//span[text()="我知道了" or text()="Got it" or text()="OK"]]',
'xpath://button[.//span[text()="跳过" or text()="Skip"]]',
'xpath://button[.//span[text()="稍后" or text()="Later" or text()="Maybe later"]]',
'xpath://div[@role="dialog"]//button[@aria-label="Close" or @aria-label="关闭"]',
'xpath://div[@role="alertdialog"]//button[@aria-label="Close" or @aria-label="关闭"]',
'xpath://div[contains(@class,"modal")]//button[@aria-label="Close" or @aria-label="关闭"]',
'xpath://div[@role="dialog"]//div[contains(@class,"justify-end")]/button[.//svg]',
'xpath:(//div[@role="dialog"]//button[.//svg])[1]',
'xpath://div[contains(@class,"cookie-banner") or contains(@class,"consent")]//button',
# 新增：TikTok "升级你的账号" 弹窗（无 aria-label，靠标题文本定位右上角 X 按钮）
'xpath://div[.//*[text()="升级你的账号"]]//button[.//svg][1]',
'xpath://div[.//*[contains(text(),"Upgrade your account")]]//button[.//svg][1]',
# 新增：通用兜底，任何 dialog 的右上角第一个无文字 svg 按钮
'xpath:(//div[@role="dialog"]//button[.//svg and not(.//span)])[1]',
]
for xp in candidates:
try:
ele = self.page_chrome.ele(xp, timeout=1.5)
if ele:
ele.click()
print(f'已关闭弹窗: {xp}')
time.sleep(1)
except Exception:
pass
def get_day(self):
self.page_chrome.get("https://www.tiktok.com/business-suite/insight/video")
self.page_chrome.set.window.max()
# 等待页面初始加载
time.sleep(random.randint(6, 10))
# 关闭可能出现的弹窗
self.dismiss_popups()
time.sleep(2)
self.dismiss_popups() # 二次复查，覆盖延迟出现的"升级你的账号"弹窗
export_orders = self.page_chrome.ele('xpath://span[text()="自定义"]', timeout=13)
export_orders.click()
print('点击自定义')
time.sleep(random.randint(5, 10))
# 先点击开始时间
self.page_chrome.ele(
f"xpath=//div[@class='tiktok-datepicker-month-title' and contains(text(), '{self.start_year} {self.start_month} 月')]"
f"/following-sibling::div[@class='tiktok-datepicker-day-wrapper']"
f"//div[@class='tiktok-datepicker-day valid in-this-month']"
f"//span[text()='{self.start_day}']/parent::div"
).click()
print(f'已输入开始时间{self.start_year} {self.start_month} 月 {self.start_day} 日')
time.sleep(random.randint(3, 5))
last_err = None
clicked = False
for i in range(10):
try:
xpath = (
f"//div[@class='tiktok-datepicker-month-title' and contains(text(), '{self.end_year} {self.end_month} 月')]"
f"/following-sibling::div[@class='tiktok-datepicker-day-wrapper']"
f"//span[text()='{self.end_day}']/parent::div"
)
print('结束日期 xpath::', xpath)
ele = self.page_chrome.ele(f"xpath={xpath}", timeout=8)
ele.click()
print(f'✅ 成功点击日期：{self.end_year}-{self.end_month}-{self.end_day}')
clicked = True
break
except Exception as e:
last_err = e
print(f'❌ 无法点击 {self.end_year}-{self.end_month}-{self.end_day}，错误：{e}')
time.sleep(random.randint(5, 15))
if not clicked:
raise RuntimeError(f'连续 10 次点击日期失败，最后错误：{last_err}')
self.get_data()
time.sleep(random.randint(3, 5))
return True
def get_data(self):
self.page_chrome.ele('xpath://div[text()="更新"]', timeout=13).click()
print('已点击更新')
sleep(randint(5, 10))
self.page_chrome.ele('xpath://span[text()="下载数据"]', timeout=13).click()
print('已点击下载数据')
sleep(randint(5, 10))
self.page_chrome.ele('xpath://span[text()="Xlsx"]', timeout=13).click()
print('已点击Xlsx')
sleep(randint(5, 10))
self.page_chrome.ele('xpath://button[text()="下载数据"]', timeout=13).click()
print('已点击下载数据')
sleep(randint(5, 10))
# 点击首页 获取店铺名称
self.page_chrome.ele('xpath://span[text()="首页"]', timeout=13).click()
print('已点击首页')
sleep(randint(5, 10))
self.shop_name = self.page_chrome.ele(
'xpath://img[contains(@class,"rounded-full")]/following-sibling::div//div[contains(@class,"text-H6-Bold")]',
timeout=15
).text
print(f'已获取店铺名: {self.shop_name}')
self.key = self.page_chrome.ele(
'xpath://img[contains(@class,"rounded-full")]/following-sibling::div//div[contains(@class,"text-P3-Regular") and contains(@class,"text-color-TextSecondary")]',
timeout=15
).text.lstrip('@')
print(f'已获取账号用户名: {self.key}')
sleep(randint(5, 10))
self.save_to_redis()
time.sleep(5)
def connect_redis(self):
"""建立 Redis 连接"""
self.r = redis.StrictRedis(**self.REDIS_CONFIG)
try:
self.r.ping() # 测试连接
print("✅ 成功连接到 Redis")
except redis.exceptions.ConnectionError as e:
print(f"❌ 无法连接到 Redis: {e}")
raise
def read_excel(self, file_path):
"""读取 Excel 文件内容，并防止数值被转为科学计数法"""
print(f"📄 正在读取文件：{file_path}")
# 定义列名映射（中文 → 英文）
column_mapping = {
'视频标题': 'video_title',
'视频链接': 'video_url',
'发布时间': 'publish_date',
'视频观看次数': 'views',
'点赞数': 'likes',
'评论数': 'comments',
'分享次数': 'shares',
'添加到收藏': 'favorites'
}
# 强制所有列为字符串类型，防止科学计数法
df = pd.read_excel(file_path, dtype=str)
# 替换列名为英文
df.rename(columns=column_mapping, inplace=True)
data = df.to_dict(orient='records') # 转换为字典列表
print(f"📊 已读取 {len(data)} 条记录")
return data
def process_data(self, data, account):
processed_data = []
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
for record in data:
# 提取 content_id
video_link = record.get('video_url', '')
if video_link:
content_id = video_link.split('/')[-1]
else:
content_id = ''
# 添加 account 和 update_time 字段
processed_record = {
'account': account,
'content_id': content_id,
'update_time': current_time,
**record # 合并原始记录
}
# 替换 NaN 和 None 为 空字符串
cleaned_record = {
key: ("" if pd.isna(value) or value is None or str(value).strip().lower() == "nan" else value)
for key, value in processed_record.items()
}
processed_data.append(cleaned_record)
return processed_data
def store_data_in_redis(self, r, data):
"""将数据存储到 Redis 中，对相同的 shop_code 清除旧数据后写入新数据"""
key = f"tk_video_data:{self.key}:order:list"
# ⚠️ 先删除旧数据（实现“覆盖”）
if r.exists(key):
r.delete(key)
print(f"🗑️ 已清除旧数据: {key}")
# 写入新数据
pipe = r.pipeline()
for record in data:
value = json.dumps(record, ensure_ascii=False)
pipe.rpush(key, value)
pipe.execute() # 批量执行，提高效率
print(f"💾 已写入新数据到键: {key}，共 {len(data)} 条记录")
def _download_name_pattern(self):
"""返回本次日期段下载文件的匹配正则（清理 / 查找复用）"""
start_month = f'0{self.start_month}' if self.start_month < 10 else self.start_month
start_day = f'0{self.start_day}' if self.start_day < 10 else self.start_day
end_month = f'0{self.end_month}' if self.end_month < 10 else self.end_month
end_day = f'0{self.end_day}' if self.end_day < 10 else self.end_day
base_prefix = f"视频({self.start_year}_{start_month}_{start_day}-{self.end_year}_{end_month}_{end_day})"
return base_prefix, re.escape(base_prefix) + r'.*$'
def cleanup_old_downloads(self):
"""清理本次日期段的旧下载文件，防止重试时读到第 1 次的脏文件"""
download_path = Path(self.download_folder)
if not download_path.exists():
return
base_prefix, pattern = self._download_name_pattern()
cleaned = 0
for f in download_path.iterdir():
if f.is_file() and re.fullmatch(pattern, f.name):
try:
f.unlink()
cleaned += 1
print(f'🗑️ 清理旧下载文件: {f.name}')
except Exception as e:
print(f'⚠️ 清理失败 {f.name}: {e}')
if cleaned == 0:
print(f'🗑️ 无匹配旧文件需要清理（前缀={base_prefix}）')
def find_specific_file(self):
download_path = Path(self.download_folder)
base_prefix, pattern = self._download_name_pattern()
print("匹配模式:", pattern)
for file in download_path.iterdir():
if file.is_file() and re.fullmatch(pattern, file.name):
return str(file)
raise FileNotFoundError(f"未找到匹配 {base_prefix} 的文件")
def save_to_redis(self):
EXCEL_FILE = self.find_specific_file()
print(f'保存文件：{EXCEL_FILE}')
# 读取 Excel 数据
data = self.read_excel(EXCEL_FILE)
processed_data = self.process_data(data, self.shop_name)
self.store_data_in_redis(self.r, processed_data)
try:
print('删除下载文件',EXCEL_FILE)
os.remove(EXCEL_FILE)
except:
print('删除数据失败')
def send_error_notification_via_wechat(self,error_message):
webhook_url = 'http://47.112.96.71:8082/selection/sendMessage' # 替换为你的企业微信机器人的Webhook URL
data = {
"account": self.receiver_name,
'title':'【TK视频数据下载异常提醒】',
'content':f'账号：{self.key}，错误信息：{error_message}, 时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
}
response = requests.post(url=webhook_url, data=data,timeout=15)
if response.status_code == 200:
print("已成功发送错误通知到企业微信")
else:
print(f"发送错误通知失败: {response.text}")
def run(self):
self.connect_redis()
self.get_datetime()
max_attempts = 2 # 失败一概重试 1 次（共 2 次）
for attempt in range(1, max_attempts + 1):
try:
print(f'=== 第 {attempt}/{max_attempts} 次尝试 ===')
self.cleanup_old_downloads() # 清理本次日期段的旧脏文件
self.get_day()
print(f'✅ 第 {attempt} 次执行成功')
break
except Exception as e:
print(f'❌ 第 {attempt} 次执行失败：{e}')
if attempt < max_attempts:
wait = random.randint(15, 30)
print(f'⏳ {wait} 秒后从头重试 ...')
time.sleep(wait)
else:
print('⛔ 已达最大重试次数，发送企业微信通知')
self.send_error_notification_via_wechat(f'重试 {max_attempts} 次仍失败：{e}')
print('完成关闭浏览器')
time.sleep(5)
self.page_chrome.quit()
if __name__ == '__main__':
TkVideo().run()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment