Skip to content

S
spider
Commit 39a2e885 by Peng
Options

no message

parent fb663d1b
Showing with 299 additions and 272 deletions
py_spider/amzon_parse_db_html/test_requests.py 0 → 100644
import requests
import json
import time
LENS_API = "http://192.168.10.8:8000"
def test_health():
r = requests.get(f"{LENS_API}/health", timeout=5)
print("Health:", r.json())
return r.status_code == 200
def test_upload_image(image_path):
print(f"\n上传图片: {image_path}")
t0 = time.time()
with open(image_path, "rb") as f:
r = requests.post(
f"{LENS_API}/api/lens",
files={"image": (image_path.split("\\")[-1], f, "image/png")},
data={"hl": "zh-CN"},
timeout=300,
)
elapsed = time.time() - t0
data = r.json()
print(f"状态: {r.status_code} | 耗时: {elapsed:.1f}s")
print(f"匹配数: {data.get('visual_matches_count', 0)}")
# AI 概览
ai = data.get("ai_overview", {})
if ai.get("available"):
print(f"\n--- AI 概览 ---")
print(f"摘要: {ai.get('summary', '')[:300]}")
for b in ai.get("bullets", [])[:10]:
print(f" - {b[:150]}")
else:
print(f"\nAI 概览: 无")
# 商品匹配
print(f"\n--- 商品匹配 (前10) ---")
for i, m in enumerate(data.get("visual_matches", [])[:10], 1):
price = f" | {m['price']}" if m.get("price") else ""
print(f" {i}. {m.get('title', '')[:60]}{price}")
print(f" {m.get('url', '')[:90]}")
return data
def test_url_image(image_url):
print(f"\nURL 搜图: {image_url[:80]}...")
t0 = time.time()
r = requests.post(
f"{LENS_API}/api/lens",
data={"image_url": image_url, "hl": "zh-CN"},
timeout=300,
)
elapsed = time.time() - t0
data = r.json()
print(f"状态: {r.status_code} | 耗时: {elapsed:.1f}s")
print(f"匹配数: {data.get('visual_matches_count', 0)}")
# AI 概览
ai = data.get("ai_overview", {})
if ai.get("available"):
print(f"\n--- AI 概览 ---")
print(f"摘要: {ai.get('summary', '')[:300]}")
for b in ai.get("bullets", [])[:10]:
print(f" - {b[:150]}")
else:
print(f"\nAI 概览: 无")
# 商品匹配
print(f"\n--- 商品匹配 (前10) ---")
for i, m in enumerate(data.get("visual_matches", [])[:10], 1):
price = f" | {m['price']}" if m.get("price") else ""
print(f" {i}. {m.get('title', '')[:60]}{price}")
print(f" {m.get('url', '')[:90]}")
return data
if __name__ == "__main__":
if not test_health():
print("服务不可用!")
exit(1)
# 上传图片搜图（走 Lens upload 路径，返回商品匹配）
test_upload_image(r"C:\Users\ASUS\Downloads\61fEImDN-kL.jpg")
# URL 搜图
# test_url_image("https://m.media-amazon.com/images/I/61fEImDN-kL.jpg")
py_spider/comment_spider/account_manager.py 0 → 100644
"""账号池管理 — 通过 RemoteEngine HTTP 代理，多机器竞争领取 + 状态回写 + 心跳"""
import os
import sys
import socket
import time
import uuid
import logging
import threading
import pandas as pd
from configparser import ConfigParser
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))
try:
from client import get_remote_engine
except ImportError:
get_remote_engine = None
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
class AccountManager:
def __init__(self, config_path="config.ini"):
cfg = ConfigParser()
cfg.read(config_path, encoding="utf-8")
mac_suffix = uuid.getnode().__format__('012x')[-4:]
self.machine_id = cfg.get("spider", "machine_id", fallback="") or f"{socket.gethostname()}-{mac_suffix}"
self.accounts_per_machine = cfg.getint("spider", "accounts_per_machine", fallback=3)
self.heartbeat_interval = cfg.getint("spider", "heartbeat_interval", fallback=600)
self._heartbeat_thread = None
self._heartbeat_stop = threading.Event()
if get_remote_engine:
self._engine = get_remote_engine("us", "mysql", database_name="selection")
else:
self._engine = None
logging.info(f"AccountManager init: machine={self.machine_id}")
def _query(self, sql, action="DB"):
for attempt in range(3):
try:
df = self._engine.read_sql(sql)
return df.to_dict("records")
except Exception as e:
logging.warning(f"{action}失败({attempt+1}/3): {e}")
time.sleep(3)
logging.error(f"{action}重试3次仍失败")
return []
def _exec(self, sql, action="DB"):
for attempt in range(3):
try:
with self._engine.begin() as tx:
tx.execute(sql)
return True
except Exception as e:
logging.warning(f"{action}失败({attempt+1}/3): {e}")
time.sleep(3)
logging.error(f"{action}重试3次仍失败")
return False
def reclaim_dead(self, timeout_minutes=60):
"""回收超时未心跳的僵尸账号"""
self._exec(
f"UPDATE share_account SET machine_id = NULL, locked_at = NULL "
f"WHERE locked_at < DATE_SUB(NOW(), INTERVAL {timeout_minutes} MINUTE) "
f"AND machine_id IS NOT NULL",
action="回收僵尸"
)
def acquire_accounts(self, count=None):
"""领取空闲账号（UPDATE 原子操作，不用 FOR UPDATE）"""
count = count or self.accounts_per_machine
self.reclaim_dead()
already = self._query(
f"SELECT * FROM share_account WHERE machine_id = '{self.machine_id}' AND status = 1",
)
if len(already) >= count:
logging.info(f"已持有 {len(already)} 个账号，无需再领")
return already
need = count - len(already)
self._exec(
f"UPDATE share_account SET machine_id = '{self.machine_id}', locked_at = NOW() "
f"WHERE status = 1 AND (machine_id IS NULL OR machine_id = '') "
f"AND (cooldown_until IS NULL OR cooldown_until < NOW()) "
f"ORDER BY fail_count ASC, account_id ASC LIMIT {need}",
action="领取账号"
)
result = self._query(
f"SELECT * FROM share_account WHERE machine_id = '{self.machine_id}' AND status = 1",
)
logging.info(f"当前持有 {len(result)} 个账号: {[r['account_id'] for r in result]}")
return result
def release_account(self, account_id):
self._exec(
f"UPDATE share_account SET machine_id = NULL, locked_at = NULL WHERE account_id = {account_id}",
action="释放账号"
)
def acquire_one_more(self):
self._exec(
f"UPDATE share_account SET machine_id = '{self.machine_id}', locked_at = NOW() "
f"WHERE status = 1 AND (machine_id IS NULL OR machine_id = '') "
f"AND (cooldown_until IS NULL OR cooldown_until < NOW()) "
f"ORDER BY fail_count ASC, account_id ASC LIMIT 1",
action="补领账号"
)
rows = self._query(
f"SELECT * FROM share_account WHERE machine_id = '{self.machine_id}' AND status = 1 "
f"ORDER BY locked_at DESC LIMIT 1",
)
if rows:
logging.info(f"补领账号 {rows[0]['account_id']}")
return rows[0]
logging.warning("无可用账号可补领")
return None
def mark_locked(self, account_id):
self._exec(
f"UPDATE share_account SET status = 2, fail_count = fail_count + 1, "
f"last_fail_at = NOW(), last_fail_reason = 'account_locked', "
f"machine_id = NULL, locked_at = NULL WHERE account_id = {account_id}",
action="标记锁定"
)
logging.warning(f"账号 {account_id} 被锁(account_locked) → status=2")
def mark_suspended(self, account_id):
self._exec(
f"UPDATE share_account SET status = 0, fail_count = fail_count + 1, "
f"last_fail_at = NOW(), last_fail_reason = 'suspended', "
f"machine_id = NULL, locked_at = NULL WHERE account_id = {account_id}",
action="标记封禁"
)
logging.warning(f"账号 {account_id} 被封禁(suspended) → status=0")
def mark_risk(self, account_id, reason="risk_detected"):
self._exec(
f"UPDATE share_account SET status = 2, fail_count = fail_count + 1, "
f"last_fail_at = NOW(), last_fail_reason = '{reason}', "
f"machine_id = NULL, locked_at = NULL, "
f"cooldown_until = DATE_ADD(NOW(), INTERVAL 30 MINUTE) "
f"WHERE account_id = {account_id}",
action="标记风控"
)
logging.warning(f"账号 {account_id} 风控({reason}) → status=2, 冷却30分钟")
def mark_login_fail(self, account_id, reason="wrong_password"):
self._exec(
f"UPDATE share_account SET fail_count = fail_count + 1, "
f"last_fail_at = NOW(), last_fail_reason = '{reason}' "
f"WHERE account_id = {account_id}",
action="标记登录失败"
)
def mark_login_ok(self, account_id):
self._exec(
f"UPDATE share_account SET fail_count = 0, last_fail_reason = NULL, "
f"locked_at = NOW() WHERE account_id = {account_id}",
action="标记登录成功"
)
def heartbeat(self):
self._exec(
f"UPDATE share_account SET locked_at = NOW() WHERE machine_id = '{self.machine_id}'",
action="心跳"
)
def start_heartbeat(self):
def _loop():
while not self._heartbeat_stop.is_set():
try:
self.heartbeat()
except Exception as e:
logging.warning(f"心跳失败: {e}")
self._heartbeat_stop.wait(self.heartbeat_interval)
self._heartbeat_thread = threading.Thread(target=_loop, daemon=True)
self._heartbeat_thread.start()
logging.info(f"心跳线程启动(每{self.heartbeat_interval}秒)")
def stop_heartbeat(self):
self._heartbeat_stop.set()
def release_all(self):
self._exec(
f"UPDATE share_account SET machine_id = NULL, locked_at = NULL "
f"WHERE machine_id = '{self.machine_id}'",
action="释放全部"
)
logging.info(f"释放本机全部账号")
def check_gateway(self):
"""轻量检查网关是否可达"""
try:
self._engine.read_sql("SELECT 1")
return True
except Exception:
return False
def get_my_accounts(self):
return self._query(
f"SELECT * FROM share_account WHERE machine_id = '{self.machine_id}' AND status = 1",
)
py_spider/threading_spider/Poll_site_search_term_pg.py deleted 100644 → 0
import sys
import os
from datetime import datetime, timedelta
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from multiprocessing import Pool
from amazon_spider.search_term_pg_week import search_temp_pg
from amazon_spider.search_term_mysql import search_temp_mysql
from threading_spider.db_connectivity import connect_db
import time
import pandas as pd
import random
import requests
import socket
from amazon_spider.VPS_IP import pppoe_ip
from threading_spider.post_to_dolphin import DolphinschedulerHelper
def select_sate_mysql(site, num=None, week=None):
db_class = connect_db(site)
engine_pg = db_class.pg_db() # pg
cursor_us, connect_us = db_class.us_mysql_db() # us站点
print('week::', week)
if int(week) < 10:
week = f'0{int(week)}'
else:
week = week
if num is None:
select_state1_sql = f"select status_val from workflow_progress where site_name='{site}' and date_info='2024-{week}' and date_type='week' and page='ABA搜索词';"
print(select_state1_sql)
cursor_us.execute(select_state1_sql)
site_sate1_list = cursor_us.fetchone()
try:
if site_sate1_list[0] == 2:
print(f"{site} 站点 {week} 周 搜索词 已完成导入到 search_term 执行下一步")
state1_sql = f"select status_val from workflow_progress where site_name='{site}' and date_info='2024-{week}' and date_type='week' and page='反查搜索词';"
print(state1_sql)
cursor_us.execute(state1_sql)
sate1_list = cursor_us.fetchone()
print('查看状态：', sate1_list)
if sate1_list[0] in (1, 2):
update_workflow_progress_2 = f"update workflow_progress set status_val=2,status='抓取中' where page='反查搜索词' and date_info='2024-{week}' and site_name='{site}' and date_type='week' and status_val=1"
print('修改状态2：：', update_workflow_progress_2)
cursor_us.execute(update_workflow_progress_2)
connect_us.commit()
return True
else:
print(33333333333336666666666666666666666666)
return False
except:
return False
if num == 1:
# 搜索词多进程已经抓完。最后执行单进程抓取。
sql_read = f'SELECT id FROM {site}_search_term_syn where state in (1,2) and week={week} LIMIT 1'
print(sql_read)
df = pd.read_sql(sql_read, con=engine_pg)
id_tuple = list(df.id)
if len(id_tuple) == 0:
id_tuple = None
print('id_tuple：', id_tuple)
if id_tuple is None:
status_sql = f"select status_val from workflow_progress where site_name='{site}' and date_info='2024-{week}' and date_type='week' and page='反查搜索词';"
print('status_sql:', status_sql)
cursor_us.execute(status_sql)
status_list = cursor_us.fetchone()
print(status_list)
if status_list:
if status_list[0] == 2:
update_workflow_progress = f"update workflow_progress set status_val=3,status='抓取结束' where page='反查搜索词' and date_info='2024-{week}' and site_name='{site}' and date_type='week'"
print('update_workflow_progress: 修改状态3 ', update_workflow_progress)
cursor_us.execute(update_workflow_progress)
connect_us.commit()
account = 'pengyanbing,chenyuanjie,hezhe,wangrui4,fangxingjun'
title = site + '站点 搜索词'
content = str(week) + ' 周 搜索词 已结束,请确认下一步流程!时间:'
db_class.send_mg(account, title, content)
DolphinschedulerHelper.start_process_instance_common(
project_name="big_data_selection",
process_df_name='ALL站点-asin同步-pg-api',
startParams={
"site_name": f"{site}",
"date_type": "week",
"date_info": f'2024-{week}'
}
)
if site == 'us':
pass
return True
else:
print('5555555555555555555555555555555555')
return False
if num == 3:
# 搜索词多进程已经抓完。最后执行单进程抓取。
sql_read = f'SELECT id FROM {site}_search_term_syn where state in (1,2) and week={week} LIMIT 1'
print(sql_read)
df = pd.read_sql(sql_read, con=engine_pg)
id_tuple = list(df.id)
if id_tuple:
id_tuple = [1]
else:
id_tuple = None
if id_tuple is None:
select_sate_mysql(site, num=1, week=week)
return False
else:
return True
def get_ip_address():
# 返回内网ip
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(('baidu.com', 0))
ip = s.getsockname()[0]
return ip
def long_time_task(site, proxy_name, week):
print("当前 抓取 站点 ", site)
if site == 'us':
spider_us = search_temp_pg(site_name=site, read_size=180, proxy_name=proxy_name, week=week)
spider_us.run_pol()
else:
spider_mysql = search_temp_mysql(site_name=site, read_size=180, proxy_name=proxy_name,
week=week)
spider_mysql.run_pol()
print('结束')
def start_run(week):
pppoe_ip()
proxy_name = None
site_list = ['us','de', 'uk']
for site in site_list:
if select_sate_mysql(site, week=week):
while True:
current_time = datetime.now()
# 计算后五分钟的时间
five_minutes_later = current_time + timedelta(minutes=1)
print('后五分钟时间', five_minutes_later)
p = Pool(5)
for i in range(3):
p.apply_async(long_time_task, args=(site, proxy_name, week))
print('等待所有子进程运行完成')
# 执行该方法后不能继续添加新的Process
p.close()
# 等待所有子进程执行完毕
p.join()
print('所有进程运行完毕！')
print('所有进程运行完毕！开始切换ip')
print('所有进程运行完毕！开始切换ip')
print('所有进程运行完毕！开始切换ip')
print('所有进程运行完毕！开始切换ip')
current_time = datetime.now()
if current_time > five_minutes_later:
pppoe_ip()
else:
time.sleep(random.uniform(200, 300))
pppoe_ip()
if select_sate_mysql(site, num=3, week=week) == False:
break
if __name__ == '__main__':
try:
week = int(sys.argv[1])
except:
week = None
start_run(week)
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