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

new file: app/blueprints/__init__.py

Browse Source 
new file:   app/blueprints/__pycache__/__init__.cpython-311.pyc
	new file:   app/blueprints/__pycache__/__init__.cpython-39.pyc
	new file:   app/blueprints/__pycache__/admin.cpython-311.pyc
	new file:   app/blueprints/__pycache__/admin.cpython-39.pyc
	new file:   app/blueprints/__pycache__/auth.cpython-311.pyc
	new file:   app/blueprints/__pycache__/auth.cpython-39.pyc
	new file:   app/blueprints/__pycache__/brand.cpython-311.pyc
	new file:   app/blueprints/__pycache__/brand.cpython-39.pyc
	new file:   app/blueprints/__pycache__/image.cpython-311.pyc
	new file:   app/blueprints/__pycache__/image.cpython-39.pyc
	new file:   app/blueprints/__pycache__/main.cpython-311.pyc
	new file:   app/blueprints/__pycache__/main.cpython-39.pyc
	new file:   app/blueprints/admin.py
	new file:   app/blueprints/auth.py
	new file:   app/blueprints/brand.py
	new file:   app/blueprints/image.py
	new file:   app/blueprints/main.py
	new file:   app/brand_spider/__pycache__/main.cpython-39.pyc
	new file:   app/brand_spider/__pycache__/web_dec.cpython-39.pyc
	new file:   app/brand_spider/main.py
	new file:   app/brand_spider/web_dec.py
	new file:   app/static/bg.jpg
	new file:   "app/static/\345\223\201\347\211\214\346\226\207\346\241\243\346\240\274\345\274\217_\346\250\241\346\235\277.xlsx"
	new file:   "app/static/\346\250\241\346\235\2772-\344\273\245\346\226\207\344\273\266\345\244\271\346\226\271\345\274\217\344\270\212\344\274\240.zip"
	new file:   app/tool/.device_id
	new file:   app/tool/__pycache__/devices.cpython-311.pyc
	new file:   app/tool/__pycache__/devices.cpython-39.pyc
	new file:   app/tool/devices.py
This commit is contained in:
铭坤
2026-03-20 23:04:19 +08:00
parent 8a64a6cd9b
commit bb06bc8f5a
29 changed files with 2899 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/blueprints/__init__.py Normal file
View File

@@ -0,0 +1 @@
# 蓝图包

BIN
app/blueprints/__pycache__/__init__.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/__init__.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/admin.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/admin.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/auth.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/auth.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/brand.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/brand.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/image.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/image.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/main.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/blueprints/__pycache__/main.cpython-39.pyc Normal file
View File

Binary file not shown.

362
app/blueprints/admin.py Normal file
View File

@@ -0,0 +1,362 @@
"""
管理员蓝图：用户管理（列表/创建/更新/删除）、生成历史、管理页
"""
import json
import pymysql
from flask import Blueprint, request, jsonify
from werkzeug.security import generate_password_hash
from app_common import get_db, _render_html, _get_current_admin_role, admin_required, login_required
from flask import session
admin_bp = Blueprint('admin', __name__)
def _parse_json(val, default=None):
if val is None:
return default if default is not None else []
if isinstance(val, (list, dict)):
return val
try:
return json.loads(val)
except Exception:
return default if default is not None else []
@admin_bp.route('/admin')
@login_required
@admin_required
def admin_page():
return _render_html('admin.html')
@admin_bp.route('/api/admin/users')
@admin_required
def admin_list_users():
"""分页获取用户列表；支持用户名模糊搜索、指定管理员所属普通用户筛选"""
role, current_row = _get_current_admin_role()
if not role:
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
page = max(1, int(request.args.get('page', 1)))
page_size = min(50, max(5, int(request.args.get('page_size', 15))))
offset = (page - 1) * page_size
search_username = (request.args.get('username') or request.args.get('search') or '').strip()
created_by_id_arg = request.args.get('created_by_id') or request.args.get('admin_id')
created_by_id = int(created_by_id_arg) if created_by_id_arg and str(created_by_id_arg).isdigit() else None
if role != 'super_admin':
created_by_id = None
try:
conn = get_db()
with conn.cursor() as cur:
if role == 'super_admin':
where_parts = ["1=1"]
params = []
if search_username:
where_parts.append("u.username LIKE %s")
params.append("%" + search_username + "%")
if created_by_id is not None:
where_parts.append("u.created_by_id = %s")
params.append(created_by_id)
where_sql = " AND ".join(where_parts)
cur.execute(
"""SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
creator.username AS creator_username
FROM users u
LEFT JOIN users creator ON creator.id = u.created_by_id
WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
tuple(params) + (page_size, offset),
)
rows = cur.fetchall()
cur.execute("SELECT COUNT(*) as total FROM users u WHERE " + where_sql, tuple(params))
total = cur.fetchone()['total']
cur.execute("SELECT id, username FROM users WHERE role = 'admin' ORDER BY id")
admins = [{'id': r['id'], 'username': r['username']} for r in cur.fetchall()]
else:
admin_id = current_row['id']
where_parts = ["(u.id = %s OR (u.role = 'normal' AND u.created_by_id = %s))"]
params = [admin_id, admin_id]
if search_username:
where_parts.append("u.username LIKE %s")
params.append("%" + search_username + "%")
where_sql = " AND ".join(where_parts)
cur.execute(
"""SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
creator.username AS creator_username
FROM users u
LEFT JOIN users creator ON creator.id = u.created_by_id
WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
tuple(params) + (page_size, offset),
)
rows = cur.fetchall()
cur.execute(
"SELECT COUNT(*) as total FROM users u WHERE " + where_sql,
tuple(params),
)
total = cur.fetchone()['total']
admins = []
items = [
{
'id': r['id'],
'username': r['username'],
'is_admin': bool(r.get('is_admin')),
'role': r.get('role') or 'normal',
'created_by_id': r.get('created_by_id'),
'creator_username': r.get('creator_username') or '',
'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
}
for r in rows
]
conn.close()
payload = {
'success': True,
'items': items,
'total': total,
'page': page,
'page_size': page_size,
'current_user_role': role,
'admins': admins,
}
return jsonify(payload)
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@admin_bp.route('/api/admin/user', methods=['POST'])
@admin_required
def admin_create_user():
data = request.get_json() or {}
username = (data.get('username') or '').strip()
password = data.get('password') or ''
role, current_row = _get_current_admin_role()
if not role:
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
want_role = (data.get('role') or 'normal').strip() or 'normal'
if want_role not in ('admin', 'normal'):
want_role = 'normal'
if role == 'admin' and want_role == 'admin':
return jsonify({'success': False, 'error': '仅超级管理员可创建管理员'})
if want_role == 'admin':
want_created_by = current_row['id']
elif role == 'super_admin':
want_created_by = data.get('created_by_id')
else:
want_created_by = current_row['id']
if not username or not password:
return jsonify({'success': False, 'error': '用户名和密码不能为空'})
if len(username) < 2:
return jsonify({'success': False, 'error': '用户名至少2个字符'})
if len(password) < 6:
return jsonify({'success': False, 'error': '密码至少6个字符'})
if want_role == 'normal' and role == 'super_admin' and want_created_by is None:
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT id FROM users WHERE role = 'admin' ORDER BY id LIMIT 1")
r = cur.fetchone()
conn.close()
want_created_by = r['id'] if r else current_row['id']
except Exception:
want_created_by = current_row['id']
if want_role == 'normal' and want_created_by is None:
want_created_by = current_row['id']
is_admin = 1 if want_role in ('super_admin', 'admin') else 0
pwd_hash = generate_password_hash(password, method='pbkdf2:sha256')
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"INSERT INTO users (username, password_hash, is_admin, role, created_by_id) VALUES (%s, %s, %s, %s, %s)",
(username, pwd_hash, is_admin, want_role, want_created_by),
)
conn.commit()
conn.close()
return jsonify({'success': True, 'msg': '用户创建成功'})
except pymysql.IntegrityError:
return jsonify({'success': False, 'error': '用户名已存在'})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@admin_bp.route('/api/admin/user/<int:uid>', methods=['PUT'])
@admin_required
def admin_update_user(uid):
data = request.get_json() or {}
password = data.get('password')
want_role = (data.get('role') or '').strip() or data.get('role')
role, current_row = _get_current_admin_role()
if not role:
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
if want_role is None and not password:
return jsonify({'success': False, 'error': '请提供要修改的内容'})
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT id, role, created_by_id FROM users WHERE id = %s", (uid,))
target = cur.fetchone()
if not target:
conn.close()
return jsonify({'success': False, 'error': '用户不存在'})
if role == 'admin':
if target['role'] != 'normal' or target.get('created_by_id') != current_row['id']:
conn.close()
return jsonify({'success': False, 'error': '只能编辑自己创建的普通用户'}), 403
want_role = None
else:
if target.get('role') == 'super_admin':
conn.close()
return jsonify({'success': False, 'error': '不能修改超级管理员'})
if want_role == 'super_admin':
return jsonify({'success': False, 'error': '不能将用户设为超级管理员'})
if want_role not in ('admin', 'normal', None, ''):
want_role = None
if password:
if len(password) < 6:
conn.close()
return jsonify({'success': False, 'error': '密码至少6个字符'})
pwd_hash = generate_password_hash(password, method='pbkdf2:sha256')
cur.execute("UPDATE users SET password_hash = %s WHERE id = %s", (pwd_hash, uid))
if want_role is not None and want_role != '':
is_admin = 1 if want_role == 'admin' else 0
cur.execute(
"UPDATE users SET is_admin = %s, role = %s WHERE id = %s",
(is_admin, want_role, uid),
)
conn.commit()
conn.close()
return jsonify({'success': True, 'msg': '更新成功'})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@admin_bp.route('/api/admin/user/<int:uid>', methods=['DELETE'])
@admin_required
def admin_delete_user(uid):
from flask import session
if session.get('user_id') == uid:
return jsonify({'success': False, 'error': '不能删除当前登录账号'})
role, current_row = _get_current_admin_role()
if not role:
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT id, role, created_by_id FROM users WHERE id = %s", (uid,))
target = cur.fetchone()
if not target:
conn.close()
return jsonify({'success': False, 'error': '用户不存在'})
if target.get('role') == 'super_admin':
conn.close()
return jsonify({'success': False, 'error': '不能删除超级管理员'})
if role == 'admin':
if target.get('role') != 'normal' or target.get('created_by_id') != current_row['id']:
conn.close()
return jsonify({'success': False, 'error': '只能删除自己创建的普通用户'}), 403
cur.execute("DELETE FROM users WHERE id = %s", (uid,))
affected = cur.rowcount
conn.commit()
conn.close()
if affected == 0:
return jsonify({'success': False, 'error': '用户不存在'})
return jsonify({'success': True, 'msg': '删除成功'})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@admin_bp.route('/api/admin/user/<int:uid>/column-permissions')
@login_required
def admin_user_column_permissions(uid):
"""获取指定用户的栏目权限列表：当前用户只能查自己，管理员可查任意用户。超级管理员返回全部栏目。"""
current_uid = session.get('user_id')
if current_uid != uid:
role, _ = _get_current_admin_role()
if not role:
return jsonify({'success': False, 'error': '无权查看该用户的栏目权限'}), 403
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT role FROM users WHERE id = %s", (uid,))
user_row = cur.fetchone()
if user_row and (user_row.get('role') or '').strip() == 'super_admin':
cur.execute("""
SELECT id, name, column_key, created_at FROM columns ORDER BY id
""")
rows = cur.fetchall()
else:
cur.execute("""
SELECT c.id, c.name, c.column_key, c.created_at
FROM columns c
INNER JOIN user_column_permission ucp ON ucp.column_id = c.id
WHERE ucp.user_id = %s
ORDER BY c.id
""", (uid,))
rows = cur.fetchall()
conn.close()
items = [
{
'id': r['id'],
'name': r['name'],
'column_key': r['column_key'],
'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
}
for r in rows
]
return jsonify({'success': True, 'items': items})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@admin_bp.route('/api/admin/history')
@admin_required
def admin_history():
"""管理员分页获取所有生成记录，支持按用户和时间筛选"""
page = max(1, int(request.args.get('page', 1)))
page_size = min(50, max(10, int(request.args.get('page_size', 15))))
offset = (page - 1) * page_size
user_id = request.args.get('user_id', type=int)
time_start = (request.args.get('time_start') or '').strip()
time_end = (request.args.get('time_end') or '').strip()
conditions, params = [], []
if user_id:
conditions.append("h.user_id = %s")
params.append(user_id)
if time_start:
conditions.append("h.created_at >= %s")
params.append(time_start)
if time_end:
conditions.append("h.created_at <= %s")
params.append(time_end + ' 23:59:59' if len(time_end) <= 10 else time_end)
where_clause = " AND ".join(conditions) if conditions else "1=1"
params_count = params[:]
params.extend([page_size, offset])
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"""SELECT h.id, h.user_id, h.created_at, h.panel_type, h.original_urls, h.params, h.result_urls,
h.long_image_url, u.username
FROM image_history h
LEFT JOIN users u ON h.user_id = u.id
WHERE """ + where_clause + """ ORDER BY h.created_at DESC LIMIT %s OFFSET %s""",
params,
)
rows = cur.fetchall()
cur.execute("SELECT COUNT(*) as total FROM image_history h WHERE " + where_clause, params_count)
total = cur.fetchone()['total']
conn.close()
items = []
for r in rows:
items.append({
'id': r['id'],
'user_id': r['user_id'],
'username': r.get('username') or '-',
'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r['created_at'] else '',
'panel_type': r['panel_type'] or '',
'original_urls': _parse_json(r['original_urls'], []),
'params': _parse_json(r['params'], {}),
'result_urls': _parse_json(r['result_urls'], []),
'long_image_url': (r.get('long_image_url') or '').strip() or None,
})
return jsonify({'success': True, 'items': items, 'total': total, 'page': page, 'page_size': page_size})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})

