bf1942
8938ce2708
- 统一移除手动创建的数据库session，统一使用models模块中的db.session - 修正项目创建接口，增加开始和结束日期的格式验证与处理 - 更新导入项目接口，使用枚举类型校验项目类型并优化异常处理 - 更新统计接口，避免多次查询假期数据，优化日期字符串处理 - 删除回滚前多余的session关闭调用，改为使用db.session.rollback() - app.py中重构数据库初始化：统一配置SQLAlchemy，动态创建数据库路径和表 - 项目模型新增开始日期和结束日期字段支持 - 添加导入批次历史记录模型支持 - 优化工具函数中日期类型提示，移除无用导入 - 更新requirements.txt依赖版本回退，确保兼容性 - 前端菜单添加导入历史导航入口，实现页面访问路由绑定
254 lines
9.8 KiB
Python
254 lines
9.8 KiB
Python
from flask import Blueprint, request, jsonify
|
||
from models.models import db, Project, ProjectType
|
||
from models.utils import *
|
||
import csv
|
||
import io
|
||
from datetime import datetime
|
||
|
||
projects_bp = Blueprint('projects', __name__)
|
||
|
||
@projects_bp.route('/api/projects', methods=['GET'])
|
||
def get_projects():
|
||
"""获取所有项目列表"""
|
||
try:
|
||
projects = db.session.query(Project).filter_by(is_active=True).all()
|
||
|
||
result = []
|
||
for project in projects:
|
||
project_dict = project.to_dict()
|
||
# 为前端显示格式化项目信息
|
||
if project.project_type == ProjectType.TRADITIONAL:
|
||
project_dict['display_name'] = f"{project.project_name} ({project.customer_name}-{project.project_code})"
|
||
else: # PSI项目
|
||
project_dict['display_name'] = f"{project.project_name} ({project.customer_name}-{project.contract_number})"
|
||
|
||
result.append(project_dict)
|
||
|
||
return jsonify({'success': True, 'data': result})
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
return jsonify({'success': False, 'error': str(e)}), 500
|
||
|
||
@projects_bp.route('/api/projects', methods=['POST'])
|
||
def create_project():
|
||
"""创建新项目"""
|
||
try:
|
||
data = request.json
|
||
|
||
# 验证必填字段
|
||
if not data.get('project_name') or not data.get('customer_name') or not data.get('project_type'):
|
||
return jsonify({'success': False, 'error': '项目名称、客户名和项目类型为必填项'}), 400
|
||
|
||
project_type_str = data['project_type']
|
||
try:
|
||
project_type = ProjectType(project_type_str)
|
||
except ValueError:
|
||
return jsonify({'success': False, 'error': f'无效的项目类型: {project_type_str}'}), 400
|
||
|
||
# 根据项目类型设置字段
|
||
if project_type == ProjectType.TRADITIONAL:
|
||
if not data.get('project_code'):
|
||
return jsonify({'success': False, 'error': '传统项目需要填写项目代码'}), 400
|
||
project_code = data['project_code']
|
||
contract_number = None
|
||
else: # PSI项目
|
||
if not data.get('contract_number'):
|
||
return jsonify({'success': False, 'error': 'PSI项目需要填写合同号'}), 400
|
||
project_code = 'PSI-PROJ' # PSI项目统一代码
|
||
contract_number = data['contract_number']
|
||
|
||
# 处理结束日期
|
||
end_date = None
|
||
if data.get('end_date') and data.get('end_date') != '' :
|
||
try:
|
||
end_date = datetime.strptime(data['end_date'], '%Y-%m-%d').date()
|
||
except ValueError:
|
||
return jsonify({'success': False, 'error': '结束日期格式不正确，应为 YYYY-MM-DD'}), 400
|
||
|
||
# 处理开始日期
|
||
start_date = None
|
||
if data.get('start_date') and data.get('start_date') != '' :
|
||
try:
|
||
start_date = datetime.strptime(data['start_date'], '%Y-%m-%d').date()
|
||
except ValueError:
|
||
return jsonify({'success': False, 'error': '开始日期格式不正确，应为 YYYY-MM-DD'}), 400
|
||
|
||
# 检查唯一性约束
|
||
existing_project = None
|
||
if project_type == ProjectType.TRADITIONAL:
|
||
# 传统项目：客户名+项目代码唯一
|
||
existing_project = db.session.query(Project).filter_by(
|
||
customer_name=data['customer_name'],
|
||
project_code=project_code,
|
||
project_type=ProjectType.TRADITIONAL
|
||
).first()
|
||
else:
|
||
# PSI项目：客户名+合同号唯一
|
||
existing_project = db.session.query(Project).filter_by(
|
||
customer_name=data['customer_name'],
|
||
contract_number=contract_number,
|
||
project_type=ProjectType.PSI
|
||
).first()
|
||
|
||
if existing_project:
|
||
return jsonify({'success': False, 'error': '项目已存在'}), 400
|
||
|
||
# 创建新项目
|
||
project = Project(
|
||
project_name=data['project_name'],
|
||
project_type=project_type,
|
||
project_code=project_code,
|
||
customer_name=data['customer_name'],
|
||
contract_number=contract_number,
|
||
description=data.get('description', ''),
|
||
start_date=start_date,
|
||
end_date=end_date
|
||
)
|
||
|
||
db.session.add(project)
|
||
db.session.commit()
|
||
|
||
result = project.to_dict()
|
||
|
||
return jsonify({'success': True, 'data': result})
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
return jsonify({'success': False, 'error': str(e)}), 500
|
||
|
||
@projects_bp.route('/api/projects/<int:project_id>', methods=['PUT'])
|
||
def update_project(project_id):
|
||
"""更新项目信息"""
|
||
try:
|
||
data = request.json
|
||
|
||
project = db.session.query(Project).get(project_id)
|
||
if not project:
|
||
return jsonify({'success': False, 'error': '项目不存在'}), 404
|
||
|
||
# 更新字段
|
||
if 'project_name' in data:
|
||
project.project_name = data['project_name']
|
||
if 'description' in data:
|
||
project.description = data['description']
|
||
|
||
db.session.commit()
|
||
result = project.to_dict()
|
||
|
||
return jsonify({'success': True, 'data': result})
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
return jsonify({'success': False, 'error': str(e)}), 500
|
||
|
||
@projects_bp.route('/api/projects/<int:project_id>', methods=['DELETE'])
|
||
def delete_project(project_id):
|
||
"""删除项目（软删除）"""
|
||
try:
|
||
project = db.session.query(Project).get(project_id)
|
||
if not project:
|
||
return jsonify({'success': False, 'error': '项目不存在'}), 404
|
||
|
||
project.is_active = False
|
||
db.session.commit()
|
||
|
||
return jsonify({'success': True, 'message': '项目已删除'})
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
return jsonify({'success': False, 'error': str(e)}), 500
|
||
|
||
@projects_bp.route('/api/projects/import', methods=['POST'])
|
||
def import_projects():
|
||
"""批量导入项目"""
|
||
try:
|
||
if 'file' not in request.files:
|
||
return jsonify({'success': False, 'error': '请选择CSV文件'}), 400
|
||
|
||
file = request.files['file']
|
||
if file.filename == '' or not file.filename.endswith('.csv'):
|
||
return jsonify({'success': False, 'error': '请选择有效的CSV文件'}), 400
|
||
|
||
# 读取CSV文件
|
||
stream = io.StringIO(file.stream.read().decode("utf-8"))
|
||
csv_reader = csv.DictReader(stream)
|
||
|
||
created_count = 0
|
||
errors = []
|
||
|
||
for row_num, row in enumerate(csv_reader, start=2):
|
||
try:
|
||
# 验证必填字段
|
||
if not row.get('项目名称') or not row.get('客户名') or not row.get('项目类型'):
|
||
errors.append(f"第{row_num}行：项目名称、客户名和项目类型为必填项")
|
||
continue
|
||
|
||
project_type_str = row['项目类型'].lower()
|
||
try:
|
||
project_type = ProjectType(project_type_str)
|
||
except ValueError:
|
||
errors.append(f"第{row_num}行：项目类型只能是 traditional 或 psi")
|
||
continue
|
||
|
||
# 根据项目类型设置字段
|
||
if project_type == ProjectType.TRADITIONAL:
|
||
if not row.get('项目代码'):
|
||
errors.append(f"第{row_num}行：传统项目需要填写项目代码")
|
||
continue
|
||
project_code = row['项目代码']
|
||
contract_number = None
|
||
else: # PSI项目
|
||
if not row.get('合同号'):
|
||
errors.append(f"第{row_num}行：PSI项目需要填写合同号")
|
||
continue
|
||
project_code = 'PSI-PROJ'
|
||
contract_number = row['合同号']
|
||
|
||
# 检查重复
|
||
existing_project = None
|
||
if project_type == ProjectType.TRADITIONAL:
|
||
existing_project = db.session.query(Project).filter_by(
|
||
customer_name=row['客户名'],
|
||
project_code=project_code,
|
||
project_type=ProjectType.TRADITIONAL
|
||
).first()
|
||
else:
|
||
existing_project = db.session.query(Project).filter_by(
|
||
customer_name=row['客户名'],
|
||
contract_number=contract_number,
|
||
project_type=ProjectType.PSI
|
||
).first()
|
||
|
||
if existing_project:
|
||
errors.append(f"第{row_num}行：项目已存在")
|
||
continue
|
||
|
||
# 创建项目
|
||
project = Project(
|
||
project_name=row['项目名称'],
|
||
project_type=project_type,
|
||
project_code=project_code,
|
||
customer_name=row['客户名'],
|
||
contract_number=contract_number,
|
||
description=row.get('描述', '')
|
||
)
|
||
|
||
db.session.add(project)
|
||
created_count += 1
|
||
|
||
except Exception as e:
|
||
errors.append(f"第{row_num}行：{str(e)}")
|
||
|
||
db.session.commit()
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': f'成功导入{created_count}个项目',
|
||
'created_count': created_count,
|
||
'errors': errors
|
||
})
|
||
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
return jsonify({'success': False, 'error': str(e)}), 500 |