feat: 新增設備維修查詢工具與修復 AJAX 認證重定向問題

設備維修查詢工具:
- 新增 job_query_routes.py 與 job_query_service.py 提供工單查詢 API
- 新增 SQL 查詢檔案 (job_list, job_txn_detail, job_txn_export)
- 新增 job_query.html 前端頁面支援設備選擇、日期範圍查詢與 CSV 匯出
- 整合 portal.html 導航與 page_status.json 頁面註冊
- 新增完整測試 (test_job_query_routes.py, test_job_query_service.py)

AJAX 認證修復:
- 修復 admin 路由對 AJAX 請求返回 302 導致前端卡住的問題
- 新增 _is_ajax_request() 偵測函式於 permissions.py
- 修改 admin_required 裝飾器對 AJAX 請求返回 JSON 401
- 修改 app.py before_request 鉤子支援 AJAX 認證失敗處理
- 更新 performance.html 使用 fetchWithAuth() 處理 401 重定向

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
beabigegg
2026-02-04 15:45:03 +08:00
parent dd0ae3ee54
commit ed60701465
20 changed files with 2523 additions and 13 deletions

View File

@@ -39,6 +39,11 @@
"route": "/excel-query",
"name": "Excel 批次查詢",
"status": "dev"
},
{
"route": "/job-query",
"name": "設備維修查詢",
"status": "released"
}
],
"api_public": true,

View File

@@ -0,0 +1,2 @@
schema: spec-driven
created: 2026-02-04

View File

@@ -0,0 +1,114 @@
## Context
目前系統已有設備即時監控 (resource dashboard) 和設備歷史分析 (resource history) 功能,皆使用 `resource_cache` 作為設備主數據來源。維修工單資料存在於:
- `DW_MES_JOB` (~125萬筆) - 工單現況,有 RESOURCEID 可關聯設備
- `DW_MES_JOBTXNHISTORY` (~955萬筆) - 工單交易歷史,透過 JOBID 關聯
現有架構已建立的模式:
- SQL 檔案集中管理於 `sql/<module>/*.sql`
- 使用 `{{ PLACEHOLDER }}``:param` 參數化查詢
- `read_sql_df()` 搭配連線池執行查詢
- Service 層處理業務邏輯Routes 層處理 HTTP
## Goals / Non-Goals
**Goals:**
- 提供設備維修工單查詢介面
- 支援單選/多選設備、時間範圍篩選
- 顯示兩層資料JOB 清單 → JOBTXNHISTORY 明細
- 完整匯出到 JOBTXNHISTORY 層級 (扁平化 CSV)
**Non-Goals:**
- 不修改既有 resource_cache 功能
- 不整合至現有 dashboard (獨立頁面)
- 不提供工單編輯功能 (唯讀查詢)
- 不處理 JOB 以外的維護類型 (如 MAINTENANCE 表)
## Decisions
### 1. 資料關聯策略:兩階段查詢
**選擇**: 先查 JOB再按需查 JOBTXNHISTORY
**替代方案**:
- 單次 JOIN 查詢JOB 和 JOBTXNHISTORY 一次 JOIN 回傳
- 問題JOBTXNHISTORY 資料量大JOIN 會產生大量重複的 JOB 欄位
**理由**:
- 前端列表只需 JOB 層級,減少傳輸量
- 展開明細時才查 JOBTXNHISTORY按需載入
- 匯出時使用單獨的 JOIN 查詢,一次產生扁平化資料
### 2. 設備選擇來源:使用 resource_cache
**選擇**: 從 `resource_cache.get_all_resources()` 取得設備清單
**替代方案**:
- 直接查詢 DW_MES_RESOURCE 表
- 問題:每次頁面載入都需查詢資料庫
**理由**:
- 與既有 resource_history 模式一致
- 利用快取減少資料庫負載
- 設備清單已有 RESOURCEID 可直接用於 JOB 關聯
### 3. SQL 管理:集中至 sql/job_query/
**選擇**: 新增 `sql/job_query/` 目錄存放 SQL 檔案
**檔案結構**:
```
sql/job_query/
├── job_list.sql # 工單清單 (前端列表)
├── job_txn_detail.sql # 單一工單的交易明細
└── job_txn_export.sql # 完整匯出 (JOB JOIN HISTORY)
```
**理由**:
- 遵循專案既有的 SQL 集中管理模式
- 便於維護和優化 SQL
### 4. 匯出格式:扁平化 CSV
**選擇**: JOB + JOBTXNHISTORY 扁平化為單一 CSV
**欄位設計**:
```
RESOURCENAME, JOBID, JOB_STATUS, JOB_CREATEDATE,
TXN_DATE, FROM_STATUS, TO_STATUS, CAUSE_CODE, REPAIR_CODE, USER_NAME
```
**理由**:
- 使用者需求是完整到 history 層級
- 扁平化格式便於 Excel 分析
- 每筆交易一列,包含對應的 JOB 資訊
### 5. API 端點設計
| 端點 | 方法 | 用途 |
|------|------|------|
| `/api/job-query/jobs` | POST | 查詢工單列表 |
| `/api/job-query/txn/{job_id}` | GET | 取得單一工單的交易歷史 |
| `/api/job-query/export` | POST | CSV 匯出 (完整到 history) |
**理由**:
- 列表查詢用 POST (帶 resource_ids 陣列)
- 明細查詢用 GET (單一 job_id)
- 匯出用 POST (同列表查詢參數)
## Risks / Trade-offs
### [風險] JOBTXNHISTORY 資料量大,匯出可能超時
**緩解**:
- 限制查詢時間範圍 (最多 365 天)
- 使用串流回應 (streaming response)
- 加入 TXNDATE 索引條件
### [風險] 多選大量設備時 IN 子句過長
**緩解**:
- 分批查詢 (每批 1000 個 RESOURCEID)
- 或使用 CTE 暫存設備清單
### [取捨] 前端不預載所有 JOBTXNHISTORY
**影響**: 展開明細需額外 API 呼叫
**優點**: 減少初始載入時間和傳輸量

View File

@@ -0,0 +1,40 @@
## Why
使用者需要查詢特定設備的維修工單歷史紀錄,目前系統缺乏專門的介面來查詢 DW_MES_JOB 和 DW_MES_JOBTXNHISTORY 資料。此工具讓維護工程師能夠:選擇單一或多台設備、指定時間範圍,查看工單清單及其完整的狀態變更軌跡,並匯出完整的交易歷史資料供分析。
## What Changes
- 新增「設備維修查詢工具」頁面 (`/job-query`)
- 新增設備選擇器,使用 resource_cache 快取的設備資料
- 新增工單查詢 API透過 RESOURCEID 關聯 DW_MES_JOB
- 新增工單交易歷史 API透過 JOBID 關聯 DW_MES_JOBTXNHISTORY
- 新增 CSV 匯出功能:完整匯出到 JOBTXNHISTORY 層級(扁平化格式)
- 新增集中管理的 SQL 檔案 (`sql/job_query/`)
## Capabilities
### New Capabilities
- `job-maintenance-query`: 設備維修工單查詢功能包含設備選擇、時間範圍篩選、工單列表顯示、交易歷史展開、CSV 匯出
### Modified Capabilities
(無修改既有規格)
## Impact
**新增檔案**:
- `src/mes_dashboard/templates/job_query.html` - 前端頁面
- `src/mes_dashboard/routes/job_query_routes.py` - API 端點
- `src/mes_dashboard/services/job_query_service.py` - 查詢邏輯
- `src/mes_dashboard/sql/job_query/*.sql` - SQL 查詢檔案
**依賴**:
- `resource_cache` - 設備快取資料 (既有)
- `DWH.DW_MES_JOB` - 維修工單表 (~125 萬筆)
- `DWH.DW_MES_JOBTXNHISTORY` - 工單交易歷史 (~955 萬筆)
**資料關聯**:
```
RESOURCE (RESOURCEID) → JOB (RESOURCEID) → JOBTXNHISTORY (JOBID)
```

View File