107
app/blueprints/auth.py Normal file
View File

@@ -0,0 +1,107 @@
"""
认证蓝图：登录、登出、登录状态校验
"""
from flask import Blueprint, request, redirect, url_for, session, jsonify
from werkzeug.security import check_password_hash
from app_common import (
get_db,
_render_html,
_is_session_user_valid,
login_required,
BASE_DIR,
)
from tool.devices import DeviceIDGenerator
auth_bp = Blueprint('auth', __name__)
@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
if session.get('user_id') and _is_session_user_valid():
return redirect(url_for('main.home'))
if request.method == 'POST':
data = request.get_json() if request.is_json else request.form
username = (data.get('username') or '').strip()
password = data.get('password') or ''
if not username or not password:
if request.is_json:
return jsonify({'success': False, 'error': '请输入用户名和密码'})
return _render_html('login.html', error='请输入用户名和密码')
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"SELECT id, password_hash, machine, is_admin FROM users WHERE username = %s",
(username,)
)
row = cur.fetchone()
if row and check_password_hash(row['password_hash'], password):
current_machine = DeviceIDGenerator().get_device_id()
stored_machine = (row.get('machine') or '').strip()
if not stored_machine:
with conn.cursor() as cur:
cur.execute("UPDATE users SET machine = %s WHERE id = %s", (current_machine, row['id']))
conn.commit()
conn.close()
session.permanent = True
session['user_id'] = row['id']
session['username'] = username
if request.is_json:
return jsonify({'success': True, 'redirect': url_for('main.home')})
return redirect(url_for('main.home'))
print("验证设备",stored_machine)
print("当前设备",current_machine)
if stored_machine != current_machine and row.get("is_admin") != 1:
conn.close()
err_msg = '当前设备与首次登录设备不一致，请在原设备上登录'
if request.is_json:
return jsonify({'success': False, 'error': err_msg})
return _render_html('login.html', error=err_msg)
conn.close()
session.permanent = True
session['user_id'] = row['id']
session['username'] = username
if request.is_json:
return jsonify({'success': True, 'redirect': url_for('main.home')})
return redirect(url_for('main.home'))
conn.close()
except Exception as e:
if request.is_json:
return jsonify({'success': False, 'error': str(e)})
return _render_html('login.html', error='登录失败，请稍后重试')
if request.is_json:
return jsonify({'success': False, 'error': '用户名或密码错误'})
return _render_html('login.html', error='用户名或密码错误')
return _render_html('login.html')
@auth_bp.route('/api/auth/check')
@login_required
def api_auth_check():
"""校验登录状态，用于页面加载时判断是否已登录；同时校验机器码是否与首次登录设备一致"""
if not session.get('user_id'):
return jsonify({'logged_in': False})
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT machine, is_admin FROM users WHERE id = %s", (session['user_id'],))
row = cur.fetchone()
conn.close()
if not row:
return jsonify({'logged_in': False})
stored_machine = (row.get('machine') or '').strip()
if stored_machine:
current_machine = DeviceIDGenerator().get_device_id()
if stored_machine != current_machine and row.get("is_admin") != 1:
session.clear()
return jsonify({'logged_in': False, 'error': '当前设备与首次登录设备不一致'})
except Exception:
return jsonify({'logged_in': False})
return jsonify({'logged_in': True, 'redirect': url_for('main.home')})
@auth_bp.route('/logout')
def logout():
session.clear()
return redirect(url_for('auth.login'))

855
app/blueprints/brand.py Normal file
View File

