super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: blueprints/__pycache__/__init__.cpython-39.pyc

Browse Source 
modified:   blueprints/__pycache__/admin.cpython-39.pyc
	modified:   blueprints/__pycache__/auth.cpython-39.pyc
	modified:   blueprints/__pycache__/brand.cpython-311.pyc
	modified:   blueprints/__pycache__/brand.cpython-39.pyc
	modified:   blueprints/brand.py
	new file:   brand_spider/__pycache__/main.cpython-311.pyc
	modified:   brand_spider/__pycache__/main.cpython-39.pyc
	new file:   brand_spider/__pycache__/web_dec.cpython-311.pyc
	modified:   brand_spider/main.py
	modified:   config.py
	modified:   web_source/admin.html
	modified:   web_source/brand.html
	modified:   web_source/home.html
	modified:   web_source/index.html
	modified:   web_source/login.html
	new file:   web_source/templates_backup/admin.html
	new file:   web_source/templates_backup/brand.html
	new file:   web_source/templates_backup/home.html
	new file:   web_source/templates_backup/index.html
	new file:   web_source/templates_backup/login.html
This commit is contained in:
铭坤
2026-03-26 16:39:39 +08:00
parent 8cfaab49fb
commit bbca159684
11 changed files with 430 additions and 685 deletions
Download Patch File Download Diff File Expand all files Collapse all files

284
app/brand_spider/main.py
View File