@@ -0,0 +1,102 @@
## ADDED Requirements
### Requirement: Equipment selection from cache
系統 SHALL 從 resource_cache 載入可用設備清單,讓使用者選擇要查詢的設備。
#### Scenario: Load equipment list
- **WHEN** 使用者進入設備維修查詢頁面
- **THEN** 系統顯示從 resource_cache 載入的設備清單,包含 RESOURCENAME 和 WORKCENTERNAME
#### Scenario: Multi-select equipment
- **WHEN** 使用者選擇多台設備
- **THEN** 系統記錄所選設備的 RESOURCEID 列表供後續查詢使用
---
### Requirement: Date range filter
系統 SHALL 提供日期範圍篩選功能,限制查詢的時間區間。
#### Scenario: Set date range
- **WHEN** 使用者指定起始日期和結束日期
- **THEN** 系統驗證日期格式為 YYYY-MM-DD 且結束日期不早於起始日期
#### Scenario: Date range limit
- **WHEN** 使用者選擇的日期範圍超過 365 天
- **THEN** 系統顯示錯誤訊息「日期範圍不可超過 365 天」
#### Scenario: Quick date preset
- **WHEN** 使用者點擊「最近 90 天」按鈕
- **THEN** 系統自動填入過去 90 天的日期範圍
---
### Requirement: Job list query
系統 SHALL 根據選擇的設備和時間範圍查詢工單清單 (DW_MES_JOB)。
#### Scenario: Query jobs by resources
- **WHEN** 使用者選擇設備並執行查詢
- **THEN** 系統查詢 DW_MES_JOB 表中 RESOURCEID 符合所選設備的工單
#### Scenario: Filter by date
- **WHEN** 使用者指定時間範圍
- **THEN** 系統篩選 CREATEDATE 在指定範圍內的工單
#### Scenario: Job list columns
- **WHEN** 工單查詢完成
- **THEN** 系統顯示欄位包含RESOURCENAME, JOBID, JOBSTATUS, CREATEDATE, COMPLETEDATE, CAUSECODENAME, REPAIRCODENAME
---
### Requirement: Job transaction history detail
系統 SHALL 提供展開功能,顯示單一工單的完整交易歷史 (DW_MES_JOBTXNHISTORY)。
#### Scenario: Expand job history
- **WHEN** 使用者點擊工單列的展開按鈕
- **THEN** 系統查詢該 JOBID 的所有 JOBTXNHISTORY 記錄並顯示
#### Scenario: History detail columns
- **WHEN** 交易歷史載入完成
- **THEN** 系統顯示欄位包含TXNDATE, FROMJOBSTATUS, JOBSTATUS, CAUSECODENAME, REPAIRCODENAME, USER_NAME
#### Scenario: History ordering
- **WHEN** 顯示交易歷史
- **THEN** 記錄依 TXNDATE 升序排列 (最早的在前)
---
### Requirement: CSV export with full history
系統 SHALL 提供 CSV 匯出功能,匯出完整到 JOBTXNHISTORY 層級的扁平化資料。
#### Scenario: Export request
- **WHEN** 使用者點擊「匯出 CSV」按鈕
- **THEN** 系統產生包含所有符合條件的工單及其交易歷史的 CSV 檔案
#### Scenario: Export format
- **WHEN** CSV 匯出完成
- **THEN** 檔案格式為 UTF-8 BOM每筆交易歷史為一列包含對應的工單資訊
#### Scenario: Export columns
- **WHEN** 檢視匯出的 CSV 內容
- **THEN** 欄位包含RESOURCENAME, JOBID, JOB_STATUS, JOB_CREATEDATE, TXN_DATE, FROM_STATUS, TO_STATUS, CAUSE_CODE, REPAIR_CODE, USER_NAME
---
### Requirement: Large dataset handling
系統 SHALL 處理大量設備選擇時的查詢效能問題。
#### Scenario: Batch resource filter
- **WHEN** 所選設備數量超過 1000 台
- **THEN** 系統將 RESOURCEID 分批處理,每批最多 1000 個
#### Scenario: Export timeout prevention
- **WHEN** 匯出大量資料
- **THEN** 系統使用串流回應 (streaming response) 避免超時
---
### Requirement: Page navigation
系統 SHALL 將設備維修查詢工具整合至主導航選單。
#### Scenario: Access from menu
- **WHEN** 使用者點擊導航選單中的「設備維修查詢」
- **THEN** 系統導航至 /job-query 頁面

View File

@@ -0,0 +1,45 @@
## 1. SQL 檔案
- [x] 1.1 建立 `sql/job_query/` 目錄
- [x] 1.2 建立 `job_list.sql` - 工單清單查詢 (JOB 層級)
- [x] 1.3 建立 `job_txn_detail.sql` - 單一工單的交易歷史查詢
- [x] 1.4 建立 `job_txn_export.sql` - 完整匯出查詢 (JOB JOIN JOBTXNHISTORY)
## 2. Service 層
- [x] 2.1 建立 `services/job_query_service.py`
- [x] 2.2 實作 `get_jobs_by_resources()` - 根據 RESOURCEID 列表查詢工單
- [x] 2.3 實作 `get_job_txn_history()` - 根據 JOBID 查詢交易歷史
- [x] 2.4 實作 `export_jobs_with_history()` - 產生完整匯出的 CSV 串流
- [x] 2.5 實作日期範圍驗證 (最多 365 天)
- [x] 2.6 實作大量 RESOURCEID 分批處理 (每批 1000)
## 3. Routes 層
- [x] 3.1 建立 `routes/job_query_routes.py`
- [x] 3.2 實作 `POST /api/job-query/jobs` - 工單列表查詢
- [x] 3.3 實作 `GET /api/job-query/txn/<job_id>` - 工單交易歷史
- [x] 3.4 實作 `POST /api/job-query/export` - CSV 匯出 (streaming response)
- [x] 3.5 註冊 Blueprint 到 app
## 4. 前端頁面
- [x] 4.1 建立 `templates/job_query.html`
- [x] 4.2 實作設備選擇器 (從 resource_cache 載入,支援多選)
- [x] 4.3 實作日期範圍選擇器 (含「最近 90 天」快速按鈕)
- [x] 4.4 實作工單列表表格 (含分頁)
- [x] 4.5 實作工單列展開功能 (顯示交易歷史)
- [x] 4.6 實作 CSV 匯出按鈕
- [x] 4.7 實作查詢中 loading 狀態
## 5. 導航整合
- [x] 5.1 新增頁面路由 `/job-query`
- [x] 5.2 將「設備維修查詢」加入導航選單
## 6. 測試
- [x] 6.1 Service 層單元測試 (mock 資料庫)
- [x] 6.2 Routes 層 API 測試
- [x] 6.3 手動 E2E 測試 - 完整查詢流程
- [x] 6.4 手動 E2E 測試 - CSV 匯出驗證

View File

@@ -0,0 +1,102 @@
## ADDED Requirements
### Requirement: Equipment selection from cache
系統 SHALL 從 resource_cache 載入可用設備清單,讓使用者選擇要查詢的設備。
#### Scenario: Load equipment list
- **WHEN** 使用者進入設備維修查詢頁面
- **THEN** 系統顯示從 resource_cache 載入的設備清單,包含 RESOURCENAME 和 WORKCENTERNAME
#### Scenario: Multi-select equipment
- **WHEN** 使用者選擇多台設備
- **THEN** 系統記錄所選設備的 RESOURCEID 列表供後續查詢使用
---
### Requirement: Date range filter
系統 SHALL 提供日期範圍篩選功能,限制查詢的時間區間。
#### Scenario: Set date range
- **WHEN** 使用者指定起始日期和結束日期
- **THEN** 系統驗證日期格式為 YYYY-MM-DD 且結束日期不早於起始日期
#### Scenario: Date range limit
- **WHEN** 使用者選擇的日期範圍超過 365 天
- **THEN** 系統顯示錯誤訊息「日期範圍不可超過 365 天」
#### Scenario: Quick date preset
- **WHEN** 使用者點擊「最近 90 天」按鈕
- **THEN** 系統自動填入過去 90 天的日期範圍
---
### Requirement: Job list query
系統 SHALL 根據選擇的設備和時間範圍查詢工單清單 (DW_MES_JOB)。
#### Scenario: Query jobs by resources
- **WHEN** 使用者選擇設備並執行查詢
- **THEN** 系統查詢 DW_MES_JOB 表中 RESOURCEID 符合所選設備的工單
#### Scenario: Filter by date
- **WHEN** 使用者指定時間範圍
- **THEN** 系統篩選 CREATEDATE 在指定範圍內的工單
#### Scenario: Job list columns
- **WHEN** 工單查詢完成
- **THEN** 系統顯示欄位包含RESOURCENAME, JOBID, JOBSTATUS, CREATEDATE, COMPLETEDATE, CAUSECODENAME, REPAIRCODENAME
---
### Requirement: Job transaction history detail
系統 SHALL 提供展開功能,顯示單一工單的完整交易歷史 (DW_MES_JOBTXNHISTORY)。
#### Scenario: Expand job history
- **WHEN** 使用者點擊工單列的展開按鈕
- **THEN** 系統查詢該 JOBID 的所有 JOBTXNHISTORY 記錄並顯示
#### Scenario: History detail columns
- **WHEN** 交易歷史載入完成
- **THEN** 系統顯示欄位包含TXNDATE, FROMJOBSTATUS, JOBSTATUS, CAUSECODENAME, REPAIRCODENAME, USER_NAME
#### Scenario: History ordering
- **WHEN** 顯示交易歷史
- **THEN** 記錄依 TXNDATE 升序排列 (最早的在前)
---
### Requirement: CSV export with full history
系統 SHALL 提供 CSV 匯出功能,匯出完整到 JOBTXNHISTORY 層級的扁平化資料。
#### Scenario: Export request
- **WHEN** 使用者點擊「匯出 CSV」按鈕
- **THEN** 系統產生包含所有符合條件的工單及其交易歷史的 CSV 檔案
#### Scenario: Export format
- **WHEN** CSV 匯出完成
- **THEN** 檔案格式為 UTF-8 BOM每筆交易歷史為一列包含對應的工單資訊
#### Scenario: Export columns
- **WHEN** 檢視匯出的 CSV 內容
- **THEN** 欄位包含RESOURCENAME, JOBID, JOB_STATUS, JOB_CREATEDATE, TXN_DATE, FROM_STATUS, TO_STATUS, CAUSE_CODE, REPAIR_CODE, USER_NAME
---
### Requirement: Large dataset handling
系統 SHALL 處理大量設備選擇時的查詢效能問題。
#### Scenario: Batch resource filter
- **WHEN** 所選設備數量超過 1000 台
- **THEN** 系統將 RESOURCEID 分批處理,每批最多 1000 個
#### Scenario: Export timeout prevention
- **WHEN** 匯出大量資料
- **THEN** 系統使用串流回應 (streaming response) 避免超時
---
### Requirement: Page navigation
系統 SHALL 將設備維修查詢工具整合至主導航選單。
#### Scenario: Access from menu
- **WHEN** 使用者點擊導航選單中的「設備維修查詢」
- **THEN** 系統導航至 /job-query 頁面

View File