@@ -0,0 +1,855 @@
"""
品牌爬虫蓝图：展开文件夹、运行任务、任务列表/详情、下载结果
"""
import os
import json
import threading
import datetime
import traceback
import zipfile
import io
import time
import tempfile
import requests
from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
from flask import Blueprint, request, jsonify, session, send_file, redirect, Response
from app_common import get_db, login_required, BASE_DIR
from config import bucket_path
from brand_spider.main import single_file_handle, TaskCancelledError
brand_bp = Blueprint('brand', __name__)
BRAND_OUTPUT_DIR = os.path.join(BASE_DIR, 'brand_output')
OSS_PREFIX = "brand_results"
# 任务完成时推送给前端的 SSE 队列：task_id -> [Queue, ...]
_task_event_queues = {}
_task_event_lock = threading.Lock()
# 任务行级进度（当前处理文件的行数/总行数），仅存内存，不落库：
# task_id -> {
# 'file_index': int, # 当前处理第几个文件
# 'file_total': int, # 总文件数
# 'file_path': str, # 当前文件路径
# 'current_line': int, # 当前行号（或当前品牌序号）
# 'total_lines': int, # 总行数（或总品牌数）
# }
_task_line_progress = {}
_task_line_progress_lock = threading.Lock()
def _push_task_event(task_id, event):
"""向订阅了该任务的所有 SSE 连接推送事件，并移除该任务的队列列表。
同时清理内存中的行级进度，不通过数据库中转行级进度。"""
with _task_event_lock:
queues = _task_event_queues.pop(task_id, [])
for q in queues:
try:
q.put_nowait(event)
except Exception:
pass
# 任务进入终态时，顺便清理行级进度缓存
with _task_line_progress_lock:
_task_line_progress.pop(task_id, None)
def _expand_folder_xlsx(folder_path):
"""返回文件夹下所有 .xlsx 文件的绝对路径列表"""
if not folder_path or not os.path.isdir(folder_path):
return []
paths = []
for name in os.listdir(folder_path):
if name.endswith('.xlsx') or name.endswith('.XLSX'):
paths.append(os.path.normpath(os.path.join(folder_path, name)))
return paths
def _upload_local_xlsx_to_oss(local_path, user_id):
"""将本地 xlsx 文件上传到 OSS，返回 (url, None) 或 (None, error_message)。"""
if not local_path or not os.path.isfile(local_path):
return None, "文件不存在"
try:
from ali_oss import upload_file as oss_upload_file
except ImportError:
return None, "OSS 模块未配置"
base = os.path.basename(local_path)
safe_base = "".join(c if c.isalnum() or c in '-_.' else '_' for c in base)
if not safe_base.endswith('.xlsx'):
safe_base = safe_base + '.xlsx'
key = f"{bucket_path}brand_input/{int(time.time() * 1000)}/{user_id}_{safe_base}"
try:
with open(local_path, "rb") as f:
content = f.read()
url = oss_upload_file(content, key)
return url, None
except Exception as e:
return None, str(e)
def _run_brand_files(file_paths, task_id=None, check_cancelled=None, strategy="Terms"):
"""对多个 xlsx 执行 single_file_handle（线程池多线程），返回 (result_paths, error_message)。
check_cancelled: 可调用对象，返回 True 时中止并返回 (result_paths, 'cancelled')。
task_id 存在时会在每处理完一个文件后更新 progress_current。
strategy: 品牌匹配方式，默认 Terms，可选 Simple。"""
os.makedirs(BRAND_OUTPUT_DIR, exist_ok=True)
subdir = str(task_id) if task_id else ('immediate_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S'))
out_dir = os.path.join(BRAND_OUTPUT_DIR, subdir)
os.makedirs(out_dir, exist_ok=True)
# 先筛出有效路径并生成 (fp, out_path) 列表
tasks = []
for fp in file_paths:
if not fp or not isinstance(fp, str) or not os.path.isfile(fp):
continue
base = os.path.splitext(os.path.basename(fp))[0]
safe_base = "".join(c if c.isalnum() or c in '-_' else '_' for c in base)
out_path = os.path.join(out_dir, safe_base + '_result.xlsx')
tasks.append((fp, out_path))
if not tasks:
return [], None
result_paths = []
result_lock = threading.Lock()
current = [0] # 用列表以便在闭包中修改
def process_one(fp, out_path, file_index, files_total):
# 行级进度回调：由 single_file_handle 在循环中调用，直接写入内存字典，不落库
def progress_callback(current_line, total_lines, extra=None):
if not task_id:
return
info = {
'file_index': file_index,
'file_total': files_total,
'file_path': fp,
'current_line': int(current_line or 0),
'total_lines': int(total_lines or 0),
}
if isinstance(extra, dict):
info.update(extra)
with _task_line_progress_lock:
_task_line_progress[task_id] = info
single_file_handle(fp, out_path, strategy, check_cancelled=check_cancelled, progress_callback=progress_callback)
return out_path
max_workers = min(8, len(tasks))
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_out = {
executor.submit(process_one, fp, out_path, idx + 1, len(tasks)): out_path
for idx, (fp, out_path) in enumerate(tasks)
}
try:
for future in as_completed(future_to_out):
if check_cancelled and check_cancelled():
for f in future_to_out:
f.cancel()
return result_paths, 'cancelled'
try:
out_path = future.result()
with result_lock:
result_paths.append(out_path)
current[0] += 1
if task_id:
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET progress_current = %s WHERE id = %s",
(current[0], task_id)
)
conn.commit()
conn.close()
except Exception:
print(traceback.format_exc())
except TaskCancelledError:
for f in future_to_out:
f.cancel()
return result_paths, 'cancelled'
except Exception as e:
print(traceback.format_exc())
print(e)
for f in future_to_out:
f.cancel()
return result_paths, str(e)
except Exception as e:
print(traceback.format_exc())
return result_paths, str(e)
return result_paths, None
def _upload_results_to_oss(source_paths, local_result_paths, task_id):
"""
将源文件和结果文件上传到 OSS，并打包成 zip（分两个文件夹：源文件、结果文件）上传。
返回 (result_data, error_message)，result_data 为 {"urls": [url1, ...], "zip_url": "..."}，出错时为 (None, err)。
"""
if not local_result_paths or not task_id:
return None, "无结果文件"
try:
from ali_oss import upload_file as oss_upload_file
except ImportError:
return None, "OSS 模块未配置"
urls = []
subdir = f"{OSS_PREFIX}/{task_id}"
for fp in local_result_paths:
if not fp or not os.path.isfile(fp):
continue
name = os.path.basename(fp)
key = f"{bucket_path}{subdir}/{name}"
try:
with open(fp, "rb") as f:
content = f.read()
url = oss_upload_file(content, key)
urls.append(url)
except Exception as e:
return None, f"上传结果文件失败: {e}"
if not urls:
return None, "没有可上传的结果文件"
# 打包成 zip：源文件 -> 源文件/，结果文件 -> 结果文件/
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
for fp in (source_paths or []):
if fp and os.path.isfile(fp):
zf.write(fp, os.path.join("源文件", os.path.basename(fp)))
for fp in local_result_paths:
if fp and os.path.isfile(fp):
zf.write(fp, os.path.join("结果文件", os.path.basename(fp)))
buf.seek(0)
zip_key = f"{bucket_path}{subdir}/result.zip"
try:
zip_url = oss_upload_file(buf.getvalue(), zip_key)
except Exception as e:
return {"urls": urls, "zip_url": None}, None # 单文件链接已保存，zip 失败不报错
return {"urls": urls, "zip_url": zip_url}, None
def _background_brand_task(task_id):
"""后台执行单个品牌爬虫任务并更新数据库"""
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"SELECT id, user_id, file_paths, strategy, `desc` FROM brand_crawl_tasks WHERE id = %s AND status = 'pending'",
(task_id,)
)
row = cur.fetchone()
conn.close()
if not row:
return
file_paths = row.get('file_paths')
if isinstance(file_paths, str):
file_paths = json.loads(file_paths) if file_paths else []
# 优先使用表里存的 strategy 字段，兼容旧数据则从 desc 解析
strategy = (row.get('strategy') or '').strip() or None
if not strategy:
desc_val = (row.get('desc') or '').strip()
strategy = "Simple" if desc_val.startswith('[Simple]') else "Terms"
if strategy not in ('Terms', 'Simple'):
strategy = 'Terms'
if not file_paths:
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
('没有有效文件路径', task_id)
)
conn.commit()
conn.close()
except Exception:
pass
_push_task_event(task_id, {'status': 'failed', 'error_message': '没有有效文件路径'})
return
# 将 file_paths 解析为本地路径：URL 下载到 BRAND_OUTPUT_DIR 对应任务文件夹下（兼容旧数据中的本地路径）
os.makedirs(BRAND_OUTPUT_DIR, exist_ok=True)
task_subdir = str(task_id) if task_id else ('immediate_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S'))
task_dir = os.path.join(BRAND_OUTPUT_DIR, task_subdir)
os.makedirs(task_dir, exist_ok=True)
local_paths = []
downloaded_files_to_cleanup_on_fail = []
for idx, p in enumerate(file_paths):
if not p or not isinstance(p, str):
continue
p = p.strip()
if _is_url(p):
try:
parsed = urlparse(p)
url_name = os.path.basename(parsed.path or '')
if not url_name:
url_name = f"input_{idx + 1}.xlsx"
safe_name = "".join(c if c.isalnum() or c in '-_.' else '_' for c in url_name)
if not safe_name.lower().endswith('.xlsx'):
safe_name = safe_name + '.xlsx'
base_no_ext, ext = os.path.splitext(safe_name)
dest = os.path.join(task_dir, safe_name)
n = 1
while os.path.exists(dest):
dest = os.path.join(task_dir, f"{base_no_ext}_{n}{ext}")
n += 1
r = requests.get(p, timeout=120, stream=True)
r.raise_for_status()
with open(dest, "wb") as f:
for chunk in r.iter_content(chunk_size=1024 * 1024):
if chunk:
f.write(chunk)
local_paths.append(dest)
downloaded_files_to_cleanup_on_fail.append(dest)
except Exception as e:
for tf in downloaded_files_to_cleanup_on_fail:
try:
os.unlink(tf)
except Exception:
pass
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
(f'下载文件失败: {str(e)}', task_id)
)
conn.commit()
conn.close()
except Exception:
pass
_push_task_event(task_id, {'status': 'failed', 'error_message': f'下载文件失败: {str(e)}'})
return
elif os.path.isfile(p):
local_paths.append(p)
if not local_paths:
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
('没有有效的本地或可下载文件', task_id)
)
conn.commit()
conn.close()
except Exception:
pass
_push_task_event(task_id, {'status': 'failed', 'error_message': '没有有效的本地或可下载文件'})
return
try:
# 原子地将 pending 改为 running，避免用户先点取消时仍被覆盖为 running
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'running', progress_total = %s, progress_current = 0 WHERE id = %s AND status = 'pending'",
(len(local_paths), task_id)
)
n = cur.rowcount
conn.commit()
conn.close()
if n == 0:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT status FROM brand_crawl_tasks WHERE id = %s", (task_id,))
r = cur.fetchone()
conn.close()
if r and (r.get('status') or '') == 'cancelled':
_push_task_event(task_id, {'status': 'cancelled'})
return
except Exception:
pass
def check_cancelled():
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT status FROM brand_crawl_tasks WHERE id = %s", (task_id,))
row = cur.fetchone()
conn.close()
return row and (row.get('status') or '') == 'cancelled'
except Exception:
return False
result_paths, err = _run_brand_files(local_paths, task_id=task_id, check_cancelled=check_cancelled, strategy=strategy)
if err:
try:
conn = get_db()
with conn.cursor() as cur:
st = 'cancelled' if err == 'cancelled' else 'failed'
cur.execute(
"UPDATE brand_crawl_tasks SET status = %s, error_message = %s WHERE id = %s",
(st, err if st == 'failed' else None, task_id)
)
conn.commit()
conn.close()
except Exception as e:
traceback.print_exc()
print(e)
pass
_push_task_event(task_id, {'status': st, 'error_message': err if st == 'failed' else None})
return
result_data, upload_err = _upload_results_to_oss(local_paths, result_paths, task_id)
if upload_err:
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
(upload_err, task_id)
)
conn.commit()
conn.close()
except Exception:
pass
_push_task_event(task_id, {'status': 'failed', 'error_message': upload_err})
return
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'success', result_paths = %s WHERE id = %s",
(json.dumps(result_data), task_id)
)
conn.commit()
conn.close()
except Exception:
pass
_push_task_event(task_id, {'status': 'success'})
finally:
pass
except Exception as e:
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
(str(e), task_id)
)
conn.commit()
conn.close()
except Exception:
pass
_push_task_event(task_id, {'status': 'failed', 'error_message': str(e)})
def _parse_json(val, default=None):
if val is None:
return default if default is not None else []
if isinstance(val, (list, dict)):
return val
try:
return json.loads(val)
except Exception:
return default if default is not None else []
def _is_url(s):
if not isinstance(s, str) or not s.strip():
return False
return s.strip().startswith('http://') or s.strip().startswith('https://')
def _normalize_result_paths(raw):
"""
将 result_paths 规范化为 (url_list, zip_url)。
支持旧格式 list 或新格式 dict {"urls": [...], "zip_url": "..."}。
"""
if not raw:
return [], None
if isinstance(raw, dict):
urls = raw.get('urls') or []
if not isinstance(urls, list):
urls = []
zip_url = raw.get('zip_url')
if zip_url and not isinstance(zip_url, str):
zip_url = None
return urls, zip_url
if isinstance(raw, list):
return raw, None
return [], None
@brand_bp.route('/api/brand/expand-folder', methods=['POST'])
@login_required
def api_brand_expand_folder():
"""展开文件夹，返回其下所有 .xlsx 文件路径列表"""
try:
data = request.get_json() or {}
folder = (data.get('folder') or '').strip()
if not folder:
return jsonify({'success': False, 'error': '请提供 folder 路径'}), 400
paths = _expand_folder_xlsx(folder)
return jsonify({'success': True, 'paths': paths})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/run', methods=['POST'])
@login_required
def api_brand_run():
"""立即运行：创建任务并后台执行，立即返回 task_id，前端可轮询进度与取消"""
try:
data = request.get_json() or {}
paths = data.get('paths') or []
# task_type: 1=立即执行，2=添加任务；此接口默认 1
try:
task_type = int(data.get('task_type') or 1)
except Exception:
task_type = 1
if task_type not in (1, 2):
task_type = 1
strategy = (data.get('strategy') or 'Terms').strip()
if strategy not in ('Terms', 'Simple'):
strategy = 'Terms'
if not isinstance(paths, list):
paths = []
paths = [p.strip() for p in paths if p and isinstance(p, str) and os.path.isfile(p.strip())]
if not paths:
return jsonify({'success': False, 'error': '没有有效的 xlsx 文件路径'}), 400
# 先上传到 OSS，获取链接再入库
user_id = session['user_id']
urls = []
for p in paths:
url, err = _upload_local_xlsx_to_oss(p, user_id)
if err:
return jsonify({'success': False, 'error': f'上传文件失败: {os.path.basename(p)} - {err}'}), 500
urls.append(url)
desc_core = ', '.join(os.path.basename(p) for p in paths)[:500] if paths else ''
desc = f'[{strategy}] ' + (desc_core or '')
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"INSERT INTO brand_crawl_tasks (user_id, file_paths, `desc`, strategy, status, task_type) VALUES (%s, %s, %s, %s, 'pending', %s)",
(user_id, json.dumps(urls), desc or None, strategy, task_type)
)
task_id = cur.lastrowid
conn.commit()
conn.close()
threading.Thread(target=_background_brand_task, args=(task_id,), daemon=True).start()
return jsonify({'success': True, 'task_id': task_id})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/tasks', methods=['GET', 'POST'])
@login_required
def api_brand_tasks():
"""GET: 获取当前用户的任务列表；POST: 添加任务（后台执行）"""
if request.method == 'GET':
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"""SELECT id, file_paths, `desc`, strategy, status, result_paths, error_message,
COALESCE(progress_current, 0) AS progress_current,
COALESCE(progress_total, 0) AS progress_total,
created_at, updated_at
FROM brand_crawl_tasks WHERE user_id = %s ORDER BY id DESC LIMIT 100""",
(session['user_id'],)
)
rows = cur.fetchall()
conn.close()
items = []
for r in rows:
items.append({
'id': r['id'],
'file_paths': _parse_json(r['file_paths'], []),
'desc': (r.get('desc') or '').strip() or None,
'strategy': (r.get('strategy') or 'Terms').strip() or 'Terms',
'status': r.get('status') or 'pending',
'result_paths': _parse_json(r['result_paths']),
'error_message': (r.get('error_message') or '').strip() or None,
'progress_current': int(r.get('progress_current') or 0),
'progress_total': int(r.get('progress_total') or 0),
'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
'updated_at': r['updated_at'].strftime('%Y-%m-%d %H:%M') if r.get('updated_at') else '',
})
return jsonify({'success': True, 'items': items})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
# POST
try:
data = request.get_json() or {}
paths = data.get('paths') or []
# task_type: 1=立即执行，2=添加任务；此接口默认 2
try:
task_type = int(data.get('task_type') or 2)
except Exception:
task_type = 2
if task_type not in (1, 2):
task_type = 2
strategy = (data.get('strategy') or 'Terms').strip()
if strategy not in ('Terms', 'Simple'):
strategy = 'Terms'
if not isinstance(paths, list):
paths = []
paths = [p.strip() for p in paths if p and isinstance(p, str)]
if not paths:
return jsonify({'success': False, 'error': '请提供至少一个文件路径或链接'}), 400
user_id = session['user_id']
urls = []
for p in paths:
if _is_url(p):
urls.append(p)
elif os.path.isfile(p):
url, err = _upload_local_xlsx_to_oss(p, user_id)
if err:
return jsonify({'success': False, 'error': f'上传文件失败: {os.path.basename(p)} - {err}'}), 500
urls.append(url)
else:
return jsonify({'success': False, 'error': f'无效路径或链接: {p[:80]}'}), 400
desc_core = ', '.join(os.path.basename(urlparse(u).path or u) for u in urls)[:500] if urls else ''
desc = f'[{strategy}] ' + (desc_core or '')
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"INSERT INTO brand_crawl_tasks (user_id, file_paths, `desc`, strategy, status, task_type) VALUES (%s, %s, %s, %s, 'pending', %s)",
(user_id, json.dumps(urls), desc or None, strategy, task_type)
)
task_id = cur.lastrowid
conn.commit()
conn.close()
# threading.Thread(target=_background_brand_task, args=(task_id,), daemon=True).start()
return jsonify({'success': True, 'task_id': task_id})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/tasks/<int:task_id>')
@login_required
def api_brand_task_detail(task_id):
"""获取单个任务详情（用于轮询状态）"""
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"""SELECT id, file_paths, `desc`, strategy, status, result_paths, error_message,
COALESCE(progress_current, 0) AS progress_current,
COALESCE(progress_total, 0) AS progress_total,
created_at, updated_at
FROM brand_crawl_tasks WHERE id = %s AND user_id = %s""",
(task_id, session['user_id'])
)
row = cur.fetchone()
conn.close()
if not row:
return jsonify({'success': False, 'error': '任务不存在'}), 404
return jsonify({
'success': True,
'task': {
'id': row['id'],
'file_paths': _parse_json(row['file_paths'], []),
'desc': (row.get('desc') or '').strip() or None,
'strategy': (row.get('strategy') or 'Terms').strip() or 'Terms',
'status': row.get('status') or 'pending',
'result_paths': _parse_json(row['result_paths']),
'error_message': (row.get('error_message') or '').strip() or None,
'progress_current': int(row.get('progress_current') or 0),
'progress_total': int(row.get('progress_total') or 0),
'created_at': row['created_at'].strftime('%Y-%m-%d %H:%M') if row.get('created_at') else '',
'updated_at': row['updated_at'].strftime('%Y-%m-%d %H:%M') if row.get('updated_at') else '',
}
})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/tasks/<int:task_id>/line-progress')
@login_required
def api_brand_task_line_progress(task_id):
"""
获取任务的“当前文件行级进度”（当前行数/总行数）。
该信息仅存放在内存字典 _task_line_progress 中，不写入数据库。
"""
try:
with _task_line_progress_lock:
info = _task_line_progress.get(task_id)
if not info:
return jsonify({'success': True, 'has_progress': False})
# 为了避免暴露完整路径，只返回文件名
file_name = os.path.basename(info.get('file_path') or '') if info.get('file_path') else ''
resp = {
'file_index': int(info.get('file_index') or 0),
'file_total': int(info.get('file_total') or 0),
'file_name': file_name,
'current_line': int(info.get('current_line') or 0),
'total_lines': int(info.get('total_lines') or 0),
}
return jsonify({'success': True, 'has_progress': True, 'info': resp})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/tasks/<int:task_id>/events')
@login_required
def api_brand_task_events(task_id):
"""SSE：任务完成时后端主动推送事件，前端监听后隐藏进度条与取消按钮"""
def _task_status_event():
conn = get_db()
try:
with conn.cursor() as cur:
cur.execute(
"SELECT status, error_message FROM brand_crawl_tasks WHERE id = %s AND user_id = %s",
(task_id, session['user_id'])
)
row = cur.fetchone()
finally:
conn.close()
if not row:
return None
st = (row.get('status') or '').lower()
if st in ('success', 'failed', 'cancelled'):
return {'status': st, 'error_message': (row.get('error_message') or '').strip() or None}
return None
# 若任务已处于终态，直接返回一条事件后结束
ev = _task_status_event()
if ev is not None:
def _one_shot():
yield "data: " + json.dumps(ev, ensure_ascii=False) + "\n\n"
return Response(
_one_shot(),
mimetype='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
)
# 否则注册队列，等待后台任务完成时推送
q = Queue()
with _task_event_lock:
_task_event_queues.setdefault(task_id, []).append(q)
def _stream():
try:
while True:
try:
event = q.get(timeout=20)
yield "data: " + json.dumps(event, ensure_ascii=False) + "\n\n"
return
except Empty:
yield ": keepalive\n\n"
finally:
with _task_event_lock:
lst = _task_event_queues.get(task_id, [])
if q in lst:
lst.remove(q)
if not lst:
_task_event_queues.pop(task_id, None)
return Response(
_stream(),
mimetype='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
)
@brand_bp.route('/api/brand/tasks/<int:task_id>/cancel', methods=['POST'])
@login_required
def api_brand_task_cancel(task_id):
"""取消正在执行或等待中的任务（pending/running 均可取消）"""
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"UPDATE brand_crawl_tasks SET status = 'cancelled' WHERE id = %s AND user_id = %s AND status IN ('pending', 'running')",
(task_id, session['user_id'])
)
n = cur.rowcount
conn.commit()
conn.close()
if n == 0:
return jsonify({'success': False, 'error': '任务不存在或无法取消'}), 400
_push_task_event(task_id, {'status': 'cancelled'})
return jsonify({'success': True})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/tasks/<int:task_id>', methods=['DELETE'])
@login_required
def api_brand_task_delete(task_id):
"""删除任务（仅限非 running 状态）"""
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"DELETE FROM brand_crawl_tasks WHERE id = %s AND user_id = %s AND status != 'running'",
(task_id, session['user_id'])
)
n = cur.rowcount
conn.commit()
conn.close()
if n == 0:
return jsonify({'success': False, 'error': '任务不存在或正在执行中无法删除'}), 400
return jsonify({'success': True})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@brand_bp.route('/api/brand/download/<int:task_id>')
@login_required
def api_brand_download(task_id):
"""下载任务结果：优先使用 OSS zip_url 重定向；否则本地/单链接/多链接按原逻辑处理"""
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute(
"SELECT result_paths FROM brand_crawl_tasks WHERE id = %s AND user_id = %s",
(task_id, session['user_id'])
)
row = cur.fetchone()
conn.close()
if not row or not row.get('result_paths'):
return jsonify({'success': False, 'error': '无结果可下载'}), 404
raw = row['result_paths']
if isinstance(raw, str):
raw = _parse_json(raw)
url_list, zip_url = _normalize_result_paths(raw)
# 新格式：有 zip_url 直接重定向到 OSS
if zip_url and _is_url(zip_url):
return redirect(zip_url, code=302)
# 兼容旧格式：result_paths 可能为 list（本地路径或 url）
if not url_list and isinstance(raw, list):
url_list = raw
local_paths = [p for p in url_list if isinstance(p, str) and not _is_url(p) and os.path.isfile(p)]
url_paths = [p.strip() for p in url_list if isinstance(p, str) and _is_url(p)]
if len(url_paths) == 1 and not local_paths:
return redirect(url_paths[0], code=302)
if len(local_paths) == 1 and not url_paths:
return send_file(
local_paths[0],
as_attachment=True,
download_name=os.path.basename(local_paths[0])
)
if local_paths or url_paths:
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
for p in local_paths:
zf.write(p, os.path.basename(p))
for url in url_paths:
try:
import requests as req
r = req.get(url, timeout=30, stream=True)
r.raise_for_status()
name = os.path.basename(urlparse(url).path) or ('file_%s' % (url_paths.index(url)))
if not name or name == 'file_%s' % url_paths.index(url):
name = 'download_%s' % url_paths.index(url)
zf.writestr(name, r.content)
except Exception:
pass
buf.seek(0)
return send_file(
buf,
mimetype='application/zip',
as_attachment=True,
download_name='brand_task_%s.zip' % task_id
)
return jsonify({'success': False, 'error': '结果文件不存在或链接不可用'}), 404
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500

