192 lines
7.9 KiB
Python
192 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
"""蝶雅采购智能体平台 - 蝶雅数据可视化平台
|
||
|
||
精简版 app.py：只保留 Flask 初始化、Blueprint 注册、静态文件路由、配置
|
||
"""
|
||
import os
|
||
from flask import Flask, request, send_file, render_template, redirect
|
||
from flask_compress import Compress
|
||
from config import SECRET_KEY, APP_PORT, APP_HOST
|
||
|
||
from routes import register_blueprints
|
||
from services.db import query
|
||
from services.audit import init_audit
|
||
|
||
app = Flask(__name__)
|
||
Compress(app)
|
||
app.secret_key = SECRET_KEY
|
||
app.config['JSON_AS_ASCII'] = False
|
||
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
|
||
app.jinja_env.auto_reload = True
|
||
app.jinja_env.cache.clear()
|
||
|
||
import os, time
|
||
@app.context_processor
|
||
def inject_build_version():
|
||
mtime = int(os.path.getmtime(os.path.join(app.static_folder, 'js', 'app.js')))
|
||
return {'app_js_version': mtime}
|
||
|
||
# 注册所有 Blueprint
|
||
register_blueprints(app)
|
||
|
||
# 注册集中式用户数据变更审计（before_request + after_request 钩子）
|
||
init_audit(app)
|
||
|
||
# 调试接口：直接查看下单前包材数据（绕过模板缓存）
|
||
|
||
|
||
# Gzip 静态文件支持
|
||
@app.route('/static/<path:filename>')
|
||
def static_with_gzip(filename):
|
||
filepath = os.path.join(app.static_folder, filename)
|
||
gz_path = filepath + '.gz'
|
||
accept_encoding = request.headers.get('Accept-Encoding', '')
|
||
if 'gzip' in accept_encoding and os.path.exists(gz_path):
|
||
return send_file(
|
||
gz_path,
|
||
mimetype='application/javascript' if filename.endswith('.js') else None,
|
||
as_attachment=False, download_name=filename,
|
||
conditional=True
|
||
), {'Content-Encoding': 'gzip'}
|
||
return app.send_static_file(filename)
|
||
|
||
|
||
@app.route('/feedback')
|
||
def removed_feedback():
|
||
"""测试问题板块已下线，旧地址回到数据概览。"""
|
||
return redirect('/dashboard')
|
||
|
||
|
||
# 上下文处理器：注入当前用户权限到所有模板
|
||
@app.context_processor
|
||
def inject_user_permissions():
|
||
from flask import session
|
||
user_permissions = {}
|
||
user_id = session.get('user_id')
|
||
role = session.get('role')
|
||
if user_id and role != 'super_admin':
|
||
try:
|
||
role_perms = query(
|
||
"SELECT page, operation, permission FROM role_operation_permissions WHERE role=%s",
|
||
('viewer' if role == '查看者' else role,)
|
||
)
|
||
from services.permission_model import ensure_position_permissions, user_position_code
|
||
ensure_position_permissions()
|
||
position = user_position_code(user_id)
|
||
position_perms = query(
|
||
"SELECT page, operation, permission FROM position_operation_permissions WHERE position_code=%s",
|
||
(position,)
|
||
) if position else []
|
||
user_perms = query(
|
||
"SELECT page, operation, permission FROM user_permissions WHERE user_id=%s",
|
||
(user_id,)
|
||
)
|
||
runtime_page_aliases = {
|
||
'purchase_pending': 'purchase', 'purchase_backup': 'purchase_ready',
|
||
'order_domestic': 'orders_domestic', 'order_overseas': 'orders_overseas',
|
||
'order_packaging': 'orders_packaging', 'order_finished': 'orders_finished'
|
||
}
|
||
for p in role_perms:
|
||
runtime_page = runtime_page_aliases.get(p['page'], p['page'])
|
||
user_permissions.setdefault(runtime_page, {})[p['operation']] = int(p['permission'])
|
||
# 岗位权限覆盖角色默认；用户级记录再覆盖岗位权限。
|
||
for p in position_perms:
|
||
runtime_page = runtime_page_aliases.get(p['page'], p['page'])
|
||
user_permissions.setdefault(runtime_page, {})[p['operation']] = int(p['permission'])
|
||
for p in user_perms:
|
||
runtime_page = runtime_page_aliases.get(p['page'], p['page'])
|
||
user_permissions.setdefault(runtime_page, {})[p['operation']] = int(p['permission'])
|
||
except Exception:
|
||
pass
|
||
# 兼容旧版页面权限表：将旧页面权限转换为统一 view 权限。
|
||
legacy_aliases = {
|
||
'data': ['data_summary', 'data_detail'],
|
||
'inventory': ['inventory_finished', 'inventory_packaging'],
|
||
'purchase': ['purchase', 'purchase_ready'],
|
||
'orders': ['orders_domestic', 'orders_overseas', 'orders_packaging', 'orders_finished'],
|
||
'login-records': ['login_records']
|
||
}
|
||
try:
|
||
legacy = query("SELECT page, permission FROM role_permissions WHERE role=%s AND permission=1", (role,))
|
||
for p in legacy:
|
||
for page_name in legacy_aliases.get(p['page'], [p['page']]):
|
||
user_permissions.setdefault(page_name, {}).setdefault('view', 1)
|
||
except Exception:
|
||
pass
|
||
# 注入字段级权限显式值（敏感字段默认隐藏 + 角色/用户字段权限）
|
||
if user_id and role not in ('super_admin',):
|
||
try:
|
||
from services.field_perms import effective_field_overrides
|
||
for runtime_page, field_map in effective_field_overrides(user_id, role).items():
|
||
base = user_permissions.setdefault(runtime_page, {})
|
||
for field_key, ops in field_map.items():
|
||
base['field:' + field_key + ':view'] = ops['view']
|
||
base['field:' + field_key + ':edit'] = ops['edit']
|
||
except Exception:
|
||
pass
|
||
# 用 JSON 序列化传给前端
|
||
import json
|
||
return {'_user_permissions_json': json.dumps(user_permissions, ensure_ascii=False)}
|
||
|
||
|
||
# SPA 路由：处理前端路由刷新时的 404
|
||
@app.route('/<path:path>')
|
||
def spa_fallback(path):
|
||
from flask import session, redirect, url_for
|
||
# 仅处理已知的前端 SPA 路径，避免覆盖 API/静态路由
|
||
SPA_PATHS = {
|
||
'knowledge', 'planning', 'products', 'bom', 'purchase',
|
||
'orders', 'inventory', 'warehouse', 'suppliers',
|
||
'users', 'data', 'config', 'logs', 'permissions',
|
||
'login-records',
|
||
'contract-templates', 'personnel',
|
||
'collaboration', 'spot', 'ai-analysis', 'product-order',
|
||
'stagnant-inventory', 'stagnant-inventory/dashboard',
|
||
'stagnant-inventory/detail', 'stagnant-inventory/plans',
|
||
'stagnant-inventory/lists',
|
||
'spot/dashboard', 'spot/application', 'spot/confirmation', 'spot/sales',
|
||
'spot/authorization',
|
||
'product-new', 'product-category', 'brand-management',
|
||
'product-packaging',
|
||
'collaboration/inventory-dashboard', 'collaboration/key-sku-forecast',
|
||
'collaboration/sku-health-index', 'collaboration/slow-moving-sku',
|
||
'collaboration/sales-management'
|
||
}
|
||
first = path.split('/')[0]
|
||
if first not in SPA_PATHS:
|
||
return redirect(url_for('auth.login'))
|
||
if 'user_id' not in session:
|
||
return redirect(url_for('auth.login'))
|
||
if session.get('must_change_password'):
|
||
return redirect(url_for('auth.login'))
|
||
# 用完整路径映射 page key（/collaboration/inventory-dashboard → collaboration_inventory_dashboard），
|
||
# 不能用第一段，否则刷新子页面会回退到父菜单占位页
|
||
return render_template('base.html', page=path.replace('/', '_').replace('-', '_'))
|
||
|
||
|
||
@app.after_request
|
||
def add_header(response):
|
||
# 静态文件：长期缓存（7天），配合版本号更新
|
||
if request.path.startswith('/static/'):
|
||
response.headers['Cache-Control'] = 'no-cache, must-revalidate'
|
||
else:
|
||
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, post-check=0, pre-check=0, max-age=0'
|
||
response.headers['Pragma'] = 'no-cache'
|
||
response.headers['Expires'] = '0'
|
||
return response
|
||
|
||
|
||
# ─── 启动 ───
|
||
if __name__ == '__main__':
|
||
# 启动时自动清洗脏数据
|
||
try:
|
||
import auto_clean
|
||
auto_clean.clean()
|
||
except:
|
||
pass
|
||
|
||
print(f"🚀 蝶雅采购智能体平台 启动中...")
|
||
print(f" http://{APP_HOST}:{APP_PORT}")
|
||
print(f" 默认账号: admin / admin123")
|
||
app.run(host=APP_HOST, port=APP_PORT, debug=True)
|