@@ -13,7 +13,7 @@ from mes_dashboard.config.tables import TABLES_CONFIG
from mes_dashboard.config.settings import get_config
from mes_dashboard.core.cache import NoOpCache
from mes_dashboard.core.database import get_table_data, get_table_columns, get_engine, init_db, start_keepalive
from mes_dashboard.core.permissions import is_admin_logged_in
from mes_dashboard.core.permissions import is_admin_logged_in, _is_ajax_request
from mes_dashboard.routes import register_routes
from mes_dashboard.routes.auth_routes import auth_bp
from mes_dashboard.routes.admin_routes import admin_bp
@@ -128,6 +128,9 @@ def create_app(config_name: str | None = None) -> Flask:
# Admin pages require login
if request.path.startswith("/admin/"):
if not is_admin_logged_in():
# For AJAX requests, return JSON error instead of redirect
if _is_ajax_request():
return jsonify({"error": "請先登入管理員帳號", "login_required": True}), 401
return redirect(url_for("auth.login", next=request.url))
return None

View File

@@ -6,7 +6,7 @@ from __future__ import annotations
from functools import wraps
from typing import TYPE_CHECKING, Callable
from flask import redirect, request, session, url_for
from flask import jsonify, redirect, request, session, url_for
if TYPE_CHECKING:
from typing import Any
@@ -30,14 +30,37 @@ def get_current_admin() -> dict | None:
return session.get("admin")
def _is_ajax_request() -> bool:
"""Check if the current request is an AJAX request.
Returns:
True if request appears to be AJAX (fetch/XHR)
"""
# Check X-Requested-With header (jQuery style)
if request.headers.get("X-Requested-With") == "XMLHttpRequest":
return True
# Check Accept header for JSON
accept = request.headers.get("Accept", "")
if "application/json" in accept:
return True
# Check Content-Type for JSON POST requests
content_type = request.headers.get("Content-Type", "")
if "application/json" in content_type:
return True
return False
def admin_required(f: Callable) -> Callable:
"""Decorator to require admin login for a route.
Redirects to login page if not logged in.
For regular requests: Redirects to login page if not logged in.
For AJAX requests: Returns JSON error with 401 status.
"""
@wraps(f)
def decorated(*args: Any, **kwargs: Any) -> Any:
if not is_admin_logged_in():
if _is_ajax_request():
return jsonify({"error": "請先登入管理員帳號", "login_required": True}), 401
return redirect(url_for("auth.login", next=request.url))
return f(*args, **kwargs)
return decorated

View File