411
app/blueprints/image.py Normal file
View File

@@ -0,0 +1,411 @@
"""
图片生成蓝图：生成、历史、下载、拼接、版本
"""
import os
import sys
import json
import re
import io
import base64
import tempfile
import threading
import subprocess
import requests
from urllib.parse import urlparse, quote
from flask import Blueprint, request, jsonify, Response, session, send_file
from PIL import Image
from app_common import get_db, login_required, BASE_DIR
from config import STITCH_WORKFLOW_ID,client_name
image_bp = Blueprint('image', __name__)
def _load_image_from_url_or_data(url_or_data):
"""从 http(s) URL 或 data URL 加载为 PIL Image，失败返回 None"""
if not url_or_data or not isinstance(url_or_data, str):
return None
try:
if url_or_data.startswith('data:'):
m = re.match(r'data:image/[^;]+;base64,(.+)', url_or_data, re.DOTALL)
if not m:
return None
raw = base64.b64decode(m.group(1).strip())
img = Image.open(io.BytesIO(raw))
elif url_or_data.startswith(('http://', 'https://')):
resp = requests.get(url_or_data, timeout=15)
resp.raise_for_status()
img = Image.open(io.BytesIO(resp.content))
else:
return None
if img.mode != 'RGB':
img = img.convert('RGB')
return img
except Exception:
return None
def _stitch_and_upload_long_image(urls):
"""将多张图片 URL 先上传获取 file_id，再调用 workflow_run 拼接长图，返回 data.merged_image_url；失败返回 None。"""
if not urls or not isinstance(urls, (list, tuple)):
return None
from coze import upload_file as coze_upload_file, workflow_run
file_ids = []
temp_paths = []
try:
for u in urls:
img = _load_image_from_url_or_data(u)
if img is None:
continue
fd, path = tempfile.mkstemp(suffix='.png')
try:
os.close(fd)
img.save(path)
temp_paths.append(path)
resp = coze_upload_file(path)
if resp.get('code') == 0 and resp.get('data', {}).get('id'):
file_ids.append(resp['data']['id'])
except Exception:
pass
if not file_ids:
return None
parameters = {"images": [{"file_id": fid} for fid in file_ids]}
resp = workflow_run(STITCH_WORKFLOW_ID, parameters, is_async=False)
if resp.get('code') != 0:
return None
data = resp.get('data') or {}
data = json.loads(data)
merged_image_url = data.get('merged_image_url')
if merged_image_url:
return merged_image_url
output_str = data.get('output') or ''
if output_str:
try:
outer = json.loads(output_str)
inner_str = outer.get('Output', '{}')
inner = json.loads(inner_str)
data_str = inner.get('data', '[]')
inner_data = json.loads(data_str)
merged_image_url = inner_data.get('merged_image_url')
return merged_image_url
except Exception:
pass
return None
except Exception:
return None
finally:
for p in temp_paths:
try:
os.unlink(p)
except Exception:
pass
def _sanitize_params_for_history(params):
"""移除 base64 大字段及敏感字段，仅保留可存储的请求参数"""
exclude = ('ref_images', 'proc_images', 'layout_image')
out = {}
for k, v in (params or {}).items():
if k == 'api_key':
continue
if k in exclude:
if isinstance(v, list):
out[f'{k}_count'] = len(v)
else:
out[f'{k}_count'] = 1 if v else 0
elif isinstance(v, (str, int, float, bool, type(None))):
out[k] = v
elif isinstance(v, list) and not v:
out[k] = []
elif isinstance(v, list) and isinstance(v[0], (str, int, float, bool)):
out[k] = v
else:
out[k] = str(v)[:200] if v else None
return out
@image_bp.route('/api/generate', methods=['POST'])
@login_required
def api_generate():
"""生成图片：调用 generate_api，上传原图到 OSS，保存历史记录"""
try:
params = request.get_json() or {}
from generate_api import generate
result = generate(params)
if result.get('success') and result.get('urls'):
long_image_url = result.get("long_image_url")
result["long_image_url"] = long_image_url
import json as _json
history_id = None
try:
hid = params.get('history_id')
if hid is not None:
try:
hid = int(hid)
except (TypeError, ValueError):
hid = None
conn = get_db()
with conn.cursor() as cur:
if hid is not None and hid > 0:
cur.execute(
"SELECT result_urls FROM image_history WHERE id=%s AND user_id=%s",
(hid, session['user_id']),
)
row = cur.fetchone()
existing_urls = []
if row and row.get('result_urls'):
try:
existing_urls = _json.loads(row['result_urls'])
except Exception:
existing_urls = []
new_urls = result.get('urls') or []
new_url = new_urls[0] if new_urls else None
idx = params.get('history_index', 0)
try:
idx = int(idx)
except (TypeError, ValueError):
idx = 0
if new_url:
if not isinstance(existing_urls, list):
existing_urls = []
while len(existing_urls) <= idx:
existing_urls.append(existing_urls[-1] if existing_urls else new_url)
existing_urls[idx] = new_url
merged_result_urls = existing_urls or new_urls
cur.execute(
"""UPDATE image_history SET panel_type=%s, original_urls=%s, params=%s, result_urls=%s
WHERE id=%s AND user_id=%s""",
(
params.get('panel_type', ''),
_json.dumps(result.get('original_urls') or []),
_json.dumps(_sanitize_params_for_history(params)),
_json.dumps(merged_result_urls),
hid,
session['user_id'],
),
)
if cur.rowcount > 0:
history_id = hid
else:
cur.execute(
"""INSERT INTO image_history (user_id, panel_type, original_urls, params, result_urls, long_image_url)
VALUES (%s, %s, %s, %s, %s, %s)""",
(
session['user_id'],
params.get('panel_type', ''),
_json.dumps(result.get('original_urls') or []),
_json.dumps(_sanitize_params_for_history(params)),
_json.dumps(result.get('urls') or []),
long_image_url,
),
)
history_id = cur.lastrowid
conn.commit()
conn.close()
except Exception:
pass
if history_id is not None:
result['history_id'] = history_id
return jsonify(result)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'success': False, 'urls': [], 'error': str(e)})
@image_bp.route('/api/version')
def api_version():
"""检测更新：请求 APP_UPDATE_URL 获取最新版本信息，与当前版本比较"""
current_version = (os.environ.get('APP_VERSION', '1.0.0') or '1.0.0').strip()
update_url = (os.environ.get('APP_UPDATE_URL', '') or '').strip()
result = {
'version': current_version,
'desc': '',
'url': '',
'has_update': False,
'latest_version': current_version,
'file_url': '',
}
if not update_url:
return jsonify(result)
try:
resp = requests.get(update_url, timeout=10)
resp.raise_for_status()
data = resp.json() or {}
latest_version = (data.get('version') or '').strip()
file_url = (data.get('file_url') or '').strip()
result['latest_version'] = latest_version
result['file_url'] = file_url
result['url'] = file_url
# 版本不一致则视为有更新
if latest_version and latest_version != current_version:
result['has_update'] = True
except Exception:
pass
return jsonify(result)
def _run_update_and_exit(zip_path, target_dir):
"""在后台延迟后启动 update.exe（脱离当前进程），然后退出当前程序"""
def _do():
import time
time.sleep(1.5) # 确保 HTTP 响应已发送
# exe_dir = target_dir
# exe_dir = "D:\\pack\\nanri\\update_exe\\main.dist"
exe_dir = os.path.join(BASE_DIR,"update")
update_exe = os.path.join(exe_dir, 'update.exe')
if not os.path.isfile(update_exe):
return
try:
creationflags = 0
if sys.platform == 'win32':
creationflags = subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP
subprocess.Popen(
[update_exe, '--zip', zip_path, '--target', target_dir, '--process',client_name],
cwd=exe_dir,
creationflags=creationflags,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
)
except Exception:
pass
os._exit(0)
t = threading.Thread(target=_do, daemon=False)
t.start()
@image_bp.route('/api/update/do', methods=['POST'])
def api_update_do():
"""执行更新：下载 zip 到 tmp，启动 update.exe 后退出程序"""
data = request.get_json() or {}
file_url = (data.get('file_url') or '').strip()
if not file_url:
return jsonify({'success': False, 'error': '缺少 file_url'}), 400
parsed = urlparse(file_url)
if parsed.scheme not in ('http', 'https'):
return jsonify({'success': False, 'error': '无效的下载地址'}), 400
tmp_dir = os.path.join(BASE_DIR, 'tmp')
try:
os.makedirs(tmp_dir, exist_ok=True)
except Exception as e:
return jsonify({'success': False, 'error': f'创建 tmp 目录失败: {e}'}), 500
# 使用 URL 中的文件名或默认版本名
filename = os.path.basename(parsed.path) or 'update.zip'
zip_path = os.path.join(tmp_dir, filename)
try:
resp = requests.get(file_url, timeout=300, stream=True)
resp.raise_for_status()
with open(zip_path, 'wb') as f:
for chunk in resp.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
except requests.RequestException as e:
return jsonify({'success': False, 'error': f'下载失败: {e}'}), 502
_run_update_and_exit(zip_path, BASE_DIR)
return jsonify({'success': True, 'message': '更新已启动，程序即将退出'})
@image_bp.route('/api/download')
@login_required
def api_download():
"""代理下载图片，解决跨域 fetch 无法下载的问题"""
url = request.args.get('url', '').strip()
filename = request.args.get('filename', 'image.png')
if not url:
return jsonify({'success': False, 'error': '缺少 url 参数'}), 400
parsed = urlparse(url)
if parsed.scheme not in ('http', 'https'):
return jsonify({'success': False, 'error': '仅支持 http/https 链接'}), 400
try:
resp = requests.get(url, timeout=30, stream=True)
resp.raise_for_status()
content_type = resp.headers.get('Content-Type', 'image/png')
encoded = quote(filename, safe='')
disposition = f"attachment; filename*=UTF-8''{encoded}"
return Response(
resp.iter_content(chunk_size=8192),
mimetype=content_type,
headers={'Content-Disposition': disposition}
)
except requests.RequestException as e:
return jsonify({'success': False, 'error': str(e)}), 502
@image_bp.route('/api/stitch/save', methods=['POST'])
@login_required
def api_stitch_save():
"""手动拼接：接收图片 URL 列表（支持 http 或 data URL），拼接并上传，返回长图 URL"""
try:
data = request.get_json() or {}
urls = data.get('urls')
if not urls or not isinstance(urls, list):
return jsonify({'success': False, 'error': '请提供 urls 数组'}), 400
urls = [u for u in urls if u and isinstance(u, str)]
if not urls:
return jsonify({'success': False, 'error': '没有有效的图片'}), 400
long_image_url = _stitch_and_upload_long_image(urls)
if not long_image_url:
return jsonify({'success': False, 'error': '拼接或上传失败'}), 500
return jsonify({'success': True, 'long_image_url': long_image_url})
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@image_bp.route('/api/history')
@login_required
def api_history():
"""分页获取当前用户的历史图库，支持按 panel_type 栏目筛选"""
page = max(1, int(request.args.get('page', 1)))
page_size = min(50, max(10, int(request.args.get('page_size', 20))))
panel_type = (request.args.get('panel_type') or '').strip()
offset = (page - 1) * page_size
try:
conn = get_db()
with conn.cursor() as cur:
where_user = "user_id = %s"
params_where = [session['user_id']]
if panel_type:
where_user += " AND panel_type = %s"
params_where.append(panel_type)
cur.execute(
"""SELECT id, created_at, panel_type, original_urls, params, result_urls, long_image_url
FROM image_history WHERE """ + where_user + """ ORDER BY created_at DESC LIMIT %s OFFSET %s""",
params_where + [page_size, offset],
)
rows = cur.fetchall()
cur.execute("SELECT COUNT(*) as total FROM image_history WHERE " + where_user, params_where)
total = cur.fetchone()['total']
conn.close()
def _parse_json(val, default=None):
if val is None:
return default if default is not None else []
if isinstance(val, (list, dict)):
return val
try:
return json.loads(val)
except Exception:
return default if default is not None else []
items = []
for r in rows:
items.append({
'id': r['id'],
'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r['created_at'] else '',
'panel_type': r['panel_type'] or '',
'original_urls': _parse_json(r['original_urls'], []),
'params': _parse_json(r['params'], {}),
'result_urls': _parse_json(r['result_urls'], []),
'long_image_url': (r.get('long_image_url') or '').strip() or None,
})
return jsonify({'success': True, 'items': items, 'total': total, 'page': page, 'page_size': page_size})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})