@@ -3,10 +3,7 @@ import re
import random
import requests
import time
import datetime
import openpyxl
import unicodedata
from openpyxl import Workbook
from brand_spider.web_dec import decrypt_via_service, get_guid
@@ -77,7 +74,7 @@ def search(hashsearch, data, proxies=None):
# print("原始结果", enc_text)
# 若返回 Forbidden，则从 config 的 proxy_url 获取代理 IP 后重试，并开启 3 分钟代理窗口
if enc_text.strip() == '{"message":"Forbidden"}' and CONFIG_PROXY_URL:
if (enc_text.strip() == '{"message":"Forbidden"}' or "Too Many Requests" in enc_text) and CONFIG_PROXY_URL:
try:
proxy_resp = requests.get(CONFIG_PROXY_URL, timeout=10)
print("代理请求结果->：", proxy_resp.text)
@@ -125,7 +122,7 @@ class TaskCancelledError(Exception):
pass
def single_file_handle(file_path, output_path, strategy="Terms", check_cancelled=None, progress_callback=None):
def single_file_handle(brand, strategy="Terms"):
status_info = {
"Ended": "已结束",
"Expired": "已过期的",
@@ -471,208 +468,123 @@ def single_file_handle(file_path, output_path, strategy="Terms", check_cancelled
"欧洲联盟"
]
check_staus = ["已过期的","已结束"]
# 1. 读取 Excel 活动工作表
wb = openpyxl.load_workbook(file_path, data_only=True)
active_sheet = wb.active
active_sheet_name = active_sheet.title
print(active_sheet_name)
# 2. 将数据转换为列表嵌套字典
rows = list(active_sheet.iter_rows(values_only=True))
if rows:
columns = list(rows[0])
data_rows = []
for row in rows[1:]:
row_dict = {}
for idx, col in enumerate(columns):
row_dict[col] = row[idx] if idx < len(row) else None
data_rows.append(row_dict)
else:
columns = []
data_rows = []
# 3. 确保必要列存在
# required_cols = ["状态", "国家", "时间"]
# for col in required_cols:
# if col not in columns:
# columns.append(col)
# for row_dict in data_rows:
# row_dict[col] = ""
# 4. 提取唯一品牌列表
brand_ls = set()
for row in data_rows:
brand = row.get("品牌")
if brand:
brand_ls.add(brand)
# 5. 初始化不符合品牌的数据列表
faild_data = []
# 初始化查询失败的品牌的数据列表
query_faild_data = []
# 6. 对每个品牌进行处理
brands = list(brand_ls)
total_brands = len(brands)
for idx, brand in enumerate(brands, start=1):
if check_cancelled and check_cancelled():
raise TaskCancelledError()
# 行级进度回调：当前第 idx 条 / total_brands
if progress_callback and total_brands > 0:
try:
progress_callback(idx, total_brands, extra={'current_brand': str(brand)})
except Exception:
# 回调失败不影响主流程
pass
# Simple 策略下最多翻 3 页（0,30,60），直到命中 554 行判断
max_pages = 3 if strategy == "Simple" else 1
page = 0
matched = False
# Simple 策略下最多翻 3 页（0,30,60），直到命中 554 行判断
max_pages = 3 if strategy == "Simple" else 1
page = 0
matched = False
while page < max_pages and not matched:
try:
as_structure_dict = {
"_id": create_code_generator(),
"boolean": "AND",
"bricks": [
{
"_id": create_code_generator(),
"key": "brandName",
"value": brand,
"strategy": strategy
}
]
}
# 构造 asStructure 内部 JSON 对象
as_structure_dict = as_structure_dict
# print(as_structure_dict)
while page < max_pages and not matched:
try:
as_structure_dict = {
"_id": create_code_generator(),
"boolean": "AND",
"bricks": [
{
"_id": create_code_generator(),
"key": "brandName",
"value": brand,
"strategy": strategy
}
]
}
# 构造 asStructure 内部 JSON 对象
as_structure_dict = as_structure_dict
# print(as_structure_dict)
# 将内部对象转为 JSON 字符串（作为 asStructure 字段的值）
as_structure_str = json.dumps(as_structure_dict, ensure_ascii=False)
data = {
"sort": "score desc",
"rows": "30",
"asStructure": as_structure_str,
"fg": "_void_"
}
# 将内部对象转为 JSON 字符串（作为 asStructure 字段的值）
as_structure_str = json.dumps(as_structure_dict, ensure_ascii=False)
data = {
"sort": "score desc",
"rows": "30",
"asStructure": as_structure_str,
"fg": "_void_"
}
# 翻页：第二页 start=30，第三页 start=60（仅 Simple 策略）
if strategy == "Simple" and page > 0:
data["start"] = str(30 * page)
# 翻页：第二页 start=30，第三页 start=60（仅 Simple 策略）
if strategy == "Simple" and page > 0:
data["start"] = str(30 * page)
# print("请求参数", as_structure_dict, "页码:", page)
hashsearch = get_guid()
enc_text = search(hashsearch, data)
print(f"【{hashsearch}】待解密-->", enc_text)
dec_text = decrypt_via_service(hashsearch, enc_text)
# print(type(dec_text))
dec_text = str(dec_text)
# print(f"解密之后的结果-->", dec_text)
data = json.loads(dec_text)
except Exception as e:
# 仅第一页请求失败时记录为查询失败品牌
if page == 0:
import traceback
traceback.print_exc()
safe_brand = brand.encode('gbk', errors='replace').decode('gbk')
print(f"品牌：{safe_brand},处理失败:{e}")
query_faild_data.append({"品牌": brand, "时间": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
break
# print("请求参数", as_structure_dict, "页码:", page)
hashsearch = get_guid()
enc_text = search(hashsearch, data)
print(f"【{hashsearch}】待解密-->", enc_text)
dec_text = decrypt_via_service(hashsearch, enc_text)
# print(type(dec_text))
dec_text = str(dec_text)
# print(f"解密之后的结果-->", dec_text)
data = json.loads(dec_text)
except Exception as e:
# 仅第一页请求失败时记录为查询失败品牌
if page == 0:
import traceback
traceback.print_exc()
safe_brand = brand.encode('gbk', errors='replace').decode('gbk')
print(f"品牌：{safe_brand},处理失败:{e}")
# query_faild_data.append({"brand": brand, "time": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
query_faild_data.append(brand)
break
# print(data)
# print(data)
if data.get("response", {}).get("numFound", 0) > 0:
if strategy == "Simple":
new_brand_name = clean_text(brand)
_brand = new_brand_name.lower()
else:
_brand = brand
docs = data["response"]["docs"]
for d in docs:
office = d.get("office")
status = d.get("status")
registrationDate = d.get("registrationDate")
status_name = status_info.get(status, status)
brand_name = d.get("brandName")
if isinstance(brand_name, list):
brand_name = "|".join(brand_name)
if data.get("response", {}).get("numFound", 0) > 0:
if strategy == "Simple":
new_brand_name = clean_text(brand)
_brand = new_brand_name.lower()
else:
_brand = brand
new_brand_name = clean_text(brand_name)
new_brand_name = new_brand_name.lower()
if _brand != new_brand_name:
continue
docs = data["response"]["docs"]
for d in docs:
office = d.get("office")
status = d.get("status")
registrationDate = d.get("registrationDate")
status_name = status_info.get(status, status)
brand_name = d.get("brandName")
if isinstance(brand_name, list):
brand_name = "|".join(brand_name)
check_office = d.get("designation")
if strategy == "Simple":
new_brand_name = clean_text(brand_name)
new_brand_name = new_brand_name.lower()
if _brand != new_brand_name:
check_office.append(office)
for i in set(check_office):
office_name = country_info.get(i, i)
# print(office_name,office_name in check_value and status_name not in check_staus)
if office_name in check_value and status_name not in check_staus:
# 命中 554 行判断，记录并结束当前品牌后续翻页
# data_rows = [row for row in data_rows if row.get("品牌") != brand]
# print(office_name, office in ["EM","QZ"] and i not in ["EM","QZ"])
if office in ["EM","QZ"] and i not in ["EM","QZ"]:
continue
check_office = d.get("designation")
if strategy == "Simple":
faild_data.append({"brand": brand, "country": office_name, "status": status_name}) # "reason":"|".join(d.get("brandName","")
matched = True
else:
# print("添加",brand,office_name)
faild_data.append({"brand": brand, "country": office_name, "status": status_name})
# break
if matched:
break
check_office.append(office)
for i in set(check_office):
office_name = country_info.get(i, i)
# print(office_name,office_name in check_value and status_name not in check_staus)
if office_name in check_value and status_name not in check_staus:
# 命中 554 行判断，记录并结束当前品牌后续翻页
data_rows = [row for row in data_rows if row.get("品牌") != brand]
# print(office_name, office in ["EM","QZ"] and i not in ["EM","QZ"])
if office in ["EM","QZ"] and i not in ["EM","QZ"]:
continue
# 如果 Simple 策略下本页未命中，则翻下一页；其他策略不翻页
if not matched:
page += 1
if strategy == "Simple":
faild_data.append({"品牌": brand, "国家": office_name, "状态": status_name, "原因":"|".join(d.get("brandName","")),"时间": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
matched = True
else:
# print("添加",brand,office_name)
faild_data.append({"品牌": brand, "国家": office_name, "状态": status_name, "时间": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
# break
if matched:
break
# 如果 Simple 策略下本页未命中，则翻下一页；其他策略不翻页
if not matched:
page += 1
# else:
# # 更新主表数据
# for row in data_rows:
# if row.get("品牌") == brand:
# row["状态"] = status_name
# row["国家"] = office_name
# row["时间"] = datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S") # 品牌匹配行
# else:
# # 非匹配行：时间 = 当前国家值（模拟原 mask 逻辑）
# row["时间"] = ""
# 7. 创建输出工作簿
out_wb = Workbook()
out_wb.remove(out_wb.active) # 删除默认 sheet
# 主工作表
main_ws = out_wb.create_sheet(title=active_sheet_name)
main_ws.append(columns)
for row_dict in data_rows:
main_ws.append([row_dict.get(col, "") for col in columns])
# 不符合品牌工作表
faild_ws = out_wb.create_sheet(title="不符合品牌")
if strategy == "Simple":
faild_columns = ["品牌", "国家", "状态","原因"]
else:
faild_columns = ["品牌", "国家", "状态"]
faild_ws.append(faild_columns)
for row_dict in faild_data:
# print("写入",row_dict)
faild_ws.append([row_dict.get(col, "") for col in faild_columns])
# 查询失败品牌工作表
query_faild_ws = out_wb.create_sheet(title="查询失败品牌")
query_faild_columns = ["品牌", "时间"]
query_faild_ws.append(query_faild_columns)
for row_dict in query_faild_data:
query_faild_ws.append([row_dict.get(col, "") for col in query_faild_columns])
out_wb.save(output_path)
print("生成完成--->", output_path)
return output_path
return faild_data,query_faild_data
if __name__ == '__main__':