""" 项目与操作日志数据层（依据 schema.sql、API 文档第 8 节） API 使用 camelCase，数据库使用 snake_case，在此做映射。 """ import re from datetime import datetime from typing import Any, Optional from .database import get_connection, new_id DATE_STR_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$") # API 字段名(camelCase) -> DB 列名(snake_case)，按 5 个分类 CONTRACT_API_TO_DB = { "serialNo": "serial_no", "contractCode": "contract_code", "powerBureauContractCode": "power_bureau_contract_code", "projectName": "project_name", "subItemCount": "sub_item_count", "subItemCode": "sub_item_code", "totalInvestment": "total_investment", "bidContractAmount": "bid_contract_amount", "warrantyRatio": "warranty_ratio", "settlementAmount": "settlement_amount", "totalCostEstimate": "total_cost_estimate", "voltageLevel": "voltage_level", "projectCategory": "project_category", "ownerUnit": "owner_unit", "ownerContact": "owner_contact", "bidType": "bid_type", "signDate": "sign_date", "startDate": "start_date", "plannedCompletionDate": "planned_completion_date", "actualCompletionDate": "actual_completion_date", "warrantyAmount": "warranty_amount", "warrantyEndDate": "warranty_end_date", "actualWarrantyRefundDate": "actual_warranty_refund_date", "projectDepartment": "project_department", "projectLeaderContact": "project_leader_contact", "paymentMethod": "payment_method", } COST_CONTROL_API_TO_DB = { "totalCost": "total_cost", "isAdjusted": "is_adjusted", "migrantWorkerPlan": "migrant_worker_plan", "migrantWorkerActual": "migrant_worker_actual", "selfSupplyMaterialControl": "self_supply_material_control", "materialPayableByRatio": "material_payable_by_ratio", "materialActualOccurred": "material_actual_occurred", "materialActualPaid": "material_actual_paid", "otherCostControl": "other_cost_control", "otherPayable": "other_payable", "otherActual": "other_actual", "tax": "tax", "profit": "profit", "actualProfit": "actual_profit", "costSettlementAmount": "cost_settlement_amount", } RECEIVABLE_API_TO_DB = { "receivableProgress": "receivable_progress", "invoiceAmount": "invoice_amount", "actualReceiptAmount": "actual_receipt_amount", "actualReceiptRate": "actual_receipt_rate", } PAYABLE_API_TO_DB = { "payableAmount": "payable_amount", "actualPaymentAmount": "actual_payment_amount", "unreceivedAmount": "unreceived_amount", "actualPaymentRate": "actual_payment_rate", "migrantWorkerArrears": "migrant_worker_arrears", } OTHER_API_TO_DB = { "settlementCostEstimate": "settlement_cost_estimate", "settlementLaborCost": "settlement_labor_cost", "settlementMaterialCost": "settlement_material_cost", "settlementOtherCost": "settlement_other_cost", "dueSettlementCount": "due_settlement_count", "overdueSettlementCount": "overdue_settlement_count", "existingProblems": "existing_problems", "suggestions": "suggestions", "cumulativeProgress": "cumulative_progress", "remark": "remark", } # 四个日期字段：API 返回时若 _str 为「日期正常」则返回日期，否则返回 _str DATE_FIELDS_WITH_STR = ("sign_date", "start_date", "planned_completion_date", "actual_completion_date") DATE_NORMAL = "日期正常" def _db_to_camel(d: dict[str, Any], api_to_db: dict[str, str]) -> dict[str, Any]: """DB 列名 -> API 字段名（反向映射）""" db_to_api = {v: k for k, v in api_to_db.items()} return {db_to_api.get(k, k): v for k, v in d.items() if k in db_to_api} def _api_to_db(d: Optional[dict], api_to_db: dict[str, str]) -> dict[str, Any]: """API 请求体（camelCase）-> DB 列（snake_case），仅包含有值的键""" if not d: return {} return {api_to_db[k]: v for k, v in d.items() if k in api_to_db and v is not None} def _normalize_contract_date_str_fields(c: dict[str, Any]) -> None: """对合同中的四个日期字段做规范化：日期则写 DATE + _str=日期正常，否则只写 _str。原地修改 c。""" for col in DATE_FIELDS_WITH_STR: if col not in c: continue val = c[col] s = (val if isinstance(val, str) else str(val)).strip() if val else "" if s and DATE_STR_PATTERN.match(s): c[col] = s c[col + "_str"] = DATE_NORMAL else: c[col + "_str"] = s or "" c[col] = None def _row_to_api_project(row: dict[str, Any]) -> dict[str, Any]: """数据库一行 -> API 项目详情（含 contract/costControl/receivable/payable/other）""" contract_cols = set(CONTRACT_API_TO_DB.values()) cost_cols = set(COST_CONTROL_API_TO_DB.values()) recv_cols = set(RECEIVABLE_API_TO_DB.values()) pay_cols = set(PAYABLE_API_TO_DB.values()) other_cols = set(OTHER_API_TO_DB.values()) contract = {k: v for k, v in row.items() if k in contract_cols} for col in DATE_FIELDS_WITH_STR: if row.get(col + "_str") == DATE_NORMAL and row.get(col): contract[col] = row[col] else: contract[col] = row.get(col + "_str") or "" cost_control = {k: v for k, v in row.items() if k in cost_cols} receivable = {k: v for k, v in row.items() if k in recv_cols} payable = {k: v for k, v in row.items() if k in pay_cols} other = {k: v for k, v in row.items() if k in other_cols} # 日期等转为 ISO 字符串 def _serialize(v: Any) -> Any: if hasattr(v, "isoformat"): return v.isoformat() return v return { "id": row.get("id"), "contract": _db_to_camel({k: _serialize(v) for k, v in contract.items()}, CONTRACT_API_TO_DB), "costControl": _db_to_camel({k: _serialize(v) for k, v in cost_control.items()}, COST_CONTROL_API_TO_DB), "receivable": _db_to_camel({k: _serialize(v) for k, v in receivable.items()}, RECEIVABLE_API_TO_DB), "payable": _db_to_camel({k: _serialize(v) for k, v in payable.items()}, PAYABLE_API_TO_DB), "other": _db_to_camel({k: _serialize(v) for k, v in other.items()}, OTHER_API_TO_DB), "createdAt": _serialize(row.get("created_at")), "updatedAt": _serialize(row.get("updated_at")), } # 时间筛选项 API -> DB 列名（仅对「日期正常」的项目按 DATE 列筛选） DATE_FILTER_COLUMNS = { "signDate": "sign_date", "startDate": "start_date", "plannedCompletionDate": "planned_completion_date", "actualCompletionDate": "actual_completion_date", } def list_projects( page: int = 1, page_size: int = 50, search_type: Optional[str] = None, keyword: Optional[str] = None, progress: Optional[str] = None, cost: Optional[str] = None, date_filter_type: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, date_abnormal: Optional[bool] = None, ) -> tuple[list[dict], int]: """分页列表，支持按项目名称/合同编号搜索、按进度/费用筛选、时间筛选、日期异常筛选。返回 (list, total)。""" with get_connection() as conn: cur = conn.cursor(dictionary=True) where_parts = [] params: list[Any] = [] if keyword and keyword.strip(): if search_type == "name": where_parts.append("project_name LIKE %s") params.append(f"%{keyword.strip()}%") elif search_type == "code": where_parts.append("contract_code LIKE %s") params.append(f"%{keyword.strip()}%") else: where_parts.append("(project_name LIKE %s OR contract_code LIKE %s)") params.extend([f"%{keyword.strip()}%", f"%{keyword.strip()}%"]) if progress and progress.strip(): where_parts.append("progress = %s") params.append(progress.strip()) if cost and cost.strip(): where_parts.append("cost = %s") params.append(cost.strip()) if date_abnormal: where_parts.append( "((sign_date_str IS NULL OR sign_date_str != %s) OR (start_date_str IS NULL OR start_date_str != %s) " "OR (planned_completion_date_str IS NULL OR planned_completion_date_str != %s) " "OR (actual_completion_date_str IS NULL OR actual_completion_date_str != %s))" ) params.extend([DATE_NORMAL, DATE_NORMAL, DATE_NORMAL, DATE_NORMAL]) date_col = DATE_FILTER_COLUMNS.get(date_filter_type) if date_filter_type else None if date_col and (date_from or date_to) and not date_abnormal: where_parts.append(f"({date_col}_str = %s)") params.append(DATE_NORMAL) if date_from and date_from.strip(): where_parts.append(f"{date_col} >= %s") params.append(date_from.strip()) if date_to and date_to.strip(): where_parts.append(f"{date_col} <= %s") params.append(date_to.strip()) where_sql = " AND ".join(where_parts) if where_parts else "1=1" cur.execute( f"SELECT COUNT(*) AS cnt FROM projects WHERE {where_sql}", tuple(params), ) total = cur.fetchone()["cnt"] offset = (page - 1) * page_size cur.execute( f"SELECT id, project_name, contract_code, progress, cost, updated_at FROM projects WHERE {where_sql} ORDER BY updated_at DESC LIMIT %s OFFSET %s", tuple(params) + (page_size, offset), ) rows = cur.fetchall() cur.close() list_ = [ { "id": r["id"], "projectName": r["project_name"] or "", "contractCode": r["contract_code"] or "", "progress": r["progress"] or "", "cost": r["cost"] or "", "updatedAt": r["updated_at"].isoformat() if r.get("updated_at") else "", } for r in rows ] return list_, total def get_project(project_id: str) -> Optional[dict[str, Any]]: """按 id 查项目，返回 API 形态详情，不存在返回 None。""" with get_connection() as conn: cur = conn.cursor(dictionary=True) cur.execute("SELECT * FROM projects WHERE id = %s", (project_id,)) row = cur.fetchone() cur.close() if not row: return None return _row_to_api_project(row) def create_project( contract: dict, cost_control: Optional[dict] = None, receivable: Optional[dict] = None, payable: Optional[dict] = None, other: Optional[dict] = None, ) -> str: """新建项目，合同信息必含 contract_code、project_name。返回新项目 id。""" pid = new_id() now = datetime.utcnow() c = _api_to_db(contract, CONTRACT_API_TO_DB) _normalize_contract_date_str_fields(c) cc = _api_to_db(cost_control, COST_CONTROL_API_TO_DB) recv = _api_to_db(receivable, RECEIVABLE_API_TO_DB) pay = _api_to_db(payable, PAYABLE_API_TO_DB) oth = _api_to_db(other, OTHER_API_TO_DB) row: dict[str, Any] = {**c, **cc, **recv, **pay, **oth} row["id"] = pid row["progress"] = (row.get("cumulative_progress") or row.get("project_name") or "")[:32] if row.get("cumulative_progress") or row.get("project_name") else "" row["cost"] = "" row["created_at"] = now row["updated_at"] = now cols = [k for k in row.keys()] vals = [row[k] for k in cols] placeholders = ", ".join(["%s"] * len(cols)) with get_connection() as conn: cur = conn.cursor() cur.execute( f"INSERT INTO projects ({', '.join(cols)}) VALUES ({placeholders})", vals, ) cur.close() return pid def update_project( project_id: str, contract: Optional[dict] = None, cost_control: Optional[dict] = None, receivable: Optional[dict] = None, payable: Optional[dict] = None, other: Optional[dict] = None, ) -> bool: """更新项目（合并传入的字段），不存在返回 False。""" c = _api_to_db(contract, CONTRACT_API_TO_DB) _normalize_contract_date_str_fields(c) cc = _api_to_db(cost_control, COST_CONTROL_API_TO_DB) recv = _api_to_db(receivable, RECEIVABLE_API_TO_DB) pay = _api_to_db(payable, PAYABLE_API_TO_DB) oth = _api_to_db(other, OTHER_API_TO_DB) merged = {**c, **cc, **recv, **pay, **oth} if not merged: return project_exists(project_id) merged["updated_at"] = datetime.utcnow() updates = [f"`{k}` = %s" for k in merged.keys()] params: list[Any] = list(merged.values()) + [project_id] with get_connection() as conn: cur = conn.cursor() cur.execute( "UPDATE projects SET " + ", ".join(updates) + " WHERE id = %s", params, ) cur.close() return True def project_exists(project_id: str) -> bool: """项目是否存在""" with get_connection() as conn: cur = conn.cursor() cur.execute("SELECT 1 FROM projects WHERE id = %s", (project_id,)) ok = cur.fetchone() is not None cur.close() return ok def insert_operation_log( project_id: str, operator_id: str, summary: str = "", detail: Optional[str] = None, ) -> None: """写入一条操作日志。""" log_id = new_id() with get_connection() as conn: cur = conn.cursor() cur.execute( "INSERT INTO operation_logs (id, project_id, operator_id, summary, detail) VALUES (%s, %s, %s, %s, %s)", (log_id, project_id, operator_id, summary or "项目信息", detail), ) cur.close() def list_operation_logs( project_id: str, page: int = 1, page_size: int = 20, ) -> tuple[list[dict], int]: """分页查询某项目的操作日志，返回 (list, total)。含 operator_name（联表 users）。""" with get_connection() as conn: cur = conn.cursor(dictionary=True) cur.execute( "SELECT COUNT(*) AS cnt FROM operation_logs WHERE project_id = %s", (project_id,), ) total = cur.fetchone()["cnt"] offset = (page - 1) * page_size cur.execute( """ SELECT l.id, l.operator_id, l.operated_at, l.summary, l.detail, u.display_name AS operator_name FROM operation_logs l LEFT JOIN users u ON l.operator_id = u.id WHERE l.project_id = %s ORDER BY l.operated_at DESC LIMIT %s OFFSET %s """, (project_id, page_size, offset), ) rows = cur.fetchall() cur.close() list_ = [ { "id": r["id"], "operatorId": r["operator_id"], "operatorName": r["operator_name"] or r["operator_id"], "operatedAt": r["operated_at"].isoformat() if r.get("operated_at") else "", "summary": r["summary"] or "", "detail": r.get("detail"), } for r in rows ] return list_, total