95
app/blueprints/main.py Normal file
View File

@@ -0,0 +1,95 @@
"""
主页面蓝图：首页、home、图片工作台、品牌页、静态文件、Logo
"""
import os
from flask import Blueprint, send_file
from app_common import (
get_db,
_render_html,
_is_session_user_valid,
login_required,
admin_required,
STATIC_DIR,
BASE_DIR,
ASSETS_DIR
)
from flask import redirect, url_for, session
from config import base_url,version
main_bp = Blueprint('main', __name__)
@main_bp.route('/')
def index():
if session.get('user_id') and _is_session_user_valid():
return redirect(url_for('main.home'))
return redirect(url_for('auth.login'))
@main_bp.route('/home')
@login_required
def home():
try:
conn = get_db()
with conn.cursor() as cur:
cur.execute("SELECT username, is_admin FROM users WHERE id = %s", (session['user_id'],))
row = cur.fetchone()
conn.close()
return _render_html('home.html', username=row.get('username', ''), is_admin=bool(row.get('is_admin')), user_id=session.get('user_id'),baseUrl=base_url,version=version)
except Exception:
return _render_html('home.html', username=session.get('username', ''), is_admin=False, user_id=session.get('user_id'),baseUrl=base_url,version=version)
@main_bp.route('/image')
@login_required
def wb():
return _render_html('index.html')
@main_bp.route('/brand')
@login_required
def brand_page():
return _render_html('brand.html')
@main_bp.route('/static/<path:filename>')
def serve_static(filename):
"""提供 static 目录及子目录下的静态文件访问。"""
filepath = os.path.normpath(os.path.join(STATIC_DIR, filename))
static_abs = os.path.abspath(STATIC_DIR)
file_abs = os.path.abspath(filepath)
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
return '', 404
return send_file(file_abs, as_attachment=False)
@main_bp.route('/assets/<path:filename>')
def serve_assets(filename):
"""提供 static 目录及子目录下的静态文件访问。"""
filepath = os.path.normpath(os.path.join(ASSETS_DIR, filename))
static_abs = os.path.abspath(ASSETS_DIR)
file_abs = os.path.abspath(filepath)
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
return '', 404
return send_file(file_abs, as_attachment=False)
@main_bp.route('/new_web_source/<path:filename>')
def serve_new_web_source(filename):
"""提供 static 目录及子目录下的静态文件访问。"""
filepath = os.path.normpath(os.path.join("new_web_source", filename))
static_abs = os.path.abspath("new_web_source")
file_abs = os.path.abspath(filepath)
if not file_abs.startswith(static_abs) or not os.path.isfile(file_abs):
return '', 404
return send_file(file_abs, as_attachment=False)
@main_bp.route('/logo.jpg', methods=['GET'])
def get_logo_image():
return send_file(os.path.join(BASE_DIR, "logo.jpg"), mimetype='image/jpeg')

