from flask import Blueprint, request, jsonify from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from backend.models.models import Project, ProjectType from backend.models.utils import * import csv import io projects_bp = Blueprint('projects', __name__) def get_db_session(): """获取数据库会话""" engine = create_engine('sqlite:///data/timetrack.db') Session = sessionmaker(bind=engine) return Session() @projects_bp.route('/api/projects', methods=['GET']) def get_projects(): """获取所有项目列表""" try: session = get_db_session() projects = 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) session.close() return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @projects_bp.route('/api/projects', methods=['POST']) def create_project(): """创建新项目""" try: data = request.json session = get_db_session() # 验证必填字段 if not data.get('project_name') or not data.get('customer_name') or not data.get('project_type'): return jsonify({'success': False, 'error': '项目名称、客户名和项目类型为必填项'}), 400 # 根据项目类型设置字段 if data['project_type'] == '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'] # 检查唯一性约束 existing_project = None if data['project_type'] == 'traditional': # 传统项目：客户名+项目代码唯一 existing_project = session.query(Project).filter_by( customer_name=data['customer_name'], project_code=project_code, project_type=ProjectType.TRADITIONAL ).first() else: # PSI项目：客户名+合同号唯一 existing_project = session.query(Project).filter_by( customer_name=data['customer_name'], contract_number=contract_number, project_type=ProjectType.PSI ).first() if existing_project: session.close() return jsonify({'success': False, 'error': '项目已存在'}), 400 # 创建新项目 project = Project( project_name=data['project_name'], project_type=ProjectType(data['project_type']), project_code=project_code, customer_name=data['customer_name'], contract_number=contract_number, description=data.get('description', '') ) session.add(project) session.commit() result = project.to_dict() session.close() return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @projects_bp.route('/api/projects/', methods=['PUT']) def update_project(project_id): """更新项目信息""" try: data = request.json session = get_db_session() project = session.query(Project).get(project_id) if not project: session.close() 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'] session.commit() result = project.to_dict() session.close() return jsonify({'success': True, 'data': result}) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @projects_bp.route('/api/projects/', methods=['DELETE']) def delete_project(project_id): """删除项目（软删除）""" try: session = get_db_session() project = session.query(Project).get(project_id) if not project: session.close() return jsonify({'success': False, 'error': '项目不存在'}), 404 project.is_active = False session.commit() session.close() return jsonify({'success': True, 'message': '项目已删除'}) except Exception as e: 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 session = get_db_session() # 读取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 = row['项目类型'].lower() if project_type not in ['traditional', 'psi']: errors.append(f"第{row_num}行：项目类型只能是 traditional 或 psi") continue # 根据项目类型设置字段 if project_type == '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 == 'traditional': existing_project = session.query(Project).filter_by( customer_name=row['客户名'], project_code=project_code, project_type=ProjectType.TRADITIONAL ).first() else: existing_project = 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=ProjectType(project_type), project_code=project_code, customer_name=row['客户名'], contract_number=contract_number, description=row.get('描述', '') ) session.add(project) created_count += 1 except Exception as e: errors.append(f"第{row_num}行：{str(e)}") session.commit() session.close() return jsonify({ 'success': True, 'message': f'成功导入{created_count}个项目', 'created_count': created_count, 'errors': errors }) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500