@@ -12,6 +12,7 @@ from .hold_routes import hold_bp
from .auth_routes import auth_bp
from .admin_routes import admin_bp
from .resource_history_routes import resource_history_bp
from .job_query_routes import job_query_bp
def register_routes(app) -> None:
@@ -22,6 +23,7 @@ def register_routes(app) -> None:
app.register_blueprint(excel_query_bp)
app.register_blueprint(hold_bp)
app.register_blueprint(resource_history_bp)
app.register_blueprint(job_query_bp)
__all__ = [
'wip_bp',
@@ -32,5 +34,6 @@ __all__ = [
'auth_bp',
'admin_bp',
'resource_history_bp',
'job_query_bp',
'register_routes',
]

View File

@@ -0,0 +1,165 @@
# -*- coding: utf-8 -*-
"""Job Query API routes.
Contains Flask Blueprint for maintenance job query endpoints:
- Job list query by resources
- Job transaction history detail
- CSV export with full history
"""
from flask import Blueprint, jsonify, request, Response, render_template
from mes_dashboard.services.job_query_service import (
get_jobs_by_resources,
get_job_txn_history,
export_jobs_with_history,
validate_date_range,
)
# Create Blueprint
job_query_bp = Blueprint('job_query', __name__)
# ============================================================
# Page Route
# ============================================================
@job_query_bp.route('/job-query')
def job_query_page():
"""Render the job query page."""
return render_template('job_query.html')
# ============================================================
# API Routes
# ============================================================
@job_query_bp.route('/api/job-query/resources', methods=['GET'])
def get_resources():
"""Get available resources for selection.
Returns resources from cache for equipment selection.
"""
from mes_dashboard.services.resource_cache import get_all_resources
try:
resources = get_all_resources()
if not resources:
return jsonify({'error': '無法載入設備資料'}), 500
# Return minimal data for selection UI
data = []
for r in resources:
data.append({
'RESOURCEID': r.get('RESOURCEID'),
'RESOURCENAME': r.get('RESOURCENAME'),
'WORKCENTERNAME': r.get('WORKCENTERNAME'),
'RESOURCEFAMILYNAME': r.get('RESOURCEFAMILYNAME'),
})
# Sort by WORKCENTERNAME, then RESOURCENAME
data.sort(key=lambda x: (x.get('WORKCENTERNAME', ''), x.get('RESOURCENAME', '')))
return jsonify({
'data': data,
'total': len(data)
})
except Exception as exc:
return jsonify({'error': f'載入設備資料失敗: {str(exc)}'}), 500
@job_query_bp.route('/api/job-query/jobs', methods=['POST'])
def query_jobs():
"""Query jobs for selected resources.
Expects JSON body:
{
"resource_ids": ["id1", "id2", ...],
"start_date": "2024-01-01",
"end_date": "2024-12-31"
}
Returns job list.
"""
data = request.get_json()
resource_ids = data.get('resource_ids', [])
start_date = data.get('start_date')
end_date = data.get('end_date')
# Validation
if not resource_ids:
return jsonify({'error': '請選擇至少一台設備'}), 400
if not start_date or not end_date:
return jsonify({'error': '請指定日期範圍'}), 400
validation_error = validate_date_range(start_date, end_date)
if validation_error:
return jsonify({'error': validation_error}), 400
result = get_jobs_by_resources(resource_ids, start_date, end_date)
if 'error' in result:
return jsonify(result), 400
return jsonify(result)
@job_query_bp.route('/api/job-query/txn/<job_id>', methods=['GET'])
def query_job_txn_history(job_id: str):
"""Query transaction history for a single job.
Args:
job_id: The JOBID to query
Returns transaction history list.
"""
if not job_id:
return jsonify({'error': '請指定工單 ID'}), 400
result = get_job_txn_history(job_id)
if 'error' in result:
return jsonify(result), 400
return jsonify(result)
@job_query_bp.route('/api/job-query/export', methods=['POST'])
def export_jobs():
"""Export jobs with full transaction history as CSV.
Expects JSON body:
{
"resource_ids": ["id1", "id2", ...],
"start_date": "2024-01-01",
"end_date": "2024-12-31"
}
Returns streaming CSV response.
"""
data = request.get_json()
resource_ids = data.get('resource_ids', [])
start_date = data.get('start_date')
end_date = data.get('end_date')
# Validation
if not resource_ids:
return jsonify({'error': '請選擇至少一台設備'}), 400
if not start_date or not end_date:
return jsonify({'error': '請指定日期範圍'}), 400
validation_error = validate_date_range(start_date, end_date)
if validation_error:
return jsonify({'error': validation_error}), 400
# Stream CSV response
return Response(
export_jobs_with_history(resource_ids, start_date, end_date),
mimetype='text/csv; charset=utf-8',
headers={
'Content-Disposition': 'attachment; filename=job_history_export.csv'
}
)

View File

@@ -0,0 +1,381 @@
# -*- coding: utf-8 -*-
"""Job Query Service.
Provides functions for querying maintenance job data:
- Job list by resource IDs
- Job transaction history detail
- CSV export with full history
Architecture:
- Uses resource_cache as the source for equipment master data
- Queries DW_MES_JOB for job current status
- Queries DW_MES_JOBTXNHISTORY for transaction history
- Supports batching for large resource lists (Oracle IN clause limit)
"""
import csv
import io
import logging
from datetime import datetime
from typing import Dict, List, Any, Optional, Generator
import pandas as pd
from mes_dashboard.core.database import read_sql_df, get_db_connection
from mes_dashboard.sql import SQLLoader
logger = logging.getLogger('mes_dashboard.job_query')
# Constants
BATCH_SIZE = 1000 # Oracle IN clause limit
MAX_DATE_RANGE_DAYS = 365
# ============================================================
# Validation Functions
# ============================================================
def validate_date_range(start_date: str, end_date: str) -> Optional[str]:
"""Validate date range.
Args:
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
Returns:
Error message if validation fails, None if valid.
"""
try:
start = datetime.strptime(start_date, '%Y-%m-%d')
end = datetime.strptime(end_date, '%Y-%m-%d')
if end < start:
return '結束日期不可早於起始日期'
diff = (end - start).days
if diff > MAX_DATE_RANGE_DAYS:
return f'日期範圍不可超過 {MAX_DATE_RANGE_DAYS}'
return None
except ValueError as e:
return f'日期格式錯誤: {e}'
# ============================================================
# Resource Filter Helpers
# ============================================================
def _build_resource_filter(resource_ids: List[str], max_chunk_size: int = BATCH_SIZE) -> List[str]:
"""Build SQL IN clause lists for resource IDs.
Oracle has a limit of ~1000 items per IN clause, so we chunk if needed.
Args:
resource_ids: List of resource IDs.
max_chunk_size: Maximum items per IN clause.
Returns:
List of SQL IN clause strings (e.g., "'ID1', 'ID2', 'ID3'").
"""
if not resource_ids:
return []
# Escape single quotes
escaped_ids = [rid.replace("'", "''") for rid in resource_ids]
# Chunk into groups
chunks = []
for i in range(0, len(escaped_ids), max_chunk_size):
chunk = escaped_ids[i:i + max_chunk_size]
chunks.append("'" + "', '".join(chunk) + "'")
return chunks
def _build_resource_filter_sql(resource_ids: List[str], column: str = 'j.RESOURCEID') -> str:
"""Build SQL WHERE clause for resource ID filtering.
Handles chunking for large resource lists.
Args:
resource_ids: List of resource IDs.
column: Column name to filter on.
Returns:
SQL condition string (e.g., "j.RESOURCEID IN ('ID1', 'ID2')").
"""
chunks = _build_resource_filter(resource_ids)
if not chunks:
return "1=0" # No resources = no results
if len(chunks) == 1:
return f"{column} IN ({chunks[0]})"
# Multiple chunks need OR
conditions = [f"{column} IN ({chunk})" for chunk in chunks]
return "(" + " OR ".join(conditions) + ")"
# ============================================================
# Query Functions
# ============================================================
def get_jobs_by_resources(
resource_ids: List[str],
start_date: str,
end_date: str
) -> Dict[str, Any]:
"""Query jobs for selected resources within date range.
Args:
resource_ids: List of RESOURCEID values to query
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
Returns:
Dict with 'data' (list of job records) and 'total' (count),
or 'error' if query fails.
"""
# Validate inputs
if not resource_ids:
return {'error': '請選擇至少一台設備'}
validation_error = validate_date_range(start_date, end_date)
if validation_error:
return {'error': validation_error}
try:
# Build resource filter
resource_filter = _build_resource_filter_sql(resource_ids)
# Load SQL template
sql = SQLLoader.load("job_query/job_list")
sql = sql.replace("{{ RESOURCE_FILTER }}", resource_filter)
# Execute query
params = {'start_date': start_date, 'end_date': end_date}
df = read_sql_df(sql, params)
# Convert to records
data = []
for _, row in df.iterrows():
record = {}
for col in df.columns:
value = row[col]
if pd.isna(value):
record[col] = None
elif isinstance(value, datetime):
record[col] = value.strftime('%Y-%m-%d %H:%M:%S')
else:
record[col] = value
data.append(record)
logger.info(f"Job query returned {len(data)} records for {len(resource_ids)} resources")
return {
'data': data,
'total': len(data),
'resource_count': len(resource_ids)
}
except Exception as exc:
logger.error(f"Job query failed: {exc}")
return {'error': f'查詢失敗: {str(exc)}'}
def get_job_txn_history(job_id: str) -> Dict[str, Any]:
"""Query transaction history for a single job.
Args:
job_id: The JOBID to query
Returns:
Dict with 'data' (list of transaction records) and 'total' (count),
or 'error' if query fails.
"""
if not job_id:
return {'error': '請指定工單 ID'}
try:
# Load SQL template
sql = SQLLoader.load("job_query/job_txn_detail")
# Execute query
params = {'job_id': job_id}
df = read_sql_df(sql, params)
# Convert to records
data = []
for _, row in df.iterrows():
record = {}
for col in df.columns:
value = row[col]
if pd.isna(value):
record[col] = None
elif isinstance(value, datetime):
record[col] = value.strftime('%Y-%m-%d %H:%M:%S')
else:
record[col] = value
data.append(record)
logger.debug(f"Transaction history query returned {len(data)} records for job {job_id}")
return {
'data': data,
'total': len(data),
'job_id': job_id
}
except Exception as exc:
logger.error(f"Transaction history query failed for job {job_id}: {exc}")
return {'error': f'查詢失敗: {str(exc)}'}
# ============================================================
# Export Functions
# ============================================================
def export_jobs_with_history(
resource_ids: List[str],
start_date: str,
end_date: str
) -> Generator[str, None, None]:
"""Generate CSV content for jobs with full transaction history.
Uses streaming to handle large datasets without memory issues.
Args:
resource_ids: List of RESOURCEID values to export
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
Yields:
CSV rows as strings (including header row first)
"""
# Validate inputs
if not resource_ids:
yield "Error: 請選擇至少一台設備\n"
return
validation_error = validate_date_range(start_date, end_date)
if validation_error:
yield f"Error: {validation_error}\n"
return
try:
# Build resource filter
resource_filter = _build_resource_filter_sql(resource_ids)
# Load SQL template
sql = SQLLoader.load("job_query/job_txn_export")
sql = sql.replace("{{ RESOURCE_FILTER }}", resource_filter)
# Execute query
params = {'start_date': start_date, 'end_date': end_date}
df = read_sql_df(sql, params)
if df is None or len(df) == 0:
yield "Error: 無符合條件的資料\n"
return
# Write CSV header with BOM for Excel UTF-8 compatibility
output = io.StringIO()
output.write('\ufeff') # UTF-8 BOM
# Define column headers (Chinese labels)
headers = [
'設備名稱', '工單ID', '工單最終狀態', '工單類型', '工單名稱',
'工單建立時間', '工單完成時間', '工單故障碼', '工單維修碼', '工單症狀碼',
'交易時間', '原狀態', '新狀態', '階段',
'交易故障碼', '交易維修碼', '交易症狀碼',
'操作者', '員工', '備註'
]
writer = csv.writer(output)
writer.writerow(headers)
yield output.getvalue()
output.truncate(0)
output.seek(0)
# Write data rows
for _, row in df.iterrows():
csv_row = []
for col in df.columns:
value = row[col]
if pd.isna(value):
csv_row.append('')
elif isinstance(value, datetime):
csv_row.append(value.strftime('%Y-%m-%d %H:%M:%S'))
else:
csv_row.append(str(value))
writer.writerow(csv_row)
yield output.getvalue()
output.truncate(0)
output.seek(0)
logger.info(f"CSV export completed: {len(df)} records")
except Exception as exc:
logger.error(f"CSV export failed: {exc}")
yield f"Error: 匯出失敗 - {str(exc)}\n"
def get_export_data(
resource_ids: List[str],
start_date: str,
end_date: str
) -> Dict[str, Any]:
"""Get export data as a dict (for non-streaming use cases).
Args:
resource_ids: List of RESOURCEID values to export
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
Returns:
Dict with 'data', 'columns', 'total', or 'error' if query fails.
"""
# Validate inputs
if not resource_ids:
return {'error': '請選擇至少一台設備'}
validation_error = validate_date_range(start_date, end_date)
if validation_error:
return {'error': validation_error}
try:
# Build resource filter
resource_filter = _build_resource_filter_sql(resource_ids)
# Load SQL template
sql = SQLLoader.load("job_query/job_txn_export")
sql = sql.replace("{{ RESOURCE_FILTER }}", resource_filter)
# Execute query
params = {'start_date': start_date, 'end_date': end_date}
df = read_sql_df(sql, params)
# Convert to records
data = []
for _, row in df.iterrows():
record = {}
for col in df.columns:
value = row[col]
if pd.isna(value):
record[col] = None
elif isinstance(value, datetime):
record[col] = value.strftime('%Y-%m-%d %H:%M:%S')
else:
record[col] = value
data.append(record)
return {
'data': data,
'columns': list(df.columns),
'total': len(data)
}
except Exception as exc:
logger.error(f"Export data query failed: {exc}")
return {'error': f'查詢失敗: {str(exc)}'}

View File

@@ -0,0 +1,33 @@
-- Job List Query
-- Retrieves maintenance jobs for selected resources within date range
-- Placeholders:
-- RESOURCE_FILTER - Resource ID filter condition (e.g., RESOURCEID IN (...))
-- Parameters:
-- :start_date - Start date (YYYY-MM-DD)
-- :end_date - End date (YYYY-MM-DD)
SELECT
j.JOBID,
j.RESOURCEID,
j.RESOURCENAME,
j.JOBSTATUS,
j.JOBMODELNAME,
j.JOBORDERNAME,
j.CREATEDATE,
j.COMPLETEDATE,
j.CANCELDATE,
j.FIRSTCLOCKONDATE,
j.LASTCLOCKOFFDATE,
j.CAUSECODENAME,
j.REPAIRCODENAME,
j.SYMPTOMCODENAME,
j.PJ_CAUSECODE2NAME,
j.PJ_REPAIRCODE2NAME,
j.PJ_SYMPTOMCODE2NAME,
j.CREATE_EMPNAME,
j.COMPLETE_EMPNAME
FROM DWH.DW_MES_JOB j
WHERE {{ RESOURCE_FILTER }}
AND j.CREATEDATE >= TO_DATE(:start_date, 'YYYY-MM-DD')
AND j.CREATEDATE < TO_DATE(:end_date, 'YYYY-MM-DD') + 1
ORDER BY j.RESOURCENAME, j.CREATEDATE DESC

View File

@@ -0,0 +1,27 @@
-- Job Transaction History Detail
-- Retrieves all transaction history for a single job
-- Parameters:
-- :job_id - The JOBID to query
SELECT
h.JOBTXNHISTORYID,
h.JOBID,
h.TXNDATE,
h.FROMJOBSTATUS,
h.JOBSTATUS,
h.STAGENAME,
h.TOSTAGENAME,
h.CAUSECODENAME,
h.REPAIRCODENAME,
h.SYMPTOMCODENAME,
h.USER_EMPNO,
h.USER_NAME,
h.EMP_EMPNO,
h.EMP_NAME,
h.COMMENTS,
h.CDONAME,
h.JOBMODELNAME,
h.JOBORDERNAME
FROM DWH.DW_MES_JOBTXNHISTORY h
WHERE h.JOBID = :job_id
ORDER BY h.TXNDATE ASC

View File

@@ -0,0 +1,35 @@
-- Job Transaction Export (Full History)
-- Joins JOB and JOBTXNHISTORY for complete CSV export
-- Placeholders:
-- RESOURCE_FILTER - Resource ID filter condition (e.g., j.RESOURCEID IN (...))
-- Parameters:
-- :start_date - Start date (YYYY-MM-DD)
-- :end_date - End date (YYYY-MM-DD)
SELECT
j.RESOURCENAME,
j.JOBID,
j.JOBSTATUS as JOB_FINAL_STATUS,
j.JOBMODELNAME,
j.JOBORDERNAME,
j.CREATEDATE as JOB_CREATEDATE,
j.COMPLETEDATE as JOB_COMPLETEDATE,
j.CAUSECODENAME as JOB_CAUSECODE,
j.REPAIRCODENAME as JOB_REPAIRCODE,
j.SYMPTOMCODENAME as JOB_SYMPTOMCODE,
h.TXNDATE,
h.FROMJOBSTATUS,
h.JOBSTATUS as TXN_TO_STATUS,
h.STAGENAME,
h.CAUSECODENAME as TXN_CAUSECODE,
h.REPAIRCODENAME as TXN_REPAIRCODE,
h.SYMPTOMCODENAME as TXN_SYMPTOMCODE,
h.USER_NAME,
h.EMP_NAME,
h.COMMENTS
FROM DWH.DW_MES_JOB j
JOIN DWH.DW_MES_JOBTXNHISTORY h ON j.JOBID = h.JOBID
WHERE {{ RESOURCE_FILTER }}
AND h.TXNDATE >= TO_DATE(:start_date, 'YYYY-MM-DD')
AND h.TXNDATE < TO_DATE(:end_date, 'YYYY-MM-DD') + 1
ORDER BY j.RESOURCENAME, j.JOBID, h.TXNDATE

View File

@@ -701,6 +701,24 @@
const PAGE_SIZE = 50;
let currentPage = 1;
let totalLogs = 0;
let authErrorShown = false;
// ============================================================
// Auth Helper
// ============================================================
async function fetchWithAuth(url, options = {}) {
const resp = await fetch(url, { ...options, cache: 'no-store' });
if (resp.status === 401) {
const json = await resp.json().catch(() => ({}));
if (!authErrorShown) {
authErrorShown = true;
alert('登入已過期,請重新登入');
window.location.href = '/auth/login?next=' + encodeURIComponent(window.location.pathname);
}
throw new Error('需要重新登入');
}
return resp;
}
// ============================================================
// Chart Setup
@@ -755,7 +773,7 @@
// ============================================================
async function loadSystemStatus() {
try {
const resp = await fetch('/admin/api/system-status', { cache: 'no-store' });
const resp = await fetchWithAuth('/admin/api/system-status');
const json = await resp.json();
if (!json.success) throw new Error(json.error);
@@ -804,7 +822,7 @@
async function loadMetrics() {
try {
const resp = await fetch('/admin/api/metrics', { cache: 'no-store' });
const resp = await fetchWithAuth('/admin/api/metrics');
const json = await resp.json();
if (!json.success) throw new Error(json.error);
@@ -841,7 +859,7 @@
if (level) url += `&level=${encodeURIComponent(level)}`;
if (q) url += `&q=${encodeURIComponent(q)}`;
const resp = await fetch(url, { cache: 'no-store' });
const resp = await fetchWithAuth(url);
const json = await resp.json();
if (!json.success) throw new Error(json.error);
@@ -933,7 +951,7 @@
// ============================================================
async function loadWorkerStatus() {
try {
const resp = await fetch('/admin/api/worker/status', { cache: 'no-store' });
const resp = await fetchWithAuth('/admin/api/worker/status');
const json = await resp.json();
if (!json.success) throw new Error(json.error);
@@ -999,10 +1017,9 @@
btn.style.opacity = '0.5';
try {
const resp = await fetch('/admin/api/worker/restart', {
const resp = await fetchWithAuth('/admin/api/worker/restart', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
cache: 'no-store'
headers: { 'Content-Type': 'application/json' }
});
const json = await resp.json();
@@ -1062,10 +1079,9 @@
btn.style.opacity = '0.5';
try {
const resp = await fetch('/admin/api/logs/cleanup', {
const resp = await fetchWithAuth('/admin/api/logs/cleanup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
cache: 'no-store'
headers: { 'Content-Type': 'application/json' }
});
const json = await resp.json();

View File

@@ -0,0 +1,918 @@
{% extends "_base.html" %}
{% block title %}設備維修查詢工具{% endblock %}
{% block head_extra %}
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: 'Microsoft JhengHei', Arial, sans-serif;
background: #f5f5f5;
padding: 20px;
}
.container {
max-width: 1600px;
margin: 0 auto;
background: white;
border-radius: 8px;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
.header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 30px;
border-radius: 8px 8px 0 0;
}
.header h1 {
font-size: 28px;
margin-bottom: 10px;
}
.header p {
opacity: 0.9;
font-size: 14px;
}
.content {
padding: 30px;
}
/* Filter Panel */
.filter-panel {
display: grid;
grid-template-columns: 1fr 1fr auto;
gap: 20px;
margin-bottom: 25px;
padding: 20px;
border: 1px solid #e0e0e0;
border-radius: 8px;
background: #fafafa;
}
.filter-group {
display: flex;
flex-direction: column;
gap: 8px;
}
.filter-group label {
font-weight: 500;
color: #333;
font-size: 14px;
}
/* Equipment Selector */
.equipment-selector {
position: relative;
}
.equipment-input {
width: 100%;
padding: 10px 15px;
border: 1px solid #ddd;
border-radius: 6px;
font-size: 14px;
cursor: pointer;
background: white;
}
.equipment-input:focus {
outline: none;
border-color: #667eea;
}
.equipment-dropdown {
position: absolute;
top: 100%;
left: 0;
right: 0;
max-height: 350px;
overflow-y: auto;
background: white;
border: 1px solid #ddd;
border-radius: 6px;
box-shadow: 0 4px 12px rgba(0,0,0,0.15);
z-index: 1000;
display: none;
}
.equipment-dropdown.show {
display: block;
}
.equipment-search {
width: 100%;
padding: 10px 15px;
border: none;
border-bottom: 1px solid #eee;
font-size: 14px;
}
.equipment-search:focus {
outline: none;
}
.equipment-list {
max-height: 280px;
overflow-y: auto;
}
.equipment-item {
display: flex;
align-items: center;
padding: 10px 15px;
cursor: pointer;
border-bottom: 1px solid #f0f0f0;
}
.equipment-item:hover {
background: #f5f7ff;
}
.equipment-item.selected {
background: #e3e8ff;
}
.equipment-item input[type="checkbox"] {
margin-right: 10px;
}
.equipment-info {
flex: 1;
}
.equipment-name {
font-weight: 500;
}
.equipment-workcenter {
font-size: 12px;
color: #666;
}
.selected-count {
font-size: 12px;
color: #667eea;
margin-top: 5px;
}
/* Date Range */
.date-range {
display: flex;
gap: 10px;
align-items: center;
}
input[type="date"] {
padding: 10px 15px;
border: 1px solid #ddd;
border-radius: 6px;
font-size: 14px;
}
input[type="date"]:focus {
outline: none;
border-color: #667eea;
}
/* Buttons */
.btn {
padding: 10px 20px;
border: none;
border-radius: 6px;
cursor: pointer;
font-size: 14px;
font-weight: bold;
transition: all 0.2s;
}
.btn-primary {
background: #667eea;
color: white;
}
.btn-primary:hover {
background: #5a6fd6;
}
.btn-success {
background: #28a745;
color: white;
}
.btn-success:hover {
background: #218838;
}
.btn-secondary {
background: #6c757d;
color: white;
}
.btn-secondary:hover {
background: #5a6268;
}
.btn-sm {
padding: 6px 12px;
font-size: 12px;
}
.btn:disabled {
background: #ccc;
cursor: not-allowed;
}
.filter-actions {
display: flex;
flex-direction: column;
gap: 10px;
justify-content: flex-end;
}
/* Result Section */
.result-section {
margin-top: 20px;
}
.result-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 15px;
}
.result-info {
font-size: 14px;
color: #666;
}
.result-actions {
display: flex;
gap: 10px;
}
/* Table */
.table-container {
overflow-x: auto;
border: 1px solid #e0e0e0;
border-radius: 8px;
}
table {
width: 100%;
border-collapse: collapse;
font-size: 13px;
}
th, td {
padding: 12px 15px;
text-align: left;
border-bottom: 1px solid #e0e0e0;
}
th {
background: #f8f9fa;
font-weight: 600;
color: #333;
position: sticky;
top: 0;
}
tr:hover {
background: #f5f7ff;
}
/* Expandable Row */
.expand-btn {
background: none;
border: none;
cursor: pointer;
padding: 4px 8px;
border-radius: 4px;
font-size: 16px;
color: #667eea;
}
.expand-btn:hover {
background: #e3e8ff;
}
.job-row.expanded {
background: #f5f7ff;
}
.txn-history-row {
display: none;
}
.txn-history-row.show {
display: table-row;
}
.txn-history-cell {
padding: 0;
background: #fafafa;
}
.txn-history-table {
margin: 10px 20px;
width: calc(100% - 40px);
border: 1px solid #ddd;
border-radius: 6px;
overflow: hidden;
}
.txn-history-table th {
background: #e9ecef;
font-size: 12px;
}
.txn-history-table td {
font-size: 12px;
padding: 8px 12px;
}
/* Status badges */
.status-badge {
display: inline-block;
padding: 3px 8px;
border-radius: 4px;
font-size: 11px;
font-weight: 500;
}
.status-badge.CREATED { background: #cfe2ff; color: #084298; }
.status-badge.ASSIGNED { background: #fff3cd; color: #664d03; }
.status-badge.IN_PROGRESS { background: #ffc107; color: #000; }
.status-badge.COMPLETED { background: #d1e7dd; color: #0a3622; }
.status-badge.CANCELLED { background: #f8d7da; color: #58151c; }
/* Loading */
.loading {
text-align: center;
padding: 40px;
color: #666;
}
.loading-spinner {
display: inline-block;
width: 40px;
height: 40px;
border: 4px solid #f3f3f3;
border-top: 4px solid #667eea;
border-radius: 50%;
animation: spin 1s linear infinite;
}
@keyframes spin {
0% { transform: rotate(0deg); }
100% { transform: rotate(360deg); }
}
.error {
background: #f8d7da;
color: #721c24;
padding: 15px;
border-radius: 6px;
margin: 10px 0;
}
.info-box {
background: #e3e8ff;
color: #667eea;
padding: 10px 15px;
border-radius: 6px;
font-size: 13px;
}
/* Empty state */
.empty-state {
text-align: center;
padding: 60px 20px;
color: #666;
}
.empty-state p {
margin-bottom: 10px;
}
/* Arrow animation */
.arrow-icon {
transition: transform 0.2s;
}
.arrow-icon.rotated {
transform: rotate(90deg);
}
</style>
{% endblock %}
{% block content %}
<div class="container">
<div class="header">
<h1>設備維修查詢工具</h1>
<p>查詢設備維修工單及交易歷史,支援匯出完整資料</p>
</div>
<div class="content">
<!-- Filter Panel -->
<div class="filter-panel">
<!-- Equipment Selector -->
<div class="filter-group">
<label>選擇設備:</label>
<div class="equipment-selector">
<div class="equipment-input" onclick="toggleEquipmentDropdown()">
<span id="equipmentDisplay">點擊選擇設備...</span>
</div>
<div class="equipment-dropdown" id="equipmentDropdown">
<input type="text" class="equipment-search" placeholder="搜尋設備名稱或站點..." oninput="filterEquipments(this.value)">
<div class="equipment-list" id="equipmentList">
<div class="loading">
<div class="loading-spinner"></div>
<br>載入設備中...
</div>
</div>
</div>
<div class="selected-count" id="selectedCount"></div>
</div>
</div>
<!-- Date Range -->
<div class="filter-group">
<label>日期範圍:</label>
<div class="date-range">
<input type="date" id="dateFrom">
<span>~</span>
<input type="date" id="dateTo">
<button class="btn btn-secondary btn-sm" onclick="setLast90Days()">最近 90 天</button>
</div>
</div>
<!-- Actions -->
<div class="filter-actions">
<button class="btn btn-primary" onclick="queryJobs()" id="queryBtn">
查詢工單
</button>
<button class="btn btn-success" onclick="exportCsv()" id="exportBtn" disabled>
匯出 CSV
</button>
</div>
</div>
<!-- Result Section -->
<div class="result-section" id="resultSection">
<div class="empty-state">
<p>請選擇設備和日期範圍後,點擊「查詢工單」</p>
</div>
</div>
</div>
</div>
{% endblock %}
{% block scripts %}
<script>
// State
let allEquipments = [];
let selectedEquipments = new Set();
let jobsData = [];
let expandedJobs = new Set();
// Initialize
document.addEventListener('DOMContentLoaded', () => {
loadEquipments();
setLast90Days();
// Close dropdown when clicking outside
document.addEventListener('click', (e) => {
const dropdown = document.getElementById('equipmentDropdown');
const selector = document.querySelector('.equipment-selector');
if (!selector.contains(e.target)) {
dropdown.classList.remove('show');
}
});
});
// Load equipments from cache
async function loadEquipments() {
try {
const data = await MesApi.get('/api/job-query/resources');
if (data.error) {
document.getElementById('equipmentList').innerHTML = `<div class="error">${data.error}</div>`;
return;
}
allEquipments = data.data;
renderEquipmentList(allEquipments);
} catch (error) {
document.getElementById('equipmentList').innerHTML = `<div class="error">載入失敗: ${error.message}</div>`;
}
}
// Render equipment list
function renderEquipmentList(equipments) {
const container = document.getElementById('equipmentList');
if (!equipments || equipments.length === 0) {
container.innerHTML = '<div class="empty-state">無設備資料</div>';
return;
}
let html = '';
let currentWorkcenter = null;
equipments.forEach(eq => {
const isSelected = selectedEquipments.has(eq.RESOURCEID);
// Group header
if (eq.WORKCENTERNAME !== currentWorkcenter) {
currentWorkcenter = eq.WORKCENTERNAME;
html += `<div style="padding: 8px 15px; background: #f0f0f0; font-weight: 600; font-size: 12px; color: #666;">${currentWorkcenter || '未分類'}</div>`;
}
html += `
<div class="equipment-item ${isSelected ? 'selected' : ''}" onclick="toggleEquipment('${eq.RESOURCEID}')">
<input type="checkbox" ${isSelected ? 'checked' : ''} onclick="event.stopPropagation(); toggleEquipment('${eq.RESOURCEID}')">
<div class="equipment-info">
<div class="equipment-name">${eq.RESOURCENAME}</div>
<div class="equipment-workcenter">${eq.RESOURCEFAMILYNAME || ''}</div>
</div>
</div>
`;
});
container.innerHTML = html;
}
// Toggle equipment dropdown
function toggleEquipmentDropdown() {
const dropdown = document.getElementById('equipmentDropdown');
dropdown.classList.toggle('show');
}
// Filter equipments by search
function filterEquipments(query) {
const q = query.toLowerCase();
const filtered = allEquipments.filter(eq =>
(eq.RESOURCENAME && eq.RESOURCENAME.toLowerCase().includes(q)) ||
(eq.WORKCENTERNAME && eq.WORKCENTERNAME.toLowerCase().includes(q)) ||
(eq.RESOURCEFAMILYNAME && eq.RESOURCEFAMILYNAME.toLowerCase().includes(q))
);
renderEquipmentList(filtered);
}
// Toggle equipment selection
function toggleEquipment(resourceId) {
if (selectedEquipments.has(resourceId)) {
selectedEquipments.delete(resourceId);
} else {
selectedEquipments.add(resourceId);
}
updateSelectedDisplay();
renderEquipmentList(allEquipments.filter(eq => {
const search = document.querySelector('.equipment-search');
if (!search || !search.value) return true;
const q = search.value.toLowerCase();
return (eq.RESOURCENAME && eq.RESOURCENAME.toLowerCase().includes(q)) ||
(eq.WORKCENTERNAME && eq.WORKCENTERNAME.toLowerCase().includes(q));
}));
}
// Update selected display
function updateSelectedDisplay() {
const display = document.getElementById('equipmentDisplay');
const count = document.getElementById('selectedCount');
if (selectedEquipments.size === 0) {
display.textContent = '點擊選擇設備...';
count.textContent = '';
} else if (selectedEquipments.size <= 3) {
const names = allEquipments
.filter(eq => selectedEquipments.has(eq.RESOURCEID))
.map(eq => eq.RESOURCENAME)
.join(', ');
display.textContent = names;
count.textContent = `已選擇 ${selectedEquipments.size} 台設備`;
} else {
display.textContent = `已選擇 ${selectedEquipments.size} 台設備`;
count.textContent = '';
}
}
// Set last 90 days
function setLast90Days() {
const today = new Date();
const past = new Date();
past.setDate(today.getDate() - 90);
document.getElementById('dateFrom').value = past.toISOString().split('T')[0];
document.getElementById('dateTo').value = today.toISOString().split('T')[0];
}
// Validate inputs
function validateInputs() {
if (selectedEquipments.size === 0) {
Toast.error('請選擇至少一台設備');
return false;
}
const dateFrom = document.getElementById('dateFrom').value;
const dateTo = document.getElementById('dateTo').value;
if (!dateFrom || !dateTo) {
Toast.error('請指定日期範圍');
return false;
}
const from = new Date(dateFrom);
const to = new Date(dateTo);
if (to < from) {
Toast.error('結束日期不可早於起始日期');
return false;
}
const daysDiff = (to - from) / (1000 * 60 * 60 * 24);
if (daysDiff > 365) {
Toast.error('日期範圍不可超過 365 天');
return false;
}
return true;
}
// Query jobs
async function queryJobs() {
if (!validateInputs()) return;
const resultSection = document.getElementById('resultSection');
resultSection.innerHTML = `
<div class="loading">
<div class="loading-spinner"></div>
<br>查詢中...
</div>
`;
document.getElementById('queryBtn').disabled = true;
document.getElementById('exportBtn').disabled = true;
try {
const data = await MesApi.post('/api/job-query/jobs', {
resource_ids: Array.from(selectedEquipments),
start_date: document.getElementById('dateFrom').value,
end_date: document.getElementById('dateTo').value
});
if (data.error) {
resultSection.innerHTML = `<div class="error">${data.error}</div>`;
return;
}
jobsData = data.data;
expandedJobs.clear();
renderJobsTable();
document.getElementById('exportBtn').disabled = jobsData.length === 0;
} catch (error) {
resultSection.innerHTML = `<div class="error">查詢失敗: ${error.message}</div>`;
} finally {
document.getElementById('queryBtn').disabled = false;
}
}
// Render jobs table
function renderJobsTable() {
const resultSection = document.getElementById('resultSection');
if (!jobsData || jobsData.length === 0) {
resultSection.innerHTML = `
<div class="empty-state">
<p>無符合條件的工單</p>
</div>
`;
return;
}
let html = `
<div class="result-header">
<div class="result-info">共 ${jobsData.length} 筆工單</div>
<div class="result-actions">
<button class="btn btn-secondary btn-sm" onclick="expandAll()">全部展開</button>
<button class="btn btn-secondary btn-sm" onclick="collapseAll()">全部收合</button>
</div>
</div>
<div class="table-container">
<table>
<thead>
<tr>
<th style="width: 40px;"></th>
<th>設備名稱</th>
<th>工單 ID</th>
<th>狀態</th>
<th>類型</th>
<th>建立時間</th>
<th>完成時間</th>
<th>故障碼</th>
<th>維修碼</th>
</tr>
</thead>
<tbody>
`;
jobsData.forEach((job, idx) => {
const isExpanded = expandedJobs.has(job.JOBID);
html += `
<tr class="job-row ${isExpanded ? 'expanded' : ''}" id="job-row-${idx}">
<td>
<button class="expand-btn" onclick="toggleJobHistory('${job.JOBID}', ${idx})">
<span class="arrow-icon ${isExpanded ? 'rotated' : ''}">▶</span>
</button>
</td>
<td>${job.RESOURCENAME || ''}</td>
<td>${job.JOBID || ''}</td>
<td><span class="status-badge ${job.JOBSTATUS || ''}">${job.JOBSTATUS || ''}</span></td>
<td>${job.JOBMODELNAME || ''}</td>
<td>${formatDate(job.CREATEDATE)}</td>
<td>${formatDate(job.COMPLETEDATE)}</td>
<td>${job.CAUSECODENAME || ''}</td>
<td>${job.REPAIRCODENAME || ''}</td>
</tr>
<tr class="txn-history-row ${isExpanded ? 'show' : ''}" id="txn-row-${idx}">
<td colspan="9" class="txn-history-cell">
<div id="txn-content-${idx}">
${isExpanded ? '<div class="loading"><div class="loading-spinner"></div></div>' : ''}
</div>
</td>
</tr>
`;
});
html += `
</tbody>
</table>
</div>
`;
resultSection.innerHTML = html;
// Load expanded histories
expandedJobs.forEach(jobId => {
const idx = jobsData.findIndex(j => j.JOBID === jobId);
if (idx >= 0) loadJobHistory(jobId, idx);
});
}
// Toggle job history
async function toggleJobHistory(jobId, idx) {
const txnRow = document.getElementById(`txn-row-${idx}`);
const jobRow = document.getElementById(`job-row-${idx}`);
const arrow = jobRow.querySelector('.arrow-icon');
if (expandedJobs.has(jobId)) {
expandedJobs.delete(jobId);
txnRow.classList.remove('show');
jobRow.classList.remove('expanded');
arrow.classList.remove('rotated');
} else {
expandedJobs.add(jobId);
txnRow.classList.add('show');
jobRow.classList.add('expanded');
arrow.classList.add('rotated');
loadJobHistory(jobId, idx);
}
}
// Load job history
async function loadJobHistory(jobId, idx) {
const container = document.getElementById(`txn-content-${idx}`);
container.innerHTML = '<div class="loading" style="padding: 20px;"><div class="loading-spinner"></div></div>';
try {
const data = await MesApi.get(`/api/job-query/txn/${jobId}`);
if (data.error) {
container.innerHTML = `<div class="error" style="margin: 10px 20px;">${data.error}</div>`;
return;
}
if (!data.data || data.data.length === 0) {
container.innerHTML = '<div style="padding: 20px; color: #666;">無交易歷史記錄</div>';
return;
}
let html = `
<table class="txn-history-table">
<thead>
<tr>
<th>交易時間</th>
<th>原狀態</th>
<th>新狀態</th>
<th>階段</th>
<th>故障碼</th>
<th>維修碼</th>
<th>操作者</th>
<th>備註</th>
</tr>
</thead>
<tbody>
`;
data.data.forEach(txn => {
html += `
<tr>
<td>${formatDate(txn.TXNDATE)}</td>
<td><span class="status-badge ${txn.FROMJOBSTATUS || ''}">${txn.FROMJOBSTATUS || '-'}</span></td>
<td><span class="status-badge ${txn.JOBSTATUS || ''}">${txn.JOBSTATUS || ''}</span></td>
<td>${txn.STAGENAME || ''}</td>
<td>${txn.CAUSECODENAME || ''}</td>
<td>${txn.REPAIRCODENAME || ''}</td>
<td>${txn.USER_NAME || txn.EMP_NAME || ''}</td>
<td>${txn.COMMENTS || ''}</td>
</tr>
`;
});
html += '</tbody></table>';
container.innerHTML = html;
} catch (error) {
container.innerHTML = `<div class="error" style="margin: 10px 20px;">載入失敗: ${error.message}</div>`;
}
}
// Expand all
function expandAll() {
jobsData.forEach((job, idx) => {
if (!expandedJobs.has(job.JOBID)) {
expandedJobs.add(job.JOBID);
}
});
renderJobsTable();
}
// Collapse all
function collapseAll() {
expandedJobs.clear();
renderJobsTable();
}
// Export CSV
async function exportCsv() {
if (!validateInputs()) return;
document.getElementById('exportBtn').disabled = true;
document.getElementById('exportBtn').textContent = '匯出中...';
try {
const response = await fetch('/api/job-query/export', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
resource_ids: Array.from(selectedEquipments),
start_date: document.getElementById('dateFrom').value,
end_date: document.getElementById('dateTo').value
})
});
if (!response.ok) {
const data = await response.json();
throw new Error(data.error || '匯出失敗');
}
// Download file
const blob = await response.blob();
const url = window.URL.createObjectURL(blob);
const a = document.createElement('a');
a.href = url;
a.download = `job_history_${document.getElementById('dateFrom').value}_${document.getElementById('dateTo').value}.csv`;
document.body.appendChild(a);
a.click();
document.body.removeChild(a);
window.URL.revokeObjectURL(url);
Toast.success('CSV 匯出完成');
} catch (error) {
Toast.error('匯出失敗: ' + error.message);
} finally {
document.getElementById('exportBtn').disabled = false;
document.getElementById('exportBtn').textContent = '匯出 CSV';
}
}
// Format date
function formatDate(dateStr) {
if (!dateStr) return '';
return dateStr.replace('T', ' ').substring(0, 19);
}
</script>
{% endblock %}

View File

@@ -322,6 +322,9 @@
{% if can_view_page('/resource-history') %}
<button class="tab" data-target="resourceHistoryFrame">設備歷史績效</button>
{% endif %}
{% if can_view_page('/job-query') %}
<button class="tab" data-target="jobQueryFrame">設備維修查詢</button>
{% endif %}
</div>
<div class="panel">
@@ -341,6 +344,9 @@
{% if can_view_page('/resource-history') %}
<iframe id="resourceHistoryFrame" data-src="/resource-history" title="設備歷史績效"></iframe>
{% endif %}
{% if can_view_page('/job-query') %}
<iframe id="jobQueryFrame" data-src="/job-query" title="設備維修查詢"></iframe>
{% endif %}
</div>
</div>
{% endblock %}

View File

@@ -0,0 +1,320 @@
# -*- coding: utf-8 -*-
"""Integration tests for Job Query API routes.
Tests the API endpoints with mocked service dependencies.
"""
import pytest
import json
from unittest.mock import patch, MagicMock
from mes_dashboard import create_app
@pytest.fixture
def app():
"""Create test Flask application."""
app = create_app()
app.config['TESTING'] = True
return app
@pytest.fixture
def client(app):
"""Create test client."""
return app.test_client()
class TestJobQueryPage:
"""Tests for /job-query page route."""
def test_page_returns_html(self, client):
"""Should return the job query page."""
response = client.get('/job-query')
assert response.status_code == 200
assert b'html' in response.data.lower()
class TestGetResources:
"""Tests for /api/job-query/resources endpoint."""
@patch('mes_dashboard.services.resource_cache.get_all_resources')
def test_get_resources_success(self, mock_get_resources, client):
"""Should return resources list."""
mock_get_resources.return_value = [
{
'RESOURCEID': 'RES001',
'RESOURCENAME': 'Machine-01',
'WORKCENTERNAME': 'WC-A',
'RESOURCEFAMILYNAME': 'FAM-01'
},
{
'RESOURCEID': 'RES002',
'RESOURCENAME': 'Machine-02',
'WORKCENTERNAME': 'WC-B',
'RESOURCEFAMILYNAME': 'FAM-02'
}
]
response = client.get('/api/job-query/resources')
assert response.status_code == 200
data = json.loads(response.data)
assert 'data' in data
assert 'total' in data
assert data['total'] == 2
assert data['data'][0]['RESOURCEID'] in ['RES001', 'RES002']
@patch('mes_dashboard.services.resource_cache.get_all_resources')
def test_get_resources_empty(self, mock_get_resources, client):
"""Should return error when no resources available."""
mock_get_resources.return_value = []
response = client.get('/api/job-query/resources')
assert response.status_code == 500
data = json.loads(response.data)
assert 'error' in data
@patch('mes_dashboard.services.resource_cache.get_all_resources')
def test_get_resources_exception(self, mock_get_resources, client):
"""Should handle exception gracefully."""
mock_get_resources.side_effect = Exception('Database error')
response = client.get('/api/job-query/resources')
assert response.status_code == 500
data = json.loads(response.data)
assert 'error' in data
class TestQueryJobs:
"""Tests for /api/job-query/jobs endpoint."""
def test_missing_resource_ids(self, client):
"""Should return error without resource_ids."""
response = client.post(
'/api/job-query/jobs',
json={
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
assert '設備' in data['error']
def test_empty_resource_ids(self, client):
"""Should return error for empty resource_ids."""
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': [],
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
def test_missing_start_date(self, client):
"""Should return error without start_date."""
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': ['RES001'],
'end_date': '2024-01-31'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
assert '日期' in data['error']
def test_missing_end_date(self, client):
"""Should return error without end_date."""
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': ['RES001'],
'start_date': '2024-01-01'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
def test_invalid_date_range(self, client):
"""Should return error for invalid date range."""
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': ['RES001'],
'start_date': '2024-12-31',
'end_date': '2024-01-01'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
assert '結束日期' in data['error'] or '早於' in data['error']
def test_date_range_exceeds_limit(self, client):
"""Should reject date range > 365 days."""
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': ['RES001'],
'start_date': '2023-01-01',
'end_date': '2024-12-31'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
assert '365' in data['error']
@patch('mes_dashboard.routes.job_query_routes.get_jobs_by_resources')
def test_query_jobs_success(self, mock_query, client):
"""Should return jobs list on success."""
mock_query.return_value = {
'data': [
{'JOBID': 'JOB001', 'RESOURCENAME': 'Machine-01', 'JOBSTATUS': 'Complete'}
],
'total': 1,
'resource_count': 1
}
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': ['RES001'],
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
)
assert response.status_code == 200
data = json.loads(response.data)
assert 'data' in data
assert data['total'] == 1
assert data['data'][0]['JOBID'] == 'JOB001'
@patch('mes_dashboard.routes.job_query_routes.get_jobs_by_resources')
def test_query_jobs_service_error(self, mock_query, client):
"""Should return error from service."""
mock_query.return_value = {'error': '查詢失敗: Database error'}
response = client.post(
'/api/job-query/jobs',
json={
'resource_ids': ['RES001'],
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
class TestQueryJobTxnHistory:
"""Tests for /api/job-query/txn/<job_id> endpoint."""
@patch('mes_dashboard.routes.job_query_routes.get_job_txn_history')
def test_get_txn_history_success(self, mock_query, client):
"""Should return transaction history."""
mock_query.return_value = {
'data': [
{
'JOBTXNHISTORYID': 'TXN001',
'JOBID': 'JOB001',
'TXNDATE': '2024-01-15 10:30:00',
'FROMJOBSTATUS': 'Open',
'JOBSTATUS': 'In Progress'
}
],
'total': 1,
'job_id': 'JOB001'
}
response = client.get('/api/job-query/txn/JOB001')
assert response.status_code == 200
data = json.loads(response.data)
assert 'data' in data
assert data['total'] == 1
assert data['job_id'] == 'JOB001'
@patch('mes_dashboard.routes.job_query_routes.get_job_txn_history')
def test_get_txn_history_service_error(self, mock_query, client):
"""Should return error from service."""
mock_query.return_value = {'error': '查詢失敗: Job not found'}
response = client.get('/api/job-query/txn/INVALID_JOB')
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
class TestExportJobs:
"""Tests for /api/job-query/export endpoint."""
def test_missing_resource_ids(self, client):
"""Should return error without resource_ids."""
response = client.post(
'/api/job-query/export',
json={
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
def test_missing_dates(self, client):
"""Should return error without dates."""
response = client.post(
'/api/job-query/export',
json={
'resource_ids': ['RES001']
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
def test_invalid_date_range(self, client):
"""Should return error for invalid date range."""
response = client.post(
'/api/job-query/export',
json={
'resource_ids': ['RES001'],
'start_date': '2024-12-31',
'end_date': '2024-01-01'
}
)
assert response.status_code == 400
data = json.loads(response.data)
assert 'error' in data
@patch('mes_dashboard.routes.job_query_routes.export_jobs_with_history')
def test_export_success(self, mock_export, client):
"""Should return CSV streaming response."""
# Mock generator that yields CSV content
def mock_generator(*args):
yield '\ufeff設備名稱,工單ID\n'
yield 'Machine-01,JOB001\n'
mock_export.return_value = mock_generator()
response = client.post(
'/api/job-query/export',
json={
'resource_ids': ['RES001'],
'start_date': '2024-01-01',
'end_date': '2024-01-31'
}
)
assert response.status_code == 200
assert 'text/csv' in response.content_type
assert 'attachment' in response.headers.get('Content-Disposition', '')
assert 'job_history_export.csv' in response.headers.get('Content-Disposition', '')

View File

@@ -0,0 +1,170 @@
# -*- coding: utf-8 -*-
"""Unit tests for Job Query service functions.
Tests the core service functions without database dependencies.
"""
import pytest
from mes_dashboard.services.job_query_service import (
validate_date_range,
_build_resource_filter,
_build_resource_filter_sql,
BATCH_SIZE,
MAX_DATE_RANGE_DAYS,
)
class TestValidateDateRange:
"""Tests for validate_date_range function."""
def test_valid_range(self):
"""Should return None for valid date range."""
result = validate_date_range('2024-01-01', '2024-01-31')
assert result is None
def test_same_day(self):
"""Should allow same day as start and end."""
result = validate_date_range('2024-01-01', '2024-01-01')
assert result is None
def test_end_before_start(self):
"""Should reject end date before start date."""
result = validate_date_range('2024-12-31', '2024-01-01')
assert result is not None
assert '結束日期' in result or '早於' in result
def test_exceeds_max_range(self):
"""Should reject date range exceeding limit."""
result = validate_date_range('2023-01-01', '2024-12-31')
assert result is not None
assert str(MAX_DATE_RANGE_DAYS) in result
def test_exactly_max_range(self):
"""Should allow exactly max range days."""
# 365 days from 2024-01-01 is 2024-12-31
result = validate_date_range('2024-01-01', '2024-12-31')
assert result is None
def test_one_day_over_max_range(self):
"""Should reject one day over max range."""
# 366 days
result = validate_date_range('2024-01-01', '2025-01-01')
assert result is not None
assert str(MAX_DATE_RANGE_DAYS) in result
def test_invalid_date_format(self):
"""Should reject invalid date format."""
result = validate_date_range('01-01-2024', '12-31-2024')
assert result is not None
assert '格式' in result or 'format' in result.lower()
def test_invalid_start_date(self):
"""Should reject invalid start date."""
result = validate_date_range('2024-13-01', '2024-12-31')
assert result is not None
assert '格式' in result or 'format' in result.lower()
def test_invalid_end_date(self):
"""Should reject invalid end date."""
result = validate_date_range('2024-01-01', '2024-02-30')
assert result is not None
assert '格式' in result or 'format' in result.lower()
def test_non_date_string(self):
"""Should reject non-date strings."""
result = validate_date_range('abc', 'def')
assert result is not None
assert '格式' in result or 'format' in result.lower()
class TestBuildResourceFilter:
"""Tests for _build_resource_filter function."""
def test_empty_list(self):
"""Should return empty list for empty input."""
result = _build_resource_filter([])
assert result == []
def test_single_id(self):
"""Should return single chunk for single ID."""
result = _build_resource_filter(['RES001'])
assert len(result) == 1
assert result[0] == "'RES001'"
def test_multiple_ids(self):
"""Should join multiple IDs with comma."""
result = _build_resource_filter(['RES001', 'RES002', 'RES003'])
assert len(result) == 1
assert "'RES001'" in result[0]
assert "'RES002'" in result[0]
assert "'RES003'" in result[0]
def test_chunking(self):
"""Should chunk when exceeding batch size."""
# Create more than BATCH_SIZE IDs
ids = [f'RES{i:05d}' for i in range(BATCH_SIZE + 10)]
result = _build_resource_filter(ids)
assert len(result) == 2
# First chunk should have BATCH_SIZE items
assert result[0].count("'") == BATCH_SIZE * 2 # 2 quotes per ID
def test_escape_single_quotes(self):
"""Should escape single quotes in IDs."""
result = _build_resource_filter(["RES'001"])
assert len(result) == 1
assert "RES''001" in result[0] # Escaped
def test_custom_chunk_size(self):
"""Should respect custom chunk size."""
ids = ['RES001', 'RES002', 'RES003', 'RES004', 'RES005']
result = _build_resource_filter(ids, max_chunk_size=2)
assert len(result) == 3 # 2+2+1
class TestBuildResourceFilterSql:
"""Tests for _build_resource_filter_sql function."""
def test_empty_list(self):
"""Should return 1=0 for empty input (no results)."""
result = _build_resource_filter_sql([])
assert result == "1=0"
def test_single_id(self):
"""Should build simple IN clause for single ID."""
result = _build_resource_filter_sql(['RES001'])
assert "j.RESOURCEID IN" in result
assert "'RES001'" in result
def test_multiple_ids(self):
"""Should build IN clause with multiple IDs."""
result = _build_resource_filter_sql(['RES001', 'RES002'])
assert "j.RESOURCEID IN" in result
assert "'RES001'" in result
assert "'RES002'" in result
def test_custom_column(self):
"""Should use custom column name."""
result = _build_resource_filter_sql(['RES001'], column='r.ID')
assert "r.ID IN" in result
def test_large_list_uses_or(self):
"""Should use OR for chunked results."""
# Create more than BATCH_SIZE IDs
ids = [f'RES{i:05d}' for i in range(BATCH_SIZE + 10)]
result = _build_resource_filter_sql(ids)
assert " OR " in result
# Should have parentheses wrapping the OR conditions
assert result.startswith("(")
assert result.endswith(")")
class TestServiceConstants:
"""Tests for service constants."""
def test_batch_size_is_reasonable(self):
"""Batch size should be <= 1000 (Oracle limit)."""
assert BATCH_SIZE <= 1000
def test_max_date_range_is_year(self):
"""Max date range should be 365 days."""
assert MAX_DATE_RANGE_DAYS == 365