BIN
app/brand_spider/__pycache__/main.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/brand_spider/__pycache__/web_dec.cpython-39.pyc Normal file
View File

Binary file not shown.

692
app/brand_spider/main.py Normal file
View File

@@ -0,0 +1,692 @@
import json
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
try:
from config import proxy_url as CONFIG_PROXY_URL, proxy_mode as CONFIG_PROXY_MODE
except ImportError:
CONFIG_PROXY_URL = None
CONFIG_PROXY_MODE = 1
# Forbidden 后 3 分钟内统一使用代理：记录代理生效截止时间与当前代理
_FORBIDDEN_PROXY_UNTIL = 0.0
_FORBIDDEN_PROXY_DICT = None
_FORBIDDEN_PROXY_MINUTES = 0.5
special_char_pattern = re.compile(r'[^\w\s]')
def remove_accents(input_str):
# 将字符分解为基础字符和重音符号
nksel = unicodedata.normalize('NFKD', input_str)
# 过滤掉非间距重音符号，并重新编码
return "".join([c for c in nksel if not unicodedata.combining(c)])
def clean_text(text):
text = remove_accents(text)
return special_char_pattern.sub(' ', text).replace(" ","")
def create_code_generator():
counter = random.randint(0, 0xFFFFFF)
def get_code():
nonlocal counter
counter = (counter + 1) % 0xFFFFFF
low_16_bits = counter & 0xFFFF
return f"{low_16_bits:04x}"
return get_code()
def search(hashsearch, data, proxies=None):
global _FORBIDDEN_PROXY_UNTIL, _FORBIDDEN_PROXY_DICT
url = "https://api.branddb.wipo.int/search"
headers = {
"accept": "application/json, text/plain, */*",
"accept-language": "zh-CN,zh;q=0.9,en;q=0.8",
"cache-control": "no-cache",
"content-type": "application/json",
"hashsearch": hashsearch,
"origin": "https://branddb.wipo.int",
"pragma": "no-cache",
"priority": "u=1, i",
"referer": "https://branddb.wipo.int/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36"
}
# 若 3 分钟内曾出现 Forbidden，则直接使用当时保存的代理
now = time.time()
if now < _FORBIDDEN_PROXY_UNTIL and _FORBIDDEN_PROXY_DICT:
proxies = _FORBIDDEN_PROXY_DICT
print("处于 Forbidden 代理窗口内，直接使用代理",proxies)
# 发送 POST 请求
response = requests.post(url, headers=headers, json=data, proxies=proxies)
enc_text = response.text
# print("原始结果", enc_text)
# 若返回 Forbidden，则从 config 的 proxy_url 获取代理 IP 后重试，并开启 3 分钟代理窗口
if enc_text.strip() == '{"message":"Forbidden"}' and CONFIG_PROXY_URL:
try:
proxy_resp = requests.get(CONFIG_PROXY_URL, timeout=10)
print("代理请求结果->：", proxy_resp.text)
# 模式 2：账号密码代理，接口返回 JSON
if CONFIG_PROXY_MODE == 2:
resp_json = proxy_resp.json()
proxy_list = resp_json.get("data", {}).get("list") or []
first_item = proxy_list[0] if proxy_list else None
if first_item:
ip = first_item.get("ip")
port = first_item.get("port")
account = first_item.get("account")
password = first_item.get("password")
if ip and port and account and password:
auth_proxy = f"{account}:{password}@{ip}:{port}"
print("获取到账号密码代理：", auth_proxy)
proxies = {
"http": f"http://{auth_proxy}",
"https": f"http://{auth_proxy}",
}
# 默认模式 1：普通 IP:port 文本
if CONFIG_PROXY_MODE != 2:
proxy_ip = (proxy_resp.text or "").strip()
if proxy_ip:
proxies = {
"http": f"http://{proxy_ip}",
"https": f"http://{proxy_ip}",
}
if proxies:
response = requests.post(url, headers=headers, json=data, proxies=proxies)
enc_text = response.text
print("代理重试结果", enc_text)
# 记录 3 分钟内都使用该代理
_FORBIDDEN_PROXY_UNTIL = now + _FORBIDDEN_PROXY_MINUTES * 60
_FORBIDDEN_PROXY_DICT = proxies
except Exception as e:
print("获取代理或重试失败:", e)
return enc_text
class TaskCancelledError(Exception):
"""用户取消任务时抛出，供上层捕获并中止"""
pass
def single_file_handle(file_path, output_path, strategy="Terms", check_cancelled=None, progress_callback=None):
status_info = {
"Ended": "已结束",
"Expired": "已过期的",
"Pending": "待决",
"Registered": "已注册",
"RegisteredMadrid": "国际注册有效",
"Unknown": "未知"
}
country_info = {
"AB": "ARABPAT",
"AD": "安道尔",
"AE": "UAE",
"AF": "阿富汗",
"AFR": "非洲",
"AG": "安提瓜和巴布达",
"AI": "安圭拉",
"AL": "阿尔巴尼亚",
"AM": "亚美尼亚",
"AN": "荷属安的列斯",
"ANT": "南极洲",
"AO": "安哥拉",
"AP": "非洲地区知识产权组织 ",
"AQ": "南极洲",
"AR": "阿根廷",
"AS": "美属萨摩亚",
"ASI": "亚洲",
"AT": "奥地利",
"AU": "澳大利亚",
"AW": "阿鲁巴",
"AX": "奥兰群岛",
"AZ": "阿塞拜疆",
"BA": "波斯尼亚和黑塞哥维纳",
"BB": "巴巴多斯",
"BD": "孟加拉国",
"BE": "比利时",
"BF": "布基纳法索",
"BG": "保加利亚",
"BH": "巴林",
"BI": "布隆迪",
"BJ": "贝宁",
"BL": "圣巴泰勒米",
"BM": "百慕大",
"BN": "文莱达鲁萨兰国",
"BO": "多民族玻利维亚国",
"BQ": "博纳尔、圣俄斯塔休斯和萨巴",
"BR": "巴西",
"BS": "巴哈马",
"BT": "不丹",
"BV": "布韦岛",
"BW": "博茨瓦纳",
"BX": "比荷卢知识产权局",
"BY": "白俄罗斯",
"BZ": "伯利兹",
"CA": "加拿大",
"CC": "科科斯群岛（基灵群岛）",
"CD": "刚果民主共和国",
"CF": "中非共和国",
"CG": "刚果",
"CH": "瑞士",
"CI": "科特迪瓦",
"CK": "库克群岛",
"CL": "智利",
"CM": "喀麦隆",
"CN": "中国",
"CO": "哥伦比亚",
"CR": "哥斯达黎加",
"CS": "捷克斯洛伐克",
"CU": "古巴",
"CV": "佛得角",
"CW": "库拉索",
"CX": "圣诞岛",
"CY": "塞浦路斯",
"CZ": "捷克共和国",
"DD": "德意志民主共和国",
"DE": "德国",
"DJ": "吉布提",
"DK": "丹麦",
"DM": "多米尼克",
"DO": "多米尼加",
"DT": "西德",
"DZ": "阿尔及利亚",
"EA": "欧亚专利组织",
"EC": "厄瓜多尔",
"EE": "爱沙尼亚",
"EG": "埃及",
"EH": "西撒哈拉",
"EM": "欧洲联盟",
"EP": "欧洲专利局",
"ER": "厄立特里亚",
"ES": "西班牙",
"ET": "埃塞俄比亚",
"EUR": "欧洲",
"FI": "芬兰",
"FJ": "斐济",
"FK": "福克兰群岛（马尔维纳斯群岛）",
"FM": "密克罗尼西亚联邦",
"FO": "法罗群岛",
"FR": "法国",
"GA": "加蓬",
"GB": "英国",
"GC": "海湾阿拉伯国家合作委员会专利局",
"GD": "格林纳达",
"GE": "格鲁吉亚",
"GF": "法属圭亚那",
"GG": "格恩西岛",
"GH": "加纳",
"GI": "直布罗陀",
"GL": "格陵兰",
"GM": "冈比亚",
"GN": "几内亚",
"GP": "瓜德罗普",
"GQ": "赤道几内亚",
"GR": "希腊",
"GS": "南乔治亚和南桑威奇群岛",
"GT": "危地马拉",
"GU": "关岛",
"GW": "几内亚比绍",
"GY": "圭亚那",
"HK": "香港",
"HM": "赫德岛和麦克唐纳群岛",
"HN": "洪都拉斯",
"HR": "克罗地亚",
"HT": "海地",
"HU": "匈牙利",
"IB": "世界知识产权组织国际局",
"ID": "印度尼西亚",
"IE": "爱尔兰",
"IL": "以色列",
"IM": "马恩岛",
"IN": "印度",
"INN": "世界卫生组织",
"IO": "英属印度洋领地",
"IQ": "伊拉克",
"IR": "伊朗伊斯兰共和国",
"IS": "冰岛",
"IT": "意大利",
"JE": "泽西岛",
"JM": "牙买加",
"JO": "约旦",
"JP": "日本",
"KE": "肯尼亚",
"KG": "吉尔吉斯斯坦",
"KH": "柬埔寨",
"KI": "基里巴斯",
"KM": "科摩罗",
"KN": "圣基茨和尼维斯",
"KP": "朝鲜民主主义人民共和国",
"KR": "大韩民国",
"KW": "科威特",
"KY": "开曼群岛",
"KZ": "哈萨克斯坦",
"LA": "老挝人民民主共和国",
"LB": "黎巴嫩",
"LC": "圣卢西亚",
"LI": "列支敦士登",
"LISBON": "WIPO",
"LK": "斯里兰卡",
"LP": "LATIPAT",
"LR": "利比里亚",
"LS": "莱索托",
"LT": "立陶宛",
"LU": "卢森堡",
"LV": "拉脱维亚",
"LY": "利比亚",
"MA": "摩洛哥",
"MC": "摩纳哥",
"MD": "摩尔多瓦共和国",
"ME": "黑山",
"MF": "圣马丁（法国部分）",
"MG": "马达加斯加",
"MH": "马绍尔群岛",
"MK": "北马其顿共和国",
"ML": "马里",
"MM": "缅甸",
"MN": "蒙古",
"MO": "澳门",
"MP": "北马里亚纳群岛",
"MQ": "马提尼克",
"MR": "毛里塔尼亚",
"MS": "蒙特塞拉特",
"MT": "马耳他",
"MU": "毛里求斯",
"MV": "马尔代夫",
"MW": "马拉维",
"MX": "墨西哥",
"MY": "马来西亚",
"MZ": "莫桑比克",
"NA": "纳米比亚",
"NAM": "北美洲",
"NC": "新喀里多尼亚",
"NE": "尼日尔",
"NF": "诺福克岛",
"NG": "尼日利亚",
"NI": "尼加拉瓜",
"NL": "荷兰",
"NO": "挪威",
"NP": "尼泊尔",
"NR": "瑙鲁",
"NU": "纽埃",
"NZ": "新西兰",
"OA": "非洲知识产权组织",
"OCE": "大洋洲",
"OM": "阿曼",
"PA": "巴拿马",
"PE": "秘鲁",
"PF": "法属波利尼西亚",
"PG": "巴布亚新几内亚",
"PH": "菲律宾",
"PK": "巴基斯坦",
"PL": "波兰",
"PM": "圣皮埃尔和密克隆",
"PN": "皮特凯恩",
"PR": "波多黎各",
"PS": "巴勒斯坦",
"PT": "葡萄牙",
"PW": "帕劳",
"PY": "巴拉圭",
"QA": "卡塔尔",
"QO": "没有ST.3代码的组织",
"QZ": "欧洲联盟",
"RE": "留尼汪",
"RO": "罗马尼亚",
"RS": "塞尔维亚",
"RU": "俄罗斯联邦",
"RW": "卢旺达",
"SA": "沙特阿拉伯",
"SAM": "南美洲",
"SB": "所罗门群岛",
"SC": "塞舌尔",
"SD": "苏丹",
"SE": "瑞典",
"SG": "新加坡",
"SH": "圣赫勒拿、阿森松和特里斯坦-达库尼亚",
"SI": "斯洛文尼亚",
"SIXTER": "WIPO",
"SJ": "斯瓦尔巴和扬马延",
"SK": "斯洛伐克",
"SL": "塞拉里昂",
"SM": "圣马力诺",
"SN": "塞内加尔",
"SO": "索马里",
"SR": "苏里南",
"SS": "南苏丹",
"ST": "圣多美和普林西比",
"SU": "苏联",
"SV": "萨尔瓦多",
"SX": "圣马丁（荷兰部分）",
"SY": "阿拉伯叙利亚共和国",
"SZ": "斯威士兰",
"TC": "特克斯和凯科斯群岛",
"TD": "乍得",
"TF": "法属南部领地",
"TG": "多哥",
"TH": "泰国",
"TJ": "塔吉克斯坦",
"TK": "托克劳",
"TL": "东帝汶",
"TM": "土库曼斯坦",
"TN": "突尼斯",
"TO": "汤加",
"TR": "土耳其",
"TT": "特立尼达和多巴哥",
"TV": "图瓦卢",
"TW": "台湾（中国的省）",
"TZ": "坦桑尼亚联合共和国",
"UA": "乌克兰",
"UG": "乌干达",
"UK": "UK",
"UM": "美国本土外小岛屿",
"US": "USA",
"UY": "乌拉圭",
"UZ": "乌兹别克斯坦",
"VA": "罗马教廷",
"VC": "圣文森特和格林纳丁斯",
"VE": "委内瑞拉玻利瓦尔共和国",
"VG": "英属维尔京群岛",
"VI": "美属维尔京群岛",
"VN": "越南",
"VU": "瓦努阿图",
"WF": "瓦利斯和富图纳",
"WHO": "世界卫生组织",
"WO": "WIPO",
"WS": "萨摩亚",
"XK": "科索沃共和国",
"XN": "北欧专利局",
"XX": "国际",
"XXX": "跨国和国际局",
"YD": "民主也门",
"YE": "也门",
"YT": "马约特岛",
"YU": "塞尔维亚和黑山",
"ZA": "南非",
"ZM": "赞比亚",
"ZW": "津巴布韦"
}
check_value = [
"瑞典",
"芬兰",
"丹麦",
"挪威",
"冰岛",
"法国",
"爱尔兰",
"荷兰",
"比利时",
"卢森堡",
"英国",
"摩纳哥",
"德国",
"波兰",
"捷克",
"斯洛伐克",
"匈牙利",
"奥地利",
"瑞士",
"列支敦士登",
"爱沙尼亚",
"拉脱维亚",
"立陶宛",
"俄罗斯",
"白俄罗斯",
"乌克兰",
"摩尔多瓦",
"西班牙",
"葡萄牙",
"意大利",
"希腊",
"斯洛文尼亚",
"克罗地亚",
"罗马尼亚",
"保加利亚",
"塞尔维亚",
"阿尔巴尼亚",
"黑山",
"马耳他",
"塞浦路斯",
"北马其顿",
"梵蒂冈",
"圣马力诺",
"安道尔",
"波黑",
"欧洲联盟"
]
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
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_"
}
# 翻页：第二页 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(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 strategy == "Simple":
new_brand_name = clean_text(brand_name)
new_brand_name = new_brand_name.lower()
if _brand != new_brand_name:
continue
check_office = d.get("designation")
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
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
if __name__ == '__main__':
file_path = r"D:\\私单交付\\maixiang_AI\\测试图片数据\\品牌采集测试\\brand_task_73\\源文件\\brand_task_75\\源文件\\3.10.xlsx"
res = single_file_handle(file_path,"结果.xlsx")
print(res)

