diff --git a/frontend-vue/convert.html b/frontend-vue/convert.html new file mode 100644 index 0000000..2639248 --- /dev/null +++ b/frontend-vue/convert.html @@ -0,0 +1,12 @@ + + + + + + 格式转换 - 数富AI + + +
+ + + diff --git a/frontend-vue/dedupe.html b/frontend-vue/dedupe.html new file mode 100644 index 0000000..12ed0d5 --- /dev/null +++ b/frontend-vue/dedupe.html @@ -0,0 +1,12 @@ + + + + + + 数据去重 - 数富AI + + +
+ + + diff --git a/frontend-vue/split.html b/frontend-vue/split.html new file mode 100644 index 0000000..ac5f76a --- /dev/null +++ b/frontend-vue/split.html @@ -0,0 +1,12 @@ + + + + + + 数据拆分 - 数富AI + + +
+ + + diff --git a/frontend-vue/src/App.vue b/frontend-vue/src/App.vue deleted file mode 100644 index 98240ae..0000000 --- a/frontend-vue/src/App.vue +++ /dev/null @@ -1,3 +0,0 @@ - diff --git a/frontend-vue/src/convert-main.ts b/frontend-vue/src/convert-main.ts new file mode 100644 index 0000000..68ec465 --- /dev/null +++ b/frontend-vue/src/convert-main.ts @@ -0,0 +1,7 @@ +import { createApp } from 'vue' +import ElementPlus from 'element-plus' +import 'element-plus/dist/index.css' +import '@/styles/main.css' +import BrandConvertTab from '@/pages/brand/components/BrandConvertTab.vue' + +createApp(BrandConvertTab).use(ElementPlus).mount('#app') diff --git a/frontend-vue/src/dedupe-main.ts b/frontend-vue/src/dedupe-main.ts new file mode 100644 index 0000000..75f0a21 --- /dev/null +++ b/frontend-vue/src/dedupe-main.ts @@ -0,0 +1,7 @@ +import { createApp } from 'vue' +import ElementPlus from 'element-plus' +import 'element-plus/dist/index.css' +import '@/styles/main.css' +import BrandDedupeTab from '@/pages/brand/components/BrandDedupeTab.vue' + +createApp(BrandDedupeTab).use(ElementPlus).mount('#app') diff --git a/frontend-vue/src/pages/admin/index.vue b/frontend-vue/src/pages/admin/index.vue deleted file mode 100644 index ba566d8..0000000 --- a/frontend-vue/src/pages/admin/index.vue +++ /dev/null @@ -1,584 +0,0 @@ - - - - - diff --git a/frontend-vue/src/pages/brand/index.vue b/frontend-vue/src/pages/brand/index.vue deleted file mode 100644 index 3e3f025..0000000 --- a/frontend-vue/src/pages/brand/index.vue +++ /dev/null @@ -1,1069 +0,0 @@ - - - - - \ No newline at end of file diff --git a/frontend-vue/src/pages/home/index.vue b/frontend-vue/src/pages/home/index.vue deleted file mode 100644 index 2dafe26..0000000 --- a/frontend-vue/src/pages/home/index.vue +++ /dev/null @@ -1,166 +0,0 @@ - - - - - diff --git a/frontend-vue/src/pages/image/index.vue b/frontend-vue/src/pages/image/index.vue deleted file mode 100644 index 9b44488..0000000 --- a/frontend-vue/src/pages/image/index.vue +++ /dev/null @@ -1,94 +0,0 @@ - - - - - diff --git a/frontend-vue/src/pages/login/index.vue b/frontend-vue/src/pages/login/index.vue deleted file mode 100644 index 5756671..0000000 --- a/frontend-vue/src/pages/login/index.vue +++ /dev/null @@ -1,137 +0,0 @@ - - - - - diff --git a/frontend-vue/src/router.ts b/frontend-vue/src/router.ts deleted file mode 100644 index 5b336fc..0000000 --- a/frontend-vue/src/router.ts +++ /dev/null @@ -1,46 +0,0 @@ -import { createRouter, createWebHistory } from 'vue-router' - -const routes = [ - { - path: '/', - redirect: '/home', - }, - { - path: '/login', - component: () => import('@/pages/login/index.vue'), - meta: { title: '登录 - 数富AI' }, - }, - { - path: '/home', - component: () => import('@/pages/home/index.vue'), - meta: { title: '首页 - 数富AI' }, - }, - { - path: '/brand', - component: () => import('@/pages/brand/index.vue'), - meta: { title: '亚马逊 - 数富AI' }, - }, - { - path: '/admin', - component: () => import('@/pages/admin/index.vue'), - meta: { title: '管理后台 - 数富AI' }, - }, - { - path: '/image', - component: () => import('@/pages/image/index.vue'), - meta: { title: '图片 - 数富AI' }, - }, -] - -const router = createRouter({ - history: createWebHistory(), - routes, -}) - -router.afterEach((to) => { - if (to.meta.title) { - document.title = String(to.meta.title) - } -}) - -export default router diff --git a/frontend-vue/src/shared/api/admin.ts b/frontend-vue/src/shared/api/admin.ts deleted file mode 100644 index c5c65e6..0000000 --- a/frontend-vue/src/shared/api/admin.ts +++ /dev/null @@ -1,96 +0,0 @@ -import { requestDeleteJson, requestGetJson, requestPostJson, requestPutJson } from '@/shared/api/http' - -export interface AdminUserItem { - id: number - username?: string - role?: 'super_admin' | 'admin' | 'normal' | string - creator_username?: string - created_at?: string -} - -export interface AdminSimpleUser { - id: number - username: string -} - -export interface AdminUsersResponse { - success: boolean - current_user_role?: 'super_admin' | 'admin' | 'normal' | string - admins?: AdminSimpleUser[] - items?: AdminUserItem[] - total?: number - page?: number - page_size?: number - error?: string -} - -export interface AdminMutationResponse { - success: boolean - msg?: string - error?: string -} - -export interface AdminHistoryItem { - id: number - username?: string - panel_type?: string - created_at?: string - long_image_url?: string | null - result_urls?: string[] -} - -export interface AdminHistoryResponse { - success: boolean - items?: AdminHistoryItem[] - total?: number - page?: number - page_size?: number - error?: string -} - -export interface AdminColumnPermissionItem { - column_key?: string -} - -export interface AdminColumnPermissionResponse { - success: boolean - items?: AdminColumnPermissionItem[] - error?: string -} - -export function getAdminUsers(query = '') { - const suffix = query ? `?${query}` : '' - return requestGetJson(`/api/admin/users${suffix}`) -} - -export function createAdminUser(payload: { - username: string - password: string - role: string - created_by_id?: number -}) { - return requestPostJson('/api/admin/user', payload) -} - -export function updateAdminUser( - userId: number | string, - payload: { - password?: string - role?: string - }, -) { - return requestPutJson(`/api/admin/user/${userId}`, payload) -} - -export function deleteAdminUser(userId: number | string) { - return requestDeleteJson(`/api/admin/user/${userId}`) -} - -export function getAdminHistory(query = '') { - const suffix = query ? `?${query}` : '' - return requestGetJson(`/api/admin/history${suffix}`) -} - -export function getUserColumnPermissions(userId: string | number) { - return requestGetJson(`/api/admin/user/${userId}/column-permissions`) -} diff --git a/frontend-vue/src/shared/api/auth.ts b/frontend-vue/src/shared/api/auth.ts deleted file mode 100644 index da7da51..0000000 --- a/frontend-vue/src/shared/api/auth.ts +++ /dev/null @@ -1,38 +0,0 @@ -import { requestGetJson, requestPostJson } from '@/shared/api/http' - -export interface LoginResponse { - success: boolean - redirect?: string - error?: string -} - -export interface AuthCheckResponse { - logged_in: boolean - redirect?: string -} - -export async function checkAuth() { - return requestGetJson('/api/auth/check') -} - -export async function login(username: string, password: string) { - const formData = new FormData() - formData.set('username', username) - formData.set('password', password) - - return requestPostJson('/login', formData, { - headers: { - 'Content-Type': 'multipart/form-data', - }, - }) -} - -export async function logout() { - try { - await requestGetJson('/logout', { - validateStatus: () => true, - }) - } catch { - // logout should still continue on the client even if backend responds abnormally - } -} diff --git a/frontend-vue/src/shared/api/convert.ts b/frontend-vue/src/shared/api/convert.ts deleted file mode 100644 index ae3ebea..0000000 --- a/frontend-vue/src/shared/api/convert.ts +++ /dev/null @@ -1,63 +0,0 @@ -import { requestGetJson, requestPostJson } from '@/shared/api/http' - -export interface UploadFileVo { - fileKey: string - originalFilename?: string - localPath?: string - size?: number -} - -export interface ConvertTemplateVo { - id: number - templateCode: string - templateName: string - outputFilename: string - isDefault?: boolean - builtIn?: boolean -} - -export interface UploadedSourceFileDto { - fileKey: string - originalFilename?: string -} - -export interface ConvertResultItemVo { - resultId?: number - sourceFilename?: string - outputFilename?: string - success: boolean - error?: string - downloadUrl?: string -} - -export interface ConvertRunVo { - total: number - successCount: number - failedCount: number - items: ConvertResultItemVo[] -} - -export interface StandardApiResponse { - success: boolean - message?: string - data?: T -} - -export function getConvertTemplates() { - return requestGetJson>('/api/convert/templates') -} - -export function uploadTempFile(file: File) { - const formData = new FormData() - formData.append('file', file) - return requestPostJson, FormData>('/api/files/upload', formData, { - headers: { 'Content-Type': 'multipart/form-data' }, - }) -} - -export function runConvert(files: UploadedSourceFileDto[], templateId: number) { - return requestPostJson>('/api/convert/run', { - files, - templateId, - }) -} diff --git a/frontend-vue/src/shared/api/dedupe.ts b/frontend-vue/src/shared/api/dedupe.ts deleted file mode 100644 index 7622020..0000000 --- a/frontend-vue/src/shared/api/dedupe.ts +++ /dev/null @@ -1,22 +0,0 @@ -import { requestGetJson } from '@/shared/api/http' - -export interface DedupeHistoryItem { - sourceFilename?: string - outputFilename?: string - downloadUrl?: string - success: boolean -} - -export interface DedupeHistoryVo { - items: DedupeHistoryItem[] -} - -export interface StandardApiResponse { - success: boolean - message?: string - data?: T -} - -export function getDedupeHistory() { - return requestGetJson>('/api/dedupe/history') -} diff --git a/frontend-vue/src/shared/api/update.ts b/frontend-vue/src/shared/api/update.ts deleted file mode 100644 index 099ed55..0000000 --- a/frontend-vue/src/shared/api/update.ts +++ /dev/null @@ -1,22 +0,0 @@ -import { requestGetJson, requestPostJson } from '@/shared/api/http' - -export interface VersionResponse { - version?: string - has_update?: boolean - latest_version?: string - desc?: string - file_url?: string -} - -export interface UpdateStartResponse { - success: boolean - error?: string -} - -export function fetchVersion() { - return requestGetJson('/api/version') -} - -export function startUpdate(fileUrl: string) { - return requestPostJson('/api/update/do', { file_url: fileUrl }) -} diff --git a/frontend-vue/src/shared/bridges/pywebview.ts b/frontend-vue/src/shared/bridges/pywebview.ts index f941311..5aa0109 100644 --- a/frontend-vue/src/shared/bridges/pywebview.ts +++ b/frontend-vue/src/shared/bridges/pywebview.ts @@ -4,7 +4,7 @@ export interface SplitExcelPayload { split_mode: 'rows_per_file' | 'parts' rows_per_file?: number parts?: number - output_dir: string + output_dir?: string } export interface SplitExcelResultItem { @@ -37,6 +37,14 @@ export interface ExcelInfoResponse { error?: string } +export interface CleanExcelPayload { + paths: string[] + selected_columns: string[] + keep_integer_ids?: boolean + keep_underscore_ids?: boolean + output_dir?: string +} + export interface CleanExcelResultItem { result_id?: number source_path: string @@ -47,7 +55,7 @@ export interface CleanExcelResultItem { export interface ConvertTxtPayload { paths: string[] - output_dir: string + output_dir?: string template_id: string } diff --git a/frontend-vue/src/shared/composables/useUpdateCheck.ts b/frontend-vue/src/shared/composables/useUpdateCheck.ts deleted file mode 100644 index ca03bf4..0000000 --- a/frontend-vue/src/shared/composables/useUpdateCheck.ts +++ /dev/null @@ -1,71 +0,0 @@ -import { ref } from 'vue' -import { fetchVersion, startUpdate } from '@/shared/api/update' - -export function useUpdateCheck() { - const loading = ref(false) - const version = ref('1.0.0') - const hint = ref('') - const downloadVisible = ref(false) - const downloadLoading = ref(false) - const fileUrl = ref('') - - const check = async () => { - loading.value = true - hint.value = '正在检测更新...' - downloadVisible.value = false - - try { - const result = await fetchVersion() - version.value = (result.version || version.value).replace(/^v/i, '') - - if (result.has_update && result.latest_version) { - hint.value = `发现新版本 v${result.latest_version}${result.desc ? `：${result.desc}` : ''}` - fileUrl.value = result.file_url || '' - downloadVisible.value = true - return - } - - hint.value = '已是最新版本' - } catch (error) { - hint.value = error instanceof Error ? error.message : '检测更新失败' - } finally { - loading.value = false - } - } - - const runUpdate = async () => { - if (!fileUrl.value) { - hint.value = '暂无下载地址，请关注官方渠道。' - return false - } - - downloadLoading.value = true - hint.value = '正在下载并准备更新，程序将自动退出...' - - try { - const result = await startUpdate(fileUrl.value) - if (result.success) { - hint.value = '更新已启动，程序即将退出...' - return true - } - - hint.value = result.error || '更新启动失败' - return false - } catch (error) { - hint.value = error instanceof Error ? error.message : '请求更新失败，请重试' - return false - } finally { - downloadLoading.value = false - } - } - - return { - loading, - version, - hint, - downloadVisible, - downloadLoading, - check, - runUpdate, - } -} diff --git a/frontend-vue/src/main.ts b/frontend-vue/src/split-main.ts similarity index 51% rename from frontend-vue/src/main.ts rename to frontend-vue/src/split-main.ts index 34be832..b74c843 100644 --- a/frontend-vue/src/main.ts +++ b/frontend-vue/src/split-main.ts @@ -2,7 +2,6 @@ import { createApp } from 'vue' import ElementPlus from 'element-plus' import 'element-plus/dist/index.css' import '@/styles/main.css' -import App from '@/App.vue' -import router from '@/router' +import BrandSplitTab from '@/pages/brand/components/BrandSplitTab.vue' -createApp(App).use(ElementPlus).use(router).mount('#app') +createApp(BrandSplitTab).use(ElementPlus).mount('#app') diff --git a/frontend-vue/vite.config.ts b/frontend-vue/vite.config.ts index cd05be6..cb1dcd7 100644 --- a/frontend-vue/vite.config.ts +++ b/frontend-vue/vite.config.ts @@ -1,4 +1,5 @@ import { fileURLToPath, URL } from 'node:url' +import { resolve } from 'node:path' import { defineConfig } from 'vite' import vue from '@vitejs/plugin-vue' import AutoImport from 'unplugin-auto-import/vite' @@ -6,6 +7,7 @@ import Components from 'unplugin-vue-components/vite' import { ElementPlusResolver } from 'unplugin-vue-components/resolvers' export default defineConfig({ + base: './', plugins: [ vue(), AutoImport({ @@ -47,5 +49,19 @@ export default defineConfig({ }, build: { outDir: 'dist', + emptyOutDir: true, + cssCodeSplit: true, + rollupOptions: { + input: { + dedupe: resolve(__dirname, 'dedupe.html'), + convert: resolve(__dirname, 'convert.html'), + split: resolve(__dirname, 'split.html'), + }, + output: { + entryFileNames: 'assets/[name].js', + chunkFileNames: 'assets/[name]-[hash].js', + assetFileNames: 'assets/[name]-[hash][extname]', + }, + }, }, }) diff --git a/ali_oss.py b/source_code/ali_oss.py similarity index 97% rename from ali_oss.py rename to source_code/ali_oss.py index 179fadd..0cb6b40 100644 --- a/ali_oss.py +++ b/source_code/ali_oss.py @@ -1,73 +1,73 @@ -import argparse -import base64 -import re -import time -import alibabacloud_oss_v2 as oss -import requests - -from config import region, endpoint, bucket, file_url_pre, bucket_path - - -def upload_file(file_content: bytes, key: str): - credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider() - cfg = oss.config.load_default() - cfg.credentials_provider = credentials_provider - cfg.region = region - cfg.endpoint = endpoint - cfg.retry_max_attempts = 3 - client = oss.Client(cfg) - - result = client.put_object( - oss.PutObjectRequest( - bucket=bucket, # 存储空间名称 - key=key, # 对象名称 - body=file_content # 读取文件内容 - ) - ) - # print(result) - return file_url_pre + key - - -def upload_data_url(data_url: str, prefix: str = "history", key_hint: str = "") -> str: - """ - 将 base64 data URL 上传到 OSS，返回图片链接 - data_url: data:image/png;base64,xxxx 或 data:image/jpeg;base64,xxxx - prefix: OSS key 前缀 - key_hint: 可选后缀避免重名，如 "_0", "_1" - """ - match = re.match(r'data:image/(\w+);base64,(.+)', data_url) - if not match: - raise ValueError('无效的 data URL 格式') - ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg' - file_content = base64.b64decode(match.group(2)) - ts = int(time.time() * 1000) - key = f"{bucket_path}{prefix}/{ts}{key_hint}.{ext}" - return upload_file(file_content, key) - - -def upload_data_urls(data_urls: list, prefix: str = "history") -> list: - """批量上传 base64 图片到 OSS，返回图片链接列表""" - urls = [] - ts = int(time.time() * 1000) - for i, data_url in enumerate(data_urls or []): - if not data_url or not isinstance(data_url, str): - continue - if not data_url.startswith("http"): - match = re.match(r'data:image/(\w+);base64,(.+)', data_url) - if not match: - continue - ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg' - file_content = base64.b64decode(match.group(2)) - else: - file_content = requests.get(data_url).content - ext = "png" - key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}" - urls.append(upload_file(file_content, key)) - return urls - - -# 脚本入口，当文件被直接运行时调用main函数 -if __name__ == "__main__": - with open("测试图片数据/IMG_2685.JPG", "rb") as f: - file_content = f.read() +import argparse +import base64 +import re +import time +import alibabacloud_oss_v2 as oss +import requests + +from config import region, endpoint, bucket, file_url_pre, bucket_path + + +def upload_file(file_content: bytes, key: str): + credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider() + cfg = oss.config.load_default() + cfg.credentials_provider = credentials_provider + cfg.region = region + cfg.endpoint = endpoint + cfg.retry_max_attempts = 3 + client = oss.Client(cfg) + + result = client.put_object( + oss.PutObjectRequest( + bucket=bucket, # 存储空间名称 + key=key, # 对象名称 + body=file_content # 读取文件内容 + ) + ) + # print(result) + return file_url_pre + key + + +def upload_data_url(data_url: str, prefix: str = "history", key_hint: str = "") -> str: + """ + 将 base64 data URL 上传到 OSS，返回图片链接 + data_url: data:image/png;base64,xxxx 或 data:image/jpeg;base64,xxxx + prefix: OSS key 前缀 + key_hint: 可选后缀避免重名，如 "_0", "_1" + """ + match = re.match(r'data:image/(\w+);base64,(.+)', data_url) + if not match: + raise ValueError('无效的 data URL 格式') + ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg' + file_content = base64.b64decode(match.group(2)) + ts = int(time.time() * 1000) + key = f"{bucket_path}{prefix}/{ts}{key_hint}.{ext}" + return upload_file(file_content, key) + + +def upload_data_urls(data_urls: list, prefix: str = "history") -> list: + """批量上传 base64 图片到 OSS，返回图片链接列表""" + urls = [] + ts = int(time.time() * 1000) + for i, data_url in enumerate(data_urls or []): + if not data_url or not isinstance(data_url, str): + continue + if not data_url.startswith("http"): + match = re.match(r'data:image/(\w+);base64,(.+)', data_url) + if not match: + continue + ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg' + file_content = base64.b64decode(match.group(2)) + else: + file_content = requests.get(data_url).content + ext = "png" + key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}" + urls.append(upload_file(file_content, key)) + return urls + + +# 脚本入口，当文件被直接运行时调用main函数 +if __name__ == "__main__": + with open("测试图片数据/IMG_2685.JPG", "rb") as f: + file_content = f.read() upload_file(file_content,key=bucket_path+"test.png") \ No newline at end of file diff --git a/app.py b/source_code/app.py similarity index 100% rename from app.py rename to source_code/app.py diff --git a/app_common.py b/source_code/app_common.py similarity index 97% rename from app_common.py rename to source_code/app_common.py index 67169e0..6928c14 100644 --- a/app_common.py +++ b/source_code/app_common.py @@ -1,261 +1,261 @@ -""" -公共模块：数据库连接、初始化、会话校验、装饰器、模板渲染 -供各蓝图复用 -""" -import os -import secrets -from datetime import timedelta -from functools import wraps - -import pymysql -from flask import request, redirect, url_for, session, jsonify, render_template, render_template_string - -from config import mysql_host, mysql_user, mysql_password, mysql_database - -BASE_DIR = os.path.dirname(os.path.abspath(__file__)) -STATIC_DIR = os.path.join(BASE_DIR, 'static') - - -def get_db(): - return pymysql.connect( - host=mysql_host, - user=mysql_user, - password=mysql_password, - database=mysql_database, - charset='utf8mb4', - cursorclass=pymysql.cursors.DictCursor - ) - - -def _render_html(template_name: str, **context): - """读取 HTML 模板：若为加密文件则先解密，再渲染。未加密或解密失败时按明文渲染。""" - path = os.path.join(BASE_DIR, "web_source", template_name) - if not os.path.isfile(path): - return render_template(template_name, **context) - with open(path, "rb") as f: - raw = f.read() - try: - from html_crypto import decrypt - content = decrypt(raw).decode("utf-8") - except Exception: - content = raw.decode("utf-8", errors="replace") - return render_template_string(content, **context) - - -def _is_session_user_valid(): - """校验 session 中的 user_id 是否在数据库中仍存在；不存在则清除 session 并返回 False""" - uid = session.get('user_id') - if not uid: - return False - try: - conn = get_db() - with conn.cursor() as cur: - cur.execute("SELECT id FROM users WHERE id = %s", (uid,)) - row = cur.fetchone() - conn.close() - if not row: - session.clear() - return False - return True - except Exception: - session.clear() - return False - - -def _get_current_admin_role(): - """获取当前登录用户的管理角色：super_admin / admin / None（非管理员）""" - try: - conn = get_db() - with conn.cursor() as cur: - cur.execute( - "SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s", - (session['user_id'],) - ) - row = cur.fetchone() - conn.close() - if not row or not row.get('is_admin'): - return None, None - return row.get('role') or ('super_admin' if row.get('created_by_id') is None else 'admin'), row - except Exception: - return None, None - - -def login_required(f): - @wraps(f) - def decorated(*args, **kwargs): - if not session.get('user_id') or not _is_session_user_valid(): - if request.headers.get('X-Requested-With') == 'XMLHttpRequest': - return jsonify({'success': False, 'error': '未登录'}), 401 - return redirect(url_for('auth.login')) - return f(*args, **kwargs) - return decorated - - -def admin_required(f): - @wraps(f) - def decorated(*args, **kwargs): - if not session.get('user_id'): - if request.headers.get('X-Requested-With') == 'XMLHttpRequest': - return jsonify({'success': False, 'error': '未登录'}), 401 - return redirect(url_for('auth.login')) - try: - conn = get_db() - with conn.cursor() as cur: - cur.execute("SELECT is_admin, role FROM users WHERE id = %s", (session['user_id'],)) - row = cur.fetchone() - conn.close() - if not row or not row.get('is_admin'): - if request.headers.get('X-Requested-With') == 'XMLHttpRequest': - return jsonify({'success': False, 'error': '需要管理员权限'}), 403 - return redirect(url_for('main.home')) - except Exception as e: - if request.headers.get('X-Requested-With') == 'XMLHttpRequest': - return jsonify({'success': False, 'error': str(e)}), 500 - return redirect(url_for('main.home')) - return f(*args, **kwargs) - return decorated - - -def init_db(): - """初始化数据库表，若不存在则创建""" - from werkzeug.security import generate_password_hash - conn = pymysql.connect( - host=mysql_host, - user=mysql_user, - password=mysql_password, - charset='utf8mb4' - ) - try: - with conn.cursor() as cur: - cur.execute(f"CREATE DATABASE IF NOT EXISTS `{mysql_database}` DEFAULT CHARSET utf8mb4") - cur.execute(f"USE `{mysql_database}`") - cur.execute(""" - CREATE TABLE IF NOT EXISTS users ( - id INT AUTO_INCREMENT PRIMARY KEY, - username VARCHAR(64) NOT NULL UNIQUE, - password_hash VARCHAR(256) NOT NULL, - is_admin TINYINT(1) DEFAULT 0, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP - ) - """) - cur.execute(""" - CREATE TABLE IF NOT EXISTS image_history ( - id INT AUTO_INCREMENT PRIMARY KEY, - user_id INT NOT NULL, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP, - panel_type VARCHAR(64) DEFAULT '', - original_urls JSON, - params JSON, - result_urls JSON, - long_image_url VARCHAR(1024) DEFAULT NULL, - INDEX idx_user_created (user_id, created_at DESC) - ) - """) - try: - cur.execute("ALTER TABLE image_history ADD COLUMN long_image_url VARCHAR(1024) DEFAULT NULL") - except Exception: - pass - try: - cur.execute("ALTER TABLE users ADD COLUMN machine VARCHAR(64) DEFAULT NULL") - except Exception: - pass - try: - cur.execute("ALTER TABLE users ADD COLUMN role VARCHAR(20) DEFAULT 'normal'") - except Exception: - pass - try: - cur.execute("ALTER TABLE users ADD COLUMN created_by_id INT NULL") - except Exception: - pass - cur.execute(""" - CREATE TABLE IF NOT EXISTS brand_crawl_tasks ( - id INT AUTO_INCREMENT PRIMARY KEY, - user_id INT NOT NULL, - file_paths JSON NOT NULL, - status VARCHAR(20) DEFAULT 'pending', - task_type TINYINT DEFAULT 1 COMMENT '1=立即执行,2=添加任务执行', - result_paths JSON NULL, - error_message TEXT NULL, - progress_current INT DEFAULT 0, - progress_total INT DEFAULT 0, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - INDEX idx_user_status (user_id, status) - ) - """) - try: - cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN progress_current INT DEFAULT 0") - except Exception: - pass - try: - cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN progress_total INT DEFAULT 0") - except Exception: - pass - try: - cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN `desc` VARCHAR(500) NULL COMMENT '上传文件名描述'") - except Exception: - pass - try: - cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN strategy VARCHAR(20) DEFAULT 'Terms' COMMENT '品牌匹配方式: Terms/Simple'") - except Exception: - pass - try: - cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN task_type TINYINT DEFAULT 1 COMMENT '1=立即执行,2=添加任务执行'") - except Exception: - pass - cur.execute(""" - CREATE TABLE IF NOT EXISTS columns ( - id INT AUTO_INCREMENT PRIMARY KEY, - name VARCHAR(128) NOT NULL COMMENT '栏目名', - column_key VARCHAR(64) NOT NULL COMMENT '栏目标识', - created_at DATETIME DEFAULT CURRENT_TIMESTAMP, - UNIQUE KEY uk_column_key (column_key) - ) - """) - cur.execute(""" - CREATE TABLE IF NOT EXISTS user_column_permission ( - user_id INT NOT NULL, - column_id INT NOT NULL, - PRIMARY KEY (user_id, column_id), - FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, - FOREIGN KEY (column_id) REFERENCES columns(id) ON DELETE CASCADE - ) - """) - try: - cur.execute("UPDATE users SET role = 'normal' WHERE (role IS NULL OR role = '') AND (is_admin = 0 OR is_admin IS NULL)") - cur.execute("SELECT MIN(id) AS mid FROM users WHERE is_admin = 1") - row = cur.fetchone() - if row and row.get('mid'): - mid = row['mid'] - cur.execute("UPDATE users SET role = 'super_admin' WHERE id = %s", (mid,)) - cur.execute("UPDATE users SET role = 'admin' WHERE is_admin = 1 AND id != %s", (mid,)) - cur.execute("UPDATE users SET created_by_id = %s WHERE role = 'admin' AND (created_by_id IS NULL)", (mid,)) - except Exception: - pass - conn.commit() - finally: - conn.close() - _create_initial_admin() - - -def _create_initial_admin(): - """若没有任何管理员，则创建默认超级管理员（首次启动时，仅一个）""" - from werkzeug.security import generate_password_hash - try: - conn = get_db() - with conn.cursor() as cur: - cur.execute("SELECT id FROM users WHERE role = 'super_admin' LIMIT 1") - if cur.fetchone(): - conn.close() - return - admin_user = os.environ.get('ADMIN_USER', 'admin') - admin_pwd = os.environ.get('ADMIN_PASSWORD', 'admin123') - pwd_hash = generate_password_hash(admin_pwd, method='pbkdf2:sha256') - cur.execute( - "INSERT INTO users (username, password_hash, is_admin, role) VALUES (%s, %s, 1, 'super_admin')", - (admin_user, pwd_hash) - ) - conn.commit() - conn.close() - except Exception: - pass +""" +公共模块：数据库连接、初始化、会话校验、装饰器、模板渲染 +供各蓝图复用 +""" +import os +import secrets +from datetime import timedelta +from functools import wraps + +import pymysql +from flask import request, redirect, url_for, session, jsonify, render_template, render_template_string + +from config import mysql_host, mysql_user, mysql_password, mysql_database + +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +STATIC_DIR = os.path.join(BASE_DIR, 'static') + + +def get_db(): + return pymysql.connect( + host=mysql_host, + user=mysql_user, + password=mysql_password, + database=mysql_database, + charset='utf8mb4', + cursorclass=pymysql.cursors.DictCursor + ) + + +def _render_html(template_name: str, **context): + """读取 HTML 模板：若为加密文件则先解密，再渲染。未加密或解密失败时按明文渲染。""" + path = os.path.join(BASE_DIR, "web_source", template_name) + if not os.path.isfile(path): + return render_template(template_name, **context) + with open(path, "rb") as f: + raw = f.read() + try: + from html_crypto import decrypt + content = decrypt(raw).decode("utf-8") + except Exception: + content = raw.decode("utf-8", errors="replace") + return render_template_string(content, **context) + + +def _is_session_user_valid(): + """校验 session 中的 user_id 是否在数据库中仍存在；不存在则清除 session 并返回 False""" + uid = session.get('user_id') + if not uid: + return False + try: + conn = get_db() + with conn.cursor() as cur: + cur.execute("SELECT id FROM users WHERE id = %s", (uid,)) + row = cur.fetchone() + conn.close() + if not row: + session.clear() + return False + return True + except Exception: + session.clear() + return False + + +def _get_current_admin_role(): + """获取当前登录用户的管理角色：super_admin / admin / None（非管理员）""" + try: + conn = get_db() + with conn.cursor() as cur: + cur.execute( + "SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s", + (session['user_id'],) + ) + row = cur.fetchone() + conn.close() + if not row or not row.get('is_admin'): + return None, None + return row.get('role') or ('super_admin' if row.get('created_by_id') is None else 'admin'), row + except Exception: + return None, None + + +def login_required(f): + @wraps(f) + def decorated(*args, **kwargs): + if not session.get('user_id') or not _is_session_user_valid(): + if request.headers.get('X-Requested-With') == 'XMLHttpRequest': + return jsonify({'success': False, 'error': '未登录'}), 401 + return redirect(url_for('auth.login')) + return f(*args, **kwargs) + return decorated + + +def admin_required(f): + @wraps(f) + def decorated(*args, **kwargs): + if not session.get('user_id'): + if request.headers.get('X-Requested-With') == 'XMLHttpRequest': + return jsonify({'success': False, 'error': '未登录'}), 401 + return redirect(url_for('auth.login')) + try: + conn = get_db() + with conn.cursor() as cur: + cur.execute("SELECT is_admin, role FROM users WHERE id = %s", (session['user_id'],)) + row = cur.fetchone() + conn.close() + if not row or not row.get('is_admin'): + if request.headers.get('X-Requested-With') == 'XMLHttpRequest': + return jsonify({'success': False, 'error': '需要管理员权限'}), 403 + return redirect(url_for('main.home')) + except Exception as e: + if request.headers.get('X-Requested-With') == 'XMLHttpRequest': + return jsonify({'success': False, 'error': str(e)}), 500 + return redirect(url_for('main.home')) + return f(*args, **kwargs) + return decorated + + +def init_db(): + """初始化数据库表，若不存在则创建""" + from werkzeug.security import generate_password_hash + conn = pymysql.connect( + host=mysql_host, + user=mysql_user, + password=mysql_password, + charset='utf8mb4' + ) + try: + with conn.cursor() as cur: + cur.execute(f"CREATE DATABASE IF NOT EXISTS `{mysql_database}` DEFAULT CHARSET utf8mb4") + cur.execute(f"USE `{mysql_database}`") + cur.execute(""" + CREATE TABLE IF NOT EXISTS users ( + id INT AUTO_INCREMENT PRIMARY KEY, + username VARCHAR(64) NOT NULL UNIQUE, + password_hash VARCHAR(256) NOT NULL, + is_admin TINYINT(1) DEFAULT 0, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + """) + cur.execute(""" + CREATE TABLE IF NOT EXISTS image_history ( + id INT AUTO_INCREMENT PRIMARY KEY, + user_id INT NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + panel_type VARCHAR(64) DEFAULT '', + original_urls JSON, + params JSON, + result_urls JSON, + long_image_url VARCHAR(1024) DEFAULT NULL, + INDEX idx_user_created (user_id, created_at DESC) + ) + """) + try: + cur.execute("ALTER TABLE image_history ADD COLUMN long_image_url VARCHAR(1024) DEFAULT NULL") + except Exception: + pass + try: + cur.execute("ALTER TABLE users ADD COLUMN machine VARCHAR(64) DEFAULT NULL") + except Exception: + pass + try: + cur.execute("ALTER TABLE users ADD COLUMN role VARCHAR(20) DEFAULT 'normal'") + except Exception: + pass + try: + cur.execute("ALTER TABLE users ADD COLUMN created_by_id INT NULL") + except Exception: + pass + cur.execute(""" + CREATE TABLE IF NOT EXISTS brand_crawl_tasks ( + id INT AUTO_INCREMENT PRIMARY KEY, + user_id INT NOT NULL, + file_paths JSON NOT NULL, + status VARCHAR(20) DEFAULT 'pending', + task_type TINYINT DEFAULT 1 COMMENT '1=立即执行,2=添加任务执行', + result_paths JSON NULL, + error_message TEXT NULL, + progress_current INT DEFAULT 0, + progress_total INT DEFAULT 0, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + INDEX idx_user_status (user_id, status) + ) + """) + try: + cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN progress_current INT DEFAULT 0") + except Exception: + pass + try: + cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN progress_total INT DEFAULT 0") + except Exception: + pass + try: + cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN `desc` VARCHAR(500) NULL COMMENT '上传文件名描述'") + except Exception: + pass + try: + cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN strategy VARCHAR(20) DEFAULT 'Terms' COMMENT '品牌匹配方式: Terms/Simple'") + except Exception: + pass + try: + cur.execute("ALTER TABLE brand_crawl_tasks ADD COLUMN task_type TINYINT DEFAULT 1 COMMENT '1=立即执行,2=添加任务执行'") + except Exception: + pass + cur.execute(""" + CREATE TABLE IF NOT EXISTS columns ( + id INT AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(128) NOT NULL COMMENT '栏目名', + column_key VARCHAR(64) NOT NULL COMMENT '栏目标识', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + UNIQUE KEY uk_column_key (column_key) + ) + """) + cur.execute(""" + CREATE TABLE IF NOT EXISTS user_column_permission ( + user_id INT NOT NULL, + column_id INT NOT NULL, + PRIMARY KEY (user_id, column_id), + FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, + FOREIGN KEY (column_id) REFERENCES columns(id) ON DELETE CASCADE + ) + """) + try: + cur.execute("UPDATE users SET role = 'normal' WHERE (role IS NULL OR role = '') AND (is_admin = 0 OR is_admin IS NULL)") + cur.execute("SELECT MIN(id) AS mid FROM users WHERE is_admin = 1") + row = cur.fetchone() + if row and row.get('mid'): + mid = row['mid'] + cur.execute("UPDATE users SET role = 'super_admin' WHERE id = %s", (mid,)) + cur.execute("UPDATE users SET role = 'admin' WHERE is_admin = 1 AND id != %s", (mid,)) + cur.execute("UPDATE users SET created_by_id = %s WHERE role = 'admin' AND (created_by_id IS NULL)", (mid,)) + except Exception: + pass + conn.commit() + finally: + conn.close() + _create_initial_admin() + + +def _create_initial_admin(): + """若没有任何管理员，则创建默认超级管理员（首次启动时，仅一个）""" + from werkzeug.security import generate_password_hash + try: + conn = get_db() + with conn.cursor() as cur: + cur.execute("SELECT id FROM users WHERE role = 'super_admin' LIMIT 1") + if cur.fetchone(): + conn.close() + return + admin_user = os.environ.get('ADMIN_USER', 'admin') + admin_pwd = os.environ.get('ADMIN_PASSWORD', 'admin123') + pwd_hash = generate_password_hash(admin_pwd, method='pbkdf2:sha256') + cur.execute( + "INSERT INTO users (username, password_hash, is_admin, role) VALUES (%s, %s, 1, 'super_admin')", + (admin_user, pwd_hash) + ) + conn.commit() + conn.close() + except Exception: + pass diff --git a/blueprints/__init__.py b/source_code/blueprints/__init__.py similarity index 92% rename from blueprints/__init__.py rename to source_code/blueprints/__init__.py index 283385b..c4c4c58 100644 --- a/blueprints/__init__.py +++ b/source_code/blueprints/__init__.py @@ -1 +1 @@ -# 蓝图包 +# 蓝图包 diff --git a/blueprints/__pycache__/__init__.cpython-311.pyc b/source_code/blueprints/__pycache__/__init__.cpython-311.pyc similarity index 100% rename from blueprints/__pycache__/__init__.cpython-311.pyc rename to source_code/blueprints/__pycache__/__init__.cpython-311.pyc diff --git a/blueprints/__pycache__/__init__.cpython-39.pyc b/source_code/blueprints/__pycache__/__init__.cpython-39.pyc similarity index 100% rename from blueprints/__pycache__/__init__.cpython-39.pyc rename to source_code/blueprints/__pycache__/__init__.cpython-39.pyc diff --git a/blueprints/__pycache__/admin.cpython-311.pyc b/source_code/blueprints/__pycache__/admin.cpython-311.pyc similarity index 100% rename from blueprints/__pycache__/admin.cpython-311.pyc rename to source_code/blueprints/__pycache__/admin.cpython-311.pyc diff --git a/blueprints/__pycache__/admin.cpython-39.pyc b/source_code/blueprints/__pycache__/admin.cpython-39.pyc similarity index 100% rename from blueprints/__pycache__/admin.cpython-39.pyc rename to source_code/blueprints/__pycache__/admin.cpython-39.pyc diff --git a/blueprints/__pycache__/auth.cpython-311.pyc b/source_code/blueprints/__pycache__/auth.cpython-311.pyc similarity index 100% rename from blueprints/__pycache__/auth.cpython-311.pyc rename to source_code/blueprints/__pycache__/auth.cpython-311.pyc diff --git a/blueprints/__pycache__/auth.cpython-39.pyc b/source_code/blueprints/__pycache__/auth.cpython-39.pyc similarity index 100% rename from blueprints/__pycache__/auth.cpython-39.pyc rename to source_code/blueprints/__pycache__/auth.cpython-39.pyc diff --git a/blueprints/__pycache__/brand.cpython-311.pyc b/source_code/blueprints/__pycache__/brand.cpython-311.pyc similarity index 100% rename from blueprints/__pycache__/brand.cpython-311.pyc rename to source_code/blueprints/__pycache__/brand.cpython-311.pyc diff --git a/blueprints/__pycache__/brand.cpython-39.pyc b/source_code/blueprints/__pycache__/brand.cpython-39.pyc similarity index 100% rename from blueprints/__pycache__/brand.cpython-39.pyc rename to source_code/blueprints/__pycache__/brand.cpython-39.pyc diff --git a/blueprints/__pycache__/image.cpython-311.pyc b/source_code/blueprints/__pycache__/image.cpython-311.pyc similarity index 100% rename from blueprints/__pycache__/image.cpython-311.pyc rename to source_code/blueprints/__pycache__/image.cpython-311.pyc diff --git a/blueprints/__pycache__/image.cpython-39.pyc b/source_code/blueprints/__pycache__/image.cpython-39.pyc similarity index 100% rename from blueprints/__pycache__/image.cpython-39.pyc rename to source_code/blueprints/__pycache__/image.cpython-39.pyc diff --git a/blueprints/__pycache__/main.cpython-311.pyc b/source_code/blueprints/__pycache__/main.cpython-311.pyc similarity index 100% rename from blueprints/__pycache__/main.cpython-311.pyc rename to source_code/blueprints/__pycache__/main.cpython-311.pyc diff --git a/blueprints/__pycache__/main.cpython-39.pyc b/source_code/blueprints/__pycache__/main.cpython-39.pyc similarity index 100% rename from blueprints/__pycache__/main.cpython-39.pyc rename to source_code/blueprints/__pycache__/main.cpython-39.pyc diff --git a/blueprints/admin.py b/source_code/blueprints/admin.py similarity index 97% rename from blueprints/admin.py rename to source_code/blueprints/admin.py index ab64fc9..cac93ec 100644 --- a/blueprints/admin.py +++ b/source_code/blueprints/admin.py @@ -1,362 +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/', 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/', 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//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)}) +""" +管理员蓝图：用户管理（列表/创建/更新/删除）、生成历史、管理页 +""" +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/', 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/', 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//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)}) diff --git a/blueprints/auth.py b/source_code/blueprints/auth.py similarity index 97% rename from blueprints/auth.py rename to source_code/blueprints/auth.py index 7b003e0..6bf5896 100644 --- a/blueprints/auth.py +++ b/source_code/blueprints/auth.py @@ -1,107 +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')) +""" +认证蓝图：登录、登出、登录状态校验 +""" +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')) diff --git a/blueprints/brand.py b/source_code/blueprints/brand.py similarity index 100% rename from blueprints/brand.py rename to source_code/blueprints/brand.py diff --git a/blueprints/image.py b/source_code/blueprints/image.py similarity index 97% rename from blueprints/image.py rename to source_code/blueprints/image.py index c454d54..ae07cbe 100644 --- a/blueprints/image.py +++ b/source_code/blueprints/image.py @@ -1,411 +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)}) - - - - - +""" +图片生成蓝图：生成、历史、下载、拼接、版本 +""" +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)}) + + + + + diff --git a/blueprints/main.py b/source_code/blueprints/main.py similarity index 88% rename from blueprints/main.py rename to source_code/blueprints/main.py index c6ec5a2..cec6167 100644 --- a/blueprints/main.py +++ b/source_code/blueprints/main.py @@ -1,69 +1,78 @@ -""" -主页面蓝图：首页、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, -) -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/') -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('/logo.jpg', methods=['GET']) -def get_logo_image(): - return send_file(os.path.join(BASE_DIR, "logo.jpg"), mimetype='image/jpeg') +""" +主页面蓝图：首页、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, +) +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('/brand-tools') +@login_required +def brand_tools_page(): + html_path = os.path.join(STATIC_DIR, 'brand-tools', 'brand-tools.html') + if os.path.isfile(html_path): + return send_file(html_path) + return _render_html('brand_tools.html') + + +@main_bp.route('/static/') +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('/logo.jpg', methods=['GET']) +def get_logo_image(): + return send_file(os.path.join(BASE_DIR, "logo.jpg"), mimetype='image/jpeg') diff --git a/brand_spider/__pycache__/main.cpython-39.pyc b/source_code/brand_spider/__pycache__/main.cpython-39.pyc similarity index 100% rename from brand_spider/__pycache__/main.cpython-39.pyc rename to source_code/brand_spider/__pycache__/main.cpython-39.pyc diff --git a/brand_spider/__pycache__/web_dec.cpython-39.pyc b/source_code/brand_spider/__pycache__/web_dec.cpython-39.pyc similarity index 100% rename from brand_spider/__pycache__/web_dec.cpython-39.pyc rename to source_code/brand_spider/__pycache__/web_dec.cpython-39.pyc diff --git a/brand_spider/main.py b/source_code/brand_spider/main.py similarity index 100% rename from brand_spider/main.py rename to source_code/brand_spider/main.py diff --git a/brand_spider/web_dec.py b/source_code/brand_spider/web_dec.py similarity index 97% rename from brand_spider/web_dec.py rename to source_code/brand_spider/web_dec.py index 7610c86..09fa965 100644 --- a/brand_spider/web_dec.py +++ b/source_code/brand_spider/web_dec.py @@ -1,126 +1,126 @@ -""" -通过 pywebview 加载本地 HTML，执行 JS 实现解密与 GUID 生成。 -""" -import json -import threading -import traceback - -import webview - -# 内嵌 HTML：加载 CryptoJS 并定义 decryptWithHashSearches、guid -_DECRYPT_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) +""" +通过 pywebview 加载本地 HTML，执行 JS 实现解密与 GUID 生成。 +""" +import json +import threading +import traceback + +import webview + +# 内嵌 HTML：加载 CryptoJS 并定义 decryptWithHashSearches、guid +_DECRYPT_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) diff --git a/config.py b/source_code/config.py similarity index 96% rename from config.py rename to source_code/config.py index 56b7ef8..84ad94a 100644 --- a/config.py +++ b/source_code/config.py @@ -1,46 +1,46 @@ -base_url = "https://api.coze.cn/v1" -coze_token = "sat_12nW40INoJxArrDXbY4lSCoudbqkOYTcphC99BP2efWyzxmsk4q81WDX3ezWgqZ5" -workflow_id = "7608812635877900322" -STITCH_WORKFLOW_ID = "7608813873483300907" - -import os -from dotenv import load_dotenv -load_dotenv() -_base_url = os.getenv("base_url","http://159.75.121.33:15124") - -# MySQL 配置 -mysql_host = os.getenv("mysql_host") -mysql_user = os.getenv("mysql_user") -mysql_password = os.getenv("mysql_password","WTFrb5y6hNLz6hNy") -mysql_database = os.getenv("mysql_database","aiimage") -proxy_url = os.getenv("proxy_url") -proxy_mode = int(os.getenv("proxy_mode",1)) - -client_name=os.getenv("client_name") + ".exe" - -cache_path = "./user_data" - -region = "cn-hangzhou" -endpoint = "oss-cn-hangzhou.aliyuncs.com" -bucket = "nanri-ai-images" -accessKeyId = "LTAI5tNpyvzMNz9f2dHarsm8" -accessKeySecret = "bQSZnFH455i8tzyOgeahJmUzwmhynz" -bucket_path = "nanri-image/" - -file_url_pre = f"https://{bucket}.oss-cn-hangzhou.aliyuncs.com/" - - - -os.environ['OSS_ACCESS_KEY_ID'] = accessKeyId -os.environ['OSS_ACCESS_KEY_SECRET'] = accessKeySecret -os.environ['SECRET_KEY'] = "ddffc7c1d02121d9554d7b080b2511b6" - - -debug = True -version = "1.0.3" -APP_UPDATE_URL = f"{_base_url}/api/version/latest" # 检测更新接口地址，返回 { "file_url": "...", "version": "x.x.x" } -os.environ['APP_VERSION'] = version -os.environ['APP_UPDATE_URL'] = APP_UPDATE_URL - - - +base_url = "https://api.coze.cn/v1" +coze_token = "sat_12nW40INoJxArrDXbY4lSCoudbqkOYTcphC99BP2efWyzxmsk4q81WDX3ezWgqZ5" +workflow_id = "7608812635877900322" +STITCH_WORKFLOW_ID = "7608813873483300907" + +import os +from dotenv import load_dotenv +load_dotenv() +_base_url = os.getenv("base_url","http://159.75.121.33:15124") + +# MySQL 配置 +mysql_host = os.getenv("mysql_host") +mysql_user = os.getenv("mysql_user") +mysql_password = os.getenv("mysql_password","WTFrb5y6hNLz6hNy") +mysql_database = os.getenv("mysql_database","aiimage") +proxy_url = os.getenv("proxy_url") +proxy_mode = int(os.getenv("proxy_mode",1)) + +client_name=os.getenv("client_name") + ".exe" + +cache_path = "./user_data" + +region = "cn-hangzhou" +endpoint = "oss-cn-hangzhou.aliyuncs.com" +bucket = "nanri-ai-images" +accessKeyId = "LTAI5tNpyvzMNz9f2dHarsm8" +accessKeySecret = "bQSZnFH455i8tzyOgeahJmUzwmhynz" +bucket_path = "nanri-image/" + +file_url_pre = f"https://{bucket}.oss-cn-hangzhou.aliyuncs.com/" + + + +os.environ['OSS_ACCESS_KEY_ID'] = accessKeyId +os.environ['OSS_ACCESS_KEY_SECRET'] = accessKeySecret +os.environ['SECRET_KEY'] = "ddffc7c1d02121d9554d7b080b2511b6" + + +debug = True +version = "1.0.3" +APP_UPDATE_URL = f"{_base_url}/api/version/latest" # 检测更新接口地址，返回 { "file_url": "...", "version": "x.x.x" } +os.environ['APP_VERSION'] = version +os.environ['APP_UPDATE_URL'] = APP_UPDATE_URL + + + diff --git a/coze.py b/source_code/coze.py similarity index 100% rename from coze.py rename to source_code/coze.py diff --git a/generate_api.py b/source_code/generate_api.py similarity index 100% rename from generate_api.py rename to source_code/generate_api.py diff --git a/html_crypto.py b/source_code/html_crypto.py similarity index 96% rename from html_crypto.py rename to source_code/html_crypto.py index cb089d9..482ff20 100644 --- a/html_crypto.py +++ b/source_code/html_crypto.py @@ -1,68 +1,68 @@ -""" -HTML 模板加密/解密模块 -使用 AES (Fernet) 对 HTML 资源进行加密存储，运行时解密后渲染 -""" - -import base64 -import os - -from cryptography.fernet import Fernet -from cryptography.hazmat.primitives import hashes -from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC - - -def _get_fernet_key() -> bytes: - """从环境变量 HTML_ENCRYPT_KEY 派生 Fernet 密钥，若未设置则使用默认开发密钥""" - raw = os.environ.get( - "HTML_ENCRYPT_KEY", - "maixiang_html_encrypt_default_key_change_in_production", - ) - # 使用 PBKDF2 派生 32 字节密钥，再转为 Fernet 所需的 base64url - kdf = PBKDF2HMAC( - algorithm=hashes.SHA256(), - length=32, - salt=b"maixiang_html_salt", - iterations=100000, - ) - key_bytes = kdf.derive(raw.encode("utf-8")) - return base64.urlsafe_b64encode(key_bytes) - - -def encrypt(plain_data: bytes) -> bytes: - """加密原始数据，返回密文（bytes）""" - key = _get_fernet_key() - f = Fernet(key) - return f.encrypt(plain_data) - - -def decrypt(encrypted_data: bytes) -> bytes: - """解密数据，返回明文（bytes）""" - key = _get_fernet_key() - f = Fernet(key) - return f.decrypt(encrypted_data) - - -def encrypt_file(path: str) -> None: - """就地加密文件：读取 UTF-8 内容，加密后写回同一路径""" - with open(path, "rb") as f: - plain = f.read() - encrypted = encrypt(plain) - with open(path, "wb") as f: - f.write(encrypted) - - -def decrypt_file(path: str) -> None: - """就地解密文件：读取密文，解密后以 UTF-8 写回同一路径""" - with open(path, "rb") as f: - encrypted = f.read() - plain = decrypt(encrypted) - with open(path, "wb") as f: - f.write(plain) - - -def is_encrypted(data: bytes) -> bool: - """简单启发式：Fernet 密文以 b'gAAAAA' 开头（base64 编码后）""" - try: - return data[:7] == b"gAAAAAB" - except Exception: - return False +""" +HTML 模板加密/解密模块 +使用 AES (Fernet) 对 HTML 资源进行加密存储，运行时解密后渲染 +""" + +import base64 +import os + +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC + + +def _get_fernet_key() -> bytes: + """从环境变量 HTML_ENCRYPT_KEY 派生 Fernet 密钥，若未设置则使用默认开发密钥""" + raw = os.environ.get( + "HTML_ENCRYPT_KEY", + "maixiang_html_encrypt_default_key_change_in_production", + ) + # 使用 PBKDF2 派生 32 字节密钥，再转为 Fernet 所需的 base64url + kdf = PBKDF2HMAC( + algorithm=hashes.SHA256(), + length=32, + salt=b"maixiang_html_salt", + iterations=100000, + ) + key_bytes = kdf.derive(raw.encode("utf-8")) + return base64.urlsafe_b64encode(key_bytes) + + +def encrypt(plain_data: bytes) -> bytes: + """加密原始数据，返回密文（bytes）""" + key = _get_fernet_key() + f = Fernet(key) + return f.encrypt(plain_data) + + +def decrypt(encrypted_data: bytes) -> bytes: + """解密数据，返回明文（bytes）""" + key = _get_fernet_key() + f = Fernet(key) + return f.decrypt(encrypted_data) + + +def encrypt_file(path: str) -> None: + """就地加密文件：读取 UTF-8 内容，加密后写回同一路径""" + with open(path, "rb") as f: + plain = f.read() + encrypted = encrypt(plain) + with open(path, "wb") as f: + f.write(encrypted) + + +def decrypt_file(path: str) -> None: + """就地解密文件：读取密文，解密后以 UTF-8 写回同一路径""" + with open(path, "rb") as f: + encrypted = f.read() + plain = decrypt(encrypted) + with open(path, "wb") as f: + f.write(plain) + + +def is_encrypted(data: bytes) -> bool: + """简单启发式：Fernet 密文以 b'gAAAAA' 开头（base64 编码后）""" + try: + return data[:7] == b"gAAAAAB" + except Exception: + return False diff --git a/main.py b/source_code/main.py similarity index 100% rename from main.py rename to source_code/main.py diff --git a/static/bg.jpg b/source_code/static/bg.jpg similarity index 100% rename from static/bg.jpg rename to source_code/static/bg.jpg diff --git a/static/品牌文档格式_模板.xlsx b/source_code/static/品牌文档格式_模板.xlsx similarity index 100% rename from static/品牌文档格式_模板.xlsx rename to source_code/static/品牌文档格式_模板.xlsx diff --git a/static/模板2-以文件夹方式上传.zip b/source_code/static/模板2-以文件夹方式上传.zip similarity index 100% rename from static/模板2-以文件夹方式上传.zip rename to source_code/static/模板2-以文件夹方式上传.zip diff --git a/tool/.device_id b/source_code/tool/.device_id similarity index 100% rename from tool/.device_id rename to source_code/tool/.device_id diff --git a/tool/__pycache__/devices.cpython-311.pyc b/source_code/tool/__pycache__/devices.cpython-311.pyc similarity index 100% rename from tool/__pycache__/devices.cpython-311.pyc rename to source_code/tool/__pycache__/devices.cpython-311.pyc diff --git a/tool/__pycache__/devices.cpython-39.pyc b/source_code/tool/__pycache__/devices.cpython-39.pyc similarity index 100% rename from tool/__pycache__/devices.cpython-39.pyc rename to source_code/tool/__pycache__/devices.cpython-39.pyc diff --git a/tool/devices.py b/source_code/tool/devices.py similarity index 96% rename from tool/devices.py rename to source_code/tool/devices.py index efdf582..acb0369 100644 --- a/tool/devices.py +++ b/source_code/tool/devices.py @@ -1,249 +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__": +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() \ No newline at end of file diff --git a/source_code/web_source/brand_tools.html b/source_code/web_source/brand_tools.html new file mode 100644 index 0000000..bd15d2a --- /dev/null +++ b/source_code/web_source/brand_tools.html @@ -0,0 +1,14 @@ + + + + + + + 品牌工具 - 数富AI + + + + +