126
app/brand_spider/web_dec.py Normal file
View File

@@ -0,0 +1,126 @@
"""
通过 pywebview 加载本地 HTML，执行 JS 实现解密与 GUID 生成。
"""
import json
import threading
import traceback
import webview
# 内嵌 HTML：加载 CryptoJS 并定义 decryptWithHashSearches、guid
_DECRYPT_HTML = """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<script src="https://cdnjs.cloudflare.com/ajax/libs/crypto-js/4.1.1/crypto-js.min.js"></script>
</head>
<body>
<script>
function decryptWithHashSearches(ciphertext, hashSearches) {
try {
var baseKey = "8?)i_~Nk6qv0IX;2";
var keyStr = baseKey + (hashSearches || "");
var key = CryptoJS.enc.Utf8.parse(keyStr);
var decrypted = CryptoJS.AES.decrypt(ciphertext, key, {
mode: CryptoJS.mode.ECB
});
return decrypted.toString(CryptoJS.enc.Utf8);
} catch (error) {
console.error('解密失败:', error.message);
return '';
}
}
function guid() {
function _p8(s) {
var p = (Math.random().toString(16) + "000000000").substr(2, 8);
return s ? "-" + p.substr(0, 4) + "-" + p.substr(4, 4) : p;
}
return _p8() + _p8(true) + _p8(true) + _p8();
}
</script>
</body>
</html>
"""
_window = None
_window_ready = threading.Event()
_init_lock = threading.Lock()
def _ensure_window():
"""首次调用时在后台线程启动 pywebview 并等待页面加载完成。"""
global _window
with _init_lock:
if _window is not None:
return
_window = webview.create_window(
"",
html=_DECRYPT_HTML,
width=1,
height=1,
hidden=True,
)
def run():
webview.start(debug=False)
t = threading.Thread(target=run, daemon=True)
t.start()
_window.events.loaded.wait()
_window_ready.set()
_window_ready.wait()
def decrypt_via_service(hash_searches, ciphertext):
"""
通过 pywebview 执行 JS decryptWithHashSearches 进行 AES-ECB 解密。
:param hash_searches: 对应 JS 的 hashSearches（密钥后缀，可为空串）
:param ciphertext: Base64 密文
:return: 解密后的 UTF-8 字符串，失败返回空串
"""
_ensure_window()
# 将参数安全注入 JS（避免注入与引号问题）
ciphertext_js = json.dumps(ciphertext)
hash_searches_js = json.dumps(hash_searches or "")
js = f"(function(){{ return decryptWithHashSearches({ciphertext_js}, {hash_searches_js}); }})();"
try:
result = _window.evaluate_js(js)
return result if result is not None else ""
except Exception as e:
raise RuntimeError(f"解密失败: {e}") from e
def get_guid():
"""通过 pywebview 执行 JS guid() 生成 UUID 格式字符串。"""
_ensure_window()
try:
result = _window.evaluate_js("(function(){ return guid(); })();")
if result is None:
raise RuntimeError("guid() 返回为空")
return result
except Exception as e:
raise RuntimeError(f"获取 GUID 失败: {e}") from e
# 使用示例
if __name__ == "__main__":
# 参数含义：hash_searches 对应原 key，ciphertext 对应原 data
hash_searches = "a6efb809-b714-7efd-4b64-8c650b9030f0"
ciphertext = "SEsfpOGa8B+sWk0ncknTNh/HNHbGiEVi/RNKxBvyHmAE5VjHonPi202c6VicC/GKfA8mLsIC5mGEpSaH2DdCaEJKeOTWD9SBHHbgtyS1O60VqjgAaptYe9LivvWKc/BU8sZOqhxPMzGUHDcKUts7d0p+hCc80XCXyM2ZT80smM1twndcDfpGkLDk2kJbzl2bGzIc60sl9MzKWfZA2sE0ztFjQ9wD2uhn5LrwoN8NnpiPLNbviMMGtHh4N6Dc0xtPzkzgfkiuxBfWnN1SeM9XVgujHvAGea/dqUWIJqLo26fZIOEFJ0MYL4c8CGLYIeP/70cT2IqJdf+IPBWsCiP29zSyAiQd3yLNvd8+xBLky7lR4ng3MZizn/vhW/5BSg1FtVglbAmNbKgHOIbtpP197Lv+uahx2IpsSPqpy4j2O2VV25YzBk9JpJ4WGG/SvF3f2ZYKKepEQ+kGmBxAfG/5Z8DTNIwnOpXhyjFpJb+fUdPV6LTCK/yFc8g31bNFAJkLW7y/VjewjFravZ3VfQjNAHCvifnrIxGG22MZ3TLVnlbh6ye6vTnd+v9GzqXu+ISR7vQGL/ZSM/bJIjuVkP2XnNUPf+NFdt75gyDmTylFmmbpg7WHaBjinPcyVjeZ38+quJhT8yEk66BOjBM47mVdfU8JLK3ToghxQ54dKWGUbc9HRzYIAQ0rIBBExcOcPM4J9DpufvajmEygoDaws4CxOQDlFoFLwNBYorJsKzAoDe1Cu8oFtk4x190Vd+leRKq6DQSNJwEmyrVkWyFpiuywSMaixFlSqve0lRs50FclXfV7gkBAAvz/DJp90i9yCJdMaihP5ZCvbhKxFGMowkU+tx5Ptnxd9hq6tUxHmuiwDI1eyY8EyjB+3MbrC3H/GY39Z5Qhkj5tkSBm4oGA/h6YjeunBlfMU/QGP3hyZIYALh/YmBlZxAcglWwcqlISUhcx1L3Dbw6ZQbpAmYHmA421xIlRJkdJuPmGoopcoFEVdIO0Ov3uP2JL5ybvUgCBMnbwcYRK9qRmvn2b8jHKch55YXerHBDCEMHz88rfgNIEL8DIAp+wMuIyJ0VrV0BwAXioLYNvMcggBp06gghq2NykGiXRZ/dz8nffFqPVkDNjzcPSGrml9fpKdO8x/GannMNBsA+oHlO2/d4dyo+Q2dZCarQ5UB/N1p+UsTW90kvITKwCbZfV/YY2NRm7HrXq+wLS1vY4mChY82Ybc3cYQT653Q=="
try:
decrypted_text = decrypt_via_service(hash_searches, ciphertext)
print("解密结果:", decrypted_text)
except Exception as e:
print("错误:", e)
try:
g = get_guid()
print("GUID:", g)
except Exception as e:
print("GUID 错误:", e)

BIN
app/static/bg.jpg Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 160 KiB

BIN
app/static/品牌文档格式_模板.xlsx Normal file
View File

Binary file not shown.

BIN
app/static/模板2-以文件夹方式上传.zip Normal file
View File

Binary file not shown.

1
app/tool/.device_id Normal file
View File

@@ -0,0 +1 @@
8cecee3bc02a178bf372ca1c3d02fc5cae3c0a51c8346f6e6a710988599c08be

BIN
app/tool/__pycache__/devices.cpython-311.pyc Normal file
View File

Binary file not shown.

BIN
app/tool/__pycache__/devices.cpython-39.pyc Normal file
View File

Binary file not shown.

249
app/tool/devices.py Normal file
View File

@@ -0,0 +1,249 @@
import hashlib
import platform
import subprocess
import uuid
import os
from typing import Optional
class DeviceIDGenerator:
"""
Windows设备唯一ID生成器
通过收集多个硬件特征来生成稳定的设备唯一标识符
"""
def __init__(self, use_cache: bool = False, cache_file: str = ".device_id"):
"""
初始化设备ID生成器
Args:
use_cache: 是否使用本地缓存
cache_file: 缓存文件名
"""
self.use_cache = use_cache
self.cache_file = cache_file
def _run_wmic_command(self, command: str) -> Optional[str]:
"""
执行WMIC命令并返回结果
Args:
command: WMIC命令
Returns:
命令执行结果，失败则返回None
"""
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0 and result.stdout.strip():
return result.stdout.strip()
except (subprocess.TimeoutExpired, Exception):
pass
return None
def _get_motherboard_serial(self) -> Optional[str]:
"""获取主板序列号"""
return self._run_wmic_command("wmic baseboard get serialnumber /value")
def _get_cpu_id(self) -> Optional[str]:
"""获取CPU ID"""
return self._run_wmic_command("wmic cpu get processorid /value")
def _get_bios_serial(self) -> Optional[str]:
"""获取BIOS序列号"""
return self._run_wmic_command("wmic bios get serialnumber /value")
def _get_disk_serial(self) -> Optional[str]:
"""获取系统盘序列号"""
return self._run_wmic_command("wmic diskdrive get serialnumber /value")
def _get_machine_guid(self) -> Optional[str]:
"""获取Windows机器GUID"""
try:
result = subprocess.run(
'reg query "HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Cryptography" /v MachineGuid',
shell=True,
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
for line in result.stdout.split('\n'):
if 'MachineGuid' in line:
return line.split()[-1]
except Exception:
pass
return None
def _extract_value(self, wmic_output: str) -> str:
"""从WMIC输出中提取实际值"""
if not wmic_output:
return ""
lines = wmic_output.split('\n')
for line in lines:
if '=' in line and not line.strip().endswith('='):
return line.split('=', 1)[1].strip()
return ""
def _collect_hardware_info(self) -> dict:
"""
收集硬件信息
Returns:
包含各种硬件信息的字典
"""
hardware_info = {}
# 主板序列号
motherboard = self._get_motherboard_serial()
hardware_info['motherboard'] = self._extract_value(motherboard) if motherboard else ""
# CPU ID
cpu_id = self._get_cpu_id()
hardware_info['cpu'] = self._extract_value(cpu_id) if cpu_id else ""
# BIOS序列号
bios = self._get_bios_serial()
hardware_info['bios'] = self._extract_value(bios) if bios else ""
# 硬盘序列号
disk = self._get_disk_serial()
hardware_info['disk'] = self._extract_value(disk) if disk else ""
# Windows机器GUID
machine_guid = self._get_machine_guid()
hardware_info['machine_guid'] = machine_guid if machine_guid else ""
# 计算机名称
hardware_info['computer_name'] = platform.node()
# MAC地址（作为备用）
hardware_info['mac_address'] = ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff)
for elements in range(0, 2*6, 2)][::-1])
return hardware_info
def _generate_device_id(self, hardware_info: dict) -> str:
"""
基于硬件信息生成设备ID
Args:
hardware_info: 硬件信息字典
Returns:
32位十六进制设备ID
"""
# 过滤掉空值，并按键排序确保一致性
filtered_info = {k: v for k, v in hardware_info.items() if v and v.strip()}
# 如果没有任何硬件信息，使用MAC地址作为后备方案
if not filtered_info:
filtered_info = {'mac_address': hardware_info.get('mac_address', str(uuid.getnode()))}
# 将所有信息连接成字符串
info_string = '|'.join(f"{k}:{v}" for k, v in sorted(filtered_info.items()))
# 使用SHA256生成哈希值
hash_object = hashlib.sha256(info_string.encode('utf-8'))
device_id = hash_object.hexdigest()
return device_id
def _load_cached_device_id(self) -> Optional[str]:
"""从缓存文件加载设备ID"""
try:
if os.path.exists(self.cache_file):
with open(self.cache_file, 'r', encoding='utf-8') as f:
cached_id = f.read().strip()
if len(cached_id) == 64: # SHA256哈希长度
return cached_id
except Exception:
pass
return None
def _save_device_id_to_cache(self, device_id: str) -> None:
"""将设备ID保存到缓存文件"""
try:
with open(self.cache_file, 'w', encoding='utf-8') as f:
f.write(device_id)
except Exception:
pass
def get_device_id(self) -> str:
"""
获取设备唯一ID
Returns:
64字符的十六进制设备ID
"""
# 如果启用缓存，先尝试从缓存加载
if self.use_cache:
cached_id = self._load_cached_device_id()
if cached_id:
return cached_id
# 收集硬件信息
hardware_info = self._collect_hardware_info()
# 生成设备ID
device_id = self._generate_device_id(hardware_info)
# 保存到缓存
if self.use_cache:
self._save_device_id_to_cache(device_id)
return device_id
def get_device_id_short(self, length: int = 16) -> str:
"""
获取短版本的设备ID
Args:
length: 返回ID的长度
Returns:
指定长度的设备ID
"""
full_id = self.get_device_id()
return full_id[:length]
def get_hardware_info(self) -> dict:
"""
获取硬件信息（用于调试）
Returns:
硬件信息字典
"""
return self._collect_hardware_info()
# 使用示例
def main():
"""使用示例"""
# 创建设备ID生成器实例
device_generator = DeviceIDGenerator()
# 获取完整设备ID（64字符）
device_id = device_generator.get_device_id()
print(f"完整设备ID: {device_id}")
# 获取短版本设备ID（16字符）
short_id = device_generator.get_device_id_short(16)
print(f"短设备ID: {short_id}")
# 查看硬件信息（调试用）
hardware_info = device_generator.get_hardware_info()
print("\n硬件信息:")
for key, value in hardware_info.items():
print(f" {key}: {value}")
if __name__ == "__main__":
main()